API 编排与工具选择

引言

当 Agent 拥有大量工具时，如何选择正确的工具、以正确的顺序调用、并处理调用中的错误，成为一个关键的工程问题。本节探讨工具选择算法、编排模式和错误处理策略。

工具选择问题

挑战

工具数量过多：上百个工具时，LLM 难以在上下文中处理所有工具定义
工具语义重叠：多个工具功能相似，需要精确区分
上下文消耗：每个工具定义消耗数百 token
选择准确性：选错工具导致任务失败

工具选择策略

1. 全量加载（适合工具少的场景）

# 工具 < 20 个时，直接全部传递给 LLM
response = llm.chat(
    messages=messages,
    tools=all_tools,  # 全部工具定义
)

2. 语义路由（Semantic Routing）

基于用户查询的语义，预先筛选相关工具：

class ToolRouter:
    def __init__(self, tools, embedding_model):
        self.tools = tools
        self.embed = embedding_model

        # 预计算所有工具描述的嵌入
        self.tool_embeddings = {
            tool["name"]: self.embed(tool["description"])
            for tool in tools
        }

    def select_tools(self, query, top_k=5):
        """基于语义相似度选择最相关的工具"""
        query_embedding = self.embed(query)

        scores = {}
        for name, emb in self.tool_embeddings.items():
            scores[name] = cosine_similarity(query_embedding, emb)

        # 返回最相关的 top_k 个工具
        sorted_tools = sorted(scores.items(), key=lambda x: x[1], reverse=True)
        selected_names = [name for name, score in sorted_tools[:top_k]]

        return [t for t in self.tools if t["name"] in selected_names]

# 使用
router = ToolRouter(all_tools, embedding_model)
relevant_tools = router.select_tools("帮我查一下北京的天气")
response = llm.chat(messages=messages, tools=relevant_tools)

3. 分类路由

使用 LLM 先分类，再加载对应类别的工具：

TOOL_CATEGORIES = {
    "information_retrieval": ["web_search", "knowledge_base_query", "database_query"],
    "data_analysis": ["execute_python", "create_chart", "statistical_test"],
    "communication": ["send_email", "send_slack", "create_document"],
    "file_operations": ["read_file", "write_file", "list_directory"],
}

def category_routing(query, llm):
    """先分类再选择工具"""
    category = llm.classify(
        f"将以下查询分类到工具类别中：{list(TOOL_CATEGORIES.keys())}\n"
        f"查询: {query}\n类别:"
    )
    return TOOL_CATEGORIES.get(category, [])

4. 两阶段选择

先用轻量模型筛选，再用强模型精确选择：

def two_stage_selection(query, all_tools):
    # Stage 1: 轻量模型快速筛选（成本低）
    candidates = light_model.select(
        query=query,
        tools=[{"name": t["name"], "description": t["description"][:100]} for t in all_tools],
        top_k=10,
    )

    # Stage 2: 强模型精确选择（带完整定义）
    selected = strong_model.select(
        query=query,
        tools=[t for t in all_tools if t["name"] in candidates],
    )
    return selected

工具链模式（Tool Chaining）

顺序链（Sequential Chain）

# 工具 A 的输出作为工具 B 的输入
async def sequential_chain(query):
    # Step 1: 搜索
    search_results = await search_tool(query)

    # Step 2: 用搜索结果查数据库
    db_results = await database_query(extract_entities(search_results))

    # Step 3: 分析
    analysis = await code_executor(generate_analysis_code(db_results))

    return analysis

并行扇出（Fan-out）

import asyncio

async def fan_out(query):
    """并行调用多个工具，汇总结果"""
    tasks = [
        web_search(query),
        knowledge_base_search(query),
        database_query(query),
    ]

    results = await asyncio.gather(*tasks, return_exceptions=True)

    # 过滤错误
    valid_results = [r for r in results if not isinstance(r, Exception)]

    return merge_results(valid_results)

条件分支

async def conditional_routing(query, context):
    """根据条件选择不同的工具链"""
    intent = classify_intent(query)

    if intent == "factual_question":
        # 知识检索链
        docs = await knowledge_base_search(query)
        return generate_answer(query, docs)

    elif intent == "data_analysis":
        # 数据分析链
        data = await database_query(extract_sql(query))
        code = generate_analysis_code(data)
        return await code_executor(code)

    elif intent == "action_request":
        # 执行操作链
        plan = plan_actions(query)
        results = []
        for step in plan:
            result = await execute_action(step)
            results.append(result)
            if result.get("error"):
                break
        return summarize_results(results)

递归工具使用

Agent 在执行中发现需要额外工具调用：

