7.4 工具选择与编排

在上一节中，我们学习了 MCP（Model Context Protocol）——模型上下文协议。MCP 的核心价值在于标准化：它定义了一套统一的协议，让 Agent 能够以一致的方式发现工具、描述工具参数、并调用工具。换言之，7.3 节解决的是"工具怎么接"的问题。那么接下来自然要问：当工具接进来之后，Agent 面对几十甚至上百个可用工具时，该挑哪个工具？按什么顺序调用？哪些可以同时跑、哪些必须排队？ 这些问题的答案，正是本节的主题。

7.4.1 从单工具到多工具：调度员面对的难题

让我们用一个类比来理解 Agent 的角色。想象一个繁忙的物流调度中心，调度员面前坐着几十位专长不同的工人——有会查天气的、有会搜网页的、有会写 SQL 的、有会操作文件的。每当一个任务单递进来，调度员必须做三件事：

1. 选人——从几十位工人里挑出最合适的那位（或几位）来干这件活；
2. 排活——如果任务有多步，安排好先后顺序：先查天气、再搜新闻、最后写报告；
3. 盯进度——哪些活可以同时派下去（比如同时查北京和上海的天气），哪些必须等前一步干完才能开始。

Agent 的工具选择与编排，本质上就是这位调度员在做的事情。它面对的不是工人，而是一个个注册好的工具函数；它依据的不是直觉，而是用户的自然语言请求、工具的描述信息、以及模型的推理能力。

这个类比揭示了一个核心挑战：工具越多，选择越难。 当只有 3 个工具时，调度员一眼就能看明白该用哪个；但当工具列表膨胀到 50 个、100 个时，把它们全部塞进 LLM 的上下文窗口既费 Token 又容易让模型"看花眼"。因此，工具选择和编排不是可选项，而是 Agent 系统从原型走向生产时必须认真设计的架构能力。

7.4.2 单工具与并行调用的基本模式

在实际 Agent 系统中，工具调用有两种基本执行模式：串行调用和并行调用。

┌─────────────────────────────────────────────────────────┐
│                    串行调用 (Sequential)                  │
│                                                         │
│  User ──▶ LLM ──▶ Tool A ──▶ LLM ──▶ Tool B ──▶ 最终输出 │
│                (基于A结果决定B)                           │
│  耗时: T_A + T_LLM + T_B + T_LLM                        │
│                                                         │
├─────────────────────────────────────────────────────────┤
│                    并行调用 (Parallel)                    │
│                                                         │
│                     ┌──▶ Tool A ──┐                     │
│  User ──▶ LLM ──▶ ──┼──▶ Tool B ──┼──▶ LLM ──▶ 最终输出 │
│                     └──▶ Tool C ──┘                     │
│  耗时: max(T_A, T_B, T_C) + T_LLM                       │
│                                                         │
└─────────────────────────────────────────────────────────┘

图 7.4-1：串行调用与并行调用的执行流程对比

串行调用的特点是"走一步看一步"：模型先调用工具 A，拿到结果后再决定是否需要调用工具 B，每一步都基于上一步的输出。这种模式的延迟是各步耗时之和。

并行调用则是"一起派出去"：模型在同一轮推理中同时发出多个工具调用请求，这些调用并发执行，最后统一收集结果。这种模式的延迟取决于最慢的那个工具——就像调度员同时派出三位工人，只需要等最慢的那位回来就行。

从生产环境的实际数据来看，并行调用的性能收益非常显著：

指标	串行调用	并行调用	提升幅度
延迟（3 个工具调用）	30+ 秒	6 秒	约 5 倍
Token 消耗	基准	减少 ~50%	2 倍节省
API 成本	基准	降低 ~40%	1.7 倍节省
对话轮次	基准	减少 ~60%	2.5 倍减少

表 7.4-1：串行与并行调用的性能对比（生产环境数据）

然而，并行调用并非万能药。当工具之间存在依赖关系时——比如工具 B 需要工具 A 的输出作为输入——盲目并行化会导致严重问题。生产数据显示，在存在依赖的场景中错误地开启并行调用，可能带来约 7.1% 的浪费调用率（即调用因为缺少前置结果而失败或返回无意义数据），延迟改善也可能从预期的 60% 缩水到仅 18%。

下面这段代码展示了如何判断一组工具调用是否适合并行执行：

def should_parallelize(tool_calls: list) -> bool:
    """判断一组工具调用是否应该并行执行。"""
    # 遍历每一对工具调用，检查是否存在依赖关系
    for i, call_a in enumerate(tool_calls):          # 外层循环取第 i 个调用
        for call_b in tool_calls[i + 1:]:            # 内层循环取第 i 个之后的调用
            if has_dependency(call_a, call_b):        # 如果两者存在依赖关系
                return False                          # 有依赖就不能并行，立即返回 False
    return True                                       # 所有调用两两独立，可以并行

