Langgraph学习2：LLM自动调用工具

在上一篇文章中，通过学习官方示例构建了一个基础的Langgraph chatbot工作流。今天更进一步，探索如何让chatbot能够调用工具来完成特定任务。

定义工具

简单定义一个无任何逻辑的工具函数，该函数直接会返回天气查询的结果来模拟LLM调用天气api的流程：

from langchain_core.tools import tool
from langchain_core.utils.function_calling import convert_to_openai_function

# 定义工具
@tool
def get_weather(query: str):
    """用于获取天气信息。"""
    return ["今天天气晴朗，温度20度", "明天天气多云，温度25度"]

# 创建工具列表
tools = [get_weather]

# 绑定工具到LLM
llm_with_tools = llm.bind_tools(tools)

# 创建OpenAI函数格式的工具描述
functions = [convert_to_openai_function(t) for t in tools]

# 创建tool节点
from langgraph.prebuilt import ToolNode
tool_node = ToolNode(tools)

使用了@tool装饰器来定义工具函数，这是Langchain提供的便捷方式。然后创建了工具列表并将其绑定到LLM，同时还创建了一个专门的ToolNode来处理工具调用。

以下是Claude-3.7-sonnet对tool装饰器的用法和解释：

@tool装饰器详解
@tool是LangChain提供的装饰器，它的作用是将普通Python函数转换为LangChain工具格式。这个装饰器会自动：
- 解析函数签名中的参数类型
- 从函数文档字符串中提取描述信息
- 创建一个符合LangChain工具标准的包装器
- 
使用@tool时需要注意：
- 函数必须有文档字符串（docstring），这将作为工具的描述传递给LLM
- 参数类型最好有明确的注解，这样LLM才能正确理解参数类型
- 参数名称应尽量具有描述性，这有助于LLM理解参数的作用
- 返回值应该是字符串或可序列化为字符串的对象，以便传递回LLM

llm_with_tools = llm.bind_tools(tools)这一步是将工具绑定到LLM实例，这样LLM在生成回复时就能感知到这些工具的存在。
convert_to_openai_function函数则是将LangChain工具格式转换为OpenAI函数调用格式，因为很多模型都兼容OpenAI的函数调用API标准。

修改聊天节点函数

需要改造原来的聊天节点函数，使其支持工具调用：

# llm的调用
async def chat_bot(state: MessagesState):
    """生成流式回复的节点函数"""
    messages = state["messages"]
    # 使用支持工具调用的方式调用LLM
    response = await llm_with_tools.ainvoke(
        messages,
        functions=functions,
        function_call="auto"
    )
    return {"messages": [response]}

通过传入functions参数和设置function_call=”auto”，让模型自行决定是否需要调用工具。

这意味着模型将根据用户的输入自动判断是否需要调用工具。如果模型认为需要调用工具，它会生成一个特殊的回复，包含工具调用信息而不是直接文本回答。

定义条件路由

from typing import Literal

# 定义边的逻辑判断（条件边），判断是否继续
def tool_router(state: MessagesState) -> Literal["tools", "__end__"]:
    print('state------>',state,'\n')
    messages = state['messages']
    last_message = messages[-1]
    print('last_message------>',last_message,'\n')
    print('last_message.tool_calls------>',last_message.tool_calls,'\n')
    if last_message.tool_calls: # 判断模型是否返回tools调用
        return "tools"
    return END

这个函数检查模型返回的最后一条消息是否包含工具调用请求。如果有，就路由到”tools”节点；如果没有，就结束工作流。

工具调用流程详解

当模型决定调用工具时，整个过程是这样的：

模型生成一个包含tool_calls属性的AIMessage对象
tool_router函数检查最新消息是否包含tool_calls
如果存在tool_calls，函数返回”tools”，指示工作流去执行工具节点
如果不存在，函数返回END，表示工作流结束

从日志中可以看到last_message.tool_calls的格式：

