开发智能体开发套件智能体

本页面介绍了如何使用智能体开发套件模板（Vertex AI SDK for Python 中的 AdkApp 类）来开发智能体。智能体会返回指定日期两种货币之间的汇率。

按照以下步骤操作：

1. 定义和配置模型
2. （可选）定义和使用工具
3. （可选）管理会话
4. （可选）管理记忆

准备工作

请确保按照设置环境中的步骤设置您的环境。

定义和配置模型

定义模型版本：

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 笔记本）中运行，则可以运行查询作为中间测试步骤：

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.
)

您可以通过对智能体执行测试查询在本地测试该智能体。运行以下命令，使用美元和瑞典克朗在本地测试代理：

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 在本地运行时使用内存中会话，在将智能体部署到 Vertex AI Agent Engine 后使用基于云的托管式会话。本部分介绍如何配置 ADK 智能体以与托管式会话搭配使用。

（可选）自定义会话数据库

注意：如果您使用自定义的内存中会话服务，则在将智能体部署到 Vertex AI Agent Engine 时，会话可能无法保持同步。只有当您能够在已部署的环境中跨会话同步状态时，我们才建议您自定义数据库。

如果您想使用自己的数据库替换默认的托管式会话服务，可以按如下方式定义 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 时，以下指令使用内存中会话：

为智能体创建会话：

session = await app.async_create_session(user_id="USER_ID")
print(session)

会话以 ADK 会话对象的字典表示形式创建。

列出与智能体关联的会话：

await app.async_list_sessions(user_id="USER_ID")

获取特定会话：

session = await app.async_get_session(user_id="USER_ID", session_id="SESSION_ID")

其中

• USER_ID 是您定义的用户 ID。例如 user-123。

• SESSION_ID 是您要检索的特定会话的 ID。

使用会话查询 AdkApp：

async for event in app.async_stream_query(
    user_id="USER_ID",
    session_id=SESSION_ID, # Optional. you can pass in the session_id when querying the agent
    message="What is the exchange rate from US dollars to Swedish currency on 2025-04-03?",
):
    print(event)

智能体可能会响应请求，以提供以下信息：

{'author': 'currency_exchange_agent',
 'content': {'parts': [{'text': 'I need to know the Swedish currency code to '
                                'provide you with the exchange rate.'}],
             'role': 'model'},
 'id': 'wIgZAtQ4',
 #...
}

您可以在与 session 对应的会话中代表 USER_ID 发送响应（例如 "SEK"），方法是指定：

async for event in app.async_stream_query(
    user_id="USER_ID",
    session_id=session.id, # Optional. you can pass in the session_id when querying the agent
    message="SEK",
):
    print(event)

您应该会收到类似如下字典序列的连续对话：

{'author': 'currency_exchange_agent',
 'content': {'parts': [{'function_call': {'args': {'currency_date': '2025-04-03',
                                                   'currency_from': 'USD',
                                                   'currency_to': 'SEK'},
                                          'id': 'adk-2b9230a6-4b92-4a1b-9a65-b708ff6c68b6',
                                          'name': 'get_exchange_rate'}}],
             'role': 'model'},
 'id': 'bOPHtzji',
 # ...
}
{'author': 'currency_exchange_agent',
 'content': {'parts': [{'function_response': {'id': 'adk-2b9230a6-4b92-4a1b-9a65-b708ff6c68b6',
                                              'name': 'get_exchange_rate',
                                              'response': {'amount': 1.0,
                                                           'base': 'USD',
                                                           'date': '2025-04-03',
                                                           'rates': {'SEK': 9.6607}}}}],
             'role': 'user'},
 'id': '9AoDFmiL',
 # ...
}
{'author': 'currency_exchange_agent',
 'content': {'parts': [{'text': 'The exchange rate from USD to SEK on '
                                '2025-04-03 is 1 USD to 9.6607 SEK.'}],
             'role': 'model'},
 'id': 'hmle7trT',
 # ...
}

（可选）管理回忆

