Jean's Blog

一个专注软件测试开发技术的个人博客

0%

LangGraph之工具调用

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

从工具执行流程图看

image-20250918170343628

有以下四步:

  • 第一步:使用@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) # ToolNode是langgraph中运行工具的节点

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) # Add the tool node to the graph

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()))

查看构成的图

image-20250918172526485

执行

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
- 天气状况:有雾

这样的天气比较潮湿,建议您外出时注意安全,能见度可能较低。