last_message.tool_calls------> [{'name': 'get_weather', 'args': {'query': '今明两天天气'}, 'id': '019619eaea388a97f7e451f4cefc4bfe', 'type': 'tool_call'}]

这个结构包含了关键信息：

name：要调用的工具名称（对应我们定义的函数名）
args：传递给工具的参数（这里是查询字符串）
id：工具调用的唯一标识符（用于后续匹配返回结果）
type：调用类型，固定为”tool_call”

ToolNode的工作原理

当工作流转到tool_node节点时，ToolNode会自动：

从最后一条消息中提取tool_calls
根据工具名称找到对应的工具函数
用提取的参数调用该函数
将结果封装为ToolMessage添加到消息历史中
返回更新后的状态给工作流
最后，工作流会再次进入chat_bot节点，让模型基于工具返回的结果生成最终回复。

扩展工作流

现在可以重新构建一个更复杂的工作流，包含工具调用和条件路由：

# 创建工作流程
workflow = StateGraph(MessagesState)
# 添加节点
workflow.add_node("chat_bot", chat_bot)
# 设置入口节点
workflow.set_entry_point("chat_bot")
# 添加tool节点
workflow.add_node("tools", tool_node)
# 从tools节点指向chat_bot节点，以便可能的后续交互
workflow.add_edge("tools", "chat_bot")
# 添加条件边
workflow.add_conditional_edges(
    "chat_bot",
    tool_router,  # 判断下一个调用的节点
)
# 编译图
app_graph = workflow.compile()

与第一篇不同这个工作流有了条件路由，执行路径现在有两种可能：
chat_bot → END（如果模型直接回答）
chat_bot → tools → chat_bot → END（如果模型需要调用工具）
add_conditional_edges方法是关键，它允许工作流根据路由函数的返回值选择不同的执行路径。

测试工作流

# 测试运行函数
async def run_streaming_chain():
    """运行graph的链"""
    print("开始生成回复...\n")
    messages = [
        SystemMessage(content="你是一个智能助手，使用专业且准确的语言回复用户的问题，且使用中文进行回复"),
        HumanMessage(content="帮我查一下今明两天的天气")
    ]
    
    # 初始化状态
    initial_state = {"messages": messages, "streamed_output": []}
    
    # 使用messages模式捕获流式输出
    async for event in app_graph.astream(initial_state, stream_mode='messages'):
        if isinstance(event, tuple):
            chunk: AIMessageChunk = event[0]
            if chunk.type == 'AIMessageChunk':
                print('event里监听到的流式输出------>',chunk.content,'\n\n')

与第一篇不同，这次选择了stream_mode=’messages’来观察模型的实时输出。

执行结果

测试结果展示了完整的工作流执行过程：

模型分析用户问题”帮我查一下今明两天的天气”，决定需要调用get_weather工具
工具调用节点执行get_weather函数，返回天气信息
模型收到天气信息后，生成了最终回复

关键是工具调用和路由部分的执行：

当模型返回包含tool_calls的消息时，tool_router函数返回”tools”
工作流转到tools节点，执行工具调用
调用的结果作为新消息被添加到对话历史
工作流返回到chat_bot节点，让模型基于工具结果生成最终回复

用户询问：
帮我查一下今明两天的天气

chatbot回答：
好的，我来帮您查询一下。请稍等片刻经过查询，今天天气晴朗，温度20度，明天天气多云，温度25度。请问还有其他需要我帮忙查询的吗？

此时的graph图：

以下是完整输出过程：

开始生成回复...

event里监听到的流式输出------>  


event里监听到的流式输出------>  


event里监听到的流式输出------>  


event里监听到的流式输出------>  


event里监听到的流式输出------>  


event里监听到的流式输出------>  


event里监听到的流式输出------>


event里监听到的流式输出------>


event里监听到的流式输出------>


event里监听到的流式输出------>  


event里监听到的流式输出------>


event里监听到的流式输出------>


