💡 本文核心：本文将从零开始，带你深入理解 Anthropic Claude 的多 Agent 协作机制，并手把手教你构建一个基于 Orchestrator-Worker 模式的自动化研究团队。更重要的是，我们将解决一个常见痛点：如何在终端中对多个并行运行的 Agent 进行可视化监控，让复杂的协作过程一目了然。无论你是 AI Agent 的初学者还是资深开发者，都能从中获得启发。

---

前言：为什么需要多 Agent 协作？

近年来，大型语言模型（LLM）的能力已从简单的文本生成扩展到执行复杂任务的自主代理（Agent）。但单个 Agent 面对日益复杂的现实世界问题时，往往会遇到瓶颈：上下文窗口的限制、顺序执行的低效、单一系统提示难以兼顾多个专业领域等。

为了突破这些限制，多 Agent 协作（Multi-Agent Collaboration） 应运而生。它通过构建一个由多个专业化 Agent 组成的团队，以分工、协作的方式共同完成单一 Agent 难以解决的宏大目标。这种范式的灵感来自人类社会的集体智慧：正如人类社会通过协作和分工实现了指数级的能力提升，Agent 团队同样可以通过协调实现远超个体的成就。

Anthropic 的内部评估数据印证了这一点：在其研究系统中，由 Claude 3 Opus 担任主导 Agent、Claude 3 Sonnet 担任子 Agent 的多 Agent 系统，在内部研究评估中比单一的 Claude 3 Opus Agent 提升了 90.2% 的性能 [1]。这一数字令人信服地证明了多 Agent 协作的价值。

核心概念：构建 Agent 系统的基石

在深入多 Agent 协作之前，我们必须首先理解构成这些系统的基本单元和原则。

1. Agent 与工作流 (Agent vs. Workflow)

Anthropic 将所有包含 LLM 和工具的系统统称为 Agentic Systems，并在此之下做出了重要的架构区分：

• 工作流 (Workflows)：是指 LLM 和工具通过预定义的代码路径被编排的系统。执行路径是固定的，模型只在规定的节点上做出决策。
• Agent (Agents)：是指 LLM 动态地指导其自身流程和工具使用，并保持对如何完成任务的控制权的系统。执行路径由模型在运行时自主决定。

简而言之，工作流是确定性的，适用于定义明确的任务；而 Agent 则是非确定性的，适用于无法预先规划所有步骤的开放式问题。多 Agent 协作主要建立在“Agent”的定义之上，强调模型的自主性和动态决策能力。

2. 工具使用与 Agent-计算机接口 (ACI)

工具（Tools） 是 Agent 与外部世界交互的桥梁。它们可以是任何函数或 API，例如执行代码、读写文件、搜索网页、查询数据库等。

Agent-计算机接口（Agent-Computer Interface, ACI） 是一个至关重要的概念，它指的是您为 Agent 设计的工具集及其文档。一个设计良好的 ACI 应该像一个为初级开发者编写的优秀函数库文档一样清晰、易于理解。模型的工具使用能力在很大程度上取决于工具定义的质量。

ACI 设计原则	描述与示例
清晰的名称和描述	工具的 name 和 description 应准确反映其功能和适用场景，避免歧义。
明确的输入模式	input_schema 应详细定义每个参数的类型、格式和含义，并说明哪些是必填项。
提供使用示例	在描述中加入具体的使用示例，能极大地帮助模型理解如何正确调用。
Poka-yoke（防错设计）	设计工具参数时，应使其难以被误用。例如，强制使用绝对路径而非相对路径，以避免因工作目录改变而出错。
明确工具边界	当存在多个功能相近的工具时，需在描述中清晰说明每个工具的适用边界，避免模型混淆。

3. 子 Agent (Subagents)

子 Agent (Subagents) 是多 Agent 协作在 Claude Agent SDK 中的核心实现。一个主 Agent 可以生成一个或多个子 Agent 来处理特定的子任务。这种模式带来了显著的优势：

