1. Agent 是什么，为什么要"多"

Agent（智能体） —— 能自主规划、调用工具、执行任务的 AI。跟普通的 LLM 调用不一样，Agent 不只是"你问我答"，它能自己决定下一步做什么。

打个比方：普通的 LLM 调用像客服热线，你问一句它答一句；Agent 更像你雇的助理——你说"帮我调研竞品"，它会自己拆解任务、上网搜索、整理结果。

两者的区别可以用一张表说清楚：

	普通 LLM 调用	Agent
交互方式	一问一答，每次独立	多轮自主决策
外部工具	需要人写代码串联	自己决定调用哪个工具
任务复杂度	单步任务	多步任务
决策权	人决定每一步	AI 自己决定下一步

那为什么要"多" Agent？

原因有两个。

一是专业化分工。 把所有工具塞给一个 Agent，就像让一个人同时当搜索工程师、数据分析师和报告撰写人。它哪个都能凑合，但哪个都不精。拆成多个专门的 Agent，每个只关注自己的领域，Prompt 更短、工具更少、输出更稳定。

二是并行效率。 多个 Agent 可以同时干活。比如竞品调研，搜索 5 个竞品的信息——一个 Agent 串行搜要五次，五个 Agent 并行搜只要一次的时间。

什么时候该用多 Agent？ 一个简单的判断标准：

• 任务能拆成 2 个以上独立的子任务 → 考虑多 Agent
• 子任务之间不需要频繁交换中间状态 → 适合多 Agent
• 任务简单、步骤少、一个 Prompt 就能搞定 → 单 Agent 足够
• 所有子任务必须共享同一份上下文 → 不适合多 Agent

还有一个概念提一下：MCP（Model Context Protocol） —— AI 与外部工具连接的开放标准。你可以理解为：REST 是 Web API 的通用语言，MCP 就是 AI 工具集成的通用语言。

你只需要知道：Agent = 能自己调工具的 AI。当任务需要多种专业能力协作，或者能拆成独立子任务并行处理时，多 Agent 就有价值。代价是 Token 消耗会大幅增加。

2. 从单 Agent 开始

先搞定单 Agent，理解它的运行机制。

ReAct：Agent 的工作循环

ReAct 模式（Reasoning + Acting） —— Agent 的核心工作方式，一个"推理→执行→观察"的循环。

用做饭来类比：看菜谱（推理）→ 切菜（执行）→ 尝味道（观察）→ 觉得淡了再加盐（推理）→ 加盐（执行）→ 再尝（观察）。Agent 的工作方式一模一样。

写你的第一个 Agent

完整代码：

import anthropic
import json

client = anthropic.Anthropic()

# 定义工具：让 Agent 能搜索网页
tools = [
    {
        "name": "web_search",
        "description": "搜索互联网获取实时信息。输入搜索关键词，返回搜索结果摘要。",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "搜索关键词"
                }
            },
            "required": ["query"]
        }
    }
]

def execute_tool(name: str, params: dict) -> str:
    """工具执行入口——目前是模拟数据，替换为 Tavily/SerpAPI 即可接入真实搜索"""
    if name == "web_search":
        return f"搜索'{params['query']}'的结果：这里是模拟的搜索结果..."
    return "未知工具"

def run_agent(task: str) -> str:
    """运行单 Agent 的 ReAct 循环"""
    messages = [{"role": "user", "content": task}]

    while True:
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            tools=tools,
            messages=messages,
        )

        # Agent 决定任务完成，返回结果
        if response.stop_reason == "end_turn":
            for block in response.content:
                if hasattr(block, "text"):
                    return block.text

        # 安全兜底：max_tokens 截止时也返回已有内容
        if response.stop_reason == "max_tokens":
            for block in response.content:
                if hasattr(block, "text"):
                    return block.text
            return "Agent 输出被截断，请增大 max_tokens"

        # Agent 要求调用工具
        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                result = execute_tool(block.name, block.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result,
                })

        # 把 Agent 的回复和工具结果都加到对话历史
        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": tool_results})

# 试一下
result = run_agent("搜索一下 2026 年最流行的 Python Web 框架有哪些")
print(result)

