google-adk

Google Agent Development Kit (ADK)

本技能介绍如何使用 Google 的 Python 版 Agent Development Kit (ADK) 构建可用于生产环境的 AI Agent，包括 Agent 组合、工具设计、会话／状态／记忆管理、制品、评估和部署。

构建 ADK Agent 的工作流

1. 定义 Agent — 创建一个 LlmAgent（或 Agent），并设置清晰的 name、model、instruction 和 description。让指令专注于一项任务。
2. 编写工具 — 编写带类型提示和文档字符串的普通 Python 函数，或使用 FunctionTool 封装现有 API。在产生副作用前验证所有输入。
3. 组合多 Agent 系统 — 对于复杂工作流，将职责拆分给子 Agent，并使用 SequentialAgent、ParallelAgent 或 LoopAgent 进行确定性编排；也可以通过 sub_agents 委派，实现由大模型驱动的路由。
4. 连接会话和状态 — 选择一个 SessionService（开发环境使用内存服务，生产环境使用 DatabaseSessionService 或 Vertex AI 托管服务），并使用 session.state 存储会话范围内的数据。
5. 添加记忆（可选） — 当 Agent 需要记住不同对话之间的信息时，配置 MemoryService 以便跨会话回忆。
6. 处理制品（可选） — 当 Agent 生成或接收文件、图像或其他二进制输出时，配置 ArtifactService。
7. 本地运行 — 使用 adk web、adk run 或 Runner API，在 Session 中测试 Agent。
8. 评估 — 编写 .evalset.json 测试用例，并运行 adk eval，检查工具调用轨迹和响应质量是否出现回退。
9. 部署 — 为 Vertex AI Agent Engine、Cloud Run 或 GKE 打包 Agent，并分别配置开发、预发布和生产环境。

Agent 设计

• 让每个 Agent 专注于一个明确的目标、角色和工具集，避免让一个 Agent 试图包办所有事情。
• 使用 LlmAgent 实现灵活、由推理驱动的行为；使用工作流 Agent（SequentialAgent、ParallelAgent、LoopAgent）实现无需大模型决定下一步的确定性编排。
• 编写指令，明确任务边界、工具使用规则（“回答订单问题前始终调用 lookup_order”）和升级处理方式（“如果无法解决问题，则转交给 human_handoff_agent”）。
• 按职责拆分多 Agent 系统（例如“分诊”Agent、“账单”Agent、“技术支持”Agent），而不是为了实现方便而拆分。
• 将模型名称设为可配置项（环境变量或配置文件），这样无需修改代码即可在 gemini-2.0-flash、gemini-2.5-pro 或其他受支持的模型之间切换。
• 当 Agent 的最终响应必须是供其他代码使用的结构化数据时，使用 output_schema（Pydantic 模型）。

工具

• 为工具提供范围明确、类型清晰的输入和输出——一个函数只做一件事，并添加文档字符串，供大模型判断何时调用。
• 执行任何副作用（写入、外部调用、支付）前验证工具参数。
• 不要将密钥、凭据和特权 API 密钥放入 Agent 指令和提示词中；应在工具实现层注入这些信息。
• 明确处理工具错误，并返回可操作的失败消息（例如 {"status": "error", "message": "order_id not found"}），而不是抛出会中断 Agent 循环的未处理异常。
• 注意 ADK 工具的限制——某些内置工具（例如 google_search、code_execution）不能与其他工具组合用于同一个 Agent；应改为委派给专用子 Agent。
• 自定义 Python 逻辑优先使用 FunctionTool；让一个 Agent 将另一个 Agent 用作工具时使用 AgentTool；仅当内置工具（google_search、BuiltInCodeExecutor）的限制符合使用场景时才使用它们。

示例：使用工具定义 Agent

from google.adk.agents import Agent
from google.adk.tools import FunctionTool
from google.adk.runners import InMemoryRunner
from google.genai import types


def get_order_status(order_id: str) -> dict:
    """Look up the current status of a customer order.

    Args:
        order_id: The unique identifier of the order, e.g. "ord-12345".

    Returns:
        A dict with 'status' ('found' or 'error') and either 'state'
        and 'updated_at', or an error 'message'.
    """
    if not order_id or not order_id.startswith("ord-"):
        return {"status": "error", "message": "invalid order_id format"}

    # In production this would call a real order service.
    orders = {"ord-12345": {"state": "shipped", "updated_at": "2026-08-30T10:00:00Z"}}
    order = orders.get(order_id)
    if order is None:
        return {"status": "error", "message": f"no order found for {order_id}"}
    return {"status": "found", **order}


