跳到主要内容

工作流图与执行引擎:WorkFlowGraph、调度器、Environment

30 秒导读: 一个多智能体工作流在 EvoAgentX 里怎么被"表示"、又怎么"跑起来"?本章讲两件事: (1) 把工作流画成一张任务图(WorkFlowGraph,基于 networkx),节点是任务、边是依赖; (2) 一个执行引擎(WorkFlow.execute)用两层调度器驱动这张图——先选下一个该跑的任务, 再为这个任务选一个 agent/动作,中间状态全记在 Environment 里。本章还会讲另一条更"硬核"的 范式:ActionGraph——不用图、直接用代码算子把流程写死,确定性执行,是自动优化器(AFlow)改写的对象。

本章上游是 02-agents-actions.md(Agent 与 Action 是被调度的执行单元), 下游是 04-auto-construction.md(这些图从一句话目标自动生成)和 05-self-evolution-optimizers.md(图如何被优化)。本章只讲图怎么表示、怎么跑,不讲它从哪来、怎么变好。

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

一句话定义: 工作流 = 把一个大目标拆成若干子任务,规定它们的先后依赖,再让合适的 agent 依次把它们做完。

它解决什么问题。 假设你要"把一段需求变成能跑的代码":得先理解需求 → 再写代码 → 再测试修 bug。 这三步有先后(没写完不能测),也有数据流(前一步的产物是后一步的输入)。EvoAgentX 把这种"多步、有依赖、 每步交给一个 AI"的活儿,统一建模成一张有向图,然后交给引擎自动跑。

图里的两个基本概念:

概念是什么类比
节点 WorkFlowNode一个子任务(有名字、输入、输出、负责它的 agent)项目里的一个"工序"
边 WorkFlowEdge一条依赖(A 的输出喂给 B)工序之间的"先后箭头"

用起来什么样。 下面是一个真实的最小用法(来自 examples/sequential_workflow.py:59-88), 读者先感受一下"定义图 → 造 agent → 跑"这条主线:

# 1) 定义一张顺序工作流图:goal + 一串带 inputs/outputs/prompt 的任务
graph = SequentialWorkFlowGraph(goal="生成代码解决编程题", tasks=tasks)

# 2) 从图里"物化"出每个任务需要的 agent,交给 AgentManager 管理
agent_manager = AgentManager()
agent_manager.add_agents_from_workflow(graph, llm_config=llm_config)

# 3) 造一个执行引擎实例,喂输入,execute 一把梭
workflow = WorkFlow(graph=graph, agent_manager=agent_manager, llm=llm)
result = workflow.execute(inputs={"problem": "写一个最长回文子串函数"})
print(result.result)   # WorkflowResult.result

一句话直觉: 把 WorkFlowGraph 当成一张施工蓝图(画了有哪些工序、谁先谁后), 把 WorkFlow 当成包工头——他不亲自干活,而是反复问自己两个问题:"下一道该开工的工序是哪个?"、 "这道工序派哪个师傅、用哪把工具?",然后盯着进度直到收工。

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

2.1 两条执行范式,先分清​

EvoAgentX 里"工作流"有两种截然不同的表示,别混淆——这是理解本章的总纲:

范式用什么表示谁来决定执行顺序典型场景
WorkFlowGraphnetworkx 任务图(节点=任务,边=依赖)LLM 动态调度(调度器每步问 LLM)多 agent 编排、自然语言子任务
ActionGraph一段代码(execute 方法里手写算子调用)代码写死,确定性执行算法式流水线、被 AFlow 优化的对象

一句话对比:WorkFlowGraph 是"给 LLM 一张图,让它边走边决定";ActionGraph 是"程序员把流程用代码焊死"。 本章 §2-§3 主讲前者,§4 专讲后者,§5 对比。

2.2 顶层结构图(WorkFlowGraph 范式)​

