跳到主要内容

工作流模式:把「building effective agents」做成可组合 agent

30 秒导读: Anthropic 那篇《Building Effective Agents》列了几种「编排 LLM」的经典套路——串联、并行、路由、编排、评审迭代。fast-agent 把每一种都写成一个普通 agent 类(几乎都是 LlmAgent 的子类),它们对外和一个单体 agent 长得一模一样,所以可以像乐高一样互相嵌套。本章逐个讲这七种模式各自解决什么问题、内部怎么转,并给出「模式 → 意图 → 类」对照表和一张组合关系图。


1. 这是什么(零基础也能懂)

一句话定义: 工作流模式 = 用固定的编排逻辑把多个 agent(或对同一个 agent 的多次采样)串起来,以换取更高的可靠性、专业分工或吞吐。

为什么需要它。 一个「裸」LLM agent 就是一个「读消息 → 调工具 → 再读 → 输出」的循环(那条循环见 工具循环引擎)。但单条循环有天花板:

  • 一个 prompt 里塞太多职责,模型会顾此失彼 → 想分工;
  • 复杂任务需要先规划再执行 → 想分解;
  • 便宜模型偶尔出错,但错误可以靠「多数投票」压下去 → 想冗余纠错;
  • 几个子任务互不依赖 → 想并行加速。

这些诉求早就被总结成了几种可复用的「编排套路」。fast-agent 的做法不是发明新东西,而是把这些套路各写成一个 agent 类,放在 src/fast_agent/agents/workflow/ 目录下。

一句话直觉: 把每种编排想象成一个「工头 agent」。工头自己不干具体活,而是按一套固定流程去指挥手下的 agent 干活,再把结果整理成一句话交出去。关键是——工头本身也是一个 agent,所以工头手下的「员工」可以是另一个工头。

用起来什么样。 声明式前端(见 声明式前端)用装饰器登记这些工头,写法和登记一个普通 agent 几乎一样:

# 示意,非源码:装饰器登记入口详见 01 章
@fast.chain(name="post", sequence=["draft", "polish", "seo"]) # 串联三个 agent
async def post(): ...

@fast.parallel(name="review", fan_out=["sec", "perf"], fan_in="summarize") # 并行分析再汇总
async def review(): ...

本节不深入实现。目标:知道「工作流模式 = 把编排套路做成 agent 类」。


2. 顶层全景(它大概怎么转)

2.1 一切的地基:模式都是 LlmAgent 子类

理解本章只需抓住一个事实:每种工作流模式都继承 LlmAgent,只重写一个方法 generate_impl(以及结构化输出的 structured_impl)。

外部调用者永远只碰公开的 generate()。它在 llm_decorator.py:540 里做两件事——把各种输入统一成消息列表,然后转交给 generate_impl:

# 真实源码 src/fast_agent/agents/llm_decorator.py:568-574 · generate
multipart_messages = normalize_to_extended_list(messages) # 归一化输入
...
with self._tracer.start_as_current_span(f"Agent: '{self._name}' generate"):
return await self.generate_impl(multipart_messages, final_request_params, tools)

这是经典的模板方法:generate() 是稳定外壳,generate_impl() 是可被子类替换的「内核」。一个裸 agent 的 generate_impl 就是跑工具循环;一个 ChainAgentgenerate_impl 则是「依次调用手下 agent」。

由此推出本章的核心结论:

  • 工作流对外满足和普通 agent 完全相同的接口(AgentProtocol);
  • 所以任何需要「一个 agent」的位置,都能塞进「一个工作流」;
  • 所以工作流可以任意嵌套——router 的某个专家可以是一条 chain,chain 的某一环可以是一个 evaluator-optimizer。

2.2 七种模式一览

fast-agent 用一个枚举 AgentType(agent_types.py:24)给每种模式挂了身份牌:

模式(白话)解决的意图类 · 文件AgentType
串联 chain把多个 agent 首尾相接,前一个的输出喂给后一个ChainAgent · chain_agent.py:35CHAIN
并行 parallel多个 agent 同时看同一输入(fan-out),再由一个 agent 汇总(fan-in)ParallelAgent · parallel_agent.py:19PARALLEL
路由 router按输入内容,选唯一最合适的专家去处理RouterAgent · router_agent.py:66ROUTER
编排 / 迭代规划把大目标分解成步骤和子任务,派给 worker,再汇总IterativePlanner · iterative_planner.py:160ITERATIVE_PLANNER
评审迭代 evaluator-optimizer「生成 → 评审 → 按反馈重做」直到达标EvaluatorOptimizerAgent · evaluator_optimizer.py:59EVALUATOR_OPTIMIZER
MAKER 投票对同一步反复采样,用「领先 k 票」共识纠错MakerAgent · maker_agent.py:90MAKER
agents-as-tools把子 agent 暴露成可并行调用的工具,让父 LLM 自己决定调谁AgentsAsToolsAgent · agents_as_tools_agent.py:397复用 MCP 工具面

