# 多 Agent 协调

当你有多个 LLM Agent 处理相关任务时，Synapse 提供协调层 — 共享记忆、任务分配与异步聊天。

## 模式

### 模式 1：共享 Mind（单一真相源）

所有 Agent 共享一个 Mind Key。它们读写同一个记忆存储。

```
┌──────────┐  ┌──────────┐  ┌──────────┐
│ Agent A  │  │ Agent B  │  │ Agent C  │
└────┬─────┘  └────┬─────┘  └────┬─────┘
     │             │             │
     └─────────────┼─────────────┘
                   ▼
           ┌──────────────┐
           │  共享 Mind    │
           │  (一个 key)   │
           └──────────────┘
```

**用例：** 一组 Agent 共同处理一个项目。

**配置：**

```bash
# 所有 Agent 使用同一 Mind Key
export SYNAPSE_MIND_KEY=mk_shared_key...
```

**通过任务协调：**

```python
# Agent A 创建任务
create_task("Review PR #42", priority="high")

# Agent B 接手
tasks = list_tasks(status="pending")
if tasks:
    task = tasks[0]
    update_task(task["id"], status="in_progress")
    # ... 执行工作 ...
    update_task(task["id"], status="done")
```

### 模式 2：专用 Mind（隔离上下文）

每个 Agent 有自己的 Mind。它们通过一个共享的“协调”Mind 通信。

```
┌──────────┐  ┌──────────┐  ┌──────────┐
│ Coder    │  │ Reviewer │  │ Deployer │
│ Agent    │  │ Agent    │  │ Agent    │
└────┬─────┘  └────┬─────┘  └────┬─────┘
     │             │             │
     ▼             ▼             ▼
┌─────────┐  ┌─────────┐  ┌─────────┐
│ Mind C  │  │ Mind R  │  │ Mind D  │
└─────────┘  └─────────┘  └─────────┘
     │             │             │
     └─────────────┼─────────────┘
                   ▼
           ┌──────────────────┐
           │  协调 Mind        │
           │ (共享)            │
           └──────────────────┘
```

**用例：** 具有不同专长（编码、审查、部署）的 Agent。

**配置：**

```bash
# Coder agent
SYNAPSE_MIND_KEY=mk_coder... MCP_TRANSPORT=stdio npx synapse-mcp-api@latest

# Reviewer agent
SYNAPSE_MIND_KEY=mk_reviewer... MCP_TRANSPORT=stdio npx synapse-mcp-api@latest

# Deployer agent
SYNAPSE_MIND_KEY=mk_deployer... MCP_TRANSPORT=stdio npx synapse-mcp-api@latest
```

**通过共享 Mind 协调：**

```python
# Coder 存储“准备好审查”
COORDINATION_KEY = "mk_coordination..."
requests.post(f"{URL}/memory",
    headers={"Authorization": f"Bearer {COORDINATION_KEY}"},
    json={
        "category": "project",
        "key": "pr_42_ready",
        "content": "PR #42 is ready for review. Branch: feature/docs-system",
        "tags": ["review", "pr-42"],
        "priority": "high"
    })

# Reviewer 轮询审查请求
r = requests.get(f"{URL}/memory/search?q=ready+for+review",
    headers={"Authorization": f"Bearer {COORDINATION_KEY}"})
```

### 模式 3：中心辐射式（Orchestrator）

中央协调 Agent 把任务分配给工作 Agent。

```
        ┌──────────────┐
        │ Orchestrator │
        │    Agent     │
        └──────┬───────┘
               │
    ┌──────────┼──────────┐
    ▼          ▼          ▼
┌──────┐  ┌──────┐  ┌──────┐
│Worker│  │Worker│  │Worker│
│  A   │  │  B   │  │  C   │
└──────┘  └──────┘  └──────┘
```

**用例：** 复杂工作流，需要并行工作。

**实现：**

```python
# Orchestrator
class Orchestrator:
    def assign_task(self, worker_id, task_description):
        # 在 worker 的 Mind（或共享协调 Mind）中存储任务
        create_task(task_description, priority="high")
        # 通过聊天通知 worker
        reply(f"@{worker_id}: New task — {task_description}")
    
    def check_progress(self):
        tasks = list_tasks(status="in_progress")
        for t in tasks:
            print(f"{t['title']}: {t['status']}")

# Worker 轮询分配给它的任务
class Worker:
    def run(self):
        while True:
            tasks = list_tasks(status="pending")
            for t in tasks:
                if assigned_to_me(t):
                    update_task(t["id"], status="in_progress")
                    result = do_work(t)
                    update_task(t["id"], status="done")
                    reply(f"Completed: {t['title']}")
            time.sleep(60)
```

## 通过聊天协调

Agent 之间可通过聊天系统通信：

```python
# Agent A 发给 Agent B
reply("@agent-b: Can you review my PR?")

# Agent B 轮询并响应
for msg in poll_messages():
    if "@agent-b" in msg["content"]:
        reply(f"@agent-a: Sure, looking at it now.")
```

> [!NOTE]
> 聊天消息有角色标记。Agent 之间消息用 role=agent，人与 Agent 之间用 role=human。

## 通过变量协调

用变量进行轻量协调（锁、标志位）：

```python
# 获取锁
def acquire_lock(name):
    r = requests.post(f"{URL}/var",
        headers={"Authorization": f"Bearer {KEY}"},
        json={"key": f"lock_{name}", "value": "acquired"})
    return True

def release_lock(name):
    requests.delete(f"{URL}/var/lock_{name}",
        headers={"Authorization": f"Bearer {KEY}"})

# 使用
if acquire_lock("deploy"):
    try:
        deploy_to_production()
    finally:
        release_lock("deploy")
```

## 最佳实践

> [!TIP]
> - **为不同关注点使用独立 Mind** — 不要把 coder 和 reviewer 的记忆混在一起
> - **在聊天中标记 Agent** — `@agent-name` 让地址清晰
> - **用任务分配工作** — 而非聊天（聊天用于讨论）
> - **实现幂等性** — Agent 可能重试失败的操作
> - **记录一切** — 把决策存入记忆以备审计

## 下一步

- [持久化 LLM Agent](/docs/guides/persistent-llm-agent)
- [LLM Cookbook](/docs/llm-cookbook/session-start-pattern)
- [Webhook 自动化](/docs/guides/webhook-automation)