最近在做一个Agent类项目,被工具调用折磨得够呛。最初我写了一大堆函数,然后靠正则和关键词把用户意图往外掰,结果越掰越乱。后来痛定思痛,把所有能力统一重构成了基于千问API的skill设计方案。核心思路是把每个工具能力封装成一份机器可读的skill描述,让模型自己决定什么时候调用、怎么填参数,我只维护一个注册表。这篇复盘会包含skill协议如何设计、参数Schema怎么写才不会让模型瞎填、Function Calling报文如何处理,以及一些官方文档不会告诉你的坑。如果你想做Agent、私有化助手,或者正准备接大模型Function Calling,这份方案可以直接拿来当起点。
1. 为什么要把能力封装成Skill,而不是直接调函数
1.1 我最初是怎么把代码写废的
先说反面教材。我早期的调用逻辑长这样:用户说“帮我查杭州到北京的高铁”,我就先分词,匹配“查”“高铁”这些词,然后从文本里正则掏出城市名。看起来能用,实际上每加一种说法就要加一条规则,模型回答稍微换个句式,正则就失配。更头疼的是,如果用户在一个句子里同时要求“查天气顺便定个闹钟”,这种规则引擎直接傻眼。
真正让我下决心重写的,是遇到一个场景:用户要求“明天早上提醒我带伞,顺便看一下杭州天气”。我原来的代码只能识别出“查天气”,提醒功能被无视。这说明意图识别和工具调用必须拆开:模型负责做意图判断,系统负责执行能力。而skill就是这个拆分的关键单位。
1.2 Skill的本质:给模型一本可操作的工具手册
Skill这东西听起来玄,本质是一份“机器可读的函数说明书”。它包含三部分:功能描述、输入参数的JSON Schema、对应的执行函数。模型在对话中看到用户需求后,会先判断“这个需求需要调用哪个skill”,然后按Schema生成参数,把调用意图返回给系统。系统再真正去执行函数,把结果回传,模型再继续组织回答。
我用一个生活类比:Skill描述文件就像给新员工看的操作手册。手册里不会写“该员工应该如何优秀”,而是清楚写“当遇到什么情况时,执行什么动作,需要哪些材料,材料格式是什么”。模型就是那个读手册的人,写得越清楚,它越不会自作主张。
1.3 设计目标:我的三个底线
在动手之前,我给自己定了三条底线,后面所有设计都围绕这三条展开。
第一,新增能力不改主流程。我的主循环代码只处理“调模型、拿结果、执行、回传”这一套逻辑,新加一个功能只需要注册一份skill描述和一个handler,主流程零改动。
第二,模型填错参数要被发现。参数校验必须放在执行前,不能让一个明显不合理的参数跑到真实业务里。
第三,过程可观测。每次模型选择了哪个skill、填了什么参数、函数返回了什么、最终怎么回答的,都要有日志。否则出了问题你根本不知道是模型理解错了,还是函数写错了。
2. 基于千问API的Skill整体架构设计
2.1 分层架构:接入层、注册中心、执行器
整套架构我分成四层。接入层负责接收用户消息,组装上下文,调用千问API;注册中心维护所有的skill描述和执行函数;执行器负责参数校验、调用handler、返回结果;决策层由模型自身充当,根据用户消息决定调用哪个skill。
调用链是这样的:用户消息进入接入层,系统把历史对话和可用skill的tools描述一起发给千问API。模型如果认为需要调用能力,会返回tool_calls,里面带skill名称和参数。接入层拿到后交给执行器,执行器做完校验和调用,把结果构造成一条tool消息追加到对话里,再发回给模型。模型拿到结果后继续生成回答或者继续调用下一个skill,直到输出最终文本。
这个循环是最核心的机制,后面所有优化都围绕它展开。
2.2 接入方式:我为什么选OpenAI兼容模式
千问API提供了两种主流的调用方式,一种是原生SDK风格,一种是兼容OpenAI协议的模式。我选了兼容模式,理由很直接:我现有的大量Python脚本和内部工具都基于openai库的调用习惯,切换到兼容模式不需要动底层代码,只需要改模型名和网关地址。
兼容模式还天然支持tools参数,也就是Function Calling的标准写法。这对skill方案是刚需,因为我的每一步设计都依赖这个参数把skill描述传给模型。原生SDK当然也能做,但我在团队内部测试时发现,兼容模式对Prompts、数据流和异常处理的经验可以复用,学习成本低得多。
2.3 注册表怎么设计:一份dict就是中心
skill注册表不需要复杂框架,我用一个Python字典就能撑起整个体系。每个skill包含三个字段:给模型看的schema、实际执行的handler、以及用于内部管理的元信息。handler从字符串映射到函数,schema直接原样传给千问API的tools参数。
SKILL_REGISTRY = { "query_ticket": { "schema": { "type": "function", "function": { "name": "query_ticket", "description": "查询火车票余票与运行时刻。用户需要查询车次、余票、出发到达时间时调用。", "parameters": { "type": "object", "properties": { "from": {"type": "string", "description": "出发城市,例如杭州"}, "to": {"type": "string", "description": "到达城市,例如北京"}, "date": {"type": "string", "description": "出发日期,格式YYYY-MM-DD"} }, "required": ["from", "to", "date"] } } }, "handler": "handle_query_ticket", "enabled": True } }为什么要多存一个enabled字段?因为灰度发布时我可以临时关掉某个skill,而不用删除代码。主流程只需要遍历这个注册表,把enabled的skill筛出来转成tools列表,维护成本非常低。
3. Skill协议定义与核心实操
3.1 一份标准的Skill描述文件长什么样
直接看示例更直观。我把skill描述文件叫schema,里面最核心的字段是name、description和parameters。
name必须全局唯一,模型就是靠这个字段去匹配执行函数。我建议命名用动词加对象,比如query_ticket、create_reminder、send_email,不要用function_1这种无意义命名,否则千问API的返回结果里你根本分不清调的是谁。
description是给模型看的,不是给人看的。它决定了模型在哪些情况下会触发这个skill。写得太宽泛,模型会过度调用;写得太窄,该调的时候不调。我一般会写清楚“什么场景下调用”和“什么场景下不要调用”。
parameters用的是JSON Schema格式。千问API支持的字段包括type、properties、required、enum、description等。千万别在这一层省事,参数定义得越细,模型填参的准确率越高。
3.2 参数Schema怎么写,模型才不容易填错
我总结了四个实战要点,每个都是拿真实对话试出来的。
第一,description里必须带格式示例。比如日期字段如果只写“出发日期”,模型可能给你返回明天、下周一这类自然语言。加上“格式YYYY-MM-DD”之后,返回结果基本就规整了。
第二,能用enum就用enum。比如性别、城市列表、服务类型这类有固定取值范围的参数,直接在Schema里列出来,比在description里写一百遍“只能从列表中选择”管用得多。
第三,不要给模型填不了的字段。有些内部参数比如用户ID、请求来源、trace_id,应该由系统在执行层注入,而不是让模型生成。模型不知道这些值,放进Schema只会诱导它编造。
第四,required里的字段必须慎重。如果某个字段并非每次都需要,就不要放进required,否则模型会为了满足约束强行填一个默认值,而你很难发现它是编的。
3.3 把Skill注入千问API的两种方式
我测试过两种注入方式。第一种是走tools参数,这是最正规的Function Calling路线。每个skill的schema放进tools列表,模型在需要时会返回tool_calls。这种方式的好处是模型知道自己在“用工具”,参数是结构化生成的,非常稳定。
第二种是把所有工具描述拼进system prompt,让模型看着描述自行输出一段JSON。这种方式适合那些老版本模型不支持Function Calling的情况,但坏处非常明显:模型可能输出格式不稳定的JSON,参数转义容易出错,而且会挤占上下文窗口。
我的结论是,只要你的千问API版本支持tools参数,就无脑用第一种。我在兼容模式下调通了完整链路之后,就再也没碰过第二种方案。
3.4 一次完整调用链的实现
下面这段代码是这套方案最核心的部分。它包含一轮完整的“模型决策-执行skill-结果回传-再生成”循环,可以直接复制去改。
import os import json from openai import OpenAI client = OpenAI( api_key=os.getenv("LLM_API_KEY"), base_url=os.getenv("LLM_BASE_URL"), # 千问API兼容网关地址 ) MODEL_NAME = "qwen-plus" def build_tools(): return [ SKILL_REGISTRY[name]["schema"] for name, skill in SKILL_REGISTRY.items() if skill["enabled"] ] def execute_skill(skill_name: str, arguments: dict): """根据skill_name找到handler并执行。""" if skill_name not in SKILL_REGISTRY: return json.dumps({"error": f"skill {skill_name} not found"}) handler_func = globals()[SKILL_REGISTRY[skill_name]["handler"]] return json.dumps(handler_func(**arguments), ensure_ascii=False) def chat_with_skills(user_message: str, history: list | None = None): messages = history or [] messages.append({"role": "user", "content": user_message}) # 防止模型陷入工具调用死循环 for _ in range(5): resp = client.chat.completions.create( model=MODEL_NAME, messages=messages, tools=build_tools(), temperature=0.2, ) choice = resp.choices[0] if choice.finish_reason == "tool_calls" and choice.message.tool_calls: # 先记录模型的选择,方便复盘 for tc in choice.message.tool_calls: fn = tc.function print(f"[tool_call] {fn.name} -> {fn.arguments}") # 把模型返回的中间消息放进上下文 messages.append(choice.message) for tc in choice.message.tool_calls: fn = tc.function # 参数解析失败时也要给模型返回错误,让它自行修正 try: args = json.loads(fn.arguments) except json.JSONDecodeError as e: args = {"_parse_error": str(e)} result = execute_skill(fn.name, args) messages.append({ "role": "tool", "tool_call_id": tc.id, "content": result, }) # 继续让模型基于工具结果生成 continue return choice.message.content return "[error] skill run exceeds max iterations"这段代码里有两个容易被忽略的细节。第一,messages.append(choice.message)这一步不能省。如果不把模型返回的中间消息放入上下文,千问API就不知道之前的tool_call是它自己发起的,可能直接报错。第二,tool_call_id必须和模型返回的id严格一致,这是关联工具结果和调用请求的唯一凭证。我一开始漏掉这个字段,结果模型反复认为自己没有调用过工具。
4. 实战中绕不开的Token预算与并发控制
4.1 Token到底花在哪里了
很多人低估了tools参数对Token消耗的影响。我实测了一次普通请求,包含5个skill描述的情况下,仅仅tools字段就吃掉了接近800个Token,占了整轮请求的四分之一。如果skill数量加到15个,光tools描述就超过2000 Token。这个问题在千问API这类按Token计费的模型上会被放大,每轮对话都带全量tools,成本很快失控。
我给过一个简单估算公式。假设单个skill的Schema平均100到150 Token,使用10个skill,每轮请求的Tools固定消耗就是1000到1500 Token。再加上system prompt、对话历史和模型输出预算,一次调用的总Token数很容易冲到3000以上。如果每天调用上万次,这就不只是技术问题了,是预算问题。
4.2 别把全部Skill一股脑塞给模型
我的解决方案是给skill做分组路由。注册表里每个skill增加一个groups字段,比如“出行”“办公”“生活”“系统管理”。每次请求之前,先根据用户的会话特征或者一个简单的意图预筛模块,决定本次请求只注入哪几个组。
比如用户上来就说“帮我看看杭州天气”,路由模块识别到天气意图,系统只注入weather和schedule这两个组的skill,而不是全部20个。这样tools字段的Token消耗从2000降到400。实现起来不复杂,最简单的做法是基于关键词做预筛,更稳的是单独用小模型做意图分类。先跑通简单版,再逐步替换大模型分类,性价比很高。
4.3 上下文裁剪:对话历史不能无限膨胀
多轮对话场景里,历史消息堆得越长,Token消耗越高,模型对早期内容的注意力也越差。我发现千问API对超长上下文的处理能力在提升,但成本是实打实的。所以我实现了滑动窗口:最多保留最近六轮对话。
如果用户早期提到过一个杭州的行程,直到第六轮以后还在追问这个行程,六轮窗口就会把关键信息丢掉。为了尽量避免这种问题,我会在窗口之外额外维护一个“关键信息摘要”,每次裁剪时把用户是否提及过的城市、时间、任务编号等压缩成一行system消息。这样既控制了Token,又不至于丢失主线信息。
4.4 并发、限流与超时控制
千问API有并发限制,具体阈值跟你的账号和模型有关。我踩过的坑是:自己写的代码在单测时很稳,一上生产就时不时收到限流错误。后来我加了两层保护。
第一层是请求超时。client.chat.completions.create要传timeout参数,我默认设为30秒。模型响应超过30秒直接按失败处理,而不是无限等下去。
第二层是重试机制。限流错误不是瞬时错误,立刻重试只会加重负担。我采用指数退避策略,第一次等1秒、第二次等2秒、第三次等4秒,最多重试三次。
import time def request_with_retry(**kwargs): for attempt in range(3): try: return client.chat.completions.create(timeout=30, **kwargs) except Exception as e: if attempt == 2: raise e time.sleep(1 * (2 ** attempt))这段代码看起来简单,但它在生产环境帮我挡住了大量偶发限流问题。重试逻辑里最重要的不是重试本身,而是退避时间。如果三个并发请求同时失败,大家一起等1秒后重试,大概率还会同时打过去。稍微加一点随机抖动会更好,但我图简单,固定退避也够用了。
5. 测试方法与问题排查实录
5.1 先写一个可重复的测试框架
做功能开发时,我最怕的是模型输出不稳定。同一个输入,上一次调用skill,下一次直接给文本答案。这给回归测试带来了麻烦。我的做法是把temperature调低到0.2,保证模型决策尽量稳定。
然后我用mock对象替换掉skill内部的真实依赖。比如query_ticket这个handler,内部其实要查一个外部票务接口。测试时我直接让handler返回固定JSON,不去真的请求外部服务。这样就能稳定验证“模型有没有正确触发skill”“参数有没有正确传入”。
测试用例我用Python的pytest组织,每个用例只测一条链路:给定某个用户输入,断言最终返回文本里包含期望的关键信息,并且断言日志里出现了期望的tool_call记录。temperature=0.2下,同一用例多次运行的结果基本一致,出问题也能稳定复现。
5.2 常见问题速查表
我把开发过程中遇到的问题汇总成一张表,方便排查。
| 问题现象 | 可能原因 | 排查方案 |
|---|---|---|
| 模型从头到尾不调用skill | schema没有传入tools、description太笼统、温度过高 | 打印实际请求的tools列表,确认schema字段名正确 |
| 该调A却调了B | 两个skill的description语义重叠 | 把description写得更互斥,明确“不要可以处理什么” |
| 参数填了模型自己编的值 | 参数缺少enum或格式约束 | 在parameters里增加enum和format示例 |
| 模型返回tool_calls后系统报缺失tool_call_id | 中间消息没有追加到messages列表 | 检查是否将choice.message放入messages |
| 工具返回结果后模型反复调用同一个skill | 函数返回了错误信息但没有说明下一步建议 | 在执行失败时,把原因和建议一并写入返回内容 |
| Token消耗突然暴增 | 全部skill无差别注入 | 做分组路由,只注入本次会话需要的skill |
| 并发一高就报限流错 | 没有退避重试或并发数超过账号限制 | 加指数退避,必要时做本地并发信号量 |
这七种问题几乎覆盖了我在整个项目里遇到的绝大多数故障。其中“该调A却调B”是最难排查的,因为它不是报错,而是逻辑错。后来我发现,与其在system prompt里拼命写“你只能……”,不如把两个skill的description互相做区分,分别写明各自的边界和禁止场景。
5.3 我踩过的几个坑
第一个坑是description写太长。最初我用一整段话描述一个skill的触发条件,结果模型几乎每次请求都调它,Token飙升。后来我把description压缩到两句话:第一句说“什么时候调用”,第二句说“不要处理什么”。准确率反而提高了。
第二个坑是函数返回纯文本而不是结构化JSON。模型拿到文本后虽然能理解,但在后续多轮处理时容易出现解析错误。我现在所有handler统一返回JSON字符串,并且包含success、data、error_message三个字段,模型处理起来非常有规律。
第三个坑是全局变量串数据。我在某个handler里用了一个模块级全局字典做缓存,上线后发现两个不同会话的数据偶尔会串。排查了很久,才意识到是多个协程共用了同一份全局状态。最后把缓存改成请求级别的上下文对象,问题立刻消失。
第四个坑是stream模式下的tool_calls处理。如果开了流式输出,千问API的tool_calls会以增量片段返回,需要自己拼接。这块逻辑容易漏字段。我的建议是优先用非流式跑通功能,等稳定了再考虑流式体验优化。
6. 落地体验与几点私人心得
这套skill方案上线后,我最直观的感受是“新增功能不再心惊胆战”。以前加一个新工具,要改主流程、加规则、补测试,半天起步。现在把schema写好、handler实现、注册表里登记,前后不到半小时。更重要的是,模型对意图的泛化能力远超我原来那套正则匹配,同一句话换十种说法,它都能稳定触发正确的skill。用户侧反馈也稳了很多,至少没再出现“说了提醒却不提醒”的情况。
最后分享一个我自己摸索出来的习惯:加skill之前先想清楚它是不是“真实高频需求”。做Agent项目的人很容易犯一个错,就是一口气注册十多个skill,觉得越多越智能。实际上,skill越多,tools字段越臃肿,模型决策的准确率反而下降。我现在的做法是先只注册三个核心skill跑通闭环,再根据真实对话日志把用户高频诉求逐个转成新skill。每加一个skill,就跑一遍回归测试,确认不会影响已有场景。这个节奏看起来慢,但比一次性铺开所有能力要稳得多。
如果你也在做类似的Agent,希望这篇复盘能帮你少走一点弯路。Skill方案最迷人的地方在于,它把“教模型用工具”这件事从模糊的调参游戏变成了可维护的软件工程。