Function Calling 实战:我用 OpenAI 工具调用重构了整个客服系统
说实话,刚接触 Function Calling 的时候,我以为它就是个"高级 JSON 解析器"。
直到我在生产环境踩了几个大坑,才明白这东西的水有多深。
背景:一个让人崩溃的客服系统
我们公司的客服系统之前是这样的:
1. 用户问问题
2. AI 生成回答
3. 如果需要查订单,AI 会说"请稍等,我帮您查一下"
4. 然后...就没有然后了
因为 AI 根本不会真的去查订单。它只是"假装"在查。
# 之前的实现 - AI 在"演戏"
def handle_customer_query(query: str) -> str:
response = openai.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": "你是客服助手,如果用户问订单问题,请说'我帮您查一下'"},
{"role": "user", "content": query}
]
)
return response.choices[0].message.content
# 用户: "我的订单 #12345 到哪了?"
# AI: "好的,我帮您查一下订单 #12345 的物流信息...您的订单正在配送中"
# 实际上 AI 根本没查,它在编!Function Calling 是什么
简单说,Function Calling 让 AI 能"调用"你定义的函数。AI 不直接执行函数,而是告诉你"我想调用这个函数,参数是这些",然后你来执行。
# 定义函数
tools = [
{
"type": "function",
"function": {
"name": "get_order_status",
"description": "查询订单状态和物流信息",
"parameters": {
"type": "object",
"properties": {
"order_id": {
"type": "string",
"description": "订单号"
}
},
"required": ["order_id"]
}
}
}
]
# 调用 API
response = openai.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": "我的订单 #12345 到哪了?"}],
tools=tools,
tool_choice="auto"
)
# AI 返回的不是文本,而是函数调用请求
message = response.choices[0].message
if message.tool_calls:
for tool_call in message.tool_calls:
if tool_call.function.name == "get_order_status":
args = json.loads(tool_call.function.arguments)
# 你来执行真正的查询
result = query_order_from_db(args["order_id"])
print(f"订单状态: {result}")踩坑记录
坑 1:参数类型不匹配
AI 有时候会把数字当字符串传,或者把字符串当数字传。
# AI 可能返回 {"order_id": 12345} 而不是 {"order_id": "12345"}
# 如果你的数据库查询要求字符串,就会报错
def safe_get_order(args: dict) -> dict:
"""安全地获取订单"""
order_id = args.get("order_id")
# 类型转换
if isinstance(order_id, int):
order_id = str(order_id)
if not order_id:
return {"error": "缺少订单号"}
return query_order_from_db(order_id)坑 2:AI 会"幻觉"函数名
有时候 AI 会调用一个不存在的函数:
# AI 可能返回 tool_call.function.name = "check_order"
# 但你定义的是 "get_order_status"
def handle_tool_call(tool_call) -> dict:
"""处理函数调用"""
func_name = tool_call.function.name
args = json.loads(tool_call.function.arguments)
# 函数路由
handlers = {
"get_order_status": handle_get_order_status,
"cancel_order": handle_cancel_order,
"refund_order": handle_refund_order,
}
handler = handlers.get(func_name)
if not handler:
return {"error": f"未知函数: {func_name}"}
return handler(args)坑 3:多轮对话的状态管理
Function Calling 需要把函数结果反馈给 AI,这个流程很容易搞乱:
class ConversationManager:
"""管理多轮对话状态"""
def __init__(self):
self.messages = []
self.tools = [...] # 函数定义
def add_user_message(self, content: str):
self.messages.append({"role": "user", "content": content})
def process(self) -> str:
"""处理对话,可能涉及多次函数调用"""
while True:
response = openai.chat.completions.create(
model="gpt-4",
messages=self.messages,
tools=self.tools
)
message = response.choices[0].message
self.messages.append(message)
# 如果没有函数调用,返回文本
if not message.tool_calls:
return message.content
# 处理所有函数调用
for tool_call in message.tool_calls:
result = self.execute_tool(tool_call)
# 把函数结果加入对话
self.messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result)
})
# 继续循环,让 AI 基于函数结果生成回答
def execute_tool(self, tool_call) -> dict:
"""执行工具调用"""
# ... 实现略
pass完整实现
这是我最终的生产级实现:
import json
import openai
from typing import Callable, Dict, Any
class FunctionCallingAgent:
"""生产级 Function Calling 代理"""
def __init__(self, model: str = "gpt-4"):
self.model = model
self.messages = []
self.functions: Dict[str, Callable] = {}
self.tool_definitions = []
def register_function(self, name: str, description: str,
parameters: dict, handler: Callable):
"""注册可调用的函数"""
self.functions[name] = handler
self.tool_definitions.append({
"type": "function",
"function": {
"name": name,
"description": description,
"parameters": parameters
}
})
def chat(self, user_message: str) -> str:
"""处理用户消息"""
self.messages.append({"role": "user", "content": user_message})
max_iterations = 5 # 防止无限循环
for _ in range(max_iterations):
response = openai.chat.completions.create(
model=self.model,
messages=self.messages,
tools=self.tool_definitions if self.tool_definitions else None
)
message = response.choices[0].message
self.messages.append(message)
# 没有函数调用,返回最终回答
if not message.tool_calls:
return message.content
# 执行所有函数调用
for tool_call in message.tool_calls:
result = self._safe_execute(tool_call)
self.messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result, ensure_ascii=False)
})
return "抱歉,处理您的请求时遇到了问题。"
def _safe_execute(self, tool_call) -> dict:
"""安全执行函数"""
try:
func_name = tool_call.function.name
args = json.loads(tool_call.function.arguments)
handler = self.functions.get(func_name)
if not handler:
return {"error": f"未知函数: {func_name}"}
return handler(**args)
except json.JSONDecodeError:
return {"error": "参数解析失败"}
except Exception as e:
return {"error": f"执行失败: {str(e)}"}
# 使用示例
agent = FunctionCallingAgent()
# 注册函数
agent.register_function(
name="get_order_status",
description="查询订单状态",
parameters={
"type": "object",
"properties": {
"order_id": {"type": "string", "description": "订单号"}
},
"required": ["order_id"]
},
handler=lambda order_id: {"status": "已发货", "eta": "明天"}
)
# 对话
response = agent.chat("我的订单 #12345 到哪了?")
print(response)性能优化
Function Calling 比普通对话慢,因为可能涉及多次 API 调用。我的优化方案:
1. 并行执行 - 多个独立的函数调用可以并行
2. 结果缓存 - 相同参数的调用可以缓存结果
3. 流式输出 - 最终回答使用流式,减少等待感
import asyncio
from functools import lru_cache
class AsyncFunctionCallingAgent(FunctionCallingAgent):
"""异步版本,支持并行函数调用"""
async def chat_async(self, user_message: str) -> str:
self.messages.append({"role": "user", "content": user_message})
while True:
response = await self._async_call()
message = response.choices[0].message
self.messages.append(message)
if not message.tool_calls:
return message.content
# 并行执行所有函数调用
tasks = [
self._async_execute(tool_call)
for tool_call in message.tool_calls
]
results = await asyncio.gather(*tasks)
for tool_call, result in zip(message.tool_calls, results):
self.messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result)
})总结
Function Calling 是个强大的工具,但要注意:
1. 参数验证 - 永远不要信任 AI 传的参数
2. 错误处理 - 函数执行可能失败,要有兜底
3. 状态管理 - 多轮对话的消息历史要管理好
4. 性能优化 - 考虑并行和缓存
希望这篇文章能帮你少走一些弯路。
读者评论 5