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

Open in Colab

本教程是多工具智能体项目的延伸。现在，你已经准备好深入探索，构建一个更复杂的多智能体系统。

我们将着手构建一个天气机器人智能体团队，在简单的基础上逐步叠加高级功能。从一个能够查询天气的单一智能体开始，我们会逐步添加各种能力：

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

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

这个看似简单的用例提供了一个实用且易于理解的平台，来探索构建复杂的真实世界智能体应用所必需的 ADK 核心概念。你将学习如何结构化交互、管理状态、确保安全性，以及编排多个 AI"大脑"协同工作。

ADK 是什么？

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

在本高级教程中，你将掌握：

✅ 工具定义与使用：编写 Python 函数（tools），赋予智能体特定能力（如获取数据），并指导智能体如何有效使用它们。
✅ 多 LLM 灵活性：通过 LiteLLM 集成配置智能体使用各种领先的 LLM（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、基本工具使用）。 本教程直接建立在这些概念之上。
✅ 你打算使用的 LLM 的 API 密钥（如 Google AI Studio 用于 Gemini、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("安装完成。")

# @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("库导入完成。")

# @title 配置 API 密钥（请替换为你自己的密钥！）

# --- 重要：请将占位符替换为你的实际 API 密钥 ---

# Gemini API 密钥（从 Google AI Studio 获取：https://aistudio.google.com/app/apikey）
os.environ["GOOGLE_API_KEY"] = "YOUR_GOOGLE_API_KEY" # <--- 替换

# [可选]
# OpenAI API 密钥（从 OpenAI Platform 获取：https://platform.openai.com/api-keys）
os.environ['OPENAI_API_KEY'] = 'YOUR_OPENAI_API_KEY' # <--- 替换

# [可选]
# Anthropic API 密钥（从 Anthropic Console 获取：https://console.anthropic.com/settings/keys）
os.environ['ANTHROPIC_API_KEY'] = 'YOUR_ANTHROPIC_API_KEY' # <--- 替换

# --- 验证密钥（可选检查） ---
print("API 密钥已设置：")
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 密钥（在此多模型配置中不使用 Agent Platform）
os.environ["GOOGLE_GENAI_USE_ENTERPRISE"] = "False"


# @markdown **安全提示：** 最佳实践是安全管理 API 密钥（例如使用 Colab Secrets 或环境变量），而不是直接在笔记本中硬编码。请替换上面的占位符字符串。

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

# 更多支持的模型可在此查阅：https://ai.google.dev/gemini-api/docs/models#model-variations
MODEL_GEMINI_2_5_FLASH = "gemini-flash-latest"

# 更多支持的模型可在此查阅：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 = "claude-sonnet-4-6" # 你也可以尝试：claude-opus-4-6 等

print("\n环境配置完成。")

第 1 步：你的第一个智能体 —— 基础天气查询¶

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

工具：一个 Python 函数，赋予智能体获取天气数据的能力。
智能体：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"--- 工具：get_weather 被调用，城市：{city} ---") # 记录工具执行
    city_normalized = city.lower().replace(" ", "") # 基本规范化

    # 模拟天气数据
    mock_weather_db = {
        "newyork": {"status": "success", "report": "纽约天气晴朗，温度为 25°C。"},
        "london": {"status": "success", "report": "伦敦多云，温度为 15°C。"},
        "tokyo": {"status": "success", "report": "东京正在下小雨，温度为 18°C。"},
    }

    if city_normalized in mock_weather_db:
        return mock_weather_db[city_normalized]
    else:
        return {"status": "error", "error_message": f"抱歉，我没有 '{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="提供特定城市的天气信息。",
    instruction="你是一个有用的天气助手。"
                "当用户询问特定城市的天气时，"
                "使用 'get_weather' 工具来查找信息。"
                "如果工具返回错误，请礼貌地告知用户。"
                "如果工具成功，请清晰地呈现天气报告。",
    tools=[get_weather], # 直接传递函数
)

print(f"智能体 '{weather_agent.name}' 使用模型 '{AGENT_MODEL}' 创建完成。")

3. 设置 Runner 和 Session Service

要管理对话并执行智能体，我们还需要两个组件：

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

# @title 设置 Session Service 和 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"会话已创建：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 为智能体 '{runner.agent.name}' 创建完成。")

4. 与智能体交互

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

我们将定义一个 async 辅助函数（call_agent_async），它：

接收一个用户查询字符串。
将其打包成 ADK Content 格式。
调用 runner.run_async，提供用户/会话上下文和新消息。
遍历 runner 生成的事件。事件代表智能体执行中的步骤（如工具调用请求、工具结果接收、中间 LLM 思考、最终响应）。
使用 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>>> 用户查询：{query}")

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

  final_response_text = "智能体未产生最终响应。" # 默认值

  # 关键概念：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"智能体升级：{event.error_message or '无具体消息。'}"
          # 如果需要可以在此添加更多检查（如特定错误代码）
          break # 找到最终响应后停止处理事件

  print(f"<<< 智能体响应：{final_response_text}")

5. 运行对话

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

观察输出：

查看用户查询。
注意智能体使用工具时的 --- 工具：get_weather 被调用... --- 日志。
观察智能体的最终响应，包括它如何处理天气数据不可用的情况（巴黎）。

# @title 运行初始对话

# 我们需要一个 async 函数来 await 我们的交互辅助函数
async def run_conversation():
    await call_agent_async("伦敦的天气怎么样？",
                                       runner=runner,
                                       user_id=USER_ID,
                                       session_id=SESSION_ID)

    await call_agent_async("巴黎呢？",
                                       runner=runner,
                                       user_id=USER_ID,
                                       session_id=SESSION_ID) # 预期工具的错误消息

    await call_agent_async("告诉我纽约的天气",
                                       runner=runner,
                                       user_id=USER_ID,
                                       session_id=SESSION_ID)

# 在异步上下文中（如 Colab/Jupyter）使用 await 执行对话
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 的统一接口。

在本步骤中，我们将：

学习如何配置 ADK Agent 使用 LiteLlm 包装器来调用 OpenAI（GPT）和 Anthropic（Claude）等提供商的模型。
定义、配置（使用各自的 session 和 runner），并立即测试我们的天气智能体实例，每个实例由不同的 LLM 支持。
与这些不同的智能体交互，观察即使使用相同的底层工具，其响应也可能存在差异。

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 和会话。这使得对话历史在此演示中保持隔离。
创建为特定智能体及其 session service 配置的 Runner。
立即调用 call_agent_async 发送查询并测试智能体。

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

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

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

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

