构建您的第一个智能智能体团队:使用 ADK 的渐进式天气机器人

在 Colab 中打开

分享到:

本教程基于 Agent Development Kit 的 Quickstart 示例 进行扩展。现在,您已经准备好深入探索并构建一个更加复杂的多智能体系统。

我们将开始构建一个天气机器人智能体团队,逐步在简单的基础上叠加高级功能。从能够查询天气的单个智能体开始,我们将逐步添加如下功能:

• 利用不同的 AI 模型(Gemini、GPT、Claude)。
• 为不同任务设计专门的子智能体(如问候和告别)。
• 在智能体之间实现智能委托。
• 使用持久会话状态赋予智能体记忆能力。
• 使用回调实现关键的安全防护栏。

为什么选择天气机器人团队?

这个用例虽然看似简单,但却提供了一个实用且易于理解的场景,用于探索构建复杂、真实世界智能体应用所需的核心 ADK 概念。您将学习如何构建交互、管理状态、确保安全性,以及协调多个 AI "大脑"协同工作。

ADK 是什么?

提醒一下,ADK 是一个 Python 框架,旨在简化由大语言模型(LLM)驱动的应用程序的开发。它提供了强大的构建块,用于创建能够推理、规划、使用工具、与用户动态交互并有效协作的智能体。

在本高级教程中,您将掌握:

• ✅ 工具定义与使用: 创建为智能体提供特定能力(如获取数据)的 Python 函数(tools),并指导智能体如何有效地使用它们。
• ✅ 多 LLM 灵活性: 通过 LiteLLM 集成配置智能体使用各种领先的大语言模型(Gemini、GPT-4o、Claude Sonnet),允许您为每个任务选择最佳模型。
• ✅ 智能体委托与协作: 设计专门的子智能体并启用用户请求的自动路由(auto flow)到团队中最合适的智能体。
• ✅ 会话状态用于记忆: 利用 Session State 和 ToolContext 使智能体能够跨对话轮次记住信息,从而实现更具上下文的交互。
• ✅ 使用回调的安全防护栏: 实现 before_model_callback 和 before_tool_callback 以根据预定义规则检查、修改或阻止请求/工具使用,增强应用程序的安全性和控制。

预期最终状态:

完成本教程后,您将构建一个功能完整的多智能体天气机器人系统。该系统不仅提供天气信息,还处理对话的礼貌用语,记住最后查询的城市,并在定义的安全边界内运行,所有这些都由 ADK 编排。

前提条件:

• ✅ 扎实的 Python 编程基础。
• ✅ 熟悉大语言模型(LLM)、API 和智能体的概念。
• ❗ 至关重要的是:已完成 ADK 快速入门教程或具备 ADK 基础知识(Agent、Runner、SessionService、基本 Tool 用法)。 本教程直接建立在这些概念之上。
• ✅ 您打算使用的 LLM 的 API 密钥(例如,Gemini 的 Google AI Studio、OpenAI Platform、Anthropic Console)。

---

关于执行环境的说明:

本教程结构适用于交互式笔记本环境,如 Google Colab、Colab Enterprise 或 Jupyter notebooks。请记住以下几点:

• 运行异步代码: 笔记本环境处理异步代码的方式不同。您将看到使用 await(适用于事件循环已在运行的情况,常见于笔记本中)或 asyncio.run()(通常在作为独立 .py 脚本运行或在特定笔记本设置中需要)的示例。代码块为这两种情况提供了指导。
• 手动 Runner/Session 设置: 步骤涉及显式创建 Runner 和 SessionService 实例。展示这种方法是因为它让您对智能体的执行生命周期、会话管理和状态持久化有细粒度的控制。

替代方案:使用 ADK 内置工具(Web UI / CLI / API Server)

如果您更喜欢使用 ADK 标准工具自动处理 runner 和会话管理的设置,可以找到为此目的构建的等效代码这里。该版本旨在直接使用 adk web(用于 Web UI)、adk run(用于 CLI 交互)或 adk api_server(用于公开 API)等命令运行。请遵循该替代资源中提供的 README.md 说明。

---

准备好构建您的智能体团队了吗?让我们开始吧!

注意: 本教程适用于 adk 版本 1.0.0 及以上

# @title 步骤 0: 设置和安装
# 安装 ADK 和 LiteLLM 以支持多模型

!pip install google-adk -q
!pip install litellm -q

print("Installation complete.")

# @title 导入必要的库
import os
import asyncio
from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm # 用于多模型支持
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner
from google.genai import types # 用于创建消息 Content/Parts

import warnings
# 忽略所有警告
warnings.filterwarnings("ignore")

import logging
logging.basicConfig(level=logging.ERROR)

print("Libraries imported.")

# @title 配置 API 密钥(请替换为您的实际密钥!)

# --- 重要:将占位符替换为您的真实 API 密钥 ---

# Gemini API Key (从 Google AI Studio 获取: https://aistudio.google.com/app/apikey)
os.environ["GOOGLE_API_KEY"] = "YOUR_GOOGLE_API_KEY" # <--- 替换

# [可选]
# OpenAI API Key (从 OpenAI Platform 获取: https://platform.openai.com/api-keys)
os.environ['OPENAI_API_KEY'] = 'YOUR_OPENAI_API_KEY' # <--- 替换

# [可选]
# Anthropic API Key (从 Anthropic Console 获取: https://console.anthropic.com/settings/keys)
os.environ['ANTHROPIC_API_KEY'] = 'YOUR_ANTHROPIC_API_KEY' # <--- 替换