前六种映射到 Anthropic《Building Effective Agents》的五个模式(chain=prompt chaining、parallel=parallelization、router=routing、orchestrator=orchestrator-workers、evaluator-optimizer),MAKER 对应 README 提到的同名机制与论文 arXiv:2511.09030。第七种是 OpenAI Agents SDK 风格的「agents as tools」,可视作把路由/并行/编排合一

2.3 两种「工头」的分野

七种模式其实分两大类,理解这条线能省很多困惑:

工头怎么决定「下一步派谁」?
|
+-----------------+------------------+
| |
代码写死的固定流程 让 LLM 临场决定
(deterministic) (LLM-driven)
| |
chain 顺序固定 router LLM 选一个专家
parallel 全都跑 orchestrator LLM 生成计划
evaluator 循环到达标 agents-as-tools LLM 发工具调用
maker 采样到 k 票领先
  • 左边(流程写死):编排逻辑是 Python 代码里的 for/while,不问 LLM「接下来干嘛」。
  • 右边(LLM 驱动):工头自己先跑一次 LLM,让模型产出「派给谁/下一步是什么」的结构化决策,再照办。

下面逐个拆。


3. 核心原理(逐个机制,由浅入深)

3.1 chain — 顺序串联

要解决的小问题: 一件事有天然的先后工序(先起草、再润色、再配 SEO),每一环最好由专精的 agent 做。

思路: 把 agent 排成一队,第一个吃用户输入,之后每个吃上一个的输出,最后一个的输出就是结果。

数据流(默认非累积模式):

user ──▶ agents[0] ──输出──▶ agents[1] ──输出──▶ agents[2] ──▶ 结果
(draft) (polish) (seo)

真实实现: ChainAgent.generate_impl(chain_agent.py:71)。第一个 agent 单独处理原始消息,其后用上一环的 content 拼成新的 user 消息传给下一环:

# 真实源码 src/fast_agent/agents/workflow/chain_agent.py:115-116
next_message = Prompt.user(*response.content)
response = await agent.generate([next_message], forward_params)

关键细节 — cumulative 开关。 构造参数 cumulative(chain_agent.py:69)切换两种语义:

模式每个 agent 看到什么输出
非累积(默认)只看上一环的输出最后一环的原始响应
累积 cumulative=True原始请求 + 之前所有环的响应<fastagent:response agent=...> XML 标签包起来的全量拼接

累积模式的拼接见 chain_agent.py:156,给每段响应打上署名标签,便于下游或人类分辨哪句是谁说的。

3.2 parallel — fan-out + fan-in

要解决的小问题: 几个视角互不依赖(安全审计、性能审计、风格审计),与其串行,不如同时跑完再汇总。

思路: 两段式——fan-out(把同一输入广播给 N 个 agent 并发执行)+ fan-in(把 N 份结果交给一个「汇总 agent」整理成一份)。

┌──▶ fan_out[0] ─┐
user ─广播──▶ ├──▶ fan_out[1] ─┼─聚合文本──▶ fan_in_agent ──▶ 结果
└──▶ fan_out[2] ─┘
(asyncio.gather 真并发) (把各家结果拼成一个 prompt)

真实实现: 并发靠 asyncio.gather,在 _execute_fan_out(parallel_agent.py:188)里:

# 真实源码 src/fast_agent/agents/workflow/parallel_agent.py:207
return await asyncio.gather(*[_run_agent(agent) for agent in self.fan_out_agents])

聚合发生在 _build_fan_in_prompt(parallel_agent.py:209):它调 _format_responses(parallel_agent.py:82)把每个 agent 的输出包进 <fastagent:response agent="..."> 标签,再整个作为一条 user 消息喂给 fan_in_agent。构造参数 include_request(parallel_agent.py:57)决定要不要把原始请求也一并放进聚合 prompt。

巧妙处: fan-in 也是一个普通 agent。想要「只拼接不总结」,给它配一个直通的 agent 即可;想要「智能综述」,给它配一个会写作的 agent。编排代码一行不改。