has_dependency 函数需要根据实际业务逻辑实现——它可能检查某个调用的输出参数是否是另一个调用的输入参数，也可能依赖工具元数据中声明的依赖关系。

适合并行的典型场景包括：多个独立的 API 查询（如同时查北京和上海的天气）、Fan-out/Fan-in 模式（将一个大查询分解为多个子查询并行执行，最后聚合结果）、以及投机执行（同时尝试多个可能的方向，取最先成功的结果）。

不适合并行的典型场景包括：依赖链（每个调用需要前一个的输出）、有副作用的操作（如写数据库、发邮件，顺序敏感）、以及需要条件判断的场景（后续步骤取决于前置结果）。

7.4.3 工具选择策略：调度员的三种选人方法

回到我们的调度员类比。当调度员面前有几十位工人时，他怎么知道该挑谁？有三种策略：

• 方法一：查档案——把每位工人的技能描述写成简历，调度员把任务要求跟简历做对比，找最匹配的。这就是语义匹配策略。
• 方法二：分类标签——每位工人身上贴了类别标签（"天气组""搜索组""数据库组"），调度员先根据任务判断属于哪组，再从组里选人。这就是意图路由策略。
• 方法三：直接问老员工——调度员自己拿不定主意，就请一位经验丰富的老员工（LLM）看任务描述，让他自己决定该叫谁。这就是 LLM 推理策略。

┌──────────────────────────────────────────────────────────┐
│                    工具选择策略                            │
├──────────────────┬──────────────────┬────────────────────┤
│  基于语义匹配     │   基于规则        │   基于 LLM 推理     │
├──────────────────┼──────────────────┼────────────────────┤
│ • 向量相似度搜索 │ • 预定义规则引擎  │ • 模型自主决策      │
│ • 适合大量工具   │ • 精确控制       │ • 最灵活           │
│ • 需要嵌入模型   │ • 维护成本高     │ • 受上下文窗口限制  │
│ • 示例: RAG 召回 │ • 示例: 意图路由 │ • 示例: ReAct Agent │
└──────────────────┴──────────────────┴────────────────────┘

图 7.4-2：三种工具选择策略对比

下面分别用代码展示这三种策略的实现。

策略一：基于语义匹配

语义匹配的思路是：给每个工具的描述生成一个嵌入向量（embedding），用户查询也生成嵌入向量，然后计算两者之间的余弦相似度，取相似度最高的若干个工具。这种方法的优点是无需人工维护规则，工具列表可以动态增减；缺点是依赖嵌入模型的质量，且对于"名字相近但功能不同"的工具可能误选。

import numpy as np
from typing import List

class SemanticToolSelector:
    """基于语义相似度的工具选择器。"""

    def __init__(self, embedding_model):
        # 保存嵌入模型实例（用于将文本转为向量）
        self.embedding_model = embedding_model
        # 工具嵌入向量缓存：{工具名: 向量}
        self.tool_embeddings = {}
        # 工具对象缓存：{工具名: 工具对象}
        self.tools = {}

    def register_tool(self, tool, embedding=None):
        """注册工具及其嵌入向量。"""
        self.tools[tool.name] = tool                      # 将工具存入字典
        if embedding is None:
            # 如果未提供预计算向量，则用描述文本生成嵌入
            embedding = self.embedding_model.embed(tool.description)
        self.tool_embeddings[tool.name] = embedding       # 缓存嵌入向量

    def select(self, query: str, top_k: int = 5) -> List:
        """根据查询语义选择最相关的 top_k 个工具。"""
        # 将用户查询转为嵌入向量
        query_embedding = self.embedding_model.embed(query)

        scores = {}                                       # {工具名: 相似度}
        for name, emb in self.tool_embeddings.items():
            # 计算余弦相似度：点积 / (模长乘积)
            similarity = np.dot(query_embedding, emb) / (
                np.linalg.norm(query_embedding) * np.linalg.norm(emb)
            )
            scores[name] = similarity

        # 按相似度从高到低排序，取前 top_k 个
        sorted_tools = sorted(scores.items(), key=lambda x: x[1], reverse=True)
        return [self.tools[name] for name, _ in sorted_tools[:top_k]]

这段代码的核心逻辑是：embed 将文本转为向量，np.dot 计算两个向量的点积，再除以各自的模长乘积得到余弦相似度（范围 -1 到 1，越接近 1 表示越相似）。最后排序取前 top_k 个。在生产环境中，这一步通常会用向量数据库（如 FAISS、Milvus）来加速，而非逐个计算。

策略二：基于意图路由

意图路由的核心思想是：先用一套规则将用户请求分类到某个意图类别，再根据意图类别取出对应的一组工具。这就像调度员看一眼任务单就知道该去哪个组找人，而不需要逐一对比每个人的简历。