怎么读这张图:左边是静态蓝图(图 + agent 池),右边是执行引擎,中间的 WorkFlow.execute 是一个反复循环的"心跳"。箭头表示一次心跳里的信息流向。

              静态蓝图                          执行引擎(每次心跳循环一次)
    ┌────────────────────────┐     ┌──────────────────────────────────────────┐
    │  WorkFlowGraph          │     │            WorkFlow.execute               │
    │   • nodes(任务+状态)    │◄────┤   while 图没跑完:                         │
    │   • edges(依赖)         │     │     ① 选下一个任务  ──► TaskScheduler     │
    │   • networkx 任务图      │     │     ② 为任务选动作  ──► ActionScheduler   │
    └────────────────────────┘     │     ③ 执行(agent 或 action_graph)        │
    ┌────────────────────────┐     │     ④ 标记节点 completed                   │
    │  AgentManager(agent池) │◄────┤                                            │
    └────────────────────────┘     └───────────────┬──────────────────────────┘
                                                    │ 每步读/写
                                                    ▼
                                    ┌──────────────────────────────┐
                                    │  Environment(执行轨迹)       │
                                    │   • trajectory(逐步消息)     │
                                    │   • execution_data(中间产物) │
                                    │   • task_execution_history    │
                                    └──────────────────────────────┘

2.3 部件一句话职责​

部件干什么在哪个文件
WorkFlowGraph任务图本体:存节点/边、校验结构、算"下一步能跑谁"workflow/workflow_graph.py:585
WorkFlowNode一个子任务:名字、inputs/outputs、agents、状态机workflow/workflow_graph.py:49
WorkFlowEdge一条有向依赖边(source→target,可带 priority)workflow/workflow_graph.py:524
WorkFlow执行引擎:主循环 execute,驱动整张图跑完workflow/workflow.py:32
WorkFlowManager双层调度器的门面:schedule_next_task + schedule_next_actionworkflow/workflow_manager.py:416
TaskScheduler调度动作①:从候选任务里选下一个该跑的任务节点workflow/workflow_manager.py:46
ActionScheduler调度动作②:为选中的任务选下一个 agent/actionworkflow/workflow_manager.py:207
Environment记录执行轨迹与中间产物,作为调度上下文workflow/environment.py:24
ActionGraph另一条范式:代码定义的确定性算子流水线workflow/action_graph.py:12
OperatorActionGraph 的积木:一次带 prompt 的 LLM 调用/算子workflow/operators.py:34

2.4 主线走一遍(高层,不进代码)​

一次 workflow.execute(inputs) 的骨架(workflow/workflow.py:140 _execute_workflow):

输入 inputs
   │  ① reset 图状态 + 清空 Environment,把 inputs 作为一条 INPUT 消息写进轨迹
   ▼
while 图没跑完(graph.is_complete 为假):
   │  ② TaskScheduler:图算出"当前可跑的候选任务",只有一个直接选,多个问 LLM
   │  ③ graph.step:把上一个完成的任务→这个任务的状态流转推进(处理循环/重跑)
   │  ④ ActionScheduler:给这个任务选一个 agent+action(或直接用它的 action_graph)
   │  ⑤ 执行:agent 干活 / action_graph 跑代码,产物写回 Environment.execution_data
   │  ⑥ graph.completed(task):把节点标成 COMPLETED
   ▼
所有 end 节点完成 → 从 Environment 取出 workflow_outputs → 包成 WorkflowResult 返回

要记住的一个词:"两层调度"。外层选任务(TaskScheduler),内层选动作(ActionScheduler)。 外层决定"做哪一道工序",内层决定"这道工序怎么下手"。

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

3.1 任务图的数据结构:节点、边、networkx​

它要解决的小问题: 怎么把"子任务 + 依赖"存成一个能查询("谁是谁的前驱?""下一步能跑谁?")的结构。

思路: 用一张 networkx MultiDiGraph(有向多重图)当底层存储,WorkFlowGraph 在它之上包一层 业务方法。节点/边的业务信息用 ref 属性挂在 networkx 节点/边上,networkx 只负责拓扑查询。

节点的字段很直白(workflow/workflow_graph.py:68-75):

字段类型含义
namestr任务唯一名(图内唯一)
descriptionstr任务描述(喂给调度器/agent 的自然语言)
inputs / outputsList[Parameter]输入/输出参数(名字用来推断依赖边)
agentsList[str/dict/Agent]能执行它的 agent(可为名字引用)
action_graphOptional[ActionGraph]或者:用一个 action_graph 执行(此时不需要 agents)
statusWorkFlowNodeState当前状态,见 §3.2

