adk-skill

Google Agent Development Kit（ADK）指南

概述

ADK 是 Google 推出的开源框架，用于构建与模型无关的 AI Agent。本指南面向 Python 2.13.0。其 Workflow Runtime 将 Agent、工具和函数组合成图，并支持记忆、工件、回调和部署。使用示例前，请先确认目标 SDK 的版本化 API；组合片段中命名的应用 Agent/工具必须由应用提供。

关键规则

1. 入口点是 root_agent — 在 agent.py 中定义的模块级变量，或用于基于图的 Agent 的 Workflow 对象。
2. 基于代码的包需要 __init__.py 导入 Agent 模块：from . import agent。仅含 YAML 的 Agent 不需要。
3. 为模型调用配置凭据。 仅运行确定性函数的工作流无需凭据。
4. 在 ADK 2.0 中，Agent 是节点。 不要重写 _run_async_impl() — 请改用 BeforeAgentCallback / AfterAgentCallback。
5. 不要直接向 context.session.events 追加事件。从节点生成事件，由框架负责持久化和路由。
6. 切勿在工具中捕获 BaseException — 这会破坏 HITL 暂停/恢复和自动重试。
7. 按职责对工具分组。 若拆分无关领域有助于路由，就应拆分；工具数量本身并不意味着需要更多 Agent。
8. 使用 output_key + output_schema 确保 Agent 间的数据可靠流转 — 不要使用自由文本。

文档与资源

• ADK 文档：https://adk.dev/llms.txt
• 官方示例：https://github.com/google/adk-samples
• 2.0 迁移指南：https://adk.dev/2.0/

支持的语言

语言	包	安装	ADK 2.x
Python	google-adk	pip install google-adk==2.13.0	2.13.0
Go	google.golang.org/adk/v2	go get google.golang.org/adk/[email protected]	2.6.0
Java	com.google.adk:google-adk	Maven/Gradle	1.13.0；图功能为实验性
Kotlin	com.google.adk:google-adk-kotlin-core	Gradle	1.4.0；核心功能稳定，图功能为实验性
TypeScript	@google/adk	npm install @google/[email protected]	2.3.0

Go 和 TypeScript 还发布了并行的 1.x LTS 版本线（google.golang.org/adk / @google/adk@^1）。

本指南提供 Python 示例。Java、Go、Kotlin 和 TypeScript 的模式请参阅 references/multi-language.md。

先决条件

Python 3.11+，并运行 pip install "google-adk==2.13.0"；模型调用需要 GOOGLE_API_KEY 或 Google Cloud 身份验证。其他语言：Java 17+、Go 1.26.6+（ADK v2.6）、Node 24+。按需安装额外依赖：[db]、[a2a]、[openai]、[mongodb]、[eval]、[bigquery-analytics]；Jinja2 为可选依赖。

快速参考

任务	ADK 2.x 方案
单个 Agent	使用包含模型、指令和工具的 LlmAgent
基于图的工作流	使用包含 Agent/Function 节点的 Workflow(edges=[...])
动态（代码驱动）工作流	使用 @node 装饰器和 ctx.run_node() 循环/分支
协作团队	使用包含 sub_agents 的 Agent，并设置 mode='single_turn'/'task'/'chat'
顺序流水线	使用 SequentialAgent 或 Workflow 链
并行执行	使用 ParallelAgent 或 asyncio.gather(ctx.run_node(...))
迭代优化	使用 LoopAgent 或动态工作流中的 while 循环
Agent 作为工具	使用 AgentTool 按需委派
支持暂停/恢复的内联专家	使用 mode="single_turn" 的 sub_agents
备用模型 / 顾问	实验性的 FallbackModel / ModelConsultTool
Google 托管的 Agent	ManagedAgent（预览版，2.4+）
Agent 作为 MCP 服务	agent.to_mcp_server()（2.5+）
远程 Agent（A2A）	RemoteA2aAgent + to_a2a()
自定义工具	带类型提示和文档字符串的 Python 函数
OpenAPI 工具	根据 OpenAPI 规范创建的 OpenAPIToolset
结构化输出	通过 output_schema + output_key 使用 Pydantic 模型
状态管理	callback_context.state 和 tool_context.state
跨会话记忆	MemoryService + PreloadMemoryTool / LoadMemory
二进制数据持久化	ArtifactService（save_artifact、load_artifact）
MCP 集成	MCPToolset；SDK 2.x 的现代 wire 路径需选择启用
YAML Agent 配置 / 会话回退	adk create --type=config / runner.rewind_async()
流式处理（Live API）	使用 LiveRequestQueue 进行双向音频/视频传输
测试	使用 InMemoryRunner 的 pytest
评估	使用 .test.json 的 EvalSet、adk eval CLI