from enum import Enum

class Intent(Enum):
    """意图分类枚举：每个值代表一种任务类型。"""
    WEATHER = "weather"
    SEARCH = "search"
    CALCULATE = "calculate"
    DATABASE = "database"
    FILE_OPERATION = "file_operation"

class IntentRouter:
    """基于意图的工具路由器。"""

    def __init__(self):
        # 每种意图映射到一组工具名
        self.routes = {
            Intent.WEATHER: ["get_weather", "get_forecast"],
            Intent.SEARCH: ["web_search", "document_search"],
            Intent.CALCULATE: ["calculator", "unit_converter"],
            Intent.DATABASE: ["db_query", "db_insert"],
            Intent.FILE_OPERATION: ["read_file", "write_file", "list_directory"],
        }

    def classify_intent(self, user_query: str) -> Intent:
        """基于关键词的简单意图分类。"""
        query_lower = user_query.lower()                  # 转小写便于匹配

        # 逐个意图检查是否命中关键词
        if any(w in query_lower for w in ["天气", "温度", "下雨", "weather"]):
            return Intent.WEATHER
        elif any(w in query_lower for w in ["搜索", "查找", "search", "find"]):
            return Intent.SEARCH
        elif any(w in query_lower for w in ["计算", "算", "等于", "+", "-", "*", "/"]):
            return Intent.CALCULATE
        elif any(w in query_lower for w in ["数据库", "查询", "sql", "database"]):
            return Intent.DATABASE
        elif any(w in query_lower for w in ["文件", "读取", "写入", "file"]):
            return Intent.FILE_OPERATION

        return None                                        # 无法匹配时返回 None

    def get_tools_for_intent(self, intent: Intent) -> list:
        """获取意图对应的工具列表。"""
        return self.routes.get(intent, [])

这个实现用了最简单的关键词匹配。在生产环境中，意图分类可以用小模型微调、训练专门的分类器、或者直接让 LLM 来分类。意图路由的优势在于精确可控——你能确切知道每个意图走哪条路；劣势在于维护成本——每加一类工具就得改代码。

策略三：基于 LLM 推理（ReAct 模式）

ReAct 是 Reasoning（推理）与 Acting（行动）的缩写，让 LLM 在"思考"与"行动"之间交替循环：先思考下一步该做什么，再选择工具并执行，然后观察结果，再思考下一步。这是当前最主流的 Agent 推理模式，也是最灵活的工具选择方式——不需要预先指定任何规则或分类，模型自己根据上下文判断该用哪个工具。

from langchain_openai import ChatOpenAI
from langchain.agents import create_react_agent, AgentExecutor
from langchain_core.prompts import PromptTemplate

# ReAct 提示模板：定义模型在每一步的输出格式
react_prompt = PromptTemplate.from_template("""
你是一个智能助手，可以使用以下工具来完成任务：

{tools}

使用以下格式回答：

Question: 用户的问题
Thought: 思考接下来应该做什么
Action: 要使用的工具名称（必须是 [{tool_names}] 之一）
Action Input: 工具的输入参数
Observation: 工具返回的结果
... (Thought/Action/Action Input/Observation 可以重复多次)
Thought: 我现在知道最终答案了
Final Answer: 对用户问题的最终回答

开始！

Question: {input}
Thought: {agent_scratchpad}
""")

# 创建 LLM 实例：temperature=0 保证输出稳定可复现
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 用 ReAct 模板创建 Agent（内部会解析模板中的占位符）
agent = create_react_agent(llm, tools, react_prompt)

# 包装为 AgentExecutor：管理多轮迭代和错误处理
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,                     # 打印每一步推理过程，便于调试
    max_iterations=5,                 # 最多迭代 5 轮，防止死循环
    handle_parsing_errors=True        # 解析失败时自动重试
)

# 执行：模型会自主决定先查天气、再搜历史
result = agent_executor.invoke({
    "input": "帮我查询北京天气，然后搜索一下长城的历史"
})

这段代码的关键在于提示模板中的 Thought → Action → Observation 循环。模型每轮输出一个 Thought（推理过程）和一个 Action（工具名+参数），系统执行工具后将结果作为 Observation 喂回模型，模型再进入下一轮思考。max_iterations=5 是安全阀——防止模型陷入"想-做-想"的死循环。在实际生产中，ReAct 模式的弱点是 Token 消耗大（每轮都要把历史思考过程塞进上下文），因此在工具数量非常多时通常结合语义匹配做预筛选。

7.4.4 工具链编排：调度员的排班表

选好了工具，接下来的问题是：工具之间怎么组织？还是回到调度员的类比——如果他需要安排一系列工序，他会画一张排班表：哪道工序先做、哪道后做、哪些可以同时做、哪些要根据前一步的结果来决定。

