多智能体协作先画清状态边界

在大型系统或复杂工作流场景中，当多 Agent 协作系统在处理跨部门审批与发票核验任务时，如果缺乏死锁防护与路径追踪机制，主控 Agent 的 Task Worker 容易陷入并发卡死状态，协程数量在短时间内可能急剧飙升。

在缺乏链路治理的情况下，系统接口可能直接返回 504 超时。检索日志会发现一秒内触发大量重复 Prompt 请求，但由于状态流转陷入死锁，没有任何一个任务能够完成并输出最终结果。

单个工具调用看起来正常，并不意味着协作链路也安全。角色拆分后，状态流转、重试和上下文格式都会增加排查难度；先补齐可关联的证据链，再判断问题是否来自模型或编排。

1. 现象复盘：环形依赖如何吞干协程池

在高并发协同异常发生时，系统的 CPU 利用率可能处于正常区间（例如 35% 左右），并非典型的计算密集型瓶颈。通过解析 Goroutine 堆栈 dump 可以发现，大量 Go 协程阻塞在通道接收（channel receive）与 HTTP 响应等待上。

标准的业务协同链路通常为单向 DAG 拓扑：用户输入诉求 -> 规划 Agent 拆解任务 -> 校验 Agent 检查权限 -> 执行 Agent 调用后端 API。然而在复杂业务场景中，若校验 Agent 发现权限不足并触发“补全权限”的辅助 Action，而该 Action 再次将请求重定向给规划 Agent 时，系统将产生隐蔽的循环依赖。

当模型在多轮对话中输出了格式微变的 JSON 结构时，解析器触发了容错重试机制。校验 Agent 若误判此请求为全新任务并重新触发规划 Agent，两个 Agent 之间便形成了环形死锁（Circular Lock）。由于请求未携带全局 TraceID 以及跳数（Hop Limit）限制，智能体间在短时间内快速交替发送消息，填充上下文缓冲区，最终导致后端 Worker 阻塞失效。

2. 缺失的证据链：非确定性系统排障的三个关键瓶颈

在传统微服务架构中，故障定位主要依赖 Prometheus 的 HTTP 5xx 指标与 OpenTelemetry 的 Trace 链路追踪。但在多 Agent 协作框架下，传统监控手段难以覆盖非确定性的上下文变化，主要存在三个关键瓶颈：

1. 状态流转丢失现场：日志虽然记录了消息的发送动作，但缺少 Agent 状态机在关键节点上的内存 Snapshot。当模型做出异常决策时，无法复原当时的历史上下文与环境变量。
2. 工具调用缺乏上下文凭证：执行 Agent 调用后端数据库接口时，在 Trace 链路中仅表现为标准的 HTTP POST 请求，未将调用凭证与模型输入的 Cause Prompt 及 Tool Call ID 进行强绑定。
3. 缺少环路计数器：系统未在 Agent 消息体中传递类似于网络协议中 TTL（Time To Live）的计数器，导致无限递归调用无法在协同入口处被及时识别与强行拦截。

3. 生产级防护：带 DAG 拓扑校验与 TTL 防线的中控代码

为了有效治理非确定性死锁，工程实践中需要在 Agent 消息分发层构建一套具备状态存根、TTL 递减与环路检测功能的 GateKeeper 中控机制。

以下是用 Go 语言实现的 Agent 协同中控处理逻辑：

package agentgate

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"sync"
	"time"
)

var (
	ErrMaxHopsExceeded    = errors.New("agent collaboration max hops exceeded")
	ErrCircularDependency = errors.New("detected circular agent execution path")
	ErrExecutionTimeout   = errors.New("agent execution context timeout")
)

// AgentMessage 封装 Agent 之间传递的控制流载荷
type AgentMessage struct {
	TraceID     string            `json:"trace_id"`
	ParentID    string            `json:"parent_id"`
	HopCount    int               `json:"hop_count"`
	MaxHops     int               `json:"max_hops"`
	Sender      string            `json:"sender"`
	Receiver    string            `json:"receiver"`
	VisitedPath []string          `json:"visited_path"`
	Payload     string            `json:"payload"`
	Metadata    map[string]string `json:"metadata"`
}

// ExecutionEvidence 异常现场捕获证据存根
type ExecutionEvidence struct {
	Timestamp    time.Time `json:"timestamp"`
	TraceID      string    `json:"trace_id"`
	SnapshotHash string    `json:"snapshot_hash"`
	PathHistory  []string  `json:"path_history"`
	LastError    string    `json:"last_error"`
}

