第03章 工具是Agent的手:Function Calling初探
第03章 工具是Agent的手:Function Calling初探
“工具调用不是一个功能,是LLM和真实世界之间的桥梁。没有这座桥,LLM永远只是一个很会说话的人,而不是一个会干活的同事。” —— Agent开发者
Agent能干活,靠的是工具。工具能被调用,靠的是Function Calling。
这一章,我们从零理解Function Calling的工作原理,并构建真正可用的工具。
3.1 Function Calling的工作原理
# Function Calling 不是魔法,是协议
# 你给LLM的东西:
REQUEST_STRUCTURE = {
"messages": [
{"role": "system", "content": "你是一个DNS管理助手"},
{"role": "user", "content": "帮我添加一个子域名 blog 指向 1.2.3.4"},
],
"tools": [
{
"type": "function",
"function": {
"name": "add_dns_record",
"description": "添加DNS A记录,将子域名指向指定IP地址",
"parameters": {
"type": "object",
"properties": {
"subdomain": {
"type": "string",
"description": "子域名,如 'blog'(不含主域名部分)"
},
"ip_address": {
"type": "string",
"description": "目标IPv4地址,如 '1.2.3.4'"
},
"proxied": {
"type": "boolean",
"description": "是否通过Cloudflare代理(默认false)",
"default": False,
}
},
"required": ["subdomain", "ip_address"]
}
}
}
]
}
# LLM返回的东西:
LLM_RESPONSE = {
"role": "assistant",
"content": None, # 有tool_calls时content通常为空
"tool_calls": [
{
"id": "call_abc123",
"type": "function",
"function": {
"name": "add_dns_record",
"arguments": '{"subdomain": "blog", "ip_address": "1.2.3.4", "proxied": false}'
}
}
]
}
# 关键理解:LLM没有"调用"任何工具
# 它只是产生了一段JSON,描述"我想调用什么工具,用什么参数"
# 真正的执行,在你的代码里
整个流程:
你的代码 → [messages + tools] → LLM API
↓
LLM决定是否需要工具
↓
tool_calls JSON(LLM输出)
↓
你的代码解析JSON → 调用真实函数
↓
函数结果 → 作为新消息发回LLM API
↓
LLM生成最终回复
3.2 构建一个真实的工具
# 好的工具设计:清晰、防御、可测试
import requests
import os
from typing import Optional
class CloudflareDNSTool:
"""
Cloudflare DNS管理工具
设计原则:
1. 每个工具做一件事
2. 返回结构化的结果(不是字符串)
3. 错误信息要对LLM友好
4. 幂等性:重复执行不产生副作用
"""
def __init__(self, api_token: str, zone_name: str):
self.api_token = api_token
self.zone_name = zone_name
self.base_url = "https://api.cloudflare.com/client/v4"
self._zone_id = None
def get_headers(self) -> dict:
return {
"Authorization": f"Bearer {self.api_token}",
"Content-Type": "application/json",
}
def get_zone_id(self) -> dict:
"""
工具:获取域名的Zone ID
这是其他DNS操作的前提
"""
if self._zone_id:
return {"success": True, "zone_id": self._zone_id}
url = f"{self.base_url}/zones"
params = {"name": self.zone_name}
try:
resp = requests.get(url, headers=self.get_headers(), params=params, timeout=10)
data = resp.json()
if data.get("success") and data.get("result"):
self._zone_id = data["result"][0]["id"]
return {"success": True, "zone_id": self._zone_id}
else:
return {
"success": False,
"error": "域名不存在或Token权限不足",
"details": data.get("errors", []),
}
except requests.exceptions.Timeout:
return {"success": False, "error": "请求超时,请稍后重试"}
except Exception as e:
return {"success": False, "error": str(e)}
def list_dns_records(self) -> dict:
"""
工具:列出所有DNS记录
返回:记录列表或错误信息
"""
zone_result = self.get_zone_id()
if not zone_result["success"]:
return zone_result
url = f"{self.base_url}/zones/{zone_result['zone_id']}/dns_records"
try:
resp = requests.get(url, headers=self.get_headers(), timeout=10)
data = resp.json()
if data.get("success"):
records = [
{
"id": r["id"],
"name": r["name"],
"type": r["type"],
"content": r["content"],
"proxied": r.get("proxied", False),
}
for r in data.get("result", [])
]
return {"success": True, "records": records, "count": len(records)}
else:
return {"success": False, "error": data.get("errors")}
except Exception as e:
return {"success": False, "error": str(e)}
def add_dns_record(
self,
subdomain: str,
ip_address: str,
proxied: bool = False,
) -> dict:
"""
工具:添加A记录(子域名 → IP)
幂等:如果记录已存在,返回现有记录
"""
zone_result = self.get_zone_id()
if not zone_result["success"]:
return zone_result
zone_id = zone_result["zone_id"]
record_name = f"{subdomain}.{self.zone_name}"
# 先检查是否已存在
existing = self._find_record(zone_id, record_name, "A")
if existing:
return {
"success": True,
"action": "already_exists",
"record_id": existing["id"],
"message": f"记录 {record_name} 已存在,指向 {existing['content']}",
}
# 创建新记录
url = f"{self.base_url}/zones/{zone_id}/dns_records"
payload = {
"type": "A",
"name": record_name,
"content": ip_address,
"proxied": proxied,
}
try:
resp = requests.post(url, headers=self.get_headers(), json=payload, timeout=10)
data = resp.json()
if data.get("success"):
return {
"success": True,
"action": "created",
"record_id": data["result"]["id"],
"name": record_name,
"ip": ip_address,
}
else:
return {
"success": False,
"error": "创建失败",
"details": data.get("errors"),
}
except Exception as e:
return {"success": False, "error": str(e)}
def _find_record(self, zone_id: str, name: str, record_type: str) -> Optional[dict]:
"""内部方法:查找特定记录"""
url = f"{self.base_url}/zones/{zone_id}/dns_records"
params = {"name": name, "type": record_type}
resp = requests.get(url, headers=self.get_headers(), params=params, timeout=10)
data = resp.json()
if data.get("success") and data.get("result"):
return data["result"][0]
return None
3.3 把工具注册给LLM
# 工具描述 = LLM能否正确调用工具的关键
def get_dns_tools_schema() -> list[dict]:
"""
为LLM生成工具的JSON Schema
描述质量直接影响LLM的调用准确率
"""
return [
{
"type": "function",
"function": {
"name": "list_dns_records",
"description": "列出当前域名的所有DNS记录。在添加新记录之前,建议先调用此工具了解现有配置。",
"parameters": {
"type": "object",
"properties": {}, # 无参数
"required": [],
}
}
},
{
"type": "function",
"function": {
"name": "add_dns_record",
"description": "添加DNS A记录,将子域名指向指定IP地址。如果记录已存在,会返回现有记录而不是报错。",
"parameters": {
"type": "object",
"properties": {
"subdomain": {
"type": "string",
"description": "子域名前缀,例如 'blog'、'api'、'www'(不含主域名)"
},
"ip_address": {
"type": "string",
"description": "目标IPv4地址,格式为 x.x.x.x",
"pattern": r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$"
},
"proxied": {
"type": "boolean",
"description": "是否启用Cloudflare代理(CDN/防火墙)。默认false。",
}
},
"required": ["subdomain", "ip_address"]
}
}
},
]
# 工具描述的最佳实践
TOOL_DESCRIPTION_BEST_PRACTICES = {
"name": "清晰的动词+名词(list_records 而不是 get)",
"description": [
"描述工具做什么(动词开头)",
"说明何时使用(在...之前/之后调用)",
"提示副作用('如果已存在则...')",
],
"parameter_description": [
"举例说明格式(如 'blog', 'api')",
"说明约束(不含主域名)",
"提供默认值说明",
],
"避免": [
"模糊描述('管理DNS')",
"技术内部术语",
"没有示例的格式要求",
],
}
3.4 完整的Function Calling执行循环
# 把工具、描述和LLM串起来
from openai import AsyncOpenAI
import json
async def dns_agent(user_request: str) -> str:
"""
一个完整的DNS管理Agent
"""
client = AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"])
dns = CloudflareDNSTool(
api_token=os.environ["CF_API_TOKEN"],
zone_name="example.com",
)
# 工具函数映射
tool_registry = {
"list_dns_records": dns.list_dns_records,
"add_dns_record": dns.add_dns_record,
}
messages = [
{"role": "system", "content": "你是一个DNS管理助手。使用工具完成用户的DNS操作,完成后简洁地报告结果。"},
{"role": "user", "content": user_request},
]
tools = get_dns_tools_schema()
# Agent循环
while True:
response = await client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=tools,
tool_choice="auto",
)
message = response.choices[0].message
messages.append(message.model_dump())
# 没有工具调用 = 任务完成
if not message.tool_calls:
return message.content
# 执行每个工具调用
for tool_call in message.tool_calls:
fn_name = tool_call.function.name
fn_args = json.loads(tool_call.function.arguments)
if fn_name in tool_registry:
result = tool_registry[fn_name](**fn_args)
else:
result = {"error": f"未知工具: {fn_name}"}
# 把结果加入对话
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result, ensure_ascii=False),
})
# 使用
# import asyncio
# result = asyncio.run(dns_agent("帮我查一下现在有哪些子域名"))
# print(result)
本章小结
五个核心认知:
-
Function Calling是协议,不是魔法:LLM生成JSON,你的代码执行——两件事分开;这意味着你对工具有完全控制权,也意味着你需要处理所有可能的错误
-
工具描述质量 = Agent准确率:LLM是否能正确调用工具,80%取决于工具描述是否清晰;模糊的描述导致错误参数,缺少示例导致格式错误
-
工具需要防御性设计:幂等性(重复执行安全)、错误信息LLM友好、超时处理——这些不是锦上添花,是生产环境的必须
-
把工具结果设计为结构化数据:返回JSON而不是字符串;LLM需要解析结果来决定下一步,结构化数据比自由文本更可靠
-
从单工具开始,逐步增加:先让一个工具可靠工作,再添加下一个;大量工具反而会让LLM的选择变得混乱
核心行动:
# 今天的任务:
# 1. 选一个你日常用的API(GitHub/Notion/Slack/任何都行)
# 2. 写一个Python函数封装它(按照本章的设计原则)
# 3. 写对应的JSON Schema(工具描述)
# 4. 用OpenAI API测试:让LLM调用它完成一个小任务
本章提示词模板
模板一:设计工具的JSON Schema
我有一个Python函数如下:
```python
[粘贴你的函数代码]
请帮我为这个函数生成OpenAI Function Calling格式的JSON Schema,要求:
- name:清晰的动词+名词命名
- description:说明用途、适用场景、副作用
- 每个参数的description:说明格式、约束、举例
- 标记required参数(必须提供的)
然后指出:
- 这个函数是否适合给Agent使用(幂等性?错误处理?)
- 描述中是否有可能导致LLM误用的地方
**模板二:调试工具调用问题**
我的Agent工具调用出现了问题:
工具定义:[粘贴工具的JSON Schema] 用户请求:[用户说了什么] LLM调用了:[LLM实际调用的工具和参数] 预期调用:[你希望LLM调用的工具和参数]
问题描述:[描述发生了什么错误]
请帮我分析:
- 是工具描述的问题还是参数定义的问题?
- LLM为什么做出了这个错误判断?
- 如何修改工具描述来避免这个问题?
- 是否需要在System Prompt中补充说明?
---
[→ 第04章:记忆是Agent的脑:上下文、技能与知识库](第04章_记忆是Agent的脑:上下文、技能与知识库.md)