LangChain实战——实现多轮对话 + Function Calling
本文介绍了如何利用LangChain框架实现大语言模型的函数调用(Function Calling)功能。通过一个完整示例,演示了定义工具函数、绑定工具到模型、构建多轮对话流以及自动调用工具完成任务的完整流程。文章详细解析了使用@tool装饰器注册工具函数、初始化Qwen模型、处理工具调用响应等关键步骤,并展示了天气查询和加法运算的实际运行效果。该方案支持多轮对话上下文记忆,模型可自动识别工具调用
随着大语言模型(LLMs)的迅猛发展,“Function Calling”(函数调用)逐渐成为一个重要的能力,它使得模型不仅能聊天,还能像“中控大脑”一样调用外部函数完成具体任务,比如查天气、调用数据库、控制机器人等。
本篇文章将以一个完整可运行的 LangChain 示例为背景,手把手教你如何:
- 定义外部工具函数(Tools)
- 将函数注册到 LLM
- 构建带有多轮消息历史的对话流
- 自动识别并调用工具完成任务
- 再将结果交由模型处理,给出最终回复
一、什么是 Function Calling?
Function Calling 是指模型根据上下文,自动生成调用特定函数的指令(如 JSON 结构的函数名 + 参数),然后由程序调用真实的函数执行,最终再将结果交还给模型生成最终答复。
LangChain 对 Function Calling 的支持非常完善,封装了 OpenAI、Qwen 等兼容 Function Calling 的模型接口,并提供工具注册、调用与上下文维护机制。
二、项目结构预览
我们实现的是一个简单的对话助手,支持两个工具:
get_weather(city: str):返回城市天气add(a: int, b: int):返回两个数的和
代码结构如下:
├── main.py # 主程序,包含模型初始化、工具绑定、消息循环
三、代码讲解
下面逐步解析代码中的关键部分。
1. 定义工具函数
使用 LangChain 提供的 @tool 装饰器即可将普通函数注册为 Tool:
from langchain_core.tools import tool
@tool
def get_weather(city: str) -> str:
"""返回指定城市的天气信息"""
return f"{city} 今天天气晴,28°C,湿度30%"
@tool
def add(a: int, b: int) -> int:
"""加法函数,返回两个整数的和"""
return a + b
LangChain 会自动基于函数签名和 docstring 为模型构建 JSON schema,模型即可调用它们。
2. 初始化模型
使用 Qwen 的 Function Calling 接口:
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
api_key="your_secret_key",
model="qwen-turbo",
temperature=0
)
Qwen 模型支持 Function Calling,且支持阿里云 API 兼容模式。
3. 绑定工具到模型
通过 bind_tools 把工具绑定到模型上:
llm_with_tools = llm.bind_tools([get_weather, add])
绑定后,模型就具备了识别调用这两个工具的能力。
4. 构建对话循环(含工具调用与多轮上下文)
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
import json
messages = [
AIMessage(content="你好!我是你的AI助手,可以帮你查天气、做加法,有什么可以帮你的?")
]
available_tools = {"get_weather": get_weather, "add": add}
while True:
user_input = input("你:").strip()
if user_input.lower() in {"exit", "quit", ""}:
print("再见!")
break
messages.append(HumanMessage(content=user_input))
response = llm_with_tools.invoke(messages)
if response.tool_calls:
print("检测到工具调用:")
print(json.dumps(response.tool_calls, indent=2, ensure_ascii=False))
messages.append(response)
for tool_call in response.tool_calls:
tool_func = available_tools.get(tool_call["name"])
if tool_func:
tool_result = tool_func.invoke(tool_call)
messages.append(tool_result)
print(f"执行 {tool_call['name']} → {tool_result.content}")
else:
messages.append(response)
print("AI:", response.content)
continue
final_response = llm_with_tools.invoke(messages)
messages.append(final_response)
print("AI:", final_response.content)
🤖 工作流程解析:
- 用户输入问题,如“北京天气如何?”
- 模型识别需要调用
get_weather(city=北京) - 返回
tool_calls,代码中通过invoke()实际执行函数 - 工具返回结果后,构造成
ToolMessage添加进对话上下文 - 再次调用模型生成基于工具结果的自然语言回答
四、运行效果演示
你:北京天气如何?
检测到工具调用:
[
{
"name": "get_weather",
"args": {
"city": "北京"
},
"id": "call_acfbc5447fa142899d4771",
"type": "tool_call"
}
]
执行 get_weather → 北京 今天天气晴,28°C,湿度30%
AI: 北京今天天气晴朗,温度是28°C,湿度为30%。
五、核心机制解析
1. 工具函数封装为 Tool 对象
LangChain 使用 ToolMessage 封装工具执行的返回值,供下一次模型使用。
ToolMessage(tool_call_id=..., content="返回结果")
2. invoke() 自动处理消息历史
模型接收完整的消息序列(系统消息、Human消息、AI消息、Tool消息),并根据上下文判断是否执行函数或直接回复。
六、总结
本文以一个简洁易懂的例子,演示了如何使用 LangChain 实现 Function Calling:
- ✅ 支持多轮对话与上下文记忆
- ✅ 模型自动识别是否需要工具调用
- ✅ 工具结果可参与后续推理
- ✅ 可快速拓展更多工具,如数据库查询、图像生成、邮件发送等
七、参考资料
附完整项目代码
import json
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
# Step 1:定义工具函数
@tool
def get_weather(city: str) -> str:
"""返回指定城市的天气信息"""
return f"{city} 今天天气晴,28°C,湿度30%"
@tool
def add(a: int, b: int) -> int:
"""加法函数,返回两个整数的和"""
return a + b
# Step 2:初始化模型
llm = ChatOpenAI(
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
api_key="your_secret_key",
model="qwen-turbo",
temperature=0
)
# Step 3:绑定工具
llm_with_tools = llm.bind_tools([get_weather, add])
# Step 4:初始化多轮消息历史
messages = [
AIMessage(content="你好!我是你的AI助手,可以帮你查天气、做加法,有什么可以帮你的?")
]
# Step 5:开始多轮对话循环
available_tools = {"get_weather": get_weather, "add": add}
while True:
user_input = input("你:").strip()
if user_input.lower() in {"exit", "quit", ""}:
print("再见!")
break
# 添加用户输入
messages.append(HumanMessage(content=user_input))
# 发送当前上下文给模型
response = llm_with_tools.invoke(messages)
# print(type(response))
# print(response)
# 判断是否需要调用工具
if response.tool_calls:
print("检测到工具调用:")
# print(type(response.tool_calls))
# print(response.tool_calls)
print(json.dumps(response.tool_calls, indent=2, ensure_ascii=False))
# 把 tool_call 消息添加到对话历史
messages.append(response)
# 执行所有工具调用并添加结果
for tool_call in response.tool_calls:
tool_name = tool_call["name"]
# print(tool_call["args"]["city"])
# print(type(tool_call["args"]["city"]))
tool_func = available_tools.get(tool_name)
if tool_func:
tool_result = tool_func.invoke(tool_call)
messages.append(tool_result)
print(f"执行 {tool_name} → {tool_result.content}")
else:
print(f"未找到工具:{tool_name}")
else:
# 如果没有 tool_call,直接是 AI 回答
messages.append(response)
print("AI:", response.content)
continue
# 工具执行完后再次交给模型生成最终回答
final_response = llm_with_tools.invoke(messages)
messages.append(final_response)
print("AI: ", final_response.content)
更多推荐
所有评论(0)