# 确保步骤 1 中的 'get_weather' 函数已在你的环境中定义。
# 确保前面定义的 '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="提供天气信息（使用 GPT-4o）。",
        instruction="你是一个由 GPT-4o 驱动的有用天气助手。"
                    "使用 'get_weather' 工具处理城市天气请求。"
                    "根据工具输出状态清晰地呈现成功的报告或礼貌的错误消息。",
        tools=[get_weather], # 重用相同的工具
    )
    print(f"智能体 '{weather_agent_gpt.name}' 使用模型 '{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"会话已创建：App='{APP_NAME_GPT}', User='{USER_ID_GPT}', Session='{SESSION_ID_GPT}'")

    # 创建为此智能体及其 session service 专用的 runner
    runner_gpt = Runner(
        agent=weather_agent_gpt,
        app_name=APP_NAME_GPT,       # 使用特定的应用名称
        session_service=session_service_gpt # 使用特定的 session service
        )
    print(f"Runner 为智能体 '{runner_gpt.agent.name}' 创建完成。")

    # --- 测试 GPT 智能体 ---
    print("\n--- 测试 GPT 智能体 ---")
    # 确保 call_agent_async 使用正确的 runner、user_id、session_id
    await call_agent_async(query = "东京的天气怎么样？",
                           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"❌ 无法创建或运行 GPT 智能体 '{MODEL_GPT_4O}'。请检查 API 密钥和模型名称。错误：{e}")

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

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

# 确保步骤 1 中的 'get_weather' 函数已在你的环境中定义。
# 确保前面定义的 '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="提供天气信息（使用 Claude Sonnet）。",
        instruction="你是一个由 Claude Sonnet 驱动的有用天气助手。"
                    "使用 'get_weather' 工具处理城市天气请求。"
                    "分析工具的字典输出（'status'、'report'/'error_message'）。"
                    "清晰地呈现成功的报告或礼貌的错误消息。",
        tools=[get_weather], # 重用相同的工具
    )
    print(f"智能体 '{weather_agent_claude.name}' 使用模型 '{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"会话已创建：App='{APP_NAME_CLAUDE}', User='{USER_ID_CLAUDE}', Session='{SESSION_ID_CLAUDE}'")

    # 创建为此智能体及其 session service 专用的 runner
    runner_claude = Runner(
        agent=weather_agent_claude,
        app_name=APP_NAME_CLAUDE,       # 使用特定的应用名称
        session_service=session_service_claude # 使用特定的 session service
        )
    print(f"Runner 为智能体 '{runner_claude.agent.name}' 创建完成。")

    # --- 测试 Claude 智能体 ---
    print("\n--- 测试 Claude 智能体 ---")
    # 确保 call_agent_async 使用正确的 runner、user_id、session_id
    await call_agent_async(query = "请告诉我伦敦的天气。",
                           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"❌ 无法创建或运行 Claude 智能体 '{MODEL_CLAUDE_SONNET}'。请检查 API 密钥和模型名称。错误：{e}")

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

每个智能体（weather_agent_gpt、weather_agent_claude）都成功创建（如果 API 密钥有效）。
每个智能体都有专用的 session 和 runner 设置。
每个智能体在处理查询时都正确识别需要使用 get_weather 工具（你会看到 --- 工具：get_weather 被调用... --- 日志）。
底层工具逻辑保持不变，始终返回我们的模拟数据。
然而，每个智能体生成的最终文本响应在措辞、语气或格式上可能略有不同。这是因为指令提示由不同的 LLM（GPT-4o 与 Claude Sonnet）解释和执行。

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

在下一步中，我们将超越单一智能体，构建一个小型团队，让智能体之间可以相互委托任务！

第 3 步：构建智能体团队——问候与告别的委托处理¶

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

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

创建多个专门的智能体，每个智能体为特定能力而设计（如一个用于天气，一个用于问候，一个用于计算）。
指定一个根智能体（或编排器）接收初始用户请求。
使根智能体能够根据用户意图将请求委托给最合适的专门子智能体。

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

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

在本步骤中，我们将：

定义用于处理问候（say_hello）和告别（say_goodbye）的简单工具。
创建两个新的专门子智能体：greeting_agent 和 farewell_agent。
更新我们的主天气智能体（weather_agent_v2）作为根智能体。
为根智能体配置其子智能体，启用自动委托。
通过向根智能体发送不同类型的请求来测试委托流程。

1. 为子智能体定义工具

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

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

# 如果独立运行此步骤，请确保步骤 1 中的 'get_weather' 可用。
# 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"你好，{name}！"
        print(f"--- 工具：say_hello 被调用，名字：{name} ---")
    else:
        greeting = "你好！" # 如果 name 为 None 或未显式传递则使用默认问候
        print(f"--- 工具：say_hello 被调用，未指定名字（name 参数值：{name}）---")
    return greeting

def say_goodbye() -> str:
    """提供简单的告别消息以结束对话。"""
    print(f"--- 工具：say_goodbye 被调用 ---")
    return "再见！祝你有美好的一天。"

print("问候和告别工具已定义。")

# 可选自测
print(say_hello("Alice"))
print(say_hello()) # 测试无参数（应使用默认 "你好！"）
print(say_hello(name=None)) # 测试 name 显式为 None（应使用默认 "你好！"）

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="你是问候智能体。你唯一的任务是使用 'say_hello' 工具向用户提供友好的问候。"
                    "如果用户提供了他们的名字，确保将其传递给工具。"
                    "不要参与任何其他对话或任务。",
        description="使用 'say_hello' 工具处理简单的问候和打招呼。", # 委托的关键
        tools=[say_hello],
    )
    print(f"✅ 智能体 '{greeting_agent.name}' 使用模型 '{greeting_agent.model}' 创建完成。")
except Exception as e:
    print(f"❌ 无法创建问候智能体。请检查 API 密钥（{greeting_agent.model}）。错误：{e}")

# --- 告别智能体 ---
farewell_agent = None
try:
    farewell_agent = Agent(
        # 可以使用相同或不同的模型
        model = MODEL_GEMINI_2_5_FLASH,
        # model=LiteLlm(model=MODEL_GPT_4O), # 如果你想尝试其他模型
        name="farewell_agent",
        instruction="你是告别智能体。你唯一的任务是提供礼貌的告别消息。"
                    "当用户表示要离开或结束对话时（如使用 'bye'、'goodbye'、'thanks bye'、'see you' 等词语），"
                    "使用 'say_goodbye' 工具。"
                    "不要执行任何其他操作。",
        description="使用 'say_goodbye' 工具处理简单的告别。", # 委托的关键
        tools=[say_goodbye],
    )
    print(f"✅ 智能体 '{farewell_agent.name}' 使用模型 '{farewell_agent.model}' 创建完成。")
except Exception as e:
    print(f"❌ 无法创建告别智能体。请检查 API 密钥（{farewell_agent.model}）。错误：{e}")

3. 定义带子智能体的根智能体（Weather Agent v2）

现在，我们升级我们的 weather_agent。关键变化是：

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

关键概念：自动委托（Auto Flow） 通过提供 sub_agents 列表，ADK 启用自动委托。当根智能体收到用户查询时，其 LLM 不仅考虑自己的指令和工具，还会考虑每个子智能体的 description。如果 LLM 确定查询更适合某个子智能体描述的能力（如"处理简单的问候"），它将自动生成一个特殊的内部操作来将控制权转移给该子智能体处理该轮次。然后子智能体使用自己的模型、指令和工具来处理查询。

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

# @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="主协调智能体。处理天气请求，并将问候/告别委托给专家。",
        instruction="你是协调团队的主天气智能体。你的主要职责是提供天气信息。"
                    "仅在具体的天气请求（如'伦敦天气'）时使用 'get_weather' 工具。"
                    "你有专门的子智能体："
                    "1. 'greeting_agent'：处理简单的问候如'你好'。将这些委托给它。"
                    "2. 'farewell_agent'：处理简单的告别如'再见'。将这些委托给它。"
                    "分析用户的查询。如果是问候，委托给 'greeting_agent'。如果是告别，委托给 'farewell_agent'。"
                    "如果是天气请求，使用 'get_weather' 自行处理。"
                    "对于其他内容，适当回应或说明你无法处理。",
        tools=[get_weather], # 根智能体仍需要天气工具来执行其核心任务
        # 关键变化：在此链接子智能体！
        sub_agents=[greeting_agent, farewell_agent]
    )
    print(f"✅ 根智能体 '{weather_agent_team.name}' 使用模型 '{root_agent_model}' 创建完成，子智能体：{[sa.name for sa in weather_agent_team.sub_agents]}")

else:
    print("❌ 无法创建根智能体，因为一个或多个子智能体初始化失败或 'get_weather' 工具缺失。")
    if not greeting_agent: print(" - 问候智能体缺失。")
    if not farewell_agent: print(" - 告别智能体缺失。")
    if 'get_weather' not in globals(): print(" - get_weather 函数缺失。")

4. 与智能体团队交互

现在我们已经定义了根智能体（weather_agent_team - 注意：确保此变量名与上一个代码块中定义的一致，可能在 # @title 定义带子智能体的根智能体 中将其命名为 root_agent）及其专门的子智能体，让我们来测试委托机制。

以下代码块将会：

定义一个 async 函数 run_team_conversation。
在此函数内部，创建一个全新的、专用的 InMemorySessionService 和一个特定的会话（session_001_agent_team），专门用于此次测试运行。这可以隔离对话历史，以便测试团队动态。
创建一个 Runner（runner_agent_team），配置为使用我们的 weather_agent_team（根智能体）和专用的 session service。
使用我们更新后的 call_agent_async 函数向 runner_agent_team 发送不同类型的查询（问候、天气请求、告别）。我们显式传递此特定测试的 runner、用户 ID 和会话 ID。
立即执行 run_team_conversation 函数。

我们期望以下流程：

"Hello there!" 查询发送到 runner_agent_team。
根智能体（weather_agent_team）接收它，并根据其指令和 greeting_agent 的描述，委托该任务。
greeting_agent 处理该查询，调用其 say_hello 工具，并生成响应。
"What is the weather in New York?" 查询不会被委托，由根智能体直接使用其 get_weather 工具处理。
"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' 或 'weather_agent_team'）。无法定义 run_team_conversation。")
    # 分配一个虚拟值以防止后续 NameError（如果代码块仍然运行）
    root_agent = None # 或设置标志以防止执行

