拆穿 tools=tools：从工具调用到 Deep Research 架构

Agent 的经典公式大家很熟悉：

Agent = LLM + Planning + Memory + Tools

但在实际写代码时，我们往往只是在 SDK 里机械地传入一个 tools 参数，却没真正理解两个关键问题：

• 工具定义是如何「进入模型脑子」的呢？
• 模型输出 tool_call 之后，又是如何拿到 Python 执行结果并完成最终回答？

这篇文章是我把工具调用的底层机制梳理清楚之后，并在这个基础上再进一步：如果把「工具函数」换成「子 Agent」，会发生什么呢？

读完之后就会发现，多 Agent 协作与工具调用在结构上高度同构：只是调用对象从「无状态函数」变成了「有状态策略单元」。

一、从 LLM 到 Agent：工具调用的真相

我们在使用 OpenAI 或 Qwen 的 SDK 时，代码通常是这样写的：

def get_response(messages):
    completion = client.chat.completions.create(
        model="qwen-plus",
        messages=messages,
        # 关键就是这行代码
        tools=tools,
    )
    return completion

看起来非常简单，只要把定义好的工具列表 JSON Schema 传进去，再加上一句 tools=tools，模型自然就知道该怎么调用工具了。可是这底层到底发生了什么？

真相可以分为以下两层。

第一层：Prompt Engineering

如果不传 tools=tools，能让模型用工具吗？完全可以。

当传入 tools 参数时，SDK 或大模型后端其实在偷偷做一件事：把这些工具的定义（JSON Schema），转换成了一段 System Prompt，塞进了对话的开头。

SDK 内部其实就是帮我们做了一次「文本格式化」。可以尝试手动实现这个过程。

第一步：欺骗模型（Prompt 注入）

直接在 System Prompt 里告诉模型：「你是一个有工具的助手。工具有这些：[这里塞入 JSON 定义]。如果要用，请按这个格式输出：<tool_call>...</tool_call>。」

# 1. 把工具定义转换成字符串，塞进 Prompt
system_prompt = f"""
你是一个助手。你可以使用以下工具：
<tools>
{tools_content}
</tools>
如果要调用工具，请务必输出 XML 格式：
<tool_call>{{"name": "函数名称", "args": {{"参数名": "参数值"}}}}</tool_call>
"""

# 2. 整理 messages 内容
messages = [
    {"role": "system", "content": system_prompt},
    {"role": "user", "content": "北京天气怎么样"}
]

# 3. 发送给 LLM（注意：这里不传 tools 参数，完全靠 Prompt 指引）
completion = client.chat.completions.create(
    model="qwen-plus",
    messages=messages,
)

第二步：手动解析（正则提取）

当模型返回文本时，它并不会真的运行代码，只是输出了上面 prompt 中要求的一段 XML 文本：

<tool_call>{"name": "get_weather", "args": {"loc": "Beijing"}}</tool_call>

这时候，我们需要用正则表达式把运行工具函数需要的信息抠出来：

if "<tool_call>" in response.content:
    match = re.search(r"<tool_call>(.*?)</tool_call>", response.content)
    tool_call_data = json.loads(match.group(1))
    # 拿到 name 和 args，手动执行 Python 函数
    execute_function(tool_call_data["name"], tool_call_data["args"])

所谓的工具调用，第一层本质就是 Prompt Engineering（提示词工程） + Structured Output（结构化输出提取）。

第二层：拼接与重传

光有 Prompt 还不够。最容易疑惑的地方在于：模型吐出 tool_call 之后，它是怎么拿到 Python 运行的结果并回答用户的呢？

（之前的文章深度解析 Memory 本质：Agent 和 Chatbot 的差异中有详细写过。这个流程的核心就是：暂停 → 执行 → 拼接 → 重传。）

这里再以「查询巴黎天气」为例简单解释一下。

第一回合：LLM 的开条子时刻

• 用户提问：「巴黎天气怎么样？」
• LLM 思考：它收到 System Prompt（你有工具）和 User Message。它发现自己不知道天气，但有个 get_weather 工具。
• LLM 暂停并输出：LLM 停止生成自然语言，而是吐出一个 JSON（或者上面定义的 XML）：

{"name": "get_weather", "args": {"city": "Paris"}}

到此为止，第一次请求结束了。模型处于挂机状态。

中场休息：Python 的跑腿时刻

这时候完全没有 LLM 的事了，全是 Python 代码在干活：代码捕获到检查单（tool_calls），去调用天气 API，拿到了 "14.2°C"。

第二回合：带着结果，重新排队

为了让 LLM 能回答用户，必须把刚才发生的一切，打包成新的历史记录，重新发给 LLM。这一步叫「拼接」，需要构造一个包含三条信息的新列表：

• User：「巴黎天气怎么样？」
• Assistant：（第一轮模型生成的 tool_calls）「我要调用 get_weather……」
• Tool：（Python 跑出来的结果）「14.2°C」

然后重传：再次调用 client.chat.completions.create。