关键设计:一个节点有两种"干活方式"——agents 或 action_graph,二选一。 这是两条范式在节点层面的 汇合点(check_agents 在 workflow/workflow_graph.py:303 明确:有 action_graph 时无需 agents 就算合法)。

边是怎么来的?多数时候不用手写。 _infer_edges_from_nodes(workflow/workflow_graph.py:664) 自动推断:只要节点 A 的某个 output 名字 == 节点 B 的某个 input 名字,就自动连一条 A→B 的边。 这段简化演示了核心逻辑:

# 示意,非源码:输出名撞上输入名 → 连边
for source in nodes:
    for target in nodes:
        if source is target:
            continue
        # A 的输出名集合 与 B 的输入名集合有交集?
        if node_outputs[source.name] & node_inputs[target.name]:
            edges.append(Edge(source=source.name, target=target.name))

用户额外写的 explicit_edges 会叠加上去,用来表达"没有数据流、但要求先后顺序"的控制依赖 (_init_from_nodes 在 workflow/workflow_graph.py:684 合并两类边,同 (source,target) 的显式边覆盖推断边的元数据)。

3.2 节点状态机:任务的一生​

它要解决的小问题: 引擎怎么知道"哪些任务还没跑、哪些正在跑、哪些跑完了"?

思路: 每个节点有个 status,在四个状态间流转(WorkFlowNodeState,workflow/workflow_graph.py:33-46):

   PENDING ──► RUNNING ──► COMPLETED
   (待执行)   (执行中)     (已完成)
                 │
                 └──► FAILED(执行失败)

图上提供了状态流转的快捷方法(workflow/workflow_graph.py:1181-1191): pending() / running() / completed() / failed()。执行引擎在主循环里调用它们推进状态。

graph.is_complete(workflow/workflow_graph.py:1147)判断整张图是否跑完:只看 end 节点是否全部 COMPLETED ——不要求每个节点都跑(有分支/可选路径时,不是所有节点都会被执行)。

3.3 图的校验:跑之前先把结构查一遍​

它要解决的小问题: 一张手搓的图可能有各种毛病——输入无来源、输出名撞车、孤立节点。 最好在执行前就报错,而不是跑到一半崩。

validate_workflow_graph(workflow/workflow_graph.py:1610)做三层检查:

检查方法查什么
结构_validate_workflow_structure:847输入/输出名唯一、无非法孤立节点、有初始节点
图 I/O ↔ 节点_check_workflow_inputs_outputs:1563声明的 workflow 输入/输出都能被某个节点消费/产生
节点 ↔ agent_check_agents:1604每个节点的 inputs/outputs 都被它的 agents 覆盖

其中最能防坑的一条是 _check_node_inputs(workflow/workflow_graph.py:827): 每个节点的每个输入,要么来自 workflow 的全局输入,要么来自另一个节点的输出——否则报错。 这保证了图里不存在"凭空冒出来的输入"。

一个巧妙点:节点的 output 名全局唯一。 _check_node_output_uniqueness(workflow/workflow_graph.py:805) 强制"同一个输出名最多由一个节点产生"。因为整个执行期共享一个扁平的 execution_data 字典 (§3.6),名字不唯一就会互相覆盖——所以这条约束是数据流正确性的基石。

3.4 调度动作①:选下一个可执行的任务节点​

它要解决的小问题: 当前这一步,图里哪些任务已经"万事俱备"可以开跑?

这分两步。先是图自己算出候选集,再由 TaskScheduler(可能借助 LLM)从候选里拍板。

第一步:图算候选(纯拓扑,不涉及 LLM)。 get_next_candidate_nodes(workflow/workflow_graph.py:1370) 的逻辑:

① 还没完成的初始节点里,前驱都完成了的 → 直接作为候选(优先跑起点)
② 否则:找到"每条路径上最后一个已完成的节点"
③ 取它们的未完成子节点
④ 过滤掉"前驱还没全完成"的 → 剩下的就是候选

