跳到主要内容

自动构建:从一句话目标到可运行的多智能体工作流

30 秒导读: 你给一句话目标(比如"生成一个能在浏览器里玩的俄罗斯方块"),WorkFlowGenerator 就自动帮你把它拆成一串带依赖的子任务、给每个子任务配好一个专属 agent,最后拼成第 3 章 讲的那种 WorkFlowGraph——直接能跑。本章讲清这条"从 goal 到图"的流水线怎么走通。

本章只讲怎么生成这张图。生成之后如何被评测、如何被优化器反复打磨,是 自演化引擎的事,这里不碰。

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

一句话定义​

WorkFlowGenerator 是 EvoAgentX 的自动建图器:输入一个高层目标字符串,输出一张 可执行的多智能体工作流图。中间的"拆任务""配 agent"全交给大模型(LLM)来做。

它解决什么问题​

平时你要搭一个多 agent 流程,得自己动手:

  • 想清楚这活儿分几步、谁先谁后;
  • 每一步写一个 agent,定义它的输入、输出、prompt;
  • 把这些步骤的输入输出对上,连成一张图。

这套手工活又慢又容易漏。WorkFlowGenerator 把它自动化:你只说"我要什么",它替你把 "怎么分工、谁配合谁"全排好。

用起来什么样​

来自 examples/workflow/workflow_demo.py:23-24 的真实用法——两行就出一张图:

wf_generator = WorkFlowGenerator(llm=llm)                       # 只需给它一个 LLM
workflow_graph = wf_generator.generate_workflow(               # 一句话目标进去
    goal="Generate html code for the Tetris game ..."          # 一张可执行图出来
)

拿到 workflow_graph 后,交给 AgentManager + WorkFlow 就能真正跑 (workflow_demo.py:33-37)。这一章只负责把 workflow_graph 造出来。

一句话直觉​

把它想成一个自动项目经理:你丢给它一个项目目标,它先写出任务分解(WBS), 再给每个任务招一个"专才"(agent),最后画出任务甘特图上的依赖箭头。

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

三步走的流水线​

WorkFlowGenerator 编排三个组件依次干活,核心就是一条从左到右的流水线:

   goal (一句话字符串)
        │
        ▼
 ┌──────────────────┐   "把大目标拆成带依赖的子任务"
 │ ① TaskPlanner    │   agents/task_planner.py
 │   任务规划器      │──► 产出 子任务列表 (List[WorkFlowNode])
 └──────────────────┘        每个子任务带 name/description/inputs/outputs
        │
        ▼
 ┌──────────────────┐   "按输入输出名对暗号,自动连边"
 │ ② build_workflow │   workflow_generator.py:201
 │   组装成图        │──► 产出 WorkFlowGraph(节点+边,暂无 agent)
 └──────────────────┘
        │
        ▼
 ┌──────────────────┐   "给每个子任务现造 / 复用一个 agent"
 │ ③ AgentGenerator │   agents/agent_generator.py
 │   智能体生成器    │──► 把 agent 规格塞回每个节点 node.set_agents(...)
 └──────────────────┘
        │
        ▼
   可执行的 WorkFlowGraph
   (第 3 章那种图,每个节点都配好了 agent)

怎么读这张图:从上到下是时间顺序,①②③ 依次执行,数据一路往下流。①出"任务", ②把任务连成"图",③给图上每个任务"配人"。

三个部件各司其职​

部件干什么在哪个文件
TaskPlanner把 goal 拆成带依赖的子任务列表agents/task_planner.py:6
AgentGenerator为每个子任务选已有 agent 或现造新 agentagents/agent_generator.py:6
WorkFlowReviewer对生成的图给反馈以迭代改进(当前是占位,未实现)agents/workflow_reviewer.py:6
WorkFlowGenerator编排上面三者、把结果拼成 WorkFlowGraphworkflow/workflow_generator.py:20

注意最后一栏:图里画的 WorkFlowReviewer 在当前 commit 其实是个占位实现, 后面 §6 会诚实交代——生成流程实际只走了 ①②③ 里的规划和生成两步。

主线走一遍(不进代码)​

