02 · Function Calling

02 · Function Calling#

LLM 本身是「纯文字」的——它不能查天气、不能查数据库、不能执行代码。 Function Calling 给模型装上「手」：模型决定调用哪个工具、传什么参数，程序执行工具，把结果再喂回模型。

这是构建 AI Agent 的核心机制。

本章从单工具开始，逐步到多工具并行、工具调用循环。

import json
import os
import litellm
from dotenv import load_dotenv

load_dotenv()
MODEL = os.getenv("LLM_MODEL")

1. 单工具调用#

Function Calling 的流程：

用户输入 → 模型判断需要调用工具 → 返回 tool_call（函数名+参数）
         → 程序执行函数 → 把结果送回模型 → 模型生成最终回答

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

# 模拟工具实现（真实场景替换为实际 API）
def get_weather(city: str, unit: str = "celsius") -> dict:
    fake_data = {
        "北京": {"temp": 18, "condition": "晴",  "humidity": 45},
        "上海": {"temp": 23, "condition": "多云", "humidity": 72},
        "广州": {"temp": 29, "condition": "阵雨", "humidity": 88},
    }
    data = fake_data.get(city, {"temp": 20, "condition": "未知", "humidity": 60})
    if unit == "fahrenheit":
        data["temp"] = data["temp"] * 9/5 + 32
    data["city"] = city
    data["unit"] = unit
    return data


def run_tool(tool_call) -> str:
    """执行 tool_call，返回结果字符串。"""
    name = tool_call.function.name
    args = json.loads(tool_call.function.arguments)
    if name == "get_weather":
        return json.dumps(get_weather(**args), ensure_ascii=False)
    return json.dumps({"error": f"Unknown tool: {name}"})


# 完整的单轮 Function Calling 流程
def chat_with_tools(user_message: str, verbose: bool = True) -> str:
    messages = [{"role": "user", "content": user_message}]

    # 第一次调用：模型决定是否要用工具
    response = litellm.completion(model=MODEL, messages=messages, tools=tools)
    msg = response.choices[0].message

    if not msg.tool_calls:
        return msg.content  # 不需要工具，直接返回

    if verbose:
        print(f"模型决定调用工具: {msg.tool_calls[0].function.name}")
        print(f"参数: {msg.tool_calls[0].function.arguments}")

    # 执行工具
    messages.append(msg)  # 加入模型的 tool_call 消息
    for tc in msg.tool_calls:
        result = run_tool(tc)
        if verbose:
            print(f"工具返回: {result}")
        messages.append({
            "role": "tool",
            "tool_call_id": tc.id,
            "content": result,
        })

    # 第二次调用：模型根据工具结果生成最终回答
    final = litellm.completion(model=MODEL, messages=messages, tools=tools)
    return final.choices[0].message.content


print(chat_with_tools("北京今天天气怎么样？"))

模型决定调用工具: get_weather
参数: {"city":"北京","unit":"celsius"}
工具返回: {"temp": 18, "condition": "晴", "humidity": 45, "city": "北京", "unit": "celsius"}

现在北京是晴天，气温约 18°C，相对湿度 45%。  

建议：  
- 早晚有些凉，出门带一件薄外套/轻便夹克比较合适；  
- 白天阳光足，注意防晒（太阳镜/防晒霜）；  
- 没有降雨，可暂不带雨具；  
- 室外活动总体适宜。  

需要我帮你查今日逐小时预报、未来几天天气或空气质量（AQI）吗？

# 不需要工具时模型直接回答
print(chat_with_tools("你好，能介绍一下你自己吗？"))

你好！我叫 ChatGPT（你也可以叫我“助手”）。下面简单介绍一下我能做的事和我的限制，方便你决定怎么用我：

我是什么
- 一个由 OpenAI 训练的大型语言模型，基于 GPT-4 架构（知识截止到 2024-06）。
- 当前日期：2026-03-14（我可以在对话中使用当前日期信息）。

我能做的事（示例）
- 回答问题：常识、科普、历史、技术原理等。
- 写作与润色：邮件、报告、演讲稿、小说、摘要等。
- 翻译与语言学习：中英互译、语法解释、写作建议。
- 编程与代码帮助：示例代码、调试思路、算法解释。
- 数据分析与公式推导：数学题、统计、建模思路（但不能直接运行代码）。
- 计划与建议：学习计划、职业建议、时间管理、旅行建议等。
- 创意工作：头脑风暴、文案、设计思路。