root_agent = Agent(
    name="order_support_agent",
    model="gemini-2.0-flash",
    description="Answers customer questions about order status.",
    instruction=(
        "You help customers check their order status. "
        "Always call get_order_status before answering an order question. "
        "If the tool returns an error, apologize and ask the customer to "
        "double-check their order ID. Never invent an order status."
    ),
    tools=[FunctionTool(func=get_order_status)],
)


async def main() -> None:
    runner = InMemoryRunner(agent=root_agent, app_name="order_support")
    session = await runner.session_service.create_session(
        app_name="order_support", user_id="user-1"
    )
    message = types.Content(
        role="user", parts=[types.Part(text="What's the status of ord-12345?")]
    )
    async for event in runner.run_async(
        user_id="user-1", session_id=session.id, new_message=message
    ):
        if event.content and event.content.parts:
            for part in event.content.parts:
                if part.text:
                    print(part.text)


if __name__ == "__main__":
    import asyncio

    asyncio.run(main())

会话、状态和记忆

• 使用 session.state 存储当前对话的数据，例如在对话过程中发现的用户偏好、计数器和中间结果。
• 使用 MemoryService 跨会话回忆和检索信息（即会话结束后仍应保留、并可在未来会话中搜索的事实）。
• 状态值应保持精简且可序列化（字符串、数字、小型字典/列表）——切勿直接在会话状态中存储大型文件或二进制数据。
• 状态键应保持稳定并有文档说明；使用前缀（user:、app:、temp:）在整个应用中一致地限定状态的生命周期。
• 在生产环境中，请使用 DatabaseSessionService 或托管式 Vertex AI 会话服务，而不要使用 InMemorySessionService，后者会在进程重启时丢失所有状态。

工件

• 使用工件存储生成的文件、上传的文件、报告、图像、音频及其他二进制数据——不要将它们存入会话状态。
• 在 Runner 上配置 ArtifactService（例如，开发环境使用 InMemoryArtifactService，生产环境使用 GcsArtifactService），然后再依赖 save_artifact 或 load_artifact 等工件操作。
• 有意识地为工件文件命名并设置版本（ADK 会按文件名自动递增版本），避免用同一个名称覆盖语义不同的输出。
• 如果完整内容应存放在工件中，则只在会话状态中保存引用或简短摘要——例如，将 report_v2.pdf 作为工件，并将 "generated report v2" 作为状态备注。

评估和部署

• 添加 .evalset.json 测试用例，涵盖工具调用轨迹（调用了哪些工具、调用顺序以及传入的参数）和最终回复质量。
• 在 CI 中运行 adk eval <agent_module> <eval_set_file>，以便在合并指令变更前发现提示词回归。
• 使用 ADK 的跟踪和事件日志（adk web 的跟踪视图或 Runner 事件）调试 Agent 的决策——检查调用了哪个工具、传入了哪些参数，以及模型看到了什么。
• 将本地开发、预发布环境和生产环境的配置分开（使用不同的 SessionService/ArtifactService 后端、不同的 API 密钥和不同的模型层级）。
• 添加延迟、工具故障、令牌用量和交接失败的可观测性——使用计时/日志装饰器包装工具函数，并将指标导出到现有监控系统。
• 部署到 Vertex AI Agent Engine，以使用托管式、自动扩缩的运行环境；如果需要更好地控制环境，则使用 adk deploy cloud_run / 自定义 Dockerfile，将应用容器化并部署到 Cloud Run 或 GKE。

常见错误

• 让一个 Agent 负责所有工作流，而不是拆分为职责明确的子 Agent。
• 让工具接受未经验证的任意 Shell、SQL 或 HTTP 输入，这会使工具调用成为注入向量。
• 依赖提示词文本实现访问控制——“绝不泄露其他用户的数据”等指令只是指导，并非强制措施；应在工具代码中检查权限。
• 将重要副作用（付款、删除、外部通知）隐藏在 process 或 handle 等通用工具名称后面——工具名称应明确说明其实际功能。
• 下游代码依赖结构化的最终回复时，忘记设置 output_key 或 output_schema，导致必须脆弱地解析字符串。
• 指令变更后跳过评估——改写提示词可能会在不知不觉中改变模型选择调用的工具。