项目结构

基于代码的项目：

my_agent/
├── my_agent/
│   ├── __init__.py          # Must import agent module
│   ├── agent.py             # Defines root_agent (entry point)
│   ├── prompts.py           # Instruction strings (optional)
│   ├── tools.py             # Custom tool functions (optional)
│   └── sub_agents/          # Sub-agent packages (optional)
├── tests/test_agent.py
├── pyproject.toml
└── .env

基于 YAML 的项目： adk create --type=config my_agent 会生成 my_agent/root_agent.yaml + .env。

重要：init.py 和 root_agent

# my_agent/__init__.py
from . import agent

在 ADK 2.0 中，root_agent 可以是 LlmAgent 或 Workflow 对象：

# my_agent/agent.py
from google.adk.agents import LlmAgent

root_agent = LlmAgent(
    name="my_agent",
    # model is optional — built-in default: gemini-3.5-flash
    description="Brief description for agent discovery",
    instruction="Detailed system prompt...",
    tools=[],
)

工作流类型

ADK 2.0 提供三种组合多 Agent 工作的方法。1.x 中的模板工作流仍可使用，但已被取代。

1. 基于图的工作流

由节点（Agent、Function、Tool）构成的声明式图，并使用显式路由边。最适用于确定性、结构化流程。

from google.adk import Workflow, Event
from google.adk.agents import LlmAgent

classifier = LlmAgent(
    name="classifier", model="gemini-flash-latest",
    instruction="Classify input as 'BUG', 'SUPPORT', or 'LOGISTICS'.",
    output_schema=str,
)

def router(node_input: str):
    routes = [r.strip() for r in node_input.split(",")]
    return Event(route=routes)

def handle_bug():    return Event(message="Handling bug...")
def handle_support(): return Event(message="Handling support...")
def handle_logistics(): return Event(message="Handling logistics...")

root_agent = Workflow(
    name="routing_workflow",
    edges=[
        ("START", classifier, router),
        (router, {"BUG": handle_bug, "SUPPORT": handle_support, "LOGISTICS": handle_logistics}),
    ],
)

2. 动态工作流

通过 @node 装饰器和 ctx.run_node() 进行程序化编排。最适用于复杂分支或迭代逻辑。

from google.adk import Context, Workflow
from google.adk.workflow import node

@node(name="process_item")
def process(node_input: str) -> str:
    return f"Processed: {node_input}"

@node(rerun_on_resume=True)
async def orchestrator(ctx: Context, node_input: list[str]) -> str:
    results = []
    for item in node_input:
        result = await ctx.run_node(process, item)
        results.append(result)
    return "\n".join(results)

root_agent = Workflow(
    name="dynamic_workflow",
    edges=[("START", orchestrator)],
)

对于 HITL：从节点中产出 RequestInput。请参阅 references/workflows.md。

3. 协作 Agent

协调器将任务委派给子 Agent，并指定运行模式：

from google.adk.agents import LlmAgent

weather_agent = LlmAgent(
    name="weather_checker", mode="single_turn", tools=[get_weather],
)
flight_agent = LlmAgent(
    name="flight_booker", mode="task",
    input_schema=FlightInput, output_schema=FlightResult,
    tools=[search_flights, book_flight],
)
root_agent = LlmAgent(
    name="travel_planner",  # Coordinator — auto-injects request_task_* tools
    sub_agents=[weather_agent, flight_agent],
)

协作模式

模式：chat（持续对话，直到移交）、task（通过 finish_task 自动返回，也可能请求澄清）、single_turn（不与用户交互，直接返回结果）。完整表格请参阅：references/workflows.md。

在子 Agent 上设置模式以控制委派。任务模式包装器从 2.12+ 起支持嵌套委派和恢复。根 Agent 是对话的入口点。

4. 模板工作流（兼容 1.x）

SequentialAgent、ParallelAgent 和 LoopAgent 仍可使用，但已被取代。请参阅 references/advanced-patterns.md。

工具

函数工具

任何带有类型提示和文档字符串的 Python 函数都会成为工具：

def get_weather(city: str, units: str = "celsius") -> dict:
    """Get current weather for a city.

    Args:
        city: The city name to look up weather for.
        units: Temperature units - 'celsius' or 'fahrenheit'.
    """
    return {"temperature": 22, "conditions": "sunny", "humidity": 45}

agent = LlmAgent(
    name="weather_agent", model="gemini-flash-latest",
    instruction="Help users check the weather.", tools=[get_weather],
)

