Function Calling 实战:让 LLM 调用外部工具

什么是 Function Calling

Function Calling(函数调用)是 LLM 与外部系统交互的能力。它让 AI 不仅能生成文本,还能调用 API、执行代码、操作数据库,真正成为一个能”做事”的智能代理。

简单来说:Function Calling = LLM 决定调用哪个工具 + 提取参数 + 执行 + 反馈结果

为什么需要 Function Calling

单纯 LLM 的局限

问题 说明
知识时效性 训练数据有截止日期,无法获取实时信息
计算能力弱 无法精确计算复杂数学问题
无外部交互 不能查数据库、发邮件、操作文件
幻觉风险 可能生成错误数据

Function Calling 能做什么

1
2
3
4
5
6
7
8
9
用户:今天北京的天气如何?

传统 LLM:无法回答实时天气

带 Function Calling

LLM 分析问题 → 调用 get_weather("北京")

返回:今天北京晴,气温 15-25°C

工作原理

调用流程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
用户提问


┌─────────────┐
│ LLM │ ← 判断需要调用什么工具
└──────┬──────┘

▼ 返回工具调用请求
┌─────────────┐
│ 解析函数名 │
│ 和参数 │
└──────┬──────┘


┌─────────────┐
│ 执行函数 │
└──────┬──────┘


┌─────────────┐
│ 返回结果 │
│ 给 LLM │
└──────┬──────┘


┌─────────────┐
│ LLM │ ← 结合结果生成最终回答
└──────┬──────┘


最终回答

工具定义格式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
# OpenAI 风格的工具定义
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的天气信息",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称,如:北京、上海"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "温度单位,默认为 celsius"
}
},
"required": ["city"]
}
}
}
]

实战:基础用法

1. 天气查询

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
import openai
import json

# 定义工具
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取城市天气",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称"
}
},
"required": ["city"]
}
}
}
]

# 模拟天气查询函数
def get_weather(city):
weather_db = {
"北京": {"weather": "晴", "temp": "25°C"},
"上海": {"weather": "多云", "temp": "23°C"},
"广州": {"weather": "雨", "temp": "20°C"},
}
return weather_db.get(city, {"weather": "未知", "temp": "未知"})

# 对话循环
messages = [{"role": "user", "content": "北京今天天气怎么样?"}]

response = openai.ChatCompletion.create(
model="gpt-4",
messages=messages,
tools=tools,
tool_choice="auto"
)

# LLM 返回的函数调用
tool_calls = response.choices[0].message.tool_calls

if tool_calls:
for call in tool_calls:
func_name = call.function.name
args = json.loads(call.function.arguments)

# 执行函数
result = get_weather(**args)

# 将结果返回给 LLM
messages.append({
"role": "assistant",
"content": None,
"tool_calls": tool_calls
})
messages.append({
"role": "tool",
"tool_call_id": call.id,
"content": json.dumps(result)
})

# 再次调用获取最终回答
final_response = openai.ChatCompletion.create(
model="gpt-4",
messages=messages,
tools=tools
)
print(final_response.choices[0].message.content)

2. 发送邮件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def send_email(to, subject, body):
"""模拟发送邮件"""
print(f"📧 发送邮件:")
print(f" To: {to}")
print(f" Subject: {subject}")
print(f" Body: {body}")
return {"status": "sent", "message_id": "123456"}

tools = [
{
"type": "function",
"function": {
"name": "send_email",
"description": "发送邮件",
"parameters": {
"type": "object",
"properties": {
"to": {
"type": "string",
"description": "收件人邮箱"
},
"subject": {
"type": "string",
"description": "邮件主题"
},
"body": {
"type": "string",
"description": "邮件正文"
}
},
"required": ["to", "subject", "body"]
}
}
}
]

# 用户请求
user_msg = "给 john@example.com 发一封邮件,告诉他会议改到下午3点"

3. 实时搜索

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
def web_search(query, max_results=5):
"""模拟网络搜索"""
# 实际使用可接入 Google/Bing API
results = [
{"title": f"关于 {query} 的结果1", "url": "https://example.com/1"},
{"title": f"关于 {query} 的结果2", "url": "https://example.com/2"},
]
return {"results": results[:max_results]}

tools = [
{
"type": "function",
"function": {
"name": "web_search",
"description": "搜索互联网获取最新信息",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词"
},
"max_results": {
"type": "integer",
"description": "返回结果数量,默认5条"
}
},
"required": ["query"]
}
}
}
]

实战:代码执行

Python 代码执行

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def execute_python(code):
"""安全地执行 Python 代码"""
import subprocess
import sys

# 基础安全检查
dangerous_patterns = [
"import os",
"import subprocess",
"import sys",
"open(",
"eval(",
"exec(",
"__import__",
]