判断"前驱是否都完成"的是 are_dependencies_complete(workflow/workflow_graph.py:1339): 默认要求所有前驱都 COMPLETED;但如果节点是循环的起点,则忽略来自"循环终点"的回边 (否则循环永远无法第二次进入)。EvoAgentX 支持带循环的工作流(如"生成→测试→不过就回去改"), _find_all_loops(workflow/workflow_graph.py:916)在建图时用 DFS 把所有环预先识别出来。

第二步:TaskScheduler 拍板。 TaskScheduler.execute(workflow/workflow_manager.py:122)的精髓在 _handle_edge_cases(workflow/workflow_manager.py:67):

# 示意,非源码:调度器的省钱哲学
candidates = graph.next()          # 图算出的候选任务
if len(candidates) == 1:
    return forward(candidates[0])   # 只有一个?直接选,不问 LLM
# 有多个候选,才组织 prompt 问 LLM 挑一个
prompt = build_prompt(graph_desc, execution_history, candidate_infos)
return llm.generate(prompt, parser=TaskSchedulerOutput)

重点看:只有候选任务 ≥2 个时才调用 LLM。 大量顺序工作流每步只有一个候选,于是调度是零 LLM 成本的。 需要 LLM 时,它拿到"图描述 + 执行历史 + 候选任务信息",输出一个 TaskSchedulerOutput (decision / task_name / reason,workflow/workflow_manager.py:36)。

3.5 调度动作②:为任务选 agent 与 action​

它要解决的小问题: 任务定了,但一个任务可能挂了多个 agent、每个 agent 又有多个 action。到底派谁、做哪个动作?

ActionScheduler(workflow/workflow_manager.py:207)负责,产出一个 NextAction (agent / action / reason / action_graph,workflow/workflow_manager.py:191)。 它的决策树同样"能不问 LLM 就不问"(_prepare_action_execution,workflow/workflow_manager.py:240):

任务有 action_graph?           ──► NextAction(action_graph=...)  直接返回,不问 LLM
上一条消息指定了 next_actions?  ──► 从这些动作里挑(唯一则直接定)
只有 1 个 agent 且 1 个 action?──► 直接定,不问 LLM
以上都不满足                    ──► 组织 prompt 问 LLM 选 (agent, action)

注意第一条:如果节点带 action_graph,这里直接把它打包返回——这就是执行引擎从"agent 编排" 切换到"确定性算子图"的开关(§4)。

3.6 Environment:执行轨迹 = 调度的上下文​

它要解决的小问题: 调度器凭什么做决策?agent 干活的输入从哪来?——都来自一块共享的"执行记录"。

Environment(workflow/environment.py:24)持有三样东西:

字段是什么谁用它
trajectoryList[TrajectoryStep],逐步的消息 + 状态(COMPLETED/FAILED)调度器读"执行历史"组 prompt
execution_data扁平的 dict,所有节点产物按参数名堆一起agent/action 按名字取输入;取最终输出
task_execution_history已执行任务名的有序列表调度器展示"走到哪了"

核心机制:每一步都往 Environment update 一条消息(workflow/environment.py:42), update 会顺手把消息里的结构化产物合并进 execution_data(update_execution_data,workflow/environment.py:62):

# 示意,非源码:产物按参数名扁平堆叠
def update_execution_data(self, message):
    if isinstance(message.content, LLMOutputParser):
        self.execution_data.update(message.content.get_structured_data())
    if isinstance(message.content, dict):
        self.execution_data.update(message.content)

这解释了 §3.3 为什么强制输出名全局唯一: 所有产物挤在同一个扁平字典里,靠名字寻址。 节点 B 需要 code 这个输入时,不管它是谁产的,直接 execution_data["code"] 就拿到了 (get_execution_data,workflow/environment.py:111)。数据流因此是基于名字的隐式连线,而非显式传参。

TrajectoryState(workflow/environment.py:9)只有两个值:COMPLETED / FAILED,标记每一步成功与否。

3.7 把它们串起来:WorkFlow.execute 主循环​

它要解决的小问题: 上面这些零件怎么协同成一次完整执行?

execute 是同步外壳(workflow/workflow.py:96),内部用 run_coroutine_sync 调 async_execute → _execute_workflow(workflow/workflow.py:140)。主循环本体只有十几行,清清爽爽:

