大语言模型中的 Function Calling

前言

         想象一下，当你向一个大语言模型（LLM）询问“今天北京的天气如何？”时，模型却只能根据其预训练的知识库给出一个模糊的回答，比如“北京的天气通常很宜人”，而无法提供具体的实时数据。

         然而，当你点击了“联网搜索”后，模型却能自动去网页搜索天气内容，并且总结后返回“今天北京天气晴朗，最高温度8℃，最低温度-1℃”，这背后的关键技术就是 Function Calling。

         Function Calling 让大模型能够突破预训练知识的局限，动态调用外部工具或函数，获取实时信息，从而提供更准确、更有用的回答。这种能力的出现，不仅让大模型的应用场景更加广泛，也为开发者提供了更强大的工具来构建智能应用。

一、Function Calling 与大模型的融合

         Function Calling 是大语言模型（LLM）生态系统中的一项关键功能，它允许模型动态调用外部函数或工具，从而扩展其能力范围。这种能力打破了传统模型仅依赖预训练知识的局限，使其能够处理实时数据和执行复杂任务。

（一）Function Calling 的定义

         Function Calling 是一种机制，使大语言模型能够根据用户的自然语言请求，动态生成并调用外部函数。这些函数可以是自定义的工具函数，也可以是调用第三方 API 的函数。例如，当用户询问“明天北京的天气如何”时，模型可以识别出这是一个获取天气信息的任务，并调用相应的天气查询函数。

（二） 与传统对话模型的区别

         与传统对话模型不同，Function Calling 赋予了模型调用外部工具的能力。传统模型主要依赖预训练知识生成回复，而 Function Calling 则允许模型动态获取最新信息，从而提供更准确、更实用的回答。这种能力的结合，让大模型在处理复杂任务和实时数据时更加得心应手。

二、Function Calling 的工作原理

         Function Calling 的工作流程可以分为以下几个关键步骤（图源：AI取经路）：

（一）定义函数

         开发者需要定义可供模型调用的函数，包括函数名、参数列表和功能描述。例如，定义一个获取城市天气的函数，其参数可能包括城市名称。

（二）用户请求

         用户通过自然语言向模型提出问题或请求。这些问题可能需要调用外部函数来获取答案或执行某些操作。

（三）模型解析与生成函数调用

         模型接收到用户输入后，会解析并理解输入内容。如果模型判断需要调用函数，它会生成一个包含函数调用所需参数的结构化输出，通常是一个 JSON 对象。这个 JSON 对象包含函数名和参数列表等信息。

（四）函数调用执行

         在代码环境中，开发者需要解析这个 JSON 对象，并使用其中的参数调用相应的函数。这个过程是在模型外部完成的。

（五）处理函数结果

         函数调用完成后，开发者需要将函数的结果返回给模型。模型会接收并处理这些结果，然后生成一个自然语言回复给用户，总结或解释函数调用的结果。

三、实战案例：Weather_EQ_Function_Calling.py 分析

         为了更好地理解 Function Calling 在实际项目中的应用，我们以一个智能问答系统为例，该系统能够动态调用天气查询和地震数据查询功能，同时利用大模型生成自然语言回答。这个系统展示了如何将 Function Calling 与大模型相结合，实现动态功能调用和实时数据处理。

（一）项目背景

         假设我们正在开发一个智能问答系统，用户可以通过自然语言向系统提问，系统需要根据问题的类型动态调用相应的函数来获取数据，并生成回答。具体来说，系统需要支持以下功能：

1. 查询指定地点的未来15天天气。
2. 查询最新的地震数据，并生成地震播报内容。

（二）系统架构

         系统的整体架构如下：

用户输入：用户通过自然语言向系统提问。
大模型解析：大模型解析用户输入，判断需要调用的函数及其参数。
动态函数调用：根据模型的解析结果，动态调用相应的函数。
外部数据获取：调用的函数通过 API 获取外部数据。
数据处理与回答生成：将获取的数据传递给大模型，生成自然语言回答。
返回结果：将生成的回答返回给用户。

（三）具体实现步骤

1.定义外部函数

         首先，我们需要定义两个外部函数：get_weather_data 和 get_earthquake_data。这些函数将通过 API 获取天气和地震数据。

get_weather_data：查询指定地点未来15天的天气。
get_earthquake_data：查询最新的地震数据。

2. 大模型解析用户输入

         用户通过自然语言向系统提问，例如：

“查询北京未来15天的天气。”
“查询最新的地震数据。”

         大模型解析这些输入，识别出需要调用的函数及其参数。例如，对于“查询北京未来15天的天气”，模型会解析出需要调用 get_weather_data 函数，并提取参数 province=“北京” 和 place=“北京”。

3. 动态函数调用

