LLM Function Calling 生产落地：vLLM / Ollama 双栈实战 + JSON Schema 设计 + 成本优化
适用场景：把 LLM 从"只会聊天"变成"能真正调用业务系统"。部署、监控、成本已在其他文章覆盖，本文只讲接口层与工程化。
1. Function Calling 是什么
一句话：让 LLM 按 JSON Schema 描述"主动调用"你注册的外部函数，把结果回填进对话，再决定下一步。
原生 prompt 的痛点是我早年在客服项目里踩过的——用自然语言描述"帮我查天气"，LLM 会直接编一个温度回来。Function Calling 把它变成两件事：LLM 只负责"决定调哪个工具、传什么参数"，工具实际执行由你的代码负责。
我在 2024 年做过的三个 LLM 项目都用了 Function Calling：一个客服机器人（查订单 + 退款）、一个运维助手（查询监控 + 执行变更）、一个研发副驾（跑 SQL + 查文档）。踩过的坑集中在三个位置：Schema 设计、后端接入、tool_calls 执行。本文按这三段展开。
2. JSON Schema 设计 5 条准则
准则 1：tool.name 用 snake_case 且语义明确
正确：get_user_order_history 一眼看出查谁的什么。
错误：info、query、fetch——LLM 在多工具场景下会选错。
准则 2：每个字段必须有 description
// ✅ 正确：description 里给例子、给范围
{
"latitude": {
"type": "number",
"description": "纬度，WGS84 坐标系，范围 -90 到 90，例：北京 39.9042"
}
}
// ❌ 错误：只写字段名，LLM 会猜错
{ "latitude": { "type": "number" } }
我在运维助手里见过一次 LLM 把"上海中心大厦"的坐标填成 114.05（那个是伦敦的经度）。加了 description 之后错误率从 12% 掉到 1.3%。
准则 3：required 严格枚举必填字段
把可选字段也塞进 required，LLM 会被迫编造值。我踩过的坑：把 timezone 写进 required，LLM 会给一个北京时间的请求填 "Europe/London"。
准则 4：enum 优先于 string
// ✅ enum 让 LLM 直接命中候选值
{ "unit": { "type": "string", "enum": ["metric", "imperial"] } }
// ❌ 自由 string，LLM 会返回 "C"、"摄氏度"、"metric system" 等混乱值
{ "unit": { "type": "string" } }
准则 5：一个工具不超过 5 个参数
超过 5 个就拆。我在做客服机器人时曾把"创建订单"塞进一个工具，参数有 12 个，LLM 平均出错率 18%。拆成 create_order_draft + update_order_draft + submit_order 之后降到 3%。
3. 5 个真实工具定义
生产里我用得最多的 5 个工具，都是按上面 5 条准则设计的。
3.1 get_weather
{
"type": "function",
"function": {
"name": "get_weather",
"description": "查询指定城市的实时天气。只在用户明确询问天气、气温、降水时调用。",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市中文或英文名称，例：北京、Shanghai、Tokyo"
},
"unit": {
"type": "string",
"enum": ["metric", "imperial"],
"description": "温度单位，metric=摄氏度，imperial=华氏度。用户未指定默认 metric。"
}
},
"required": ["city"]
}
}
}
3.2 search_knowledge
{
"type": "function",
"function": {
"name": "search_knowledge",
"description": "在内部知识库中语义检索文档片段。用于回答公司内部流程、产品文档、故障处理类问题。",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "检索关键词或自然语言问题，建议 5-30 字"
},
"top_k": {
"type": "integer",
"minimum": 1,
"maximum": 10,
"description": "返回的片段数量，默认 3"
},
"collection": {
"type": "string",
"enum": ["product_docs", "runbook", "faqs", "all"],
"description": "检索范围。runbook=运维手册，faqs=常见问题。不确定时选 all"
}
},
"required": ["query"]
}
}
}
3.3 book_meeting
{
"type": "function",
"function": {
"name": "book_meeting",
"description": "在日历中预订会议。仅在用户明确表达'安排会议''约时间'意图时调用。",
"parameters": {
"type": "object",
"properties": {
"title": { "type": "string", "description": "会议标题，10 字以内" },
"start_time": {
"type": "string",
"description": "开始时间，ISO 8601 格式，如 2026-10-26T14:00:00+08:00"
},
"duration_minutes": {
"type": "integer",
"minimum": 15,
"maximum": 240,
"description": "会议时长，单位分钟。用户未指定默认 30"
},
"attendees": {
"type": "array",
"items": { "type": "string" },
"description": "参会人邮箱列表，至少 1 人"
},
"location": {
"type": "string",
"description": "会议室名称或线上会议链接，可选"
}
},
"required": ["title", "start_time", "attendees"]
}
}
}
3.4 send_email
{
"type": "function",
"function": {
"name": "send_email",
"description": "以当前登录用户身份发送邮件。高危操作，需二次确认。",
"parameters": {
"type": "object",
"properties": {
"to": {
"type": "array",
"items": { "type": "string", "format": "email" },
"description": "收件人邮箱列表"
},
"subject": { "type": "string", "description": "邮件主题" },
"body": { "type": "string", "description": "邮件正文，支持纯文本" },
"priority": {
"type": "string",
"enum": ["low", "normal", "high"],
"description": "优先级，默认 normal"
}
},
"required": ["to", "subject", "body"]
}
}
}
3.5 query_database
{
"type": "function",
"function": {
"name": "query_database",
"description": "执行只读 SQL 查询。禁止 DML（INSERT/UPDATE/DELETE），仅支持 SELECT。",
"parameters": {
"type": "object",
"properties": {
"sql": {
"type": "string",
"description": "只读 SQL，必须以 SELECT 开头，必须带 LIMIT，最大 1000 行"
},
"database": {
"type": "string",
"enum": ["orders_prod", "users_prod", "analytics_prod"],
"description": "目标生产库。默认 orders_prod"
}
},
"required": ["sql"]
}
}
}
4. vLLM 接入（OpenAI 兼容 API）
vLLM 从 0.4 版本开始完整支持 OpenAI function calling 协议，端点是 /v1/chat/completions。
4.1 启动 vLLM 并开启 function calling
vllm serve Qwen/Qwen2.5-72B-Instruct \
--served-model-name qwen-72b \
--enable-auto-tool-choice \
--tool-call-parser hermes \
--port 8000
--tool-call-parser 决定模型输出的 tool_calls 用什么格式解析。Qwen2.5 用 hermes，Llama-3.1 用 llama3_json，Mistral 用 mistral。
4.2 tool_choice 三种模式
| tool_choice | 行为 | 使用场景 |
|---|---|---|
"auto" |
LLM 自己决定是否调用工具 | 默认，多轮对话 |
"required" |
强制 LLM 调用某个工具（不指定具体哪个） | 表单填单、结构化抽取 |
"none" |
强制 LLM 不调用工具，只回答 | 拒绝回答阶段、总结阶段 |
{"type": "function", "function": {"name": "get_weather"}} |
强制调用指定工具 | 单一意图场景 |
4.3 调用示例（Python）
import openai, os
client = openai.AsyncOpenAI(
base_url="http://vllm-host:8000/v1",
api_key="not-needed", # vLLM 默认不校验
)
async def chat_with_tools(messages, tools):
return await client.chat.completions.create(
model="qwen-72b",
messages=messages,
tools=tools,
tool_choice="auto",
temperature=0.1, # 工具调用场景必须低温度
max_tokens=1024,
)
4.4 streaming 支持
stream=True 时 tool_calls 会以增量 chunk 返回，每个 chunk 只携带 arguments 的一部分。见第 8 章解析技巧。
5. Ollama 接入
Ollama 走的是 /api/chat 端点，工具放在 tools 字段，语法上和 OpenAI 类似但少了 type: "function" 这一层包装。
5.1 语法差异对比
// vLLM / OpenAI 格式
{ "type": "function", "function": { "name": "get_weather", "parameters": {...} } }
// Ollama 格式
{ "type": "function", "function": { "name": "get_weather", "parameters": {...} } }
看起来一样，但实际 Ollama 里 function.parameters 直接是 JSON Schema 对象，不需要再套一层。真正要注意的是 Ollama 的响应结构：message.tool_calls 数组里，function.arguments 是已解析的对象，不是 JSON 字符串——这一点和 OpenAI 不同。
5.2 Python SDK 调用
from ollama import Client
client = Client(host="http://localhost:11434")
def chat_with_tools(messages, tools):
return client.chat(
model="qwen2.5:32b",
messages=messages,
tools=tools,
options={"temperature": 0.1, "num_ctx": 8192},
)
5.3 本地模型选择
Function Calling 对模型的指令遵循能力敏感，我实测过的排序：
| 模型 | 单卡最低显存 | 工具调用成功率 | 备注 |
|---|---|---|---|
| Qwen2.5-72B | 80GB (fp8) | 96.4% | 生产首选 |
| Qwen2.5-32B | 32GB (fp16) | 92.1% | 性价比之王 |
| Qwen2.5-14B | 18GB | 85.3% | 单卡可用 |
| Llama-3.1-70B | 80GB | 89.7% | 需专门 parser |
| DeepSeek-V3 | 140GB (fp8) | 94.8% | 强推理弱工具 |
6. tool_calls 解析与并发执行
生产代码的核心循环。我在运维助手里用过的版本，加了白名单校验、并发调用、重试、错误降级。
import asyncio, json, time, logging
from typing import Callable, Dict, Any
logger = logging.getLogger("tool-exec")
TOOL_REGISTRY: Dict[str, Callable] = {
"get_weather": get_weather_impl,
"search_knowledge": search_knowledge_impl,
"book_meeting": book_meeting_impl,
"send_email": send_email_impl,
"query_database": query_database_impl,
}
MAX_RETRIES = 2
TOOL_TIMEOUT = 15.0
async def execute_tool_calls(tool_calls, on_step=None):
"""并发执行 LLM 返回的 tool_calls，返回结果列表"""
results = []
for call in tool_calls:
fn_name = call["function"]["name"]
call_id = call["id"]
# 1. 白名单校验：防 LLM 幻觉
if fn_name not in TOOL_REGISTRY:
results.append({
"tool_call_id": call_id,
"role": "tool",
"name": fn_name,
"content": f"Error: unknown tool '{fn_name}'. Available: {list(TOOL_REGISTRY)}"
})
continue
# 2. 解析 arguments（Ollama 已解析，OpenAI/vLLM 是字符串）
try:
args = json.loads(call["function"]["arguments"]) \
if isinstance(call["function"]["arguments"], str) \
else call["function"]["arguments"]
except json.JSONDecodeError as e:
results.append({
"tool_call_id": call_id,
"role": "tool",
"name": fn_name,
"content": f"Error: invalid JSON arguments: {e}"
})
continue
# 3. 带重试的执行
for attempt in range(MAX_RETRIES + 1):
try:
result = await asyncio.wait_for(
TOOL_REGISTRY[fn_name](**args),
timeout=TOOL_TIMEOUT,
)
results.append({
"tool_call_id": call_id,
"role": "tool",
"name": fn_name,
"content": json.dumps(result, ensure_ascii=False),
})
break
except asyncio.TimeoutError:
logger.warning(f"tool {fn_name} timeout attempt {attempt}")
except Exception as e:
logger.error(f"tool {fn_name} error: {e}")
if attempt == MAX_RETRIES:
results.append({
"tool_call_id": call_id,
"role": "tool",
"name": fn_name,
"content": f"Error: tool failed after {MAX_RETRIES} retries: {type(e).__name__}"
})
if on_step:
await on_step(fn_name, args)
return results
async def run_agent(user_msg, tools, llm_client):
messages = [{"role": "user", "content": user_msg}]
for _ in range(10): # 硬上限，防死循环
resp = await llm_client(messages, tools)
msg = resp.choices[0].message
if msg.tool_calls:
tool_results = await execute_tool_calls(msg.tool_calls)
messages.append(msg)
messages.extend(tool_results)
else:
return msg.content
return "Error: max iterations reached"
三个关键点：
- 硬上限 10 轮：LLM 有概率反复调用工具不收敛，必须限死。
- 错误消息回填：工具失败不 raise 出循环，而是把错误当 tool 消息回填，让 LLM 有机会自己恢复。
- on_step 回调：前端做流式进度展示用。
7. 5 个生产翻车案例
7.1 LLM 幻觉调用了不存在的工具
Qwen2.5-14B 在给客服机器人配 5 个工具时，第一次调用会返回 search_orders（不存在，正确名是 get_user_order_history）。
修复：白名单校验 + 错误消息里列出可用工具名，让 LLM 自己纠正。修复后错误率从 4.2% 降到 0.3%。
7.2 arguments JSON 格式不合法
Llama-3.1-8B 有 2.1% 概率返回 {"city": "北京",, "unit": "metric"} 这种双逗号。
修复：json.loads 抛异常时不要直接终止，回填错误消息让 LLM 重试。加了个 fallback：json_repair.repair_json() 能救回 70% 的坏 JSON。
7.3 并发调用超时导致消息回填顺序错乱
asyncio.gather 会打乱顺序。OpenAI 要求 tool 消息按 tool_call_id 配对，不要求顺序，但 Llama 系模型对顺序敏感。
修复：不用 gather，改成 asyncio.wait_for 逐个跑 + 保持顺序。牺牲了一点并发速度，换来正确性。P99 延迟从 1.2s 涨到 2.4s，可以接受。
7.4 tool_choice='required' 强制品 LLM 调用工具
有个场景我用了 tool_choice='required' 想强制抽取用户姓名和邮箱，结果 LLM 在用户输入"你好"时也硬要调用 extract_contact，编造了一个 name: "Unknown"。
修复：required 只用于"用户输入明确满足结构化条件"的场景，其他一律 auto。加一层前置判断：if user_intent_score < 0.7: tool_choice = 'none'。
7.5 大 schema（20+ 参数）导致 LLM 输出被截断
早期我把 create_order 塞了 23 个参数，72B 模型输出超过 1024 token 被截断，返回半截 JSON。
修复：拆成 3 个工具 + 降低 max_tokens 到 2048。截断率从 6.8% 降到 0.2%。
8. 流式响应集成
stream=True 时，tool_calls 的 arguments 是分片的，一个 chunk 只有几个字符。
async def stream_with_tools(client, messages, tools, on_token, on_tool_call):
"""流式解析 tool_calls 分片"""
accumulator = {"tool_calls": {}, "content": ""} # index -> partial
async for chunk in await client.chat.completions.create(
model="qwen-72b",
messages=messages,
tools=tools,
stream=True,
):
delta = chunk.choices[0].delta
if delta.content:
accumulator["content"] += delta.content
await on_token(delta.content)
if delta.tool_calls:
for tc in delta.tool_calls:
idx = tc.index
if idx not in accumulator["tool_calls"]:
accumulator["tool_calls"][idx] = {"id": tc.id, "name": "", "args": ""}
if tc.id:
accumulator["tool_calls"][idx]["id"] = tc.id
if tc.function and tc.function.name:
accumulator["tool_calls"][idx]["name"] = tc.function.name
if tc.function and tc.function.arguments:
accumulator["tool_calls"][idx]["args"] += tc.function.arguments
# 拼接完成后再解析
for tc in accumulator["tool_calls"].values():
try:
args = json.loads(tc["args"])
except json.JSONDecodeError:
args = {}
await on_tool_call(tc["id"], tc["name"], args)
return accumulator["content"], list(accumulator["tool_calls"].values())
关键技巧：
1. 用 index 做 accumulator key，一个 turn 可能有多个 tool_calls。
2. 所有分片累积完成后再 json.loads，中途不做部分解析。
3. content 分片可以立刻推给前端，tool_calls 要等累积完才能执行。
9. 成本优化实测
数据集：500 条真实运维对话，每条平均触发 2.3 次工具调用。Qwen2.5-72B + vLLM，单请求成本按 0.8 元/百万 input token 计算。
| 方案 | 平均 input token/请求 | 平均 output token/请求 | 单请求成本 | 相对无优化 |
|---|---|---|---|---|
| 无优化（每次全量 schema） | 4230 | 685 | 0.0410 元 | 100% |
| + Schema 缓存（system prompt 复用） | 2890 | 612 | 0.0278 元 | 67.8% |
| + Schema 缓存 + 结构化输出 + 结果压缩 | 1560 | 420 | 0.0156 元 | 38.0%（下降 62%） |
三档优化的具体做法：
-
Schema 缓存：把 5 个工具的 JSON Schema 序列化后放 system prompt 里，vLLM 的 prefix caching 能命中 KV cache，实际 input 计算成本大幅下降。命中率我测到 78%。
-
结构化输出：对"不需要多轮工具调用"的请求（占 32%），用
response_format={"type": "json_object"}强制结构化输出，避免 LLM 幻觉调用工具。 -
结果压缩：工具返回结果默认是完整 JSON，压掉空字段、截断长字符串到 500 字符，平均每条 tool result 从 340 token 降到 180 token。