news 2026/9/13 19:46:41

ADK single_turn 子代理实战:把无状态子 Agent 当结构化工具调用,同时保留完整会话痕迹

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
ADK single_turn 子代理实战:把无状态子 Agent 当结构化工具调用,同时保留完整会话痕迹

ADK single_turn 子代理实战:把无状态子 Agent 当结构化工具调用,同时保留完整会话痕迹

【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python

在 Google ADK(adk-python)的多 Agent 体系中,mode="single_turn"子代理是一种"即插即用"的委托机制:父 Agent 把它当普通工具(function)调用,子 Agent 在隔离分支内独立跑完"输入 → 工具调用 → 结构化输出"的完整闭环,全程不与用户对话,且其内部事件仍保留在会话历史中。本文以仓库中的完整示例 single_turn_sub_agent 为主体,完整拆解"Pixel 手机推荐"样例的实现代码、事件流转过程与框架源码原理,读完你可以掌握:如何定义带input_schema/output_schema的 single-turn 子代理、如何在父代理中挂载它、以及它与传统AgentTool模式的本质区别。

1. 样例概览:为什么用 single_turn 子代理替代 AgentTool

README 明确指出,该样例演示的是一个single_turn模式代理如何作为 LLM 代理的自主子代理运行——"利用 schema 和工具,且从不与用户交互"。它同时强调了这一机制的定位:

这是推荐用来替代旧AgentTool模式的机制。与AgentTool不同,使用single_turn子代理时,子代理的内部交互(如工具调用)会保留在会话历史(session history)中。

这一点是两者的核心分水岭:旧的AgentTool把子代理包装成一个"黑盒函数",子代理内部发生了多少次模型推理、调用了哪些工具,在父级会话里看不到;而 single-turn 子代理的执行事件会以独立分支(branch)的形式落盘到同一个 session 中,便于审计、回放和调试。

样例中两个角色的分工如下:

  1. phone_recommender:single-turn 子代理。接收结构化输入(UserPreferences),调用 mock 工具(check_phone_price)查询价格,最终返回结构化输出(PhoneRecommendation)。
  2. root_agent:主代理。负责与用户多轮对话、把自然语言请求翻译成UserPreferences,然后委托phone_recommender

对应的代理拓扑(来自样例 README 的 Mermaid 图):

2. 完整实现:从 Schema 到代理定义

样例全部代码见 agent.py,结构非常紧凑,共四部分:输入/输出 Schema、mock 工具、子代理定义、根代理定义。

2.1 结构化输入 Schema:UserPreferences

single-turn 子代理的第一个特征是"契约化输入"——用 Pydantic 模型显式声明父代理必须传入什么字段:

from pydantic import BaseModel from pydantic import Field class UserPreferences(BaseModel): budget: int = Field(description="The user's maximum budget in USD") primary_use: str = Field( description=( "What the user primarily uses their phone for (e.g., photography," " gaming, basics)" ) ) preferred_size: str = Field( description="Preferred phone size (e.g., small, large, any)" )

三个字段(预算、主要用途、尺寸偏好)的description很关键:它们会被框架转成工具调用的参数说明,直接决定父 LLM 能否正确填参。

2.2 结构化输出 Schema:PhoneRecommendation

class PhoneRecommendation(BaseModel): """Output schema for the phone recommendation.""" model_name: str price: float reason: str

2.3 Mock 价格工具

def check_phone_price(model_name: str) -> float: """Mock tool to check the current price of a Pixel phone model.""" prices = { "Pixel 10a": 499.0, "Pixel 10": 799.0, "Pixel 10 Pro": 999.0, "Pixel 10 Pro XL": 1199.0, "Pixel 10 Pro Fold": 1799.0, } # Simple mock logic, defaulting to 799 if not found exactly for key, value in prices.items(): if key.lower() in model_name.lower(): return value return 799.0

注意 mock 逻辑的细节:按"键名是否为入参子串"做模糊匹配(如Pixel 10会同时命中Pixel 10Pixel 10 Pro等键),未命中时默认返回 799。这解释了后面事件流中反复出现的799.0结果。

2.4 子代理定义:phone_recommender

这是本文的核心配置,四个参数缺一不可:

phone_recommender = Agent( name="phone_recommender", mode="single_turn", input_schema=UserPreferences, output_schema=PhoneRecommendation, tools=[check_phone_price], instruction=("""\ You are an expert Google Pixel hardware recommender. Based on the provided UserPreferences, recommend exactly one Pixel phone model. You must use the `check_phone_price` tool to find the exact current price of the model you are recommending before you finish your task. Only recommend these phones: Pixel 10a, Pixel 10, Pixel 10 Pro, Pixel 10 Pro XL, Pixel 10 Pro Fold. """), description="Recommends a Pixel phone based on preferences.", )
  • mode="single_turn":声明执行模式。作为子代理时 LlmAgent 默认是chat模式,所以要显式写出;
  • input_schema/output_schema:定义进出契约,框架据此生成工具声明;
  • tools=[check_phone_price]:子代理内部可自主调用的工具;
  • description:父代理决定"何时该调用这个工具"时依赖这段描述,务必写清适用场景。