这段代码的核心是：定义工具——ReAct 循环——对话历史——Prompt

Agent 自动规划搜索顺序——你不需要告诉它"先搜 FastAPI 再搜 Django"。

3. 升级到多 Agent

搞定了单 Agent。但当任务涉及多个领域——比如既要搜索信息、又要分析数据、还要生成报告——一个 Agent 的 Prompt 就得又长又复杂。每加一个工具，它对每个工具的理解都会被"稀释"一点。这时候多 Agent 就有意义了。

编排器-子 Agent 架构

多 Agent 系统的核心架构叫 Orchestrator-Subagent（编排器-子 Agent）。

编排器（Orchestrator） —— 负责任务分配和协调的主 Agent。像项目经理，不亲自干活，负责拆解任务、分配给合适的人、汇总结果。

子 Agent（Subagent） —— 执行具体任务的专门 Agent。每个只管自己的领域。

上下文隔离：为什么要分开

这里有个关键设计：上下文隔离（Context Isolation） —— 每个 Agent 有独立的对话历史，互不干扰。

上下文隔离就像分会议室开会——每个人先在自己的小会议室研究自己的部分，最后只向项目经理汇报结论。不需要知道其他人搜了什么、看了哪些资料。

在技术层面，上下文隔离带来两个好处：

1. 每个 Agent 的上下文窗口只包含自己需要的信息，不会被其他任务的内容"稀释"
2. 子 Agent 只返回压缩后的结论给编排器，不是把搜索中的所有中间结果都传回去

代码实现：

import anthropic
import json
import re
from concurrent.futures import ThreadPoolExecutor

client = anthropic.Anthropic()

def create_subagent(system_prompt: str, task: str, tools: list = None) -> str:
    """创建并运行一个独立的子 Agent"""
    # 关键：每个子 Agent 都有全新的 messages，这就是上下文隔离
    messages = [{"role": "user", "content": task}]

    while True:
        kwargs = {
            "model": "claude-sonnet-4-20250514",
            "max_tokens": 4096,
            "system": system_prompt,
            "messages": messages,
        }
        if tools:
            kwargs["tools"] = tools

        response = client.messages.create(**kwargs)

        if response.stop_reason in ("end_turn", "max_tokens"):
            for block in response.content:
                if hasattr(block, "text"):
                    return block.text

        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                result = execute_tool(block.name, block.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result,
                })

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


def extract_json(text: str) -> dict:
    """从 LLM 输出中提取 JSON——处理 markdown 代码块包裹的情况"""
    # 先尝试直接解析
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        pass
    # 尝试提取 ```json ... ``` 中的内容
    match = re.search(r'```(?:json)?\s*([\s\S]*?)```', text)
    if match:
        try:
            return json.loads(match.group(1).strip())
        except json.JSONDecodeError:
            pass
    # 尝试提取第一个 { ... } 块
    match = re.search(r'\{[\s\S]*\}', text)
    if match:
        try:
            return json.loads(match.group(0))
        except json.JSONDecodeError:
            pass
    raise ValueError(f"无法从 LLM 输出中提取 JSON: {text[:200]}")


def run_orchestrator(task: str) -> str:
    """编排器：拆解任务 → 并行分配 → 汇总结果"""

    # 第一步：让编排器分析任务，拆成子任务
    plan_response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=2048,
        system="""你是一个任务编排器。分析用户任务，拆解成可并行的子任务。
用 JSON 格式输出，格式如下：
{"subtasks": [{"name": "任务名", "prompt": "子Agent的角色描述", "task": "具体任务"}]}""",
        messages=[{"role": "user", "content": task}],
    )

    plan = extract_json(plan_response.content[0].text)

    # 第二步：并行启动子 Agent
    results = {}
    with ThreadPoolExecutor(max_workers=5) as executor:
        futures = {}
        for subtask in plan["subtasks"]:
            future = executor.submit(
                create_subagent,
                system_prompt=subtask["prompt"],
                task=subtask["task"],
            )
            futures[subtask["name"]] = future

        for name, future in futures.items():
            results[name] = future.result()

    # 第三步：汇总
    summary = create_subagent(
        system_prompt="你是信息汇总专家。把多个研究结果整合成结构清晰的报告。",
        task=f"以下是各子任务的结果，请综合整理：\n{json.dumps(results, ensure_ascii=False, indent=2)}",
    )

    return summary

