前言

Function Calling(函数调用 / 工具调用)是大模型落地到实际业务的关键能力。没有它,模型只能输出文本,有了它,模型才能查数据库、发 API、操作文件。很多开发者直接套 LangChain 或 Semantic Kernel,但出了问题根本不知道从哪排起。本文带你用纯 Python 手写一套可用的 Function Calling 机制,去掉所有框架封装,理解本质。

核心概念:Function Calling 不是什么黑魔法

先说清楚一件事:Function Calling 不是让大模型去执行代码。模型只是个文本生成器,它做的事情很简单——在输出中夹一段 JSON,告诉你"我想调用这个工具,参数是这样"。

所以完整的流程只有三步:

  1. 告诉模型有哪些工具可用——把工具定义成模型能看懂的描述格式
  2. 模型决定要不要调工具——如果模型觉得有必要,输出一个特殊格式的调用请求
  3. 你拿到请求后自己执行——解析 JSON,调函数,把结果塞回去让模型继续

整个过程不需要什么特殊 API,本质就是字符串拼接 + JSON 解析。

协议原理:工具描述到底长什么样

不论你用 OpenAI、Claude 还是本地模型,工具描述的核心字段都差不多。以 OpenAI 兼容格式为例,一个工具定义长这样:

{
  "type": "function",
  "function": {
    "name": "get_weather",
    "description": "获取指定城市的当前天气",
    "parameters": {
      "type": "object",
      "properties": {
        "city": {
          "type": "string",
          "description": "城市名称,如 北京、上海"
        },
        "unit": {
          "type": "string",
          "enum": ["celsius", "fahrenheit"],
          "description": "温度单位"
        }
      },
      "required": ["city"]
    }
  }
}

模型看到这个描述后,会理解:有一个叫 get_weather 的函数可用,它需要 city 参数,可选 unit,如果决定调用就会在回复中输出类似这样的内容:

{
  "function_call": {
    "name": "get_weather",
    "arguments": "{\"city\": \"北京\", \"unit\": \"celsius\"}"
  }
}

在最新 API 中,这个字段改成了 tool_calls,每个元素是一个 ToolCall 对象。我们后面按照新版协议来实现。

动手实现:从零开始写一个 Function Calling 引擎

第一步:定义工具 Schema

我们先用 Python 类把工具描述结构化,这样后续组装请求就更方便。

from typing import Any, Callable, Dict, List, Optional, Union

class ToolParameter:
    """描述函数的一个参数"""
    def __init__(
        self,
        name: str,
        type_: str,
        description: str = "",
        required: bool = True,
        enum: Optional[List[str]] = None,
    ):
        self.name = name
        self.type = type_
        self.description = description
        self.required = required
        self.enum = enum

    def to_dict(self) -> dict:
        d: dict = {"type": self.type, "description": self.description}
        if self.enum:
            d["enum"] = self.enum
        return d


class Tool:
    """封装一个可被模型调用的工具"""
    def __init__(
        self,
        name: str,
        fn: Callable,
        description: str,
        parameters: List[ToolParameter],
    ):
        self.name = name
        self.fn = fn  # 真正的 Python 函数
        self.description = description
        self.parameters = parameters

    def to_openai_tool(self) -> dict:
        """生成 OpenAI 兼容的 tool 描述"""
        props = {}
        required = []
        for p in self.parameters:
            props[p.name] = p.to_dict()
            if p.required:
                required.append(p.name)
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": {
                    "type": "object",
                    "properties": props,
                    "required": required,
                },
            },
        }

    def execute(self, **kwargs) -> str:
        """执行绑定的 Python 函数,将结果转为字符串"""
        result = self.fn(**kwargs)
        return str(result)

这样做的好处是:注册一个新工具只需要定义函数 + 描述参数,后面生成请求 JSON 和调用执行都可以自动化。

第二步:实现实际工具函数

我们写三个简单的工具用来演示:

import random
import json
from datetime import datetime

def get_weather(city: str, unit: str = "celsius") -> str:
    """模拟查询天气(实际项目中替换为真实 API 调用)"""
    weathers = ["晴", "多云", "阴", "小雨", "大雨", "小雪"]
    temps = {"celsius": (15, 35), "fahrenheit": (59, 95)}
    low, high = temps.get(unit, (15, 35))
    temp = random.randint(low, high)
    weather = random.choice(weathers)
    return f"{city}当前天气:{weather},温度 {temp}°{'C' if unit == 'celsius' else 'F'}"