# 示意,非源码:主循环骨架(对应 _execute_workflow / execute_task)
graph.reset_graph(); environment.reset()          # 每次执行前清干净,可复用实例
environment.update(INPUT 消息)                     # 输入作为第一条轨迹

while not graph.is_complete:
    task = await workflow_manager.schedule_next_task(graph, env)   # 调度①
    if task is None:
        break
    graph.step(source=上一个完成的任务, target=task)  # 推进状态机(处理循环/重跑)
    next_action = await workflow_manager.schedule_next_action(...)  # 调度②
    if next_action.action_graph is not None:
        await 用 action_graph 跑(确定性)
    else:
        await 用 agent 跑(可能多轮,直到任务产出齐全)
    graph.completed(task)                          # 标记完成

return environment.get_execution_data(输出名)      # 取最终输出

真实的 execute_task(workflow/workflow.py:240)里,用 agent 执行时是一个内层小循环 (_async_execute_task_by_agents,workflow/workflow.py:303):反复"选动作→执行→检查任务是否产出齐全", 直到 is_task_completed(workflow/workflow.py:400,即该任务所有 output 名都进了 execution_data)为真, 或撞上 max_execution_steps 上限(默认 5,workflow/workflow.py:42)报错。这是防止 agent 反复徒劳的护栏。

收尾成 WorkflowResult。 async_execute(workflow/workflow.py:110)用 try/except 包住整个执行, 成功返回 WorkflowResult(status="success", result=...),失败则捕获异常、附上 traceback 与一条可展示给用户的 错误信息(WorkflowResult,workflow/workflow.py:26)。

4. 另一条范式:ActionGraph 与 Operator(用代码而非图)​

前面整章讲的都是 WorkFlowGraph——LLM 每步动态决定该干嘛。但很多任务(尤其是被自动优化器改写的那些) 需要确定性、可复现的流程。这就是 ActionGraph。

4.1 ActionGraph:把流程写进 execute 方法​

核心区别一句话: ActionGraph 里没有图数据结构、没有调度器——流程就是你在 execute 方法里 手写的一段代码,顺序、循环、分支全用 Python 表达。

ActionGraph(workflow/action_graph.py:12)基类只有三个字段(name / description / llm_config)和一个 待子类实现的 execute(workflow/action_graph.py:26,基类直接 raise NotImplementedError)。看内置的 QAActionGraph(workflow/action_graph.py:110)就懂了:

# 示意,非源码:QAActionGraph.execute —— 自洽性投票问答
def execute(self, problem: str) -> dict:
    solutions = []
    for _ in range(3):                          # 同一问题采样 3 个答案
        response = self.answer_generate(input=problem)
        solutions.append(response["answer"])
    best = self.sc_ensemble(solutions=solutions)  # 投票选最一致的
    return {"answer": best["response"]}

流程("采样 3 次 → 投票")是代码焊死的,不问任何调度器。真实实现见 workflow/action_graph.py:121。

ActionGraph 也能被节点当作执行器:WorkFlowNode.action_graph 一旦设置,该节点就走确定性执行分支 (§3.5 的 ActionScheduler 第一条规则直接返回它)。执行引擎里 _async_execute_task_by_action_graph (workflow/workflow.py:261)按 execute 的签名从 execution_data 里取参数、调用、把返回 dict 写回环境。

4.2 Operator:ActionGraph 的积木​

ActionGraph 里那些 self.answer_generate(...)、self.sc_ensemble(...) 是什么?是 Operator (workflow/operators.py:34)——每个 Operator 封装一次带特定 prompt 的 LLM 调用 + 结构化解析。

Operator 有个贴心设计:__call__(workflow/operators.py:56)自动判断同步/异步环境—— 在异步事件循环里就返回协程,否则同步执行,所以同一个算子在两种上下文都能直接 operator(...) 调用。

内置算子(都改编自 AFlow 的预定义算子,workflow/operators.py:1 顶部致谢):