• 上下文隔离：每个子 Agent 拥有独立的上下文窗口，避免主 Agent 的上下文被中间过程污染。
• 并行处理：可以同时运行多个子 Agent，并行执行不同的任务，极大地提升了效率。
• 专业化：可以为每个子 Agent 定义专门的系统提示（System Prompt）和工具集，使其成为特定领域的专家。
• 安全性：可以限制子 Agent 的工具权限，例如创建一个只有“只读”权限的分析器。

主 Agent 通过调用一个名为 Task 的内置工具来生成和管理子 Agent。这是实现“Orchestrator-Worker”架构模式的关键机制。

架构模式：设计多 Agent 协作系统

构建一个高效的多 Agent 系统，关键在于选择合适的架构模式。Anthropic 在其研究中总结了几种简单且可组合的模式。

1. Orchestrator-Worker 模式（核心模式）

这是最重要也是最灵活的多 Agent 架构模式。在该模式中，一个中心的 协调器 Agent (Orchestrator) 负责接收高级任务，将其动态分解为更小的子任务，然后将这些子任务委派给一个或多个 工作器 Agent (Workers) 并行处理，最后综合工作器的产出形成最终结果。

适用场景：无法预先确定所有步骤的复杂、开放式任务（如代码编写、市场研究）；需要从多个独立来源收集和分析信息的任务；任务可以被清晰地分解为多个并行子任务的场景。

2. 其他协作模式

架构模式	描述	典型适用场景
并行化 (Parallelization)	将任务分解为多个独立子任务并行运行（分片），或让多个 Agent 对同一任务执行多次以获得多样化输出（投票）。	内容生成与内容审查同步进行；多角度代码安全审查。
评估器-优化器 (Evaluator-Optimizer)	一个 Agent 生成内容，另一个 Agent 根据明确标准进行评估并提供反馈，形成迭代优化循环。	高质量文学翻译；需要多轮迭代的复杂代码生成。
路由 (Routing)	一个分类 Agent 将输入引导至最合适的专用 Agent 或工作流。	客服系统按意图分流（售前/技术支持/退款）；按难度路由至不同规格的模型。
提示链 (Prompt Chaining)	将任务分解为顺序步骤，每个 LLM 调用处理上一步的输出，可在中间步骤加入验证门控。	先生成营销文案，再翻译成多语言；先生成文档大纲，验证后再撰写正文。

这些模式可以相互组合，构建出适应不同需求的复杂 Agent 系统。

实战应用：构建与监控多 Agent 系统

理论知识需要通过实践来巩固。本章节将通过一个具体的例子，从零开始构建一个基于 Orchestrator-Worker 模式的多 Agent 研究系统，并最终实现在终端中对所有 Agent 进行可视化监控。

1. 环境准备

在开始之前，请确保您的开发环境已安装以下工具：

Python 3.9+ 和 pip 是基础要求。随后，通过以下命令安装所需的 Python 库：

# 安装 Anthropic 客户端（包含 Claude Agent SDK 的依赖）
pip install anthropic

# 安装用于终端可视化的 Rich 库
pip install rich

Tmux 是一个强大的终端复用工具，用于创建和管理多个终端会话和窗格，是实现可视化监控的基础：

# on macOS
brew install tmux

# on Debian/Ubuntu
sudo apt-get install tmux

最后，从您的 Anthropic Console 获取 API 密钥，并设置为环境变量：

export ANTHROPIC_API_KEY="sk-ant-api03-..."

2. 方案：使用 Anthropic 原生 API 实现

如果您不想使用 Agent SDK，也可以直接使用 anthropic 库通过手动实现工具调用循环来构建多 Agent 系统。这种方式更底层，但提供了更大的控制灵活性，也更有助于理解多 Agent 系统的工作原理。

创建文件 multi_agent_raw.py：

# multi_agent_raw.py
# 使用 Anthropic 原生 API 手动实现多 Agent 协作
# 依赖: pip install anthropic rich

import anthropic
import json
import time
from rich.console import Console
from rich.panel import Panel

