在多Agent系统中,最让人头疼的问题往往不是单个Agent的能力,而是如何让多个Agent协同工作。当任务A的输出是任务B的输入,任务C必须等待A和B都完成后才能启动,一旦依赖关系没有管理好,整个流程就会陷入混乱。MiMo Code的任务编排引擎正是为了解决这个问题而设计的,它把复杂的Agent协作抽象成一张可计算的任务图,让依赖解析、调度执行、结果传递都有章可循。

任务编排引擎的核心设计:为什么选择有向无环图
MiMo Code的编排引擎底层采用了DAG(Directed Acyclic Graph,有向无环图)作为任务模型。每个Agent任务被抽象为一个节点,节点之间的边表示依赖关系。选择DAG而非简单的线性队列,原因在于真实的多Agent场景很少是纯线性的:代码审查任务可能同时依赖静态分析Agent和测试生成Agent的结果,汇总Agent又要等前面两个都完成,这种扇入扇出的结构用DAG表达最自然。
DAG的第二个好处是可以自动推导执行顺序。引擎在构建完任务图之后,会先执行一次拓扑排序,把所有节点排成若干批次,同一批次内的任务没有相互依赖,可以并行调度。比如A、B、C三个任务,其中C依赖A和B,拓扑排序后得到两个批次:第一批是A和B,第二批是C。这种批次化的调度方式让引擎不用为每个任务单独维护等待逻辑,只需要按批次推进,实现简单且不容易出错。
同时,DAG的无环约束也帮助引擎在构建阶段就能发现问题。如果开发者在声明依赖时不小心写出了循环依赖,比如A依赖B、B又依赖A,引擎会在提交任务图时立即抛出异常,而不是等到运行时卡死。这种快速失败的设计在调试复杂工作流时非常宝贵。
依赖声明的三种方式与适用场景
MiMo Code提供了三种声明依赖的方式,分别应对不同的使用场景。第一种是显式声明,直接在任务节点上指定它依赖哪些前置任务:
from mimo_code.orchestrator import TaskGraph, AgentTask
graph = TaskGraph()
# 创建三个Agent任务节点
analysis_task = AgentTask(
name="static_analysis",
agent="code-analyzer",
params={"repo_path": "./src"}
)
test_task = AgentTask(
name="test_generation",
agent="test-writer",
params={"target_module": "core"}
)
review_task = AgentTask(
name="final_review",
agent="reviewer"
)
# 显式声明依赖:review 依赖前两个任务
graph.add(analysis_task)
graph.add(test_task)
graph.add(review_task, depends_on=["static_analysis", "test_generation"])
# 提交任务图,引擎会自动做拓扑排序和循环检测
graph.submit()显式声明的好处是依赖关系一目了然,适合任务数量不多、结构相对固定的场景。但当任务图变得庞大时,逐个写依赖容易遗漏。这时可以用第二种方式:基于数据流推断。引擎会分析每个任务的输入参数来源,如果某个参数引用了另一个任务的输出,依赖关系会自动建立。
from mimo_code.orchestrator import TaskGraph, AgentTask, TaskOutput
graph = TaskGraph()
analysis = graph.add(AgentTask(
name="static_analysis",
agent="code-analyzer"
))
# review 任务的 input 引用了 analysis 的输出,依赖自动建立
review = graph.add(AgentTask(
name="final_review",
agent="reviewer",
params={
"analysis_report": TaskOutput(analysis, field="report")
}
))
# 打印自动推断的依赖关系,确认是否符合预期
print(review.depends_on) # 输出: ['static_analysis']第三种是条件依赖,用于表达分支逻辑。比如只有当静态分析发现问题时,才需要修复Agent介入。条件依赖通过when参数声明,引擎会在运行时根据前置任务的实际输出决定是否激活下游节点,被跳过的节点状态会标记为skipped,其下游依赖也会级联跳过。
调度执行与结果传递机制
依赖关系建立之后,引擎进入调度阶段。MiMo Code采用了批次并行加节点内重试的调度策略:每个批次内的任务会被提交到线程池或分布式执行器并发执行,批次内所有任务完成后才进入下一批次。这种调度方式在保证依赖正确性的同时最大化了并行度。
结果传递是依赖管理的另一半。前置任务的输出需要以某种形式注入到下游任务的上下文中。MiMo Code使用了一个共享的上下文存储,每个任务完成后,其输出会按照任务名加字段名的格式写入存储,下游任务启动时,引擎根据依赖声明把相关数据注入到任务的参数里。这样做的好处是解耦:下游任务不需要知道上游任务在哪个进程执行,只需要声明需要什么数据。
对于大数据量的场景,直接把完整输出注入参数并不划算。引擎提供了引用传递模式,此时注入的是结果的存储句柄,下游Agent可以在需要时按需读取。这在处理代码仓库全量分析这类输出体积较大的任务时尤其重要,可以显著降低内存占用和序列化开销。
工程实践:超时、重试与循环依赖的排查
依赖管理的实际工程难点通常出现在异常路径上。首先是超时控制,某个前置Agent卡住会阻塞整条链路。建议为每个任务设置独立的超时时间,引擎超时后会将该任务标记为failed,并根据失败策略决定是终止整个图还是触发降级分支:
task = AgentTask(
name="static_analysis",
agent="code-analyzer",
timeout=300, # 单任务超时 300 秒
retry=2, # 失败后最多重试 2 次
retry_delay=10, # 每次重试间隔 10 秒
on_failure="degrade" # 失败策略:走降级分支而不是终止整个图
)
graph.add(task, depends_on=["prepare"])其次是循环依赖的排查。虽然引擎会在提交时报错,但错误信息有时只指出两个任务名,定位完整的环需要借助引擎提供的可视化工具。调用graph.visualize()可以输出任务图的结构,配合graph.find_cycles()能直接列出所有的环,快速定位是哪条依赖声明写反了。
最后一点实践建议:依赖关系应该尽量反映数据流而不是执行顺序的猜测。很多开发者习惯按照设想的执行时序去声明依赖,结果导致本可以并行的任务被串行化,白白损失性能。正确的做法是问自己:这个任务真正需要哪些上游数据?如果答案是不需要,就不要添加依赖,让引擎的拓扑排序去决定最优的执行顺序。理解了这一点,就能把MiMo Code编排引擎的能力真正发挥出来。