3.3 router — LLM 选一个专家

要解决的小问题: 有一堆专家 agent(退款、技术支持、销售),来一条请求,应该交给最合适的那一个,别都跑。

思路: 让一次 LLM 调用做「分类」——读请求 + 读每个 agent 的能力卡片,吐出「选谁 + 置信度 + 理由」的结构化结果,然后把原请求原样转交给中选者。

结构化决策模型 RoutingResponse(router_agent.py:58):

# 真实源码 src/fast_agent/agents/workflow/router_agent.py:58-63
class RoutingResponse(BaseModel):
agent: str
confidence: str
reasoning: str | None = None

真实实现: 决策在 _route_request(router_agent.py:347),它把候选 agent 的 A2A 能力卡片塞进系统提示(_generate_routing_instruction,router_agent.py:153),再让 LLM 以 RoutingResponse 结构化输出选人:

# 真实源码 src/fast_agent/agents/workflow/router_agent.py:377-381
response, _ = await self._require_llm().structured(
[message], RoutingResponse, self._default_request_params,
)

选出后在 generate_impl(router_agent.py:200)里 self.agent_map[route.agent].generate(...) 把请求转交。

两个务实细节:

  • 单专家短路:只有一个候选时,跳过 LLM 直接返回该 agent(router_agent.py:364),省一次调用。
  • 未知名兜底:LLM 若返回一个不在册的 agent 名,记警告并返回 None(router_agent.py:387),而不是崩。

router 与 orchestrator 的差别:router 只选一个、不分解;orchestrator 会分解成多步多任务

3.4 orchestrator / iterative_planner — 分解-执行-汇总

要解决的小问题: 目标太大,一步做不完,需要先「想清楚下一步派谁做什么」,做完看结果再想下一步。

思路: 一个 while 循环。每轮让 planner LLM 看「目标 + 已完成进度」,产出下一个步骤(含若干可并行子任务);派给 worker agent 执行;把结果并回进度;直到 planner 宣布 is_complete 或触顶。这是迭代式规划(一次只出一步),而非一次性出完整计划。

数据模型(orchestrator_models.py)是这套的骨架:

模型 · 行含义
AgentTask · orchestrator_models.py:15一个子任务:描述 + 指定哪个 agent 来做
Step · orchestrator_models.py:23一个步骤:一句描述 + 一组可并行AgentTask
PlanningStep · orchestrator_models.py:34迭代模式下 planner 每轮产出的「下一步」,多带一个 is_complete
Plan · orchestrator_models.py:40若干 Step 的序列 + is_complete
PlanResult · orchestrator_models.py:76整个计划的执行结果:目标 + 各步结果 + 是否完成 / 是否触顶

主循环 _execute_plan(iterative_planner.py:315)。伪代码把它讲清楚:

# 示意,非源码:提炼自 iterative_planner.py:336-393
while not objective_met and not terminate_plan:
next_step = await self._get_next_step(objective, plan_result, params) # planner 出下一步
if next_step.is_complete: # planner 说完事了
objective_met = True; break
if self._validate_agent_names(...): # 计划里点名了不存在的 agent → 终止
break
step_result = await self._execute_step(step, plan_result, params) # 执行这一步
plan_result.add_step_result(step_result)
if 触及 plan_iterations 预算: # 迭代预算用完 → 终止
break
# 最后再让自己的 LLM 把整个 PlanResult 综述成一段回答

步骤内的并发策略(_execute_step,iterative_planner.py:401)有个巧妙取舍:按 agent 分组——不同 agent 之间并行,同一 agent 的多个任务串行(保住它自己的对话历史连续):

# 真实源码 src/fast_agent/agents/workflow/iterative_planner.py:490-491
all_results = await asyncio.gather(*agent_futures) # 不同 agent 并行
task_results = [r for agent_results in all_results for r in agent_results]

终止条件(见 _execute_plan):planner 主动 is_complete、点到未知 agent、plan_iterations 预算耗尽(iterative_planner.py:385)、或生成下一步失败。收尾时用 PLAN_RESULT_TEMPLATE(iterative_planner.py:137)让 planner 自己的 LLM 把 PlanResult 综述成最终答案。

3.5 evaluator-optimizer — 生成-评审-迭代

要解决的小问题: 一次生成质量不稳,但如果有个「审稿人」给出具体反馈,让作者照着改,几轮后能显著变好。