算子干什么位置
Custom按自定义 instruction+input 生成任意内容operators.py:95
AnswerGenerate逐步思考(thought)+ 给最终答案(answer)operators.py:121
ScEnsemble / QAScEnsemble自洽性投票:从多个候选解里选最一致的一个operators.py:183 / :148
Programmer写 Python 代码 + 在沙箱里执行,失败带反馈重试 3 次operators.py:406
Test用公开测试用例跑解,失败则反思修正,循环最多 3 轮operators.py:277

Programmer(workflow/operators.py:447)值得一看:它不只生成代码,还用 ProcessPoolExecutor + 超时把代码真跑一遍(run_code,workflow/operators.py:371,并禁掉 os/sys/subprocess 等危险导入), 失败就把错误信息塞回 prompt 重试。"生成即验证",而不是把没验证过的代码直接吐出来。

4.3 为什么单拎出 ActionGraph:它是优化器改写的对象​

关键动机:ActionGraph 的算子(尤其 prompt)是可以被程序自动改写的。Operator.set_prompt / get_graph_info(workflow/action_graph.py:32)把每个算子的 prompt、interface 暴露成可序列化配置。 AFlow 之类的优化器就是直接改这些算子的组合与 prompt 来优化工作流——这属于 05-self-evolution-optimizers.md 的内容,本章只需知道: ActionGraph 之所以用"代码图"而非"任务图",正是为了让它成为一个结构清晰、可被机器改写的确定性对象。

5. 两种范式对比(WorkFlowGraph vs ActionGraph)​

这是本章最该带走的一张表:

维度WorkFlowGraphActionGraph
表示形式networkx 任务图(节点+边)Python 代码(execute 方法体)
执行顺序谁定调度器 + LLM 动态决定代码写死,确定性
执行单元Agent(带多个 Action)Operator(带 prompt 的 LLM 算子)
是否每步问 LLM候选多时问(调度决策)否(只在算子内部调 LLM 干活)
中间状态存在共享 Environment存在方法内局部变量 / 返回 dict
典型用途多 agent 协作、自然语言子任务编排算法式流水线、可复现基准
谁改写它工作流生成器(第 4 章)AFlow 等优化器(第 5 章)
关键类WorkFlowGraph / WorkFlowActionGraph / Operator

判断口诀: 需要"让 LLM 边走边编排、灵活分支"→ WorkFlowGraph;需要"确定性、可复现、可被自动调优的 算子流水线"→ ActionGraph。两者能在节点层面桥接:一个 WorkFlowNode 可以用 action_graph 作为执行器, 把一段确定性子流程嵌进一张动态编排图里。

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

① 数据流靠"名字对齐"隐式连线,不显式传参。 输出名撞输入名就自动连边(_infer_edges_from_nodes, workflow/workflow_graph.py:664),所有产物堆进扁平 execution_data 按名字取用。代价是强约束 "输出名全局唯一"(_check_node_output_uniqueness,workflow/workflow_graph.py:805),换来的是图定义极简。

② 调度"能省则省":单候选/单 agent 时零 LLM 调用。 TaskScheduler._handle_edge_cases (workflow/workflow_manager.py:67)和 ActionScheduler._prepare_action_execution (workflow/workflow_manager.py:240)都优先走确定性快路径,只有真有歧义(多候选/多 agent)才掏钱问 LLM。

③ 建图时预先识别所有循环,让带反馈的工作流成为一等公民。 _find_all_loops (workflow/workflow_graph.py:916)用 DFS 找环并规范化;are_dependencies_complete (workflow/workflow_graph.py:1339)对循环起点忽略回边前驱,step(workflow/workflow_graph.py:1403)在 "循环终点→循环起点"时把整圈节点重置为 PENDING——于是"生成→测试→改→再测"这种回路能自然运转。

④ 节点校验会 monkey-patch action_graph 的 execute 做运行时契约检查。 check_action_graph (workflow/workflow_graph.py:126)在建节点时包裹 execute,运行时校验"返回是 dict 且含所有声明的 output 名", 把契约违约在第一时间抛出来。

⑤ 生成代码"即时执行验证"。 Programmer(workflow/operators.py:406)在受限沙箱里真跑生成的代码、 带超时与危险导入拦截,失败反馈重试——输出的是验证过的解,而非猜测。

