ADK Python 实战：用 YAML 配置构建带反馈回路的 Workflow（loop

ADK Python 实战：用 YAML 配置构建带反馈回路的 Workflow（loop_config 样本详解）

【免费下载链接】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

本文围绕 ADK（Agent Development Kit）Python 版中的loop_config官方样本，完整讲解如何用 YAML 配置文件定义一个“生成—评估—反馈—重试”的反馈回路工作流：包括root_agent.yaml中 edges 的完整写法、代码引用（Code References）、边内函数引用（Function References in Edges）与外部 Agent 文件（External Agent Files）三处特殊语法的语义和解析原理，以及通过黄金测试事件序列验证回路实际运行的方法。读完后你将能够直接用配置代替 Python 代码来搭建带循环路由的 ADK Workflow。

一、样本解决什么问题

loop_config是 contributing/samples/workflows/loop_config/README.md 所描述的官方样本，它演示如何用 YAML 定义一个带反馈回路（feedback loop）的 Workflow：

process_input：把用户输入（主题）写入会话状态；
generate_headline：LLM Agent 根据主题生成标题，并参考上一轮反馈；
evaluate_headline：LLM Agent 对标题打分（tech-related或unrelated）并给出改进建议；
route_headline：按打分结果路由——若为unrelated，携带反馈回到generate_headline重新生成，形成回路；若为tech-related，流程结束。

该样本与 Python 版本的 contributing/samples/workflows/loop/agent.py 行为完全对应（README 称其 "mirrors" 该样本），区别仅在于图结构用 YAML 而非 Python 字面量表达。README 明确指出：加载root_agent.yaml后，agent_class: Workflow会被解析为Workflow类，edges会映射到Workflow同名字段，最终得到五条边、最后一条为unrelated回到generate_headline的Workflow实例。

二、样本文件结构与职责

contributing/samples/workflows/loop_config/ ├── README.md # 样本说明（本文主体来源） ├── root_agent.yaml # 工作流图定义：agent_class + edges ├── agent.py # Feedback 模型 + process_input / route_headline 函数 ├── generate_headline.yaml # 生成标题的 LlmAgent 定义 ├── evaluate_headline.yaml # 评估标题的 LlmAgent 定义（含 output_schema） └── tests/ ├── flower.json # 黄金事件序列：非科技主题触发回路 └── computer.json # 黄金事件序列

各文件职责：

文件	作用
`root_agent.yaml`	根工作流声明，`agent_class: Workflow`，用`edges`描述整张执行图（含回路边）
`agent.py`	存放 Pydantic 结构化输出模型`Feedback`，以及两个普通 Python 函数`process_input`、`route_headline`
`generate_headline.yaml`/`evaluate_headline.yaml`	独立 YAML 定义的`LlmAgent`，被 root 边引用
`tests/*.json`	从目录结构看是录制的完整会话事件流（事件列表 + 最终 state），可用来逐步核对每个节点的输入输出

三、完整配置文件逐文件解读

3.1 根工作流：root_agent.yaml

contributing/samples/workflows/loop_config/root_agent.yaml 全文（去掉许可证头）：

agent_class: Workflow name: root_agent edges: - - START - .agent.process_input - generate_headline.yaml - evaluate_headline.yaml - .agent.route_headline - - .agent.route_headline - unrelated: generate_headline.yaml

两个要点：

第一条边是一个线性序列：START → process_input → generate_headline → evaluate_headline → route_headline。列表中混用了三种节点表示法（函数引用.agent.process_input、外部 Agent 文件generate_headline.yaml），这正是下面第四节三种特殊语法的综合演示。
第二条边是回路边：.agent.route_headline之后的unrelated: generate_headline.yaml是一个路由字典——当route_headline产出的route值等于unrelated时，流转到generate_headline.yaml节点；其他路由值（tech-related）没有出边，流程自然终止。

3.2 生成节点：generate_headline.yaml

contributing/samples/workflows/loop_config/generate_headline.yaml：

agent_class: LlmAgent name: generate_headline instruction: | Write a headline about the topic "{topic}". If feedback is provided, take it into account. The feedback: {feedback?}