# --- 验证密钥(可选检查) ---
print("API Keys Set:")
print(f"Google API Key set: {'Yes' if os.environ.get('GOOGLE_API_KEY') and os.environ['GOOGLE_API_KEY'] != 'YOUR_GOOGLE_API_KEY' else 'No (REPLACE PLACEHOLDER!)'}")
print(f"OpenAI API Key set: {'Yes' if os.environ.get('OPENAI_API_KEY') and os.environ['OPENAI_API_KEY'] != 'YOUR_OPENAI_API_KEY' else 'No (REPLACE PLACEHOLDER!)'}")
print(f"Anthropic API Key set: {'Yes' if os.environ.get('ANTHROPIC_API_KEY') and os.environ['ANTHROPIC_API_KEY'] != 'YOUR_ANTHROPIC_API_KEY' else 'No (REPLACE PLACEHOLDER!)'}")

# 配置 ADK 直接使用 API 密钥(而不是此多模型设置的 Vertex AI)
os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "False"


# @markdown **安全提示:** 最佳做法是安全地管理 API 密钥(例如使用 Colab Secrets 或环境变量),而不是在笔记本中硬编码。替换上面的占位符字符串。

# --- 定义模型常量以便于使用 ---

# 可以在此处参考更多支持的模型: https://ai.google.dev/gemini-api/docs/models#model-variations
MODEL_GEMINI_2_5_FLASH = "gemini-2.5-flash"

# 可以在此处参考更多支持的模型: https://docs.litellm.ai/docs/providers/openai#openai-chat-completion-models
MODEL_GPT_4O = "openai/gpt-4.1" # 您也可以尝试: gpt-4.1-mini, gpt-4o 等。

# 可以在此处参考更多支持的模型: https://docs.litellm.ai/docs/providers/anthropic
MODEL_CLAUDE_SONNET = "anthropic/claude-sonnet-4-20250514" # 您也可以尝试: claude-opus-4-20250514 , claude-3-7-sonnet-20250219 等

print("\nEnvironment configured.")

---

步骤 1:您的第一个智能体 - 基本天气查询

让我们从构建天气机器人的基本组件开始:一个能够执行特定任务的单个智能体 - 查询天气信息。这涉及创建两个核心部分:

1. 工具: 为智能体提供获取天气数据能力的 Python 函数。
2. 智能体: 理解用户请求、知道自己有天气工具并决定何时以及如何使用它的 AI "大脑"。

---

1. 定义工具(get_weather)

在 ADK 中,工具是赋予智能体超出文本生成之外的具体能力的构建块。它们通常是执行特定操作的常规 Python 函数,如调用 API、查询数据库或执行计算。

我们的第一个工具将提供模拟天气报告。这允许我们专注于智能体结构,而暂时不需要外部 API 密钥。稍后,您可以轻松地将此模拟函数替换为调用真实天气服务的函数。

关键概念:文档字符串至关重要! 智能体的 LLM 严重依赖函数的文档字符串来理解:

• 工具做什么。
• 何时使用它。
• 需要什么参数(city: str)。
• 返回什么信息。

最佳实践: 为您的工具编写清晰、描述性且准确的文档字符串。这对 LLM 正确使用工具至关重要。

# @title 定义 get_weather 工具
def get_weather(city: str) -> dict:
    """检索指定城市的当前天气报告。

    Args:
        city (str): 城市名称(例如,"New York"、"London"、"Tokyo")。

    Returns:
        dict: 包含天气信息的字典。
              包含 'status' 键('success' 或 'error')。
              如果是 'success',包含带有天气详情的 'report' 键。
              如果是 'error',包含 'error_message' 键。
    """
    print(f"--- Tool: get_weather called for city: {city} ---") # 记录工具执行
    city_normalized = city.lower().replace(" ", "") # 基本规范化

    # 模拟天气数据
    mock_weather_db = {
        "newyork": {"status": "success", "report": "The weather in New York is sunny with a temperature of 25°C."},
        "london": {"status": "success", "report": "It's cloudy in London with a temperature of 15°C."},
        "tokyo": {"status": "success", "report": "Tokyo is experiencing light rain and a temperature of 18°C."},
    }

    if city_normalized in mock_weather_db:
        return mock_weather_db[city_normalized]
    else:
        return {"status": "error", "error_message": f"Sorry, I don't have weather information for '{city}'."}

# 示例工具用法(可选测试)
print(get_weather("New York"))
print(get_weather("Paris"))

---

2. 定义智能体(weather_agent)

现在,让我们创建智能体本身。ADK 中的 Agent 协调用户、LLM 和可用工具之间的交互。

我们使用几个关键参数对其进行配置:

• name: 此智能体的唯一标识符(例如,"weather_agent_v1")。
• model: 指定使用哪个 LLM(例如,MODEL_GEMINI_2_5_FLASH)。我们将从特定的 Gemini 模型开始。
• description: 智能体整体目的的简明摘要。当其他智能体需要决定是否将任务委托给此智能体时,这一点变得至关重要。
• instruction: 为 LLM 提供关于如何表现、其角色、其目标的详细指导,特别是如何以及何时使用其分配的 tools。
• tools: 包含智能体允许使用的实际 Python 工具函数的列表(例如,[get_weather])。

最佳实践: 提供清晰且具体的 instruction 提示。说明越详细,LLM 就能更好地理解其角色以及如何有效地使用其工具。如果需要,请明确说明错误处理。

最佳实践: 选择描述性的 name 和 description 值。这些值被 ADK 内部使用,对于自动委托等功能至关重要(稍后介绍)。

# @title 定义天气智能体
# 使用之前定义的模型常量之一
AGENT_MODEL = MODEL_GEMINI_2_5_FLASH # 从 Gemini 开始