7. 边界与局限​

  • 输出名必须全局唯一。 两个节点想产出同名字段(哪怕语义不同)会在校验期直接报错 (workflow/workflow_graph.py:805)。这是扁平 execution_data 的代价。

  • 每个子任务的 agent 执行有硬上限。 max_execution_steps 默认 5(workflow/workflow.py:42); agent 若在 5 步内没凑齐任务的全部 output,直接抛错而非无限重试。结构设计不当(inputs/outputs 对不上)容易撞这个墙。

  • HITL 节点限制多。 带人工介入 agent 的节点目前必须是该节点唯一的 agent,且不能是终止节点 (workflow/workflow.py:481、:534)——代码注释明说这是当前实现约束,未来可能放宽。

  • 调度依赖 LLM 遵守输出格式。 多候选时 TaskScheduler / ActionScheduler 要求 LLM 返回结构化的 task_name / (agent, action);LLM 幻觉出不存在的名字时,ActionScheduler.check_candidate_action (workflow/workflow_manager.py:224)会拦下未知 action,但整体正确性仍系于 LLM 的格式遵从。

  • ActionGraph 无内建循环/分支调度。 它的控制流就是 Python 代码——灵活性来自写代码的人,框架本身不为它提供 依赖管理或状态可视化(那些是 WorkFlowGraph 的地盘)。

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

用符号名 grep 定位,比行号抗漂移。

主题文件路径符号名
节点定义与两种执行器evoagentx/workflow/workflow_graph.pyWorkFlowNode
节点状态机evoagentx/workflow/workflow_graph.pyWorkFlowNodeState
依赖边evoagentx/workflow/workflow_graph.pyWorkFlowEdge
任务图本体evoagentx/workflow/workflow_graph.pyWorkFlowGraph
边自动推断(输出名↔输入名)evoagentx/workflow/workflow_graph.py_infer_edges_from_nodes
全图校验(结构/IO/agent)evoagentx/workflow/workflow_graph.pyvalidate_workflow_graph
输出名全局唯一约束evoagentx/workflow/workflow_graph.py_check_node_output_uniqueness
算候选任务(拓扑)evoagentx/workflow/workflow_graph.pyget_next_candidate_nodes
前驱完成判断(含循环)evoagentx/workflow/workflow_graph.pyare_dependencies_complete
循环识别evoagentx/workflow/workflow_graph.py_find_all_loops
状态推进(循环/重跑)evoagentx/workflow/workflow_graph.pystep
顺序工作流(便捷子类)evoagentx/workflow/workflow_graph.pySequentialWorkFlowGraph
执行引擎 / 主循环evoagentx/workflow/workflow.pyWorkFlow._execute_workflow
单任务执行(agent/action_graph 分派)evoagentx/workflow/workflow.pyexecute_task
agent 执行内层循环 + 步数上限evoagentx/workflow/workflow.py_async_execute_task_by_agents
执行结果封装evoagentx/workflow/workflow.pyWorkflowResult
双层调度器门面evoagentx/workflow/workflow_manager.pyWorkFlowManager
调度①:选任务节点evoagentx/workflow/workflow_manager.pyTaskScheduler
调度①省 LLM 的快路径evoagentx/workflow/workflow_manager.py_handle_edge_cases
调度②:选 agent/actionevoagentx/workflow/workflow_manager.pyActionScheduler
调度②决策树evoagentx/workflow/workflow_manager.py_prepare_action_execution
下一步动作数据结构evoagentx/workflow/workflow_manager.pyNextAction
执行环境 / 轨迹evoagentx/workflow/environment.pyEnvironment
产物按名字扁平合并evoagentx/workflow/environment.pyupdate_execution_data
轨迹步状态evoagentx/workflow/environment.pyTrajectoryState
确定性算子图基类evoagentx/workflow/action_graph.pyActionGraph
内置自洽性问答图evoagentx/workflow/action_graph.pyQAActionGraph
算子基类(同步/异步自适应)evoagentx/workflow/operators.pyOperator
代码生成+沙箱执行算子evoagentx/workflow/operators.pyProgrammer
反思修正测试算子evoagentx/workflow/operators.pyTest