大模型多Agent协同中的状态机管理用 Go 实现一个轻量级 DAG 任务流引擎一、深度引言与场景痛点当大模型应用从单兵作战的 Prompt 调优走向复杂业务场景时单个 Agent智能体的能力上限便会迅速暴露。为了处理复杂的长链路业务业界普遍转向多 Agent 协同架构Multi-Agent Collaboration。在这种架构下不同的 Agent 扮演不同的角色一个负责用户需求拆解一个负责从向量数据库检索数据另一个负责代码生成或逻辑校验最后由一个专家 Agent 进行最终汇总。然而一旦多个 Agent 开始频繁交互工程上的管理灾难便接踵而至。最显著的痛点是协同状态的失控。多个 Agent 之间的调用关系极其容易陷入死锁或无限递归。例如Agent A 发现生成的数据不合格将其打回给 Agent B 重构而 Agent B 又因为某些前置条件未满足再次请求 Agent A最终导致在有限的 Token 预算内发生无休止的“套娃调用”。其次是并发与拓扑依赖管理的混乱。在长链路业务中有些 Agent 的执行是相互独立的如同时进行代码安全扫描和逻辑合规性审查可以并发运行而有些 Agent 必须等待前置任务全部完成如必须等扫描和审查都通过后才能交由发布 Agent 处理。如果仅通过在代码里手工嵌套go channel或硬编码状态字段很快就会把代码库变成不可维护的“意大利面条”。大厂的常用解法是直接引入 Temporal、Apache Airflow 等重量级分布式工作流引擎。但对于追求高 ROI、资源吃紧的创业团队或小研发团队来说这无异于高射炮打蚊子。引入这些庞然大物不仅意味着成倍的服务器运维成本更会带来漫长的开发调试链路。最务实、最轻量级的工程方案是在应用内存中构建一个基于有向无环图DAG, Directed Acyclic Graph的任务流引擎配合有限状态机FSM, Finite State Machine来统一接管多 Agent 协同的生命周期与数据依赖。二、底层机制与原理深度剖析要在内存中优雅地管理多 Agent 协同我们必须将协同流程抽象为一个有向无环图DAG其中节点Node/Vertex代表单个 Agent 的执行任务。有向边Edge代表任务之间的执行先后顺序及数据依赖。在 DAG 中任何节点在被触发执行前必须确保其所有的前置依赖节点In-degree 参入度都已成功执行完毕。为了实现这一点引擎底层的核心算法是拓扑排序Topological Sort通过计算每个节点的入度来决定任务的并发调度顺序。以下是该并发任务流协作机制的 Mermaid 原理架构图清晰地展示了一个用户请求进来后DAG 引擎是如何动态调度并隔离各个 Agent 节点的flowchart TD A[用户请求/目标输入] -- B[DAG 引擎解析依赖] B -- C{计算拓扑排序} C --|无前置依赖| D[Agent-1 需求拆解] C --|无前置依赖| E[Agent-2 知识库检索] D --|产生依赖关系| F{依赖校验中心} E --|产生依赖关系| F F --|前置节点全部完成| G[Agent-3 逻辑校验与生成] G -- H[Agent-4 专家评审/最终输出] H -- I[执行完毕/状态释放] subgraph 有限状态机 FSM 控制器 D -.-|更新状态| FSM[FSM: Init - Running - Success/Failed] E -.-|更新状态| FSM G -.-|更新状态| FSM H -.-|更新状态| FSM end这个架构的底层流转逻辑如下FSM 控制状态变更每个 Agent 节点都有自己的生命周期Pending、Running、Success、Failed、Skipped。引擎通过原子的 CASCompare-And-Swap操作来保证在多协程并发下状态流转的安全性。信号量与通道调度引擎在检测到某个节点的前置依赖全部变为 Success 后会自动将其送入执行通道由协程池并发处理。动态背压与数据传递前置 Agent 的输出Output会自动绑定为后置 Agent 的输入Input并且通过上下文 Context 传递超时控制。三、生产级代码实现与最佳实践大厂的烂代码才喜欢通过堆砌抽象类来展示架构设计我们的原则是KISS。下面我将分享一个用 Go 语言实现的高性能、就地运行的轻量级 DAG 任务流调度引擎核心代码package dagengine import ( context errors fmt sync ) // State 代表任务节点的执行状态 type State string const ( StatePending State PENDING StateRunning State RUNNING StateSuccess State SUCCESS StateFailed State FAILED ) // AgentFunc 代表具体的 Agent 执行函数接受上下文和输入数据返回输出数据或错误 type AgentFunc func(ctx context.Context, input interface{}) (interface{}, error) // TaskNode 代表 DAG 中的任务节点 type TaskNode struct { ID string Action AgentFunc DependsOn []string // 依赖的前置节点 ID 列表 State State // 当前状态 Input interface{} Output interface{} Err error mu sync.RWMutex } // SetState 安全地更新节点状态 func (n *TaskNode) SetState(s State) { n.mu.Lock() defer n.mu.Unlock() n.State s } // GetState 安全地获取节点状态 func (n *TaskNode) GetState() State { n.mu.RLock() defer n.mu.RUnlock() return n.State } // DAGEngine 负责管理并并发执行有向无环图任务流 type DAGEngine struct { nodes map[string]*TaskNode mu sync.RWMutex } // NewDAGEngine 创建一个新的 DAG 引擎实例 func NewDAGEngine() *DAGEngine { return DAGEngine{ nodes: make(map[string]*TaskNode), } } // AddTask 向引擎中注册一个任务节点 func (e *DAGEngine) AddTask(id string, fn AgentFunc, dependsOn []string) error { e.mu.Lock() defer e.mu.Unlock() if _, exists : e.nodes[id]; exists { return fmt.Errorf(任务节点 %s 已经存在, id) } e.nodes[id] TaskNode{ ID: id, Action: fn, DependsOn: dependsOn, State: StatePending, } return nil } // Run 并发执行 DAG 任务流直到所有任务执行完成或者检测到环、执行失败 func (e *DAGEngine) Run(ctx context.Context, initialInput map[string]interface{}) error { if err : e.hasCycle(); err ! nil { return fmt.Errorf(DAG 任务流存在循环依赖错误: %w, err) } // 初始化初始节点的输入数据 for id, val : range initialInput { if node, exists : e.nodes[id]; exists { node.Input val } } var wg sync.WaitGroup errChan : make(chan error, len(e.nodes)) ctx, cancel : context.WithCancel(ctx) defer cancel() for { e.mu.RLock() allDone : true anyFailed : false var tasksToRun []*TaskNode for _, node : range e.nodes { state : node.GetState() if state StateFailed { anyFailed true break } if state ! StateSuccess { allDone false } // 如果是 pending 状态且前置依赖已全部成功则本轮可以运行 if state StatePending e.dependenciesSatisfied(node) { tasksToRun append(tasksToRun, node) } } e.mu.RUnlock() if anyFailed { return errors.New(任务流中存在节点执行失败引擎已终止运行) } if allDone { break } // 如果没有任务可以运行但还有任务没运行成功说明发生死锁可能有环漏检但前面已有检测保证 if len(tasksToRun) 0 { timeChan : make(chan struct{}) // 阻塞等待并发任务通知 _ timeChan // 生产中可以用 channel 信号通知这里简化为轮询间隔 break } // 并发调度本轮准备就绪的任务 for _, node : range tasksToRun { node.SetState(StateRunning) wg.Add(1) go func(n *TaskNode) { defer wg.Done() // 收集前置依赖节点的输出作为输入数据源简单合并 n.Input e.gatherInputs(n) output, err : n.Action(ctx, n.Input) if err ! nil { n.Err err n.SetState(StateFailed) errChan - fmt.Errorf(节点 %s 执行失败: %w, n.ID, err) cancel() // 发生错误撤销整个链路的上下文 return } n.Output output n.SetState(StateSuccess) }(node) } wg.Wait() } select { case err : -errChan: return err default: return nil } } // dependenciesSatisfied 检查节点的所有依赖是否已全部 SUCCESS func (e *DAGEngine) dependenciesSatisfied(node *TaskNode) bool { for _, depID : range node.DependsOn { depNode, exists : e.nodes[depID] if !exists || depNode.GetState() ! StateSuccess { return false } } return true } // gatherInputs 汇总前置依赖节点的 Output func (e *DAGEngine) gatherInputs(node *TaskNode) map[string]interface{} { inputs : make(map[string]interface{}) for _, depID : range node.DependsOn { if depNode, exists : e.nodes[depID]; exists { inputs[depID] depNode.Output } } return inputs } // hasCycle 拓扑排序进行环路检测Kahn 算法思想 func (e *DAGEngine) hasCycle() error { inDegree : make(map[string]int) adjList : make(map[string][]string) for _, node : range e.nodes { inDegree[node.ID] len(node.DependsOn) for _, dep : range node.DependsOn { adjList[dep] append(adjList[dep], node.ID) } } queue : make([]string, 0) for id, deg : range inDegree { if deg 0 { queue append(queue, id) } } visitedCount : 0 for len(queue) 0 { curr : queue[0] queue queue[1:] visitedCount for _, neighbor : range adjList[curr] { inDegree[neighbor]-- if inDegree[neighbor] 0 { queue append(queue, neighbor) } } } if visitedCount ! len(e.nodes) { return errors.New(依赖关系中检测到循环引用 (Cycle Detected)) } return nil }这段代码实现了引擎的核心精髓防爆并发没有像微服务编排那样引入复杂的分布式事务锁而是依赖于sync.RWMutex和状态机流转控制。环路预防在执行前进行 Kahn 拓扑图环路检测从源头防止出现死锁循环。统一上下文取消任何一个 Agent 执行失败立即通过取消 Context 广播到所有子任务极大释放系统算力并防止 Token 资源被白白浪费。四、边界分析与架构权衡基于内存的轻量级 DAG 任务流引擎极其适合快速上线和中小规模流量但为了在生产中做好防护必须了解其技术边界与架构的妥协1. 单点故障与状态丢失由于该引擎的所有节点运行状态和数据交互都保存在 Go 进程的内存中一旦服务器发生宕机或进行灰度重启运行中的任务状态将彻底丢失且没有断点续传能力。妥协与应对因此本方案不适用于执行周期在数小时以上的超长事务。对于多 Agent 协同而言由于大部分 Agent 的执行周期在数十秒内我们可以通过在业务上层加入重试重入逻辑来容忍单点重启。2. 跨容器水平扩展的限制该调度器无法直接在多个容器实例Pod间进行任务分发与状态同步。如果需要让 Agent-1 在机器 A 运行Agent-2 在机器 B 运行本引擎无能为力。妥协与应对对于大多数应用而言将不同的 Agent 作为同一进程内的不同组件载入采用进程内通信是最高效、也是延迟最低的方案。只有当单个 Agent 节点需要的物理环境截然不同如某些需要大显存 GPU时才需要考虑将节点重构为 RPC 微服务调用。五、总结多 Agent 协同能不能在线上平稳支撑复杂的业务逻辑关键在于能否为无序的调用建立清晰的规矩。基于 DAG 与状态机的小型调度引擎能够以极低的设计与运维成本为进程内的多个 Agent 提供健壮的并发隔离、防环机制与超时传递。在落地实施时有三条原则需要牢记严防“坏节点”阻塞务必为每个 Agent 节点的内部 HTTP 调用设置不大于 15 秒的局部超时以防某一个 Agent 假死拖垮整个 DAG 的执行。设计降级分支对于易失败的节点可以通过在 DAG 中设计兜底的 Mock 节点进行容错避免单个节点的失败频繁中断整条业务链路。数据清洗与隔离前置节点的 Output 在传递给后置节点时必须设计简单的字段清洗过滤器避免大模型产生的冗余 JSON 数据导致后续节点解析失败。综上所述通过轻量级 DAG 与有限状态机控制能够以极低成本接管多 Agent 协同生命周期保障长链路协作系统的稳定与健壮运行。
大模型多Agent协同中的状态机管理：用 Go 实现一个轻量级 DAG 任务流引擎
大模型多Agent协同中的状态机管理用 Go 实现一个轻量级 DAG 任务流引擎一、深度引言与场景痛点当大模型应用从单兵作战的 Prompt 调优走向复杂业务场景时单个 Agent智能体的能力上限便会迅速暴露。为了处理复杂的长链路业务业界普遍转向多 Agent 协同架构Multi-Agent Collaboration。在这种架构下不同的 Agent 扮演不同的角色一个负责用户需求拆解一个负责从向量数据库检索数据另一个负责代码生成或逻辑校验最后由一个专家 Agent 进行最终汇总。然而一旦多个 Agent 开始频繁交互工程上的管理灾难便接踵而至。最显著的痛点是协同状态的失控。多个 Agent 之间的调用关系极其容易陷入死锁或无限递归。例如Agent A 发现生成的数据不合格将其打回给 Agent B 重构而 Agent B 又因为某些前置条件未满足再次请求 Agent A最终导致在有限的 Token 预算内发生无休止的“套娃调用”。其次是并发与拓扑依赖管理的混乱。在长链路业务中有些 Agent 的执行是相互独立的如同时进行代码安全扫描和逻辑合规性审查可以并发运行而有些 Agent 必须等待前置任务全部完成如必须等扫描和审查都通过后才能交由发布 Agent 处理。如果仅通过在代码里手工嵌套go channel或硬编码状态字段很快就会把代码库变成不可维护的“意大利面条”。大厂的常用解法是直接引入 Temporal、Apache Airflow 等重量级分布式工作流引擎。但对于追求高 ROI、资源吃紧的创业团队或小研发团队来说这无异于高射炮打蚊子。引入这些庞然大物不仅意味着成倍的服务器运维成本更会带来漫长的开发调试链路。最务实、最轻量级的工程方案是在应用内存中构建一个基于有向无环图DAG, Directed Acyclic Graph的任务流引擎配合有限状态机FSM, Finite State Machine来统一接管多 Agent 协同的生命周期与数据依赖。二、底层机制与原理深度剖析要在内存中优雅地管理多 Agent 协同我们必须将协同流程抽象为一个有向无环图DAG其中节点Node/Vertex代表单个 Agent 的执行任务。有向边Edge代表任务之间的执行先后顺序及数据依赖。在 DAG 中任何节点在被触发执行前必须确保其所有的前置依赖节点In-degree 参入度都已成功执行完毕。为了实现这一点引擎底层的核心算法是拓扑排序Topological Sort通过计算每个节点的入度来决定任务的并发调度顺序。以下是该并发任务流协作机制的 Mermaid 原理架构图清晰地展示了一个用户请求进来后DAG 引擎是如何动态调度并隔离各个 Agent 节点的flowchart TD A[用户请求/目标输入] -- B[DAG 引擎解析依赖] B -- C{计算拓扑排序} C --|无前置依赖| D[Agent-1 需求拆解] C --|无前置依赖| E[Agent-2 知识库检索] D --|产生依赖关系| F{依赖校验中心} E --|产生依赖关系| F F --|前置节点全部完成| G[Agent-3 逻辑校验与生成] G -- H[Agent-4 专家评审/最终输出] H -- I[执行完毕/状态释放] subgraph 有限状态机 FSM 控制器 D -.-|更新状态| FSM[FSM: Init - Running - Success/Failed] E -.-|更新状态| FSM G -.-|更新状态| FSM H -.-|更新状态| FSM end这个架构的底层流转逻辑如下FSM 控制状态变更每个 Agent 节点都有自己的生命周期Pending、Running、Success、Failed、Skipped。引擎通过原子的 CASCompare-And-Swap操作来保证在多协程并发下状态流转的安全性。信号量与通道调度引擎在检测到某个节点的前置依赖全部变为 Success 后会自动将其送入执行通道由协程池并发处理。动态背压与数据传递前置 Agent 的输出Output会自动绑定为后置 Agent 的输入Input并且通过上下文 Context 传递超时控制。三、生产级代码实现与最佳实践大厂的烂代码才喜欢通过堆砌抽象类来展示架构设计我们的原则是KISS。下面我将分享一个用 Go 语言实现的高性能、就地运行的轻量级 DAG 任务流调度引擎核心代码package dagengine import ( context errors fmt sync ) // State 代表任务节点的执行状态 type State string const ( StatePending State PENDING StateRunning State RUNNING StateSuccess State SUCCESS StateFailed State FAILED ) // AgentFunc 代表具体的 Agent 执行函数接受上下文和输入数据返回输出数据或错误 type AgentFunc func(ctx context.Context, input interface{}) (interface{}, error) // TaskNode 代表 DAG 中的任务节点 type TaskNode struct { ID string Action AgentFunc DependsOn []string // 依赖的前置节点 ID 列表 State State // 当前状态 Input interface{} Output interface{} Err error mu sync.RWMutex } // SetState 安全地更新节点状态 func (n *TaskNode) SetState(s State) { n.mu.Lock() defer n.mu.Unlock() n.State s } // GetState 安全地获取节点状态 func (n *TaskNode) GetState() State { n.mu.RLock() defer n.mu.RUnlock() return n.State } // DAGEngine 负责管理并并发执行有向无环图任务流 type DAGEngine struct { nodes map[string]*TaskNode mu sync.RWMutex } // NewDAGEngine 创建一个新的 DAG 引擎实例 func NewDAGEngine() *DAGEngine { return DAGEngine{ nodes: make(map[string]*TaskNode), } } // AddTask 向引擎中注册一个任务节点 func (e *DAGEngine) AddTask(id string, fn AgentFunc, dependsOn []string) error { e.mu.Lock() defer e.mu.Unlock() if _, exists : e.nodes[id]; exists { return fmt.Errorf(任务节点 %s 已经存在, id) } e.nodes[id] TaskNode{ ID: id, Action: fn, DependsOn: dependsOn, State: StatePending, } return nil } // Run 并发执行 DAG 任务流直到所有任务执行完成或者检测到环、执行失败 func (e *DAGEngine) Run(ctx context.Context, initialInput map[string]interface{}) error { if err : e.hasCycle(); err ! nil { return fmt.Errorf(DAG 任务流存在循环依赖错误: %w, err) } // 初始化初始节点的输入数据 for id, val : range initialInput { if node, exists : e.nodes[id]; exists { node.Input val } } var wg sync.WaitGroup errChan : make(chan error, len(e.nodes)) ctx, cancel : context.WithCancel(ctx) defer cancel() for { e.mu.RLock() allDone : true anyFailed : false var tasksToRun []*TaskNode for _, node : range e.nodes { state : node.GetState() if state StateFailed { anyFailed true break } if state ! StateSuccess { allDone false } // 如果是 pending 状态且前置依赖已全部成功则本轮可以运行 if state StatePending e.dependenciesSatisfied(node) { tasksToRun append(tasksToRun, node) } } e.mu.RUnlock() if anyFailed { return errors.New(任务流中存在节点执行失败引擎已终止运行) } if allDone { break } // 如果没有任务可以运行但还有任务没运行成功说明发生死锁可能有环漏检但前面已有检测保证 if len(tasksToRun) 0 { timeChan : make(chan struct{}) // 阻塞等待并发任务通知 _ timeChan // 生产中可以用 channel 信号通知这里简化为轮询间隔 break } // 并发调度本轮准备就绪的任务 for _, node : range tasksToRun { node.SetState(StateRunning) wg.Add(1) go func(n *TaskNode) { defer wg.Done() // 收集前置依赖节点的输出作为输入数据源简单合并 n.Input e.gatherInputs(n) output, err : n.Action(ctx, n.Input) if err ! nil { n.Err err n.SetState(StateFailed) errChan - fmt.Errorf(节点 %s 执行失败: %w, n.ID, err) cancel() // 发生错误撤销整个链路的上下文 return } n.Output output n.SetState(StateSuccess) }(node) } wg.Wait() } select { case err : -errChan: return err default: return nil } } // dependenciesSatisfied 检查节点的所有依赖是否已全部 SUCCESS func (e *DAGEngine) dependenciesSatisfied(node *TaskNode) bool { for _, depID : range node.DependsOn { depNode, exists : e.nodes[depID] if !exists || depNode.GetState() ! StateSuccess { return false } } return true } // gatherInputs 汇总前置依赖节点的 Output func (e *DAGEngine) gatherInputs(node *TaskNode) map[string]interface{} { inputs : make(map[string]interface{}) for _, depID : range node.DependsOn { if depNode, exists : e.nodes[depID]; exists { inputs[depID] depNode.Output } } return inputs } // hasCycle 拓扑排序进行环路检测Kahn 算法思想 func (e *DAGEngine) hasCycle() error { inDegree : make(map[string]int) adjList : make(map[string][]string) for _, node : range e.nodes { inDegree[node.ID] len(node.DependsOn) for _, dep : range node.DependsOn { adjList[dep] append(adjList[dep], node.ID) } } queue : make([]string, 0) for id, deg : range inDegree { if deg 0 { queue append(queue, id) } } visitedCount : 0 for len(queue) 0 { curr : queue[0] queue queue[1:] visitedCount for _, neighbor : range adjList[curr] { inDegree[neighbor]-- if inDegree[neighbor] 0 { queue append(queue, neighbor) } } } if visitedCount ! len(e.nodes) { return errors.New(依赖关系中检测到循环引用 (Cycle Detected)) } return nil }这段代码实现了引擎的核心精髓防爆并发没有像微服务编排那样引入复杂的分布式事务锁而是依赖于sync.RWMutex和状态机流转控制。环路预防在执行前进行 Kahn 拓扑图环路检测从源头防止出现死锁循环。统一上下文取消任何一个 Agent 执行失败立即通过取消 Context 广播到所有子任务极大释放系统算力并防止 Token 资源被白白浪费。四、边界分析与架构权衡基于内存的轻量级 DAG 任务流引擎极其适合快速上线和中小规模流量但为了在生产中做好防护必须了解其技术边界与架构的妥协1. 单点故障与状态丢失由于该引擎的所有节点运行状态和数据交互都保存在 Go 进程的内存中一旦服务器发生宕机或进行灰度重启运行中的任务状态将彻底丢失且没有断点续传能力。妥协与应对因此本方案不适用于执行周期在数小时以上的超长事务。对于多 Agent 协同而言由于大部分 Agent 的执行周期在数十秒内我们可以通过在业务上层加入重试重入逻辑来容忍单点重启。2. 跨容器水平扩展的限制该调度器无法直接在多个容器实例Pod间进行任务分发与状态同步。如果需要让 Agent-1 在机器 A 运行Agent-2 在机器 B 运行本引擎无能为力。妥协与应对对于大多数应用而言将不同的 Agent 作为同一进程内的不同组件载入采用进程内通信是最高效、也是延迟最低的方案。只有当单个 Agent 节点需要的物理环境截然不同如某些需要大显存 GPU时才需要考虑将节点重构为 RPC 微服务调用。五、总结多 Agent 协同能不能在线上平稳支撑复杂的业务逻辑关键在于能否为无序的调用建立清晰的规矩。基于 DAG 与状态机的小型调度引擎能够以极低的设计与运维成本为进程内的多个 Agent 提供健壮的并发隔离、防环机制与超时传递。在落地实施时有三条原则需要牢记严防“坏节点”阻塞务必为每个 Agent 节点的内部 HTTP 调用设置不大于 15 秒的局部超时以防某一个 Agent 假死拖垮整个 DAG 的执行。设计降级分支对于易失败的节点可以通过在 DAG 中设计兜底的 Mock 节点进行容错避免单个节点的失败频繁中断整条业务链路。数据清洗与隔离前置节点的 Output 在传递给后置节点时必须设计简单的字段清洗过滤器避免大模型产生的冗余 JSON 数据导致后续节点解析失败。综上所述通过轻量级 DAG 与有限状态机控制能够以极低成本接管多 Agent 协同生命周期保障长链路协作系统的稳定与健壮运行。