weather_agent = Agent(
    name="weather_agent_v1",
    model=AGENT_MODEL, # 可以是 Gemini 的字符串或 LiteLlm 对象
    description="Provides weather information for specific cities.",
    instruction="You are a helpful weather assistant. "
                "When the user asks for the weather in a specific city, "
                "use the 'get_weather' tool to find the information. "
                "If the tool returns an error, inform the user politely. "
                "If the tool is successful, present the weather report clearly.",
    tools=[get_weather], # 直接传递函数
)

print(f"Agent '{weather_agent.name}' created using model '{AGENT_MODEL}'.")

---

3. 设置 Runner 和会话服务

要管理对话和执行智能体,我们需要另外两个组件:

• SessionService: 负责管理不同用户和会话的对话历史和状态。InMemorySessionService 是一个简单的实现,将所有内容存储在内存中,适用于测试和简单应用程序。它跟踪交换的消息。我们将在步骤 4 中更多地探索状态持久化。
• Runner: 编排交互流的引擎。它接受用户输入,将其路由到适当的智能体,根据智能体逻辑管理对 LLM 和工具的调用,通过 SessionService 处理会话更新,并产生表示交互进度的事件。

# @title 设置会话服务和 Runner

# --- 会话管理 ---
# 关键概念:SessionService 存储对话历史和状态。
# InMemorySessionService 是本教程的简单、非持久化存储。
session_service = InMemorySessionService()

# 定义用于标识交互上下文的常量
APP_NAME = "weather_tutorial_app"
USER_ID = "user_1"
SESSION_ID = "session_001" # 使用固定 ID 以简化

# 创建将发生对话的特定会话
session = await session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)
print(f"Session created: App='{APP_NAME}', User='{USER_ID}', Session='{SESSION_ID}'")

# --- 或 ---

# 如果作为标准 Python 脚本(.py 文件)运行,请取消以下行的注释:

# async def init_session(app_name:str,user_id:str,session_id:str) -> InMemorySessionService:
#     session = await session_service.create_session(
#         app_name=app_name,
#         user_id=user_id,
#         session_id=session_id
#     )
#     print(f"Session created: App='{app_name}', User='{user_id}', Session='{session_id}'")
#     return session
#
# session = asyncio.run(init_session(APP_NAME,USER_ID,SESSION_ID))

# --- Runner ---
# 关键概念:Runner 编排智能体执行循环。
runner = Runner(
    agent=weather_agent, # 我们要运行的智能体
    app_name=APP_NAME,   # 将运行与我们的应用程序关联
    session_service=session_service # 使用我们的会话管理器
)
print(f"Runner created for agent '{runner.agent.name}'.")

---

4. 与智能体交互

我们需要一种向智能体发送消息并接收其响应的方法。由于 LLM 调用和工具执行可能需要时间,ADK 的 Runner 异步运行。

我们将定义一个 async 辅助函数(call_agent_async),该函数将:

1. 接受用户查询字符串。
2. 将其打包成 ADK Content 格式。
3. 调用 runner.run_async,提供用户/会话上下文和新消息。
4. 遍历 runner 产生的事件。事件代表智能体执行中的步骤(例如,请求的工具调用、接收的工具结果、中间 LLM 思考、最终响应)。
5. 使用 event.is_final_response() 识别并打印最终响应事件。

为什么使用 async? 与 LLM 和潜在工具(如外部 API)的交互是 I/O 密集型操作。使用 asyncio 允许程序有效地处理这些操作,而不会阻塞执行。

# @title 定义智能体交互函数

from google.genai import types # 用于创建消息 Content/Parts

async def call_agent_async(query: str, runner, user_id, session_id):
  """向智能体发送查询并打印最终响应。"""
  print(f"\n>>> User Query: {query}")

  # 准备用户的 ADK 格式消息
  content = types.Content(role='user', parts=[types.Part(text=query)])

  final_response_text = "Agent did not produce a final response." # 默认

  # 关键概念:run_async 执行智能体逻辑并产生事件。
  # 我们遍历事件以找到最终答案。
  async for event in runner.run_async(user_id=user_id, session_id=session_id, new_message=content):
      # 您可以取消注释下面的行以查看执行期间的*所有*事件
      # print(f"  [Event] Author: {event.author}, Type: {type(event).__name__}, Final: {event.is_final_response()}, Content: {event.content}")

      # 关键概念:is_final_response() 标记该轮次的结束消息。
      if event.is_final_response():
          if event.content and event.content.parts:
             # 假设第一部分是文本响应
             final_response_text = event.content.parts[0].text
          elif event.actions and event.actions.escalate: # 处理潜在的错误/升级
             final_response_text = f"Agent escalated: {event.error_message or 'No specific message.'}"
          # 如有需要,在此添加更多检查(例如,特定错误代码)
          break # 找到最终响应后停止处理事件

  print(f"<<< Agent Response: {final_response_text}")

---

5. 运行对话

最后,让我们通过向智能体发送几个查询来测试我们的设置。我们将 async 调用包装在一个主 async 函数中,并使用 await 运行它。

观察输出:

• 查看用户查询。
• 注意当智能体使用工具时出现 --- Tool: get_weather called... --- 日志。
• 观察智能体的最终响应,包括它如何处理天气数据不可用的情况(巴黎)。

# @title 运行初始对话

# 我们需要一个异步函数来 await 我们的交互助手
async def run_conversation():
    await call_agent_async("What is the weather like in London?",
                                       runner=runner,
                                       user_id=USER_ID,
                                       session_id=SESSION_ID)

    await call_agent_async("How about Paris?",
                                       runner=runner,
                                       user_id=USER_ID,
                                       session_id=SESSION_ID) # 期望工具的错误消息

    await call_agent_async("Tell me the weather in New York",
                                       runner=runner,
                                       user_id=USER_ID,
                                       session_id=SESSION_ID)