state------> {'messages': [SystemMessage(content='你是一个智能助手，使用专业且准确的语言回复用户的问题，且使用中文进行回复', additional_kwargs={}, response_metadata={}, id='76b77c29-117e-4a0f-b36c-896e2eaf13ee'), HumanMessage(content='帮我查一下今明两天的天气', additional_kwargs={}, response_metadata={}, id='e0b6808b-a4ad-4ee5-829b-7718dbe06d88'), AIMessage(content='', additional_kwargs={'tool_calls': [{'index': 0, 'id': '019619eaea388a97f7e451f4cefc4bfe', 'function': {'arguments': '{"query": "今明两天天气"}', 'name': 'get_weather'}, 'type': 'function'}]}, response_metadata={'finish_reason': 'tool_calls', 'model_name': 'THUDM/glm-4-9b-chat'}, id='run-6b761404-e237-4cc4-a203-2316e0ecb3e3', tool_calls=[{'name': 'get_weather', 'args': {'query': '今明两天天气'}, 'id': '019619eaea388a97f7e451f4cefc4bfe', 'type': 'tool_call'}], usage_metadata={'input_tokens': 2076, 'output_tokens': 65, 'total_tokens': 2141, 'input_token_details': {}, 'output_token_details': {}})]}

last_message------> content='' additional_kwargs={'tool_calls': [{'index': 0, 'id': '019619eaea388a97f7e451f4cefc4bfe', 'function': {'arguments': '{"query": "今明两天天气"}', 'name': 'get_weather'}, 'type': 'function'}]} response_metadata={'finish_reason': 'tool_calls', 'model_name': 'THUDM/glm-4-9b-chat'} id='run-6b761404-e237-4cc4-a203-2316e0ecb3e3' tool_calls=[{'name': 'get_weather', 'args': {'query': '今明两天天气'}, 'id': '019619eaea388a97f7e451f4cefc4bfe', 'type': 'tool_call'}] usage_metadata={'input_tokens': 2076, 'output_tokens': 65, 'total_tokens': 2141, 'input_token_details': {}, 'output_token_details': {}}

last_message.tool_calls------> [{'name': 'get_weather', 'args': {'query': '今明两天天气'}, 'id': '019619eaea388a97f7e451f4cefc4bfe', 'type': 'tool_call'}]

event里监听到的流式输出------>  


event里监听到的流式输出------>



event里监听到的流式输出------> 好的


event里监听到的流式输出------> ，


event里监听到的流式输出------> 我来


event里监听到的流式输出------> 帮


event里监听到的流式输出------> 您


event里监听到的流式输出------> 查询 


event里监听到的流式输出------> 一下


event里监听到的流式输出------> 。


event里监听到的流式输出------> 请


event里监听到的流式输出------> 稍


event里监听到的流式输出------> 等


event里监听到的流式输出------> 片刻


event里监听到的流式输出------> 经过 


event里监听到的流式输出------> 查询


event里监听到的流式输出------> ，


event里监听到的流式输出------> 今天


event里监听到的流式输出------> 天气


event里监听到的流式输出------> 晴


event里监听到的流式输出------> 朗


event里监听到的流式输出------> ，


event里监听到的流式输出------> 温度


event里监听到的流式输出------> 20


event里监听到的流式输出------> 度 


event里监听到的流式输出------> ，


event里监听到的流式输出------> 明天


event里监听到的流式输出------> 天气


event里监听到的流式输出------> 多云


event里监听到的流式输出------> ，


event里监听到的流式输出------> 温度


event里监听到的流式输出------> 25


event里监听到的流式输出------> 度


event里监听到的流式输出------> 。


event里监听到的流式输出------> 请问


event里监听到的流式输出------> 还有


event里监听到的流式输出------> 其他 


event里监听到的流式输出------> 需要 


event里监听到的流式输出------> 我 


event里监听到的流式输出------> 帮忙 


event里监听到的流式输出------> 查询 


event里监听到的流式输出------> 的吗 


event里监听到的流式输出------> ？ 