# 仅在根智能体存在时定义和运行
if root_agent_var_name in globals() and globals()[root_agent_var_name]:
    # 定义对话逻辑的主 async 函数。
    # 此函数内部的 'await' 关键字是异步操作所必需的。
    async def run_team_conversation():
        print("\n--- 测试智能体团队委托 ---")
        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"会话已创建：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 为智能体 '{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` async 函数 ---
    # 根据你的环境选择以下方法之一。
    # 注意：这可能需要所使用模型的 API 密钥！

    # 方法 1：直接 await（笔记本/异步 REPL 的默认方式）
    # 如果你的环境支持顶层 await（如 Colab/Jupyter 笔记本），
    # 意味着事件循环已在运行，因此你可以直接 await 函数。
    print("尝试使用 'await' 执行（笔记本默认方式）...")
    await run_team_conversation()

    # 方法 2：asyncio.run（用于标准 Python 脚本 [.py]）
    # 如果从终端以标准 Python 脚本运行此代码，
    # 脚本上下文是同步的。需要 `asyncio.run()` 来
    # 创建和管理事件循环以执行你的 async 函数。
    # 要使用此方法：
    # 1. 注释掉上面的 `await run_team_conversation()` 行。
    # 2. 取消注释以下代码块：
    """
    import asyncio
    if __name__ == "__main__": # 确保仅在脚本直接执行时运行
        print("使用 'asyncio.run()' 执行（用于标准 Python 脚本）...")
        try:
            # 这会创建一个事件循环，运行你的 async 函数，然后关闭循环。
            asyncio.run(run_team_conversation())
        except Exception as e:
            print(f"发生错误：{e}")
    """

else:
    # 如果之前未找到根智能体变量则打印此消息
    print("\n⚠️ 跳过智能体团队对话执行，因为根智能体未在之前的步骤中成功定义。")

仔细观察输出日志，特别是 --- 工具：... 被调用 --- 消息。你应该观察到：

对于 "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）。
它在该会话中跨多个对话轮次持久化信息。
智能体和工具可以读取和写入此状态，使它们能够记住细节、调整行为和个性化响应。

智能体如何与状态交互：

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

在本步骤中，我们将通过以下方式增强天气机器人团队：

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

1. 初始化新的 Session Service 和状态

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

# @title 1. 初始化新的 Session Service 和状态

# 导入必要的会话组件
from google.adk.sessions import InMemorySessionService

# 为此次状态演示创建新的 session service 实例
session_service_stateful = InMemorySessionService()
print("✅ 为状态演示创建了新的 InMemorySessionService。")

# 为本教程的此部分定义新的 SESSION ID
SESSION_ID_STATEFUL = "session_state_demo_001"
USER_ID_STATEFUL = "user_state_demo"

# 定义初始状态数据 - 用户初始偏好摄氏度
initial_state = {
    "user_preference_temperature_unit": "Celsius"
}

# 创建会话，提供初始状态
session_stateful = await session_service_stateful.create_session(
    app_name=APP_NAME, # 使用一致的应用名称
    user_id=USER_ID_STATEFUL,
    session_id=SESSION_ID_STATEFUL,
    state=initial_state # <<< 在创建时初始化状态
)
print(f"✅ 会话 '{SESSION_ID_STATEFUL}' 为用户 '{USER_ID_STATEFUL}' 创建完成。")

# 验证初始状态已正确设置
retrieved_session = await session_service_stateful.get_session(app_name=APP_NAME,
                                                         user_id=USER_ID_STATEFUL,
                                                         session_id = SESSION_ID_STATEFUL)
print("\n--- 初始会话状态 ---")
if retrieved_session:
    print(retrieved_session.state)
else:
    print("错误：无法获取会话。")

2. 创建状态感知天气工具（get_weather_stateful）

现在，我们创建天气工具的新版本。其关键特性是接受 tool_context: ToolContext，允许它访问 tool_context.state。它将读取 user_preference_temperature_unit 并相应地格式化温度。

关键概念：ToolContext 此对象是允许你的工具逻辑与会话上下文交互的桥梁，包括读取和写入状态变量。如果定义为工具函数的最后一个参数，ADK 会自动注入它。
最佳实践：从状态读取时，使用 dictionary.get('key', default_value) 处理键可能尚不存在的情况，确保你的工具不会崩溃。

from google.adk.tools.tool_context import ToolContext

def get_weather_stateful(city: str, tool_context: ToolContext) -> dict:
    """检索天气信息，并根据会话状态转换单位。"""
    print(f"--- 工具：正在为 {city} 调用 get_weather_stateful ---")

    # --- 从状态中读取偏好设置 ---
    preferred_unit = tool_context.state.get("user_preference_temperature_unit", "Celsius") # 默认为摄氏度
    print(f"--- 工具：读取状态 'user_preference_temperature_unit': {preferred_unit} ---")

    city_normalized = city.lower().replace(" ", "")

    # 模拟天气数据（内部始终存储为摄氏度）
    mock_weather_db = {
        "newyork": {"temp_c": 25, "condition": "晴天"},
        "london": {"temp_c": 15, "condition": "多云"},
        "tokyo": {"temp_c": 18, "condition": "小雨"},
    }

    if city_normalized in mock_weather_db:
        data = mock_weather_db[city_normalized]
        temp_c = data["temp_c"]
        condition = data["condition"]

        # 根据状态偏好格式化温度
        if preferred_unit == "Fahrenheit":
            temp_value = (temp_c * 9/5) + 32 # 计算华氏温度
            temp_unit = "°F"
        else: # 默认为摄氏度
            temp_value = temp_c
            temp_unit = "°C"

        report = f"{city.capitalize()}的天气为{condition}，温度为{temp_value:.0f}{temp_unit}。"
        result = {"status": "success", "report": report}
        print(f"--- 工具：已生成 {preferred_unit} 单位的报告。结果: {result} ---")

        # 写回状态的示例（该工具的可选操作）
        tool_context.state["last_city_checked_stateful"] = city
        print(f"--- 工具：已更新状态 'last_city_checked_stateful': {city} ---")

        return result
    else:
        # 处理未找到城市的情况
        error_msg = f"抱歉，我没有 '{city}' 的天气信息。"
        print(f"--- 工具：未找到城市 '{city}'。 ---")
        return {"status": "error", "error_message": error_msg}

print("✅ 状态感知 'get_weather_stateful' 工具已定义。")

3. 重新定义子智能体并更新根智能体

为确保这一步是自包含的并能正确构建，我们首先按照第 3 步中的方式重新定义 greeting_agent 和 farewell_agent。然后，我们定义新的根智能体（weather_agent_v4_stateful）：

它使用新的 get_weather_stateful 工具。
它包含问候和告别子智能体用于委托。
关键的是，它设置了 output_key="last_weather_report"，这会自动将其最终的天气响应保存到会话状态中。

# @title 3. 重新定义子智能体并使用 output_key 更新根智能体

# 确保必要导入: Agent, LiteLlm, Runner
from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm
from google.adk.runners import Runner
# 确保工具 'say_hello', 'say_goodbye' 已定义（来自第 3 步）
# 确保模型常量 MODEL_GPT_4O, MODEL_GEMINI_2_5_FLASH 等已定义

# --- 重新定义问候智能体（来自第 3 步） ---
greeting_agent = None
try:
    greeting_agent = Agent(
        model=MODEL_GEMINI_2_5_FLASH,
        name="greeting_agent",
        instruction="你是问候智能体。你的唯一任务是使用 'say_hello' 工具提供友好的问候。不要做其他任何事情。",
        description="使用 'say_hello' 工具处理简单的问候和打招呼。",
        tools=[say_hello],
    )
    print(f"✅ 智能体 '{greeting_agent.name}' 已重新定义。")
except Exception as e:
    print(f"❌ 无法重新定义问候智能体。错误: {e}")

# --- 重新定义告别智能体（来自第 3 步） ---
farewell_agent = None
try:
    farewell_agent = Agent(
        model=MODEL_GEMINI_2_5_FLASH,
        name="farewell_agent",
        instruction="你是告别智能体。你的唯一任务是使用 'say_goodbye' 工具提供礼貌的告别信息。不要执行任何其他操作。",
        description="使用 'say_goodbye' 工具处理简单的告别和再见。",
        tools=[say_goodbye],
    )
    print(f"✅ 智能体 '{farewell_agent.name}' 已重新定义。")