client = anthropic.Anthropic()
console = Console()
LOG_FILE = "agent_logs.jsonl"

# ============================================================
# 工具定义：这些工具将被 Agent 调用
# ============================================================
TOOLS = [
    {
        "name": "web_search",
        "description": "搜索互联网获取关于某个主题的最新信息。返回相关的搜索结果摘要。",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "搜索查询词，应简洁明确。例如：'2025年AI Agent市场规模'"
                }
            },
            "required": ["query"]
        }
    },
    {
        "name": "spawn_subagent",
        "description": "生成一个专门的子 Agent 来处理特定的子任务。子 Agent 会独立完成任务并返回结果。",
        "input_schema": {
            "type": "object",
            "properties": {
                "agent_role": {
                    "type": "string",
                    "enum": ["researcher", "analyst", "writer"],
                    "description": "子 Agent 的角色：researcher（研究员）、analyst（分析师）、writer（写作员）"
                },
                "task": {
                    "type": "string",
                    "description": "子 Agent 需要完成的具体任务描述，应详细清晰。"
                },
                "context": {
                    "type": "string",
                    "description": "传递给子 Agent 的背景信息或已有数据。"
                }
            },
            "required": ["agent_role", "task"]
        }
    }
]

def log_action(agent_name: str, action_type: str, content: str):
    """记录 Agent 动作到日志文件。"""
    with open(LOG_FILE, "a", encoding="utf-8") as f:
        entry = {
            "timestamp": time.time(),
            "agent_name": agent_name,
            "action_type": action_type,
            "content": str(content)[:500]
        }
        f.write(json.dumps(entry, ensure_ascii=False) + "\n")

def execute_web_search(query: str) -> str:
    """模拟网络搜索（实际应用中可接入真实搜索 API）。"""
    mock_results = {
        "AI Agent": "2025年AI Agent市场规模预计达到280亿美元，主要玩家包括 Anthropic、OpenAI、Google DeepMind。",
        "半导体": "2025年全球半导体市场预计达到6000亿美元，台积电、三星、英特尔是主要厂商。",
    }
    for keyword, result in mock_results.items():
        if keyword.lower() in query.lower():
            return result
    return f"关于'{query}'的搜索结果：[模拟数据] 相关市场正在快速增长，多家科技公司正在积极布局。"

def run_subagent(agent_role: str, task: str, context: str = "") -> str:
    """运行一个子 Agent 来处理特定任务。"""
    role_prompts = {
        "researcher": "你是一个专业的研究员，擅长收集和整理信息。请根据任务要求，提供详细的研究结果。",
        "analyst": "你是一个数据分析师，擅长从数据中发现规律和洞见。请对提供的信息进行深度分析。",
        "writer": "你是一个专业的技术写作员，擅长将复杂信息转化为清晰易读的报告。请撰写一份结构清晰的报告。"
    }

    system_prompt = role_prompts.get(agent_role, "你是一个通用助手。")
    user_message = f"任务：{task}"
    if context:
        user_message += f"\n\n背景信息：{context}"

    log_action(agent_role, "start", f"开始执行任务: {task[:100]}")
    console.print(f"[cyan]  → 子 Agent [{agent_role}] 启动[/cyan]")

    sub_tools = [TOOLS[0]]  # 只给子 Agent 搜索工具
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.messages.create(
            model="claude-3-haiku-20240307",
            max_tokens=2048,
            system=system_prompt,
            tools=sub_tools,
            messages=messages
        )

        if response.stop_reason == "end_turn":
            result = ""
            for block in response.content:
                if hasattr(block, "text"):
                    result = block.text
            log_action(agent_role, "result", result[:200])
            console.print(f"[green]  ✓ 子 Agent [{agent_role}] 完成[/green]")
            return result

        elif response.stop_reason == "tool_use":
            messages.append({"role": "assistant", "content": response.content})
            tool_results = []

            for block in response.content:
                if block.type == "tool_use":
                    log_action(agent_role, "tool_use", f"{block.name}: {str(block.input)[:100]}")
                    if block.name == "web_search":
                        result = execute_web_search(block.input["query"])
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": result
                        })

            messages.append({"role": "user", "content": tool_results})
        else:
            break

    return "子 Agent 未能完成任务。"