for pattern in dangerous_patterns:
if pattern in code:
return {"error": f"禁止使用: {pattern}"}

# 执行代码
try:
result = subprocess.run(
[sys.executable, "-c", code],
capture_output=True,
text=True,
timeout=10
)
return {
"stdout": result.stdout,
"stderr": result.stderr,
"returncode": result.returncode
}
except Exception as e:
return {"error": str(e)}

tools = [
{
"type": "function",
"function": {
"name": "execute_python",
"description": "执行 Python 代码并返回输出",
"parameters": {
"type": "object",
"properties": {
"code": {
"type": "string",
"description": "要执行的 Python 代码"
}
},
"required": ["code"]
}
}
}
]

# 使用示例
user_msg = "帮我计算 1 到 100 的和"
# LLM 会调用 execute_python 计算

SQL 查询

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def execute_sql(query):
"""执行 SQL 查询"""
import sqlite3

# 创建内存数据库演示
conn = sqlite3.connect(':memory:')
cursor = conn.cursor()

# 实际使用时连接真实数据库
try:
cursor.execute(query)
results = cursor.fetchall()
columns = [desc[0] for desc in cursor.description] if cursor.description else []
return {"columns": columns, "rows": results}
except Exception as e:
return {"error": str(e)}
finally:
conn.close()

tools = [
{
"type": "function",
"function": {
"name": "execute_sql",
"description": "执行 SQL 查询,返回查询结果",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "SQL 查询语句"
}
},
"required": ["query"]
}
}
}
]

实战:多工具协作

Agent 模式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
import json
from typing import List, Callable

class ToolAgent:
def __init__(self, tools: List[dict], model: str = "gpt-4"):
self.tools = tools
self.model = model
self.tools_map = {}

def register(self, name: str, func: Callable):
"""注册工具函数"""
self.tools_map[name] = func

def run(self, user_message: str, max_iterations: int = 10):
"""运行 Agent"""
messages = [{"role": "user", "content": user_message}]

for i in range(max_iterations):
# 调用 LLM
response = openai.ChatCompletion.create(
model=self.model,
messages=messages,
tools=self.tools,
tool_choice="auto"
)

message = response.choices[0].message

# 检查是否需要调用工具
if not message.tool_calls:
# 没有工具调用,返回结果
return message.content

# 处理每个工具调用
for call in message.tool_calls:
func_name = call.function.name
args = json.loads(call.function.arguments)

# 执行工具
if func_name in self.tools_map:
result = self.tools_map[func_name](**args)
else:
result = {"error": f"Unknown tool: {func_name}"}

# 添加到消息历史
messages.append({
"role": "assistant",
"content": None,
"tool_calls": message.tool_calls
})
messages.append({
"role": "tool",
"tool_call_id": call.id,
"content": json.dumps(result)
})

return "达到最大迭代次数"

# 使用示例
agent = ToolAgent(tools=tools)
agent.register("get_weather", get_weather)
agent.register("send_email", send_email)
agent.register("web_search", web_search)
agent.register("execute_python", execute_python)

# 运行
result = agent.run("北京天气怎么样?适合穿什么衣服?")
print(result)

复杂任务处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# 复杂任务:查询天气、穿衣建议
user_task = """
我的出差计划如下:
- 5月20日 北京
- 5月21日 上海
- 5月22日 广州

请帮我:
1. 查询这三个城市的天气
2. 根据天气建议我带什么衣服
3. 最后总结一下行程建议
"""

# LLM 会自动分解任务并调用工具
# 可能需要多次工具调用

工具设计最佳实践

1. 清晰的描述

1
2
3
4
5
6
7
8
9
10
11
# ❌ 模糊的描述
{
"name": "search",
"description": "搜索"
}

# ✅ 清晰完整的描述
{
"name": "search_hotels",
"description": "搜索符合条件的酒店列表,返回酒店名称、评分、价格和地址"
}

2. 明确的参数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
# ❌ 缺乏约束
{
"properties": {
"query": {"type": "string"}
}
}

# ✅ 有约束和示例
{
"properties": {
"query": {
"type": "string",
"description": "搜索关键词,如:朝阳区五星级酒店",
"minLength": 2,
"maxLength": 50
},
"sort_by": {
"type": "string",
"enum": ["price", "rating", "distance"],
"default": "rating",
"description": "排序方式"
}
},
"required": ["query"]
}

3. 参数验证

1
2
3
4
5
6
7
8
9
10
11
def get_weather(city: str, unit: str = "celsius"):
# 参数验证
valid_cities = ["北京", "上海", "广州", "深圳"]
if city not in valid_cities:
return {"error": f"暂不支持城市:{city},支持:{valid_cities}"}