# 使用 await 在异步上下文中执行对话(如 Colab/Jupyter)
await run_conversation()

# --- 或 ---

# 如果作为标准 Python 脚本(.py 文件)运行,请取消以下行的注释:
# import asyncio
# if __name__ == "__main__":
#     try:
#         asyncio.run(run_conversation())
#     except Exception as e:
#         print(f"An error occurred: {e}")

---

恭喜!您已成功构建并与您的第一个 ADK 智能体进行了交互。它理解用户的请求,使用工具查找信息,并根据工具的结果适当地响应。

在下一步中,我们将探索如何轻松切换为此智能体提供支持的基础语言模型。

步骤 2:使用 LiteLLM 实现多模型 [可选]

在步骤 1 中,我们构建了一个由特定 Gemini 模型驱动的功能性天气智能体。虽然有效,但实际应用程序通常受益于使用不同大语言模型(LLM)的灵活性。为什么?

• 性能: 某些模型在特定任务(例如,编码、推理、创意写作)方面表现出色。
• 成本: 不同的模型有不同的价格点。
• 功能: 模型提供各种功能、上下文窗口大小和微调选项。
• 可用性/冗余: 拥有替代方案可确保即使某个提供商出现问题,您的应用程序仍能正常运行。

ADK 通过与 LiteLLM 库的集成使模型之间的切换变得无缝。LiteLLM 充当超过 100 种不同 LLM 的一致接口。

在此步骤中,我们将:

1. 学习如何配置 ADK Agent 使用来自 OpenAI(GPT)和 Anthropic(Claude)等提供商的模型,使用 LiteLlm 包装器。
2. 定义、配置(使用它们自己的会话和运行器),并立即测试我们的天气智能体实例,每个实例都由不同的 LLM 支持。
3. 与这些不同的智能体进行交互,以观察它们的响应可能存在的潜在差异,即使使用相同的基础工具。

---

1. 导入 LiteLlm

我们在初始设置(步骤 0)期间导入了此内容,但它是多模型支持的关键组件:

# @title 1. 导入 LiteLlm
from google.adk.models.lite_llm import LiteLlm

2. 定义和测试多模型智能体

不仅传递模型名称字符串(默认为 Google 的 Gemini 模型),我们将所需的模型标识符字符串包装在 LiteLlm 类中。

• 关键概念:LiteLlm 包装器: LiteLlm(model="provider/model_name") 语法告诉 ADK 通过 LiteLLM 库将针对此智能体的请求路由到指定的模型提供商。

确保您在步骤 0 中为 OpenAI 和 Anthropic 配置了必要的 API 密钥。我们将使用 call_agent_async 函数(之前定义的,现在接受 runner、user_id 和 session_id)在设置后立即与每个智能体交互。

下面的每个代码块将:

• 使用特定的 LiteLLM 模型(MODEL_GPT_4O 或 MODEL_CLAUDE_SONNET)定义智能体。
• 为该智能体的测试运行创建一个新的、单独的 InMemorySessionService 和特定的会话。这为了演示目的使对话历史保持隔离。
• 创建一个为特定智能体及其会话服务配置的 Runner。
• 立即调用 call_agent_async 发送查询并测试智能体。

最佳实践: 使用模型名称的常量(如步骤 0 中定义的 MODEL_GPT_4O、MODEL_CLAUDE_SONNET)以避免拼写错误并使代码更易于管理。

错误处理: 我们将智能体定义包装在 try...except 块中。这可以防止如果特定提供商的 API 密钥缺失或无效时整个代码单元失败,允许教程继续使用已配置的模型。

首先,让我们创建并测试使用 OpenAI 的 GPT-4o 的智能体。

# @title 定义和测试 GPT 智能体

# 确保 'get_weather' 函数(来自步骤 1)已在您的环境中定义。
# 确保 'call_agent_async' 已在之前定义。

# --- 使用 GPT-4o 的智能体 ---
weather_agent_gpt = None # 初始化为 None
runner_gpt = None      # 将 runner 初始化为 None

try:
    weather_agent_gpt = Agent(
        name="weather_agent_gpt",
        # 关键更改:包装 LiteLLM 模型标识符
        model=LiteLlm(model=MODEL_GPT_4O),
        description="Provides weather information (using GPT-4o).",
        instruction="You are a helpful weather assistant powered by GPT-4o. "
                    "Use the 'get_weather' tool for city weather requests. "
                    "Clearly present successful reports or polite error messages based on the tool's output status.",
        tools=[get_weather], # 重用相同的工具
    )
    print(f"Agent '{weather_agent_gpt.name}' created using model '{MODEL_GPT_4O}'.")

    # InMemorySessionService 是本教程的简单、非持久化存储。
    session_service_gpt = InMemorySessionService() # 创建专用服务

    # 定义用于标识交互上下文的常量
    APP_NAME_GPT = "weather_tutorial_app_gpt" # 此测试的唯一应用名称
    USER_ID_GPT = "user_1_gpt"
    SESSION_ID_GPT = "session_001_gpt" # 使用固定 ID 以简化

    # 创建将发生对话的特定会话
    session_gpt = await session_service_gpt.create_session(
        app_name=APP_NAME_GPT,
        user_id=USER_ID_GPT,
        session_id=SESSION_ID_GPT
    )
    print(f"Session created: App='{APP_NAME_GPT}', User='{USER_ID_GPT}', Session='{SESSION_ID_GPT}'")

    # 为该智能体及其会话服务创建专用 runner
    runner_gpt = Runner(
        agent=weather_agent_gpt,
        app_name=APP_NAME_GPT,       # 使用特定应用名称
        session_service=session_service_gpt # 使用特定会话服务
        )
    print(f"Runner created for agent '{runner_gpt.agent.name}'.")

    # --- 测试 GPT 智能体 ---
    print("\n--- Testing GPT Agent ---")
    # 确保 call_agent_async 使用正确的 runner、user_id、session_id
    await call_agent_async(query = "What's the weather in Tokyo?",
                           runner=runner_gpt,
                           user_id=USER_ID_GPT,
                           session_id=SESSION_ID_GPT)
    # --- 或 ---

    # 如果作为标准 Python 脚本(.py 文件)运行,请取消以下行的注释:
    # import asyncio
    # if __name__ == "__main__":
    #     try:
    #         asyncio.run(call_agent_async(query = "What's the weather in Tokyo?",
    #                      runner=runner_gpt,
    #                       user_id=USER_ID_GPT,
    #                       session_id=SESSION_ID_GPT)
    #     except Exception as e:
    #         print(f"An error occurred: {e}")