// CircuitMediator 中控调配器
type CircuitMediator struct {
	mu           sync.RWMutex
	evidencePool map[string]*ExecutionEvidence
}

func NewCircuitMediator() *CircuitMediator {
	return &CircuitMediator{
		evidencePool: make(map[string]*ExecutionEvidence),
	}
}

// Dispatch 校验并分发 Agent 消息，阻断死循环
func (m *CircuitMediator) Dispatch(ctx context.Context, msg *AgentMessage) (*ExecutionEvidence, error) {
	// 1. 检查 TTL 限制
	if msg.HopCount >= msg.MaxHops {
		evidence := m.recordEvidence(msg, ErrMaxHopsExceeded)
		return evidence, fmt.Errorf("%w: current %d, max %d", ErrMaxHopsExceeded, msg.HopCount, msg.MaxHops)
	}

	// 2. 检查 DAG 环路依赖
	for _, node := range msg.VisitedPath {
		if node == msg.Receiver {
			evidence := m.recordEvidence(msg, ErrCircularDependency)
			return evidence, fmt.Errorf("%w: node %s already in path %v", ErrCircularDependency, msg.Receiver, msg.VisitedPath)
		}
	}

	// 3. 更新执行链路路径
	msg.HopCount++
	msg.VisitedPath = append(msg.VisitedPath, msg.Sender)

	// 4. 带 Timeout 约束的执行上下文
	evalCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
	defer cancel()

	done := make(chan error, 1)
	go func() {
		// 模拟 Agent 实际推导与 Tool 执行
		done <- m.executeStep(evalCtx, msg)
	}()

	select {
	case <-evalCtx.Done():
		evidence := m.recordEvidence(msg, ErrExecutionTimeout)
		return evidence, ErrExecutionTimeout
	case err := <-done:
		if err != nil {
			evidence := m.recordEvidence(msg, err)
			return evidence, err
		}
	}

	return nil, nil
}

func (m *CircuitMediator) executeStep(ctx context.Context, msg *AgentMessage) error {
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-time.After(100 * time.Millisecond):
		return nil
	}
}

func (m *CircuitMediator) recordEvidence(msg *AgentMessage, execErr error) *ExecutionEvidence {
	m.mu.Lock()
	defer m.mu.Unlock()

	hasher := sha256.New()
	hasher.Write([]byte(fmt.Sprintf("%s:%s:%s", msg.TraceID, msg.Sender, msg.Payload)))
	hashStr := hex.EncodeToString(hasher.Sum(nil))

	evidence := &ExecutionEvidence{
		Timestamp:    time.Now(),
		TraceID:      msg.TraceID,
		SnapshotHash: hashStr,
		PathHistory:  append(msg.VisitedPath, msg.Receiver),
		LastError:    execErr.Error(),
	}

	m.evidencePool[msg.TraceID] = evidence
	return evidence
}

上述实现的的核心原则是：不依赖 LLM 模型的自我判断来结束对话，而是通过外层的确定性程序进行强收口。当检测到 HopCount 超过预设上限或 VisitedPath 出现重复节点时，系统立即阻断当前调用流，并生成包含快照 Hash 与链路历史的 ExecutionEvidence 证据存根。

4. 性能验证与指标对比

部署防线机制后，使用压力测试工具对系统进行连续高并发场景模拟。在测试中故意注入含有模糊歧义的 Prompt 以诱发模型死循环。

在演练中，可以注入模糊输入来观察环路检测是否会触发，并记录阻断位置、耗时和降级结果。不同模型与下游依赖下的数值不应直接套用。

完成上述改造后，原本难以捕获的非确定性交互异常转化为了可被 Prometheus 实时监控与告警的确定性指标。

5. 治理多 Agent 系统的三条硬规则

要确保多 Agent 协作系统在生产环境中稳定运行，工程设计上需要严格遵循以下原则：

• 永远不要信任模型能自我终止：必须在 Protocol 传输协议层显式设置强硬的 Hop Limit 与全局超时机制。
• Trace 必须携带模型快照 Hash：仅仅记录日志无法满足推断可复现性要求，必须同时记录决策上下文的完整 Hash 指纹。
• 降级兜底路径必须基于确定性代码：当 Agent 协作发生冲突或死锁时，系统应直接回退至确定性的规则引擎或运维流程，严禁使用新的 Agent 去修补上一级 Agent 产生的异常。