LLM 看到这三句话，恍然大悟：「哦，我刚才要查巴黎天气，现在温度（14.2 度）来了。」于是它生成最终回复：「巴黎现在 14.2 度，比较凉爽。」

理解了这两层真相（Prompt 定义 + 拼接重传），就彻底搞懂了单 Agent 的工具调用。接下来就可以进阶了：如果把工具换成另一个 Agent，会发生什么呢？

二、从单 Agent 到多 Agent：架构的同构性

如果任务太复杂，一个 Agent 搞不定，这时就会想到用多 Agent 协作完成。

多 Agent 听起来很深奥，但其实，多 Agent 本质上可以看成把「工具调用」提升了一个抽象层级：

• 单 Agent：LLM 调用一个个具体的 Python 函数（Tools）
• 多 Agent：总控 Agent 调用一个个具备推理、记忆和工具能力的子系统（Sub-Agents）

从这两张图就会发现两者的同构性。并且单 Agent 的四种工具调用形态，与多 Agent 的架构模式可以完美对应。

1、单一工具调用 ≈ 单一子 Agent

这是最简单的形态。

• 工具层：用户问天气 → 模型生成 Tool Call → 执行 weather 函数
• Agent 层：简单的 SearchAgent。用户提问 → Agent 规划 → 搜索并回答

线性执行，一次往返。

2、多工具选择（Router）≈ 路由式多 Agent

• 工具层：给模型一堆工具（计算器、翻译、搜索），模型在单轮对话中决定用哪一个
• Agent 层：给总控 Agent 一堆子 Agent（数学专家、翻译专家、搜索专家），总控根据问题分发任务

这是分类问题。不需要复杂的编排，只需要准确的「意图识别」。

3、多轮链式工具（Chain）≈ 链式 / 接力 Agent

这是最常见的进阶形态。

工具层（While 循环）：LLM 先调 web_search → System 回填搜索结果 → LLM 读结果，发现全是英文，决定调 translate_text → System 回填翻译结果 → LLM 最终总结。

Agent 层：策划 Agent → 撰写 Agent → 审核 Agent。后一个 Agent 的输入严格依赖前一个 Agent 的输出。

这是状态机。核心在于 while True 循环，直到满足停止条件（如 tool_calls 为空）。

4、单轮并行工具 ≈ 并行多 Agent

工具层：LLM 一次性输出多个 Tool Call：[get_weather(Beijing), get_weather(Paris), get_stock(AAPL)]，代码层用 ThreadPoolExecutor 并发执行，最后汇总。

Agent 层：这也是 Deep Research 的核心雏形。面对一个大问题，拆解成 n 个子问题，由 n 个 Worker Agent 同时并行处理，最后由 Leader Agent 汇总。

目的是提升吞吐，利用 LLM 的并发规划能力。

总结一下：

• 工具是无状态的函数，输入即输出。
• 子 Agent 是有状态的「大工具」，它内部包含了「目标、上下文、反思循环」。

多 Agent 结构就像工具调用的升级版：调度器调用的不再是死函数，而是一个个能独立完成子任务的策略单元。

（无论外面的架构怎么变——路由、并行、串行——每一个干活的 Agent 内部，基本都遵循一个经典的闭环：Plan-Act-Reflect。之前的文章「规划-行动-反思」架构：Agent 是如何进行反思的？有写到相关细节。）

三、拆解 Deep Research 架构

现在，就可以用上面的知识，来拆解一个真实的多 Agent 项目了。

这个项目是标准的 Orchestrator-Workers 模式，能够做到：用户给一个相对模糊的课题，它能自动拆解子任务、并行检索、归纳分析、发现缺口并自动补全，最终生成一份结构化的研究报告。

它的核心代码结构如下：

• orchestrator.py：大脑。负责规划、派发、汇总、验收与迭代。
• workers.py：手脚。负责执行具体的检索、整理与单点研究（包含 OODA 循环）。
• tools.py：工具箱。封装搜索 API 与 LLM 调用，并提供 JSON 提取等基础能力。

下面跟踪一条指令：「分析 2026 年人工智能的发展趋势」，看看它是如何在代码中流转的。

1、大脑规划：从模糊指令到 JSON 规划

一切始于 orchestrator.py。主调度 Agent 拿到的第一个动作不是去搜，而是先想清楚怎么搜。

在代码中，这对应一个 decompose_task 函数流程：它把模糊任务交给 LLM，让模型输出一个可执行的 TaskPlan，其中最关键的是 subtasks 列表与 query_type / recommended_worker_count 等字段：

{
  "query_type": "breadth_first",
  "subtasks": [
    {"id": 1, "description": "2026 AI 技术架构趋势（端侧模型、推理加速）"},
    {"id": 2, "description": "2026 AI 产业落地（医疗、金融、制造）"},
    {"id": 3, "description": "全球 AI 监管与合规政策"}
  ],
  "recommended_worker_count": 3
}

2、手脚的并发：朴素的 ThreadPoolExecutor