我有什么限制
- 知识截止到 2024-06，无法得知之后发生的事实性事件或最新数据（可以尝试联网工具或你提供信息）。
- 可能会犯错或给出不准确的答案，重要结论请交叉核实。
- 我不是人类，没有个人经历或情感，不能做出法律、医学等权威诊断（可提供一般信息和建议，必要时请咨询专业人士）。
- 无法执行物理世界的动作或直接访问你设备上的私人文件，除非你把内容粘贴到对话中。

隐私与安全
- 请避免在对话中分享敏感个人信息（如身份证号、银行账号、密码等）。
- 我不会主动记住你在本次会话外的个人信息，但对话内容可能会被用于改进模型（请参阅平台隐私政策）。

需要我怎么帮你？
- 你可以直接问问题或告诉我你想要的任务样式（比如“帮我写一封求职信”“解释一下量子纠缠”“把这段中文翻译成英文”）。
- 如果想让我示范某种风格或长度（正式/口语、300字/一页等），请一并说明。

你现在想从哪方面开始？

2. 多工具并行调用#

现代 LLM 支持在一次响应里调用多个工具（并行），减少往返次数。

# 扩展工具集
multi_tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的当前天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string"},
                    "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
                },
                "required": ["city"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "执行数学计算，支持加减乘除和简单表达式",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {"type": "string", "description": "数学表达式，如 '(18+23)/2'"},
                },
                "required": ["expression"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "search_news",
            "description": "搜索最新新闻",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "搜索关键词"},
                    "limit": {"type": "integer", "description": "返回条数", "default": 3},
                },
                "required": ["query"],
            },
        },
    },
]

def calculate(expression: str) -> dict:
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return {"expression": expression, "result": result}
    except Exception as e:
        return {"error": str(e)}

def search_news(query: str, limit: int = 3) -> dict:
    # 模拟新闻搜索
    return {"query": query, "results": [
        f"[模拟新闻 {i+1}] 关于'{query}'的最新报道" for i in range(limit)
    ]}

def dispatch_tool(tool_call) -> str:
    name = tool_call.function.name
    args = json.loads(tool_call.function.arguments)
    fn_map = {"get_weather": get_weather, "calculate": calculate, "search_news": search_news}
    result = fn_map[name](**args) if name in fn_map else {"error": f"Unknown: {name}"}
    return json.dumps(result, ensure_ascii=False)


def chat_multi_tools(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]
    response = litellm.completion(model=MODEL, messages=messages, tools=multi_tools)
    msg = response.choices[0].message

    if not msg.tool_calls:
        return msg.content

    print(f"模型并行调用 {len(msg.tool_calls)} 个工具:")
    messages.append(msg)

    for tc in msg.tool_calls:
        result = dispatch_tool(tc)
        print(f"  ✓ {tc.function.name}({tc.function.arguments}) → {result[:80]}")
        messages.append({"role": "tool", "tool_call_id": tc.id, "content": result})

    final = litellm.completion(model=MODEL, messages=messages, tools=multi_tools)
    return final.choices[0].message.content

# 一个需要多工具的问题
print(chat_multi_tools("北京和上海今天气温分别是多少？平均气温是多少度？"))

模型并行调用 2 个工具:
  ✓ get_weather({"city": "北京", "unit": "celsius"}) → {"temp": 18, "condition": "晴", "humidity": 45, "city": "北京", "unit": "celsius"}
  ✓ get_weather({"city": "上海", "unit": "celsius"}) → {"temp": 23, "condition": "多云", "humidity": 72, "city": "上海", "unit": "celsius"}

北京今天气温：18°C（晴）。  
上海今天气温：23°C（多云）。  
两地平均气温为：(18 + 23) / 2 = 20.5°C（摄氏）。

3. 工具调用循环（Agent 雏形）#

复杂任务需要多轮工具调用。让模型持续调用工具，直到认为任务完成。这是 ReAct Agent 的核心结构。

def agent_loop(task: str, max_steps: int = 5) -> str:
    """
    工具调用循环：模型持续调用工具直到给出最终答案。
    max_steps: 最大工具调用轮数（防止无限循环）
    """
    messages = [
        {"role": "system", "content": "你是一个有用的助手，可以使用工具来完成任务。请尽可能使用工具获取准确信息。"},
        {"role": "user", "content": task},
    ]

    for step in range(max_steps):
        response = litellm.completion(model=MODEL, messages=messages, tools=multi_tools)
        msg = response.choices[0].message
        finish_reason = response.choices[0].finish_reason

        # 模型不再调用工具，给出最终答案
        if finish_reason == "stop" or not msg.tool_calls:
            print(f"[Step {step+1}] 任务完成")
            return msg.content

        # 执行工具
        print(f"[Step {step+1}] 调用 {len(msg.tool_calls)} 个工具")
        messages.append(msg)
        for tc in msg.tool_calls:
            result = dispatch_tool(tc)
            print(f"  → {tc.function.name}: {result[:60]}...")
            messages.append({"role": "tool", "tool_call_id": tc.id, "content": result})

    return "[达到最大步数，未完成任务]"


