用LangChain重构Function Calling：从手搓代码到流水线作业

m0_62603533

600人浏览 · 2025-12-17 08:00:00

m0_62603533 · 2025-12-17 08:00:00 发布

文章目录

0 前言
1 给函数加个“帽子”：@tool
2 绑定与链式调用：LCEL 的魔法
3 高级玩法：自定义输出解析器
总结

0 前言

上一篇刚把手动调用 Function Calling 的坑踩完，写了一堆 auto_functions 和 run_conversation，代码虽然能跑，但说实话，看着那一坨 JSON Schema 的转换逻辑还是挺累人的。

今天折腾了一下 LangChain，发现这玩意儿确实能省不少事。它把那些繁琐的代码都封装好了，而且提供了一套很简洁的“链式调用”写法。把之前的代码重构了一遍，感觉清爽了不少。

这里记录一下怎么用 LangChain 实现同样的天气查询功能，重点聊聊 @tool、bind_tools 和那个看着有点晕的 LCEL（LangChain Expression Language）。

1 给函数加个“帽子”：@tool

首先还是得有工具。在 LangChain 里，把一个普通的 Python 函数变成大模型能识别的 Tool，最简单的办法就是加个装饰器。

from langchain.tools import tool
import requests
import json
import os

# 这里的 key 还是得自己配
open_weather_key = "5c939a7cc59eb8696f4cd77bf75c5a9a"

# 使用@tool装饰器可以直接将get_weather函数转换成工具
@tool
def get_weather(loc):
    """
    查询即时天气函数
    :param loc: 必要参数，字符串类型，用于表示查询天气的具体城市名称，\
    注意，中国的城市需要用对应城市的英文名称代替，例如如果需要查询北京市天气，则loc参数需要输入'Beijing'；
    :return：OpenWeather API查询即时天气的结果...
    """
    # ... 省略中间构建请求的代码，和之前一模一样 ...
    
    response = requests.get(url, params=params)
    data = response.json()
    return json.dumps(data)

这点很关键：之前我们需要自己写个函数去解析函数的解释，然后拼凑成 OpenAI 需要的那个 JSON 格式。现在这个 @tool 装饰器把活全干了。它会自动读取你的函数签名、注释，在后台生成 Schema。你只管写业务逻辑，剩下的交给框架。

2 绑定与链式调用：LCEL 的魔法

工具定义好了，怎么让大模型知道有这个工具？以前得在 API 参数里传 tools=[…]，现在用 bind_tools。

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers.openai_tools import JsonOutputKeyToolsParser

# 实例化大模型，我这里还是用的本地 Ollama 跑的 Qwen
openai_chat = ChatOpenAI(model_name="qwen3:4b-instruct", api_key='None' ,base_url="http://localhost:11434/v1")

# 1. 绑定外部工具
# 这一步就把 tool 的定义注入到模型里了，相当于告诉模型：你有这些技能包
llm_with_tools = openai_chat.bind_tools([get_weather])

# 2. 构建执行链
# 这就是 LCEL (LangChain Expression Language)，用竖线 | 把步骤串起来
chain = llm_with_tools | JsonOutputKeyToolsParser(key_name='get_weather', first_tool_only=True) | get_weather

# 3. 触发
weather_data = chain.invoke("今天杭州的天气好吗？")

这段代码里最有意思的就是那个 chain 的定义。

llm_with_tools 是带了工具说明书的模型。当它收到“杭州天气”的请求时，它不会吐出“你好”，而是会吐出一个 Tool Call 的结构。

紧接着，管道符 | 把这个结果传给了 JsonOutputKeyToolsParser。这玩意儿是个解析器，专门用来从模型的回复里提取出“它想调用的函数名”和“参数”。这里我们指定只提取 get_weather 的参数。

最后，再通过 | 把参数传给 get_weather 函数本身。

这就形成了一条流水线：用户输入 -> 模型决策 -> 提取参数 -> 执行函数 -> 拿到JSON数据。

但是！注意这里有个坑。到这一步为止，变量 weather_data 里存的只是一堆干巴巴的 JSON 字符串（比如温度、湿度数值），大模型还没把这些数据翻译成人话。

所以，如果你想要像 ChatGPT 那样自然的回复，还得再加一步手动拼接：

from langchain_core.prompts import ChatPromptTemplate

# 构造输入模版，把刚才拿到的 JSON 数据塞进去
chat_template = ChatPromptTemplate.from_messages(
    [
        ("system", "天气信息来源于OpenWeather API：https://api.openweathermap.org/data/2.5/weather"),
        ("system", "这是实时的天气数据：{weather_data}"),
        ("human", "{user_input}"),
    ]
)

messages = chat_template.format_messages(weather_data=weather_data, user_input="今天杭州的天气好吗？")
response = openai_chat.invoke(messages)
print(response.content)

这就有点像我们之前手动写的 run_conversation 里的“第二次调用”。

3 高级玩法：自定义输出解析器

上面那种写法还是不够优雅，因为把“查数据”和“生成回复”割裂开了。理想情况是，我 invoke 一下，直接给我最终的人话结果。

这时候就可以自定义一个解析器函数，把它也串到 Chain 里去。

def final_resonse(ai_message: str) -> str:
    """
    这个函数负责接盘。
    它接收上一步（get_weather）返回的 JSON 字符串，
    然后重新构造 prompt，再次呼叫大模型进行总结。
    """
    data = json.loads(ai_message)
   
    chat_template = ChatPromptTemplate.from_messages(
        [
            ("system", "这是实时的{city}的天气数据...详细的数据是：{detail}",),
            ("system", "请你解析该数据，以自然语言的形式回复"),
        ]
    )
    
    # 这里动态构建 Prompt
    messages = chat_template.format_messages(city=data["name"], detail=data)

    # 再次调用模型
    # 注意：这里重新实例化一个模型或者复用之前的都行
    openai_chat = ChatOpenAI(model_name="qwen3:4b-instruct", api_key='None' ,base_url="http://localhost:11434/v1")
    response = openai_chat.invoke(messages)
    return response.content

# 现在的 Chain 变成了全自动流水线
chain = llm_with_tools | JsonOutputKeyToolsParser(key_name='get_weather', first_tool_only=True) | get_weather | final_resonse

final_reponse = chain.invoke("北京现在的天气怎么样？")
# 去掉多余的换行符
print(final_reponse.replace('\n', ''))