拿到 subtasks 列表后，怎么让多个 Worker 同时干活呢？

代码中没有引入复杂的 Asyncio 或多进程，而是选择了对于网络检索型 IO 任务非常实用的 ThreadPoolExecutor。在 dispatch_workers 函数中，Orchestrator 会为每个 subtask 创建一个独立 Worker 实例，把它们扔进线程池并行跑：

def dispatch_workers(self, subtasks):
    actual_workers = min(len(subtasks), self.max_workers)
    with ThreadPoolExecutor(max_workers=actual_workers) as executor:
        futures = []
        for subtask in subtasks:
            worker = self.worker_factory(subtask)   # 每个 subtask 一个 worker 实例
            futures.append(executor.submit(worker.execute, subtask))
        results = [f.result() for f in as_completed(futures)]
    return results

这就是「单轮并行」的真相：把 n 个子任务扔进线程池，让它们同时去调用搜索 API 与 LLM。每个 Worker 都是独立实例，互不共享内部状态。

3、Worker 的核心：带预算的 OODA 循环

现在视角切换到 workers.py。每个被扔进线程池的 Worker 内部是在做什么呢？

代码中负责实现的 SearchWorker 采用了一个 OODA + 预算的研究回路：

• 预算限制（max_queries / max_cycles）由任务复杂度（simple / medium / complex）自动估算并配置
• OODA 每轮会「观察（搜索）→ 定向（粗评估结果）→ 决策（是否补搜、补什么）→ 行动（生成下一轮 query）」

def execute(self, subtask):
    budget = adjust_budget(assess_complexity(subtask))   # max_queries + max_cycles
    queries = subtask.search_queries
    if not queries:
        queries = expand_queries_with_llm(subtask)       # 可能会调用一次 LLM

    all_results = []
    for cycle in range(budget.max_cycles):
        if budget.remaining_queries <= 0:
            break

        # Observe：执行若干 query 的搜索
        new_results = []
        for q in pick_queries(queries, budget.remaining_queries):
            new_results += search_client.search(q)
            budget.consume_query()
        all_results += new_results

        # Orient / Decide：来源粗评估 + 信息缺口识别（偏启发式）
        gaps = identify_information_gaps(subtask, all_results)
        if not gaps:
            break

        # Act：基于 gaps 生成下一轮 queries
        queries = generate_followup_queries(gaps)

    # 最后：调用 LLM 把 all_results 格式化 + 归纳为结构化输出
    context = format_search_context(all_results)
    return llm_summarize_to_structured_result(subtask, context)

主程序只给目标，Worker 在预算约束下通过多轮搜索补齐信息，再把证据上下文整理成结构化结果交回 Orchestrator 进行汇总与撰写。

4、闭环的灵魂：迭代与查漏补缺

Deep Research 之所以是 Deep，在于它不只有拆解 → 并行 → 汇总，还引入了质量检查 + 差距分析 + 补充研究 + 修订的迭代闭环。

在 orchestrator.run() 中，主流程大致是：

def run(self, task):
    # 1) 初次研究
    task_plan = self.decompose_task(task)
    worker_results = self.dispatch_workers(task_plan.subtasks)
    report = self.synthesize_results(worker_results)

    # 2) 迭代优化（注意：有 min_iterations + quality_threshold 双重条件）
    for i in range(self.max_iterations):
        quality = self.check_quality(report)
        if i + 1 >= self.min_iterations and quality.score >= self.quality_threshold:
            break

        # （可选）收益递减检测：模型判断是否「继续已不划算」
        if self.enable_diminishing_returns_check:
            if self.check_diminishing_returns(...).should_stop:
                break

        # Gap Analysis：模型指出缺口 + 生成 supplementary_tasks
        gap_info = self.analyze_gaps(report, task, quality)
        supplementary_tasks = gap_info["supplementary_tasks"]

        # 再并行补漏
        new_results = self.dispatch_supplementary_workers(supplementary_tasks)

        # 修订报告
        report = self.refine_report(report, gap_info, new_results)

    return report

报告不是一次写成，而是写完初稿后自己评审，发现缺口（比如监管条款没说清、关键数据缺引用、对比维度不够），然后动态生成补充子任务，让 Worker 去补证据，补回来后再修订报告。

在拆解完整个 orchestrator.py 和 workers.py 后，是否发现了一个有趣的细节？

Agent 之间从不「聊天」，而是互传 JSON。

Orchestrator 下发的是结构化的 SubTask，Worker 回传的是结构化的 Dict。多 Agent 协作的本质，不是让模型互相聊天，而是定义一套清晰的 API 接口标准。

最后

从 tools=tools 到 Deep Research，本质是同一条链路的放大：Prompt 约束结构化输出 + 暂停 → 执行 → 拼接 → 重传。

单 Agent 在循环里按开关，多 Agent 在总控调度下并行按开关、再用评估与补漏迭代优化。

理解了这些底层机制与调度原理，就可以自己摸索着设计出真正解决复杂问题的系统了。