Function Calling编程实践(OpenAI API)

目录

• Function Calling编程实践
  • 一、Function Calling概述
    • 1. Function Calling 工作流程概述（OpenAI API）
  • 二、Function Calling实现详解
    • 1. 定义Function Schema
    • 2. 系统提示词设计
    • 3. Function Calling API 调用实现
    • 4. OpenAI客户端配置
  • 三、在客服系统中的集成应用
    • 1. 路由节点实现
    • 2. 在客服工作流中的完整调用链
  • 四、案例流程解析
    • 用户查询处理示例
      • 场景：用户询问订单发货状态

Function Calling编程实践

一、Function Calling概述

在大语言模型（LLM）应用中，Function Calling允许我们以 JSON 格式向 LLM 模型描述预定义函数，并使用模型的固有推理能力来决定在生成响应之前是否调用该函数。模型本身不执行函数，而是生成包含函数名称和执行函数所需的参数的 JSON （注意：模型可能会幻觉参数）。

本文通过一个电商客服系统，以其中的路由智能体RouterAgent通过Function Calling实现智能路由为例子，来加深对Function Calling的理解。

参考文章：
OpenAI function-calling中文文档链接
一篇不错的博客

1. Function Calling 工作流程概述（OpenAI API）

1. 定义 Function Schema（向模型描述预定义的函数）
   ↓
2. 构建系统提示词（指导模型的功能，如何调用函数）
   ↓
3. 准备用户消息和上下文
   ↓
4. 调用 OpenAI/DeepSeek API 并传入以下参数：
   • messages（系统提示 + 用户消息）
   • tools（function schema 定义）
   • tool_choice="required"（强制调用）
     ↓
5. 模型返回 tool_calls 响应
   ↓
6. 解析 function 调用参数
   ↓
7. 使用解析后的数据进行下一步操作（智能体路由）

二、Function Calling实现详解

1. 定义Function Schema

首先，我们需要为预定义函数（路由功能）定义一个清晰的函数接口规范（向模型描述函数的调用时机，如何调用等）：

# 定义function call的schema
self.function_schema = {"name": "route_user_query","description": "将用户查询路由到合适的智能体","parameters": {"type": "object","properties": {"target_agent": {"type": "string","enum": ["order_agent", "product_agent", "after_sales_agent", "promotion_agent", "general_agent", "END"],"description": "目标智能体的名称"},"confidence": {"type": "number","minimum": 0,"maximum": 1,"description": "路由决策的置信度，0到1之间的数值"},"user_intent": {"type": "string","description": "用户意图的简洁总结，便于下游智能体理解"},"original_query": {"type": "string","description": "用户的原始查询内容"}},"required": ["target_agent", "confidence", "user_intent", "original_query"]}
}

核心参数如下：

• tool_choice：默认情况下，tool_choice 为 tool_choice: "auto"，模型可以决定是否调用函数以及如果要调用的话，哪些函数要调用。

1. 要强制模型始终调用一个或多个函数，可以将 tool_choice: "required"。
2. 要强制模型仅调用一个特定的函数，可以将 tool_choice: {"type": "function", "function": {"name": "my_function"}}}。
3. 要禁用函数调用并强制模型仅生成面向用户的消息，可以将 tool_choice: "none"。

这个schema定义了一个名为route_user_query的函数，包含四个关键参数：

• target_agent：目标智能体名称（枚举类型，确保选项的规范性）
• confidence：路由决策的置信度（数值类型，范围0-1）
• user_intent：用户意图总结（字符串类型）
• original_query：用户原始查询内容（字符串类型）

2. 系统提示词设计

为了引导模型正确使用函数，本例设计了详细的系统提示词：

self.prompt = """# 角色
你是一个电商业务智能路由中心。你的核心职责是精准理解用户的提问意图，并将其分类到指定的业务模块中。...# 决策流程
在做出路由判断时，请遵循以下思考链：
1. **分析意图**：用户的核心诉求是什么？他最想获得哪方面的信息或服务？
2. **匹配分类**：这个诉求最符合哪个智能体的专业领域？
3. **设置置信度**：你有多确定这个路由选择是正确的？根据问题与分类的匹配程度，给出一个0到1之间的置信度分数。
4. **总结意图**：用一句话简洁总结用户意图，便于下游智能体快速理解。
5. **调用函数**：你必须调用route_user_query函数来输出路由结果。
..."""

这个提示词明确了模型的角色、职责和决策流程，特别是强调了必须调用指定的函数并返回结构化数据。

3. Function Calling API 调用实现

接下来，在 RouterAgent 的 run 方法中，我们实现具体的API调用逻辑：