def calculate(expression: str) -> str:
    """安全计算数学表达式"""
    allowed = set("0123456789+-*/.()% ")
    if not all(c in allowed for c in expression):
        return "错误:表达式包含非法字符"
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return f"{expression} = {result}"
    except Exception as e:
        return f"计算错误:{str(e)}"

def get_current_time(timezone: str = "Asia/Shanghai") -> str:
    """获取当前时间"""
    from datetime import timezone as tz
    import pytz
    try:
        tz_obj = pytz.timezone(timezone)
        now = datetime.now(tz_obj)
        return f"当前时间({timezone}):{now.strftime('%Y-%m-%d %H:%M:%S')}"
    except Exception:
        return f"不支持的时区:{timezone}"

然后注册成工具:

tools = [
    Tool(
        name="get_weather",
        fn=get_weather,
        description="获取指定城市的当前天气信息",
        parameters=[
            ToolParameter(name="city", type_="string", description="城市名称"),
            ToolParameter(name="unit", type_="string", description="温度单位", required=False, enum=["celsius", "fahrenheit"]),
        ],
    ),
    Tool(
        name="calculate",
        fn=calculate,
        description="计算数学表达式,支持四则运算和括号",
        parameters=[
            ToolParameter(name="expression", type_="string", description="数学表达式,如 2+3*4"),
        ],
    ),
    Tool(
        name="get_current_time",
        fn=get_current_time,
        description="获取指定时区的当前时间",
        parameters=[
            ToolParameter(name="timezone", type_="string", description="时区名称", required=False),
        ],
    ),
]

这样工具就注册好了,后面不管是组装请求给模型,还是执行回调,都基于这套结构。

第三步:组装 API 请求

我们的引擎需要支持接入任何兼容 OpenAI 的 API(包括本地部署的模型)。关键是把 tool schema 拼进 tools 字段:

import requests
import json

class FunctionCallingEngine:
    """手写 Function Calling 引擎"""

    def __init__(
        self,
        api_key: str,
        base_url: str = "https://api.openai.com/v1",
        model: str = "gpt-4o",
        tools: Optional[List[Tool]] = None,
    ):
        self.api_key = api_key
        self.base_url = base_url.rstrip("/")
        self.model = model
        self.tools = {t.name: t for t in (tools or [])}

    def _build_messages(
        self,
        user_input: str,
        history: Optional[List[dict]] = None,
    ) -> List[dict]:
        """构建完整的消息列表"""
        messages = list(history or [])
        messages.append({"role": "user", "content": user_input})
        return messages

    def _call_llm(self, messages: List[dict]) -> dict:
        """调用大模型 API,传入工具描述"""
        tool_defs = [t.to_openai_tool() for t in self.tools.values()]

        payload = {
            "model": self.model,
            "messages": messages,
            "tools": tool_defs,
            "tool_choice": "auto",  # 让模型自己决定是否调工具
        }

        resp = requests.post(
            f"{self.base_url}/chat/completions",
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json",
            },
            json=payload,
            timeout=60,
        )
        resp.raise_for_status()
        return resp.json()

核心代码就这么几行。注意 tool_choice: "auto" 是告诉模型:你觉得需要就用工具,不需要就直接回答。你也可以改成 "required" 强制模型每次都用工具,或者 {"type": "function", "function": {"name": "get_weather"}} 强制指定某个工具。

第四步:解析模型响应中的工具调用

模型返回后,我们需要检查 choices[0].message 中是否包含 tool_calls

    def _parse_tool_calls(self, message: dict) -> List[Dict[str, Any]]:
        """从模型回复中解析出工具调用请求"""
        tool_calls = message.get("tool_calls", [])
        if not tool_calls:
            return []

        results = []
        for tc in tool_calls:
            fn_info = tc.get("function", {})
            name = fn_info.get("name", "")
            raw_args = fn_info.get("arguments", "{}")

            # 解析参数 JSON
            try:
                args = json.loads(raw_args)
            except json.JSONDecodeError:
                args = {}

            tool = self.tools.get(name)
            if tool is None:
                result = f"错误:未知工具 {name}"
            else:
                try:
                    result = tool.execute(**args)
                except TypeError as e:
                    result = f"参数错误:{str(e)}"
                except Exception as e:
                    result = f"执行异常:{str(e)}"

            results.append({
                "tool_call_id": tc.get("id", ""),
                "tool_name": name,
                "result": result,
            })

        return results