要求： 所有参数都必须有类型提示；文档字符串必须包含说明和 Args；必须标注返回类型。

支持状态访问的工具

from google.adk.tools import ToolContext

def add_to_cart(item: str, quantity: int, tool_context: ToolContext) -> dict:
    """Add an item to the shopping cart."""
    cart = tool_context.state.get("cart", [])
    cart.append({"item": item, "quantity": quantity})
    tool_context.state["cart"] = cart
    return {"status": "added", "cart_size": len(cart)}

AgentTool

from google.adk.tools.agent_tool import AgentTool

specialist = LlmAgent(
    name="code_reviewer", model="gemini-flash-latest",
    instruction="Review code for bugs and best practices.",
)
coordinator = LlmAgent(
    name="coordinator", model="gemini-flash-latest",
    instruction="Coordinate tasks. Use code_reviewer for code reviews.",
    tools=[AgentTool(agent=specialist)],
)

内置工具、OpenAPI、MCP

from google.adk.tools import google_search, load_memory
from google.adk.tools.openapi_tool.openapi_spec_parser.openapi_toolset import OpenAPIToolset
from google.adk.tools.mcp_tool import MCPToolset, StdioConnectionParams
from mcp import StdioServerParameters

# Built-in
agent = LlmAgent(name="researcher", tools=[google_search, load_memory])

# OpenAPI (generate tools from REST API spec)
toolset = OpenAPIToolset(spec_str=openapi_spec_json, spec_str_type="json")

# MCP
mcp_toolset = MCPToolset(
    connection_params=StdioConnectionParams(
        server_params=StdioServerParameters(command="npx", args=["-y", "some-mcp-server"]),
    ),
)

有关工具身份验证、确认、MCP、稳定版 SkillToolset、实验性的技能生命周期、脚本制品和顾问预算，请参阅 references/tools-reference.md。

---

记忆与制品

记忆（跨会话）

长期记忆会跨会话保留。Agent 使用内置工具检索记忆：

from google.adk.memory import InMemoryMemoryService
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools.preload_memory_tool import PreloadMemoryTool

memory_service = InMemoryMemoryService()  # Prototyping (or VertexAiMemoryBankService for prod)

agent = LlmAgent(
    name="memory_agent", model="gemini-flash-latest",
    instruction="Answer questions using past conversation history.",
    tools=[PreloadMemoryTool()],
)

runner = Runner(
    agent=agent, app_name="my_app",
    session_service=InMemorySessionService(),
    memory_service=memory_service,
)

后端包括内存、SQLite、Vertex Memory Bank/RAG 和实验性的 MongoDB。各后端支持的可选变更 API 不尽相同。请参阅 references/memory-artifacts.md。

工件（二进制数据）

带版本管理的二进制数据（文件、图像、音频）；继续使用上面的运行器示例：

from google.adk.artifacts import InMemoryArtifactService

artifact_service = InMemoryArtifactService()  # Or GcsArtifactService

runner = Runner(
    agent=agent, app_name="my_app",
    session_service=InMemorySessionService(),
    artifact_service=artifact_service,
)
# In async tools: await tool_context.save_artifact(filename, part)
#                 await tool_context.load_artifact(filename)
#                 await tool_context.list_artifacts()

请参阅 references/memory-artifacts.md。

---

状态管理

状态是 Agent、工具和回调之间共享的字典。作用域：state["key"]（会话级）、app:key（应用级）、user:key（用户级）。通过 output_key + output_schema 在 Agent 之间传递数据。指令支持 {var} / {prefix:var} 占位符；选择启用 Jinja2 模板（use_jinja2=True，2.7+，可选依赖）后，可使用条件语句、循环和点号访问：

from google.adk.agents import LlmAgent, SequentialAgent
from pydantic import BaseModel

class ResearchOutput(BaseModel):
    findings: list[str]

researcher = LlmAgent(name="researcher", output_key="findings", output_schema=ResearchOutput)
writer = LlmAgent(name="writer", instruction="Write a report based on {findings}.")
pipeline = SequentialAgent(name="pipeline", sub_agents=[researcher, writer])

---

回调

回调会拦截 Agent 生命周期。返回 None 以继续，返回值则会短路执行。在 ADK 2.0 中，请使用回调，而不是重写 _run_async_impl()。

回调	签名	用例
before_agent_callback	(CallbackContext)	初始化状态
before_tool_callback	(BaseTool, dict, ToolContext) -> dict|None	验证输入
after_tool_callback	(BaseTool, dict, ToolContext, dict) -> dict|None	后处理结果
before_model_callback	(CallbackContext, LlmRequest)	速率限制、安全

