工具调用机制：Function Calling实战

1

定义工具函数 → 描述工具给LLM → LLM选择并生成调用 → 宿主执行工具 → 结果回灌LLM

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

用户输入
   │
   ▼
[LLM] ──看到工具描述──> 决定调用 get_weather(location="上海")
   │
   ▼
宿主代码解析 tool_calls
   │
   ▼
执行 get_weather("上海") ──> {"temp": 32, "condition": "晴"}
   │
   ▼
把结果作为 tool 消息追加到 messages
   │
   ▼
[LLM] ──基于结果──> "上海今天32度，晴天，注意防晒"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

{
  "type": "function",
  "function": {
    "name": "get_weather",
    "description": "查询指定城市的实时天气。支持中国地级市及以上行政区。",
    "parameters": {
      "type": "object",
      "properties": {
        "location": {
          "type": "string",
          "description": "城市名称，例如：上海、北京、深圳"
        },
        "unit": {
          "type": "string",
          "enum": ["celsius", "fahrenheit"],
          "description": "温度单位，默认摄氏度"
        }
      },
      "required": ["location"]
    }
  }
}

1

pip install openai python-dotenv

1
2
3
4
5

# .env 文件（项目根目录，每行一个变量，格式：变量名=变量值）
# 写入一行：OPENAI_API_KEY 等号 你的密钥
#   - 变量名固定为 OPENAI_API_KEY（SDK据此自动读取）
#   - 密钥从 https://platform.openai.com/api-keys 获取，以 sk- 开头
#   - 等号两边不要加空格，整行不加引号

 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

# tools_impl.py
import json
from typing import Dict, Any

# 模拟天气数据库（真实项目里这里会改成调用第三方天气API）
WEATHER_DB = {
    "上海": {"temp": 32, "condition": "晴", "humidity": 65},
    "北京": {"temp": 28, "condition": "多云", "humidity": 50},
    "深圳": {"temp": 30, "condition": "雷阵雨", "humidity": 80},
}

def get_weather(location: str, unit: str = "celsius") -> Dict[str, Any]:
    """查询指定城市的实时天气"""
    # 去掉"市"字并去空格，让"上海市"和"上海"都能命中
    city = location.replace("市", "").strip()
    if city not in WEATHER_DB:
        return {"error": f"暂不支持查询 {location} 的天气"}
    data = WEATHER_DB[city].copy()
    # 如果用户要华氏度，做个换算
    if unit == "fahrenheit":
        data["temp"] = round(data["temp"] * 9 / 5 + 32, 1)
    data["unit"] = unit
    return data

def calculator(expression: str) -> Dict[str, Any]:
    """安全计算数学表达式"""
    # 白名单机制：只允许数字和基本运算符，防止代码注入
    allowed = "0123456789+-*/().% "
    if not all(c in allowed for c in expression):
        return {"error": "表达式包含非法字符"}
    try:
        # 用受限的eval执行：禁用所有内置函数，防止逃逸
        result = eval(expression, {"__builtins__": {}}, {})
        return {"expression": expression, "result": result}
    except Exception as e:
        return {"error": f"计算失败: {e}"}

 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

# agent.py
import os
import json
from dotenv import load_dotenv
from openai import OpenAI
from tools_impl import get_weather, calculator

load_dotenv()  # 从.env文件加载环境变量
# 从环境变量读取API密钥，绝不硬编码到代码里
# 用**把字典展开成关键字参数传给OpenAI（Python传参的等价写法）
_opts = {"api_key": os.getenv("OPENAI_API_KEY")}
client = OpenAI(**_opts)
MODEL = "gpt-5"

# 工具schema定义：这就是给LLM看的"菜单"
TOOLS_SCHEMA = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询指定城市的实时天气，包括温度、天气状况和湿度。当用户询问天气、温度、是否下雨、穿衣建议时调用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "城市名称，如：上海、北京、深圳"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度单位，默认摄氏度"
                    }
                },
                "required": ["location"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculator",
            "description": "计算数学表达式。当用户需要做加减乘除、百分比、混合运算时调用。不支持代数或微积分。",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "数学表达式，如：1234 * 5678 或 (100+200)/3"
                    }
                },
                "required": ["expression"]
            }
        }
    }
]

# 工具名 → 实现函数 的映射表：根据LLM点到的"菜名"找到对应的"厨师"
TOOL_REGISTRY = {
    "get_weather": get_weather,
    "calculator": calculator,
}

 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

