AI Agent 系统设计与多 Agent 协作架构：别让演示效果骗了你

Demo 演示视频里，三个 Agent 互相发消息、写代码、做 Code Review，过程流畅得就像三个顶尖程序员在配合。然而把这套代码放到本地多进程环境里一测，并发调到 5 个 Task，系统瞬间卡死。看日志，Agent A 在等待 Agent B 输出 JSON 校验结果，而 Agent B 却卡在 Agent A 的锁文件释放通知上。

这就是多 Agent 系统开发中最常见的陷阱：把演示状态下的单线程、顺序化脚本，误以为是可复现、可调度的多 Agent 协作架构。

---

Demo 演示顺畅，一上并发就死锁：多 Agent 状态交接的现场死穴

在单机本地测试多 Agent 协作（例如 Planner -> Executor -> Reviewer）时，开发人员往往习惯通过 Python 全局变量或简陋的内存队列（queue.Queue）传递上下文。如果在 Agent 交互中加入 LLM 调用延迟，整个系统极其脆弱。

看一个典型的卡死现场日志片段：

2026-08-10 10:14:02.102 | ERROR    | agent.scheduler:dispatch:89 - Deadlock detected!
2026-08-10 10:14:02.103 | DEBUG    | agent.planner:run:45 - [Planner-01] Waiting for lock: 'task_ref_9921.lock'
2026-08-10 10:14:02.103 | DEBUG    | agent.executor:run:67 - [Executor-03] Blocked on API stream response...
2026-08-10 10:14:02.104 | WARNING  | agent.reviewer:run:12 - [Reviewer-02] Recv message failed: TimeoutError after 300s

问题根源通常体现在三点：

1. 隐式共享状态：Agent 之间没有严格的数据隔离界限，往往直接修改同一个 context 对象。
2. 缺乏确定性回放机制：LLM 输出的不确定性导致每次运行的 Token 长度、响应时间不同，无法复现死锁条件。
3. 消息传递缺乏 Ack 机制：异步消息投递后，发送方以为接收方已处理完毕，导致状态机严重不同步。

为了直观展示 Agent 状态死锁与解耦后的事件总线交互流程，我们可以对比以下时序模型：

sequenceDiagram
    autonumber
    participant P as Planner Agent
    participant E as Executor Agent
    participant EB as Event Bus (Redis/ZeroMQ)
    participant State as State Store (SQLite/Memory)

    P->>State: 1. 写入任务 Payload 并锁住 TaskID
    P->>EB: 2. 发布事件 "TASK_CREATED"
    EB->>E: 3. 广播分发 TASK_CREATED
    alt 异步处理
        E->>State: 4. 尝试获取 TaskID 状态
        State-->>E: 5. 状态被 P 锁住，触发 Retrying...
        E-->>EB: 6. 抛出 Timeout 错误并释放连接
    end
    Note over P,E: 解耦后：摒弃共享锁，改为状态变更通知与版本号控制

---

从单进程 Mock 到分布式拓扑：可复现实验脚手架的目录设计

要搞多 Agent 协作，第一件事不是急着写 Prompt，而是先搭建一个能够在本地无 API 密钥（或 Mock 环境）下跑通的实验脚手架。工程脚手架的目录结构应当清晰隔离 Agent 逻辑、环境模拟器与测试回放。

推荐的生产级脚手架工程目录：

agent_sandbox/
├── config/
│   ├── env.development.yaml
│   └── topology_mock.json
├── core/
│   ├── bus.py            # 事件总线实现
│   ├── state.py          # 状态持久化与 snapshot
│   └── router.py         # Agent 消息路由
├── agents/
│   ├── base.py
│   ├── planner.py
│   └── executor.py
├── mocks/
│   ├── llm_server.py     # 本地 mock llm HTTP 服务
│   └── mock_responses/   # 预录制的 JSON 交互文件
├── tests/
│   ├── test_replay.py    # 确定性回放测试
│   └── test_deadlock.py  # 并发死锁压力测试
└── run_harness.py        # 脚手架启动入口

核心思路在于：将 Agent 内部逻辑（Prompt 组合、工具调用决策）与 Agent 间的通信机制完全剥离。Agent 只向 EventBus 发送 Struct 格式的 Message，由 Router 完成投递。

---

AgentEnvironment 状态隔离与 Event Bus 抓包验证