result = agent_loop(
    "帮我查一下北京、上海、广州三个城市的天气，然后告诉我哪个城市最适合今天户外活动？"
)
print("\n最终答案：")
print(result)

[Step 1] 调用 3 个工具
  → get_weather: {"temp": 18, "condition": "晴", "humidity": 45, "city": "北京",...
  → get_weather: {"temp": 23, "condition": "多云", "humidity": 72, "city": "上海"...
  → get_weather: {"temp": 29, "condition": "阵雨", "humidity": 88, "city": "广州"...

[Step 2] 任务完成

最终答案：
我查到了三座城市今天的实时天气（摄氏）：

- 北京：18°C，晴，湿度45%  
- 上海：23°C，多云，湿度72%  
- 广州：29°C，阵雨，湿度88%

结论与建议：
- 最适合今天户外活动：北京。天气晴朗、气温舒适（约18°C）、湿度低，适合散步、跑步或短途郊游。白天紫外线可能有，外出可备薄外套和防晒。  
- 次选：上海。温度适中（23°C）、无降雨预报但多云，适合大多数户外活动；湿度偏高，做高强度运动时要注意补水。  
- 不建议：广州。气温偏高且有阵雨、湿度很大，可能闷热且易下雨，户外活动受限，建议改室内活动或带雨具并注意防暑。

需要我再帮你查今天各城市的空气质量（AQI）或逐小时预报/未来几天的天气吗？

4. 错误处理#

工具调用可能失败——网络超时、参数错误、权限不足等。正确的做法是把错误信息传回模型，让它决定如何处理。

def safe_dispatch(tool_call) -> str:
    """带错误处理的工具执行，把错误信息传回模型。"""
    try:
        return dispatch_tool(tool_call)
    except Exception as e:
        # 不要抛出异常，把错误作为工具结果传回模型
        error_result = {
            "error": str(e),
            "tool": tool_call.function.name,
            "suggestion": "请检查参数是否正确，或尝试其他方式",
        }
        return json.dumps(error_result, ensure_ascii=False)


# 模拟一个会失败的工具调用
def get_stock_price(symbol: str) -> dict:
    raise ConnectionError(f"无法连接到股票数据服务（symbol={symbol}）")

stock_tools = [{
    "type": "function",
    "function": {
        "name": "get_stock_price",
        "description": "获取股票实时价格",
        "parameters": {
            "type": "object",
            "properties": {"symbol": {"type": "string", "description": "股票代码"}},
            "required": ["symbol"],
        },
    },
}]

messages = [{"role": "user", "content": "苹果公司股票现在多少钱？"}]
response = litellm.completion(model=MODEL, messages=messages, tools=stock_tools)
msg = response.choices[0].message

if msg.tool_calls:
    messages.append(msg)
    tc = msg.tool_calls[0]
    # 工具执行失败，把错误传回
    try:
        args = json.loads(tc.function.arguments)
        result = get_stock_price(**args)
    except Exception as e:
        result = json.dumps({"error": str(e), "suggestion": "服务暂时不可用"}, ensure_ascii=False)
    
    print(f"工具错误: {result}")
    messages.append({"role": "tool", "tool_call_id": tc.id, "content": result})
    
    # 模型根据错误信息给出回答
    final = litellm.completion(model=MODEL, messages=messages, tools=stock_tools)
    print(f"\n模型的处理: {final.choices[0].message.content}")

工具错误: {"error": "无法连接到股票数据服务（symbol=AAPL）", "suggestion": "服务暂时不可用"}

模型的处理: 抱歉，我刚才无法获取实时数据 —— 数据服务暂时不可用（错误：无法连接到股票数据服务）。你想怎么做？我可以：

- 现在再试一次去获取 AAPL 的实时价格（要我重试吗？）
- 或者我可以告诉你在哪里马上查看：Google 财经、Yahoo Finance、新浪财经/东方财富/同花顺、雪球、你的券商 APP（Robinhood、Fidelity、Interactive Brokers 等）。
- 如果你愿意，也可以让我给出我截止到 2024‑06 的最后已知价格（但那不是实时数据，可能已过时）。

请选择你想要的操作。

小结#

概念	要点
Function Calling	模型输出工具名+参数，程序执行后把结果送回
并行工具调用	一次响应调用多个工具，减少往返次数
工具调用循环	多轮调用直到完成任务，是 Agent 的核心
错误处理	工具失败时把错误传回模型，让它决定如何应对

下一章 → Vision：给模型「眼睛」，处理图片输入