工具链编排就是给 Agent 画这张排班表。主要有四种模式：

┌─────────────────────────────────────────────────────────┐
│                    工具链编排模式                          │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  1. 串行流水线 (Sequential Pipeline)                     │
│     Tool A -> Tool B -> Tool C                            │
│     适用: 有依赖关系的步骤                                │
│                                                         │
│  2. 并行扇出 (Parallel Fan-out)                          │
│              ┌-> Tool A                                   │
│     Trigger ─┼-> Tool B ─-> Aggregate                      │
│              └-> Tool C                                   │
│     适用: 独立子任务                                      │
│                                                         │
│  3. 条件分支 (Conditional Branch)                        │
│                   ┌-> Tool A (条件1)                       │
│     Trigger ──判断─┼-> Tool B (条件2)                       │
│                   └-> Tool C (条件3)                       │
│     适用: 根据结果动态选择路径                              │
│                                                         │
│  4. 循环迭代 (Loop)                                      │
│     Trigger -> Tool A -> 判断 ──(继续)──-> Tool A            │
│                      └──(完成)──-> 输出                    │
│     适用: 需要多次尝试直到满足条件                          │
│                                                         │
└─────────────────────────────────────────────────────────┘

图 7.4-3：四种工具链编排模式

串行流水线是最基础的模式：A 的输出是 B 的输入，B 的输出是 C 的输入，像工厂流水线一样一环扣一环。适用场景是步骤之间有严格的依赖关系，比如"先查天气→再根据天气决定是否带伞→最后生成出行建议"。

并行扇出模式将一个触发器分发给多个独立工具并行执行，最后汇总结果。比如用户问"帮我查北京、上海、广州三地的天气"，三个城市的天气查询互不依赖，可以同时发出。关键在于最后要有一个聚合步骤把结果合并。

条件分支模式根据中间结果的值决定走哪条路径。比如查完天气后，如果是晴天就走"推荐户外活动"的工具，如果是雨天就走"推荐室内活动"的工具。

循环迭代模式让某个工具反复执行，直到满足退出条件。比如搜索引擎每次返回一页结果，如果不够再翻页继续搜。这种模式要特别注意设置最大循环次数，避免无限循环。

使用 LangGraph 实现工具编排

LangGraph 是 LangChain 生态中的图编排框架，它把工具编排问题转化为有向图问题：每个工具是一个节点，工具之间的执行顺序是边，条件判断是条件边。下面用 LangGraph 实现一个"查天气→条件判断→搜索新闻→生成报告"的工作流：

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator
import json

# ========== 定义状态 ==========
class AgentState(TypedDict):
    """工作流的全局状态，所有节点共享。"""
    messages: Annotated[List[dict], operator.add]  # 消息列表，用 operator.add 累加
    weather_data: dict                              # 天气查询结果
    search_results: list                            # 新闻搜索结果
    final_report: str                               # 最终报告文本

# ========== 定义节点（工具执行函数） ==========
def query_weather(state: AgentState) -> AgentState:
    """节点1：查询天气。"""
    # 从状态中取出最后一条用户消息
    last_msg = state["messages"][-1]["content"]
    # 简单规则提取城市名（实际可用 NLU 模型）
    city = "北京" if "北京" in last_msg else "上海"
    # 模拟调用天气 API 返回的数据
    weather = {"city": city, "temp": 28, "condition": "晴"}
    # 只返回要更新的字段（LangGraph 会自动合并到状态）
    return {"weather_data": weather}

def search_news(state: AgentState) -> AgentState:
    """节点2：搜索相关新闻。"""
    # 从状态中读取上一步存的天气数据
    city = state.get("weather_data", {}).get("city", "北京")
    # 模拟新闻搜索结果
    news = [
        f"{city}今日头条: AI技术突破",
        f"{city}天气预报: 明日有雨",
        f"{city}交通: 地铁新线路开通"
    ]
    return {"search_results": news}

def generate_report(state: AgentState) -> AgentState:
    """节点3：综合天气和新闻生成报告。"""
    # 读取天气数据
    weather = state.get("weather_data", {})
    # 读取新闻列表
    news = state.get("search_results", [])
    # 拼装报告文本
    report = f"""
📊 综合日报

🌤️ 天气: {weather.get('city', '未知')} {weather.get('temp', 'N/A')}°C {weather.get('condition', '')}

📰 相关新闻:
""" + "\n".join(f"  • {n}" for n in news)
    return {"final_report": report}

def should_continue(state: AgentState) -> str:
    """条件路由函数：判断是否需要搜索新闻。"""
    # 读取天气状况
    weather = state.get("weather_data", {})
    # 恶劣天气时搜索更多信息
    if weather.get("condition") in ["雨", "暴雨", "雪"]:
        return "search_news"
    # 晴天直接生成报告
    return "generate_report"