if unit not in ["celsius", "fahrenheit"]:
return {"error": "温度单位仅支持 celsius 或 fahrenheit"}

# 正常处理...
return weather_data

主流框架支持

LangChain

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
from langchain.tools import Tool
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain import hub

# 定义工具
tools = [
Tool(
name="Calculator",
func=calculate,
description="用于数学计算,支持加减乘除、指数等"
),
Tool(
name="Search",
func=search,
description="搜索最新新闻和资讯"
),
Tool(
name="Weather",
func=get_weather,
description="查询城市天气"
)
]

# 创建 Agent
prompt = hub.pull("hwchase17/openai-functions-agent")
agent = create_openai_functions_agent(llm, tools, prompt)

# 运行
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
result = agent_executor.invoke({"input": "北京今天适合户外运动吗?"})

LlamaIndex

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from llama_index.tools import FunctionTool
from llama_index.agent import OpenAIAgent

# 定义工具
weather_tool = FunctionTool.from_defaults(
fn=get_weather,
name="get_weather",
description="获取指定城市的天气"
)

code_tool = FunctionTool.from_defaults(
fn=execute_code,
name="execute_code",
description="执行 Python 代码"
)

# 创建 Agent
agent = OpenAIAgent.from_tools([weather_tool, code_tool])
response = agent.chat("帮我计算一下 2 的 10 次方")

安全考虑

1. 工具权限控制

1
2
3
4
5
6
7
8
9
10
11
12
13
14
class SecureToolAgent:
def __init__(self):
self.allowed_tools = set()

def grant_permission(self, tool_name: str):
self.allowed_tools.add(tool_name)

def revoke_permission(self, tool_name: str):
self.allowed_tools.discard(tool_name)

def execute_tool(self, tool_name: str, **kwargs):
if tool_name not in self.allowed_tools:
return {"error": f"没有权限调用: {tool_name}"}
# 执行工具...

2. 输入过滤

1
2
3
4
5
6
7
def sanitize_sql_input(user_input: str) -> str:
"""SQL 注入防护"""
dangerous = [";", "--", "DROP", "DELETE", "INSERT", "UPDATE", "UNION"]
for keyword in dangerous:
if keyword.upper() in user_input.upper():
raise ValueError(f"禁止输入: {keyword}")
return user_input

3. 审计日志

1
2
3
4
5
6
7
8
9
10
11
import logging

logging.basicConfig(filename='tool_calls.log', level=logging.INFO)

def log_tool_call(tool_name: str, args: dict, result: any):
logging.info({
"tool": tool_name,
"args": args,
"result": str(result)[:200], # 截断避免日志过大
"timestamp": datetime.now().isoformat()
})

应用场景

1. 智能助手

1
2
3
4
5
6
7
assistant_tools = [
web_search, # 搜索
get_weather, # 天气
set_reminder, # 提醒
send_message, # 发消息
create_calendar_event # 日历
]

2. 数据分析助手

1
2
3
4
5
6
analysis_tools = [
query_database, # SQL 查询
execute_python, # 数据处理
generate_chart, # 生成图表
export_report # 导出报告
]

3. 代码开发助手

1
2
3
4
5
6
7
coding_tools = [
read_file, # 读文件
write_file, # 写文件
execute_tests, # 运行测试
deploy_application, # 部署
git_commit # Git 提交
]

4. 客服机器人

1
2
3
4
5
6
7
customer_service_tools = [
query_order, # 查订单
query_product, # 查商品
process_refund, # 处理退款
create_ticket, # 创建工单
transfer_to_human # 转人工
]

常见问题

1. 工具选择错误

1
2
3
4
# LLM 可能选错工具,需要:
# 1. 工具描述要准确
# 2. 参数要有足够约束
# 3. 添加错误处理和重试

2. 参数提取错误

1
2
3
4
5
6
7
8
# LLM 可能提取错误参数
# 解决:参数验证 + 返回清晰错误信息
def tool_with_validation(**kwargs):
try:
validate_params(kwargs)
return execute(**kwargs)
except ValidationError as e:
return {"error": str(e)}

3. 循环调用

1
2
# 设置最大迭代次数,防止无限循环
MAX_ITERATIONS = 10

总结

Function Calling 让 LLM 从「知识输出」进化到「行动执行」:

  1. 定义清晰的工具:描述准确、参数明确、有约束
  2. 完善的错误处理:参数验证、异常捕获、友好提示
  3. 安全保障:权限控制、输入过滤、审计日志
  4. 迭代优化:根据实际使用持续改进工具设计

掌握 Function Calling,是构建 AI Agent 的基础,也是 LLM 真正落地的关键能力。


有问题欢迎留言讨论!