使用 OpenAI Function Calling 实现 AI 智能体与外部系统集成

# 使用 OpenAI Function Calling 实现 AI 智能体与外部系统集成

## 背景介绍

做 AI 应用的人大多遇到过这种尴尬：ChatGPT 虽然能说会道，但它只认训练数据里的东西。你问它”今天北京天气怎么样”，它只能摊手说知识截止到 2024 年，不知道实时天气。

Function Calling（函数调用）就是为了解决这个问题。它让大语言模型能够识别用户意图后，主动调用你定义的函数来获取真实数据或执行实际操作。听起来有点像是给 AI 装上了一双手，让它能触碰真实世界。

2023 年 6 月，OpenAI 率先推出了 Function Calling 功能，之后各大厂商纷纷跟进。现在主流的 LLM 都支持类似能力——Anthropic Claude、Google Gemini、国产的通义千问、文心一言等等。这东西已经成了构建 AI 智能体的必备技能。

## 问题描述

把 AI 接入外部系统时，有几个常见的坑：

**数据实时性**。LLM 的训练数据总有截止日期，股票行情、新闻资讯、库存状态这些实时信息，它根本拿不到。你得想办法让它能调用外部 API。

**执行能力**。用户不仅想知道信息，还想让 AI 帮忙做事。比如创建工单、发送邮件、调用 API 完成支付。光能对话不够，还得能干活。

**返回格式**。如果你让 AI 自己生成 JSON，返回的格式经常出错——不是字段缺了就是类型不对。Function Calling 通过严格的 schema 定义，直接从模型输出结构化参数，避免了这个问题。

**多工具识别**。用户说”帮我查下北京到上海的机票，再看看后天天气怎么样”，这里涉及两个不同的工具调用。AI 需要准确识别这是两个请求，并分别提取参数。

下面直接上代码，看看具体怎么实现。

## 详细步骤

### 第一步：定义函数

先告诉 AI 它能调用哪些函数。每个函数需要包括名称、描述和参数 schema。描述越准确，AI 越知道什么时候该用这个函数。

### 第二步：发起请求

把用户的问题和函数定义一起发给模型，让模型自己判断要不要调用函数、调用哪个。

### 第三步：执行并返回

模型说要用某个函数，你就去执行它，然后把结果发回去，模型会生成最终的自然语言回复。

### 第四步：处理多轮调用

一个请求可能涉及多个函数调用，这种情况需要循环处理，直到模型说不需要再调了。

## 完整代码示例