# ========== 构建图 ==========
workflow = StateGraph(AgentState)

# 添加三个节点
workflow.add_node("query_weather", query_weather)     # 注册天气查询节点
workflow.add_node("search_news", search_news)         # 注册新闻搜索节点
workflow.add_node("generate_report", generate_report)  # 注册报告生成节点

# 设置入口节点（图的起点）
workflow.set_entry_point("query_weather")

# 添加条件边：从天气节点出发，根据 should_continue 的返回值决定下一步
workflow.add_conditional_edges(
    "query_weather",                                  # 条件边的起点
    should_continue,                                  # 路由判断函数
    {                                                 # 返回值 -> 目标节点的映射
        "search_news": "search_news",
        "generate_report": "generate_report"
    }
)

# 添加普通边：搜索完新闻后去生成报告
workflow.add_edge("search_news", "generate_report")
# 报告生成后结束
workflow.add_edge("generate_report", END)

# 编译图（LangGraph 会做拓扑排序和验证）
app = workflow.compile()

# ========== 执行 ==========
result = app.invoke({
    "messages": [{"role": "user", "content": "查询北京天气并生成日报"}],
    "weather_data": {},
    "search_results": [],
    "final_report": ""
})

print(result["final_report"])

逐行解读关键部分：

• AgentState(TypedDict) 定义了整个工作流共享的状态结构。Annotated[List[dict], operator.add] 表示 messages 字段在多个节点返回时会自动拼接而非覆盖——这是 LangGraph 的"状态归约"机制。
• 每个 add_node(name, func) 调用注册一个节点，func 接收当前状态、返回状态增量（只包含要更新的字段）。
• set_entry_point 指定图的起点。执行时 LangGraph 从这个节点开始遍历。
• add_conditional_edges 是条件分支的核心：第二个参数是路由函数，返回一个字符串键；第三个参数是 {返回值: 目标节点} 的映射字典。路由函数返回 "search_news" 就走搜索分支，返回 "generate_report" 就直接出报告。
• add_edge("search_news", "generate_report") 是普通边——搜索完无条件进入报告生成。
• workflow.compile() 做的是编译：验证图的连通性、检查有没有死节点、生成可执行的计算图。
• app.invoke(initial_state) 从入口节点开始执行，直到到达 END 终点，返回最终状态。

7.4.5 并行工具调用实战

前面的 LangGraph 例子展示了串行和条件分支，但如何真正实现并行扇出？在 Python 中，最自然的方式是使用 asyncio 异步并发。下面是一个完整的并行工具执行器实现：

import asyncio
from concurrent.futures import ThreadPoolExecutor
from typing import Dict, Any, List
import json

class ParallelToolExecutor:
    """并行工具执行器：同时发出多个工具调用并收集结果。"""

    def __init__(self, max_workers: int = 10, timeout: float = 30.0):
        self.max_workers = max_workers               # 最大并发线程数
        self.timeout = timeout                       # 单个工具超时时间（秒）
        self.executor = ThreadPoolExecutor(max_workers=max_workers)  # 线程池

    async def execute_parallel(self, tool_calls: List[Dict]) -> Dict[str, Any]:
        """并行执行多个工具调用，返回 {call_id: result} 字典。"""

        async def execute_single(tool_call: Dict) -> tuple:
            """执行单个工具调用（内部协程）。"""
            tool_name = tool_call["name"]             # 取出工具名
            tool_args = tool_call.get("arguments", {}) # 取出参数字典
            call_id = tool_call.get("id", tool_name)   # 取出调用标识

            try:
                # 带超时地执行工具，防止某个工具卡死拖慢整体
                result = await asyncio.wait_for(
                    self._run_tool(tool_name, tool_args),
                    timeout=self.timeout
                )
                return call_id, {"success": True, "result": result}
            except asyncio.TimeoutError:
                return call_id, {"success": False, "error": "超时"}
            except Exception as e:
                return call_id, {"success": False, "error": str(e)}

        # 为每个工具调用创建协程任务
        tasks = [execute_single(call) for call in tool_calls]
        results = {}
        # asyncio.as_completed 按完成顺序（而非提交顺序）返回协程
        for completed in asyncio.as_completed(tasks):
            call_id, result = await completed          # 等待某个工具完成
            results[call_id] = result                  # 按调用 ID 存储结果

        return results

    async def _run_tool(self, tool_name: str, args: Dict) -> Any:
        """在独立线程中执行工具（模拟异步效果）。"""
        loop = asyncio.get_event_loop()
        # run_in_executor 把同步函数丢到线程池中执行，避免阻塞事件循环
        return await loop.run_in_executor(
            self.executor,
            self._tool_executor, tool_name, args
        )

    def _tool_executor(self, tool_name: str, args: Dict) -> Any:
        """实际工具执行逻辑（模拟不同工具的响应时间）。"""
        if tool_name == "get_weather":
            import time
            time.sleep(0.5)                            # 模拟 0.5 秒网络延迟
            return {"city": args.get("city"), "temp": 28}
        elif tool_name == "search_web":
            time.sleep(0.8)                            # 搜索更慢，0.8 秒
            return {"query": args.get("query"), "results": 5}
        elif tool_name == "calculate":
            return {"result": eval(args.get("expression", "0"))}
        else:
            return {"error": f"未知工具: {tool_name}"}


