夜雨聆风Google ADK 源码深度解析：官方 Agent 框架的架构哲学与工程实现
系列文章第三篇 | 三大 Agent SDK 源码对比：OpenAI Agents SDK · Anthropic Claude Code · Google ADK
一、项目概览：Google 官方的 Agent 开发套件
Google Agent Development Kit（ADK）是 Google 官方推出的 Python Agent 框架，托管于 google/adk-python 仓库。与 OpenAI Agents SDK 的极简主义和 Claude Code 的工具链哲学不同，ADK 的设计理念是为企业级生产环境提供全栈解决方案——它不仅仅是一个 Agent 调用库，而是一个完整的 Agent 生命周期管理平台。
ADK 的核心特征可以概括为：
原生 Gemini 集成：深度绑定 google.genaiSDK，利用 Gemini 模型的原生能力（函数调用、上下文缓存、实时音频等）丰富的 Agent 类型系统：提供六种 Agent 类型，是三大框架中 Agent 类型最丰富的 完整的基础设施层：内置 Session 管理、Artifact 存储、认证系统、A2A 协议支持 企业级特性：插件系统、可观测性、可恢复执行、上下文缓存优化
# ADK 的典型使用方式
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
agent = LlmAgent(
name="assistant",
model="gemini-3.5-flash",
instruction="You are a helpful assistant.",
tools=[my_tool],
)
runner = Runner(
app_name="my_app",
agent=agent,
session_service=InMemorySessionService(),
)
async for event in runner.run_async(
user_id="user1",
session_id="session1",
new_message=user_content,
):
print(event.content)
二、目录结构：模块化的工程架构
ADK 的源码目录结构体现了其"全栈框架"的定位：
google/adk/
├── agents/ # Agent 核心定义
│ ├── base_agent.py # BaseAgent 基类
│ ├── llm_agent.py # LlmAgent（核心 Agent 类型）
│ ├── loop_agent.py # LoopAgent（循环执行）
│ ├── parallel_agent.py # ParallelAgent（并行执行）
│ ├── sequential_agent.py # SequentialAgent（顺序执行）
│ ├── remote_a2a_agent.py # RemoteA2aAgent（远程 A2A Agent）
│ ├── invocation_context.py # 调用上下文
│ ├── context.py # 运行上下文
│ ├── run_config.py # 运行配置
│ └── callback_context.py # 回调上下文
├── tools/ # 工具系统
│ ├── base_tool.py # BaseTool 基类
│ ├── function_tool.py # FunctionTool（函数工具）
│ ├── base_toolset.py # BaseToolset（工具集）
│ ├── openapi_toolset.py # OpenAPI 工具集
│ ├── rest_api_tool.py # REST API 工具
│ └── tool_auth_handler.py # 工具认证处理
├── flows/ # 执行流程
│ └── llm_flows/
│ ├── base_llm_flow.py # BaseLlmFlow 基类
│ ├── single_flow.py # SingleFlow（单 Agent 流）
│ └── auto_flow.py # AutoFlow（自动流转）
├── models/ # LLM 模型抽象
│ ├── base_llm.py # BaseLlm 基类
│ ├── llm_request.py # LLM 请求
│ └── llm_response.py # LLM 响应
├── sessions/ # 会话管理
│ ├── base_session_service.py
│ └── session.py
├── artifacts/ # 制品存储
│ ├── base_artifact_service.py
│ ├── gcs_artifact_service.py
│ └── in_memory_artifact_service.py
├── auth/ # 认证系统
│ ├── auth_credential.py
│ ├── auth_handler.py
│ └── auth_tool.py
├── a2a/ # Agent-to-Agent 协议
│ ├── converters/
│ └── agent/
├── apps/ # 应用封装
│ └── app.py
├── runners.py # Runner 执行器
├── events/ # 事件系统
├── plugins/ # 插件系统
├── memory/ # 记忆服务
├── code_executors/ # 代码执行器
└── planners/ # 规划器
这种模块化的目录结构清晰地分离了关注点：agents/ 定义"是什么"，flows/ 定义"怎么跑"，tools/ 定义"能做什么"，runners.py 负责"谁来驱动"。
三、Agent 类型体系：六大 Agent 类型的分层设计
ADK 的 Agent 类型体系是三大框架中最丰富的。它采用经典的继承层次结构：
BaseNode (workflow 基类)
└── BaseAgent (抽象基类)
├── LlmAgent (LLM 驱动的 Agent，核心)
├── SequentialAgent (顺序执行)
├── ParallelAgent (并行执行)
├── LoopAgent (循环执行)
└── RemoteA2aAgent (远程 A2A Agent)
3.1 BaseAgent：万物之基
BaseAgent 是所有 Agent 的抽象基类，继承自 BaseNode（workflow 系统的节点基类）和 abc.ABC。它定义了 Agent 的核心接口：
class BaseAgent(BaseNode, abc.ABC):
"""Base class for all agents in Agent Development Kit."""
name: str
"""The agent's name. Must be a Python identifier and unique within the agent tree."""
description: str = ''
"""Description about the agent's capability.
The model uses this to determine whether to delegate control to the agent."""
parent_agent: Optional[BaseAgent] = Field(default=None, init=False, exclude=True)
"""The parent agent of this agent."""
sub_agents: list[BaseAgent] = Field(default_factory=list)
"""The sub-agents of this agent."""
before_agent_callback: Optional[BeforeAgentCallback] = None
after_agent_callback: Optional[AfterAgentCallback] = None
BaseAgent 的 run_async 方法是所有 Agent 的统一入口，它实现了模板方法模式：
async def run_async(self, parent_context: InvocationContext) -> AsyncGenerator[Event, None]:
ctx = self._create_invocation_context(parent_context)
async with _instrumentation.record_agent_invocation(ctx, self):
try:
# 1. 前置回调
if event := await self._handle_before_agent_callback(ctx):
yield event
if ctx.end_invocation:
return
# 2. 核心执行（子类实现）
async with Aclosing(self._run_async_impl(ctx)) as agen:
async for event in agen:
yield event
# 3. 后置回调
if event := await self._handle_after_agent_callback(ctx):
yield event
except Exception as e:
await self._handle_agent_error_callback(ctx, e)
raise
这个设计模式确保了所有 Agent 类型共享统一的生命周期管理：前回调 → 核心执行 → 后回调，同时支持遥测和错误处理。
3.2 LlmAgent：框架的核心引擎
LlmAgent 是 ADK 中最重要、最复杂的 Agent 类型。它是唯一一个能够调用 LLM 的 Agent，所有其他 Agent 类型本质上都是编排容器。
class LlmAgent(BaseAgent, abc.ABC):
"""LLM-based Agent."""
DEFAULT_MODEL: ClassVar[str] = 'gemini-3.5-flash'
DEFAULT_LIVE_MODEL: ClassVar[str] = 'gemini-live-2.5-flash-native-audio'
model: Union[str, BaseLlm] = ''
instruction: Union[str, InstructionProvider] = ''
tools: list[ToolUnion] = Field(default_factory=list)
generate_content_config: Optional[types.GenerateContentConfig] = None
sub_agents: list[BaseAgent] = Field(default_factory=list)
output_schema: Optional[SchemaType] = None
output_key: Optional[str] = None
planner: Optional[BasePlanner] = None
code_executor: Optional[BaseCodeExecutor] = None
LlmAgent 的字段设计体现了 Google 对 LLM Agent 的深度理解：
instruction：支持静态字符串和动态 InstructionProvider函数，后者可以根据上下文动态生成指令tools： ToolUnion类型别名统一了Callable、BaseTool、BaseToolset三种工具形态output_schema：支持 Pydantic 模型、列表类型、字典等多种 Schema 类型，实现结构化输出 planner：内置规划器支持，可以启用模型的思考能力
LlmAgent 的模型解析逻辑特别值得注意：
@property
def canonical_model(self) -> BaseLlm:
if isinstance(self.model, BaseLlm):
return self.model
elif self.model: # model is non-empty str
return LLMRegistry.new_llm(self.model)
else: # find model from ancestors.
ancestor_agent = self.parent_agent
while ancestor_agent is not None:
if isinstance(ancestor_agent, LlmAgent):
return ancestor_agent.canonical_model
ancestor_agent = ancestor_agent.parent_agent
return self._resolve_default_model()
这段代码展示了模型继承机制：如果当前 Agent 没有指定模型，它会沿着父 Agent 链向上查找，直到找到一个指定了模型的祖先，或者使用全局默认模型。这种设计允许在 Agent 树的顶层统一配置模型，子 Agent 自动继承。
3.3 SequentialAgent、ParallelAgent、LoopAgent：编排三兄弟
这三个 Agent 类型都是编排容器，它们本身不调用 LLM，而是按照不同的策略执行子 Agent。
SequentialAgent 按顺序逐个执行子 Agent：
class SequentialAgent(BaseAgent):
async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event, None]:
if not self.sub_agents:
return
agent_state = self._load_agent_state(ctx, SequentialAgentState)
start_index = self._get_start_index(agent_state)
for i in range(start_index, len(self.sub_agents)):
sub_agent = self.sub_agents[i]
async with Aclosing(sub_agent.run_async(ctx)) as agen:
async for event in agen:
yield event
ParallelAgent 使用 asyncio.TaskGroup 并行执行所有子 Agent：
class ParallelAgent(BaseAgent):
async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event, None]:
agent_runs = []
for sub_agent in self.sub_agents:
sub_agent_ctx = _create_branch_ctx_for_sub_agent(self, sub_agent, ctx)
agent_runs.append(sub_agent.run_async(sub_agent_ctx))
merge_func = _merge_agent_run if sys.version_info >= (3, 11) else _merge_agent_run_pre_3_11
async with Aclosing(merge_func(agent_runs)) as agen:
async for event in agen:
yield event
ParallelAgent 的事件合并机制非常精巧：它使用 asyncio.Queue 和 asyncio.Event 实现了背压控制——每个子 Agent 在生成事件后会等待 Runner 消费完毕再继续，避免了内存溢出。
LoopAgent 在子 Agent 列表上循环执行，直到某个子 Agent 发出 escalate 信号或达到最大迭代次数：
class LoopAgent(BaseAgent):
max_iterations: Optional[int] = None
async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event, None]:
while (not self.max_iterations or times_looped < self.max_iterations) and not should_exit:
for i in range(start_index, len(self.sub_agents)):
sub_agent = self.sub_agents[i]
async with Aclosing(sub_agent.run_async(ctx)) as agen:
async for event in agen:
yield event
if event.actions.escalate:
should_exit = True
值得注意的是，这三个编排 Agent 在源码中都被标记了 @deprecated，Google 正在将它们迁移到更强大的 Workflow 系统。
3.4 RemoteA2aAgent：跨网络的 Agent 协作
RemoteA2aAgent 是 ADK 中最具前瞻性的 Agent 类型，它实现了 Google 的 Agent-to-Agent（A2A）协议，允许本地 Agent 调用远程 Agent：
class RemoteA2aAgent(BaseAgent):
def __init__(
self,
name: str,
agent_card: Union[AgentCard, str], # 支持 AgentCard 对象、URL 或文件路径
*,
description: str = "",
httpx_client: Optional[httpx.AsyncClient] = None,
timeout: float = DEFAULT_TIMEOUT,
genai_part_converter: GenAIPartToA2APartConverter = convert_genai_part_to_a2a_part,
a2a_part_converter: A2APartToGenAIPartConverter = convert_a2a_part_to_genai_part,
...
):
RemoteA2aAgent 的核心职责包括：
Agent Card 解析：从 URL、文件或直接传入的 AgentCard对象中解析远程 Agent 的能力描述消息格式转换：在 ADK 的 Event格式和 A2A 的Message格式之间双向转换会话状态管理：跟踪远程 Agent 的 context_id，维护跨请求的会话连续性错误处理：处理网络超时、Agent Card 解析失败、远程 Agent 错误等异常场景
四、LlmAgent 深度剖析：指令、工具与回调的精密协作
4.1 指令系统：三层指令架构
LlmAgent 的指令系统分为三层：
static_instruction：静态指令，直接作为 system instruction 发送，适合不变化的内容（支持上下文缓存优化） instruction：动态指令，支持占位符替换和动态生成函数 global_instruction（已废弃）：全局指令，影响 Agent 树中的所有 Agent
static_instruction: Optional[types.ContentUnion] = None
"""Static instruction content sent literally as system instruction at the beginning.
This field is primarily for context caching optimization."""
instruction: Union[str, InstructionProvider] = ''
"""Dynamic instructions for the LLM model.
These instructions can contain placeholders like {variable_name} that will be
resolved at runtime using session state and context."""
global_instruction: Union[str, InstructionProvider] = ''
"""DEPRECATED: Use GlobalInstructionPlugin instead."""
InstructionProvider 类型别名允许指令成为一个函数：
InstructionProvider: TypeAlias = Callable[
[ReadonlyContext], Union[str, Awaitable[str]]
]
这意味着指令可以根据当前会话状态、用户信息等上下文动态生成。
4.2 工具系统：统一的工具抽象
LlmAgent 的工具系统通过 ToolUnion 类型别名统一了三种工具形态：
ToolUnion: TypeAlias = Union[Callable, BaseTool, BaseToolset]
工具解析的核心逻辑在 _convert_tool_union_to_tools 函数中：
async def _convert_tool_union_to_tools(
tool_union: ToolUnion,
ctx: Optional[ReadonlyContext],
model: Union[str, BaseLlm],
multiple_tools: bool = False,
) -> list[BaseTool]:
# 处理 GoogleSearchTool 的特殊限制
if multiple_tools and isinstance(tool_union, GoogleSearchTool):
if tool_union.bypass_multi_tools_limit:
return [GoogleSearchAgentTool(create_google_search_agent(model))]
# 处理 BaseNode（Workflow 节点）
if isinstance(tool_union, BaseNode):
if isinstance(tool_union, BaseAgent):
raise ValueError("Agent cannot be wrapped as a NodeTool.")
return [NodeTool(node=tool_union, name=tool_union.name, description=description)]
# 处理 BaseTool
if isinstance(tool_union, BaseTool):
return [tool_union]
# 处理 Callable（普通函数）
if callable(tool_union):
return [FunctionTool(func=tool_union)]
# 处理 BaseToolset
return await tool_union.get_tools_with_prefix(ctx)
这段代码揭示了 ADK 工具系统的几个重要设计决策：
普通 Python 函数会被自动包装为 FunctionToolBaseNode（Workflow 节点）可以作为工具使用，但BaseAgent不行（Agent 应该作为子 Agent 而非工具）GoogleSearchTool有特殊处理，因为它不能与其他工具混用
4.3 回调系统：六层回调钩子
LlmAgent 提供了六层回调钩子，覆盖了 Agent 执行的每一个关键节点：
# Agent 级别回调
before_agent_callback: Optional[BeforeAgentCallback] = None
after_agent_callback: Optional[AfterAgentCallback] = None
# Model 级别回调
before_model_callback: Optional[BeforeModelCallback] = None
after_model_callback: Optional[AfterModelCallback] = None
on_model_error_callback: Optional[OnModelErrorCallback] = None
# Tool 级别回调
before_tool_callback: Optional[BeforeToolCallback] = None
after_tool_callback: Optional[AfterToolCallback] = None
on_tool_error_callback: Optional[OnToolErrorCallback] = None
每个回调都支持单个函数或函数列表，列表中的回调会按顺序执行，直到某个回调返回非 None 值。这种设计允许插件系统和用户自定义回调共存。
4.4 结构化输出：output_schema 的强大能力
LlmAgent 的 output_schema 支持多种 Schema 类型：
output_schema: Optional[SchemaType] = None
"""Supports all schema types:
- type[BaseModel]: e.g., MySchema
- list[type[BaseModel]]: e.g., list[MySchema]
- list[primitive]: e.g., list[str], list[int]
- dict: Raw dict schemas
- Schema: Google's Schema type
NOTE: The ADK supports using `output_schema` and `tools` together.
It works by exposing tools during the thought loop and enforcing
structure only on the final output."""
最后一段注释特别重要——ADK 支持同时使用 output_schema 和 tools，这在其他框架中通常不支持。它通过在思考循环中暴露工具，仅在最终输出时强制结构化来实现。
五、执行模型：Runner 与 Flow 的协同
5.1 Runner：执行引擎
Runner 是 ADK 的执行引擎，负责管理 Agent 的完整生命周期：
class Runner:
app_name: str
agent: Optional[BaseAgent | 'BaseNode'] = None
artifact_service: Optional[BaseArtifactService] = None
plugin_manager: PluginManager
session_service: BaseSessionService
memory_service: Optional[BaseMemoryService] = None
credential_service: Optional[BaseCredentialService] = None
context_cache_config: Optional[ContextCacheConfig] = None
resumability_config: Optional[ResumabilityConfig] = None
Runner 的 run_async 方法是整个框架的入口点。它的工作流程如下：
会话获取/创建：通过 session_service获取或创建会话用户消息处理：将用户消息添加到会话历史 InvocationContext 构建：创建包含所有服务引用的调用上下文 Agent 执行：调用 agent.run_async(ctx)并收集事件事件持久化：将事件追加到会话历史 状态管理：处理状态变更、Artifact 保存等
Runner 还支持通过 App 对象进行配置：
def _resolve_app(app, app_name, agent, node, plugins) -> App:
"""Exactly one of app, agent, or node must be provided."""
provided = sum(x is not None for x in (app, agent, node))
if provided > 1:
raise ValueError('Only one of app, agent, or node may be provided.')
if provided == 0:
raise ValueError('One of app, agent, or node must be provided.')
# Normalize to App
if agent is not None:
return App.model_construct(name=app_name, root_agent=agent, plugins=plugins or [])
return app
5.2 Flow：LlmAgent 的执行流程
当 Runner 执行一个 LlmAgent 时，它不会直接调用 LLM，而是委托给 BaseLlmFlow。LlmAgent 的 _llm_flow 属性决定了使用哪种 Flow：
@property
def _llm_flow(self) -> BaseLlmFlow:
if (
self.disallow_transfer_to_parent
and self.disallow_transfer_to_peers
and not self.sub_agents
):
return SingleFlow()
else:
return AutoFlow()
SingleFlow：当 Agent 不能转移控制权且没有子 Agent 时使用，是一个简单的"调用 LLM → 处理工具 → 返回结果"循环 AutoFlow：当 Agent 可能需要转移控制权时使用，包含完整的 Agent 转移逻辑
Flow 的核心执行逻辑在 BaseLlmFlow 中：
# base_llm_flow.py 中的核心流程
async def run_async(self, ctx: InvocationContext) -> AsyncGenerator[Event, None]:
# 1. 解析工具集认证
async for event in _resolve_toolset_auth(ctx, agent):
yield event
# 2. 主循环：LLM 调用 → 工具执行 → 重复
while not ctx.end_invocation:
# 构建 LLM 请求
llm_request = await self._build_llm_request(ctx)
# 调用 LLM
async for llm_response in self._call_llm(ctx, llm_request):
# 处理 before_model_callback
callback_response = await _handle_before_model_callback(ctx, llm_request, event)
if callback_response:
yield callback_response
continue
# 处理 after_model_callback
callback_response = await _handle_after_model_callback(ctx, llm_response, event)
if callback_response:
llm_response = callback_response
# 构建响应事件
model_response_event = _finalize_model_response_event(llm_request, llm_response, event)
yield model_response_event
# 如果有函数调用，执行工具
if function_calls := model_response_event.get_function_calls():
tool_events = await self._execute_tools(ctx, function_calls, llm_request)
for tool_event in tool_events:
yield tool_event
这个循环揭示了 ADK 的核心执行模型：LLM 调用 → 工具执行 → 将工具结果反馈给 LLM → 重复，直到 LLM 生成最终响应或转移控制权。
5.3 工具处理的并发优化
Flow 中的工具处理有一个重要的并发优化：
async def _process_agent_tools(invocation_context, llm_request):
# 并发解析所有工具
resolved_tools_per_union = await asyncio.gather(*(
_convert_tool_union_to_tools(tool_union, ReadonlyContext(invocation_context), model, multiple_tools)
for tool_union in agent.tools
))
# 串行提交（保持原始顺序）
for tool_union, tools in zip(agent.tools, resolved_tools_per_union):
tool_context = ToolContext(invocation_context)
if isinstance(tool_union, BaseToolset):
await tool_union.process_llm_request(tool_context=tool_context, llm_request=llm_request)
for tool in tools:
await tool.process_llm_request(tool_context=tool_context, llm_request=llm_request)
工具解析阶段使用 asyncio.gather 并发执行（特别是 MCP 工具的 list_tools 需要网络 I/O），但提交阶段保持串行以确保 llm_request 的状态一致性。
六、工具系统：从函数到 REST API 的全栈支持
6.1 FunctionTool：最基础的工具
FunctionTool 将普通 Python 函数包装为 ADK 工具。它通过函数签名和类型注解自动生成工具声明：
# 一个普通的 Python 函数
def search_weather(city: str) -> dict:
"""Search for weather information.
Args:
city: The city name to search for.
Returns:
Weather information dictionary.
"""
return {"city": city, "temp": 25}
# 自动包装为 FunctionTool
agent = LlmAgent(tools=[search_weather]) # search_weather 被自动包装
6.2 OpenAPIToolset：从 OpenAPI Spec 到工具
ADK 的 OpenAPI 工具集是其企业级特性的典型体现。它可以从 OpenAPI 规范自动生成 REST API 工具：
class OpenAPIToolset(BaseToolset):
def __init__(
self,
*,
spec_dict: Optional[Dict[str, Any]] = None,
spec_str: Optional[str] = None,
spec_str_type: Literal["json", "yaml"] = "json",
auth_scheme: Optional[AuthScheme] = None,
auth_credential: Optional[AuthCredential] = None,
credential_key: Optional[str] = None,
tool_filter: Optional[Union[ToolPredicate, List[str]]] = None,
tool_name_prefix: Optional[str] = None,
ssl_verify: Optional[Union[bool, str, ssl.SSLContext]] = None,
header_provider: Optional[Callable[[ReadonlyContext], Dict[str, str]]] = None,
httpx_client_factory: Optional[HttpxClientFactory] = None,
preserve_property_names: bool = False,
):
OpenAPIToolset 的初始化流程：
解析 OpenAPI Spec（支持 JSON 和 YAML） 使用 OpenApiSpecParser提取所有操作将每个操作转换为 RestApiTool实例配置认证方案和凭证
6.3 RestApiTool：通用 REST API 工具
RestApiTool 是 ADK 中最复杂的工具类型，它能够将 OpenAPI 操作转换为 LLM 可调用的工具：
class RestApiTool(BaseTool):
def _get_declaration(self) -> FunctionDeclaration:
"""Returns the function declaration in the Gemini Schema format."""
schema_dict = self._operation_parser.get_json_schema()
if is_feature_enabled(FeatureName.JSON_SCHEMA_FOR_FUNC_DECL):
return FunctionDeclaration(
name=self.name, description=self.description,
parameters_json_schema=schema_dict,
)
else:
parameters = _to_gemini_schema(schema_dict)
return FunctionDeclaration(
name=self.name, description=self.description, parameters=parameters
)
async def call(self, *, args: dict[str, Any], tool_context: Optional[ToolContext]) -> Dict[str, Any]:
# 准备认证凭证
tool_auth_handler = ToolAuthHandler.from_tool_context(
tool_context, self.auth_scheme, self.auth_credential, credential_key=self.credential_key
)
auth_result = await tool_auth_handler.prepare_auth_credentials()
if auth_result.state == "pending":
return {"pending": True, "message": "Waiting for auth credential"}
# 准备请求参数
request_params = self._prepare_request_params(parameters, args)
# 执行 HTTP 请求
async with httpx.AsyncClient(verify=self._ssl_verify) as client:
response = await client.request(**request_params)
return response.json()
6.4 工具认证处理
ADK 的工具认证系统是其企业级特性的核心。ToolAuthHandler 负责处理各种认证流程：
class ToolAuthHandler:
async def prepare_auth_credentials(self) -> AuthPreparationResult:
if not self.auth_scheme:
return AuthPreparationResult(state="done")
# 检查已有凭证
existing_credential = await self._get_existing_credential()
credential = existing_credential or self.auth_credential
# 处理 OAuth2/OIDC 的多步骤交换
if not credential or self._external_exchange_required(credential):
credential = self._get_auth_response()
if credential:
self._store_credential(credential)
else:
self._request_credential() # 中断执行，等待用户授权
return AuthPreparationResult(state="pending", ...)
# 交换凭证
exchanged_credential = self._exchange_credential(credential)
return AuthPreparationResult(state="done", auth_credential=exchanged_credential)
认证流程支持：
API Key 认证 OAuth2（Authorization Code、Client Credentials） OpenID Connect Service Account HTTP Bearer Token
七、Session 与 State：会话管理的精细设计
7.1 InvocationContext：执行上下文的中枢
InvocationContext 是 ADK 中最核心的数据结构之一，它承载了一次 Agent 调用的全部状态：
class InvocationContext(BaseModel):
artifact_service: Optional[BaseArtifactService] = None
session_service: BaseSessionService
memory_service: Optional[BaseMemoryService] = None
credential_service: Optional[BaseCredentialService] = None
invocation_id: str
branch: Optional[str] = None
agent: Optional[BaseAgent | BaseNode] = None
user_content: Optional[types.Content] = None
session: Session
agent_states: dict[str, dict[str, Any]] = Field(default_factory=dict)
end_of_agents: dict[str, bool] = Field(default_factory=dict)
end_invocation: bool = False
run_config: Optional[RunConfig] = None
plugin_manager: PluginManager = Field(default_factory=PluginManager)
credential_by_key: dict[str, AuthCredential] = Field(default_factory=dict)
InvocationContext 的注释中有一段非常有价值的架构说明：
An invocation:
1. Starts with a user message and ends with a final response.
2. Can contain one or multiple agent calls.
3. Is handled by runner.run_async().
An LLM agent call can contain one or multiple steps.
A step:
1. Calls the LLM only once and yields its response.
2. Calls the tools and yields their responses if requested.
┌─────────────────────── invocation ──────────────────────────┐
┌──────────── llm_agent_call_1 ────────────┐ ┌─ agent_call_2 ─┐
┌──── step_1 ────────┐ ┌───── step_2 ──────┐
[call_llm] [call_tool] [call_llm] [transfer]
7.2 Context：运行时上下文
Context 是 ReadonlyContext 的可变子类，提供了运行时操作接口：
class Context(ReadonlyContext):
@property
def state(self) -> State:
"""The delta-aware state of the current session.
For any state change, you can mutate this object directly,
e.g. `ctx.state['foo'] = 'bar'`"""
return self._state
@property
def actions(self) -> EventActions:
"""The event actions for the current context."""
return self._event_actions
async def run_node(self, node: NodeLike, node_input: Any = None, **kwargs) -> Any:
"""Executes a node dynamically within a workflow."""
return await self._run_node_internal(node, node_input, **kwargs)
State 对象是 delta-aware 的——它不会直接修改会话状态，而是将变更记录在 state_delta 中，由 Runner 在事件处理阶段合并到会话状态。
7.3 会话状态的分支隔离
ADK 通过 branch 字段实现了会话状态的分支隔离，这对于并行 Agent 执行至关重要：
branch: Optional[str] = None
"""The format is like agent_1.agent_2.agent_3.
Branch is used when multiple sub-agents shouldn't see their peer agents'
conversation history."""
在 InvocationContext._get_events 中，分支过滤逻辑确保每个 Agent 只能看到自己分支的事件：
def _is_branch_match(event: Event) -> bool:
if getattr(event, "author", None) == "user":
# 验证用户响应是否针对当前分支的函数调用
fr_ids = {fr.id for fr in event.get_function_responses() if fr.id is not None}
branch_fc_ids = {fc.id for e in branch_events for fc in e.get_function_calls()}
if not (fr_ids & branch_fc_ids):
return False
return event.branch == self.branch
八、Artifact 系统：二进制制品的存储抽象
ADK 的 Artifact 系统提供了统一的二进制制品存储接口：
class BaseArtifactService(abc.ABC):
@abc.abstractmethod
async def save_artifact(
self, *, app_name: str, user_id: str, session_id: str,
filename: str, artifact: types.Part,
) -> int:
"""Save an artifact and return the version number."""
@abc.abstractmethod
async def load_artifact(
self, *, app_name: str, user_id: str, session_id: str,
filename: str, version: Optional[int] = None,
) -> Optional[types.Part]:
"""Load an artifact. If version is None, load the latest version."""
@abc.abstractmethod
async def list_artifact_keys(
self, *, app_name: str, user_id: str, session_id: str,
) -> list[str]:
"""List all artifact filenames for a session."""
ADK 内置了多种 Artifact 服务实现：
InMemoryArtifactService：内存存储，适合开发和测试 GcsArtifactService：Google Cloud Storage 存储，适合生产环境 BaseArtifactService 的文件系统实现：本地文件存储
Artifact 系统支持版本控制——每次保存都会返回一个版本号，加载时可以指定版本。
九、Auth 系统：企业级认证的全面支持
ADK 的认证系统是其企业级定位的核心体现。它由三个核心组件构成：
9.1 AuthCredential：认证凭证
class AuthCredential(BaseModel):
auth_type: AuthCredentialTypes
http: Optional[HttpCredentials] = None
oauth2: Optional[OAuth2Auth] = None
service_account: Optional[ServiceAccountCredential] = None
api_key: Optional[str] = None
9.2 AuthScheme：认证方案
class AuthScheme(BaseModel):
type_: AuthSchemeType # apiKey, http, oauth2, openIdConnect
# ... 具体配置
9.3 认证辅助函数
ADK 提供了便捷的辅助函数来创建常见的认证配置：
def token_to_scheme_credential(
token_type: Literal["apikey", "oauth2Token"],
location: Optional[Literal["header", "query", "cookie"]] = None,
name: Optional[str] = None,
credential_value: Optional[str] = None,
) -> Tuple[AuthScheme, AuthCredential]:
"""Creates AuthScheme and AuthCredential for API key or bearer token."""
认证凭证的存储通过 ToolContextCredentialStore 实现，它利用会话状态（session.state）来跨请求持久化凭证：
class ToolContextCredentialStore:
def get_credential_key(self, auth_scheme, auth_credential) -> str:
"""Generates a unique key based on scheme and credential digest."""
scheme_name = f"{auth_scheme.type_.name}_{_stable_model_digest(auth_scheme)}"
credential_name = f"{auth_credential.auth_type.value}_{_stable_model_digest(auth_credential)}"
return f"{scheme_name}_{credential_name}_existing_exchanged_credential"
def get_credential(self, auth_scheme, auth_credential) -> Optional[AuthCredential]:
token_key = self.get_credential_key(auth_scheme, auth_credential)
serialized_credential = self.tool_context.state.get(token_key)
if serialized_credential:
return AuthCredential.model_validate(serialized_credential)
十、A2A 协议：Agent 互操作的未来
ADK 的 A2A（Agent-to-Agent）协议是 Google 推动的 Agent 互操作标准。RemoteA2aAgent 是其客户端实现。
10.1 消息格式转换
A2A 协议的核心挑战是消息格式的双向转换。ADK 通过 converters 模块处理这个问题：
from ..a2a.converters.event_converter import convert_a2a_message_to_event
from ..a2a.converters.event_converter import convert_a2a_task_to_event
from ..a2a.converters.event_converter import convert_event_to_a2a_message
from ..a2a.converters.part_converter import convert_a2a_part_to_genai_part
from ..a2a.converters.part_converter import convert_genai_part_to_a2a_part
10.2 会话状态管理
RemoteA2aAgent 通过 context_id 维护与远程 Agent 的会话连续性：
def _construct_message_parts_from_session(self, ctx: InvocationContext):
message_parts = []
context_id = None
for event in reversed(ctx.session.events):
if self._is_remote_response(event):
if event.custom_metadata:
context_id = event.custom_metadata.get(A2A_METADATA_PREFIX + "context_id")
if not self._full_history_when_stateless or context_id:
break
# 转换事件为 A2A 消息部分
for part in processed_event.content.parts:
converted_parts = self._genai_part_converter(part)
message_parts.extend(converted_parts)
return message_parts, context_id
10.3 假函数调用机制
当远程 Agent 需要用户输入时，ADK 使用巧妙的"假函数调用"机制：
MOCK_FUNCTION_CALL_FOR_REQUIRED_USER_INPUT = "__adk_mock_fc_for_required_user_input"
MOCK_FUNCTION_CALL_FOR_REQUIRED_USER_AUTH = "__adk_mock_fc_for_required_user_auth"
这些假函数调用模拟了本地工具的行为，使得 Runner 可以用统一的方式处理本地和远程 Agent 的用户交互需求。
十一、多 Agent 模式：编排的艺术
ADK 支持多种多 Agent 编排模式：
11.1 层级委派模式
LlmAgent 的 sub_agents 字段和自动转移机制实现了层级委派：
coordinator = LlmAgent(
name="coordinator",
model="gemini-3.5-flash",
instruction="You coordinate tasks between specialists.",
sub_agents=[
LlmAgent(name="researcher", instruction="You research topics.", ...),
LlmAgent(name="writer", instruction="You write content.", ...),
],
)
当 Coordinator 判断某个任务适合由 Researcher 处理时，它会调用内置的 transfer_to_agent 函数将控制权转移给 Researcher。
11.2 流水线模式
SequentialAgent 实现了流水线模式，子 Agent 按顺序处理数据：
pipeline = SequentialAgent(
name="pipeline",
sub_agents=[
LlmAgent(name="extractor", instruction="Extract key information."),
LlmAgent(name="analyzer", instruction="Analyze the extracted data."),
LlmAgent(name="reporter", instruction="Generate a report."),
],
)
11.3 并行探索模式
ParallelAgent 实现了并行探索模式，多个 Agent 同时处理同一任务：
explorer = ParallelAgent(
name="explorer",
sub_agents=[
LlmAgent(name="approach_a", instruction="Try approach A."),
LlmAgent(name="approach_b", instruction="Try approach B."),
LlmAgent(name="approach_c", instruction="Try approach C."),
],
)
11.4 迭代优化模式
LoopAgent 实现了迭代优化模式，Agent 在循环中不断改进结果：
optimizer = LoopAgent(
name="optimizer",
max_iterations=5,
sub_agents=[
LlmAgent(name="generator", instruction="Generate a solution."),
LlmAgent(name="evaluator", instruction="Evaluate the solution. Escalate if good enough."),
],
)
十二、Apps 与部署：从开发到生产的桥梁
12.1 App 对象：应用的封装
App 是 ADK 应用的顶层封装：
class App(BaseModel):
name: str
root_agent: BaseAgent | BaseNode
plugins: list[BasePlugin] = Field(default_factory=list)
context_cache_config: Optional[ContextCacheConfig] = None
resumability_config: Optional[ResumabilityConfig] = None
Runner 推荐使用 App 对象来初始化：
app = App(
name="my_app",
root_agent=agent,
plugins=[GlobalInstructionPlugin("You are a helpful assistant.")],
)
runner = Runner(app=app, session_service=session_service)
12.2 RunConfig：运行时配置
RunConfig 控制 Agent 的运行时行为：
class RunConfig(BaseModel):
streaming_mode: StreamingMode = StreamingMode.NONE # NONE, SSE, BIDI
max_llm_calls: int = 500
speech_config: Optional[types.SpeechConfig] = None
response_modalities: Optional[list[types.Modality]] = None
tool_thread_pool_config: Optional[ToolThreadPoolConfig] = None
save_live_blob: bool = False
context_window_compression: Optional[types.ContextWindowCompressionConfig] = None
RunConfig 特别值得注意的配置：
streaming_mode：支持 NONE（非流式）、SSE（Server-Sent Events）、BIDI（双向流）三种模式 max_llm_calls：限制 LLM 调用次数，防止无限循环 tool_thread_pool_config：工具线程池配置，用于 Live 模式下的并发工具执行
12.3 可恢复执行
ADK 支持可恢复执行（Resumability），这对于长时间运行的 Agent 任务至关重要：
class ResumabilityConfig(BaseModel):
is_resumable: bool = False
# InvocationContext 中的状态管理
def set_agent_state(self, agent_name, *, agent_state=None, end_of_agent=False):
if end_of_agent:
self.end_of_agents[agent_name] = True
self.agent_states.pop(agent_name, None)
elif agent_state is not None:
self.agent_states[agent_name] = agent_state.model_dump(mode="json")
self.end_of_agents[agent_name] = False
每个编排 Agent（Sequential、Parallel、Loop）都维护自己的状态，可以在中断后从断点恢复执行。
总结：ADK 的架构哲学
通过深入分析 ADK 的源码，我们可以总结出 Google 在 Agent 框架设计上的核心哲学：
全栈优先：ADK 不仅仅是一个 Agent 调用库，它提供了从开发到部署的完整基础设施——Session 管理、Artifact 存储、认证系统、A2A 协议、插件系统、可观测性。
类型安全：大量使用 Pydantic 模型、类型别名和泛型，确保类型安全和代码可维护性。
ToolUnion、InstructionProvider、BeforeModelCallback等类型别名体现了对开发者体验的关注。企业级设计：OpenAPI 工具集、OAuth2/OIDC 认证、Service Account 支持、SSL 配置、线程池等特性明确面向企业级生产环境。
可扩展性：通过回调系统（六层回调钩子）、插件系统、BaseToolset 抽象等机制，ADK 提供了丰富的扩展点。
Agent 互操作：A2A 协议和 RemoteA2aAgent 体现了 Google 对 Agent 生态互操作性的重视，这是其他框架尚未深入探索的领域。
渐进式复杂度：从简单的
LlmAgent+FunctionTool到完整的App+Runner+ 多种 Agent 类型，ADK 允许开发者根据需求选择合适的复杂度级别。
ADK 的代码量和复杂度远超 OpenAI Agents SDK 和 Claude Code，这既是它的优势（功能全面）也是它的挑战（学习曲线陡峭）。但对于需要构建企业级 Agent 系统的团队来说，ADK 提供了最完整的解决方案。
本文基于 Google ADK Python 源码分析撰写，涵盖 agents/、tools/、flows/、runners.py 等核心模块的 26 个源文件。