看 generate_workflow 的骨架(workflow_generator.py:100),整条主线就四步:

  1. 规划:调 generate_plan,拿到子任务列表(workflow_generator.py:109-116);
  2. 建图:调 build_workflow_from_plan,把子任务连成图(:120-126);
  3. 校验:workflow._validate_workflow_structure() 确认图结构合法(:131);
  4. 配 agent:调 generate_agents,给每个节点塞进 agent 规格(:136-143),再校验每个节点都有 agent(:149-151)。

每一步都被包在 _execute_with_retry(workflow_generator.py:70)里——LLM 输出不稳定, 失败了就指数退避重试,重试次数由 retry 参数控制。

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

3.1 第一步:把 goal 拆成带依赖的子任务(TaskPlanner)​

它要解决的小问题: 一句话目标太笼统,没法直接执行。得先切成一个个"目标明确、 自成一体"的小任务,并且理清谁依赖谁。

思路/直觉: 这活儿本身就是 LLM 擅长的——给它目标 + 一套拆解原则(简单、模块化、 不冗余……),让它吐出结构化的子任务清单。关键设计是:子任务之间不靠显式"连线"来 表达依赖,而是靠"输入输出的名字对暗号"。

规划的产物类型是 TaskPlanningOutput,里面就一个字段(actions/task_planning.py:20-24):

class TaskPlanningOutput(ActionOutput):
    sub_tasks: List[WorkFlowNode] = Field(...)   # 直接就是一串 WorkFlowNode

也就是说,LLM 吐出的每个子任务,解析后直接就是一个 WorkFlowNode——图的节点在 规划这一步就已经成型了,只是还没连边、还没配 agent。

每个子任务长什么样。 prompt 强制 LLM 按固定 JSON 结构输出每个子任务 (prompts/task_planner.py:141-167),字段如下:

字段含义
name子任务唯一名字
description这个子任务要干什么
reason为什么需要它、如何贡献于总目标
inputs需要哪些输入(每个带 name/type/required/description)
outputs产出哪些输出

两条关键规则(依赖关系的命门)。 prompt 里两条硬约束决定了后面怎么连边:

  • 一个子任务的 inputs 只能来自用户的 goal 或它前置子任务的 outputs (prompts/task_planner.py:177);
  • 第一个子任务必须只有一个输入,名字就叫 goal(prompts/task_planner.py:180-190)。

这两条保证了子任务串成一条"输入输出首尾相接"的链——正是下一步自动连边的依据。

真实实现。 generate_plan 把 goal 交给 TaskPlanner 执行,LLM 以 JSON 模式解析出结构化结果:

# workflow_generator.py:155  generate_plan
message = task_planner.execute(
    action_name=task_planning_action_name,
    action_input_data={"goal": goal, "history": history, "suggestion": suggestion},
    return_msg_type=MessageType.REQUEST
)
return message.content          # 即 TaskPlanningOutput

底层是 TaskPlanning.execute(actions/task_planning.py:43)把 prompt 填好后调 llm.generate(..., parser=self.outputs_format, parse_mode="json")(:70-75)—— parse_mode="json" 让 LLM 输出被解析进 TaskPlanningOutput。这套"结构化输出解析" 的机制,见 地基章。

TaskPlanner 本身是个标准 Agent(agents/task_planner.py:6),它随身带一个 TaskPlanning action(task_planner.py:24),prompt/system_prompt 来自 prompts/task_planner.py。Agent 与 Action 的关系见 执行单元章。

3.2 第二步:按输入输出名"对暗号",自动连边(build_workflow_from_plan)​

它要解决的小问题: 有了一堆子任务节点,怎么知道该在哪两个之间画箭头?

思路/直觉: 上一步已经埋好线索——子任务的 inputs 只能来自前置子任务的 outputs。 所以只要 A 的某个输出名 出现在 B 的输入名里,就说明 B 依赖 A,连一条 A→B 的边。 不需要再问 LLM,纯靠名字匹配就能推出全部依赖。

图示:

  任务A                          任务B
  outputs: [problem_analysis] ───┐
                                 │ "problem_analysis" 这个名字
                                 │  同时出现在 A 的输出和 B 的输入里
                                 ▼
  inputs: [goal, problem_analysis]        ==>  连一条边  A ──► B

原理演示(示意,非源码):

