Langgraph核心组件中的工具调用分为四步:定义工具、绑定工具、生成工具调用(2coin)和执行工具。通过拆解过程,用户可干预每个步骤,尤其在多智能体模式下,可加入人工节点检查调用准确性,提升工具使用的灵活性与控制力。
从工具执行流程图看

有以下四步:
- 第一步:使用@tool修饰符创建工具
- 第二步:绑定工具到模型(注意:并不是所有大模型都支持工具绑定)
- 第三步:工具调用:将自然语言转为正确的参数
- 第四步:工具执行:将工具执行并将结果传递给下一步
根据以上说明,下面根据这四步进行代码示例
第一步:工具定义
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19
| from langchain_core.tools import tool from langgraph.prebuilt import ToolNode
@tool def get_weather(location: str): """调用此函数获取当前天气。""" if location.lower() in ["北京", "深圳"]: return "现在是20度,有雾。" else: return "现在是10度,晴朗。"
@tool def get_coolest_cities(): """获取最冷城市列表""" return "北京,哈尔滨"
tools = [get_weather, get_coolest_cities] tool_node = ToolNode(tools)
|
langgraph提供了低层面的封装,可以直接手动执行工具
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
| from langchain_core.messages import AIMessage
message_with_single_tool_call = AIMessage( content="", tool_calls=[ { "name": "get_weather", "args": {"location": "北京"}, "id": "tool_call_id", "type": "tool_call", } ], )
tool_node.invoke({"messages": [message_with_single_tool_call]})
|
执行结果
1
| {'messages': [ToolMessage(content='现在是20度,有雾。', name='get_weather', tool_call_id='tool_call_id')]}
|
第二步:工具绑定
1 2 3 4 5 6 7 8 9 10 11 12
| from langchain_deepseek import ChatDeepSeek import os
model_with_tools = ChatDeepSeek( model="deepseek-chat", temperature=0, api_key=os.environ.get("DEEPSEEK_API_KEY"), base_url=os.environ.get("DEEPSEEK_API_BASE"), ).bind_tools(tools)
model_with_tools.invoke("深圳的天气如何?").tool_calls
|
第三步:工具调用
返回ToolMessage
1
| tool_node.invoke({"messages": [model_with_tools.invoke("深圳的天气如何?")]})
|
执行结果
1
| {'messages': [ToolMessage(content='现在是20度,有雾。', name='get_weather', tool_call_id='call_00_0vppp9Iy24ryGv9a89hsxVHk')]}
|
第四步:工具执行
在ReAct智能体中执行
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31
| from langgraph.graph import StateGraph, MessagesState, START, END
def should_continue(state: MessagesState): messages = state["messages"] last_message = messages[-1] if last_message.tool_calls: return "tools" return END
def call_model(state: MessagesState): messages = state["messages"] response = model_with_tools.invoke(messages) return {"messages": [response]}
workflow = StateGraph(MessagesState)
workflow.add_node("agent", call_model) workflow.add_node("tools", tool_node)
workflow.add_edge(START, "agent") workflow.add_conditional_edges("agent", should_continue, ["tools", END]) workflow.add_edge("tools", "agent")
app = workflow.compile()
from IPython.display import Image, display
display(Image(app.get_graph().draw_mermaid_png()))
|
查看构成的图

执行
1 2 3 4
| for chunk in app.stream( {"messages": [("human", "深圳的天气如何?")]}, stream_mode="values" ): chunk["messages"][-1].pretty_print()
|
执行结果
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22
| ================================ Human Message =================================
深圳的天气如何? ================================== Ai Message ==================================
我来帮您查询深圳的天气情况。 Tool Calls: get_weather (call_00_a1KyKFbzutzxaD52Q9BHy4ep) Call ID: call_00_a1KyKFbzutzxaD52Q9BHy4ep Args: location: 深圳 ================================= Tool Message ================================= Name: get_weather
现在是20度,有雾。 ================================== Ai Message ==================================
根据查询结果,深圳目前的天气情况是: - 温度:20°C - 天气状况:有雾
这样的天气比较潮湿,建议您外出时注意安全,能见度可能较低。
|