# ========== 使用示例 ==========
async def demo_parallel():
    """演示并行执行 4 个工具调用。"""
    executor = ParallelToolExecutor()

    # 4 个独立的工具调用
    tool_calls = [
        {"id": "call_1", "name": "get_weather", "arguments": {"city": "Beijing"}},
        {"id": "call_2", "name": "get_weather", "arguments": {"city": "Shanghai"}},
        {"id": "call_3", "name": "search_web", "arguments": {"query": "AI news"}},
        {"id": "call_4", "name": "calculate", "arguments": {"expression": "15 * 23"}},
    ]

    print("开始并行执行 4 个工具调用...")
    import time
    start = time.time()

    results = await executor.execute_parallel(tool_calls)

    elapsed = time.time() - start
    # 如果串行执行：0.5 + 0.5 + 0.8 + 0 = 1.8 秒
    # 并行执行：max(0.5, 0.5, 0.8, 0) = 0.8 秒
    print(f"并行执行完成，耗时: {elapsed:.2f}s")
    print(f"结果: {json.dumps(results, ensure_ascii=False, indent=2)}")

asyncio.run(demo_parallel())

代码的关键设计点逐一说明：

• ThreadPoolExecutor 创建了一个线程池。工具函数本身是同步的（比如调 requests.get），通过 run_in_executor 包装成异步，避免阻塞事件循环。
• asyncio.wait_for(..., timeout=self.timeout) 给每个工具调用加了超时保护。如果某个工具卡住，不会拖垮整个并行批次。
• asyncio.as_completed 是核心——它返回一个迭代器，按完成顺序而非提交顺序产出结果。这意味着哪个工具先回来就先处理哪个，不用等最慢的。
• 每个工具调用的结果用 call_id 做键存储，这样上层可以按调用 ID 对号入座，不依赖返回顺序。
• 从示例注释可以看到性能对比：4 个工具串行需要约 1.8 秒，并行只需约 0.8 秒（取最慢的那个）。

7.4.6 综合实战：智能工具选择器

实际生产中，单一策略往往不够用。最有效的做法是混合策略：先用 LLM 做意图分析缩小候选范围，再从候选工具中精确匹配。下面是一个结合 LLM 推理和工具筛选的智能选择器：

from typing import List, Dict, Any
import json

class SmartToolSelector:
    """智能工具选择器：结合 LLM 意图分析和工具筛选。"""

    def __init__(self, llm, embedding_model=None):
        self.llm = llm                                  # LLM 实例用于意图分析
        self.embedding_model = embedding_model          # 嵌入模型（可选）
        self.tools = {}                                # {工具名: 工具定义}

    def register(self, name: str, description: str, schema: Dict):
        """注册工具：name 为工具名，description 为描述，schema 为参数定义。"""
        self.tools[name] = {
            "name": name,
            "description": description,
            "schema": schema
        }

    def select_tools(self, query: str, max_tools: int = 5) -> List[Dict]:
        """
        两步选择策略：
        Step 1: LLM 分析意图，输出需要的工具类型
        Step 2: 根据意图类型筛选候选工具
        """
        # ---- Step 1: LLM 分析意图 ----
        intent_prompt = f"""
        分析用户查询的意图，返回需要的工具类型（可多选）：

        用户查询: {query}

        可选工具类型: weather, search, calculation, database, file, api

        只返回 JSON 数组，如 ["weather", "search"]
        """

        try:
            response = self.llm.invoke(intent_prompt)   # 调用 LLM 分析意图
            intent_categories = json.loads(response.content)  # 解析返回的 JSON 数组
        except:
            intent_categories = []                      # 解析失败时降级为空列表

        # ---- Step 2: 根据意图筛选候选工具 ----
        candidates = []
        for name, tool in self.tools.items():
            for intent in intent_categories:
                # 检查工具描述中是否包含意图关键词
                if intent.lower() in tool["description"].lower():
                    candidates.append(tool)
                    break                               # 匹配到一个意图即可，避免重复

        # 如果没有匹配到任何候选，降级为返回全部工具
        if not candidates:
            candidates = list(self.tools.values())

        # 截断到 max_tools 个（防止上下文过长）
        return candidates[:max_tools]

    def format_for_llm(self, selected_tools: List[Dict]) -> str:
        """将选中的工具格式化为文本，供 LLM 上下文使用。"""
        lines = []
        for i, tool in enumerate(selected_tools, 1):
            lines.append(f"{i}. {tool['name']}: {tool['description']}")
            lines.append(f"   参数: {json.dumps(tool['schema'], ensure_ascii=False)}")
        return "\n".join(lines)