agent_class: LlmAgent告诉 loader 该文件描述的是一个LlmAgent实例；
指令中的{topic}来自process_input写入状态的键；
{feedback?}中的?是可选插值语法：首轮执行时 state 中还没有feedback键，使用可选占位符可避免缺失报错，第二轮起则自动带入evaluate_headline写入的反馈内容。

3.3 评估节点：evaluate_headline.yaml

contributing/samples/workflows/loop_config/evaluate_headline.yaml：

agent_class: LlmAgent name: evaluate_headline instruction: | Grade whether the headline is related to technology or software engineering. output_schema: name: loop_config.agent.Feedback output_key: feedback

output_schema采用name条目携带全限定类名loop_config.agent.Feedback，loader 会动态导入该对象——这是 README 所述 “Code References” 语法的实际用例；
output_key: feedback指定评估结果写入会话状态中的键名，正是回路下一轮generate_headline指令里{feedback?}读取的键。

3.4 Python 侧：agent.py

contributing/samples/workflows/loop_config/agent.py 定义了图节点之外的全部 Python 代码：

class Feedback(BaseModel): grade: Literal["tech-related", "unrelated"] = Field( description=( "Decide if the headline is related to technology or software" " engineering." ) ) feedback: str = Field( description=( "If the headline is unrelated to technology, provide feedback on how" " to make it more tech-focused." ) ) def process_input(node_input: str): """Puts user input in the state.""" return Event(state={"topic": node_input}) def route_headline(node_input: Feedback): return Event(route=node_input.grade)

Feedback：结构化输出模型，grade限定为两个取值，正好对应路由字典的两个键之一；
process_input：接收节点输入（用户消息文本），通过Event(state={...})把主题写入状态；
route_headline：读取上一节点写入的feedback（反序列化为Feedback），通过Event(route=...)声明路由值，驱动Workflow按出边字典分发。

与 Python 版样本对照：contributing/samples/workflows/loop/agent.py 中同样的图是用Workflow(name="root_agent", edges=[(...), (route_headline, {"unrelated": generate_headline})])字面量写出的，两者的edges结构一一对应，可交叉验证 YAML 语义。

四、README 定义的三处特殊语法

README（contributing/samples/workflows/loop_config/README.md）明确归纳了本样本用到的三种动态解析语法：

4.1 代码引用（Code References）

持有 Python 对象的字段（如evaluate_headline.yaml中的output_schema）可以写成带name条目的结构，name是该对象的全限定名，loader 负责导入：

解析基于sys.path，其中包含 agent 目录所在目录；
示例：name: loop_config.agent.Feedback解析为本目录下agent.py中的FeedbackPydantic 模型。

4.2 边内函数引用（Function References in Edges）

边列表中的字符串如果不以.yaml结尾、也不是'START'，就会被当作函数引用处理：

以.开头时，相对当前 agent 目录的 Python 包路径解析。示例：.agent.process_input解析为agent.py中的process_input函数（即先展开为loop_config.agent.process_input再导入）；
loader 会自动用函数名作为节点名创建FunctionNode（本例中节点名即process_input）；
不带前导点的引用（如.agent.route_headline展开后的包路径形式）走同样的sys.path导入逻辑。

4.3 外部 Agent 文件（External Agent Files）

Agent 可以定义在独立 YAML 文件中，边列表中直接用文件名引用：

示例：generate_headline.yaml即引用该文件定义的LlmAgent；
实例复用：mapper 按字符串值缓存已解析节点，因此在多条边中重复使用同一文件名（回路场景必然如此：generate_headline.yaml同时出现在首条边和unrelated出边中）会命中同一个节点实例，从而正确保持图结构——这是回路能够成立的关键机制。

五、源码级原理：loader 如何把字符串解析成节点

上述三种语法的实现集中在 src/google/adk/agents/config_agent_utils.py 的_resolve_node_like方法（约 L334–L394），可以直接印证 README 的描述：