代码结构分两部分：

1. create_subagent —— 创建独立的子 Agent。每次调用都是全新的 messages，这就是上下文隔离。注意 stop_reason 同时处理了 end_turn 和 max_tokens。
2. run_orchestrator —— 编排器分三步走：分析任务 → 并行执行 → 汇总结果。

你只需要知道：多 Agent 系统的核心是"编排器 + 子 Agent"。编排器负责拆解和分配，子 Agent 各自在独立的上下文中执行。用 ThreadPoolExecutor 可以让子 Agent 并行运行。

4. 避坑指南

搭多 Agent 系统的框架不复杂，但从 demo 到稳定运行，有几个坑需要提前知道。

坑 1：Token 消耗失控

Token（令牌） —— AI 处理文本的基本单位，也是计费单位。一个中文字大约 1-2 个 Token。

具体消耗取决于任务复杂度、子 Agent 数量和工具调用轮次。上面的倍数是 Anthropic 在研究系统中的平均值。

怎么控制：

• 子 Agent 只返回精简摘要，不要把完整搜索结果传给编排器
• 用 max_tokens 限制每个 Agent 的输出长度
• 简单任务别用多 Agent——杀鸡别用牛刀

坑 2：上下文污染

如果你不小心让多个 Agent 共享了同一个 messages 列表，它们的对话历史会互相污染。A Agent 搜索的结果跑到 B Agent 的上下文里，B Agent 就可能产出莫名其妙的回答。

解法： 每个子 Agent 都用全新的 messages 列表。回头看第 3 章的 create_subagent 函数——每次调用都 messages = [{"role": "user", "content": task}]，这就是上下文隔离。

坑 3：工具描述太模糊

工具的描述决定了 Agent 能不能正确使用它。描述写得模糊，Agent 就不知道什么时候该用、怎么用。

工具设计三条规则：

• 描述写清楚：说明输入什么、返回什么、什么场景下用
• 功能要专一：一个工具做一件事，别搞"万能工具"
• 参数要约束：用 JSON Schema 的 required 和类型限制

坑 4：调试困难

多 Agent 系统最大的问题不是搭建，而是出了 bug 很难定位。多个 Agent 并行运行，某个给了奇怪的结果——你怎么知道是哪个 Agent 的问题？

我的经验是三个方法最管用：

1. 打日志 : 记录每个 Agent 的输入和输出，出问题时能回溯。
2. 开发时关掉并行 : 让 Agent 一个个跑，方便定位。上线后再开并行。
3. 设置 temperature=0 : 让结果尽量可复现。

坑 5：Prompt 越堆越长

随着系统变复杂，你可能会不断给 Agent 加指令。Prompt 越来越长，Agent 反而越来越笨——因为注意力被分散了。

解法： 把复杂指令拆到多个专门的 Agent 里，而不是全塞到一个 Prompt。这也是多 Agent 架构的核心价值之一：通过分工降低单个 Agent 的认知负担。

你只需要知道：多 Agent 的主要成本是 Token。保持上下文隔离、写好工具描述、打好日志，能避开大部分坑。

5. 想想可以做什么

到这里，你已经掌握了多 Agent 系统的核心：

• ReAct 循环让 Agent 能自主推理和行动
• 编排器-子 Agent 架构让多个 Agent 分工协作
• 上下文隔离保证了每个 Agent 的工作质量
• Token 消耗和调试是实际落地时要重点关注的

如果你想继续深入，可以看看这几个方向：

• 接入真实搜索 API：把模拟搜索换成 Tavily 或 SerpAPI，让系统能真正上网搜信息
• MCP 协议：标准化的 AI 工具集成方案。CNCF 在 KubeCon 2026 首次设立了 Agentics Day，云原生和 Agent 系统正在快速融合
• Anthropic Agent SDK：官方的 Agent 开发工具包，封装了本文手写的很多逻辑，适合生产环境
• 生产级部署：认证管理、速率限制、错误重试、成本监控——从 demo 到生产还有不少路要走