这里要注意几个坑:

  • JSON 解析可能失败:部分模型输出的 arguments 不是严格的 JSON(比如尾随逗号、单引号),最好用 json.loads 包一层 try
  • 参数可能缺字段:模型可能只传了必填参数中的一部分,tool.execute(**args) 会抛 TypeError,要捕获
  • 执行也会出错:网络超时、除零、权限等异常,都不能让整个流程挂掉

第五步:实现完整的调用循环

这是整个引擎的核心——自动在"模型思考→调工具→返回结果→模型再思考"之间循环,直到模型给出最终答案:

    def chat(
        self,
        user_input: str,
        history: Optional[List[dict]] = None,
        max_turns: int = 10,
    ) -> dict:
        """
        完整的对话 + 工具调用循环

        Args:
            user_input: 用户输入
            history: 历史消息
            max_turns: 最大工具调用轮数(防止死循环)

        Returns:
            包含最终回复和完整消息历史的 dict
        """
        messages = self._build_messages(user_input, history)
        turn = 0

        while turn < max_turns:
            turn += 1

            # 1. 调用模型
            raw = self._call_llm(messages)
            choice = raw["choices"][0]
            msg = choice["message"]

            # 2. 把模型的回复追加到消息列表
            messages.append({
                "role": "assistant",
                "content": msg.get("content"),
                "tool_calls": msg.get("tool_calls"),
            })

            # 3. 检查是否有工具调用请求
            tool_results = self._parse_tool_calls(msg)
            if not tool_results:
                # 模型直接回答了,没有调工具,结束循环
                break

            # 4. 有工具调用:把每个工具的结果作为 tool 消息追加
            for tr in tool_results:
                messages.append({
                    "role": "tool",
                    "tool_call_id": tr["tool_call_id"],
                    "content": tr["result"],
                })

            # 5. 循环继续:模型看到 tool 结果后会决定下一步

        return {
            "reply": messages[-1].get("content", ""),
            "messages": messages,
            "turns": turn,
        }

这个循环的逻辑其实很简单:

  1. 把消息发给模型
  2. 模型要么直接回答,要么请求调工具
  3. 如果要调工具,执行函数并把结果塞回消息列表
  4. 带着结果再问模型一次
  5. 重复直到模型直接回答,或者到达最大轮数

这就是 Function Calling 的全部本质。没有魔法,就是一轮一轮的消息交换。

第六步:跑一个完整的例子

把上面的代码整合到一起,写一个可运行的 demo:

import os

def run_demo():
    # 从环境变量读取 API Key(也可以直接填字符串测试)
    api_key = os.getenv("OPENAI_API_KEY", "sk-your-key-here")
    base_url = os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")

    engine = FunctionCallingEngine(
        api_key=api_key,
        base_url=base_url,
        model="gpt-4o",
        tools=tools,
    )

    questions = [
        "北京今天天气怎么样?",
        "计算 (25 + 37) * 3 的结果",
        "现在几点了?",
        "北京天气怎么样,换算成华氏度是多少?",
    ]

    for q in questions:
        print(f"\n{'='*60}")
        print(f"用户:{q}")
        result = engine.chat(q, max_turns=5)
        print(f"模型:{result['reply']}")
        print(f"(共调用 {result['turns']} 轮)")

if __name__ == "__main__":
    run_demo()

如果 API Key 配置正确,输出大概是这样:

============================================================
用户:北京今天天气怎么样?
模型:北京今天天气:多云,温度 28°C
(共调用 2 轮)

============================================================
用户:计算 (25 + 37) * 3 的结果
模型:(25 + 37) * 3 = 186
(共调用 2 轮)

============================================================
用户:现在几点了?
模型:当前时间(Asia/Shanghai):2026-04-27 21:17:00
(共调用 2 轮)

============================================================
用户:北京天气怎么样,换算成华氏度是多少?
模型:北京今天天气:晴,温度 32°C,换算成华氏度为 89.6°F
(共调用 2 轮)

注意第二问和第四问,模型会先调工具拿到原始数据,然后用自然语言组织回复。第 4 问更典型——模型先调 get_weather 拿到摄氏温度,然后在回复中帮你算了华氏度,没有额外调 calculate

这就是模型在 理解上下文:它觉得摄氏度转华氏度这种简单换算自己能算,不需要再调工具。

进阶:更健壮的工具调用