event里监听到的流式输出------>  


state------> {'messages': [SystemMessage(content='你是一个智能助手，使用专业且准确的语言回复用户的问题，且使用中文进行回复', additional_kwargs={}, response_metadata={}, id='76b77c29-117e-4a0f-b36c-896e2eaf13ee'), HumanMessage(content='帮我查一下今明两天的天气', additional_kwargs={}, response_metadata={}, id='e0b6808b-a4ad-4ee5-829b-7718dbe06d88'), AIMessage(content='', additional_kwargs={'tool_calls': [{'index': 0, 'id': '019619eaea388a97f7e451f4cefc4bfe', 'function': {'arguments': '{"query": "今明两天天气"}', 'name': 'get_weather'}, 'type': 'function'}]}, response_metadata={'finish_reason': 'tool_calls', 'model_name': 'THUDM/glm-4-9b-chat'}, id='run-6b761404-e237-4cc4-a203-2316e0ecb3e3', tool_calls=[{'name': 'get_weather', 'args': {'query': '今明两天天气'}, 'id': '019619eaea388a97f7e451f4cefc4bfe', 'type': 'tool_call'}], usage_metadata={'input_tokens': 2076, 'output_tokens': 65, 'total_tokens': 2141, 'input_token_details': {}, 'output_token_details': {}}), ToolMessage(content='["今天天气晴朗，温度20度", "明天天气多云，温度25度"]', name='get_weather', id='4d8564fa-8301-4672-8e8d-64f6cecbf936', tool_call_id='019619eaea388a97f7e451f4cefc4bfe'), AIMessage(content='\n好的，我来帮您查询一下。请稍等片 刻经过查询，今天天气晴朗，温度20度，明天天气多云，温度25度。请问还有其他需要我帮忙查询的吗？', additional_kwargs={}, response_metadata={'finish_reason': 'stop', 'model_name': 'THUDM/glm-4-9b-chat'}, id='run-b9e66b49-9301-4343-8e51-67d3d48737c1', usage_metadata={'input_tokens': 8668, 'output_tokens': 945, 'total_tokens': 9613, 'input_token_details': {}, 'output_token_details': {}})]}

last_message------> content='\n好的，我来帮您查询一下。请稍等片刻经过查询，今天天气晴朗，温度20度，明天天气多云，温度25度。请问还有其他需要我帮忙查询的吗？' additional_kwargs={} response_metadata={'finish_reason': 'stop', 'model_name': 'THUDM/glm-4-9b-chat'} id='run-b9e66b49-9301-4343-8e51-67d3d48737c1' usage_metadata={'input_tokens': 8668, 'output_tokens': 945, 'total_tokens': 9613, 'input_token_details': {}, 'output_token_details': {}}

last_message.tool_calls------> []

以下是完整代码：

import os
from typing import Literal
from datetime import datetime
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, AIMessageChunk
from langgraph.graph import Graph, StateGraph, MessagesState, END
from langchain_community.chat_models import QianfanChatEndpoint
from langchain_core.tools import tool
from langgraph.prebuilt import ToolNode
import asyncio
from langchain_core.runnables.graph import MermaidDrawMethod
from langchain_core.utils.function_calling import convert_to_openai_function

# 百度千帆的调用方式
# llm = QianfanChatEndpoint(
#     model="ERNIE-Speed-128K",
#     streaming=True,  # 启用流式输出
#     api_key=os.getenv('QIANFAN_AK', ''),
#     secret_key=os.getenv('QIANFAN_SK', '')
# )

# 硅基流动的api调用方式
llm = ChatOpenAI(
    #THUDM/glm-4-9b-chat
    #Qwen/Qwen2.5-7B-Instruct
    model="THUDM/glm-4-9b-chat",
    streaming=False,  # 启用流式输出
    api_key=os.getenv('SILICONFLOW_API_KEY', ''), 
    base_url=os.getenv('SILICONFLOW_BASE_URL', ''),
    temperature=0.1,
)