def before_tool(tool, args, tool_context) -> dict | None:
    if tool.name == "approve_discount" and args.get("value", 0) > 50:
        return {"status": "rejected", "reason": "Discount too large"}
    return None  # Proceed normally

---

结构化输出

使用 Pydantic 模型生成类型明确且经过验证的 Agent 输出：

from pydantic import BaseModel
from google.adk.agents import LlmAgent

class AnalysisResult(BaseModel):
    summary: str
    key_findings: list[str]
    confidence: float
    recommendations: list[str]

agent = LlmAgent(
    name="analyzer", model="gemini-flash-latest",
    instruction="Analyze the provided data and return structured results.",
    output_schema=AnalysisResult,
    output_key="analysis",
)

运行和测试

pip install "google-adk==2.13.0"
export GOOGLE_API_KEY="your-key"
adk run my_agent          # Interactive CLI
adk web my_agent          # Web UI (includes Visual Builder)
adk api_server my_agent   # REST API server
adk create --type=config my_agent  # YAML-based agent

测试

import pytest
from google.adk.runners import InMemoryRunner
from google.genai import types

@pytest.mark.asyncio
async def test_agent():
    runner = InMemoryRunner(agent=root_agent, app_name="test")
    session = await runner.session_service.create_session(
        user_id="test_user", app_name="test",
    )
    content = types.Content(
        role="user", parts=[types.Part.from_text(text="Hello")],
    )
    events = []
    async for event in runner.run_async(
        user_id="test_user", session_id=session.id, new_message=content,
    ):
        events.append(event)
    text = " ".join(
        part.text for event in events if event.is_final_response() and event.content
        for part in (event.content.parts or []) if part.text and not part.thought
    )
    assert "expected" in text.lower()

评估

在 .test.json 文件中定义评估用例，通过 adk eval、pytest 或 Web UI 运行。ADK 2.2+ 新增 RubricBasedMultiTurnTrajectoryEvaluator。请参阅 references/evaluation.md。

模型选择

内置默认模型仍为 gemini-3.5-flash。gemini-flash-latest 是一个滚动别名；为确保部署可复现，请选择提供商/区域支持的固定版本模型。不同提供商的采样/推理参数有所不同：请检查所选模型支持的配置。OpenAI 适配器使用 google.adk.integrations.openai 和 [openai] 扩展；其他提供商的用法请参阅高级参考文档。

设计模式

模式	适用场景	ADK 2.0 实现方式
图工作流	确定性的多步骤流程	使用带有显式 edges 的 Workflow
动态工作流	复杂分支、迭代	使用 @node + ctx.run_node() 代码
协作团队	自主管理的专家团队	使用协调器 + 带有 mode 的 sub_agents
扇出 / 扇入	独立的并行任务	使用 ParallelAgent 或 asyncio.gather
反思循环	质量优先于速度	使用 LoopAgent 或动态工作流中的 while
动态路由	根据输入选择处理程序	在 Workflow 边中使用路由函数
模型故障转移	符合条件的模型错误	使用实验性的 FallbackModel；工具需显式恢复
带护栏的 Agent	强制执行安全 / 合规要求	使用 before_model_callback + before_tool_callback
资源分级	优化成本	为不同 Agent 配置不同的 model

关键设计规则：

• 新开发优先使用基于图的工作流或动态工作流，而非模板工作流。
• 按职责对 Agent 分组；如果路由能从中受益，则拆分无关领域。
• 通过 output_key + output_schema（Pydantic）在 Agent 之间传递数据——绝不要依赖自由文本。
• 将生成与评估分开——使用不同的 Agent 进行评议。
• 编写精确的子 Agent description 字段——它们决定 Auto-Flow 路由决策。
• 在指令中嵌入推理步骤：“1. 分析 2. 规划 3. 执行 4. 验证”。
• 在图边中为不明确的输入添加备用路由。

另请参阅：references/workflows.md（图路由、HITL、动态循环） | references/advanced-patterns.md（Agent Config、部署、AG-UI） | references/design-patterns.md（链式处理、反思、规划） | references/a2a-protocol.md（A2A 服务 / 客户端） | references/memory-artifacts.md（记忆后端、资源） | references/troubleshooting.md（调试、从 1.x 迁移到 2.0）。

限制

图工作流：不支持 Live 流式传输；旧版编排器不能作为节点；仅 START 可使用聊天模式 Agent；图中的任务模式需要 2.5+ 才支持恢复。完整列表：references/workflows.md。

故障排除

有关设置、运行时、A2A、MCP、会话以及从 1.x 迁移到 2.0 的错误 → 原因 → 修复模式，请参阅 references/troubleshooting.md。