您可能需要为新的运行使用不同的配置重建图。例如,您可能需要根据配置使用不同的图状态或图结构。本指南将展示如何实现这一点。
注意 在大多数情况下,根据配置自定义行为应该由一个单独的图来处理,其中每个节点都可以读取配置并根据其改变其行为。
先决条件
请务必先查阅此操作指南,了解如何设置您的应用程序以进行部署。
定义图
假设您有一个应用程序,其中包含一个简单的图,它调用一个LLM并将响应返回给用户。应用程序文件目录如下所示:
my-app/
|-- requirements.txt
|-- .env
|-- openai_agent.py # code for your graph
其中图在openai_agent.py中定义。
不重建
在标准LangGraph API配置中,服务器使用在openai_agent.py顶级定义的已编译图实例,如下所示:
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, MessageGraph
model = ChatOpenAI(temperature=0)
graph_workflow = MessageGraph()
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如下:
from typing import Annotated
from typing_extensions import TypedDict
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, MessageGraph
from langgraph.graph.state import StateGraph
from langgraph.graph.message import add_messages
from langchain.tools import tool
from langgraph.prebuilt import ToolNode
from langchain.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():
"""Make a simple LLM agent"""
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():
"""Make a tool-calling agent"""
@tool
def add(a: float, b: float):
"""Adds two numbers."""
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
# this is the graph making function that will decide which graph to
# build based on the provided config
def make_graph(config: RunnableConfig):
user_id = config.get("configurable", {}).get("user_id")
# route to different graph state / structure based on the user 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配置文件的更多信息,请参阅此处