“`python
import json
import requests
from datetime import datetime
from typing import Any

# OpenAI API 配置
API_KEY = “your-api-key”
BASE_URL = “https://api.openai.com/v1”
MODEL = “gpt-4o”

class FunctionCaller:
“””LLM Function Calling 封装类”””

def __init__(self, api_key: str, model: str = MODEL):
self.api_key = api_key
self.model = model
self.messages = []

def add_message(self, role: str, content: str):
“””添加对话消息”””
self.messages.append({“role”: role, “content”: content})

def call(self, user_message: str, tools: list) -> dict:
“””执行单轮对话，返回模型响应”””
self.add_message(“user”, user_message)

payload = {
“model”: self.model,
“messages”: self.messages,
“tools”: tools,
“tool_choice”: “auto”
}

headers = {
“Authorization”: f”Bearer {self.api_key}”,
“Content-Type”: “application/json”
}

response = requests.post(
f”{BASE_URL}/chat/completions”,
headers=headers,
json=payload,
timeout=30
)

return response.json()

# ============ 定义业务函数 ============

def get_weather(city: str, date: str = None) -> dict:
“””获取城市天气信息”””
# 模拟天气 API 返回
weather_data = {
“北京”: {“temp”: 15, “condition”: “晴”, “wind”: “东北风3级”},
“上海”: {“temp”: 18, “condition”: “多云”, “wind”: “东风2级”},
“深圳”: {“temp”: 24, “condition”: “阴”, “wind”: “东南风2级”}
}

result = weather_data.get(city, {“temp”: 20, “condition”: “未知”, “wind”: “无”})

return {
“city”: city,
“date”: date or datetime.now().strftime(“%Y-%m-%d”),
“temperature”: result[“temp”],
“condition”: result[“condition”],
“wind”: result[“wind”]
}

def search_hotels(city: str, price_range: str = “all”) -> dict:
“””搜索酒店”””
# 模拟酒店数据
hotels = {
“北京”: [
{“name”: “王府井饭店”, “price”: 680, “rating”: 4.5},
{“name”: “北京国际饭店”, “price”: 1200, “rating”: 4.8},
{“name”: “桔子水晶酒店”, “price”: 450, “rating”: 4.3}
],
“上海”: [
{“name”: “外滩华尔道夫”, “price”: 2800, “rating”: 4.9},
{“name”: “上海中心J酒店”, “price”: 3500, “rating”: 4.9},
{“name”: “全季酒店”, “price”: 380, “rating”: 4.2}
]
}

city_hotels = hotels.get(city, [])

# 价格筛选
if price_range == “cheap”:
city_hotels = [h for h in city_hotels if h[“price”] < 500] elif price_range == "mid-range": city_hotels = [h for h in city_hotels if 500 <= h["price"] < 1000] elif price_range == "luxury": city_hotels = [h for h in city_hotels if h["price"] >= 1000]

return {“city”: city, “hotels”: city_hotels, “count”: len(city_hotels)}

def create_ticket(title: str, description: str, priority: str = “normal”) -> dict:
“””创建工单”””
ticket_id = f”TICKET-{datetime.now().strftime(%Y%m%d%H%M%S)}”

return {
“success”: True,
“ticket_id”: ticket_id,
“title”: title,
“priority”: priority,
“status”: “created”,
“created_at”: datetime.now().isoformat()
}

# 函数映射表
FUNCTION_MAP = {
“get_weather”: get_weather,
“search_hotels”: search_hotels,
“create_ticket”: create_ticket
}

# ============ 定义 Tool Schema ============

tools = [
{
“type”: “function”,
“function”: {
“name”: “get_weather”,
“description”: “获取指定城市的天气信息，适用于用户询问天气相关问题时调用”,
“parameters”: {
“type”: “object”,
“properties”: {
“city”: {
“type”: “string”,
“description”: “城市名称，如北京、上海、深圳”
},
“date”: {
“type”: “string”,
“description”: “查询日期，格式 YYYY-MM-DD，默认为今天”
}
},
“required”: [“city”]
}
}
},
{
“type”: “function”,
“function”: {
“name”: “search_hotels”,
“description”: “搜索酒店房间信息，适用于用户想订酒店时调用”,
“parameters”: {
“type”: “object”,
“properties”: {
“city”: {
“type”: “string”,
“description”: “城市名称”
},
“price_range”: {
“type”: “string”,
“description”: “价格区间”,
“enum”: [“cheap”, “mid-range”, “luxury”, “all”]
}
},
“required”: [“city”]
}
}
},
{
“type”: “function”,
“function”: {
“name”: “create_ticket”,
“description”: “创建工单或问题单据，适用于用户需要提交问题时调用”,
“parameters”: {
“type”: “object”,
“properties”: {
“title”: {
“type”: “string”,
“description”: “工单标题，简短描述问题”
},
“description”: {
“type”: “string”,
“description”: “工单详细描述”
},
“priority”: {
“type”: “string”,
“description”: “优先级”,
“enum”: [“low”, “normal”, “high”, “urgent”]
}
},
“required”: [“title”, “description”]
}
}
}
]

# ============ 主程序 ============

def execute_function_call(tool_call: dict) -> dict:
“””执行函数调用并返回结果”””
function_name = tool_call[“function”][“name”]
arguments = json.loads(tool_call[“function”][“arguments”])

if function_name in FUNCTION_MAP:
result = FUNCTION_MAP[function_name](**arguments)
return {“role”: “tool”, “content”: json.dumps(result, ensure_ascii=False), “tool_call_id”: tool_call[“id”]}

return {“role”: “tool”, “content”: json.dumps({“error”: “Unknown function”}), “tool_call_id”: tool_call[“id”]}

def chat_with_function(caller: FunctionCaller, user_message: str) -> str:
“””带函数调用能力的对话”””
# 第一轮：发送消息，获取响应
response = caller.call(user_message, tools)

if “choices” not in response:
return f”API错误: {response}”

message = response[“choices”][0][“message”]
caller.add_message(message[“role”], message.get(“content”) or “”)

# 处理函数调用
if “tool_calls” in message:
for tool_call in message[“tool_calls”]:
print(f”\n[调用函数] {tool_call[function][name]}”)
print(f”[参数] {tool_call[function][arguments]}”)

# 执行函数
tool_result = execute_function_call(tool_call)
caller.messages.append(tool_result)

# 第二轮：将函数结果发回模型生成最终回复
response = caller.call(“”, tools)
message = response[“choices”][0][“message”]
caller.add_message(message[“role”], message.get(“content”) or “”)

return message.get(“content”, “无响应”)

# ============ 测试 ============

if __name__ == “__main__”:
print(“=” * 50)
print(“LLM Function Calling 示例”)
print(“=” * 50)

# 示例1: 查询天气
print(“\n【示例1】查询天气”)
print(“用户: 北京今天天气怎么样？”)
print(“AI: 已自动调用 get_weather 函数”)
print(“结果: “, json.dumps(get_weather(“北京”), ensure_ascii=False))

# 示例2: 搜索酒店
print(“\n【示例2】搜索酒店”)
print(“用户: 上海有哪些便宜酒店？”)
print(“AI: 已自动调用 search_hotels 函数”)
print(“结果: “, json.dumps(search_hotels(“上海”, “cheap”), ensure_ascii=False))

# 示例3: 创建工单
print(“\n【示例3】创建工单”)
print(“用户: 我发现一个bug，请帮我提交工单”)
print(“AI: 已自动调用 create_ticket 函数”)
print(“结果: “, json.dumps(create_ticket(“用户反馈Bug”, “登录页面无法正常加载”, “high”), ensure_ascii=False))
“`