def run_orchestrator(user_prompt: str):
    """运行协调器 Agent，它会动态生成和管理子 Agent。"""
    open(LOG_FILE, 'w').close()
    log_action("Orchestrator", "start", f"目标: {user_prompt}")

    console.print(Panel(f"[bold]用户请求：[/bold]{user_prompt}", title="[bold blue]协调器 Agent 启动[/bold blue]"))

    system_prompt = """你是一个研究团队的协调员（Orchestrator）。
你的职责是分析用户需求，并通过调用工具来协调多个专业子 Agent 完成任务。

工作流程：
1. 分析用户需求，制定研究计划
2. 使用 spawn_subagent 生成 researcher 子 Agent 收集信息
3. 使用 spawn_subagent 生成 analyst 子 Agent 分析数据
4. 使用 spawn_subagent 生成 writer 子 Agent 撰写最终报告
5. 将最终报告呈现给用户

重要：你必须通过 spawn_subagent 工具来委派任务，不要自己直接完成所有工作。"""

    messages = [{"role": "user", "content": user_prompt}]

    iteration = 0
    while iteration < 10:
        iteration += 1

        response = client.messages.create(
            model="claude-3-opus-20240229",
            max_tokens=4096,
            system=system_prompt,
            tools=TOOLS,
            messages=messages
        )

        if response.stop_reason == "end_turn":
            final_output = ""
            for block in response.content:
                if hasattr(block, "text"):
                    final_output = block.text
            log_action("Orchestrator", "result", final_output[:500])
            console.print(Panel(final_output, title="[bold magenta]最终报告[/bold magenta]"))
            break

        elif response.stop_reason == "tool_use":
            messages.append({"role": "assistant", "content": response.content})
            tool_results = []

            for block in response.content:
                if hasattr(block, "text"):
                    console.print(f"[yellow]💭 协调器思考：[/yellow]{block.text[:200]}")
                    log_action("Orchestrator", "thinking", block.text[:200])

                elif block.type == "tool_use":
                    console.print(f"[green]🛠️ 协调器调用工具：[/green]{block.name}")
                    log_action("Orchestrator", "tool_use", f"{block.name}: {str(block.input)[:100]}")

                    if block.name == "spawn_subagent":
                        agent_role = block.input["agent_role"]
                        task = block.input["task"]
                        context = block.input.get("context", "")
                        log_action("Orchestrator", "delegate", f"委派给 {agent_role}: {task[:100]}")

                        subagent_result = run_subagent(agent_role, task, context)
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": f"子 Agent [{agent_role}] 完成任务，结果：\n{subagent_result}"
                        })

                    elif block.name == "web_search":
                        search_result = execute_web_search(block.input["query"])
                        log_action("Orchestrator", "tool_use", f"搜索: {block.input['query']}")
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": search_result
                        })

            messages.append({"role": "user", "content": tool_results})

if __name__ == "__main__":
    run_orchestrator("请帮我研究2025年全球AI Agent市场的主要趋势、头部公司和投资机会。")

3. 步骤三：在终端中实现可视化监控

当多个 Agent 并行工作时，交错的日志输出会变得混乱不堪。我们需要一个能够将不同 Agent 的活动清晰地展示在独立空间中的仪表盘。我们将使用 tmux 和 rich 来构建这样一个 TUI（文本用户界面）监控系统。

创建监控仪表盘脚本 monitor.py：

# monitor.py
# 实时监控多 Agent 系统的运行状态
# 依赖: pip install rich

import json
import time
import os
from collections import defaultdict
from rich.console import Console
from rich.layout import Layout
from rich.live import Live
from rich.panel import Panel
from rich.text import Text
from rich.table import Table
from rich import box

LOG_FILE = "agent_logs.jsonl"
REFRESH_RATE = 0.3