instruction 里"必须先用check_phone_price查价"这条约束保证了子代理的推荐结果有工具调用佐证,而不是模型凭记忆报价。

2.5 根代理定义:root_agent

挂载只需一行sub_agents=[phone_recommender]

root_agent = Agent( name="root_agent", sub_agents=[phone_recommender], instruction=("""\ You are a helpful phone sales associate. If the user is asking for a phone recommendation, use the `phone_recommender` to get a structured recommendation. Once the recommender finishes, present the model, price, and reason to the user in a friendly way. """), )

注意root_agent本身没有设置mode——按源码约定,LlmAgent子代理不声明mode时默认为chat(见下文源码分析),因此root_agent仍作为标准聊天代理直接与用户对话。

3. 框架原理:single_turn 子代理如何变成"工具"

3.1 挂载阶段:sub_agent 被包装为 _SingleTurnAgentTool

在 llm_agent.py 中,父代理初始化时会对每个sub_agent检查mode字段并做相应包装:

if self.sub_agents: for sub_agent in self.sub_agents: mode = getattr(sub_agent, 'mode', None) # LlmAgent sub-agents default to chat mode (unchanged behavior). if isinstance(sub_agent, LlmAgent) and mode is None: sub_agent.mode = 'chat' mode = 'chat' if mode == 'single_turn': self.tools.append(_SingleTurnAgentTool(sub_agent)) elif mode == 'task': self.tools.append(_TaskAgentTool(sub_agent))

由此可以确认三件事:

  1. 声明了mode='single_turn'的子代理会被包进_SingleTurnAgentTool并追加到父代理的tools列表——父代理因此把它当普通函数工具调用;
  2. single_turn子代理不是transfer 目标,父代理不能通过transfer_to_agent把会话控制权交给它(这一点官方指南 single_turn.md 在 Limitations 一节同样明确);
  3. 未声明modeLlmAgent子代理默认chat,仍走传统的 transfer/对话式委托路径。

3.2 执行阶段:输入校验、子分支与作用域隔离

包装工具的核心实现在 agent_tool.py 的_SingleTurnAgentTool.run_async中:

class _SingleTurnAgentTool(AgentTool): """A tool that wraps a single-turn agent and runs it via ctx.run_node.""" @override async def run_async(self, *, args, tool_context) -> Any: input_schema = _get_input_schema(self.agent) node_input if input_schema: try: node_input = input_schema.model_validate(args) except Exception as e: return f'Error validating input: {e}' else: node_input = args.get('request') # Align subagent branch scoping with node execution using function_call_id. fc_id = tool_context.function_call_id base_branch = tool_context.get_invocation_context().branch tool_branch = _BranchPath.create_sub_branch( base_branch, name=self.agent.name, run_id=fc_id ) try: return await tool_context.run_node( self.agent, node_input=node_input, override_branch=tool_branch, use_sub_branch=False, ) except Exception as e: return f'Error running sub-agent: {e}'

从源码结构看,一次工具调用内部完成了四步:

  1. Schema 校验:父代理传来的args先经input_schema.model_validate校验(本例即UserPreferences),校验失败直接返回错误字符串,不会让子代理"带病运行";
  2. 子分支创建:用本次function_call_id(fc_id)作为 run_id,从父分支派生出形如父分支/phone_recommender@fc-1的子分支,子代理的所有事件都写在这个分支下;
  3. run_node 执行:通过tool_context.run_node把子代理当作工作流节点执行,node_input就是校验后的结构化对象;
  4. 异常兜底:子代理抛错时把异常信息作为工具返回值传回父代理,父对话不会中断。

配合官方指南 single_turn.md 对上下文隔离的说明,可见 single-turn 子代理默认include_contents="none":它不加载会话历史,只能看到工具调用传入的入参。子分支对父分支是"单向可见"的(父分支读不到子分支的内部推理事件),父代理拿到的只是最终返回值。如果希望子代理能感知父级对话历史,需显式设置include_contents="default"

关键参数速查(综合样例与官方指南):

参数类型默认值说明
modeLiteral['single_turn', 'task', 'chat']子代理默认'chat';工作流节点默认'single_turn'single_turn隔离执行、作为工具调用;task支持委托;chat保留完整历史
include_contentsLiteral['default', 'none']single_turn未显式设置时为'none'控制历史可见性;显式设'default'可让子代理感知父级对话
input_schema/output_schemaPydantic 模型定义进出契约;有input_schema时父代理按字段传参,否则退化为单个request字符串参数