# ========== 测试 ==========
selector = SmartToolSelector(llm=None)  # 实际使用时传入 LLM 实例

# 注册 4 个工具
selector.register("get_weather", "获取城市天气信息",
    {"city": {"type": "string", "description": "城市名"}})
selector.register("web_search", "搜索互联网内容",
    {"query": {"type": "string", "description": "搜索关键词"}})
selector.register("calculator", "执行数学计算",
    {"expression": {"type": "string", "description": "算式"}})
selector.register("db_query", "查询数据库",
    {"sql": {"type": "string", "description": "SQL语句"}})

print("已注册工具:", list(selector.tools.keys()))

这个选择器的设计思路是"先粗后细"：第一步用 LLM 做粗粒度分类（只返回类型标签，不消耗太多 Token），第二步在候选工具里做细粒度筛选。这种两阶段架构在生产环境中能有效控制 Token 成本，同时保持选择准确率。

7.4.7 综合实战：旅游规划工作流

将前面的概念综合起来，用 LangGraph 构建一个旅游规划 Agent。这个工作流先查天气，再并行获取机票、酒店、景点信息，最后生成旅行方案：

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator

class TravelState(TypedDict):
    """旅游规划工作流的状态定义。"""
    destination: str          # 目的地
    weather: dict             # 天气信息
    flights: list             # 航班列表
    hotels: list              # 酒店列表
    attractions: list         # 景点列表
    itinerary: str            # 最终行程文本

# ---- 模拟工具函数 ----
def get_weather(destination: str) -> dict:
    """模拟天气 API 返回。"""
    return {"city": destination, "temp": 25, "condition": "晴"}

def search_flights(destination: str) -> list:
    """模拟航班搜索。"""
    return [{"flight": "CA123", "price": 1200}, {"flight": "MU456", "price": 980}]

def search_hotels(destination: str) -> list:
    """模拟酒店搜索。"""
    return [{"name": "酒店A", "rating": 4.5}, {"name": "酒店B", "rating": 4.2}]

def search_attractions(destination: str) -> list:
    """模拟景点搜索。"""
    return [{"name": "景点1", "type": "自然"}, {"name": "景点2", "type": "文化"}]

# ---- 图节点 ----
def fetch_weather(state: TravelState) -> TravelState:
    """节点1：查目的地天气。"""
    dest = state["destination"]                    # 取出目的地
    return {"weather": get_weather(dest)}          # 返回天气数据

def fetch_travel_options(state: TravelState) -> TravelState:
    """节点2：并行获取机票、酒店、景点（在图中表现为一个节点内同时调用三个函数）。"""
    dest = state["destination"]                    # 取出目的地
    return {
        "flights": search_flights(dest),           # 搜索航班
        "hotels": search_hotels(dest),             # 搜索酒店
        "attractions": search_attractions(dest)    # 搜索景点
    }

def generate_itinerary(state: TravelState) -> TravelState:
    """节点3：综合所有信息生成行程方案。"""
    weather = state.get("weather", {})             # 读取天气
    flights = state.get("flights", [])             # 读取航班
    hotels = state.get("hotels", [])               # 读取酒店
    attractions = state.get("attractions", [])     # 读取景点

    itinerary = f"""
🏖️ {state['destination']} 旅行规划

🌤️ 天气: {weather.get('temp', 'N/A')}°C {weather.get('condition', '')}

✈️ 推荐航班:
""" + "\n".join(f"  • {f['flight']} ¥{f['price']}" for f in flights) + f"""

🏨 推荐酒店:
""" + "\n".join(f"  • {h['name']} ⭐{h['rating']}" for h in hotels) + f"""

🎯 推荐景点:
""" + "\n".join(f"  • {a['name']} ({a['type']})" for a in attractions)

    return {"itinerary": itinerary}

# ---- 构建图 ----
workflow = StateGraph(TravelState)

workflow.add_node("fetch_weather", fetch_weather)            # 注册天气节点
workflow.add_node("fetch_travel", fetch_travel_options)      # 注册旅行选项节点
workflow.add_node("generate_itinerary", generate_itinerary)  # 注册行程生成节点