except Exception as e:
    print(f"❌ Could not create or run GPT agent '{MODEL_GPT_4O}'. Check API Key and model name. Error: {e}")

接下来,我们将为 Anthropic 的 Claude Sonnet 做同样的操作。

# @title 定义和测试 Claude 智能体

# 确保 'get_weather' 函数(来自步骤 1)已在您的环境中定义。
# 确保 'call_agent_async' 已在之前定义。

# --- 使用 Claude Sonnet 的智能体 ---
weather_agent_claude = None # 初始化为 None
runner_claude = None      # 将 runner 初始化为 None

try:
    weather_agent_claude = Agent(
        name="weather_agent_claude",
        # 关键更改:包装 LiteLLM 模型标识符
        model=LiteLlm(model=MODEL_CLAUDE_SONNET),
        description="Provides weather information (using Claude Sonnet).",
        instruction="You are a helpful weather assistant powered by Claude Sonnet. "
                    "Use the 'get_weather' tool for city weather requests. "
                    "Analyze the tool's dictionary output ('status', 'report'/'error_message'). "
                    "Clearly present successful reports or polite error messages.",
        tools=[get_weather], # 重用相同的工具
    )
    print(f"Agent '{weather_agent_claude.name}' created using model '{MODEL_CLAUDE_SONNET}'.")

    # InMemorySessionService 是本教程的简单、非持久化存储。
    session_service_claude = InMemorySessionService() # 创建专用服务

    # 定义用于标识交互上下文的常量
    APP_NAME_CLAUDE = "weather_tutorial_app_claude" # 唯一应用名称
    USER_ID_CLAUDE = "user_1_claude"
    SESSION_ID_CLAUDE = "session_001_claude" # 使用固定 ID 以简化

    # 创建将发生对话的特定会话
    session_claude = await session_service_claude.create_session(
        app_name=APP_NAME_CLAUDE,
        user_id=USER_ID_CLAUDE,
        session_id=SESSION_ID_CLAUDE
    )
    print(f"Session created: App='{APP_NAME_CLAUDE}', User='{USER_ID_CLAUDE}', Session='{SESSION_ID_CLAUDE}'")

    # 为该智能体及其会话服务创建专用 runner
    runner_claude = Runner(
        agent=weather_agent_claude,
        app_name=APP_NAME_CLAUDE,       # 使用特定应用名称
        session_service=session_service_claude # 使用特定会话服务
        )
    print(f"Runner created for agent '{runner_claude.agent.name}'.")

    # --- 测试 Claude 智能体 ---
    print("\n--- Testing Claude Agent ---")
    # 确保 call_agent_async 使用正确的 runner、user_id、session_id
    await call_agent_async(query = "Weather in London please.",
                           runner=runner_claude,
                           user_id=USER_ID_CLAUDE,
                           session_id=SESSION_ID_CLAUDE)

    # --- 或 ---

    # 如果作为标准 Python 脚本(.py 文件)运行,请取消以下行的注释:
    # import asyncio
    # if __name__ == "__main__":
    #     try:
    #         asyncio.run(call_agent_async(query = "Weather in London please.",
    #                      runner=runner_claude,
    #                       user_id=USER_ID_CLAUDE,
    #                       session_id=SESSION_ID_CLAUDE)
    #     except Exception as e:
    #         print(f"An error occurred: {e}")


except Exception as e:
    print(f"❌ Could not create or run Claude agent '{MODEL_CLAUDE_SONNET}'. Check API Key and model name. Error: {e}")

仔细观察两个代码块的输出。您应该看到:

1. 每个智能体(weather_agent_gpt、weather_agent_claude)都已成功创建(如果 API 密钥有效)。
2. 为每个智能体设置了专用会话和运行器。
3. 每个智能体在处理查询时都正确识别了使用 get_weather 工具的需要(您将看到 --- Tool: get_weather called... --- 日志)。
4. 基础工具逻辑保持不变,始终返回我们的模拟数据。
5. 然而,每个智能体生成的最终文本响应可能在措辞、语气或格式上略有不同。这是因为指令提示由不同的 LLM(GPT-4o 与 Claude Sonnet)解释和执行。

此步骤演示了 ADK + LiteLLM 提供的强大功能和灵活性。您可以轻松地使用各种 LLM 进行实验和部署智能体,同时保持核心应用程序逻辑(工具、基本智能体结构)一致。

在下一步中,我们将超越单个智能体,构建一个智能体可以相互委托任务的小团队!

---

步骤 3:构建智能体团队 - 用于问候和告别的委托

在步骤 1 和 2 中,我们构建并测试了一个仅专注于天气查询的单个智能体。虽然在其特定任务上很有效,但实际应用程序通常涉及处理更广泛的用户交互。我们可以继续向单个天气智能体添加更多工具和复杂的指令,但这可能会很快变得难以管理和效率低下。

更稳健的方法是构建智能体团队。这涉及:

1. 创建多个专门的智能体,每个智能体设计用于特定功能(例如,一个用于天气,一个用于问候,一个用于计算)。
2. 指定根智能体(或编排器)以接收初始用户请求。
3. 使根智能体能够根据用户的意图将请求委托给最合适的专用子智能体。

为什么要构建智能体团队?

• 模块化: 更易于开发、测试和维护单个智能体。
• 专业化: 每个智能体都可以针对其特定任务进行微调(指令、模型选择)。
• 可扩展性: 通过添加新智能体更简单地添加新功能。
• 效率: 允许对更简单的任务(如问候)使用潜在更简单/更便宜的模型。

在此步骤中,我们将:

1. 定义用于处理问候(say_hello)和告别(say_goodbye)的简单工具。
2. 创建两个新的专用子智能体:greeting_agent 和 farewell_agent。
3. 更新我们的主天气智能体(weather_agent_v2)以充当根智能体。
4. 使用其子智能体配置根智能体,启用自动委托。
5. 通过向根智能体发送不同类型的请求来测试委托流程。

---

1. 定义子智能体的工具

首先,让我们创建将用作我们新的专家智能体工具的简单 Python 函数。请记住,清晰的文档字符串对于使用它们的智能体至关重要。

# @title 定义问候和告别智能体的工具
from typing import Optional # 确保导入 Optional

# 确保如果独立运行此步骤,'get_weather'(来自步骤 1)可用。
# def get_weather(city: str) -> dict: ... (来自步骤 1)

def say_hello(name: Optional[str] = None) -> str:
    """提供简单的问候。如果提供了名称,将使用该名称。

    Args:
        name (str, optional): 要问候的人的名称。如果未提供,默认为通用问候。

    Returns:
        str: 友好的问候消息。
    """
    if name:
        greeting = f"Hello, {name}!"
        print(f"--- Tool: say_hello called with name: {name} ---")
    else:
        greeting = "Hello there!" # 如果 name 为 None 或未显式传递的默认问候
        print(f"--- Tool: say_hello called without a specific name (name_arg_value: {name}) ---")
    return greeting

def say_goodbye() -> str:
    """提供简单的告别消息以结束对话。"""
    print(f"--- Tool: say_goodbye called ---")
    return "Goodbye! Have a great day."

print("Greeting and Farewell tools defined.")

# 可选自测
print(say_hello("Alice"))
print(say_hello()) # 测试无参数(应使用默认"Hello there!")
print(say_hello(name=None)) # 测试 name 显式为 None(应使用默认"Hello there!")

---

2. 定义子智能体(问候和告别)

现在,为我们的专家创建 Agent 实例。注意它们高度专注的 instruction,关键是它们清晰的 description。description 是根智能体用于决定何时委托给这些子智能体的主要信息。

最佳实践: 子智能体 description 字段应准确且简洁地总结其特定功能。这对于有效的自动委托至关重要。

最佳实践: 子智能体 instruction 字段应针对其有限范围进行定制,确切告诉它们做什么以及不做什么(例如,"您的唯一任务是...")。

# @title 定义问候和告别子智能体

# 如果您想使用 Gemini 以外的模型,确保已导入 LiteLlm 并设置了 API 密钥(来自步骤 0/2)
# from google.adk.models.lite_llm import LiteLlm
# MODEL_GPT_4O、MODEL_CLAUDE_SONNET 等应已定义
# 否则,继续使用: model = MODEL_GEMINI_2_5_FLASH

# --- 问候智能体 ---
greeting_agent = None
try:
    greeting_agent = Agent(
        # 为简单任务使用潜在的不同/更便宜的模型
        model = MODEL_GEMINI_2_5_FLASH,
        # model=LiteLlm(model=MODEL_GPT_4O), # 如果您想尝试其他模型
        name="greeting_agent",
        instruction="You are the Greeting Agent. Your ONLY task is to provide a friendly greeting to the user. "
                    "Use the 'say_hello' tool to generate the greeting. "
                    "If the user provides their name, make sure to pass it to the tool. "
                    "Do not engage in any other conversation or tasks.",
        description="Handles simple greetings and hellos using the 'say_hello' tool.", # 对委托至关重要
        tools=[say_hello],
    )
    print(f"✅ Agent '{greeting_agent.name}' created using model '{greeting_agent.model}'.")
except Exception as e:
    print(f"❌ Could not create Greeting agent. Check API Key ({greeting_agent.model}). Error: {e}")

# --- 告别智能体 ---
farewell_agent = None
try:
    farewell_agent = Agent(
        # 可以使用相同或不同的模型
        model = MODEL_GEMINI_2_5_FLASH,
        # model=LiteLlm(model=MODEL_GPT_4O), # 如果您想尝试其他模型
        name="farewell_agent",
        instruction="You are the Farewell Agent. Your ONLY task is to provide a polite goodbye message. "
                    "Use the 'say_goodbye' tool when the user indicates they are leaving or ending the conversation "
                    "(e.g., using words like 'bye', 'goodbye', 'thanks bye', 'see you'). "
                    "Do not perform any other actions.",
        description="Handles simple farewells and goodbyes using the 'say_goodbye' tool.", # 对委托至关重要
        tools=[say_goodbye],
    )
    print(f"✅ Agent '{farewell_agent.name}' created using model '{farewell_agent.model}'.")
except Exception as e:
    print(f"❌ Could not create Farewell agent. Check API Key ({farewell_agent.model}). Error: {e}")

---

3. 定义带有子智能体的根智能体(天气智能体 v2)

现在,我们升级 weather_agent。关键更改是:

• 添加 sub_agents 参数:我们传递包含刚创建的 greeting_agent 和 farewell_agent 实例的列表。
• 更新 instruction: 我们明确告诉根智能体关于其子智能体的信息以及何时应该将任务委托给它们。