支持强制指定工具

有时我们希望强制模型使用某个工具(比如客服场景必须先查用户信息):

def call_with_forced_tool(
    engine, messages, tool_name: str
) -> dict:
    """强制模型使用指定工具"""
    payload = {
        "model": engine.model,
        "messages": messages,
        "tools": [engine.tools[tool_name].to_openai_tool()],
        "tool_choice": {
            "type": "function",
            "function": {"name": tool_name},
        },
    }
    resp = requests.post(
        f"{engine.base_url}/chat/completions",
        headers={
            "Authorization": f"Bearer {engine.api_key}",
            "Content-Type": "application/json",
        },
        json=payload,
        timeout=60,
    )
    return resp.json()

并行工具调用

GPT-4o 和较新的模型支持一次返回多个 tool_calls,上面的引擎已经天然支持了——_parse_tool_calls 用循环处理所有 tool_calls,然后把结果全部追加回消息列表。不需要额外改造。

流式(Stream)处理

流式场景下的 tool calls 处理要麻烦一些,因为信息是分块到达的:

def chat_streaming(engine, user_input: str):
    """流式处理工具调用"""
    messages = engine._build_messages(user_input)
    tool_defs = [t.to_openai_tool() for t in engine.tools.values()]

    payload = {
        "model": engine.model,
        "messages": messages,
        "tools": tool_defs,
        "stream": True,
    }

    resp = requests.post(
        f"{engine.base_url}/chat/completions",
        headers={"Authorization": f"Bearer {engine.api_key}", "Content-Type": "application/json"},
        json=payload,
        stream=True,
        timeout=60,
    )

    # 流式解析:聚合 tool_calls 数据
    tool_calls = {}  # index -> accumulated data
    content_chunks = []

    for line in resp.iter_lines():
        if not line or line.startswith(b":") or line == b"data: [DONE]":
            continue
        if line.startswith(b"data: "):
            chunk = json.loads(line[6:])
            delta = chunk["choices"][0]["delta"]

            if delta.get("content"):
                content_chunks.append(delta["content"])

            if delta.get("tool_calls"):
                for tc in delta["tool_calls"]:
                    idx = tc.get("index", 0)
                    if idx not in tool_calls:
                        tool_calls[idx] = {"id": "", "function": {"name": "", "arguments": ""}}
                    call_data = tool_calls[idx]
                    if tc.get("id"):
                        call_data["id"] = tc["id"]
                    if tc.get("function", {}).get("name"):
                        call_data["function"]["name"] += tc["function"]["name"]
                    if tc.get("function", {}).get("arguments"):
                        call_data["function"]["arguments"] += tc["function"]["arguments"]

    # 处理工具调用
    if tool_calls:
        fake_msg = {"tool_calls": [v for _, v in sorted(tool_calls.items())]}
        results = engine._parse_tool_calls(fake_msg)
        for tr in results:
            messages.append({"role": "tool", "tool_call_id": tr["tool_call_id"], "content": tr["result"]})
        # 递归调用,但注意这里只是演示,实际需要重入
        return chat_streaming(engine, messages)

    return "".join(content_chunks)

流式处理的难点在于 tool_calls 是分片到达的,需要逐 chunk 拼装 function.namefunction.arguments,直到收到完整的 tool_calls 对象。

错误重试与退避

模型生成 JSON 时可能不严谨,加一层重试能提升可靠性:

import time

def _call_llm_with_retry(self, messages, max_retries=3):
    """带重试的 API 调用"""
    for attempt in range(max_retries):
        try:
            return self._call_llm(messages)
        except (requests.exceptions.Timeout,
                requests.exceptions.ConnectionError) as e:
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)  # 指数退避
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 429:  # 限流
                retry_after = int(e.response.headers.get("Retry-After", 3))
                time.sleep(retry_after)
            else:
                raise

完整的可运行代码

下面是把所有代码合并后的完整版本,保存为 function_calling.py 即可直接运行:

#!/usr/bin/env python3
"""
手写 Function Calling 引擎 - 完整可运行版
依赖:pip install requests pytz
"""
import json
import os
import random
import time
from datetime import datetime
from typing import Any, Callable, Dict, List, Optional, Tuple

import requests
import pytz

# ──────────────────── 工具定义层 ────────────────────

