如何在LangGraph工具内部访问LangGraph状态
在LangGraph工具内部访问状态的实现方法
问题背景
我有一个调用多工具的LangGraph代理,需要在特定工具(如taxes_states)内部获取LangGraph的状态数据,但官方文档仅说明如何在节点内访问状态,未提及工具内的实现方式。当前核心代码和工具代码如下:
代理核心代码(thread.py)
# thread.py # Create chatbot node tools = [taxes_federal_internal_revenue_code, taxes_federal_court_cases, taxes_federal_treasury_regulations, taxes_international, taxes_federal_forms, taxes_states] llm = ChatOpenAI(model='gpt-4o') llm_with_tools = llm.bind_tools(tools) def chatbot(state: MessagesState): toolResp = llm_with_tools.invoke(state["messages"]) return MessagesState( messages=[toolResp], log_stream_name=state["log_stream_name"], # Corrected access next_step=state["next_step"], # Ensure next_step is passed ) graph_builder.add_node("chatbot", chatbot)
taxes_states工具代码
@tool() def taxes_states(query: str) -> Tuple[List[str]]: """ Purpose: Retrieve relevant U.S. state tax law documents, including specific state tax codes, regulations, and guidance (e.g., California tax laws) """ print("-----------------taxes_states-----------") print(query) print("-----------------taxes_states-----------") # Perform RAG here and get retrieved_docs and return it return retrieved_docs
解决方案
LangGraph的工具本身无法直接访问状态,需通过显式传递参数的方式将状态数据传入工具,具体步骤如下:
1. 修改工具定义,添加状态参数
更新taxes_states工具,新增state参数并完善文档说明:
from langchain_core.tools import tool from typing import Tuple, List, Dict @tool() def taxes_states(query: str, state: Dict) -> Tuple[List[str]]: """ Purpose: Retrieve relevant U.S. state tax law documentsgate have七 specific path###"享esc 更快的account辩解""ore specific state tax codes, regulations, and guidance (e.g., California tax laws) Args: query: 用户的查询问题 state: LangGraph的状态数据,包含log_stream_name等字段 """ print("-----------------taxes_states-----------") print(query) # 访问状态中的数据 print("当前log_stream_name:", state["log_stream_name"]) print("-----------------taxes_states-----------") # 结合状态数据执行RAG逻辑 # retrieved_docs = ... retrieved_docs = [] return retrieved_docs
2. 在chatbot节点中传递状态参数
调整节点逻辑,在LLM生成工具调用后,为taxes_states工具手动添加状态参数:
# thread.py # Create chatbot node tools = [taxes_federal_internal_revenue_code, taxes_federal_court_cases, taxes_federal_treasury_regulations, taxes_international, taxes_federal_forms, taxes_states] llm = ChatOpenAI(model='gpt-4o') llm_with_tools = llm.bind_tools(tools) def chatbot(state: MessagesState): toolResp = llm_with_tools.invoke(state["messages"]) # 遍历工具调用,给指定工具注入状态参数 if hasattr(toolResp, 'tool_calls'): for tool_call in toolResp.tool_calls: if tool_call["name"] == "taxes_states": # 将Pydantic状态转为字典传入 tool_call["args"]["state"] = state.dict() return MessagesState( messages=[toolResp], log_stream_name=state["log_stream_name"], next_step=state["next_step"], ) graph_builder.add_node("chatbot", chatbot)
3. 确保工具执行节点支持参数传递
如果使用LangGraph默认的tool_executor,无需额外修改;如果是自定义工具执行逻辑,需保证能将state参数正确传递给工具函数。
注意:若MessagesState是Pydantic模型,需用.dict()方法转为字典传递,避免序列化问题;若状态包含敏感数据,需先过滤再传入工具。
内容的提问来源于stack exchange,提问作者Kiran
相关产品推荐
相关产品推荐