# 对每一对不同的子任务,检查"A 的输出名"是否落进"B 的输入名"
for A in nodes:
    for B in nodes:
        if A is B:
            continue
        a_outputs = [p.name for p in A.outputs]     # A 产出的名字
        b_inputs  = [p.name for p in B.inputs]      # B 需要的名字
        if any(name in b_inputs for name in a_outputs):
            edges.append(edge(A.name, B.name))      # 有交集 => A 依赖被 B 用 => 连边

真实实现。 与上面几乎一一对应(workflow_generator.py:201-214):

def build_workflow_from_plan(self, goal, plan):
    nodes = plan.sub_tasks
    edges = []
    for node in nodes:
        for another_node in nodes:
            if node.name == another_node.name:
                continue
            node_output_params = [param.name for param in node.outputs]
            another_node_input_params = [param.name for param in another_node.inputs]
            if any([param in another_node_input_params for param in node_output_params]):
                edges.append(WorkFlowEdge(edge_tuple=(node.name, another_node.name)))
    return WorkFlowGraph(goal=goal, nodes=nodes, edges=edges)

边的表示是 WorkFlowEdge,构造时用一个 (source, target) 元组(workflow/workflow_graph.py:524、 :542)。节点 + 边一起塞进 WorkFlowGraph(goal=..., nodes=..., edges=...) (workflow/workflow_graph.py:585)——这就是第 3 章的那种图,已经有结构了,只差 agent。

关键细节: 建完图立刻校验结构 _validate_workflow_structure() (workflow/workflow_graph.py:847)——查输入输出重名、孤立节点、有没有初始节点等。 比如"初始节点"就是那些**只依赖 workflow 输入(即 goal)**的节点(workflow_graph.py:864-877), 正好对应 §3.1 里"第一个子任务只有 goal 输入"那条规则。

3.3 第三步:为每个子任务现造或复用一个 agent(AgentGenerator)​

它要解决的小问题: 图有了、任务清楚了,但每个任务得有"人"来干。这个"人"从哪来?

思路/直觉: 逐个子任务问 LLM 两件事——

  1. 现有的候选 agent 里有没有能直接用的?有就选它(记名字);
  2. 没有合适的,就照着子任务的输入输出现造一个新 agent,连 prompt 一起生成。

产出类型 AgentGenerationOutput 正好两个字段,对应这两条路(actions/agent_generation.py:120-123):

class AgentGenerationOutput(ActionOutput):
    selected_agents: List[str]              # 选中的已有 agent 的名字
    generated_agents: List[GeneratedAgent]  # 现造的新 agent 规格

逐个子任务地生成。 generate_agents 遍历图里每个节点,把"总目标 + 整张图描述 + 当前 子任务"喂给 AgentGenerator,拿回 agent 规格后塞回该节点(workflow_generator.py:180-197):

for subtask in workflow.nodes:                      # 逐个子任务
    subtask_desc = json.dumps(...)                  # 把这个子任务序列化成 JSON
    agents = agent_generator.execute(               # 问 LLM:选谁 / 造谁
        action_name=agent_generation_action_name,
        action_input_data={"goal": goal, "workflow": workflow_desc, "task": subtask_desc},
        ...
    ).content
    generated_agents = [a.to_dict(...) for a in agents.generated_agents]
    subtask.set_agents(agents=generated_agents)     # 把 agent 规格挂到节点上

一个诚实的实现细节: 代码里有句注释 # todo I only handle generated agents (workflow_generator.py:191)——当前只把新造的 generated_agents 挂回节点, selected_agents(复用已有 agent 的那条路)在这个编排里还没被真正接上。所以实践中, 每个子任务基本都是"现造"出来的。

新造的 agent 长什么样。 每个 GeneratedAgent 是一份完整规格 (actions/agent_generation.py:29-39):name、description、inputs、outputs、prompt、 可选的 tool_names。也就是说,LLM 不只是决定"要个 agent",还把这个 agent 的执行 prompt 都替你写好了。

一个巧妙的自我修复: GeneratedAgent 有个 validate_prompt 校验器 (actions/agent_generation.py:52-117),在 agent 生成后自动做三件事:

  • 检查所有输入变量都在 prompt 里被引用了(缺了就报错,:74-79);
  • 把 prompt 里 ### Instructions 段的输入占位符规整成 <input>{name}</input> 格式(:82-92);
  • 检查 ### Output Format 段声明的输出名和 agent 的 outputs 对得上;对不上时,用词相似度 找最接近的输出名自动替换(:99-115)。