class ToolParameter:
    def __init__(self, name: str, type_: str, description: str = "",
                 required: bool = True, enum: Optional[List[str]] = None):
        self.name = name
        self.type = type_
        self.description = description
        self.required = required
        self.enum = enum

    def to_dict(self) -> dict:
        d: dict = {"type": self.type, "description": self.description}
        if self.enum:
            d["enum"] = self.enum
        return d


class Tool:
    def __init__(self, name: str, fn: Callable, description: str,
                 parameters: List[ToolParameter]):
        self.name = name
        self.fn = fn
        self.description = description
        self.parameters = parameters

    def to_openai_tool(self) -> dict:
        props = {}
        required = []
        for p in self.parameters:
            props[p.name] = p.to_dict()
            if p.required:
                required.append(p.name)
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": {"type": "object", "properties": props, "required": required},
            },
        }

    def execute(self, **kwargs) -> str:
        result = self.fn(**kwargs)
        return str(result)


# ──────────────────── 实际工具函数 ────────────────────

def get_weather(city: str, unit: str = "celsius") -> str:
    weathers = ["晴", "多云", "阴", "小雨", "大雨"]
    temps = {"celsius": (15, 35), "fahrenheit": (59, 95)}
    low, high = temps.get(unit, (15, 35))
    temp = random.randint(low, high)
    weather = random.choice(weathers)
    return f"{city}当前天气:{weather},温度 {temp}°{'C' if unit == 'celsius' else 'F'}"

def calculate(expression: str) -> str:
    allowed = set("0123456789+-*/.()% ")
    if not all(c in allowed for c in expression):
        return "错误:表达式包含非法字符"
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return f"{expression} = {result}"
    except Exception as e:
        return f"计算错误:{str(e)}"

def get_current_time(timezone: str = "Asia/Shanghai") -> str:
    try:
        tz_obj = pytz.timezone(timezone)
        now = datetime.now(tz_obj)
        return f"当前时间({timezone}):{now.strftime('%Y-%m-%d %H:%M:%S')}"
    except Exception:
        return f"不支持的时区:{timezone}"


# ──────────────────── 引擎核心 ────────────────────

class FunctionCallingEngine:
    def __init__(self, api_key: str, base_url: str = "https://api.openai.com/v1",
                 model: str = "gpt-4o", tools: Optional[List[Tool]] = None):
        self.api_key = api_key
        self.base_url = base_url.rstrip("/")
        self.model = model
        self.tools = {t.name: t for t in (tools or [])}

    def _build_messages(self, user_input: str,
                        history: Optional[List[dict]] = None) -> List[dict]:
        messages = list(history or [])
        messages.append({"role": "user", "content": user_input})
        return messages

    def _call_llm(self, messages: List[dict]) -> dict:
        tool_defs = [t.to_openai_tool() for t in self.tools.values()]
        payload = {
            "model": self.model,
            "messages": messages,
            "tools": tool_defs,
            "tool_choice": "auto",
        }
        resp = requests.post(
            f"{self.base_url}/chat/completions",
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json",
            },
            json=payload,
            timeout=60,
        )
        resp.raise_for_status()
        return resp.json()

    def _call_llm_with_retry(self, messages: List[dict],
                              max_retries: int = 3) -> dict:
        for attempt in range(max_retries):
            try:
                return self._call_llm(messages)
            except (requests.exceptions.Timeout,
                    requests.exceptions.ConnectionError) as e:
                if attempt == max_retries - 1:
                    raise
                time.sleep(2 ** attempt)
            except requests.exceptions.HTTPError as e:
                if e.response.status_code == 429:
                    retry_after = int(e.response.headers.get("Retry-After", 3))
                    time.sleep(retry_after)
                else:
                    raise

    def _parse_tool_calls(self, message: dict) -> List[Dict[str, Any]]:
        tool_calls = message.get("tool_calls", [])
        if not tool_calls:
            return []

        results = []
        for tc in tool_calls:
            fn_info = tc.get("function", {})
            name = fn_info.get("name", "")
            raw_args = fn_info.get("arguments", "{}")
            try:
                args = json.loads(raw_args)
            except json.JSONDecodeError:
                args = {}

            tool = self.tools.get(name)
            if tool is None:
                result = f"错误:未知工具 {name}"
            else:
                try:
                    result = tool.execute(**args)
                except TypeError as e:
                    result = f"参数错误:{str(e)}"
                except Exception as e:
                    result = f"执行异常:{str(e)}"

            results.append({
                "tool_call_id": tc.get("id", ""),
                "tool_name": name,
                "result": result,
            })
        return results

    def chat(self, user_input: str, history: Optional[List[dict]] = None,
             max_turns: int = 10) -> dict:
        messages = self._build_messages(user_input, history)
        turn = 0

        while turn < max_turns:
            turn += 1
            raw = self._call_llm_with_retry(messages)
            choice = raw["choices"][0]
            msg = choice["message"]

            messages.append({
                "role": "assistant",
                "content": msg.get("content"),
                "tool_calls": msg.get("tool_calls"),
            })

            tool_results = self._parse_tool_calls(msg)
            if not tool_results:
                break

            for tr in tool_results:
                messages.append({
                    "role": "tool",
                    "tool_call_id": tr["tool_call_id"],
                    "content": tr["result"],
                })

        return {
            "reply": messages[-1].get("content", ""),
            "messages": messages,
            "turns": turn,
        }