根据模型的解析结果，系统动态调用相应的函数。例如：

• 如果模型解析出需要调用 get_weather_data，系统会调用该函数，并传递解析出的参数。
• 如果模型解析出需要调用 get_earthquake_data，系统会调用该函数，无需传递参数。

4. 外部数据获取

调用的函数通过 API 获取外部数据。例如：

• get_weather_data 函数通过天气 API 获取指定地点的天气数据。
• get_earthquake_data 函数通过地震数据 API 获取最新的地震数据。

5. 数据处理与回答生成

将获取的数据传递给大模型，生成自然语言回答。例如：

• 对于天气数据，模型会生成未来15天的天气总结。
• 对于地震数据，模型会生成地震播报内容。

6. 返回结果

将生成的回答返回给用户。例如：

• “未来15天北京的天气总结：周一晴朗，最高温度25℃，最低温度15℃；周二多云……”
• “最新的地震播报：地震等级5.0，地震时间2025年3月16日14:00，位置：四川省某地，深度10千米。”
• “无需调用函数，正常回答用户问题”

（四）代码实现

import requests
import json
from zhipuai import ZhipuAI

# 配置接口参数
EARTHQUAKE_API_URL = "https://example.com/api/earthquake"  # 替换为实际的地震数据API
WEATHER_API_URL = "https://example.com/api/weather"       # 替换为实际的天气数据API
USER_ID = "YOUR_USER_ID"                                  # 替换为你的用户ID
USER_KEY = "YOUR_USER_KEY"                                # 替换为你的用户KEY
GLM_API_KEY = "YOUR_GLM_API_KEY"                          # 替换为你的智谱GLM-4-Flash API密钥

# 初始化大模型客户端
client = ZhipuAI(api_key=GLM_API_KEY)
model_name = "glm-4-flash"

# 获取地震数据
def get_earthquake_data():
    params = {
        "id": USER_ID,
        "key": USER_KEY
    }
    response = requests.get(EARTHQUAKE_API_URL, params=params)
    if response.status_code == 200:
        data = response.json()
        print("API返回的地震数据：", data)
        return data
    else:
        print("获取地震数据失败，状态码：", response.status_code)
        return None

# 使用大模型生成地震播报内容
def generate_earthquake_report(earthquake_data):
    if earthquake_data and earthquake_data.get("code") == 200:
        earthquake_reports = []
        for data in earthquake_data["data"][:5]:
            earthquake_info = f"地震等级: {data['leve']}, 地震时间: {data['addtime']}, 位置: {data['weizhi']}, 深度: {data['shendu']}千米"
            earthquake_reports.append(earthquake_info)

        response = client.chat.completions.create(
            model=model_name,
            messages=[
                {"role": "system", "content": "你是地震方面的专家，对用户提出的地震问题给出准确的答复。"},
                {"role": "user", "content": f"根据中国地震台网提供的地震数据：{earthquake_reports},回答用户提出的问题:{user_input}"}
            ]
        )
        answer = response.choices[0].message.content
    else:
        answer = "获取地震数据失败，请稍后再试。"
    return "AI:" + answer

# 查询指定地点的天气信息
def get_weather_data(province, place):
    params = {
        "id": USER_ID,
        "key": USER_KEY,
        "sheng": province,
        "place": place
    }
    response = requests.get(WEATHER_API_URL, params=params)
    if response.status_code == 200:
        data = response.json()
        print("API返回的数据：", data)
        return data
    else:
        print("获取天气数据失败，状态码：", response.status_code)
        return None

# 使用大模型总结天气
def summarize_weather(weather_data):
    if weather_data and weather_data.get("code") == 200:
        weather_reports = []
        for data in weather_data["data"]:
            weather_info = f"{data['week1']} {data['week2']}，白天：{data['wea1']}，夜间：{data['wea2']}，温度：{data['wendu1']} / {data['wendu2']}"
            weather_reports.append(weather_info)

        response = client.chat.completions.create(
            model=model_name,
            messages=[
                {"role": "system", "content": "生成未来15天的天气总结，每一天都简单说一下，每行代表一天天气,可以加点表情"},
                {"role": "user", "content": "根据以下天气数据生成总结：\n" + "\n".join(weather_reports)}
            ]
        )
        answer = response.choices[0].message.content
    else:
        answer = "获取天气数据失败，请稍后再试。"
    return "AI:" + answer

