Agent 工具系统：Function Calling 背后的真实世界

你有没有想过当ChatGPT帮你查天气、写代码、搜资料的时候它到底是怎么知道该调哪个接口的答案大家都知道——Function Calling。但说实话大部分人只看到了冰山一角。模型返回一个函数名和参数你执行一下把结果塞回去完事。真要做一个生产级的Agent工具系统你会发现坑多得离谱。从一个真实的问题说起去年我帮一个团队搭Agent系统需求很简单让Agent能查数据库、调API、读写文件。一开始我们的做法很naive——直接把所有工具的JSON Schema塞进system prompt让模型自己选。三个工具的时候还好等加到十五个工具模型开始犯迷糊了明明该调search的它非要调database参数格式也经常传错。后来工具数量涨到三十多个彻底崩了。模型选工具的准确率掉到了60%以下token消耗暴涨光工具描述就吃了大几千token响应速度也慢得让人抓狂。那一刻我才意识到Function Calling只是工具系统的冰山一角。真正的挑战在水面之下——工具注册、发现、描述、校验、编排、缓存……每一层都有学问。工具系统的三层架构经过反复踩坑我把Agent工具系统抽象成了三层┌─────────────────────────────────────┐│ 编排层 (Orchestration) ││ 工具选择 → 并行/串行 → 结果聚合 │├─────────────────────────────────────┤│ 描述层 (Description) ││ Schema定义 → 语义描述 → 检索匹配 │├─────────────────────────────────────┤│ 执行层 (Execution) ││ 参数校验 → 调用执行 → 结果格式化 │└─────────────────────────────────────┘别小看这个分层。很多团队的工具系统出问题都是因为这三层的职责混在一起了。描述层让模型看懂你的工具这是最容易被忽视但影响最大的一层。工具Schema不只是JSONOpenAI的Function Calling要求你用JSON Schema描述工具。但很多人不知道Schema的写法直接影响模型的选择准确率。看个反面教材{ name: query, description: Query the database, parameters: { type: object, properties: { input: { type: string } } }}这个Schema有三个致命问题名字太泛query什么query数据库query搜索引擎query API模型看到这个名字根本不知道你干嘛描述太短“Query the database”——什么数据库查什么表什么场景下用参数名没意义input是什么SQL语句自然语言表名改一下{ name: query_user_database, description: 在用户管理数据库中执行SQL查询。用于查找用户信息、订单记录、账户状态等。仅支持SELECT语句不支持写操作。当用户询问某某用户的信息、最近的订单等问题时使用此工具。, parameters: { type: object, properties: { sql: { type: string, description: 要执行的SQL SELECT语句例如SELECT * FROM users WHERE email xxx } }, required: [sql] }}差别在哪名字有语义、描述有场景、参数有示例。模型选工具的准确率能从60%直接拉到90%以上。工具数量爆炸怎么办当工具超过10个直接全塞prompt就不现实了。两个思路思路一语义检索先用embedding把所有工具描述向量化用户提问时先检索top-k相关工具只把这几个工具的Schema发给模型。import numpy as npfrom sentence_transformers import SentenceTransformerclass ToolRegistry: def __init__(self): self.tools [] self.embeddings None self.encoder SentenceTransformer(all-MiniLM-L6-v2) def register(self, tool_schema: dict): 注册工具 self.tools.append(tool_schema) # 用 name description 生成embedding text f{tool_schema[name]}: {tool_schema[description]} embedding self.encoder.encode(text) if self.embeddings is None: self.embeddings np.array([embedding]) else: self.embeddings np.vstack([self.embeddings, embedding]) def discover(self, query: str, top_k: int 5) - list: 根据用户查询发现最相关的工具 query_embedding self.encoder.encode(query) # 余弦相似度 scores np.dot(self.embeddings, query_embedding) / ( np.linalg.norm(self.embeddings, axis1) * np.linalg.norm(query_embedding) ) top_indices np.argsort(scores)[-top_k:][::-1] return [self.tools[i] for i in top_indices]# 使用registry ToolRegistry()registry.register({ name: search_web, description: 搜索互联网获取最新信息})registry.register({ name: query_database, description: 查询内部数据库获取业务数据})registry.register({ name: send_email, description: 发送邮件给指定收件人})# 用户问最近有什么新闻relevant_tools registry.discover(最近有什么新闻, top_k2)# → [search_web, ...] 不会返回send_email思路二工具分类把工具按领域分组文件操作、网络请求、数据库、系统命令等先让模型选类别再在类别内选具体工具。两层选择每层的候选集都小很多。MCP工具描述的标准化尝试2025年底Anthropic推出了MCPModel Context Protocol想解决的就是这个问题——让工具描述有一个标准格式任何Agent框架都能即插即用。MCP的核心思路是把工具提供者Tool Provider和工具消费者Agent解耦。工具提供者通过MCP Server暴露工具Agent通过MCP Client发现和调用。说白了就是给AI工具生态搞了个USB-C接口。不过说实话MCP目前还处于早期阶段。协议本身设计得不错但生态远没成熟。我现在的建议是关注MCP的发展但别急着all in。自己的工具系统先把描述层做好等MCP生态成熟了再对接也不迟。执行层别信任模型的参数模型选对了工具不代表参数就对了。这一层要做的事很朴素但很重要。参数校验永远不要直接把模型返回的参数丢给你的函数。模型会撒谎。它会传字符串给你期望的整数会漏掉必填字段会在枚举值里编一个不存在的选项。from pydantic import BaseModel, ValidationError, Fieldfrom typing import Optionalfrom enum import Enumclass QueryType(str, Enum): SELECT select INSERT insert UPDATE updateclass DatabaseQueryParams(BaseModel): sql: str Field(..., descriptionSQL查询语句) database: str Field(default_db, description目标数据库名) query_type: QueryType Field(QueryType.SELECT, description查询类型) limit: Optional[int] Field(None, ge1, le1000, description结果数量限制) timeout_ms: Optional[int] Field(5000, ge100, le30000, description超时时间)def execute_tool(tool_name: str, raw_params: dict): 安全的工具执行入口 # 1. 参数校验 try: params DatabaseQueryParams(**raw_params) except ValidationError as e: return { success: False, error: f参数校验失败: {e}, suggestion: 请检查SQL语句格式和参数类型 } # 2. 额外的业务规则校验 if params.query_type ! QueryType.SELECT: return { success: False, error: 只允许SELECT查询, suggestion: 如需修改数据请使用专门的数据修改工具 } # 3. 执行 try: result db.execute(params.sql, timeoutparams.timeout_ms) return {success: True, data: result} except Exception as e: return { success: False, error: f执行失败: {str(e)}, sql: params.sql # 方便调试 }注意几个细节用Pydantic做校验别手写if-else。Pydantic的类型转换和错误提示都比手写的好校验失败时返回suggestion字段告诉模型哪里错了、怎么改。模型拿到这个反馈后下一轮可以自动修正参数业务规则校验和类型校验分开职责清晰结果格式化工具返回的结果模型不一定能理解。比如数据库返回的datetime对象、API返回的嵌套JSON、文件系统返回的路径对象……这些都需要格式化成模型能理解的文本。def format_tool_result(tool_name: str, raw_result) - str: 把工具返回值格式化成模型友好的文本 if isinstance(raw_result, dict) and raw_result.get(success): data raw_result.get(data) # 数据库查询结果 if isinstance(data, list): if len(data) 0: return 查询结果为空没有匹配的记录。 # 只返回前10条避免token爆炸 preview data[:10] formatted f查询到 {len(data)} 条记录显示前10条\n for i, row in enumerate(preview, 1): formatted f{i}. {row}\n if len(data) 10: formatted f...还有 {len(data) - 10} 条记录未显示。 return formatted # 单条记录 if isinstance(data, dict): return \n.join(f{k}: {v} for k, v in data.items()) return str(data) elif isinstance(raw_result, dict) and not raw_result.get(success): error raw_result.get(error, 未知错误) suggestion raw_result.get(suggestion, ) return f工具执行失败: {error}\n{suggestion} return str(raw_result)这个格式化函数做了几件关键的事截断长结果数据库返回一万条记录你不能全塞给模型。取前10条告诉模型总数错误信息友好不只是返回错误码还告诉模型怎么修正统一格式不管底层工具返回什么类型最终都变成模型能理解的文本编排层工具调用的指挥家这是最有趣的一层。当Agent需要调用多个工具时怎么编排串行 vs 并行最简单的场景Agent一次只调一个工具拿到结果再决定下一步。这就是串行。但很多时候工具之间没有依赖关系。比如用户问帮我查一下北京明天的天气和上海的航班信息查天气和查航班完全可以并行。import asynciofrom typing import Anyclass ToolOrchestrator: def __init__(self, executor): self.executor executor # 工具执行器 async def execute_parallel(self, tool_calls: list[dict]) - list: 并行执行多个工具调用 tasks [ self.executor.execute( call[tool_name], call[parameters] ) for call in tool_calls ] results await asyncio.gather(*tasks, return_exceptionsTrue) # 处理异常不让一个失败拖垮全部 formatted [] for call, result in zip(tool_calls, results): if isinstance(result, Exception): formatted.append({ tool: call[tool_name], success: False, error: str(result) }) else: formatted.append(result) return formatted async def execute_sequential(self, tool_calls: list[dict]) - list: 串行执行前一个的结果可以传给后一个 context {} results [] for call in tool_calls: # 支持参数引用前一个工具的结果 params self._resolve_references( call[parameters], context ) result await self.executor.execute( call[tool_name], params ) context[call[tool_name]] result results.append(result) # 如果某个工具失败了看是否继续 if not result.get(success) and call.get(required, True): break return results def _resolve_references(self, params: dict, context: dict) - dict: 解析参数中的引用比如 {user_id: {{get_user.id}}} import re resolved {} for key, value in params.items(): if isinstance(value, str): matches re.findall(r\{\{(\w)\.(\w)\}\}, value) for tool_name, field in matches: if tool_name in context: replacement context[tool_name].get(data, {}) if isinstance(replacement, dict): value value.replace( f{{{{{tool_name}.{field}}}}}, str(replacement.get(field, )) ) resolved[key] value return resolved模型驱动的动态编排更高级的做法是让模型自己决定调用顺序。这就是ReAct模式的核心——模型在每一轮思考后决定下一步调什么工具。用户: 帮我分析一下竞品A最近一个月的用户增长情况模型思考: 我需要先获取竞品A的信息然后查它的用户数据 → 调用 search_company(竞品A) ← 返回: {name: A公司, id: 12345}模型思考: 拿到了公司ID现在查用户增长数据 → 调用 query_user_growth(company_id12345, period30d) ← 返回: {growth_rate: 15.2%, new_users: 12000}模型思考: 数据拿到了再查一下行业平均数据做对比 → 调用 query_industry_average(industrySaaS, metricuser_growth) ← 返回: {average_growth: 8.5%}模型: 竞品A最近一个月用户增长15.2%新增用户12000人 远超行业平均的8.5%……这个模式的好处是灵活——模型可以根据中间结果动态调整策略。坏处是慢——每一轮都要等模型思考而且token消耗大。一个实用的折中方案在实际项目中我发现一个效果不错的折中方案预定义编排模板 模型填充参数。# 预定义常见的工作流模板WORKFLOW_TEMPLATES { competitor_analysis: { description: 竞品分析工作流, steps: [ {tool: search_company, param_source: user_input}, {tool: query_user_metrics, depends_on: 0}, {tool: query_revenue_metrics, depends_on: 0}, {tool: compare_with_industry, depends_on: [1, 2]}, ] }, user_support: { description: 用户问题排查工作流, steps: [ {tool: query_user_info, param_source: user_input}, {tool: query_recent_logs, depends_on: 0}, {tool: query_system_status, parallel_with: 1}, {tool: generate_diagnosis, depends_on: [1, 2]}, ] }}模型只需要判断用户意图属于哪个模板然后填充参数。编排逻辑是确定性的速度快、可预测、好调试。几个血泪教训1. 工具描述要写什么时候不用大部分人写工具描述只说这个工具能干什么。但模型更需要知道这个工具什么时候不该用。{ name: search_knowledge_base, description: 搜索内部知识库。适用于查找公司政策、产品文档、FAQ等内部资料。不要用于搜索互联网信息请用search_web不要用于查询实时数据请用query_database。}加了不要用于之后模型选错工具的概率明显下降。2. 给工具加元数据除了Schema工具还应该携带一些元数据帮助编排层做决策tool( namedelete_user, description删除用户账户, danger_levelhigh, # 危险等级 requires_confirmationTrue, # 需要人工确认 rate_limit10/min, # 频率限制 estimated_latency_ms200, # 预估延迟 cost_per_call0.01, # 每次调用成本 categoryuser_management # 分类)async def delete_user(user_id: str, reason: str): ...这些元数据在编排时非常有用。比如危险等级高的工具可以自动触发人工审批频率限制可以防止模型陷入循环调用。3. 工具结果的缓存有些工具的结果是幂等的——同样的参数短时间内返回的结果不会变。这种工具的结果应该缓存。from functools import lru_cacheimport hashlibimport jsonclass CachedToolExecutor: def __init__(self, ttl_seconds: int 300): self.cache {} self.ttl ttl_seconds def _cache_key(self, tool_name: str, params: dict) - str: raw f{tool_name}:{json.dumps(params, sort_keysTrue)} return hashlib.md5(raw.encode()).hexdigest() async def execute(self, tool_name: str, params: dict, cacheable: bool False) - dict: if cacheable: key self._cache_key(tool_name, params) if key in self.cache: entry self.cache[key] if time.time() - entry[timestamp] self.ttl: entry[hits] 1 return {**entry[result], cached: True} result await self._actual_execute(tool_name, params) if cacheable: self.cache[key] { result: result, timestamp: time.time(), hits: 0 } return result模型在多轮对话中经常会重复问同样的问题。缓存能省掉大量重复的工具调用和token消耗。4. 工具调用的可观测性每个工具调用都应该被记录。不是简单的日志而是结构化的traceclass ToolCallTrace: def __init__(self): self.calls [] def record(self, tool_name, params, result, latency_ms, tokens_before, tokens_after): self.calls.append({ timestamp: time.time(), tool: tool_name, params_summary: self._summarize(params), result_summary: self._summarize(result), success: result.get(success, False), latency_ms: latency_ms, token_delta: tokens_after - tokens_before }) def get_stats(self): total len(self.calls) successes sum(1 for c in self.calls if c[success]) total_latency sum(c[latency_ms] for c in self.calls) total_tokens sum(c[token_delta] for c in self.calls) return { total_calls: total, success_rate: f{successes/total*100:.1f}%, avg_latency_ms: total_latency / total, total_tokens_used: total_tokens, most_called_tool: self._most_called(), slowest_tool: self._slowest() }这些数据不仅能帮你排查问题还能用来优化工具描述哪些工具经常被选错、调整编排策略哪些工具可以并行、控制成本哪些工具最耗token。工具系统的未来我觉得Agent工具系统正在经历类似Web API从SOAP到REST的演进。早期大家都是把所有东西塞给模型就像SOAP一样——完整但笨重。现在开始出现分层、检索、标准化协议MCP就像REST时代的到来。几个值得关注的趋势MCP生态成熟当越来越多的工具以MCP Server的形式提供Agent就能像手机装App一样即插即用工具市场类似App Store开发者发布工具Agent按需安装。Anthropic的MCP Server目录已经是这个方向了自适应工具选择模型不再只是被动选择工具而是能根据执行结果动态调整策略甚至自己发明新的工具组合工具安全标准随着Agent调用的工具越来越多工具本身的安全审计、权限控制、沙箱隔离会成为刚需说到底工具系统是Agent连接真实世界的桥梁。桥修得好不好直接决定了Agent是能聊天的玩具还是能干活的助手。Function Calling只是桥上的一块砖。要把桥修好你还需要描述、校验、编排、缓存、监控……每一块都不能少。学AI大模型的正确顺序千万不要搞错了2026年AI风口已来各行各业的AI渗透肉眼可见超多公司要么转型做AI相关产品要么高薪挖AI技术人才机遇直接摆在眼前有往AI方向发展或者本身有后端编程基础的朋友直接冲AI大模型应用开发转岗超合适就算暂时不打算转岗了解大模型、RAG、Prompt、Agent这些热门概念能上手做简单项目也绝对是求职加分王给大家整理了超全最新的AI大模型应用开发学习清单和资料手把手帮你快速入门学习路线:✅大模型基础认知—大模型核心原理、发展历程、主流模型GPT、文心一言等特点解析✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑✅开发基础能力—Python进阶、API接口调用、大模型开发框架LangChain等实操✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经以上6大模块看似清晰好上手实则每个部分都有扎实的核心内容需要吃透我把大模型的学习全流程已经整理好了抓住AI时代风口轻松解锁职业新可能希望大家都能把握机遇实现薪资/职业跃迁这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】