except Exception as e:
    print(f"❌ 无法重新定义告别智能体。错误: {e}")

# --- 定义更新后的根智能体 ---
root_agent_stateful = None
runner_root_stateful = None # 初始化 Runner

# 创建根智能体前检查前提条件
if greeting_agent and farewell_agent and 'get_weather_stateful' in globals():

    root_agent_model = MODEL_GEMINI_2_5_FLASH # 选择编排模型

    root_agent_stateful = Agent(
        name="weather_agent_v4_stateful", # 新版本名称
        model=root_agent_model,
        description="主智能体：提供天气（状态感知单位）、委托问候/告别、将报告保存到状态。",
        instruction="你是主天气智能体。你的任务是使用 'get_weather_stateful' 提供天气信息。"
                    "该工具会根据存储在状态中的用户偏好格式化温度。"
                    "将简单问候委托给 'greeting_agent'，将告别委托给 'farewell_agent'。"
                    "只处理天气请求、问候和告别。",
        tools=[get_weather_stateful], # 使用状态感知工具
        sub_agents=[greeting_agent, farewell_agent], # 包含子智能体
        output_key="last_weather_report" # <<< 自动保存智能体的最终天气响应
    )
    print(f"✅ 根智能体 '{root_agent_stateful.name}' 已使用状态感知工具和 output_key 创建。")

    # --- 为此根智能体创建 Runner 和新的会话服务 ---
    runner_root_stateful = Runner(
        agent=root_agent_stateful,
        app_name=APP_NAME,
        session_service=session_service_stateful # 使用新的状态感知会话服务
    )
    print(f"✅ 已为状态感知根智能体 '{runner_root_stateful.agent.name}' 创建 Runner，使用状态感知会话服务。")

else:
    print("❌ 无法创建状态感知根智能体。缺少前提条件。")
    if not greeting_agent: print(" - greeting_agent 定义缺失。")
    if not farewell_agent: print(" - farewell_agent 定义缺失。")
    if 'get_weather_stateful' not in globals(): print(" - get_weather_stateful 工具缺失。")

4. 交互并测试状态流转

现在，让我们执行一段对话来测试状态交互，使用 runner_root_stateful（与我们的状态感知智能体和 session_service_stateful 关联）。我们将使用之前定义的 call_agent_async 函数，确保传入正确的 Runner、用户 ID（USER_ID_STATEFUL）和会话 ID（SESSION_ID_STATEFUL）。

对话流程如下：

检查天气（伦敦）： get_weather_stateful 工具应从第 1 节初始化的会话状态中读取初始的 "Celsius" 偏好。根智能体的最终响应（以摄氏度为单位的天气报告）应通过 output_key 配置保存到 state['last_weather_report']。
手动更新状态： 我们将直接修改存储在 InMemorySessionService 实例（session_service_stateful）中的状态。
- 为什么要直接修改？ session_service.get_session() 方法返回的是会话的副本。修改该副本不会影响后续智能体运行中使用的状态。对于使用 InMemorySessionService 的测试场景，我们访问内部 sessions 字典来更改实际存储的 user_preference_temperature_unit 状态值为 "Fahrenheit"。注意：在实际应用中，状态更改通常由工具或智能体逻辑返回 EventActions(state_delta=...) 来触发，而不是直接手动更新。
再次检查天气（纽约）： get_weather_stateful 工具现在应从状态中读取更新后的 "Fahrenheit" 偏好并相应地转换温度。根智能体的新响应（以华氏度为单位的天气）将由于 output_key 而覆盖 state['last_weather_report'] 中的先前值。
问候智能体： 验证委托给 greeting_agent 的功能在状态感知操作期间仍然正常工作。
检查最终状态： 对话结束后，我们最后一次检索会话（获取副本）并打印其状态，以确认 user_preference_temperature_unit 确实为 "Fahrenheit"，观察 output_key 保存的最终值（在本次运行中将是上一次天气报告），以及查看工具写入的 last_city_checked_stateful 值。

# @title 4. 交互以测试状态流转和 output_key
import asyncio # 确保已导入 asyncio

# 确保状态感知 Runner（runner_root_stateful）在上一个单元中可用
# 确保 call_agent_async, USER_ID_STATEFUL, SESSION_ID_STATEFUL, APP_NAME 已定义

if 'runner_root_stateful' in globals() and runner_root_stateful:
    # 定义主异步函数用于状态感知对话逻辑。
    # 该函数内部的 'await' 关键字对于异步操作是必需的。
    async def run_stateful_conversation():
        print("\n--- 测试状态：温度单位转换和 output_key ---")

        # 1. 检查天气（使用初始状态：摄氏度）
        print("--- 第 1 轮：请求伦敦天气（预期为摄氏度） ---")
        await call_agent_async(query= "What's the weather in London?",
                               runner=runner_root_stateful,
                               user_id=USER_ID_STATEFUL,
                               session_id=SESSION_ID_STATEFUL
                              )

        # 2. 手动将状态偏好更新为华氏度 - 直接修改存储
        print("\n--- 手动更新状态：设置单位为华氏度 ---")
        try:
            # 直接访问内部存储 - 这是 InMemorySessionService 测试专用的
            # 注意：在使用持久化服务（数据库、VertexAI）的生产环境中，你通常会
            # 通过智能体操作或特定的服务 API（如果可用）来更新状态，
            # 而不是直接操作内部存储。
            stored_session = session_service_stateful.sessions[APP_NAME][USER_ID_STATEFUL][SESSION_ID_STATEFUL]
            stored_session.state["user_preference_temperature_unit"] = "Fahrenheit"
            # 可选：如果有逻辑依赖时间戳，你可能还需要更新时间戳
            # import time
            # stored_session.last_update_time = time.time()
            print(f"--- 已更新存储会话状态。当前 'user_preference_temperature_unit': {stored_session.state.get('user_preference_temperature_unit', 'Not Set')} ---") # 使用 .get 保证安全
        except KeyError:
            print(f"--- 错误：无法从内部存储中检索会话 '{SESSION_ID_STATEFUL}'（用户 '{USER_ID_STATEFUL}'，应用 '{APP_NAME}'）以更新状态。请检查 ID 和会话是否已创建。 ---")
        except Exception as e:
             print(f"--- 更新内部会话状态时出错: {e} ---")

        # 3. 再次检查天气（工具现在应使用华氏度）
        # 这也会通过 output_key 更新 'last_weather_report'
        print("\n--- 第 2 轮：请求纽约天气（预期为华氏度） ---")
        await call_agent_async(query= "Tell me the weather in New York.",
                               runner=runner_root_stateful,
                               user_id=USER_ID_STATEFUL,
                               session_id=SESSION_ID_STATEFUL
                              )

        # 4. 测试基本委托（应仍然有效）
        # 这将再次更新 'last_weather_report'，覆盖纽约的天气报告
        print("\n--- 第 3 轮：发送问候 ---")
        await call_agent_async(query= "Hi!",
                               runner=runner_root_stateful,
                               user_id=USER_ID_STATEFUL,
                               session_id=SESSION_ID_STATEFUL
                              )

    # --- 执行 `run_stateful_conversation` 异步函数 ---
    # 根据你的环境选择以下方法之一。

    # 方法 1：直接 await（笔记本/异步 REPL 的默认方式）
    # 如果你的环境支持顶层 await（如 Colab/Jupyter 笔记本），
    # 说明事件循环已在运行，你可以直接 await 该函数。
    print("正在尝试使用 'await' 执行（笔记本默认方式）...")
    await run_stateful_conversation()

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

    # --- 对话后检查最终会话状态 ---
    # 该代码块在任一执行方法完成后运行。
    print("\n--- 检查最终会话状态 ---")
    final_session = await session_service_stateful.get_session(app_name=APP_NAME,
                                                         user_id= USER_ID_STATEFUL,
                                                         session_id=SESSION_ID_STATEFUL)
    if final_session:
        # 使用 .get() 更安全地访问可能缺失的键
        print(f"最终偏好: {final_session.state.get('user_preference_temperature_unit', 'Not Set')}")
        print(f"最终天气报告（来自 output_key）: {final_session.state.get('last_weather_report', 'Not Set')}")
        print(f"最终检查的城市（来自工具）: {final_session.state.get('last_city_checked_stateful', 'Not Set')}")
        # 打印完整状态以获取详细视图
        # print(f"完整状态字典: {final_session.state}") # 用于详细视图
    else:
        print("\n❌ 错误：无法检索最终会话状态。")