这一步很关键:LLM 写 prompt 时经常把输出名写走样,这个校验器兜底修正,让生成的 agent 更可能真的能跑。

真实执行。 AgentGeneration.execute(actions/agent_generation.py:147)填好 prompt 后同样 调 llm.generate(..., parser=outputs_format, parse_mode="json")(:187-192)。它的 prompt (prompts/agent_generator.py:170)详细规定了 agent 该怎么选、怎么造、prompt 模板长什么样。

3.4 生成的 agent 规格,如何变成能跑的 agent​

上一步挂到节点上的只是字典规格,不是活的 agent 对象。真正"复活"发生在使用侧:

AgentManager.add_agents_from_workflow(agents/agent_manager.py:204)遍历图里每个节点的 node.agents,对每个规格调 add_agent(:174);字典规格最终经 create_agent_from_dict(utils/utils.py:222)被实例化成一个 CustomizeAgent(utils/utils.py:233)。

也就是说:"现造一个 agent"= 生成一份 CustomizeAgent 的配置字典。 CustomizeAgent 这个"用配置就能造出 agent"的机制,正是 执行单元章 的主角—— 自动构建之所以能"凭空造 agent",全靠它。

用法上就一行(examples/workflow/workflow_demo.py:33-34):

agent_manager = AgentManager()
agent_manager.add_agents_from_workflow(workflow_graph, llm_config=openrouter_config)

4. 端到端一条真实路径​

把一整条路串起来,以 demo 的目标为例(examples/workflow/workflow_demo.py:20):

goal = "Generate html code for the Tetris game ..."
   │
   │  ① TaskPlanner (task_planning.py) —— LLM 拆解
   ▼
sub_tasks = [
   {name: "requirements_analysis",  inputs:[goal],
                                     outputs:[analyzed_requirements]},
   {name: "code_generation",        inputs:[goal, analyzed_requirements],
                                     outputs:[html_code]},
   ...
]                                    # 示意结构,实际子任务由 LLM 决定
   │
   │  ② build_workflow_from_plan (workflow_generator.py:201) —— 名字对暗号连边
   ▼
WorkFlowGraph:
   requirements_analysis ──(analyzed_requirements)──► code_generation ──► ...
   │                        └ 因为该名字同时是前者输出、后者输入
   │
   │  ③ AgentGenerator (agent_generation.py) —— 逐节点现造 agent
   ▼
每个节点挂上一个 GeneratedAgent 规格(name/inputs/outputs/prompt/tool_names)
   │
   │  校验:每个节点都必须有 agent(workflow_generator.py:149-151)
   ▼
可执行 WorkFlowGraph  ──► AgentManager 把规格实例化成 CustomizeAgent ──► WorkFlow 执行

对照三种数据形态的转换:

阶段数据形态由谁产出
输入goal 字符串用户
规划后List[WorkFlowNode](子任务)TaskPlanner / TaskPlanningOutput
建图后WorkFlowGraph(节点+边,无 agent)build_workflow_from_plan
配 agent 后WorkFlowGraph(每节点带 agent 规格)AgentGenerator / generate_agents
使用侧每节点的规格 → CustomizeAgent 实例AgentManager

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

  • 依赖不靠 LLM 显式画,靠"输入输出名对暗号"推出来。 让 LLM 只管拆任务、约定好 "输入只能来自前置输出",连边就退化成一次纯名字匹配(workflow_generator.py:205-212)。 少问 LLM 一次,还更不容易出错。

  • 生成即校验,坏输出当场修。 GeneratedAgent.validate_prompt(actions/agent_generation.py:52) 在 agent 一生成就检查 prompt 里输入输出对不对,用相似度自动纠正走样的输出名(:99-115)—— 把 LLM 的不靠谱在源头兜住。

  • 每步都能重试、指数退避。 _execute_with_retry(workflow_generator.py:70)把规划、 建图、配 agent 每一步都包起来,失败按 2 ** cur_retries 秒退避重试(:95-97), 对付 LLM 偶发的解析失败。

  • "造 agent"复用了 CustomizeAgent 这块地基。 生成器只需吐一份配置字典,复活交给 create_agent_from_dict(utils/utils.py:222)——生成逻辑和 agent 运行时彻底解耦。