START哨兵：字符串"START"直接映射为Workflow的START常量（L340–L341）；
缓存优先：任何字符串/内联节点在解析前都会先查_resolved_nodes_cache（L349–L350），命中即返回同一对象——这正是“同一文件名在多边中复用同一实例”的源码依据（L355、L391–L393 写入缓存，且同时以原始字符串和节点name两个键登记）；
.yaml/.yml后缀：构造AgentRefConfig(config_path=...)并调用resolve_agent_reference加载外部 Agent 文件（L352–L358）；
裸单词的限制：不含.的字符串被视为已存在的节点名；若缓存里没有，会抛出ValueError提示“节点必须由更早的边先定义”，即不支持前向引用（L360–L368）。这意味着写 edges 时节点必须先定义后引用；
前导点相对路径：以.开头的引用会取当前 agent 目录名拼成包路径（L371–L374），再经resolve_fully_qualified_name导入；若结果是可调用对象（非类），则用最后一段作为节点名创建FunctionNode（L376–L379），与 README “自动创建 FunctionNode” 的描述一致；
非法引用快速失败：解析结果既非可调用也非节点时立即抛出带引用名的ValueError（L386–L389），避免问题延迟到图校验阶段才暴露。

由此也能解释 README 的运行前提：因为代码引用是对sys.path解析的，必须从 agent 文件夹所在目录（即contributing/samples/workflows/）作为工作目录运行，与 ADK CLI 加载 agent 的约定一致。

对应的图运行时实现位于 src/google/adk/workflow/_workflow.py（Workflow类与edges字段）及 src/google/adk/workflow/_function_node.py（FunctionNode），edges 列表中的元组/字典形式与 YAML 中- - START / .agent.process_input / ...的结构完全同构。

六、用黄金事件序列验证回路真的发生了

tests/目录下的 contributing/samples/workflows/loop_config/tests/flower.json 录入了主题flower（非科技主题，注定触发回路）的完整事件流，可逐步对照验证：

e-2（process_input@1）：stateDelta.topic = "flower"——主题写入状态；
e-3（generate_headline@1）：首轮产出"Petal Power: The Timeless Allure of Flowers"；
e-4（evaluate_headline@1）：模型输出 JSON{"grade": "unrelated", "feedback": "..."}，并写入stateDelta.feedback——注意state中的feedback值正是下一轮生成的输入；
e-5（route_headline@1）：actions.route = "unrelated"——命中unrelated: generate_headline.yaml出边，回到生成节点；
e-6（generate_headline@2）：节点名后缀@2表明同一节点第二次执行，新标题AI-Powered Petals: ...明显吸收了反馈；
e-7（evaluate_headline@2）：grade = "tech-related"；
e-8（route_headline@2）：route = "tech-related"，无对应出边，调用结束。

最终state同时包含topic: "flower"与feedback.grade: "tech-related"，完整还原了“生成 → 评估 → 反馈 → 再生成 → 通过”的回路闭环。tests/computer.json提供了另一组主题的事件记录可作对照。

七、执行方式与运行前提

运行目录：从contributing/samples/workflows/目录执行 ADK 命令（README 原话：代码引用相对sys.path解析，须“从持有 agent 文件夹的目录运行，与 CLI 的约定一致”）。由于root_agent.yaml声明name: root_agent且文件位于loop_config/目录，按 ADK CLI 的 agent 目录约定，该目录即一个可加载的 agent 项目；
输入示例：README 给出的两个测试输入为Python programming（科技主题，预期一轮通过）与Baking cookies（非科技主题，预期触发unrelated回路）；
前提与限制：
- 模型访问由LlmAgent默认配置决定，本样本自身未固定模型与 API 配置；
- 回路边只声明了unrelated路由，tech-related无出边即终止，因此不会出现死循环；若自行扩展路由字典，需自行保证终止性；
- 节点必须先定义后引用（前向引用会直接报错），新增节点时应保持“定义在前的边”出现在引用它的边之前。

八、小结：配置式 Workflow 回路的三个可复用模式

对照 contributing/samples/workflows/loop_config/ 全套文件，可以把该样本抽象为三个可直接复用的模式：

图结构完全 YAML 化：agent_class: Workflow+edges即可表达含循环的执行图，LLM 节点拆成独立.yaml文件按文件名引用，Python 侧只保留路由函数与结构化模型；
动态解析三语法：name全限定名（任意 Python 对象，如output_schema）、前导点函数引用（自动转FunctionNode）、外部 Agent 文件名（带实例级缓存保证图结构正确）；
用状态键串联回路数据：output_key（feedback）写入状态，下一轮指令中的{feedback?}可选插值读出，配合路由函数读取output_key的值决定走向——“状态即回路记忆”，无需任何额外会话管理代码。