# 定义工具
@tool
def get_weather(query: str):
    """用于获取天气信息。"""
    return ["今天天气晴朗，温度20度", "明天天气多云，温度25度"]

tools = [get_weather]
llm_with_tools = llm.bind_tools(tools)

# 创建工具列表的函数版本
functions = [convert_to_openai_function(t) for t in tools]
# 创建tool节点
tool_node = ToolNode(tools)

# llm的调用
async def chat_bot(state: MessagesState):
    """生成流式回复的节点函数"""
    messages = state["messages"]
    # response = await llm.ainvoke(messages)
    response = await llm_with_tools.ainvoke(
        messages,
        functions=functions,
        function_call="auto"
    )
    return {"messages": [response]}

# 4.定义边的逻辑判断（条件边），判断是否继续
def tool_router(state: MessagesState) -> Literal["tools", "__end__"]: #Literal用于限制返回的值的可选值
    print('state------>',state,'\n')
    messages = state['messages']
    last_message = messages[-1]
    print('last_message------>',last_message,'\n')
    print('last_message.tool_calls------>',last_message.tool_calls,'\n')
    if last_message.tool_calls: #判断models是否返回tools调用，有则告诉调用tools节点，否则结束
        return "tools"
    return END

# 创建工作流程
workflow = StateGraph(MessagesState)
# 添加节点
workflow.add_node("chat_bot", chat_bot)
# 设置入口节点
workflow.set_entry_point("chat_bot")
# 添加tool节点
workflow.add_node("tools", tool_node)
# 从tools节点指向chat_bot节点，以便可能的后续交互
workflow.add_edge("tools", "chat_bot")
workflow.add_conditional_edges(
    "chat_bot",
    tool_router,    #判断下一个调用的节点
)
# 编译图
app_graph = workflow.compile()



# 定义一个将图导出为PNG的函数
def export_graph_to_png():
    """
    将LangGraph图导出为PNG格式
    Returns:
        str: 生成的PNG文件路径
    """
    try:
        output_file='简单的chatbot-'+datetime.now().strftime("%Y-%m-%d_%H-%M-%S")+".png"
        app_graph.get_graph().draw_mermaid_png(
            draw_method=MermaidDrawMethod.API,
            output_file_path=output_file
        )
        # return True
    except Exception as e:
        print(f"导出PNG图形时出错: {e}")
        return None

# 测试运行函数
async def run_streaming_chain():
    """运行graph的链"""
    print("开始生成回复...\n")
    messages = [
        SystemMessage(content="你是一个智能助手，使用专业且准确的语言回复用户的问题，且使用中文进行回复"),
        HumanMessage(content="帮我查一下今明两天的天气")
    ]
    
    # 初始化状态
    initial_state = {"messages": messages, "streamed_output": []}
    
    # stream_mode values的效果
    # async for event in app_graph.astream(initial_state, config={"configurable": {"thread_id": "1"}}, stream_mode="values"):
    #     # print('event------>',event,'\n\n')
    #     if "messages" in event:
    #         event["messages"][-1].pretty_print()
    #     pass
    
    # stream_mode messages的流式效果
    async for event in app_graph.astream(initial_state, stream_mode='messages'):
        # print('event------>',event,'\n\n')
        if isinstance(event, tuple):
            chunk: AIMessageChunk = event[0]
            if chunk.type == 'AIMessageChunk':
                print('event里监听到的流式输出------>',chunk.content,'\n\n')
    
    # print("\n回复完成")
    
    # 展示图形
    try:
        # 导出为PNG
        export_graph_to_png()
    except Exception as e:
        print(f"图表绘制出错: {e}")

# 运行流式输出
if __name__ == "__main__":
    asyncio.run(run_streaming_chain())

AI相关

Langgraph

本博客所有文章除特别声明外，均采用 CC BY-SA 4.0 协议，转载请注明出处！

Langgraph学习3：AI搜索工具的比较上一篇

Langgraph学习1：基础流程下一篇