async def run(self, messages: list[Dict[str, str]]) -> Dict[str, Any]:# 获取OpenAI客户端openai_client = client.async_openai_client# 调用OpenAI API with function callingmessages_history = [{"role": "system", "content": self.prompt}]messages_history.extend(messages)response = await openai_client.chat.completions.create(model="deepseek-chat",  # 支持function calling的模型messages=messages_history,tool_choice="required",tools=[{"type": "function", "function": self.function_schema}]  # 强制调用指定函数)# 解析function call结果if response.choices[0].message.tool_calls:router_list = []for function_info in response.choices[0].message.tool_calls:function_args = json.loads(function_info.function.arguments)router_list.append(function_args)return {"success": True,"data": router_list}else:return {"success": False,"error": "模型未返回function call结果"}

这段代码实现了以下关键步骤：

1. 构建包含系统提示词和用户消息的对话历史
2. 调用支持function calling的模型API
3. 设置tool_choice="required"强制要求模型调用函数
4. 提供函数schema定义
5. 解析API返回的函数调用结果
6. 返回结构化的处理结果

4. OpenAI客户端配置

系统使用异步客户端连接模型服务：

class Client:def __init__(self):settings = Settings()self.api_key = settings.OPENAI_API_KEYself.base_url = settings.OPENAI_BASE_URLself.model_name = settings.MODEL_NAME# ...@propertydef async_openai_client(self):if self._async_openai_client is None:self._async_openai_client = AsyncOpenAI(api_key=self.api_key,base_url=self.base_url)return self._async_openai_clientclient = Client()

在配置中，我们使用DeepSeek作为模型提供商：

OPENAI_BASE_URL: Optional[str] = Field(default="https://api.deepseek.com/v1", description="OpenAI API基础URL")
MODEL_NAME: str = Field(default="deepseek-chat", min_length=1, description="AI模型名称")

三、在客服系统中的集成应用

1. 路由节点实现

在客服系统的工作流中，RouterAgent被集成到路由节点：

# 路由节点（增强版，包含记忆上下文）
async def router_node(state: CustomerState):"""路由节点：分析用户意图并决定使用哪个agent"""messages = state["messages"]memory_context = state.get("memory_context", "")# 如果有记忆上下文，添加到消息中router = RouterAgent()if memory_context:router.prompt += "用户的记忆信息：" + "\n" + memory_contextresult = await router.run(messages)if result["success"]:router_data = result["data"][0]  # 取第一个路由结果return {"router_result": router_data,"current_agent": router_data["target_agent"]}else:# 路由失败，默认使用general_agentreturn {"router_result": {"target_agent": "general_agent","confidence": 0.5,"user_intent": "通用咨询","original_query": str(messages[-1].content if messages else "")},"current_agent": "general_agent"}

这个路由节点展示了几个重要特性：

• 上下文增强：集成用户记忆信息，提高意图识别准确性
• 容错处理：当路由失败时提供默认行为
• 状态管理：将路由结果保存到系统状态中

2. 在客服工作流中的完整调用链

整个客服查询处理流程如下：

async def process_customer_query(user_message: str, user_id: Optional[str] = None,session_id: Optional[str] = None) -> dict:# 创建工作流app = create_customer_service_workflow()# 初始化状态initial_state = {"messages": [HumanMessage(content=user_message)],"router_result": None,"current_agent": None,"final_response": None,"user_id": user_id or f"user_{uuid.uuid4().hex[:8]}","session_id": session_id or f"session_{uuid.uuid4().hex[:8]}","memory_context": "","historical_memories": []}# 执行工作流result = await app.ainvoke(initial_state)return {"success": True,"user_message": user_message,"user_id": initial_state["user_id"],"session_id": initial_state["session_id"],"router_result": result.get("router_result"),"current_agent": result.get("current_agent"),"final_response": result.get("final_response"),"memory_context": result.get("memory_context"),"messages": result.get("messages", [])}

四、案例流程解析

用户查询处理示例

场景：用户询问订单发货状态

当用户发送"我的订单什么时候发货？"时，系统的处理流程如下：

1. 请求接收：process_customer_query接收用户消息
2. 记忆加载：系统加载用户的历史记忆（如有）
3. 路由处理：RouterAgent通过Function Calling分析意图
4. 函数调用结果：

   {"name": "route_user_query","arguments": "{\"target_agent\": \"order_agent\", \"confidence\": 0.95, \"user_intent\": \"查询订单发货状态\", \"original_query\": \"我的订单什么时候发货？\"}"
   }
   

5. 智能体调用：系统将请求路由到order_agent
6. 业务处理：订单智能体处理发货状态查询
7. 响应返回：向用户返回订单发货信息