4. 事件流实证:一次推荐请求在 session 里长什么样

样例的tests/目录提供了完整的事件快照,如 gaming_1000.json 与 go.json,记录了从用户输入到最终回复的全部Event。以 gaming_1000 会话为例,事件链清晰呈现了 single-turn 委托的完整生命周期:

  1. 多轮澄清(父分支root_agent@1):用户说 "gaming, $1000",root_agent先反问手机尺寸偏好,用户答 "large";

  2. 触发子代理工具root_agent发出functionCall,名字就是子代理名phone_recommender,参数为校验前的结构化入参:

    {"functionCall": {"args": {"budget": 1000, "preferred_size": "large", "primary_use": "gaming"}, "name": "phone_recommender"}}
  3. 子代理在子分支内执行:后续事件全部带有"branch": "phone_recommender@fc-1"与路径root_agent@1/phone_recommender@1。子代理连续多次调用check_phone_price(依次查 Pixel 10 Pro XL、Pro Fold、Pro、10、10a 的价格),每次工具返回都以functionResponse事件落盘——这些正是旧AgentTool模式下不可见的"内部交互";

  4. 结构化输出:子代理调用set_model_response提交{model_name, price, reason},事件actions.setModelResponse记录结构化结果,并生成一条messageAsOutput: true的输出事件;

  5. 回传父代理:父分支收到phone_recommender工具的functionResponse(即PhoneRecommendation的 dict),root_agent据此向用户给出友好文案:"I recommend thePixel 10 Pro XL. Price: $799 ..."。

这套事件快照也是从源码_SingleTurnAgentTool的分支命名逻辑(name + run_id=function_call_id)得到印证的——事件里的fc-1正是那次functionCall的 id。

5. 运行样例

样例支持 README "Sample Inputs" 一节给出的典型问法,例如:

  • I need a phone mostly for gaming. I have about $1000 to spend.
  • What is a good cheap phone from Google for basic tasks?
  • I love photography but prefer smaller phones. My budget is $600.

运行方式(前提:已安装 ADK 并配置好可用的 LLM 凭据,如 Gemini API key):

# 在仓库根目录启动 ADK 开发 UI,选择 contributing/samples/multi_agent/single_turn_sub_agent 后在对话框输入上面的问题 adk web # 或直接在命令行运行该代理目录 adk run contributing/samples/multi_agent/single_turn_sub_agent

多轮交互中,root_agent会主动补齐缺失的UserPreferences字段(如示例中反问尺寸偏好),凑齐后才发起对phone_recommender的工具调用——这正是结构化input_schema带来的行为:缺参时父代理宁可多问一轮,也不会猜。

6. 小结与延伸阅读

  • 何时选 single_turn 子代理:任务有清晰的输入/输出契约、不需要与用户多轮往返、但希望保留内部执行痕迹(工具调用、推理事件)供审计时,它就是推荐做法;
  • AgentTool的本质区别:事件留痕在 session 中,且执行走统一的run_node节点机制,与工作流节点共享分支/作用域语义;
  • 局限:不能用transfer_to_agent直接转移给 single-turn 代理;默认无状态(include_contents="none"),需要上下文时必须显式打开。

进一步阅读:

  • 官方指南 LlmAgent Single-Turn Mode:完整讲解工作流节点、子代理两种部署形态、mode/include_contents配置及上下文感知(context-aware)高级用法;
  • 样例源码 agent.py 与事件快照 tests/gaming_1000.json、tests/go.json;
  • 框架实现 llm_agent.py 子代理包装逻辑 与 agent_tool.py _SingleTurnAgentTool。

【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python

创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考

版权声明: 本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若内容造成侵权/违法违规/事实不符,请联系邮箱:809451989@qq.com进行投诉反馈,一经查实,立即删除!
网站建设 2026/9/13 19:44:26

Argo CD 如何执行部分资源的选择性同步(Selective Sync)

Argo CD 如何执行部分资源的选择性同步(Selective Sync) 【免费下载链接】argo-cd Declarative Continuous Deployment for Kubernetes 项目地址: https://gitcode.com/GitHub_Trending/ar/argo-cd 当一次同步只想作用于 Application 中的部分资源…

作者头像 李华
网站建设 2026/9/13 19:39:26

darktable 实战指南:RAW 到成片的 5 步工作流

darktable 实战指南:RAW 到成片的 5 步工作流 【免费下载链接】darktable darktable is an open source photography workflow application and raw developer 项目地址: https://gitcode.com/GitHub_Trending/da/darktable 把一堆 RAW 倒进文件夹&#xff0…

作者头像 李华