ACTION_COLORS = {
    "start":    ("🚀", "bold blue"),
    "thinking": ("🤔", "yellow"),
    "tool_use": ("🛠️", "green"),
    "delegate": ("🤝", "cyan"),
    "result":   ("✅", "bold magenta"),
    "error":    ("❌", "bold red"),
}

agent_logs = defaultdict(list)
last_file_size = 0

def read_new_logs():
    global last_file_size
    if not os.path.exists(LOG_FILE):
        return

    current_size = os.path.getsize(LOG_FILE)
    if current_size <= last_file_size:
        return

    with open(LOG_FILE, 'r', encoding='utf-8') as f:
        f.seek(last_file_size)
        new_content = f.read()
        last_file_size = f.tell()

    for line in new_content.strip().split('\n'):
        if not line:
            continue
        try:
            entry = json.loads(line)
            agent_logs[entry["agent_name"]].append(entry)
        except json.JSONDecodeError:
            continue

def format_time(timestamp: float) -> str:
    return time.strftime("%H:%M:%S", time.localtime(timestamp))

def generate_agent_panel(agent_name: str, max_lines: int = 15) -> Panel:
    logs = agent_logs.get(agent_name, [])
    text = Text()

    if not logs:
        text.append("等待中...\n", style="dim")
    else:
        for log in logs[-max_lines:]:
            action_type = log.get("action_type", "unknown")
            content = log.get("content", "")
            timestamp = log.get("timestamp", 0)
            time_str = format_time(timestamp)

            icon, color = ACTION_COLORS.get(action_type, ("•", "white"))
            display_content = content[:80] + "..." if len(content) > 80 else content
            text.append(f"[{time_str}] {icon} ", style="dim")
            text.append(f"{display_content}\n", style=color)

    status = "运行中..." if logs and logs[-1].get("action_type") != "result" else "已完成 ✓"
    status_color = "yellow" if "运行中" in status else "green"

    return Panel(
        text,
        title=f"[bold]{agent_name}[/bold]",
        subtitle=f"[{status_color}]{status}[/{status_color}] | 共 {len(logs)} 条日志",
        border_style="blue" if "Orchestrator" in agent_name else "green"
    )

def generate_summary_table() -> Table:
    table = Table(box=box.SIMPLE, show_header=True, header_style="bold")
    table.add_column("Agent", style="bold")
    table.add_column("状态", justify="center")
    table.add_column("日志数", justify="right")
    table.add_column("最后动作", style="dim")

    for agent_name, logs in agent_logs.items():
        if not logs:
            continue
        last_log = logs[-1]
        action_type = last_log.get("action_type", "")
        icon, color = ACTION_COLORS.get(action_type, ("•", "white"))
        status = f"[{color}]{icon} {action_type}[/{color}]"
        last_content = last_log.get("content", "")[:40]
        table.add_row(agent_name, status, str(len(logs)), last_content)

    return table

def main():
    layout = Layout()
    layout.split(
        Layout(name="header", size=3),
        Layout(name="orchestrator", ratio=2),
        Layout(name="workers", ratio=3),
        Layout(name="footer", size=5),
    )
    layout["workers"].split_row(Layout(name="worker_left"), Layout(name="worker_right"))

    with Live(layout, screen=True, refresh_per_second=4) as live:
        while True:
            read_new_logs()
            layout["header"].update(Panel(f"[bold blue]Claude 多 Agent 监控仪表盘[/bold blue] | 按 Ctrl+C 退出", style="on dark_blue"))
            layout["orchestrator"].update(generate_agent_panel("Orchestrator", max_lines=10))
            worker_agents = [name for name in agent_logs.keys() if name != "Orchestrator"]
            if len(worker_agents) >= 1:
                layout["worker_left"].update(generate_agent_panel(worker_agents[0]))
            else:
                layout["worker_left"].update(Panel("等待工作器启动...", title="Worker 1"))
            if len(worker_agents) >= 2:
                layout["worker_right"].update(generate_agent_panel(worker_agents[1]))
            else:
                layout["worker_right"].update(Panel("等待工作器启动...", title="Worker 2"))
            layout["footer"].update(Panel(generate_summary_table(), title="Agent 状态汇总"))
            time.sleep(REFRESH_RATE)