else:
    print("\n⚠️ 跳过状态测试对话。状态感知根智能体 Runner（'runner_root_stateful'）不可用。")

通过审查对话流程和最终会话状态输出，你可以确认：

状态读取： 天气工具（get_weather_stateful）正确地从状态中读取了 user_preference_temperature_unit，对伦敦初始使用 "Celsius"。
状态更新： 直接修改成功地将存储的偏好更改为 "Fahrenheit"。
状态读取（更新后）： 当请求纽约天气时，工具随后读取了 "Fahrenheit" 并进行了转换。
工具状态写入： 工具通过 tool_context.state 成功将 last_city_checked_stateful（第二次天气检查后为 "New York"）写入状态。
委托： 在状态修改后，对 greeting_agent 处理 "Hi!" 的委托仍然正常工作。
output_key： output_key="last_weather_report" 成功为每个回合中根智能体最终响应的情况保存了最终响应。在此序列中，最后的问候（"Hello, there!"）是由委托的子智能体生成的，而不是根智能体，因此 output_key 在最后一轮没有被触发，上一次天气报告在会话状态中保持不变。
最终状态： 最终检查确认偏好持久化为 "Fahrenheit"。

你现在已经成功集成了会话状态，使用 ToolContext 来个性化智能体行为，手动操作了 InMemorySessionService 的状态进行测试，并观察了 output_key 如何提供一种简单机制将智能体的最后响应保存到状态中。这种对状态管理的基础理解是我们继续在下一步使用回调实现安全防护的关键。

第 5 步：添加安全防护——使用 `before_model_callback` 的输入安全防护¶

我们的智能体团队正在变得更加强大，能够记住偏好并有效地使用工具。然而，在实际场景中，我们通常需要安全机制来在可能有问题的请求到达核心大语言模型（LLM）之前控制智能体的行为。

ADK 提供了回调——允许你在智能体执行生命周期的特定点进行钩入的函数。before_model_callback 对于输入安全特别有用。

什么是 before_model_callback？

它是你定义的一个 Python 函数，ADK 会在智能体将其编译的请求（包括对话历史、指令和最新用户消息）发送到底层 LLM 之前执行它。
目的： 检查请求，必要时修改它，或根据预定义规则完全阻止它。

常见用例：

输入验证/过滤： 检查用户输入是否满足条件或是否包含不允许的内容（如 PII 或关键词）。
安全防护： 防止有害的、偏离主题的或违反策略的请求被 LLM 处理。
动态提示修改： 在发送前及时向 LLM 请求上下文中添加信息（例如来自会话状态的信息）。

工作原理：

定义一个接受 callback_context: CallbackContext 和 llm_request: LlmRequest 的函数。
- callback_context：提供对智能体信息、会话状态（callback_context.state）等的访问。
- llm_request：包含准备发送给 LLM 的完整载荷（contents、config）。
在函数内部：
- 检查： 检查 llm_request.contents（特别是最后一条用户消息）。
- 修改（谨慎使用）： 你可以更改 llm_request 的部分内容。
- 阻止（安全防护）： 返回一个 LlmResponse 对象。ADK 会立即发送此响应，跳过该轮的 LLM 调用。
- 允许： 返回 None。ADK 将继续使用（可能已修改的）请求调用 LLM。

在本步骤中，我们将：

定义一个 before_model_callback 函数（block_keyword_guardrail），检查用户输入中是否包含特定关键词（"BLOCK"）。
更新我们的状态感知根智能体（第 4 步中的 weather_agent_v4_stateful）以使用此回调。
创建一个与该更新后智能体关联的新 Runner，但使用相同的状态感知会话服务以保持状态连续性。
通过发送正常请求和包含关键词的请求来测试安全防护。

1. 定义安全防护回调函数

此函数将检查 llm_request 内容中的最后一条用户消息。如果发现 "BLOCK"（不区分大小写），它将构建并返回一个 LlmResponse 以阻止流程；否则返回 None。

# @title 1. 定义 before_model_callback 安全防护

# 确保必要导入可用
from google.adk.agents.callback_context import CallbackContext
from google.adk.models.llm_request import LlmRequest
from google.adk.models.llm_response import LlmResponse
from google.genai import types # 用于创建响应内容
from typing import Optional

def block_keyword_guardrail(
    callback_context: CallbackContext, llm_request: LlmRequest
) -> Optional[LlmResponse]:
    """
    检查最新用户消息中是否包含 'BLOCK'。如果找到，则阻止 LLM 调用
    并返回预定义的 LlmResponse。否则返回 None 继续执行。
    """
    agent_name = callback_context.agent_name # 获取模型调用被拦截的智能体名称
    print(f"--- 回调：block_keyword_guardrail 正在为智能体 {agent_name} 运行 ---")

    # 从请求历史中提取最新用户消息的文本
    last_user_message_text = ""
    if llm_request.contents:
        # 查找最近一条角色为 'user' 的消息
        for content in reversed(llm_request.contents):
            if content.role == 'user' and content.parts:
                # 为简化起见，假设文本在第一部分
                if content.parts[0].text:
                    last_user_message_text = content.parts[0].text
                    break # 找到最新用户消息文本

    print(f"--- 回调：正在检查最新用户消息: '{last_user_message_text[:100]}...' ---") # 记录前 100 个字符

    # --- 安全防护逻辑 ---
    keyword_to_block = "BLOCK"
    if keyword_to_block in last_user_message_text.upper(): # 不区分大小写检查
        print(f"--- 回调：找到 '{keyword_to_block}'。阻止 LLM 调用！ ---")
        # 可选：在状态中设置标志以记录阻止事件
        callback_context.state["guardrail_block_keyword_triggered"] = True
        print(f"--- 回调：已设置状态 'guardrail_block_keyword_triggered': True ---")

        # 构建并返回 LlmResponse 以停止流程，并将其发送回去
        return LlmResponse(
            content=types.Content(
                role="model", # 从智能体的角度模拟响应
                parts=[types.Part(text=f"我无法处理此请求，因为它包含被阻止的关键词 '{keyword_to_block}'。")],
            )
            # 注意：如果需要，你也可以在此处设置 error_message 字段
        )
    else:
        # 未找到关键词，允许请求继续发送到 LLM
        print(f"--- 回调：未找到关键词。允许为 {agent_name} 调用 LLM。 ---")
        return None # 返回 None 表示 ADK 继续正常执行

print("✅ block_keyword_guardrail 函数已定义。")

2. 更新根智能体以使用回调

我们重新定义根智能体，添加 before_model_callback 参数并指向我们的新安全防护函数。为清晰起见，我们将赋予它一个新的版本名称。

重要： 如果子智能体（greeting_agent、farewell_agent）和状态感知工具（get_weather_stateful）在此上下文中尚未从前面的步骤中可用，我们需要在此处重新定义它们，确保根智能体定义可以访问其所有组件。

# @title 2. 使用 before_model_callback 更新根智能体


# --- 重新定义子智能体（确保它们存在于此上下文中） ---
greeting_agent = None
try:
    # 使用已定义的模型常量
    greeting_agent = Agent(
        model=MODEL_GEMINI_2_5_FLASH,
        name="greeting_agent", # 保持原始名称以保持一致性
        instruction="你是问候智能体。你的唯一任务是使用 'say_hello' 工具提供友好的问候。不要做其他任何事情。",
        description="使用 'say_hello' 工具处理简单的问候和打招呼。",
        tools=[say_hello],
    )
    print(f"✅ 子智能体 '{greeting_agent.name}' 已重新定义。")
except Exception as e:
    print(f"❌ 无法重新定义问候智能体。请检查模型/API 密钥 ({greeting_agent.model})。错误: {e}")

farewell_agent = None
try:
    # 使用已定义的模型常量
    farewell_agent = Agent(
        model=MODEL_GEMINI_2_5_FLASH,
        name="farewell_agent", # 保持原始名称
        instruction="你是告别智能体。你的唯一任务是使用 'say_goodbye' 工具提供礼貌的告别信息。不要执行任何其他操作。",
        description="使用 'say_goodbye' 工具处理简单的告别和再见。",
        tools=[say_goodbye],
    )
    print(f"✅ 子智能体 '{farewell_agent.name}' 已重新定义。")
except Exception as e:
    print(f"❌ 无法重新定义告别智能体。请检查模型/API 密钥 ({farewell_agent.model})。错误: {e}")