6. 边界与局限(诚实交代)​

  • WorkFlowReviewer 尚未接入。 类本身只是占位,execute 直接 raise NotImplementedError(agents/workflow_reviewer.py:12-13)。WorkFlowGenerator.init_module 里初始化它的代码是被注释掉的 TODO(workflow_generator.py:54-58)。所以"生成→给反馈→ 再改进"的 review 迭代当前并未真正运行。

  • num_turns 定义了却没用上。 字段声明为"精修迭代次数"(workflow_generator.py:40), 但 generate_workflow(:100)的主流程里没有围绕 num_turns 的循环——真正的多轮 精修依赖尚未实现的 reviewer。当前生成本质是一趟过(规划一次、建图一次、逐节点配 agent 一次)。

  • 只处理"新造"的 agent。 如 §3.3 所述,generate_agents 有 # todo I only handle generated agents(workflow_generator.py:191),selected_agents(复用已有 agent)在这条 编排链里还没被真正消费。

  • goal 太短会被直接拒。 少于 10 个字符就 raise ValueError(workflow_generator.py:102-103)—— 它要的是一句描述清楚的目标,不是一个词。

  • 强依赖 LLM 遵守输出契约。 整条流水线建立在 LLM 能吐出合规 JSON 之上;虽然有重试和 校验兜底,但子任务拆得好不好、agent prompt 写得对不对,最终取决于所用模型的能力。

7. 横向对比(与本组其它章的关系)​

想了解去哪一章
生成出来的图长什么样、怎么被调度执行03-workflow-execution.md
"现造一个 agent"里的 CustomizeAgent 到底怎么工作02-agents-actions.md
parse_mode="json" 这套结构化输出解析的底层01-core-foundation.md
图生成之后如何被评测、被优化器打磨05-self-evolution-optimizers.md
生成的 agent 如何用工具 / 记忆 / RAG06-tools-memory-rag-hitl.md

一句话定位:本章是 EvoAgentX "自动化"链条的入口——从 goal 造出图;第 3 章负责跑这张图; 第 5 章负责让这张图越跑越好。

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

主题文件路径符号名
编排入口:三步流水线evoagentx/workflow/workflow_generator.py:100WorkFlowGenerator.generate_workflow
生成器类 & 三个组件字段evoagentx/workflow/workflow_generator.py:20WorkFlowGenerator(task_planner/agent_generator/workflow_reviewer/num_turns)
第①步:调规划器evoagentx/workflow/workflow_generator.py:155WorkFlowGenerator.generate_plan
第②步:按名字对暗号连边evoagentx/workflow/workflow_generator.py:201WorkFlowGenerator.build_workflow_from_plan
第③步:逐节点配 agentevoagentx/workflow/workflow_generator.py:168WorkFlowGenerator.generate_agents
每步重试/退避evoagentx/workflow/workflow_generator.py:70WorkFlowGenerator._execute_with_retry
规划器 agentevoagentx/agents/task_planner.py:6TaskPlanner
规划 action & 产出结构evoagentx/actions/task_planning.py:27 / :20TaskPlanning / TaskPlanningOutput
规划 prompt(拆解规则)evoagentx/prompts/task_planner.py:123TASK_PLANNING_ACTION_INST
生成器 agentevoagentx/agents/agent_generator.py:6AgentGenerator
生成 action & 产出结构evoagentx/actions/agent_generation.py:126 / :120AgentGeneration / AgentGenerationOutput
单个生成 agent + 自修复校验evoagentx/actions/agent_generation.py:29 / :52GeneratedAgent / GeneratedAgent.validate_prompt
评审器(占位,未实现)evoagentx/agents/workflow_reviewer.py:6WorkFlowReviewer
图/节点/边数据模型evoagentx/workflow/workflow_graph.py:585 / :49 / :524WorkFlowGraph / WorkFlowNode / WorkFlowEdge
结构校验evoagentx/workflow/workflow_graph.py:847WorkFlowGraph._validate_workflow_structure
规格→CustomizeAgent 实例化evoagentx/agents/agent_manager.py:204 → evoagentx/utils/utils.py:222add_agents_from_workflow → create_agent_from_dict
端到端使用示例examples/workflow/workflow_demo.py:23-37main