运行时重建图¶
你可能需要为新的运行使用不同的配置来重建你的图。例如,你可能需要根据配置使用不同的图状态或图结构。本指南将展示如何实现这一点。
注意
在大多数情况下,基于配置自定义行为应该通过一个图来处理,每个节点都可以读取配置并根据配置改变其行为。
先决条件¶
请先查看这个操作指南,了解如何为部署设置你的应用。
定义图¶
假设你有一个简单的应用程序,该程序调用一个大型语言模型(LLM)并返回响应给用户。该应用的文件目录结构如下所示:
其中图在 openai_agent.py
中定义。
不重新构建¶
在标准的 LangGraph API 配置中,服务器使用的是在 openai_agent.py
文件顶层定义的编译后的图实例,其外观如下:
API Reference: ChatOpenAI | END | START | StateGraph
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, StateGraph, MessagesState
model = ChatOpenAI(temperature=0)
graph_workflow = StateGraph(MessagesState)
graph_workflow.add_node("agent", model)
graph_workflow.add_edge("agent", END)
graph_workflow.add_edge(START, "agent")
agent = graph_workflow.compile()
为了让服务器知道你的图,你需要在 LangGraph API 配置文件(langgraph.json
)中指定包含 CompiledStateGraph
实例变量的路径,例如:
{
"dependencies": ["."],
"graphs": {
"openai_agent": "./openai_agent.py:agent",
},
"env": "./.env"
}
重新构建¶
如果你想让图在每次新运行时根据自定义配置重新构建,需要重写 openai_agent.py
,以提供一个函数,该函数接受一个配置并返回图(或编译后的图)实例。假设我们希望为用户 ID '1' 返回现有的图,而为其他用户返回一个工具调用代理。我们可以这样修改 openai_agent.py
:
API Reference: ChatOpenAI | END | START | StateGraph | add_messages | ToolNode | tool | BaseMessage | RunnableConfig
from typing import Annotated
from typing_extensions import TypedDict
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START
from langgraph.graph.state import StateGraph
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langchain_core.tools import tool
from langchain_core.messages import BaseMessage
from langchain_core.runnables import RunnableConfig
class State(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
model = ChatOpenAI(temperature=0)
def make_default_graph():
"""创建一个简单的 LLM 代理"""
graph_workflow = StateGraph(State)
def call_model(state):
return {"messages": [model.invoke(state["messages"])]}
graph_workflow.add_node("agent", call_model)
graph_workflow.add_edge("agent", END)
graph_workflow.add_edge(START, "agent")
agent = graph_workflow.compile()
return agent
def make_alternative_graph():
"""创建一个工具调用代理"""
@tool
def add(a: float, b: float):
"""将两个数字相加。"""
return a + b
tool_node = ToolNode([add])
model_with_tools = model.bind_tools([add])
def call_model(state):
return {"messages": [model_with_tools.invoke(state["messages"])]}
def should_continue(state: State):
if state["messages"][-1].tool_calls:
return "tools"
else:
return END
graph_workflow = StateGraph(State)
graph_workflow.add_node("agent", call_model)
graph_workflow.add_node("tools", tool_node)
graph_workflow.add_edge("tools", "agent")
graph_workflow.add_edge(START, "agent")
graph_workflow.add_conditional_edges("agent", should_continue)
agent = graph_workflow.compile()
return agent
# 这是图生成函数,它会根据提供的配置决定构建哪个图
def make_graph(config: RunnableConfig):
user_id = config.get("configurable", {}).get("user_id")
# 根据用户 ID 路由到不同的图状态 / 结构
if user_id == "1":
return make_default_graph()
else:
return make_alternative_graph()
最后,你需要在 langgraph.json
中指定图生成函数(make_graph
)的路径:
{
"dependencies": ["."],
"graphs": {
"openai_agent": "./openai_agent.py:make_graph",
},
"env": "./.env"
}
有关 LangGraph API 配置文件的更多信息,请参阅 此处