# --- 定义带有回调的根智能体 ---
root_agent_model_guardrail = None
runner_root_model_guardrail = None

# 在继续之前检查所有组件
if greeting_agent and farewell_agent and 'get_weather_stateful' in globals() and 'block_keyword_guardrail' in globals():

    # 使用已定义的模型常量
    root_agent_model = MODEL_GEMINI_2_5_FLASH

    root_agent_model_guardrail = Agent(
        name="weather_agent_v5_model_guardrail", # 新版本名称以清晰区分
        model=root_agent_model,
        description="主智能体：处理天气、委托问候/告别、包含输入关键词安全防护。",
        instruction="你是主天气智能体。使用 'get_weather_stateful' 提供天气信息。"
                    "将简单问候委托给 'greeting_agent'，将告别委托给 'farewell_agent'。"
                    "只处理天气请求、问候和告别。",
        tools=[get_weather_stateful],
        sub_agents=[greeting_agent, farewell_agent], # 引用重新定义的子智能体
        output_key="last_weather_report", # 保留第 4 步的 output_key
        before_model_callback=block_keyword_guardrail # <<< 分配安全防护回调
    )
    print(f"✅ 根智能体 '{root_agent_model_guardrail.name}' 已使用 before_model_callback 创建。")

    # --- 为此智能体创建 Runner，使用相同的状态感知会话服务 ---
    # 确保 session_service_stateful 存在于第 4 步中
    if 'session_service_stateful' in globals():
        runner_root_model_guardrail = Runner(
            agent=root_agent_model_guardrail,
            app_name=APP_NAME, # 使用一致的 APP_NAME
            session_service=session_service_stateful # <<< 使用第 4 步的服务
        )
        print(f"✅ 已为安全防护智能体 '{runner_root_model_guardrail.agent.name}' 创建 Runner，使用状态感知会话服务。")
    else:
        print("❌ 无法创建 Runner。缺少第 4 步的 'session_service_stateful'。")

else:
    print("❌ 无法创建带有模型安全防护的根智能体。一个或多个前提条件缺失或初始化失败：")
    if not greeting_agent: print("   - 问候智能体")
    if not farewell_agent: print("   - 告别智能体")
    if 'get_weather_stateful' not in globals(): print("   - 'get_weather_stateful' 工具")
    if 'block_keyword_guardrail' not in globals(): print("   - 'block_keyword_guardrail' 回调")

3. 交互以测试安全防护

让我们测试安全防护的行为。我们将使用与第 4 步中相同的会话（SESSION_ID_STATEFUL）来证明状态在这些更改之间保持持久。

发送一个正常的天气请求（应通过安全防护并执行）。
发送一个包含 "BLOCK" 的请求（应被回调拦截）。
发送一个问候（应通过根智能体的安全防护，被委托，并正常执行）。

# @title 3. 交互以测试模型输入安全防护
import asyncio # 确保已导入 asyncio

# 确保安全防护智能体的 Runner 可用
if 'runner_root_model_guardrail' in globals() and runner_root_model_guardrail:
    # 定义主异步函数用于安全防护测试对话。
    # 该函数内部的 'await' 关键字对于异步操作是必需的。
    async def run_guardrail_test_conversation():
        print("\n--- 测试模型输入安全防护 ---")

        # 使用带有回调的智能体的 Runner 和现有的状态感知会话 ID
        # 定义辅助 lambda 以使交互调用更简洁
        interaction_func = lambda query: call_agent_async(query,
                                                         runner_root_model_guardrail,
                                                         USER_ID_STATEFUL, # 使用现有用户 ID
                                                         SESSION_ID_STATEFUL # 使用现有会话 ID
                                                        )
        # 1. 正常请求（回调允许，应使用上次状态更改后的华氏度）
        print("--- 第 1 轮：请求伦敦天气（预期允许，华氏度） ---")
        await interaction_func("What is the weather in London?")

        # 2. 包含被阻止关键词的请求（回调拦截）
        print("\n--- 第 2 轮：包含被阻止关键词的请求（预期被阻止） ---")
        await interaction_func("BLOCK the request for weather in Tokyo") # 回调应捕获 "BLOCK"

        # 3. 正常问候（回调允许根智能体，委托正常执行）
        print("\n--- 第 3 轮：发送问候（预期允许） ---")
        await interaction_func("Hello again")

    # --- 执行 `run_guardrail_test_conversation` 异步函数 ---
    # 根据你的环境选择以下方法之一。

    # 方法 1：直接 await（笔记本/异步 REPL 的默认方式）
    # 如果你的环境支持顶层 await（如 Colab/Jupyter 笔记本），
    # 说明事件循环已在运行，你可以直接 await 该函数。
    print("正在尝试使用 'await' 执行（笔记本默认方式）...")
    await run_guardrail_test_conversation()

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

    # --- 对话后检查最终会话状态 ---
    # 该代码块在任一执行方法完成后运行。
    # 可选：检查由回调设置的触发标志
    print("\n--- 检查最终会话状态（安全防护测试后） ---")
    # 使用与此状态感知会话关联的会话服务实例
    final_session = await session_service_stateful.get_session(app_name=APP_NAME,
                                                         user_id=USER_ID_STATEFUL,
                                                         session_id=SESSION_ID_STATEFUL)
    if final_session:
        # 使用 .get() 更安全地访问
        print(f"安全防护触发标志: {final_session.state.get('guardrail_block_keyword_triggered', 'Not Set (or False)')}")
        print(f"最后天气报告: {final_session.state.get('last_weather_report', 'Not Set')}") # 如果成功应为伦敦天气
        print(f"温度单位: {final_session.state.get('user_preference_temperature_unit', 'Not Set')}") # 应为华氏度
        # print(f"完整状态字典: {final_session.state}") # 用于详细视图
    else:
        print("\n❌ 错误：无法检索最终会话状态。")

else:
    print("\n⚠️ 跳过模型安全防护测试。Runner（'runner_root_model_guardrail'）不可用。")

观察执行流程：

伦敦天气： 回调为 weather_agent_v5_model_guardrail 运行，检查消息，打印 "未找到关键词。允许调用 LLM。"，并返回 None。智能体继续执行，调用 get_weather_stateful 工具（该工具使用第 4 步状态更改中的 "Fahrenheit" 偏好），并返回天气。此响应通过 output_key 更新 last_weather_report。
BLOCK 请求： 回调再次为 weather_agent_v5_model_guardrail 运行，检查消息，找到 "BLOCK"，打印 "阻止 LLM 调用！"，设置状态标志，并返回预定义的 LlmResponse。智能体的底层 LLM 在此轮从未被调用。用户看到的是回调的阻止消息。
再次问候： 回调为 weather_agent_v5_model_guardrail 运行，允许请求。根智能体随后委托给 greeting_agent。注意：定义在根智能体上的 before_model_callback 不会自动应用于子智能体。 greeting_agent 正常继续，调用其 say_hello 工具，并返回问候。

你已经成功实现了输入安全层！before_model_callback 提供了一个强大的机制，可以在昂贵或有风险的 LLM 调用之前强制执行规则和控制智能体行为。接下来，我们将应用类似的概念来添加围绕工具使用本身的安全防护。

第 6 步：添加安全防护——工具参数安全防护（`before_tool_callback`）¶

在第 5 步中，我们添加了一个安全防护来检查和潜在阻止用户输入在它到达 LLM 之前。现在，我们将在 LLM 决定使用工具之后但该工具实际执行之前添加另一层控制。这对于验证 LLM 想要传递给工具的参数非常有用。

ADK 为此提供了 before_tool_callback。

什么是 before_tool_callback？

它是在特定工具函数运行之前执行的 Python 函数，在 LLM 请求使用该工具并决定参数之后执行。
目的： 验证工具参数、根据特定输入阻止工具执行、动态修改参数或强制执行资源使用策略。

常见用例：

参数验证： 检查 LLM 提供的参数是否有效、在允许范围内或符合预期格式。
资源保护： 防止工具被以可能昂贵、访问受限数据或导致不必要副作用的输入调用（例如，阻止某些参数的 API 调用）。
动态参数修改： 在工具运行之前根据会话状态或其他上下文信息调整参数。

工作原理：