def run_agent(user_query: str, max_turns: int = 5) -> str:
    # 初始化对话历史：system设定人设，user放用户问题
    messages = [
        {"role": "system", "content": "你是一个能查天气和做计算的助手。回答要简洁。"},
        {"role": "user", "content": user_query},
    ]

    # 最多循环max_turns轮，防止意外死循环
    for _ in range(max_turns):
        # 把当前所有消息和工具清单一起发给LLM
        response = client.chat.completions.create(
            model=MODEL,
            messages=messages,
            tools=TOOLS_SCHEMA,
        )
        msg = response.choices[0].message

        # 关键判断：如果LLM没有请求工具，说明它已经准备好直接回答用户
        if not msg.tool_calls:
            return msg.content

        # 把assistant这条消息（含tool_calls）原样加入历史
        # 注意：必须append原始msg对象，因为里面带着tool_calls引用
        messages.append(msg)

        # 依次执行LLM点名的每个工具调用
        for tool_call in msg.tool_calls:
            name = tool_call.function.name                   # 工具名
            args = json.loads(tool_call.function.arguments)  # 参数是JSON字符串，要解析
            print(f"[调用工具] {name}({args})")

            func = TOOL_REGISTRY.get(name)  # 从注册表找到对应Python函数
            if func is None:
                result = {"error": f"未知工具: {name}"}
            else:
                try:
                    result = func(**args)   # 用**解包把字典变成关键字参数
                except Exception as e:
                    result = {"error": str(e)}

            # 关键：把工具结果作为role="tool"的消息回灌
            # tool_call_id必须和上面的tool_call.id对应，LLM靠它把结果归位
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result, ensure_ascii=False),
            })

    return "达到最大轮数，处理中止。"

1
2

if __name__ == "__main__":
    print(run_agent("上海和北京今天多少度？哪个更热？顺便算一下 1234 × 5678"))

1
2
3
4

[调用工具] get_weather({'location': '上海'})
[调用工具] get_weather({'location': '北京'})
[调用工具] calculator({'expression': '1234 * 5678'})
上海32°C，北京28°C，上海更热。1234 × 5678 = 7006652。

 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

import json
from concurrent.futures import ThreadPoolExecutor

def run_tool_call(tool_call):
    """执行单个工具调用，返回对应的tool消息"""
    name = tool_call.function.name
    args = json.loads(tool_call.function.arguments)  # 解析参数JSON
    func = TOOL_REGISTRY.get(name)                    # 查注册表
    try:
        result = func(**args) if func else {"error": f"未知工具: {name}"}
    except Exception as e:
        result = {"error": str(e)}
    # 返回标准格式的tool消息，tool_call_id用来和原调用配对
    return {
        "role": "tool",
        "tool_call_id": tool_call.id,
        "content": json.dumps(result, ensure_ascii=False),
    }

def run_agent_parallel(user_query: str, max_turns: int = 5) -> str:
    messages = [
        {"role": "system", "content": "你是一个能查天气和做计算的助手。"},
        {"role": "user", "content": user_query},
    ]
    for _ in range(max_turns):
        response = client.chat.completions.create(
            model=MODEL, messages=messages, tools=TOOLS_SCHEMA
        )
        msg = response.choices[0].message
        if not msg.tool_calls:
            return msg.content
        messages.append(msg)

        # 关键改动：用线程池并行执行本轮所有工具调用
        with ThreadPoolExecutor() as pool:
            tool_msgs = list(pool.map(run_tool_call, msg.tool_calls))
        messages.extend(tool_msgs)  # 一次性把所有结果回灌

    return "达到最大轮数，处理中止。"

1

pip install google-search-results

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

# search_tool.py
import os
from serpapi import GoogleSearch

def web_search(query: str, num_results: int = 5) -> dict:
    """使用SerpAPI进行网络搜索，返回前N条结果摘要"""
    search = GoogleSearch({
        "q": query,
        "num": num_results,
        "api_key": os.getenv("SERPAPI_KEY")   # 密钥同样从环境变量读
    })
    data = search.get_dict()
    results = []
    # 只取organic_results（自然搜索结果），裁剪成精简结构再返回
    for item in data.get("organic_results", [])[:num_results]:
        results.append({
            "title": item.get("title"),
            "snippet": item.get("snippet"),
            "link": item.get("link"),
        })
    return {"query": query, "results": results}

 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

# file_tool.py
import os
from pathlib import Path

# 把所有文件操作锁死在这个workspace目录里
WORK_DIR = Path("./workspace").resolve()

def read_file(path: str) -> dict:
    """读取指定路径的文本文件内容。路径相对于workspace目录。"""
    target = (WORK_DIR / path).resolve()
    # 防御路径穿越：解析后的真实路径必须仍然在WORK_DIR之下
    if not str(target).startswith(str(WORK_DIR)):
        return {"error": "禁止访问workspace之外的文件"}
    if not target.exists():
        return {"error": f"文件不存在: {path}"}
    return {"path": path, "content": target.read_text(encoding="utf-8")}