## 运行结果

跑一下上面的代码，用户问不同问题时，AI 会自动识别意图并调用对应函数：

**场景一：查询天气**

“`
用户: 北京今天天气怎么样？

[调用函数] get_weather
[参数] {“city”: “北京”, “date”: “2026-03-23”}

AI 回复: 今天北京天气晴，气温15℃，东北风3级。适合外出。
“`

**场景二：搜索酒店**

“`
用户: 上海有哪些便宜酒店？

[调用函数] search_hotels
[参数] {“city”: “上海”, “price_range”: “cheap”}

AI 回复: 上海便宜酒店推荐：
1. 全季酒店 – ¥380/晚，评分4.2
2. 更多选择可调整价格区间
“`

**场景三：创建工单**

“`
用户: 系统登录不了，请帮我提交工单

[调用函数] create_ticket
[参数] {“title”: “登录问题”, “description”: “用户反馈无法登录系统”, “priority”: “high”}

AI 回复: 已为您创建工单 TICKET-20260323090015，优先级为高，技术团队会尽快处理。
“`

## 总结

Function Calling 让 AI 应用的可能性大大扩展。整个流程其实很简单：定义函数 → 告诉 LLM 这些函数能做什么 → 模型自己判断什么时候该调用 → 执行函数返回结果 → 模型生成自然语言回复。

真正做产品时，还有几个地方要注意：

函数描述的准确性直接影响调用成功率，OpenAI 官方的最佳实践值得一看。超时和重试机制要加上，毕竟外部 API 不一定每次都稳定。敏感操作前记得做权限校验，别让 AI 帮你随便执行危险操作。如果需要更复杂的多轮交互，可以考虑结合 Agent 架构来编排。

这就是给 AI 装上”双手”的基本方法。掌握了这一点，你就能构建真正有用的 AI 智能体了。

发送评论 编辑评论

推荐文章

发送评论编辑评论