第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)

本章小结

五个核心认知:

  1. Function Calling是协议,不是魔法:LLM生成JSON,你的代码执行——两件事分开;这意味着你对工具有完全控制权,也意味着你需要处理所有可能的错误

  2. 工具描述质量 = Agent准确率:LLM是否能正确调用工具,80%取决于工具描述是否清晰;模糊的描述导致错误参数,缺少示例导致格式错误

  3. 工具需要防御性设计:幂等性(重复执行安全)、错误信息LLM友好、超时处理——这些不是锦上添花,是生产环境的必须

  4. 把工具结果设计为结构化数据:返回JSON而不是字符串;LLM需要解析结果来决定下一步,结构化数据比自由文本更可靠

  5. 从单工具开始,逐步增加:先让一个工具可靠工作,再添加下一个;大量工具反而会让LLM的选择变得混乱

核心行动

# 今天的任务:
# 1. 选一个你日常用的API(GitHub/Notion/Slack/任何都行)
# 2. 写一个Python函数封装它(按照本章的设计原则)
# 3. 写对应的JSON Schema(工具描述)
# 4. 用OpenAI API测试:让LLM调用它完成一个小任务

本章提示词模板

模板一:设计工具的JSON Schema

我有一个Python函数如下:

```python
[粘贴你的函数代码]

请帮我为这个函数生成OpenAI Function Calling格式的JSON Schema,要求:

  1. name:清晰的动词+名词命名
  2. description:说明用途、适用场景、副作用
  3. 每个参数的description:说明格式、约束、举例
  4. 标记required参数(必须提供的)

然后指出:

  • 这个函数是否适合给Agent使用(幂等性?错误处理?)
  • 描述中是否有可能导致LLM误用的地方

**模板二:调试工具调用问题**

我的Agent工具调用出现了问题:

工具定义:[粘贴工具的JSON Schema] 用户请求:[用户说了什么] LLM调用了:[LLM实际调用的工具和参数] 预期调用:[你希望LLM调用的工具和参数]

问题描述:[描述发生了什么错误]

请帮我分析:

  1. 是工具描述的问题还是参数定义的问题?
  2. LLM为什么做出了这个错误判断?
  3. 如何修改工具描述来避免这个问题?
  4. 是否需要在System Prompt中补充说明?

---

[→ 第04章:记忆是Agent的脑:上下文、技能与知识库](第04章_记忆是Agent的脑:上下文、技能与知识库.md)