# ──────────────────── 工具注册 ────────────────────

tools = [
    Tool("get_weather", get_weather, "获取指定城市的当前天气信息", [
        ToolParameter("city", "string", "城市名称"),
        ToolParameter("unit", "string", "温度单位", required=False, enum=["celsius", "fahrenheit"]),
    ]),
    Tool("calculate", calculate, "计算数学表达式,支持四则运算和括号", [
        ToolParameter("expression", "string", "数学表达式"),
    ]),
    Tool("get_current_time", get_current_time, "获取指定时区的当前时间", [
        ToolParameter("timezone", "string", "时区名称", required=False),
    ]),
]


# ──────────────────── 运行 ────────────────────

if __name__ == "__main__":
    api_key = os.getenv("OPENAI_API_KEY", "sk-your-key-here")
    base_url = os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")

    engine = FunctionCallingEngine(
        api_key=api_key, base_url=base_url, model="gpt-4o", tools=tools,
    )

    questions = [
        "北京今天天气怎么样?",
        "计算 (25 + 37) * 3 的结果",
        "现在几点了?",
        "北京天气怎么样,换算成华氏度显示?",
    ]

    for q in questions:
        print(f"\n{'='*60}")
        print(f"用户:{q}")
        result = engine.chat(q, max_turns=5)
        print(f"模型:{result['reply']}")
        print(f"调用轮数:{result['turns']}")

一些重要的设计经验和坑

为什么模型有时不调工具?

即使你传了工具定义,模型也可能选择直接回答。原因通常是:
- 问题太简单,模型觉得自己能答
- 工具描述写得太模糊,模型没意识到该用
- 模型 size 太小,不支持 tool calling(比如某些 7B 以下的模型本身就不支持)

解决办法:把 tool_choice"auto" 改成 "required",或者优化工具描述,在 description 里加上使用场景提示。

参数不是你想的那样

模型生成的参数值可能跟你预设的 enum 不完全匹配。比如你定了 unit: ["celsius", "fahrenheit"],模型可能传 "C""摄氏度"。收到参数后建议做一层规整化处理。

系统提示词里也可以提工具

不一定只在 tools 字段里描述。在 system prompt 里写一句"你可以使用 get_weather 查询天气"也能提升工具的调用率。我的实践中,系统提示 + 工具描述的双重引导效果最好。

总结

Function Calling 的本质就是三个步骤的自动循环:

  1. 注册:用标准格式告诉模型有哪些工具可用
  2. 解析:从模型回复中提取工具调用请求
  3. 执行:调用真实函数,把结果塞回去让模型继续

我们没有依赖任何框架,用不到 300 行纯 Python 就实现了一个完整可用的引擎。这套代码可以直接跑,也可以按需扩展——加认证、加缓存、加多模态工具,都是在这个基础上做加法。

理解了这个底层机制,不管将来换什么模型、用什么框架,你都能一眼看穿它在背后做了什么。


💡 读者福利

手写系列的所有代码都需要模型推理来验证效果。如果你手头缺少 API 额度,推荐使用 硅基流动 的 AI 云平台:

  • 支持 DeepSeek、Qwen、GLM 等主流开源模型
  • 提供 OpenAI 兼容接口,本文代码无需修改即可接入
  • 新用户完成实名认证即赠 ¥16 代金券

📎 邀请链接:https://cloud.siliconflow.cn/i/qQMjNGt7
🔑 邀请码:qQMjNGt7

这 ¥16 足够跑完本文所有示例,甚至还能多试几个模型对比效果。


更多推荐