关键概念:自动委托(Auto Flow) 通过提供 sub_agents 列表,ADK 启用自动委托。当根智能体收到用户查询时,其 LLM 不仅考虑自己的指令和工具,还考虑每个子智能体的 description。如果 LLM 确定查询与子智能体描述的功能(例如,"Handles simple greetings")更一致,它将自动生成一个特殊的内部操作来转移控制权到该子智能体以进行该轮次。然后,子智能体使用其自己的模型、指令和工具处理查询。

最佳实践: 确保根智能体的指令清楚地指导其委托决策。按名称提及子智能体,并描述应进行委托的条件。

# @title 定义带有子智能体的根智能体

# 确保子智能体已成功创建,然后再定义根智能体。
# 还确保原始的 'get_weather' 工具已定义。
root_agent = None
runner_root = None # 初始化 runner

if greeting_agent and farewell_agent and 'get_weather' in globals():
    # 让我们为根智能体使用强大的 Gemini 模型来处理编排
    root_agent_model = MODEL_GEMINI_2_5_FLASH

    weather_agent_team = Agent(
        name="weather_agent_v2", # 给它一个新的版本名称
        model=root_agent_model,
        description="The main coordinator agent. Handles weather requests and delegates greetings/farewells to specialists.",
        instruction="You are the main Weather Agent coordinating a team. Your primary responsibility is to provide weather information. "
                    "Use the 'get_weather' tool ONLY for specific weather requests (e.g., 'weather in London'). "
                    "You have specialized sub-agents: "
                    "1. 'greeting_agent': Handles simple greetings like 'Hi', 'Hello'. Delegate to it for these. "
                    "2. 'farewell_agent': Handles simple farewells like 'Bye', 'See you'. Delegate to it for these. "
                    "Analyze the user's query. If it's a greeting, delegate to 'greeting_agent'. If it's a farewell, delegate to 'farewell_agent'. "
                    "If it's a weather request, handle it yourself using 'get_weather'. "
                    "For anything else, respond appropriately or state you cannot handle it.",
        tools=[get_weather], # 根智能体仍然需要天气工具来完成其核心任务
        # 关键更改:在此链接子智能体!
        sub_agents=[greeting_agent, farewell_agent]
    )
    print(f"✅ Root Agent '{weather_agent_team.name}' created using model '{root_agent_model}' with sub-agents: {[sa.name for sa in weather_agent_team.sub_agents]}")

else:
    print("❌ Cannot create root agent because one or more sub-agents failed to initialize or 'get_weather' tool is missing.")
    if not greeting_agent: print(" - Greeting Agent is missing.")
    if not farewell_agent: print(" - Farewell Agent is missing.")
    if 'get_weather' not in globals(): print(" - get_weather function is missing.")

---

4. 与智能体团队交互