async def recursive_tool_use(query, tools, llm, depth=0, max_depth=5):
    """递归工具调用"""
    if depth >= max_depth:
        return "达到最大递归深度"

    response = await llm.chat(
        messages=[{"role": "user", "content": query}],
        tools=tools
    )

    if not response.tool_calls:
        return response.content

    # 执行工具调用
    results = []
    for tc in response.tool_calls:
        result = await execute_tool(tc.name, tc.arguments)
        results.append(result)

    # 将结果反馈给 LLM，可能触发更多工具调用
    follow_up = format_results(results)
    return await recursive_tool_use(follow_up, tools, llm, depth + 1, max_depth)

错误处理与重试

错误分类

错误类型	示例	处理策略
参数错误	缺少必填参数	让 LLM 修正参数
认证失败	API Key 过期	刷新认证后重试
速率限制	429 Too Many Requests	指数退避重试
服务不可用	500 Server Error	等待后重试或降级
逻辑错误	查询无结果	改写查询或换工具
超时	执行时间过长	超时重试或简化请求

重试策略

import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10),
)
async def resilient_tool_call(tool_name, arguments):
    """带重试的工具调用"""
    return await execute_tool(tool_name, arguments)

class SmartRetry:
    def __init__(self, llm):
        self.llm = llm

    async def execute_with_recovery(self, tool_name, arguments, error=None):
        """智能错误恢复"""
        if error:
            # 让 LLM 分析错误并修正参数
            fix = await self.llm.generate(
                f"工具 {tool_name} 调用失败。\n"
                f"参数: {arguments}\n"
                f"错误: {error}\n"
                f"请修正参数或建议替代方案。"
            )
            arguments = fix.get("corrected_arguments", arguments)

        try:
            return await execute_tool(tool_name, arguments)
        except Exception as e:
            if self.is_retryable(e):
                return await self.execute_with_recovery(tool_name, arguments, str(e))
            raise

    def is_retryable(self, error):
        retryable_codes = [429, 500, 502, 503, 504]
        return getattr(error, 'status_code', None) in retryable_codes

降级策略

class ToolWithFallback:
    def __init__(self, primary_tool, fallback_tools):
        self.primary = primary_tool
        self.fallbacks = fallback_tools

    async def execute(self, arguments):
        """主工具失败时尝试备用工具"""
        try:
            return await self.primary(arguments)
        except Exception as primary_error:
            for fallback in self.fallbacks:
                try:
                    return await fallback(arguments)
                except:
                    continue
            raise primary_error

# 示例：搜索工具的降级链
search = ToolWithFallback(
    primary_tool=google_search,
    fallback_tools=[bing_search, brave_search, duckduckgo_search]
)

工具使用决策框架

何时使用何种工具

TOOL_DECISION_TREE = """
用户查询 → 
├─ 需要最新信息？ → web_search
├─ 需要内部知识？ → knowledge_base_search  
├─ 需要精确计算？ → code_interpreter
├─ 需要数据分析？ → code_interpreter + data_tools
├─ 需要执行操作？ → 
│  ├─ 发送消息？ → email/slack_tool
│  ├─ 文件操作？ → file_tools
│  └─ 系统操作？ → bash_tool
├─ 需要多步推理？ → 组合多个工具
└─ 纯知识问答？ → 不使用工具，直接回答
"""

实践建议

编排清单

[ ] 实现工具路由减少无关工具的干扰
[ ] 为每个工具定义清晰的使用场景和排除条件
[ ] 实现错误处理和重试机制
[ ] 设置工具调用的超时和资源限制
[ ] 添加工具使用的日志和监控
[ ] 对高风险工具添加人工确认步骤

性能优化

并行调用独立的工具减少延迟
缓存常用工具的结果
使用轻量模型做预筛选
对工具结果进行截断避免上下文浪费

延伸阅读

框架选型综述 - Agent 框架如何处理工具编排
Qin, Y., et al. (2024). "Tool Learning with Large Language Models: A Survey"
Hao, S., et al. (2024). "ToolkenGPT: Augmenting Frozen Language Models with Massive Tools via Tool Embeddings"

API 编排与工具选择

引言

工具选择问题

挑战

工具选择策略

1. 全量加载（适合工具少的场景）

2. 语义路由（Semantic Routing）

3. 分类路由

4. 两阶段选择

工具链模式（Tool Chaining）

顺序链（Sequential Chain）

并行扇出（Fan-out）

条件分支

递归工具使用

错误处理与重试

错误分类

重试策略

降级策略

工具使用决策框架

何时使用何种工具

实践建议

编排清单

性能优化

延伸阅读

评论 #