默认情况下，AdkApp 在本地运行时使用代理记忆的内存中实现，在将代理部署到 Vertex AI Agent Engine 后使用 Vertex AI Agent Engine 记忆库。

开发 ADK 智能体时，您可以添加一个 PreloadMemoryTool，用于控制智能体何时检索记忆以及如何在提示中包含记忆。以下示例代理总会在每个回合开始时检索记忆，然后将记忆包含在系统指令中：

from google import adk
from vertexai.agent_engines import AdkApp

agent = adk.Agent(
    model="gemini-2.0-flash",
    name='stateful_agent',
    instruction="""You are a Vehicle Voice Agent, designed to assist users with information and in-vehicle actions.

1.  **Direct Action:** If a user requests a specific vehicle function (e.g., "turn on the AC"), execute it immediately using the corresponding tool. You don't have the outcome of the actual tool execution, so provide a hypothetical tool execution outcome.
2.  **Information Retrieval:** Respond concisely to general information requests with your own knowledge (e.g., restaurant recommendation).
3.  **Clarity:** When necessary, try to seek clarification to better understand the user's needs and preference before taking an action.
4.  **Brevity:** Limit responses to under 30 words.
""",
    tools=[adk.tools.preload_memory_tool.PreloadMemoryTool()],
)
app = AdkApp(agent=agent)

（可选）自定义记忆服务

注意： 如果您使用自定义的内存中记忆服务，则在将智能体部署到 Vertex AI Agent Engine 时，记忆可能无法保持同步。只有在部署代理时，您能够在分布式设置中跨实例同步记忆，我们才建议您自定义记忆服务。

如果您想替换默认的内存服务，可以定义一个返回 BaseMemoryService 的 memory_service_builder 函数，如下所示：

def memory_service_builder():
  from google.adk.memory import InMemoryMemoryService

  return InMemoryMemoryService()

将数据库作为 memory_service_builder= 传递给 AdkApp：

from vertexai.agent_engines import AdkApp

app = AdkApp(
   agent=agent,                                    # Required.
   memory_service_builder=memory_service_builder,  # Optional.
)

将智能体与记忆搭配使用

测试具有记忆功能的 ADK 智能体：

1. 创建会话并与代理互动：

   initial_session = await app.async_create_session(user_id="USER_ID")
   
   async for event in app.async_stream_query(
       user_id="USER_ID",
       session_id=initial_session.id,
       message="Can you update the temperature to my preferred temperature?",
   ):
       print(event)
   

   由于在第一个会话期间没有可用的记忆，并且代理不知道任何用户偏好，因此代理可能会回复“您偏好的温度是多少？”等消息。您可以使用以下命令进行回复：

   async for event in app.async_stream_query(
       user_id="USER_ID",
       session_id=initial_session.id,
       message="I like it at 71 degrees",
   ):
       print(event)
   

   代理可能会返回“将温度设为 71 华氏度”之类的回答。已成功更改温度。”智能体的回答可能会因您使用的模型而异。

2. 根据会话生成记忆。如需存储会话中的信息以供日后会话使用，请使用 async_add_session_to_memory 方法：

   await app.async_add_session_to_memory(session=initial_session)
   

3. 通过创建新会话并提示智能体，测试智能体是否保留了会话的记忆（使用 PreloadMemoryTool）：

   new_session = await app.async_create_session(user_id="USER_ID")
   async for event in app.async_stream_query(
       user_id="USER_ID",
       session_id=initial_session.id,
       message="Fix the temperature!",
   ):
       print(event)
   

   智能体可能会返回“将温度设为 71 度”之类的回答。这是否正确？"智能体的回答可能会因您使用的模型和记忆服务提供商而异。

4. 使用 async_search_memory 方法显示代理的记忆：

   response = await app.async_search_memory(
       user_id="USER_ID",
       query="Fix the temperature!",
   )
   print(response)
   

后续步骤

• 会话概览
• Memory Bank 概览
• 使用智能体开发套件智能体。
• 评估智能体。
• 部署代理。
• 排查代理开发问题。
• 获取支持。