def write_file(path: str, content: str) -> dict:
    """向指定路径写入文本内容。"""
    target = (WORK_DIR / path).resolve()
    if not str(target).startswith(str(WORK_DIR)):
        return {"error": "禁止写入workspace之外的文件"}
    # 自动创建父目录，避免"目录不存在"报错
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(content, encoding="utf-8")
    return {"path": path, "bytes_written": len(content.encode("utf-8"))}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

# db_tool.py
import sqlite3
import re

DB_PATH = "./data/agent.db"

def query_database(sql: str) -> dict:
    """执行只读SQL查询，返回结果行。仅允许SELECT语句。"""
    # 用正则把非SELECT语句挡在门外（不区分大小写）
    if not re.match(r"^\s*select\b", sql, re.IGNORECASE):
        return {"error": "仅允许SELECT查询"}
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row  # 让查询结果可以按列名取值
    try:
        cur = conn.execute(sql)            # 执行一次，拿到游标
        columns = [d[0] for d in cur.description]  # 从游标取列名
        rows = cur.fetchall()              # 再从同一游标取所有行
        return {"columns": columns, "rows": [list(r) for r in rows]}
    except Exception as e:
        return {"error": str(e)}
    finally:
        conn.close()  # 无论成功失败都关闭连接，防止连接泄漏

 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

# base_tool.py
import inspect
import json
from typing import Callable, Dict, Any

class BaseTool:
    """工具基类：子类实现run方法，自动从签名生成schema"""
    name: str = ""
    description: str = ""

    def run(self, **kwargs) -> Dict[str, Any]:
        raise NotImplementedError  # 子类必须实现

    def to_schema(self) -> Dict[str, Any]:
        """从run方法的参数签名自动生成OpenAI工具schema"""
        sig = inspect.signature(self.run)
        properties = {}
        required = []
        for pname, param in sig.parameters.items():
            properties[pname] = {"type": "string", "description": pname}
            # 没有默认值的参数标记为必填
            if param.default is inspect.Parameter.empty:
                required.append(pname)
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": {
                    "type": "object",
                    "properties": properties,
                    "required": required,
                },
            },
        }

class WeatherTool(BaseTool):
    name = "get_weather"
    description = "查询指定城市实时天气"

    def run(self, location: str, unit: str = "celsius") -> Dict[str, Any]:
        from tools_impl import get_weather
        return get_weather(location, unit)

class ToolRegistry:
    """统一管理多个工具：注册、生成schema列表、按名执行"""
    def __init__(self):
        self._tools: Dict[str, BaseTool] = {}

    def register(self, tool: BaseTool):
        self._tools[tool.name] = tool
        return self  # 链式调用：registry.register(A).register(B)

    def schemas(self):
        return [t.to_schema() for t in self._tools.values()]

    def execute(self, name: str, args: dict):
        tool = self._tools.get(name)
        return tool.run(**args) if tool else {"error": f"未知工具: {name}"}

# 使用示例
registry = ToolRegistry()
registry.register(WeatherTool())
# registry.register(CalculatorTool()).register(SearchTool())...

1

{"description": "查询天气"}

1

{"description": "查询指定城市的实时天气，返回温度、天气状况、湿度。当用户询问天气、温度、是否需要带伞、穿衣建议时调用。仅支持中国地级市及以上行政区，不支持乡镇。"}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

import time

def safe_execute(func, args, retries=3):
    for attempt in range(retries):
        try:
            result = func(**args)
            # 业务级错误：不重试，直接返回给LLM让它自己处理
            if isinstance(result, dict) and "error" in result:
                return result
            return result
        except Exception as e:
            # 系统/网络级错误：重试，指数退避避免雪崩
            if attempt == retries - 1:
                return {"error": f"工具执行失败({attempt+1}次): {e}"}
            time.sleep(2 ** attempt)  # 1秒、2秒、4秒...
    return {"error": "未知错误"}

1
2
3
4
5
6
7
8
9

CONFIRM_REQUIRED = {"send_email", "delete_file", "transfer_money"}

def maybe_confirm(tool_name, args):
    """高危工具调用前先问一下人"""
    if tool_name in CONFIRM_REQUIRED:
        # 实际项目弹UI或发审批消息，这里用input演示
        if input(f"确认执行 {tool_name}({args})? (y/n): ").lower() != "y":
            return {"error": "用户取消执行"}
    return None

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

import json
import jsonschema

def validate_tool_args(args_json, schema):
    """容错解析+校验：先看JSON能不能解析，再看符不符合schema"""
    try:
        args = json.loads(args_json)
    except json.JSONDecodeError as e:
        return None, f"JSON解析失败: {e}"
    try:
        jsonschema.validate(args, schema)
    except jsonschema.ValidationError as e:
        return None, f"参数校验失败: {e}"
    return args, None