当前位置：首页 > news >正文

langgraph tool如何获取上下文和RunableConfig

news 来源：原创 2024/9/21 14:59:05

在构造langraph工具函数时，常需要使用到运行时的配置RunableConfig以及对话的上下文数据（常命名为State）。文章记录下在工具函数中该如何获取上述配置

获取运行时配置RunableConfig

获取运行时配置，比较简单，只需要将RunableConfig作为函数的一个入参即可，一些确定的元数据常常用RunableConfig进行传递。

from langchain_core.runnables.config import RunnableConfig
from langchain_core.tools import tool@tool
def get_fligh_info(config: RunnableConfig):"""获取乘客的航班信息"""configuration = config.get("configurable", {})passenger_id = configuration.get("passenger_id")print(f"传入的passenger_id:{passenger_id}")return {"name":"老谢", "flight": "AX123"}

传入的passenger_id是123

获取对话上下文

tools中获取对话上下文(state)，需要使用InjectedState来注解。使用该注解后，函数的入参将对模型调用隐藏，意味着该参数不会有大模型的funcation calling生成，而是由state直接注入。可选注入整个state或者state中的某个字段，改写上面的函数

from typing import List
from typing_extensions import Annotated, TypedDictfrom langgraph.graph.message import AnyMessage, add_messages
from langgraph.prebuilt import InjectedState
from langchain_core.runnables import RunnableConfig
from langchain_core.tools import toolclass State(TypedDict):messages: Annotated[list[AnyMessage], add_messages]foo: str@tool
def get_fligh_info(config: RunnableConfig,state: Annotated[dict, InjectedState],foo: Annotated[str, InjectedState("foo")]):"""获取乘客的航班信息"""configuration = config.get("configurable", {})passenger_id = configuration.get("passenger_id")print(f"传入的passenger_id:{passenger_id}")print(f"传入的state:{state}")print(f"传入的foo:{foo}")return {"name":"老谢", "flight": "AX123"}

测试代码

graph定义

定义一个工具用于获取乘客的航班信息，定义一个助手Agent用于工具的调用，同时创建graph如下图

from typing_extensions import Annotated, TypedDictfrom langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import Runnable, RunnableConfig
from langchain_core.tools import tool
from langgraph.graph.message import AnyMessage, add_messages
from langgraph.graph import StateGraph, START
from langgraph.prebuilt import ToolNode, InjectedState, tools_condition
from langgraph.checkpoint.memory import MemorySaverclass State(TypedDict):messages: Annotated[list[AnyMessage], add_messages]foo: str@tool
def get_fligh_info(config: RunnableConfig,state: Annotated[dict, InjectedState],foo: Annotated[str, InjectedState("foo")]):"""获取乘客的航班信息"""configuration = config.get("configurable", {})passenger_id = configuration.get("passenger_id")print(f"传入的passenger_id:{passenger_id}")print(f"传入的state:{state}")print(f"传入的foo:{foo}")return {"name":"老谢", "flight": "AX123"}class Assistant:def __init__(self, runnable: Runnable):self.runnable = runnabledef __call__(self, state: State, config: RunnableConfig):result = self.runnable.invoke(state)return {"messages": result, "foo": "bar"}assistant_system_prompt = "你是航空公司的专业助手，你需要协助用户完成查询操作并解答相关问题"
assistant_prompt = ChatPromptTemplate.from_messages([("system",assistant_system_prompt,),("placeholder", "{messages}"),],
)
assistant_tool = [get_fligh_info]
assistant_runnable = assistant_prompt | llm.bind_tools(assistant_tool)
assistant = Assistant(assistant_runnable)# node定义
builder = StateGraph(State)
builder.add_node("assistant", assistant)
builder.add_node("tools", ToolNode(assistant_tool))# edge定义
builder.add_edge(START, "assistant")
builder.add_conditional_edges("assistant", tools_condition)
builder.add_edge("tools", "assistant")
memory = MemorySaver()
graph = builder.compile(checkpointer=memory)

graph调用

import uuid
thread_id = str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id, "passenger_id": 1234}}
events= graph.stream({"messages": ("user", "帮我查询一下航班信息")}, config, stream_mode="updates")
for event in events:continue

传入的passenger_id:1234
传入的state:{'messages': [HumanMessage(content='帮我查询一下航班信息', id='ceaf9e61-0141-4a4a-af93-2973c4a6f6ec'), AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_dMELZC2UM7gWMatoPwTZQwpx', 'function': {'arguments': '{}', 'name': 'get_fligh_info'}, 'type': 'function'}]}, response_metadata={'token_usage': {'completion_tokens': 29, 'prompt_tokens': 86, 'total_tokens': 115}, 'model_name': 'gpt-4', 'system_fingerprint': 'fp_e49e4201a9', 'finish_reason': 'tool_calls', 'logprobs': None, 'content_filter_results': {}}, id='run-d68e4eaa-cd94-49fa-9200-2360a786a4fe-0', tool_calls=[{'name': 'get_fligh_info', 'args': {}, 'id': 'call_dMELZC2UM7gWMatoPwTZQwpx', 'type': 'tool_call'}], usage_metadata={'input_tokens': 86, 'output_tokens': 29, 'total_tokens': 115})], 'foo': 'bar'}
传入的foo:bar