# 主函数
def main():
    global user_input
    user_input = input("User：")

    # 定义可用的工具列表
    tools = [
        {
            "type": "function",
            "function": {
                "name": "get_weather_data",
                "description": "查询指定地点未来15天的天气",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "province": {"type": "string", "description": "省份"},
                        "place": {"type": "string", "description": "城市"}
                    },
                    "required": ["province", "place"]
                }
            }
        },
        {
            "type": "function",
            "function": {
                "name": "get_earthquake_data",
                "description": "根据用户的地震相关问题给出相关回答",
                "parameters": {
                    "type": "object",
                    "properties": {},
                    "required": []
                }
            }
        }
    ]

    # 第一次调用模型，让模型提取参数并决定是否调用函数
    response = client.chat.completions.create(
        model=model_name,
        messages=[
            {"role": "user", "content": f"{user_input}"}
        ],
        tools=tools,
        tool_choice="auto"
    )

    # 检查是否需要调用函数
    if response.choices[0].message.tool_calls:
        tool_call = response.choices[0].message.tool_calls[0]
        if tool_call.function.name == "get_weather_data":
            function_args = json.loads(tool_call.function.arguments)
            province = function_args.get("province")
            place = function_args.get("place")

            weather_data = get_weather_data(province, place)
            if weather_data:
                summary = summarize_weather(weather_data)
                print(f"未来15天{place}的天气总结：\n{summary}")
            else:
                print("无法获取天气数据，请检查输入的地点是否正确。")

        elif tool_call.function.name == "get_earthquake_data":
            earthquake_data = get_earthquake_data()
            if earthquake_data:
                report = generate_earthquake_report(earthquake_data)
                print("地震播报内容：")
                print(report)
            else:
                print("无法获取地震数据。")
    else:
        print("工具判断：模型判断无需调用天气查询函数。")
        print("AI:" + response.choices[0].message.content)

if __name__ == "__main__":
    main()

（五）运行示例

1. 查询天气

假设用户输入以下问题：
用户输入：查询北京未来15天的天气。
模型解析：需要调用 get_weather_data 函数，参数为 province=“北京” 和 place=“北京”。
函数调用：调用 get_weather_data 函数，获取北京未来15天的天气数据。
数据处理：将天气数据传递给大模型，生成天气总结。
返回结果：未来15天北京的天气总结：周一晴朗，最高温度25℃，最低温度15℃；周二多云……

2. 查询地震情况

用户输入：查询最新的地震数据。
模型解析：需要调用 get_earthquake_data 函数，无需参数。
函数调用：调用 get_earthquake_data 函数，获取最新的地震数据。
数据处理：将地震数据传递给大模型，生成地震播报内容。
返回结果：最新的地震播报：地震等级5.0，地震时间2025年3月16日14:00，位置：四川省某地，深度10千米。

3. 正常问答

用户输入：你叫什么名字
模型解析：模型判断无需调用函数。
函数调用：无
数据处理：将用户输入数据传递给大模型，生成回复。
返回结果：我是一个名为 ChatGLM 的人工智能助手，是基于清华大学 KEG 实验室和智谱 AI 公司于 2024 年共同训练的语言模型开发的。我的任务是针对用户的问题和要求提供适当的答复和支持。。

四、总结

         通过这个实战案例，我们可以看到 Function Calling 如何使大模型能够动态调用外部函数，获取实时数据，并生成自然语言回答。这种能力不仅扩展了模型的功能，还提高了系统的灵活性和实用性。以下是 Function Calling 在这个项目中的具体优势：

1. 实时性
   Function Calling 允许模型动态调用外部 API 获取最新的数据，而不是依赖预训练的知识库。例如，在查询天气或地震数据时，系统能够实时获取最新的信息，从而提供最准确的回答。
2. 灵活性
   Function Calling 使系统能够根据用户输入动态选择和调用不同的函数。这种灵活性使得系统能够处理多种类型的请求，而无需为每种请求编写固定的逻辑。例如，用户可以查询天气、地震数据或其他任何支持的功能，系统都能动态响应。
3. 扩展性
   通过定义更多的外部函数，系统可以轻松扩展其功能。开发者可以添加新的功能模块，而无需修改核心逻辑。例如，如果需要添加空气质量查询功能，只需定义一个新的函数即可。
4. 模块化
   Function Calling 促进了代码的模块化设计。每个外部函数都封装了特定的功能，使得代码更加清晰、易于维护。例如，get_weather_data 和 get_earthquake_data 函数分别处理天气和地震数据的获取，使得代码逻辑更加简洁。
5. 高效性
   Function Calling 通过动态调用外部函数，避免了在模型中硬编码复杂的逻辑，从而提高了系统的运行效率。同时，模型可以专注于自然语言处理和生成回答，而外部函数负责数据获取和处理，这种分工使得系统更加高效。
6. 用户体验
   Function Calling 提供了更自然、更智能的用户体验。用户可以通过自然语言提出问题，系统能够理解并调用相应的函数来提供答案。这种交互方式更加直观，用户无需学习复杂的命令或接口。