思路: 两个 agent 打配合——generator_agent 写,evaluator_agent 打分并给反馈;循环「评审 → 若不达标则带反馈重写」,直到达到质量门槛或用完次数。

质量分级 QualityRating(evaluator_optimizer.py:30)是个有序枚举 POOR < FAIR < GOOD < EXCELLENT;评审输出结构 EvaluationResult(evaluator_optimizer.py:48)含 rating / feedback / needs_improvement / focus_areas

循环图:

generator 首稿


evaluator 评分 ──达标 or 不需改进?──是──▶ 返回「历史最佳」
│否

达到 max_refinements?──是──▶ 返回「历史最佳」
│否

generator 带反馈重写 ──▶ 回到 evaluator 评分

真实实现: 循环在 generate_impl(evaluator_optimizer.py:111)。三条退出闸门:评审说 needs_improvement=False(evaluator_optimizer.py:213)、评分已达 min_rating(evaluator_optimizer.py:219)、或改够 max_refinements 次(evaluator_optimizer.py:226)。

巧妙处 — 留住历史最佳。 迭代不保证单调变好,所以它按数值分持续记录 best_response(evaluator_optimizer.py:204),最终返回见过的最好版本,而非最后一版:

# 真实源码 src/fast_agent/agents/workflow/evaluator_optimizer.py:204-209
if QUALITY_RATING_VALUES[evaluation_result.rating] > QUALITY_RATING_VALUES[best_rating]:
best_rating = evaluation_result.rating
best_response = response

3.6 maker — first-to-ahead-by-k 投票纠错

要解决的小问题: 便宜/小模型单次成功率不够高,但如果对同一步反复采样、多数投票,错误率会随投票余量指数下降,而成本只对数增长。这样便宜模型也能跑「百万步零错误」的任务。

出处: 论文《Solving a Million-Step LLM Task with Zero Errors》(arXiv:2511.09030),对应 README 里的 MAKER(Massively decomposed Agentic processes with K-voting Error Reduction)。文件头 maker_agent.py:1-18 记了三个概念:最大化分解(MAD)、first-to-ahead-by-k 投票、红旗剔除(red-flagging)。

思路: 反复采样同一个 worker_agent,给相同的响应「计票」;谁比第二名领先 k 票,谁就赢——不是「采样固定次数取多数」,而是「谁先拉开 k 票差谁先胜出」,这样简单的题几票就收敛,难题才多采几次。

判胜逻辑 _check_winner(maker_agent.py:221):

# 真实源码 src/fast_agent/agents/workflow/maker_agent.py:236-243
sorted_items = sorted(votes.items(), key=lambda x: x[1], reverse=True)
leader_key, leader_votes = sorted_items[0]
runner_up_votes = sorted_items[1][1] if len(sorted_items) > 1 else 0
if leader_votes - runner_up_votes >= self.k: # 领先第二名 k 票即胜
return leader_key
return None

采样主循环 generate_impl(maker_agent.py:245):每采一次就先过红旗剔除 _is_red_flagged(maker_agent.py:194)——太长或不合自定义校验的响应直接丢(论文说这类响应更可能是错的);过关的响应先归一化再计票。

归一化 _normalize_response(maker_agent.py:165)决定「什么算同一个答案」,由 MatchStrategy(maker_agent.py:58)控制:

策略何时算同票
EXACT逐字相同
NORMALIZED忽略大小写和空白差异
STRUCTURED解析为 JSON 后结构相同(键排序)

触顶兜底: 采到 max_samples 还没拉开 k 票,就退化为多数票(plurality),并在 MakerResult(maker_agent.py:73)里把 converged=False 记下来,便于事后分析(maker_agent.py:339)。若所有样本都被红旗剔除,则直接报错提示放宽红旗条件。

3.7 agents-as-tools — 把子 agent 当工具

要解决的小问题: 前面几种都是「代码或 planner 决定派谁」。但现代 LLM 本来就擅长 function calling——干脆把每个子 agent 直接做成一个工具,让父 LLM 在正常的工具循环里自己决定调谁、要不要并行调多个。

定位: AgentsAsToolsAgent(agents_as_tools_agent.py:397)是唯一不直接继承 LlmAgentgenerate_impl 的模式——它继承 McpAgent(见 Agent 类栈MCP 集成),把「子 agent」和「MCP 工具」揉进同一张工具面。因此它某种意义上把路由 / 并行 / 编排合一:选谁 = 父 LLM 发哪个工具调用,并行 = 父 LLM 一次发多个,编排 = 父 LLM 的多轮工具循环。

