Agent Platform SDK for Python 中的 AdkApp 类，您可以创建返回汇率并管理有状态交互的代理。
本文档介绍了如何开发 ADK 智能体，包括定义模型、添加工具以及管理会话和记忆。
如需详细了解如何管理已部署的代理，请参阅管理已部署的代理。
准备工作
请确保按照设置环境中的步骤设置您的环境。
定义和配置模型
指定要使用的模型：
model = "gemini-2.0-flash"
可选：配置模型的安全设置。如需详细了解可用于 Gemini 中安全设置的选项，请参阅配置安全属性。 以下示例展示了如何配置安全设置：
from google.genai import types
safety_settings = [
types.SafetySetting(
category=types.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,
threshold=types.HarmBlockThreshold.OFF,
),
]
可选：指定内容生成参数：
from google.genai import types
generate_content_config = types.GenerateContentConfig(
safety_settings=safety_settings,
temperature=0.28,
max_output_tokens=1000,
top_p=0.95,
)
使用模型配置创建 AdkApp：
from google.adk.agents import Agent
from vertexai.agent_engines import AdkApp
agent = Agent(
model=model, # Required.
name='currency_exchange_agent', # Required.
generate_content_config=generate_content_config, # Optional.
)
app = AdkApp(agent=agent)
如果您在交互式环境（例如终端或 Colab 笔记本）中运行，则可以使用 AdkApp.async_stream_query 方法运行查询作为中间测试步骤：
async for event in app.async_stream_query(
user_id="USER_ID", # Required
message="What is the exchange rate from US dollars to Swedish currency?",
):
print(event)
- USER_ID：选择您自己的用户 ID，字符数限制为 128。
例如
user-123。
响应是类似于以下示例的 Python 字典：
{'actions': {'artifact_delta': {},
'requested_auth_configs': {},
'state_delta': {}},
'author': 'currency_exchange_agent',
'content': {'parts': [{'text': 'To provide you with the most accurate '
'exchange rate, I need to know the specific '
'currencies you\'re asking about. "Swedish '
'currency" could refer to:\n'
'\n'
'* **Swedish Krona (SEK):** This is the '
'official currency of Sweden.\n'
'\n'
"Please confirm if you're interested in the "
'exchange rate between USD and SEK. Once you '
'confirm, I can fetch the latest exchange rate '
'for you.\n'}],
'role': 'model'},
'id': 'LYg7wg8G',
'invocation_id': 'e-113ca547-0f19-4d50-9dde-f76cbc001dce',
'timestamp': 1744166956.925927}
可选：定义和使用工具
定义模型后，定义模型用于推理的工具。
定义函数时，请务必添加能完整而清晰地描述函数的参数、函数的用途以及函数返回内容的注释。模型会使用此信息来确定要使用的函数。您还必须在本地测试函数，以确认其是否正常运行。
使用以下代码定义一个返回汇率的函数：
def get_exchange_rate(
currency_from: str = "USD",
currency_to: str = "EUR",
currency_date: str = "latest",
):
"""Retrieves the exchange rate between two currencies on a specified date.
Uses the Frankfurter API (https://api.frankfurter.app/) to obtain
exchange rate data.
Args:
currency_from: The base currency (3-letter currency code).
Defaults to "USD" (US Dollar).
currency_to: The target currency (3-letter currency code).
Defaults to "EUR" (Euro).
currency_date: The date for which to retrieve the exchange rate.
Defaults to "latest" for the most recent exchange rate data.
Can be specified in YYYY-MM-DD format for historical rates.
Returns:
dict: A dictionary containing the exchange rate information.
Example: {"amount": 1.0, "base": "USD", "date": "2023-11-24",
"rates": {"EUR": 0.95534}}
"""
import requests
response = requests.get(
f"https://api.frankfurter.app/{currency_date}",
params={"from": currency_from, "to": currency_to},
)
return response.json()
如需在代理中使用函数之前先对其进行测试，请运行以下命令：
get_exchange_rate(currency_from="USD", currency_to="SEK")
响应应该类似以下内容：
{'amount': 1.0, 'base': 'USD', 'date': '2025-04-03', 'rates': {'SEK': 9.6607}}
如需在 AdkApp 内使用该工具，请将其添加到 tools= 参数下的工具列表中：
from google.adk.agents import Agent
agent = Agent(
model=model, # Required.
name='currency_exchange_agent', # Required.
tools=[get_exchange_rate], # Optional.
)
您可以使用 AdkApp.async_stream_query 方法对智能体执行测试查询，从而在本地测试该智能体。运行以下命令，使用美元和瑞典克朗在本地测试代理：
from vertexai.agent_engines import AdkApp
app = AdkApp(agent=agent)
async for event in app.async_stream_query(
user_id="USER_ID",
message="What is the exchange rate from US dollars to SEK on 2025-04-03?",
):
print(event)
其中，USER_ID 是您定义的用户 ID。例如 user-123。
响应是一个类似于以下内容的字典序列：
{'author': 'currency_exchange_agent',
'content': {'parts': [{'function_call': {'args': {'currency_date': '2025-04-03',
'currency_from': 'USD',
'currency_to': 'SEK'},
'id': 'adk-e39f3ba2-fa8c-4169-a63a-8e4c62b89818',
'name': 'get_exchange_rate'}}],
'role': 'model'},
'id': 'zFyIaaif',
# ...
}
{'author': 'currency_exchange_agent',
'content': {'parts': [{'function_response': {'id': 'adk-e39f3ba2-fa8c-4169-a63a-8e4c62b89818',
'name': 'get_exchange_rate',
'response': {'amount': 1.0,
'base': 'USD',
'date': '2025-04-03',
'rates': {'SEK': 9.6607}}}}],
'role': 'user'},
'id': 'u2YR4Uom',
# ...
}
{'author': 'currency_exchange_agent',
'content': {'parts': [{'text': 'The exchange rate from USD to SEK on '
'2025-04-03 is 9.6607.'}],
'role': 'model'},
'id': 'q3jWA3wl',
# ...
}
可选：管理会话
AdkApp 在本地运行时使用内存中会话，在将智能体部署到 Agent Runtime 后使用基于云的托管式会话。本部分介绍如何配置 ADK 智能体以与托管式会话搭配使用。
可选：自定义会话数据库
如果您想使用自己的数据库替换默认的托管式会话服务，可以按如下方式定义 session_service_builder 函数：
def session_service_builder():
from google.adk.sessions import InMemorySessionService
return InMemorySessionService()
将数据库作为 session_service_builder= 传递给 AdkApp：
from vertexai.agent_engines import AdkApp
app = AdkApp(
agent=agent, # Required.
session_service_builder=session_service_builder, # Optional.
)
将智能体与会话搭配使用
在本地运行 AdkApp 时，以下指令使用内存中会话。
如需为智能体创建会话，请使用 AdkApp.async_create_session 方法：
session = await app.async_create_session(user_id="USER_ID")
print(session)
会话以 ADK 会话对象的字典表示形式创建。
如需列出与智能体关联的会话，请使用 AdkApp.async_list_sessions 方法：
await app.async_list_sessions(user_id="USER_ID")
如需获取特定会话，请使用 AdkApp.async_get_session 方法：
session = await app.async_get_session(user_id="