定义一个接受 tool: BaseTool、args: Dict[str, Any] 和 tool_context: ToolContext 的函数。
- tool：即将被调用的工具对象（检查 tool.name）。
- args：LLM 为该工具生成的参数字典。
- tool_context：提供对会话状态（tool_context.state）、智能体信息等的访问。
在函数内部：
- 检查： 检查 tool.name 和 args 字典。
- 修改： 直接更改 args 字典中的值。如果你返回 None，工具将使用这些修改后的参数运行。
- 阻止/覆盖（安全防护）： 返回一个字典。ADK 将此字典视为工具调用的结果，完全跳过原始工具函数的执行。该字典理想情况下应匹配被阻止工具的预期返回格式。
- 允许： 返回 None。ADK 将继续使用（可能已修改的）参数执行实际工具函数。

在本步骤中，我们将：

定义一个 before_tool_callback 函数（block_paris_tool_guardrail），专门检查 get_weather_stateful 工具是否以城市 "Paris" 被调用。
如果检测到 "Paris"，回调将阻止工具并返回自定义错误字典。
更新我们的根智能体（weather_agent_v6_tool_guardrail）以同时包含 before_model_callback 和这个新的 before_tool_callback。
为此智能体创建一个新的 Runner，使用相同的状态感知会话服务。
通过请求允许城市和被阻止城市（"Paris"）的天气来测试流程。

1. 定义工具安全防护回调函数

此函数针对 get_weather_stateful 工具。它检查 city 参数。如果是 "Paris"，它返回一个看起来像工具自身错误响应的错误字典。否则，它通过返回 None 允许工具运行。

# @title 1. 定义 before_tool_callback 安全防护

# 确保必要导入可用
from google.adk.tools.base_tool import BaseTool
from google.adk.tools.tool_context import ToolContext
from typing import Optional, Dict, Any # 用于类型提示

def block_paris_tool_guardrail(
    tool: BaseTool, args: Dict[str, Any], tool_context: ToolContext
) -> Optional[Dict]:
    """
    检查是否以 'Paris' 调用 'get_weather_stateful'。
    如果是，则阻止工具执行并返回特定错误字典。
    否则，通过返回 None 允许工具调用继续执行。
    """
    tool_name = tool.name
    agent_name = tool_context.agent_name # 尝试调用工具的智能体
    print(f"--- 回调：block_paris_tool_guardrail 正在为智能体 '{agent_name}' 中的工具 '{tool_name}' 运行 ---")
    print(f"--- 回调：正在检查参数: {args} ---")

    # --- 安全防护逻辑 ---
    target_tool_name = "get_weather_stateful" # 与 FunctionTool 使用的函数名匹配
    blocked_city = "paris"

    # 检查是否是正确的工具且城市参数匹配被阻止的城市
    if tool_name == target_tool_name:
        city_argument = args.get("city", "") # 安全获取 'city' 参数
        if city_argument and city_argument.lower() == blocked_city:
            print(f"--- 回调：检测到被阻止的城市 '{city_argument}'。阻止工具执行！ ---")
            # 可选：更新状态
            tool_context.state["guardrail_tool_block_triggered"] = True
            print(f"--- 回调：已设置状态 'guardrail_tool_block_triggered': True ---")

            # 返回与工具错误预期输出格式匹配的字典
            # 此字典将成为工具的结果，跳过实际工具运行。
            return {
                "status": "error",
                "error_message": f"策略限制：工具安全防护当前禁用了对 '{city_argument.capitalize()}' 的天气查询。"
            }
        else:
             print(f"--- 回调：城市 '{city_argument}' 对工具 '{tool_name}' 是允许的。 ---")
    else:
        print(f"--- 回调：工具 '{tool_name}' 不是目标工具。允许执行。 ---")


    # 如果上面的检查没有返回字典，则允许工具执行
    print(f"--- 回调：允许工具 '{tool_name}' 继续执行。 ---")
    return None # 返回 None 允许实际工具函数运行

print("✅ block_paris_tool_guardrail 函数已定义。")

2. 更新根智能体以同时使用两个回调

我们再次重新定义根智能体（weather_agent_v6_tool_guardrail），这次在第 5 步的 before_model_callback 基础上添加 before_tool_callback 参数。

自包含执行说明： 与第 5 步类似，在定义此智能体之前，确保所有前提条件（子智能体、工具、before_model_callback）在执行上下文中已定义或可用。

# @title 2. 使用两个回调更新根智能体（自包含）

# --- 确保前提条件已定义 ---
# （包含或确保以下定义已执行：Agent, LiteLlm, Runner, ToolContext,
#  模型常量, say_hello, say_goodbye, greeting_agent, farewell_agent,
#  get_weather_stateful, block_keyword_guardrail, block_paris_tool_guardrail）

# --- 重新定义子智能体（确保它们存在于此上下文中） ---
greeting_agent = None
try:
    # 使用已定义的模型常量
    greeting_agent = Agent(
        model=MODEL_GEMINI_2_5_FLASH,
        name="greeting_agent", # 保持原始名称以保持一致性
        instruction="你是问候智能体。你的唯一任务是使用 'say_hello' 工具提供友好的问候。不要做其他任何事情。",
        description="使用 'say_hello' 工具处理简单的问候和打招呼。",
        tools=[say_hello],
    )
    print(f"✅ 子智能体 '{greeting_agent.name}' 已重新定义。")
except Exception as e:
    print(f"❌ 无法重新定义问候智能体。请检查模型/API 密钥 ({greeting_agent.model})。错误: {e}")

farewell_agent = None
try:
    # 使用已定义的模型常量
    farewell_agent = Agent(
        model=MODEL_GEMINI_2_5_FLASH,
        name="farewell_agent", # 保持原始名称
        instruction="你是告别智能体。你的唯一任务是使用 'say_goodbye' 工具提供礼貌的告别信息。不要执行任何其他操作。",
        description="使用 'say_goodbye' 工具处理简单的告别和再见。",
        tools=[say_goodbye],
    )
    print(f"✅ 子智能体 '{farewell_agent.name}' 已重新定义。")
except Exception as e:
    print(f"❌ 无法重新定义告别智能体。请检查模型/API 密钥 ({farewell_agent.model})。错误: {e}")

# --- 定义带有两个回调的根智能体 ---
root_agent_tool_guardrail = None
runner_root_tool_guardrail = None

if ('greeting_agent' in globals() and greeting_agent and
    'farewell_agent' in globals() and farewell_agent and
    'get_weather_stateful' in globals() and
    'block_keyword_guardrail' in globals() and
    'block_paris_tool_guardrail' in globals()):

    root_agent_model = MODEL_GEMINI_2_5_FLASH

    root_agent_tool_guardrail = Agent(
        name="weather_agent_v6_tool_guardrail", # 新版本名称
        model=root_agent_model,
        description="主智能体：处理天气、委托、包含输入和工具安全防护。",
        instruction="你是主天气智能体。使用 'get_weather_stateful' 提供天气信息。"
                    "将问候委托给 'greeting_agent'，将告别委托给 'farewell_agent'。"
                    "只处理天气、问候和告别。",
        tools=[get_weather_stateful],
        sub_agents=[greeting_agent, farewell_agent],
        output_key="last_weather_report",
        before_model_callback=block_keyword_guardrail, # 保留模型安全防护
        before_tool_callback=block_paris_tool_guardrail # <<< 添加工具安全防护
    )
    print(f"✅ 根智能体 '{root_agent_tool_guardrail.name}' 已使用两个回调创建。")

    # --- 创建 Runner，使用相同的状态感知会话服务 ---
    if 'session_service_stateful' in globals():
        runner_root_tool_guardrail = Runner(
            agent=root_agent_tool_guardrail,
            app_name=APP_NAME,
            session_service=session_service_stateful # <<< 使用第 4/5 步的服务
        )
        print(f"✅ 已为工具安全防护智能体 '{runner_root_tool_guardrail.agent.name}' 创建 Runner，使用状态感知会话服务。")
    else:
        print("❌ 无法创建 Runner。缺少第 4/5 步的 'session_service_stateful'。")

else:
    print("❌ 无法创建带有工具安全防护的根智能体。缺少前提条件。")

3. 交互以测试工具安全防护

让我们测试交互流程，再次使用之前步骤中相同的有状态会话（SESSION_ID_STATEFUL）。

请求 "New York" 的天气：通过两个回调，工具执行（使用状态中的华氏度偏好）。
请求 "Paris" 的天气：通过 before_model_callback。LLM 决定调用 get_weather_stateful(city='Paris')。before_tool_callback 拦截，阻止工具执行，并返回错误字典。智能体传递此错误。
请求 "London" 的天气：通过两个回调，工具正常执行。