workflow.set_entry_point("fetch_weather")                   # 从天气开始
workflow.add_edge("fetch_weather", "fetch_travel")          # 天气→旅行选项
workflow.add_edge("fetch_travel", "generate_itinerary")      # 旅行选项→行程
workflow.add_edge("generate_itinerary", END)                # 行程→结束

app = workflow.compile()

# ---- 执行 ----
result = app.invoke({
    "destination": "东京",
    "weather": {},
    "flights": [],
    "hotels": [],
    "attractions": [],
    "itinerary": ""
})

print(result["itinerary"])

这个工作流的执行流程是线性的：fetch_weather → fetch_travel → generate_itinerary → END。在真实场景中，fetch_travel_options 节点内部会用 asyncio 并行调用三个搜索函数，而整个图的拓扑结构保持简洁。

7.4.8 常见误区

在实际工程实践中，工具选择与编排容易踩到以下几个坑：

误区一：盲目开启并行调用

很多人看到 parallel_tool_calls=true 就开启，以为并行一定比串行快。但如果两个工具存在依赖——比如"先查用户 ID，再用 ID 查订单"——并行化会导致第二个工具收到空参数，要么报错要么返回无意义数据。正确做法是：先分析工具间的数据依赖关系，只有确认无依赖时才并行。

误区二：把所有工具塞进上下文窗口

当工具列表有 100 个时，每个工具的描述加参数 schema 可能占用上万 Token，直接塞进 LLM 上下文窗口会导致：输入过长推理变慢、模型注意力分散导致选错工具、Token 成本飙升。正确做法是：先用语义匹配或意图路由做预筛选，只把最相关的 5~10 个工具送入 LLM。

误区三：工具描述写得过于简略或过于冗长

工具描述是 LLM 做选择的唯一依据。如果描述只有"计算器"三个字，模型无法判断它支持什么运算；如果描述写了 500 字的使用说明，又浪费 Token 且容易让模型困惑。正确做法是：描述控制在 1~3 句话，说明"做什么""什么时候用""输入输出是什么"。

误区四：循环编排不设上限

ReAct 模式或 LangGraph 的循环边如果没有设最大迭代次数，模型可能陷入"思考→调用→观察→再思考"的无限循环，持续消耗 Token 直到被外部超时机制杀掉。正确做法是：永远设置 max_iterations（通常 5~10 轮），并在每轮检查是否有进展（如果连续两轮 Action 相同则强制退出）。

误区五：忽略工具调用的幂等性

并行执行时，同一个工具可能被多次调用（比如查多个城市的天气）。如果工具内部有副作用（比如写日志到数据库），重复调用可能产生重复数据。正确做法是：读操作设计为幂等的（多次调用结果一致），写操作避免在并行批次中执行。

误区六：条件分支逻辑硬编码在工具函数里

有些开发者把"如果是晴天就走 A 路径"的逻辑写死在工具函数内部，导致更换条件时要改工具代码。正确做法是：条件判断逻辑放在路由函数（如 should_continue）中，与工具函数解耦，这样修改条件不用动工具代码。

7.4.9 本节小结

本节围绕"Agent 如何选择和编排工具"展开，核心知识点梳理如下：

序号	核心要点
1	工具调用有两种模式：串行调用（走一步看一步，延迟为各步之和）和并行调用（同时发出，延迟取最慢者）。并行调用可带来约 5 倍延迟改善和 50% Token 节省。
2	并行调用需谨慎：存在依赖关系的工具不能并行，盲目并行化可能导致约 7% 的浪费调用率。用 should_parallelize 函数检查依赖关系是必要的防线。
3	工具选择有三种策略：语义匹配（查档案，适合大量工具）、意图路由（看标签，精确可控但维护成本高）、LLM 推理（问老员工，最灵活但消耗 Token）。
4	混合策略最优：先用 LLM 做粗粒度意图分析缩小候选范围，再做细粒度筛选，兼顾准确率和 Token 成本。
5	工具链编排有四种模式：串行流水线（有依赖）、并行扇出（独立子任务）、条件分支（动态路径）、循环迭代（多次尝试）。
6	LangGraph 是编排利器：通过有向图定义工具执行流程，节点是工具、边是执行顺序、条件边是动态路由。StateGraph + add_conditional_edges 是核心 API。
7	ReAct 模式是当前最主流的 Agent 推理模式，在 Thought→Action→Observation 循环中自主决策，但需设 max_iterations 防止死循环。

承前启后：在 7.3 节中，我们学习了 MCP 协议如何标准化工具的接入方式；本节解决了"选哪些工具、怎么编排"的问题。但还有一层关键问题没有回答：Agent 调用工具时的权限边界在哪里？ 如果 Agent 调用了一个删文件的工具却误删了系统文件怎么办？如果工具涉及用户隐私数据，如何防止泄露？这些问题将在下一节 7.5 权限与安全 中深入讨论。