现在,我们已经定义了带有专用子智能体的根智能体(weather_agent_team - 注意:确保此变量名称与上一个代码块中定义的名称匹配,可能为 # @title Define the Root Agent with Sub-Agents,这可能将其命名为 root_agent),让我们测试委托机制。

以下代码块将:

1. 定义一个 async 函数 run_team_conversation。
2. 在此函数内,为此测试运行创建一个新的、专用的 InMemorySessionService 和特定会话(session_001_agent_team)。这将隔离对话历史以测试团队动态。
3. 创建一个配置为使用我们的 weather_agent_team(根智能体)和专用会话服务的 Runner(runner_agent_team)。
4. 使用我们更新的 call_agent_async 函数向 runner_agent_team 发送不同类型的查询(问候、天气请求、告别)。我们为此特定测试显式传递 runner、用户 ID 和会话 ID。
5. 立即执行 run_team_conversation 函数。

我们期望以下流程:

1. "Hello there!" 查询转到 runner_agent_team。
2. 根智能体(weather_agent_team)接收它,并根据其指令和 greeting_agent 的描述,委托任务。
3. greeting_agent 处理查询,调用其 say_hello 工具,并生成响应。
4. "What is the weather in New York?" 查询未被委托,由根智能体使用其 get_weather 工具直接处理。
5. "Thanks, bye!" 查询被委托给 farewell_agent,后者使用其 say_goodbye 工具。

# @title 与智能体团队交互
import asyncio # 确保 asyncio 已导入

# 确保根智能体(例如,来自上一个单元的 'weather_agent_team' 或 'root_agent')已定义。
# 确保 call_agent_async 函数已定义。

# 检查根智能体变量是否存在于定义对话函数之前
root_agent_var_name = 'root_agent' # 步骤 3 指南中的默认名称
if 'weather_agent_team' in globals(): # 检查用户是否使用了此名称
    root_agent_var_name = 'weather_agent_team'
elif 'root_agent' not in globals():
    print("⚠️ Root agent ('root_agent' or 'weather_agent_team') not found. Cannot define run_team_conversation.")
    # 分配一个虚拟值,以防止如果代码块无论如何都运行时出现 NameError
    root_agent = None # 或设置一个标志以防止执行

# 仅在根智能体存在时定义和运行
if root_agent_var_name in globals() and globals()[root_agent_var_name]:
    # 定义对话逻辑的主异步函数。
    # 此函数内的 'await' 关键字对于异步操作是必要的。
    async def run_team_conversation():
        print("\n--- Testing Agent Team Delegation ---")
        session_service = InMemorySessionService()
        APP_NAME = "weather_tutorial_agent_team"
        USER_ID = "user_1_agent_team"
        SESSION_ID = "session_001_agent_team"
        session = await session_service.create_session(
            app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID
        )
        print(f"Session created: App='{APP_NAME}', User='{USER_ID}', Session='{SESSION_ID}'")

        actual_root_agent = globals()[root_agent_var_name]
        runner_agent_team = Runner( # 或使用 InMemoryRunner
            agent=actual_root_agent,
            app_name=APP_NAME,
            session_service=session_service
        )
        print(f"Runner created for agent '{actual_root_agent.name}'.")

        # --- 使用 await 的交互(在 async def 内正确) ---
        await call_agent_async(query = "Hello there!",
                               runner=runner_agent_team,
                               user_id=USER_ID,
                               session_id=SESSION_ID)
        await call_agent_async(query = "What is the weather in New York?",
                               runner=runner_agent_team,
                               user_id=USER_ID,
                               session_id=SESSION_ID)
        await call_agent_async(query = "Thanks, bye!",
                               runner=runner_agent_team,
                               user_id=USER_ID,
                               session_id=SESSION_ID)

    # --- 执行 `run_team_conversation` 异步函数 ---
    # 根据您的环境选择以下方法之一。
    # 注意:这可能需要所用模型的 API 密钥!

    # 方法 1:直接 await(笔记本/异步 REPL 的默认设置)
    # 如果您的环境支持顶层 await(如 Colab/Jupyter 笔记本),
    # 这意味着事件循环已在运行,因此您可以直接 await 该函数。
    print("Attempting execution using 'await' (default for notebooks)...")
    await run_team_conversation()

    # 方法 2:asyncio.run(用于标准 Python 脚本 [.py])
    # 如果从终端作为标准 Python 脚本运行此代码,
    # 脚本上下文是同步的。需要 `asyncio.run()` 来
    # 创建和管理事件循环以执行您的异步函数。
    # 要使用此方法:
    # 1. 注释掉上面的 `await run_team_conversation()` 行。
    # 2. 取消以下块的注释:
    """
    import asyncio
    if __name__ == "__main__": # 确保仅在直接执行脚本时运行
        print("Executing using 'asyncio.run()' (for standard Python scripts)...")
        try:
            # 这会创建事件循环,运行您的异步函数,并关闭循环。
            asyncio.run(run_team_conversation())
        except Exception as e:
            print(f"An error occurred: {e}")
    """

else:
    # 如果根智能体变量早些时候未找到,则打印此消息
    print("\n⚠️ Skipping agent team conversation execution as the root agent was not successfully defined in a previous step.")

---

仔细观察输出日志,尤其是 --- Tool: ... called --- 消息。您应该观察:

• 对于 "Hello there!",调用了 say_hello 工具(表明 greeting_agent 处理了它)。
• 对于 "What is the weather in New York?",调用了 get_weather 工具(表明根智能体处理了它)。
• 对于 "Thanks, bye!",调用了 say_goodbye 工具(表明 farewell_agent 处理了它)。

这确认了成功的自动委托!根智能体在其指令和 sub_agents 的 description 的指导下,正确地将用户请求路由到团队内合适的专家智能体。

您现在已经构建了具有多个协作智能体的应用程序结构。这种模块化设计是构建更复杂和功能强大的智能体系统的基础。在下一步中,我们将通过会话状态赋予我们的智能体跨轮次记忆信息的能力。

步骤 4:使用会话状态添加记忆和个性化

到目前为止,我们的智能体团队可以通过委托处理不同的任务,但每次交互都从头开始 - 智能体对会话中的过去对话或用户偏好没有记忆。要创建更复杂和具有上下文意识的体验,智能体需要记忆。ADK 通过会话状态提供此功能。

什么是会话状态?

• 它是绑定到特定用户会话(由 APP_NAME、USER_ID、SESSION_ID 标识)的 Python 字典(session.state)。
• 它在该会话内跨多个对话轮次持久化信息。
• 智能体和工具可以读取和写入此状态,从而允许它们记住详细信息、调整行为和个性化响应。

智能体如何与状态交互:

1. ToolContext(主要方法): 工具可以接受 ToolContext 对象(如果声明为最后一个参数,则由 ADK 自动提供)。此对象通过 tool_context.state 直接访问会话状态,允许工具在执行期间读取偏好或保存结果。
2. output_key(自动保存智能体响应): 可以使用 output_key="your_key" 配置 Agent。ADK 然后将智能体该轮次的最终文本响应自动保存到 session.state["your_key"]。

在此步骤中,我们将通过以下方式增强我们的天气机器人团队:

1. 使用新的 InMemorySessionService 来隔离演示状态。
2. 使用用户的 temperature_unit 偏好初始化会话状态。
3. 创建天气工具的状态感知版本(get_weather_stateful),该版本通过 ToolContext 读取此偏好并调整其输出格式(摄氏度/华氏度)。
4. 更新根智能体以使用此状态感知工具,并使用 output_key 配置它,以自动将其最终天气报告保存到会话状态。
5. 运行对话以观察初始状态如何影响工具,手动状态更改如何改变后续行为,以及 output_key 如何持久化智能体的响应。

---

1. 初始化新的会话服务和状态

为了在不干扰先前步骤的情况下清楚地演示状态管理,我们将实例化一个新的 InMemorySessionService。我们还将创建一个带有初始状态的会话,定义用户的首选温度单位。

```python

@title 1. 初始化新的会话服务和状态

导入必要的会话组件

from google.adk.sessions import InMemorySessionService

为此状态演示创建新的会话服务实例

session_service_stateful = InMemorySessionService() print("✅ New InMemorySessionService created for state demonstration.")

为本教程的此部分定义新的会话 ID

SESSION_ID_STATEFUL = "session_state_demo_001" USER_ID_STATEFUL = "user_state_demo"

定义初始状态数据 - 用户最初偏好摄氏度

initial_state = { "user_preference_temperature_unit": "Celsius" }