三步机制:

  1. 登记为工具 — 每个子 agent 映射成合成工具名 agent__{child_name}(_make_tool_name,agents_as_tools_agent.py:447)。

  2. 工具发现合并(list_tools,agents_as_tools_agent.py:542)— 先取基类的 MCP/本地工具,再把不冲突的子 agent 工具追加上去,父 LLM 看到的是一张混合清单:

# 真实源码 src/fast_agent/agents/workflow/agents_as_tools_agent.py:545-565
base = await super().list_tools() # MCP + 本地工具
tools = list(base.tools)
for tool_name, agent in self._child_agents.items():
if tool_name in existing_names: continue
tools.append(Tool(name=tool_name, description=..., inputSchema=...))
  1. 并行执行(run_tools,agents_as_tools_agent.py:1609)— 把父 LLM 这一轮的工具调用分区:是子 agent 的走并行专用路径 _run_child_tools(agents_as_tools_agent.py:1660);其余 MCP/本地工具下沉给基类 McpAgent.run_tools(agents_as_tools_agent.py:1646);两边结果合并成一条消息返回。
# 真实源码 src/fast_agent/agents/workflow/agents_as_tools_agent.py:1624-1628
if tool_name not in base_tool_names and self._resolve_child_agent(tool_name):
child_ids.append(correlation_id) # 归为「子 agent 工具」
if not child_ids:
return await super().run_tools(...) # 全是普通工具,交给基类

巧妙处 — 每次调用一个「分身」。 为了让同一子 agent 被父 LLM 并行调用多次时互不串味,run_tools 会为每次调用克隆出一个短命的「detached instance」,各带独立的 LLM + MCP 栈和后缀名 Child[i],跑完把用量并回模板 agent(设计说明见 agents_as_tools_agent.py:130-152)。行为旋钮集中在 AgentsAsToolsOptions(agents_as_tools_agent.py:369):max_parallel(并发上限)、child_timeout_sec(单个子 agent 超时)、history_source / history_merge_target(子 agent 起始/回并历史)。


4. 组合关系:它们如何互相嵌套

因为每个工作流都满足同一个 agent 接口,它们能自由套娃。下图给一个真实可行的组合(自上而下就是「谁包着谁」):

router「客服总台」 ← 先按请求类型选一条线
├─ 专家A: chain「退款处理」 ← 选中后走一条固定工序
│ ├─ verify-agent
│ └─ evaluator-optimizer「话术打磨」 ← 工序里的一环又是生成-评审循环
│ ├─ generator
│ └─ evaluator
└─ 专家B: iterative_planner「疑难工单」 ← 另一条线是分解-执行
├─ worker: maker「关键判断」 ← 某个 worker 用投票纠错保可靠
│ └─ cheap-model-agent
└─ worker: parallel「多源核查」 ← 另一个 worker 是并行 fan-out/fan-in
├─ fan_out: kb-agent
├─ fan_out: web-agent
└─ fan_in: summarizer

怎么读这张图: 每个非叶节点都是一个「工头」工作流,它的孩子既可以是裸 agent(叶子),也可以是另一个工头。之所以能这样,是因为 §2.1 那条:工头对外就是个 generate(),填进任何「需要一个 agent」的槽位都合法。

模式 → 意图 → 类 速查(本章核心对照表):

想要的效果选哪个模式类 · 文件:行
固定工序、逐环加工chainChainAgent · chain_agent.py:35
多视角并发 + 汇总parallelParallelAgent · parallel_agent.py:19
只交给最合适的一个专家routerRouterAgent · router_agent.py:66
大目标分解成多步多任务orchestratorIterativePlanner · iterative_planner.py:160
生成后反复评审改进evaluator-optimizerEvaluatorOptimizerAgent · evaluator_optimizer.py:59
便宜模型靠投票纠错makerMakerAgent · maker_agent.py:90
让父 LLM 自主调用/并行子 agentagents-as-toolsAgentsAsToolsAgent · agents_as_tools_agent.py:397

