工具
定义工具
ToolNode
是一个 LangChain Runnable,它是一个预构建的 API,可以用于进行工具调用
需要与带有适当的 reducer
的 messages
键使用(如 MessagesState
)
from langchain_core.messages import AIMessage
from langchain_core.tools import tool
from langgraph.prebuilt import ToolNode
定义工具函数
@tool
def get_weather(location: str):
"""获取当前城市的天气状态"""
if location.lower() in ["北京", "beijing"]:
return "今天阳光明媚"
else:
return "今天在下雨"
@tool
def get_coolest_cities() -> list[str]:
"""获取一组最近的城市列表"""
return ["北京"]
tools = [get_weather, get_coolest_cities]
tool_node = ToolNode(tools)
输入到工具节点中
tools = [get_weather, get_coolest_cities]
tool_node = ToolNode(tools)
ToolNode
运行,期望传入的是messages
最后一条消息是带有tool_calls
参数的AIMessage
下面mock:
message = AIMessage( content="", tool_calls=[ { "name": "get_weather", "args": {"location": "北京"}, "id": "tool_call_id", "type": "tool_call", }, { "name": "get_coolest_cities", "args": {}, "id": "tool_call_id_2", "type": "tool_call", }, ] ) tool_node.invoke({ "messages": [message] })
{'messages': [ToolMessage(content='今天阳光明媚', name='get_weather', tool_call_id='tool_call_id'), ToolMessage(content='["北京"]', name='get_coolest_cities', tool_call_id='tool_call_id_2')]}
和chatModel一起使用
通过调用模型的 .bind_tools
来让模型知道有哪些可用的工具
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model_name="gpt-4.1-mini-2025-04-14",
temperature=0
).bind_tools(tools)
llm.invoke("北京天气怎么样").tool_calls
[{'id': 'call_0_d06ffec1-1e88-41c7-960d-4f6ddfc614af', 'function': {'arguments': '{"location":"北京"}', 'name': 'get_weather'}, 'type': 'function', 'index': 0}]
聊天模型生成的 AI 消息已经填充了 tool_calls
,因此可以直接将其传递给 ToolNode
res = tool_node.invoke({"messages": [llm.invoke("北京天气怎么样")]})
{'messages': [ToolMessage(content='今天阳光明媚', name='get_weather', tool_call_id='call_0_191ea70f-9bd5-471d-8654-5293de0ca01f')]}
tools_condition
快速嵌入进图中
tools_condition
是预构建好的工具决策函数
实现原理逻辑:
- 判断最后一条消息是否有
tool_calls
,如果有则返回tools
- 否则返回
END
def tools_condition(state: MessagesState): last_message = state["messages"][-1] if last_message.tool_calls: return "tools" return END
使用例子
from langgraph.graph.message import MessagesState
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from my_openai.deepseekv3 import get_deepseek_v3_client
from langgraph.graph import StateGraph, START
from langgraph.prebuilt import ToolNode, tools_condition
@tool
def wheather(city: str):
"""模拟一个天气搜索工具"""
if "北京" in city.lower() or "Beijing" in city.lower():
return "北京天气晴朗,气温20度"
elif "上海" in city.lower() or "Shanghai" in city.lower():
return "上海多云,气温15度"
else:
return "现在天气很好"
tools = [wheather]
tool_node = ToolNode(tools)
llm = get_deepseek_v3_client().bind_tools(tools)
def chatbot(state: MessagesState):
return { "messages": [llm.invoke(state["messages"])] }
builder = StateGraph(MessagesState)
builder.add_node("chatbot", chatbot)
builder.add_node("tools", tool_node)
builder.add_edge(START, "chatbot")
# 使用 tools_condition 简化工具路由边
builder.add_conditional_edges("chatbot", tools_condition)
builder.add_edge("tools", "chatbot")
graph = builder.compile()
result = graph.invoke({ "messages": [HumanMessage(content="北京天气怎么样?")] })
print(result)
# result
# {
# 'messages': [
# HumanMessage(content='北京天气怎么样?'),
# AIMessage(
# content='',
# additional_kwargs={
# 'tool_calls': [
# {
# 'id': 'call_0_b91082b1-c9c4-4187-9d7a-57553d7044dd',
# 'function': {'arguments': '{"city":"北京"}', 'name': 'wheather'},
# 'type': 'function',
# 'index': 0
# }
# ],
# 'refusal': None
# },
# response_metadata={'token_usage': {'completion_tokens': 19, 'prompt_tokens': 105, 'total_tokens': 124, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 64}, 'prompt_cache_hit_tokens': 64, 'prompt_cache_miss_tokens': 41}, 'model_name': 'deepseek-chat', 'system_fingerprint': 'fp_3d5141a69a_prod0225', 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-4ff91a2d-1c9d-4f60-bd54-57297afa1793-0', tool_calls=[{'name': 'wheather', 'args': {'city': '北京'}, 'id': 'call_0_b91082b1-c9c4-4187-9d7a-57553d7044dd', 'type': 'tool_call'}], usage_metadata={'input_tokens': 105, 'output_tokens': 19, 'total_tokens': 124, 'input_token_details': {'cache_read': 64}, 'output_token_details': {}}
# ),
# ToolMessage(content='北京天气晴朗,气温20度', name='wheather', id='67f660c2-c919-45ec-8bbf-f35b681c0e5e', tool_call_id='call_0_b91082b1-c9c4-4187-9d7a-57553d7044dd'),
# AIMessage(content='北京的天气晴朗,气温为20度,非常适合外出活动!', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 13, 'prompt_tokens': 143, 'total_tokens': 156, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 64}, 'prompt_cache_hit_tokens': 64, 'prompt_cache_miss_tokens': 79}, 'model_name': 'deepseek-chat', 'system_fingerprint': 'fp_3d5141a69a_prod0225', 'finish_reason': 'stop', 'logprobs': None}, id='run-9b22059b-0e19-4803-98a3-b00792e262e9-0', usage_metadata={'input_tokens': 143, 'output_tokens': 13, 'total_tokens': 156, 'input_token_details': {'cache_read': 64}, 'output_token_details': {}})
# ]
# }
访问State、Config、Store
from langgraph.prebuilt import InjectedState, InjectedStore
from typing_extensions import Annotated
from langgraph.store.base import BaseStore
from langchain_core.runnables import RunnableConfig
@tool
def wheather(
# 参数正常传入
city: str,
# 状态
state: Annotated[MessagesState, InjectedState],
# 配置
config: RunnableConfig,
# 数据库
store: Annotated[BaseStore, InjectedStore()],
):
return "好天气"
更新状态
当工具返回 Command
时,可以更新 state
不要忘记
messages
返回一个ToolMessage
class state(MessagesState):
user_info: dict[str, Any]
@tool
def lookup_user_info(tool_call_id: Annotated[str, InjectedToolCallId], config: RunnableConfig):
"""用这个来查找用户信息,以便更好地帮助他们解答问题。"""
user_info = get_user_info(config)
return Command(
update={
# 更新用户信息
"user_info": user_info,
# 更新消息列表
"messages": [ToolMessage(f"用户名称是{user_info['name']}。用户居住在{user_info['location']}", tool_call_id=tool_call_id)]
}
)
工具函数构造器
可以使用 StructuredTool.from_function
构造一个工具函数
from langchain_core.tools import StructuredTool
def get_weather(location: str):
if location.lower() in ["北京", "beijing"]:
return "今天阳光明媚"
else:
return "今天在下雨"
tool = StructuredTool.from_function(
get_weather,
name="get_weather",
description="模拟一个天气搜索工具",
)
利用这个方法可以遍历生成一批相似的工具函数
from langchain_core.tools import StructuredTool import uuid animals = { "rat": "被视为机警应变,善处逆境,子孙繁衍,家业兴旺的象征。有生生不息,繁盛不衰之吉祥寓意", "cattle": "被视为勤奋朴实,诚挚忠厚,忍辱负重,勇武倔强的象征。有勤劳致富,风调雨顺之吉祥寓意", "tiger": "被视为威武勇猛,豪爽正义,文彩华美,气宇轩昂的象征。有辟邪降魅,四方安康之吉祥寓意", "rabbit": "被视为温柔文静,纯洁高雅,机智灵敏,忠厚善良的象征。有自然超脱,长生不老之吉祥寓意", "dragon": "被视为尊贵神圣,志趣高远,能屈能伸,通达旷放的象征。有惩邪镇恶,国泰民安之吉祥寓意" } def create_tool(animal: str): info = animals[animal] def animal_info_tool(): return f"{animal}的寓意是:{info}" return StructuredTool.from_function( animal_info_tool, name=f"meaning_of_{animal}", description=f"关于{animal}的寓意" ) tool_registry = { str(uuid.uuid4()): create_tool(animal) for animal in animals.keys() }