if __name__ == "__main__":
    main()

创建 tmux 启动脚本 start_monitor.sh：

#!/bin/bash
# start_monitor.sh
# 一键启动多 Agent 系统和监控仪表盘

SESSION="multi_agent"
tmux kill-session -t $SESSION 2>/dev/null
tmux new-session -d -s $SESSION -x 220 -y 50

tmux split-window -v -p 40 -t $SESSION:0
tmux split-window -h -p 50 -t $SESSION:0.1

tmux send-keys -t $SESSION:0.0 "python3 monitor.py" C-m
sleep 1
tmux send-keys -t $SESSION:0.1 "python3 multi_agent_raw.py 2>&1 | tee agent_run.log" C-m
tmux send-keys -t $SESSION:0.2 "touch agent_logs.jsonl && tail -f agent_logs.jsonl | python3 -c \"import sys, json; [print(f\"[{d['agent_name']}] {d['action_type']}: {d['content'][:60]}\”) for l in sys.stdin for d in [json.loads(l)] if d]\"" C-m

tmux attach-session -t $SESSION

4. 运行与观察

完成所有文件的创建后，按以下步骤启动系统：

# 第一步：确保所有文件在同一目录下
ls -la multi_agent_raw.py monitor.py start_monitor.sh

# 第二步：赋予启动脚本执行权限
chmod +x start_monitor.sh

# 第三步：一键启动
./start_monitor.sh

您的终端将被 tmux 接管，呈现一个三窗格的布局，您可以清晰地看到协调器如何“思考”和“委派”任务，以及每个工作器如何“启动”并执行自己的工具，整个多 Agent 协作的过程一目了然。

进阶技巧与最佳实践

• 提示词工程：教会协调器如何委派任务是多 Agent 系统成功的关键。协调器在给子 Agent 分配任务时，必须提供足够清晰的指令，包括：明确的目标、期望的输出格式、可用的工具和数据源，以及与其他子 Agent 任务的边界划分。
• 工具设计：工具的设计质量直接决定了 Agent 系统的可靠性。当系统中存在多个功能相近的工具时，必须在描述中清晰说明每个工具的适用场景和限制。
• 评估与调试：在 Agent 开发的早期阶段，20 个测试用例通常就足以发现主要问题。对于自由文本输出，可以使用另一个 LLM 来评估输出质量。
• 生产环境的可靠性：Agent 系统是有状态的，错误会在多个步骤中累积。应当实现检查点机制（Checkpointing），使系统能够从错误发生处恢复，而不是从头开始。
• 成本控制：多 Agent 系统消耗的 Token 数量通常是单次对话的 15 倍以上。应当为每个任务设置合理的 Token 预算，并在协调器的提示词中嵌入资源分配规则，避免系统在简单任务上过度消耗。

总结

多 Agent 协作是释放大型语言模型全部潜力的关键。通过将复杂的任务分解，并利用专门的 Agent 团队进行并行处理和迭代优化，我们可以构建出远超单个 Agent 能力的强大系统。然而，随着系统复杂度的增加，理解和调试多个并行 Agent 的行为成为了一个新的挑战。本手册通过引入 tmux 和 rich 库，提供了一套完整的终端可视化监控解决方案。该方案通过结构化日志和动态 TUI 仪表盘，将原本无形的 Agent 协作过程变得具体、可追溯，极大地提升了开发和调试效率。

从核心概念的理解，到架构模式的选择，再到编码实现和可视化监控，本手册为您铺设了一条完整的学习路径。我们希望通过这份指南，您不仅能掌握构建 Claude 多 Agent 系统的技术，更能获得一种观察和理解复杂 AI 系统行为的全新视角。

---

参考文献

[1] Anthropic. (2025, June 13). How we built our multi-agent research system. https://www.anthropic.com/engineering/multi-agent-research-system