5. 巧妙之处(可借鉴的技术)

  • 模板方法把「编排」降维成「重写一个函数」。 只要重写 generate_impl,就得到一个可嵌套的工头,无需另立一套编排框架(llm_decorator.py:574 分发到子类)。这是全章能成立的支点。

  • fan-in / router / planner 的「决策」都用结构化输出。 router 的 RoutingResponse(router_agent.py:58)、planner 的 PlanningStep(orchestrator_models.py:34)、evaluator 的 EvaluationResult(evaluator_optimizer.py:48)——把「让 LLM 做决定」变成「让 LLM 填一个 Pydantic 模型」,决策因此可校验、可兜底。

  • orchestrator 的分组并发:不同 agent 并行、同一 agent 串行(iterative_planner.py:485-491),在吞吐与「保住单个 agent 历史连续」之间取了个漂亮平衡。

  • evaluator-optimizer 记「历史最佳」而非「最后一版」(evaluator_optimizer.py:204),防止迭代把结果改坏。

  • maker 的红旗剔除(maker_agent.py:194):投票前先扔掉「疑似跑偏」的样本,用极低成本抬高有效成功率。

  • agents-as-tools 的每调用克隆(agents_as_tools_agent.py:130-152):用短命分身替代「改名/共享状态」的 hack,让同一子 agent 的并行调用天然隔离。


6. 边界与局限

  • 左派模式不「思考」路径。 chain / parallel / maker 的流程是写死的:chain 不会跳过某环,parallel 永远全跑,maker 只会投票不会换策略。要「临场决定」得用右派(router / orchestrator / agents-as-tools)。

  • 迭代规划是「一次一步」,不是一次性出完整 DAG(_execute_plan 循环 iterative_planner.py:336);代码注释也留了 # this will only be one for iterative(iterative_planner.py:374),说明当前每轮只执行一个 step。

  • maker 只在答案「可比较」时有效。 它靠归一化后计票,适合有收敛答案的步骤;开放式长文本很难有两次采样落到同一票桶,容易一路走到 max_samples 退化成多数票(maker_agent.py:326)。

  • 成本随嵌套层级相乘。 每层工头都可能触发多次 LLM 调用;router 里套 parallel 里套 evaluator,调用数是各层的乘积,要留意预算。

  • agents-as-tools 把控制权交给父 LLM。 好处是灵活,代价是流程不可预测——模型可能漏调、错调或过度并行,需要靠 max_parallel(agents_as_tools_agent.py:369)等旋钮兜底。


7. 横向对比

  • 与本库其它 agent 框架:多数框架把「编排」做成独立于 agent 的 graph/DAG 引擎;fast-agent 的取舍是不引入第二套抽象——编排本身就是 agent,复用同一条 generate()/工具循环,因而组合成本极低。代价是没有「显式计划图」这种一眼可视化的中间产物(orchestrator 靠收尾时让 LLM 画 Mermaid 补足,iterative_planner.py:155)。

  • maker vs evaluator-optimizer:两者都追求「更可靠的输出」,但路子相反——maker 靠冗余采样 + 投票(适合有收敛答案、用便宜模型),evaluator-optimizer 靠单路生成 + 反馈重写(适合开放式创作、需要具体改进意见)。

  • 装饰器登记入口(@fast.chain / @fast.parallel / @fast.router / @fast.orchestrator / @fast.evaluator_optimizer 等)见 声明式前端,本章不重复;各模式最终实例化的 LlmAgent 继承链见 Agent 类栈


8. 代码地图(导航索引)

主题文件路径符号名
模式身份枚举src/fast_agent/agents/agent_types.pyAgentType
模板方法分发点src/fast_agent/agents/llm_decorator.pygenerategenerate_impl
串联src/fast_agent/agents/workflow/chain_agent.pyChainAgent, _chain_messages_with_responses
并行 fan-out/fan-insrc/fast_agent/agents/workflow/parallel_agent.pyParallelAgent, _execute_fan_out, _format_responses
路由src/fast_agent/agents/workflow/router_agent.pyRouterAgent, RoutingResponse, _route_request
迭代规划循环src/fast_agent/agents/workflow/iterative_planner.pyIterativePlanner, _execute_plan, _execute_step, _get_next_step
规划数据模型src/fast_agent/agents/workflow/orchestrator_models.pyPlan, Step, AgentTask, PlanningStep, PlanResult
评审迭代src/fast_agent/agents/workflow/evaluator_optimizer.pyEvaluatorOptimizerAgent, QualityRating, EvaluationResult
MAKER 投票src/fast_agent/agents/workflow/maker_agent.pyMakerAgent, MatchStrategy, MakerResult, _check_winner, _is_red_flagged
agents-as-toolssrc/fast_agent/agents/workflow/agents_as_tools_agent.pyAgentsAsToolsAgent, AgentsAsToolsOptions, list_tools, run_tools, _run_child_tools
子请求参数下发src/fast_agent/agents/workflow/request_params.pychild_request_params