下面是用 Python asyncio 实现的一个隔离型 AgentEnvironment 与简易消息总线。它利用事件驱动机制替代互斥锁，彻底规避死锁风险：

import asyncio
import json
import logging
from typing import Dict, Any, Callable
from dataclasses import dataclass, asdict

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")

@dataclass
class AgentMessage:
    msg_id: str
    sender: str
    receiver: str
    action: str
    payload: Dict[str, Any]
    version: int = 1

class AgentEventBus:
    def __init__(self):
        self._subscribers: Dict[str, list[Callable]] = {}
        self._message_log: list[AgentMessage] = []

    def subscribe(self, event_type: str, callback: Callable):
        if event_type not in self._subscribers:
            self._subscribers[event_type] = []
        self._subscribers[event_type].append(callback)

    async def publish(self, message: AgentMessage):
        self._message_log.append(message)
        logging.info(f"[Bus] Published {message.msg_id} from {message.sender} -> {message.receiver}")
        
        if message.action in self._subscribers:
            tasks = [cb(message) for cb in self._subscribers[message.action]]
            await asyncio.gather(*tasks)

class BaseAgent:
    def __init__(self, name: str, bus: AgentEventBus):
        self.name = name
        self.bus = bus
        self.memory: Dict[str, Any] = {}

    async def send(self, receiver: str, action: str, payload: Dict[str, Any]):
        msg = AgentMessage(
            msg_id=f"msg_{len(self.bus._message_log)+1:04d}",
            sender=self.name,
            receiver=receiver,
            action=action,
            payload=payload
        )
        await self.bus.publish(msg)

class PlannerAgent(BaseAgent):
    async def handle_task(self, msg: AgentMessage):
        logging.info(f"[{self.name}] Received task response from {msg.sender}")
        self.memory["last_result"] = msg.payload.get("status")

async def main():
    bus = AgentEventBus()
    planner = PlannerAgent("Planner-01", bus)
    
    # 订阅响应事件，避免阻塞轮询
    bus.subscribe("EXECUTION_DONE", planner.handle_task)
    
    # 模拟触发
    await planner.send("Executor-01", "EXECUTE_TASK", {"cmd": "build_index", "target": "/data"})

if __name__ == "__main__":
    asyncio.run(main())

运行该脚本后，可以通过系统的 TCP / Mock 日志命令进行事件流捕获。如果将 Event Bus 映射到 ZeroMQ 或 Redis Pub/Sub，我们可以在终端中使用以下命令进行现场抓包诊断：

# 查看 Redis 频道中的 Agent 消息流动
redis-cli monitor | grep "AgentMessage"

# 或使用 pytest 验证高并发场景下是否有 Message 丢失
pytest tests/test_deadlock.py -n 4 --log-cli-level=INFO

---

避免上下文死锁：本地 Mock Server 与 Deterministic Replay 配置

要验证多 Agent 系统在大规模测试下的稳定性，决不能依赖真实的 OpenAI API 或开源 LLM 实时生成。原因很直接：网络延迟不确定，API 限流（429 Too Many Requests）以及生成文本微小的随机性，会导致排查死锁变成一场噩梦。

我们需要使用轻量化的 Mock API Server 替换真实模型请求。一个简单且高效的方法是使用 Python aiohttp 或 FastAPI 搭配预先录制的响应日志文件。

# 启动本地 Mock LLM API 服务
python mocks/llm_server.py --port 8080 --mock-data mocks/mock_responses/case_01.json &

# 使用 curl 验证 Mock 响应延迟是否稳定在 10ms 以内
curl -X POST http://localhost:8080/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{"model": "mock-llm", "messages": [{"role": "user", "content": "ping"}]}'

在测试脚手架中设置 DETERMINISTIC_SEED=42 和 MOCK_LLM_URL=http://localhost:8080/v1。当并发测试跑到 100 次循环时，只要出现一次 Message ACK 丢失或死锁，终端脚手架就能精确输出当前运行步骤的步数与 snapshot 保存路径：

# 运行确定性回放断言脚本
python run_harness.py --replay-log logs/crash_20260810_0912.json --check-deadlock

一旦实验脚手架能够在本地 100% 确定性地复现并发碰撞问题，多 Agent 系统的协作架构设计才算真正完成了第一步。别被演示画面中的“表面和谐”骗了，只有建立起严格的事件总线与确定性测试脚手架，Agent 的工程落地才谈得上可靠。

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/2401_87746054/article/details/163643572