# @title 3. 交互以测试工具参数安全防护
import asyncio # 确保已导入 asyncio

# 确保工具安全防护智能体的 Runner 可用
if 'runner_root_tool_guardrail' in globals() and runner_root_tool_guardrail:
    # 定义主异步函数用于工具安全防护测试对话。
    # 该函数内部的 'await' 关键字对于异步操作是必需的。
    async def run_tool_guardrail_test():
        print("\n--- 测试工具参数安全防护（'Paris' 被阻止） ---")

        # 使用带有两个回调的智能体的 Runner 和现有的有状态会话
        # 定义辅助 lambda 以使交互调用更简洁
        interaction_func = lambda query: call_agent_async(query,
                                                         runner_root_tool_guardrail,
                                                         USER_ID_STATEFUL, # 使用现有用户 ID
                                                         SESSION_ID_STATEFUL # 使用现有会话 ID
                                                        )
        # 1. 允许的城市（应通过两个回调，使用华氏度状态）
        print("--- 第 1 轮：请求纽约天气（预期允许） ---")
        await interaction_func("What's the weather in New York?")

        # 2. 被阻止的城市（应通过模型回调，但被工具回调阻止）
        print("\n--- 第 2 轮：请求巴黎天气（预期被工具安全防护阻止） ---")
        await interaction_func("How about Paris?") # 工具回调应拦截此请求

        # 3. 另一个允许的城市（应再次正常工作）
        print("\n--- 第 3 轮：请求伦敦天气（预期允许） ---")
        await interaction_func("Tell me the weather in London.")

    # --- 执行 `run_tool_guardrail_test` 异步函数 ---
    # 根据你的环境选择以下方法之一。

    # 方法 1：直接 await（笔记本/异步 REPL 的默认方式）
    # 如果你的环境支持顶层 await（如 Colab/Jupyter 笔记本），
    # 说明事件循环已在运行，你可以直接 await 该函数。
    print("正在尝试使用 'await' 执行（笔记本默认方式）...")
    await run_tool_guardrail_test()

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

    # --- 对话后检查最终会话状态 ---
    # 该代码块在任一执行方法完成后运行。
    # 可选：检查工具阻止触发标志
    print("\n--- 检查最终会话状态（工具安全防护测试后） ---")
    # 使用与此有状态会话关联的会话服务实例
    final_session = await session_service_stateful.get_session(app_name=APP_NAME,
                                                         user_id=USER_ID_STATEFUL,
                                                         session_id= SESSION_ID_STATEFUL)
    if final_session:
        # 使用 .get() 更安全地访问
        print(f"工具安全防护触发标志: {final_session.state.get('guardrail_tool_block_triggered', 'Not Set (or False)')}")
        print(f"最后天气报告: {final_session.state.get('last_weather_report', 'Not Set')}") # 如果成功应为伦敦天气
        print(f"温度单位: {final_session.state.get('user_preference_temperature_unit', 'Not Set')}") # 应为华氏度
        # print(f"完整状态字典: {final_session.state}") # 用于详细视图
    else:
        print("\n❌ 错误：无法检索最终会话状态。")

else:
    print("\n⚠️ 跳过工具安全防护测试。Runner（'runner_root_tool_guardrail'）不可用。")

分析输出：

纽约： before_model_callback 允许请求。LLM 请求 get_weather_stateful。before_tool_callback 运行，检查参数（{'city': 'New York'}），看到不是 "Paris"，打印 "允许工具..." 并返回 None。实际的 get_weather_stateful 函数执行，从状态中读取 "Fahrenheit"，并返回天气报告。智能体传递此报告，通过 output_key 保存。
巴黎： before_model_callback 允许请求。LLM 请求 get_weather_stateful(city='Paris')。before_tool_callback 运行，检查参数，检测到 "Paris"，打印 "阻止工具执行！"，设置状态标志，并返回错误字典 {'status': 'error', 'error_message': '策略限制...'}。实际的 get_weather_stateful 函数从未被执行。智能体接收到错误字典，就好像它是工具的输出，并根据该错误消息构建响应。
伦敦： 行为与纽约类似，通过两个回调并成功执行工具。新的伦敦天气报告覆盖了状态中的 last_weather_report。

你现在已经添加了一个关键的安全层，不仅控制了什么能到达 LLM，还控制了智能体的工具如何基于 LLM 生成的特定参数被使用。before_model_callback 和 before_tool_callback 这样的回调对于构建稳健、安全且符合策略的智能体应用至关重要。

总结：你的智能体团队已准备就绪！¶

恭喜！你已经成功地从构建一个单一的基础天气智能体，到使用智能体开发工具包（ADK）构建了一个复杂的多智能体团队。

让我们回顾一下你所取得的成就：

你从一个配备单一工具（get_weather）的基础智能体开始。
你使用 LiteLLM 探索了 ADK 的多模型灵活性，使用 Gemini、GPT-4o 和 Claude 等不同的 LLM 运行相同的核心逻辑。
你通过创建专门的子智能体（greeting_agent、farewell_agent）并从根智能体启用自动委托，拥抱了模块化设计。
你使用会话状态赋予了智能体记忆能力，使它们能够记住用户偏好（temperature_unit）和过去的交互（output_key）。
你使用 before_model_callback（阻止特定输入关键词）和 before_tool_callback（基于参数如城市 "Paris" 阻止工具执行）实现了关键的安全防护。

通过构建这个渐进式的天气机器人团队，你获得了开发复杂智能型应用所必需的 ADK 核心概念的实践经验。

关键要点：

智能体与工具：定义能力和推理的基本构建模块。清晰的指令和文档字符串至关重要。
Runner 与会话服务：编排智能体执行和维护对话上下文的引擎与记忆管理系统。
委托：设计多智能体团队可以实现专业化、模块化以及更好地管理复杂任务。智能体的 description 是自动流程的关键。
会话状态（ToolContext、output_key）：对于创建上下文感知、个性化和多轮对话的智能体至关重要。
回调（before_model、before_tool）：在关键操作（LLM 调用或工具执行）之前实现安全、验证、策略执行和动态修改的强大钩子。
灵活性（LiteLlm）：ADK 使你能够选择最适合任务的 LLM，在性能、成本和功能之间取得平衡。

接下来去哪里？

你的天气机器人团队是一个很好的起点。以下是一些进一步探索 ADK 和增强应用的思路：

真实天气 API：将 get_weather 工具中的 mock_weather_db 替换为调用真实天气 API（如 OpenWeatherMap、WeatherAPI）。
更复杂的状态：在会话状态中存储更多用户偏好（如首选位置、通知设置）或对话摘要。
优化委托：尝试不同的根智能体指令或子智能体描述，以微调委托逻辑。你能添加一个 "forecast"（预报）智能体吗？
高级回调：
- 使用 after_model_callback 在 LLM 生成响应之后重新格式化或净化其输出。
- 使用 after_tool_callback 处理或记录工具返回的结果。
- 实现 before_agent_callback 或 after_agent_callback 用于智能体级别的进入/退出逻辑。
错误处理：改善智能体处理工具错误或意外 API 响应的方式。也许可以在工具中添加重试逻辑。
持久化会话存储：探索 InMemorySessionService 的替代方案，用于持久化存储会话状态（如使用 Firestore 或 Cloud SQL 等数据库——需要自定义实现或未来的 ADK 集成）。
流式 UI：将你的智能体团队与 Web 框架（如 FastAPI，如 ADK 流式快速入门中所示）集成，以创建实时聊天界面。

智能体开发工具包为构建复杂的 LLM 驱动应用提供了坚实的基础。通过掌握本教程中涵盖的概念——工具、状态、委托和回调——你已经具备了应对日益复杂的智能体系统的能力。

祝你构建愉快！

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

第 1 步：你的第一个智能体 —— 基础天气查询¶

第 2 步：使用 LiteLLM 支持多模型 [可选]¶

第 3 步：构建智能体团队——问候与告别的委托处理¶

第 4 步：使用会话状态添加记忆和个性化¶

第 5 步：添加安全防护——使用 before_model_callback 的输入安全防护¶

第 6 步：添加安全防护——工具参数安全防护（before_tool_callback）¶

总结：你的智能体团队已准备就绪！¶

第 5 步：添加安全防护——使用 `before_model_callback` 的输入安全防护¶

第 6 步：添加安全防护——工具参数安全防护（`before_tool_callback`）¶