MultiAgent-Coop: 低开销多智能体协作系统

面向多智能体协作的低开销通信、状态传递与共享记忆机制 —— 赛题9。

简介

基于 ZeroMQ 消息总线 + shared_memory 共享状态 + FAISS 向量记忆的三层架构，实现三个专用 Agent（协调者/研究者/执行者）之间的高效协作。通信延迟低至亚毫秒级，通信开销仅占总耗时的 6.8%。

核心亮点

🚀 低开销通信 — ZeroMQ 消息总线

Ping-Pong 同步模式实测单程延迟（100次取平均）： | 消息大小 | 平均延迟 | P95延迟 | 吞吐量 | |———|———|——–|——-| | 100B | 0.27 ms | 0.39 ms | 0.35 MB/s | | 1KB | 0.31 ms | 0.42 ms | 3.11 MB/s | | 10KB | 0.35 ms | 0.50 ms | 27.8 MB/s | | 100KB | 0.68 ms | 0.95 ms | 142.9 MB/s | | 500KB | 3.32 ms | 4.58 ms | 147.2 MB/s |

通信传输占总耗时的 **6.8%**（< 20% 目标），证明通信不是系统瓶颈。

⚡ 高效状态传递 — SharedMemory

共享内存传递 float 数组比 ZeroMQ 消息总线序列化传递快 1.5x ~ 4.0x，数据量越大优势越明显（1024 floats 加速 4.0 倍），适合高频向量传递场景。

🧠 智能记忆复用 — FAISS 向量检索

基于 BGE-small-zh 的语义向量存储 + FAISS 近似最近邻搜索，在 3 轮递进任务对比中：

有记忆组：3 轮准确率 100% → 100% → **100%**（逐轮积累知识）
无记忆组：3 轮准确率 100% → 100% → **40%**（无法跨越检索历史信息）
差异：**+20%**（第3轮）

系统架构

三 Agent 角色

┌──────────────────────────────────────────────────────────┐
│                   CoordAgent（协调者）                      │
│  负责任务拆解（LLM）、子任务调度、结果汇总、记忆存储        │
└────────────┬────────────────────────────┬────────────────┘
             │                            │
    ┌────────▼────────┐          ┌────────▼────────┐
    │  ResearchAgent  │          │   ExecAgent     │
    │ （研究者）       │          │ （执行者）       │
    │ 记忆检索 + LLM  │          │ 沙箱代码执行    │
    │ 分析 + 网页搜索 │          │ 工具调用       │
    └────────┬────────┘          └────────┬────────┘
             │                            │
    ┌────────▼────────────────────────────▼────────┐
    │              三层基础架构                      │
    │  ┌──────────────────────────────────────┐    │
    │  │ 通信层：ZeroMQ PUB/SUB 消息总线       │    │
    │  ├──────────────────────────────────────┤    │
    │  │ 状态层：SharedMemory 共享内存         │    │
    │  ├──────────────────────────────────────┤    │
    │  │ 记忆层：SQLite + FAISS 向量记忆       │    │
    │  └──────────────────────────────────────┘    │
    └─────────────────────────────────────────────┘

三层架构说明

通信层 — ZeroMQ PUB/SUB 消息总线
- 单例模式，进程内/跨进程零配置通信
- 基于 topic 的路由 + 回调注册
- 自动端口分配，避免冲突
状态层 — multiprocessing.shared_memory 共享内存
- 零拷贝传递 numpy 数组（避免序列化开销）
- TTL 自动过期，惰性清理
- pack_state_ref / unpack_state_ref 跨进程引用传递
记忆层 — SQLite + FAISS 向量记忆
- BGE-small-zh 中文语义向量（512维）
- FAISS IndexFlatL2 精确最近邻搜索
- 访问计数 + 热度追踪

目录结构

submit_final/
├── README.md                  # 本文件
├── requirements.txt            # 依赖清单
├── src/                        # 核心源代码
│   ├── config.py               # 全局配置（路径/API Key维一管理）
│   ├── main.py                 # 系统入口
│   ├── agents/                 # 三个 Agent 实现
│   │   ├── base_agent.py       # Agent 基类（抽象）
│   │   ├── coord_agent.py      # 协调者 Agent
│   │   ├── research_agent.py   # 研究者 Agent
│   │   ├── exec_agent.py       # 执行者 Agent
│   │   └── retriever.py        # 网页检索器
│   ├── comm/
│   │   └── message_bus.py      # ZeroMQ 消息总线
│   ├── embedding/
│   │   └── embedding_engine.py # BGE 向量编码引擎
│   ├── llm/
│   │   └── deepseek_client.py  # DeepSeek API 客户端
│   ├── memory/
│   │   ├── memory_store.py     # SQLite 记忆存储
│   │   ├── memory_unit.py      # 记忆单元数据结构
│   │   └── vector_index.py     # FAISS 向量索引
│   ├── sandbox/
│   │   ├── code_executor.py    # 安全代码执行沙箱
│   │   └── security.py         # AST 安全检查器
│   ├── state/
│   │   └── shared_state.py     # 共享状态管理器
│   ├── tasks/
│   │   ├── task_group_a.py     # 任务组 A
│   │   └── task_group_b.py     # 任务组 B
│   └── evaluation/
│       └── run_benchmark.py    # 评测脚本
└── test/
    ├── test_integration.py     # 整合测试（6项）
    ├── benchmark.py            # 性能基准测试
    └── test_token_saving.py    # Token 消耗对比测试（需 DeepSeek API Key）

环境要求

Python 3.8+
操作系统：Linux / macOS / Windows
内存：≥ 4GB（加载 BGE 模型约需 2GB）
磁盘：≥ 500MB

安装

# 1. 进入项目目录
cd submit_final

# 2. 安装依赖
pip install -r requirements.txt

# 3. 验证安装
python -m pytest test/test_integration.py -v

快速开始

注意：运行 LLM 功能前需要先设置 DEEPSEEK_API_KEY 环境变量。可在 platform.deepseek.com 申请 API Key。

# 设置 API Key（Windows 用 set 替换 export）
export DEEPSEEK_API_KEY="your-api-key"

# 运行单任务
cd submit_final
python src/main.py --task A1

# 运行所有任务
python src/main.py --all

# 运行整合测试
python -m pytest test/test_integration.py -v

# 运行性能基准测试
python test/benchmark.py --quick

# 运行 Token 消耗对比测试
python test/test_token_saving.py

环境变量

变量名	必需	说明
`DEEPSEEK_API_KEY`	必需	DeepSeek API Key。运行 LLM 功能（`main.py`、`test_token_saving.py`）时必须设置。可在 platform.deepseek.com 申请

性能测试

# 基础性能基准测试（不依赖 LLM）
python test/benchmark.py --quick

# Token 消耗对比测试（需设置 DEEPSEEK_API_KEY）
export DEEPSEEK_API_KEY="your-key"
python test/test_token_saving.py

test/benchmark.py：4 项基础性能测试，输出详尽数据和汇总表格，结果保存为 output/results/benchmark_data.json。
test/test_token_saving.py：对比纯文本通信与向量通信两种模式在 3 轮递进任务下的 LLM token 消耗差异，必须设置 DEEPSEEK_API_KEY 后才能运行。使用 tiktoken（cl100k_base）逐次统计 token，结果保存为 output/results/token_saving.json。

整合测试

python -m pytest test/test_integration.py -v

6 项测试覆盖：消息总线、Embedding 编码、记忆存读、Agent 通信、共享状态、代码沙箱。

团队

松鼠突击队