> ## Documentation Index > Fetch the complete documentation index at: https://cndoc-langchain.site/llms.txt > Use this file to discover all available pages before exploring further. # 使用图 API 本指南演示了 LangGraph 图 API 的基础知识。它将引导你了解[状态](#定义和更新状态)，以及如何组合常见的图结构，如[序列](#创建步骤序列)、[分支](#创建分支)和[循环](#创建和控制循环)。它还涵盖了 LangGraph 的控制功能，包括用于 Map-Reduce 工作流的 \[Send API]\(#map-reduce 和 send api) 以及用于将状态更新与节点间“跳转”相结合的 \[Command API]\(#使用 command 结合控制流和状态更新)。 ## 设置 安装 `langgraph`： ```bash pip theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} pip install -U langgraph ``` ```bash uv theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} uv add langgraph ``` **设置 LangSmith 以获得更好的调试体验** 注册 [LangSmith](https://smith.langchain.com) 以快速发现问题并提升你的 LangGraph 项目的性能。LangSmith 允许你使用跟踪数据来调试、测试和监控使用 LangGraph 构建的 LLM 应用——在[文档](/langsmith/observability)中了解更多入门信息。 ## 定义和更新状态 这里我们展示如何在 LangGraph 中定义和更新[状态](/oss/python/langgraph/graph-api#state)。我们将演示： 1. 如何使用状态来定义图的[模式](/oss/python/langgraph/graph-api#schema) 2. 如何使用[归约器](/oss/python/langgraph/graph-api#reducers)来控制状态更新的处理方式。 ### 定义状态 LangGraph 中的[状态](/oss/python/langgraph/graph-api#state)可以是 `TypedDict`、`Pydantic` 模型或数据类。下面我们将使用 `TypedDict`。有关使用 Pydantic 的详细信息，请参阅\[使用 Pydantic 模型作为图状态]\(#使用 pydantic 模型作为图状态)。 默认情况下，图将具有相同的输入和输出模式，状态决定了该模式。有关如何定义不同的输入和输出模式，请参阅[定义输入和输出模式](#定义输入和输出模式)。 让我们考虑一个使用[消息](/oss/python/langgraph/graph-api#messagesstate)的简单示例。这代表了用于许多 LLM 应用程序的一种通用状态表述。更多详情请参阅我们的[概念页面](/oss/python/langgraph/graph-api#working-with-messages-in-graph-state)。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langchain.messages import AnyMessage from typing_extensions import TypedDict class State(TypedDict): messages: list[AnyMessage] extra_field: int ``` 此状态跟踪一个[消息](https://python.langchain.com/docs/concepts/messages/)对象列表，以及一个额外的整数字段。 ### 更新状态 让我们构建一个包含单个节点的示例图。我们的[节点](/oss/python/langgraph/graph-api#nodes)只是一个读取图状态并对其进行更新的 Python 函数。此函数的第一个参数始终是状态： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langchain.messages import AIMessage def node(state: State): messages = state["messages"] new_message = AIMessage("Hello!") return {"messages": messages + [new_message], "extra_field": 10} ``` 此节点只是向我们的消息列表追加一条消息，并填充一个额外的字段。 节点应直接返回对状态的更新，而不是修改状态。 接下来，让我们定义一个包含此节点的简单图。我们使用 [`StateGraph`](/oss/python/langgraph/graph-api#stategraph) 来定义一个操作此状态的图。然后我们使用 [`add_node`](/oss/python/langgraph/graph-api#nodes) 来填充我们的图。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.graph import StateGraph builder = StateGraph(State) builder.add_node(node) builder.set_entry_point("node") graph = builder.compile() ``` LangGraph 提供了内置的可视化工具来查看你的图。让我们检查一下我们的图。有关可视化的详细信息，请参阅[可视化你的图](#可视化你的图)。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from IPython.display import Image, display display(Image(graph.get_graph().draw_mermaid_png())) ``` 具有单个节点的简单图 在这种情况下，我们的图只执行一个节点。让我们进行一个简单的调用： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langchain.messages import HumanMessage result = graph.invoke({"messages": [HumanMessage("Hi")]}) result ``` ``` {'messages': [HumanMessage(content='Hi'), AIMessage(content='Hello!')], 'extra_field': 10} ``` 请注意： * 我们通过更新状态的单个键来启动调用。 * 我们在调用结果中接收到整个状态。 为了方便，我们经常通过漂亮打印来检查[消息对象](https://python.langchain.com/docs/concepts/messages/)的内容： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} for message in result["messages"]: message.pretty_print() ``` ``` ================================ Human Message ================================ Hi ================================== Ai Message ================================== Hello! ``` ### 使用归约器处理状态更新 状态中的每个键都可以有自己的独立[归约器](/oss/python/langgraph/graph-api#reducers)函数，该函数控制如何应用来自节点的更新。如果没有明确指定归约器函数，则假定对该键的所有更新都应覆盖它。 对于 `TypedDict` 状态模式，我们可以通过用归约器函数注解状态的相应字段来定义归约器。 在前面的示例中，我们的节点通过向其追加消息来更新状态中的 `"messages"` 键。下面，我们为这个键添加一个归约器，以便更新自动追加： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from typing_extensions import Annotated def add(left, right): """也可以从内置的 `operator` 导入 `add`。""" return left + right class State(TypedDict): messages: Annotated[list[AnyMessage], add] # [!code highlight] extra_field: int ``` 现在我们的节点可以简化： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} def node(state: State): new_message = AIMessage("Hello!") return {"messages": [new_message], "extra_field": 10} # [!code highlight] ``` ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.graph import START graph = StateGraph(State).add_node(node).add_edge(START, "node").compile() result = graph.invoke({"messages": [HumanMessage("Hi")]}) for message in result["messages"]: message.pretty_print() ``` ``` ================================ Human Message ================================ Hi ================================== Ai Message ================================== Hello! ``` #### MessagesState 实际上，更新消息列表还有额外的考虑因素： * 我们可能希望更新状态中的现有消息。 * 我们可能希望接受[消息格式](/oss/python/langgraph/graph-api#using-messages-in-your-graph)的简写，例如 [OpenAI 格式](https://python.langchain.com/docs/concepts/messages/#openai-format)。 LangGraph 包含一个内置的归约器 [`add_messages`](https://reference.langchain.com/python/langgraph/graph/message/add_messages)，它处理这些考虑因素： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.graph.message import add_messages class State(TypedDict): messages: Annotated[list[AnyMessage], add_messages] # [!code highlight] extra_field: int def node(state: State): new_message = AIMessage("Hello!") return {"messages": [new_message], "extra_field": 10} graph = StateGraph(State).add_node(node).set_entry_point("node").compile() ``` ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} input_message = {"role": "user", "content": "Hi"} # [!code highlight] result = graph.invoke({"messages": [input_message]}) for message in result["messages"]: message.pretty_print() ``` ``` ================================ Human Message ================================ Hi ================================== Ai Message ================================== Hello! ``` 这是涉及[聊天模型](https://python.langchain.com/docs/concepts/chat_models/)的应用程序的一种通用状态表示。为了方便，LangGraph 包含一个预构建的 `MessagesState`，这样我们可以有： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.graph import MessagesState class State(MessagesState): extra_field: int ``` ### 使用 `Overwrite` 绕过归约器 在某些情况下，你可能希望绕过归约器并直接覆盖状态值。LangGraph 提供了 [`Overwrite`](https://reference.langchain.com/python/langgraph/types/) 类型来实现此目的。当节点返回一个用 `Overwrite` 包装的值时，归约器将被绕过，通道将直接设置为该值。 当你想要重置或替换累积的状态，而不是将其与现有值合并时，这很有用。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.graph import StateGraph, START, END from langgraph.types import Overwrite from typing_extensions import Annotated, TypedDict import operator class State(TypedDict): messages: Annotated[list, operator.add] def add_message(state: State): return {"messages": ["first message"]} def replace_messages(state: State): # 绕过归约器并替换整个消息列表 return {"messages": Overwrite(["replacement message"])} builder = StateGraph(State) builder.add_node("add_message", add_message) builder.add_node("replace_messages", replace_messages) builder.add_edge(START, "add_message") builder.add_edge("add_message", "replace_messages") builder.add_edge("replace_messages", END) graph = builder.compile() result = graph.invoke({"messages": ["initial"]}) print(result["messages"]) ``` ``` ['replacement message'] ``` 你也可以使用带有特殊键 `"__overwrite__"` 的 JSON 格式： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} def replace_messages(state: State): return {"messages": {"__overwrite__": ["replacement message"]}} ``` 当节点并行执行时，在给定的超级步骤中，只有一个节点可以使用 `Overwrite` 来处理同一个状态键。如果多个节点尝试在同一个超级步骤中覆盖同一个键，将引发 `InvalidUpdateError`。 ### 定义输入和输出模式 默认情况下，`StateGraph` 使用单一模式操作，所有节点都期望使用该模式进行通信。但是，也可以为图定义不同的输入和输出模式。 当指定了不同的模式时，内部模式仍将用于节点之间的通信。输入模式确保提供的输入符合预期的结构，而输出模式根据定义的输出模式过滤内部数据，仅返回相关信息。 下面，我们将看到如何定义不同的输入和输出模式。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.graph import StateGraph, START, END from typing_extensions import TypedDict # 定义输入的模式 class InputState(TypedDict): question: str # 定义输出的模式 class OutputState(TypedDict): answer: str # 定义整体模式，结合输入和输出 class OverallState(InputState, OutputState): pass # 定义处理输入并生成答案的节点 def answer_node(state: InputState): # 示例答案和一个额外的键 return {"answer": "bye", "question": state["question"]} # 构建指定了输入和输出模式的图 builder = StateGraph(OverallState, input_schema=InputState, output_schema=OutputState) builder.add_node(answer_node) # 添加答案节点 builder.add_edge(START, "answer_node") # 定义起始边 builder.add_edge("answer_node", END) # 定义结束边 graph = builder.compile() # 编译图 # 使用输入调用图并打印结果 print(graph.invoke({"question": "hi"})) ``` ``` {'answer': 'bye'} ``` 请注意，调用的输出仅包含输出模式。 ### 在节点之间传递私有状态 在某些情况下，你可能希望节点交换对中间逻辑至关重要但不需要成为图主要模式一部分的信息。这些私有数据与图的整体输入/输出无关，只应在特定节点之间共享。 下面，我们将创建一个由三个节点（node\_1、node\_2 和 node\_3）组成的示例顺序图，其中私有数据在前两个步骤（node\_1 和 node\_2）之间传递，而第三个步骤（node\_3）只能访问公共的整体状态。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.graph import StateGraph, START, END from typing_extensions import TypedDict # 图的整体状态（这是节点间共享的公共状态） class OverallState(TypedDict): a: str # node_1 的输出包含不属于整体状态的私有数据 class Node1Output(TypedDict): private_data: str # 私有数据仅在 node_1 和 node_2 之间共享 def node_1(state: OverallState) -> Node1Output: output = {"private_data": "set by node_1"} print(f"Entered node `node_1`:\n\tInput: {state}.\n\tReturned: {output}") return output # Node 2 输入仅请求 node_1 之后可用的私有数据 class Node2Input(TypedDict): private_data: str def node_2(state: Node2Input) -> OverallState: output = {"a": "set by node_2"} print(f"Entered node `node_2`:\n\tInput: {state}.\n\tReturned: {output}") return output # Node 3 只能访问整体状态（无法访问来自 node_1 的私有数据） def node_3(state: OverallState) -> OverallState: output = {"a": "set by node_3"} print(f"Entered node `node_3`:\n\tInput: {state}.\n\tReturned: {output}") return output # 按顺序连接节点 # node_2 接受来自 node_1 的私有数据，而 # node_3 看不到私有数据。 builder = StateGraph(OverallState).add_sequence([node_1, node_2, node_3]) builder.add_edge(START, "node_1") graph = builder.compile() # 使用初始状态调用图 response = graph.invoke( { "a": "set at start", } ) print() print(f"Output of graph invocation: {response}") ``` ``` Entered node `node_1`: Input: {'a': 'set at start'}. Returned: {'private_data': 'set by node_1'} Entered node `node_2`: Input: {'private_data': 'set by node_1'}. Returned: {'a': 'set by node_2'} Entered node `node_3`: Input: {'a': 'set by node_2'}. Returned: {'a': 'set by node_3'} Output of graph invocation: {'a': 'set by node_3'} ``` ### 使用 pydantic 模型作为图状态 [StateGraph](https://langchain-ai.github.io/langgraph/reference/graphs.md#langgraph.graph.StateGraph) 在初始化时接受一个 [`state_schema`](https://reference.langchain.com/python/langchain/middleware/#langchain.agents.middleware.AgentMiddleware.state_schema) 参数，该参数指定图中节点可以访问和更新的状态的“形状”。 在我们的示例中，我们通常使用 Python 原生的 `TypedDict` 或 [`dataclass`](https://docs.python.org/3/library/dataclasses.html) 作为 `state_schema`，但 [`state_schema`](https://reference.langchain.com/python/langchain/middleware/#langchain.agents.middleware.AgentMiddleware.state_schema) 可以是任何[类型](https://docs.python.org/3/library/stdtypes.html#type-objects)。 这里，我们将看到如何使用 [Pydantic BaseModel](https://docs.pydantic.dev/latest/api/base_model/) 作为 [`state_schema`](https://reference.langchain.com/python/langchain/middleware/#langchain.agents.middleware.AgentMiddleware.state_schema) 来添加对**输入**的运行时验证。 **已知限制** \* 目前，图的输出**不会**是 pydantic 模型的实例。 \* 运行时验证仅发生在图第一个节点的输入上，而不是后续节点或输出上。 \* pydantic 的验证错误跟踪不会显示错误发生在哪个节点。 \* Pydantic 的递归验证可能很慢。对于性能敏感的应用程序，你可能需要考虑改用 `dataclass`。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.graph import StateGraph, START, END from typing_extensions import TypedDict from pydantic import BaseModel # 图的整体状态（这是节点间共享的公共状态） class OverallState(BaseModel): a: str def node(state: OverallState): return {"a": "goodbye"} # 构建状态图 builder = StateGraph(OverallState) builder.add_node(node) # node_1 是第一个节点 builder.add_edge(START, "node") # 从 node_1 开始图 builder.add_edge("node", END) # 在 node_1 之后结束图 graph = builder.compile() # 使用有效输入测试图 graph.invoke({"a": "hello"}) ``` 使用**无效**输入调用图 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} try: graph.invoke({"a": 123}) # 应该是字符串 except Exception as e: print("An exception was raised because `a` is an integer rather than a string.") print(e) ``` ``` An exception was raised because `a` is an integer rather than a string. 1 validation error for OverallState a Input should be a valid string [type=string_type, input_value=123, input_type=int] For further information visit https://errors.pydantic.dev/2.9/v/string_type ``` 有关 Pydantic 模型状态的更多功能，请参见下文： 当使用 Pydantic 模型作为状态模式时，了解序列化的工作方式非常重要，特别是在以下情况下： * 将 Pydantic 对象作为输入传递 * 接收图的输出 * 处理嵌套的 Pydantic 模型 让我们看看这些行为的实际应用。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.graph import StateGraph, START, END from pydantic import BaseModel class NestedModel(BaseModel): value: str class ComplexState(BaseModel): text: str count: int nested: NestedModel def process_node(state: ComplexState): # 节点接收一个经过验证的 Pydantic 对象 print(f"Input state type: {type(state)}") print(f"Nested type: {type(state.nested)}") # 返回字典更新 return {"text": state.text + " processed", "count": state.count + 1} # 构建图 builder = StateGraph(ComplexState) builder.add_node("process", process_node) builder.add_edge(START, "process") builder.add_edge("process", END) graph = builder.compile() # 创建一个 Pydantic 实例作为输入 input_state = ComplexState(text="hello", count=0, nested=NestedModel(value="test")) print(f"Input object type: {type(input_state)}") # 使用 Pydantic 实例调用图 result = graph.invoke(input_state) print(f"Output type: {type(result)}") print(f"Output content: {result}") # 如果需要，转换回 Pydantic 模型 output_model = ComplexState(**result) print(f"Converted back to Pydantic: {type(output_model)}") ``` Pydantic 对某些数据类型执行运行时类型强制转换。这可能很有帮助，但如果你没有意识到，也可能导致意外行为。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.graph import StateGraph, START, END from pydantic import BaseModel class CoercionExample(BaseModel): # Pydantic 会将字符串数字强制转换为整数 number: int # Pydantic 会将字符串布尔值解析为 bool flag: bool def inspect_node(state: CoercionExample): print(f"number: {state.number} (type: {type(state.number)})") print(f"flag: {state.flag} (type: {type(state.flag)})") return {} builder = StateGraph(CoercionExample) builder.add_node("inspect", inspect_node) builder.add_edge(START, "inspect") builder.add_edge("inspect", END) graph = builder.compile() # 演示将被转换的字符串输入的强制转换 result = graph.invoke({"number": "42", "flag": "true"}) # 这将因验证错误而失败 try: graph.invoke({"number": "not-a-number", "flag": "true"}) except Exception as e: print(f"\nExpected validation error: {e}") ``` 在状态模式中使用 LangChain 消息类型时，序列化有一些重要的考虑因素。你应该使用 `AnyMessage`（而不是 `BaseMessage`）来正确序列化/反序列化通过网络传输的消息对象。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.graph import StateGraph, START, END from pydantic import BaseModel from langchain.messages import HumanMessage, AIMessage, AnyMessage from typing import List class ChatState(BaseModel): messages: List[AnyMessage] context: str def add_message(state: ChatState): return {"messages": state.messages + [AIMessage(content="Hello there!")]} builder = StateGraph(ChatState) builder.add_node("add_message", add_message) builder.add_edge(START, "add_message") builder.add_edge("add_message", END) graph = builder.compile() # 创建带有消息的输入 initial_state = ChatState( messages=[HumanMessage(content="Hi")], context="Customer support chat" ) result = graph.invoke(initial_state) print(f"Output: {result}") # 转换回 Pydantic 模型以查看消息类型 output_model = ChatState(**result) for i, msg in enumerate(output_model.messages): print(f"Message {i}: {type(msg).__name__} - {msg.content}") ``` ## 添加运行时配置 有时你希望在调用图时能够配置它。例如，你可能希望能够在运行时指定使用哪个 LLM 或系统提示，*而不将这些参数污染到图状态中*。 要添加运行时配置： 1. 为你的配置指定一个模式 2. 将配置添加到节点或条件边的函数签名中 3. 将配置传递到图中。 请参见下面的简单示例： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.graph import END, StateGraph, START from langgraph.runtime import Runtime from typing_extensions import TypedDict # 1. 指定配置模式 class ContextSchema(TypedDict): my_runtime_value: str # 2. 定义一个在节点中访问配置的图 class State(TypedDict): my_state_value: str def node(state: State, runtime: Runtime[ContextSchema]): # [!code highlight] if runtime.context["my_runtime_value"] == "a": # [!code highlight] return {"my_state_value": 1} elif runtime.context["my_runtime_value"] == "b": # [!code highlight] return {"my_state_value": 2} else: raise ValueError("Unknown values.") builder = StateGraph(State, context_schema=ContextSchema) # [!code highlight] builder.add_node(node) builder.add_edge(START, "node") builder.add_edge("node", END) graph = builder.compile() # 3. 在运行时传入配置： print(graph.invoke({}, context={"my_runtime_value": "a"})) # [!code highlight] print(graph.invoke({}, context={"my_runtime_value": "b"})) # [!code highlight] ``` ``` {'my_state_value': 1} {'my_state_value': 2} ``` 下面我们演示一个实际示例，我们在运行时配置使用哪个 LLM。我们将同时使用 OpenAI 和 Anthropic 模型。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from dataclasses import dataclass from langchain.chat_models import init_chat_model from langgraph.graph import MessagesState, END, StateGraph, START from langgraph.runtime import Runtime from typing_extensions import TypedDict @dataclass class ContextSchema: model_provider: str = "anthropic" MODELS = { "anthropic": init_chat_model("claude-haiku-4-5-20251001"), "openai": init_chat_model("gpt-5.4-mini"), } def call_model(state: MessagesState, runtime: Runtime[ContextSchema]): model = MODELS[runtime.context.model_provider] response = model.invoke(state["messages"]) return {"messages": [response]} builder = StateGraph(MessagesState, context_schema=ContextSchema) builder.add_node("model", call_model) builder.add_edge(START, "model") builder.add_edge("model", END) graph = builder.compile() # 用法 input_message = {"role": "user", "content": "hi"} # 没有配置时，使用默认值（Anthropic） response_1 = graph.invoke({"messages": [input_message]}, context=ContextSchema())["messages"][-1] # 或者，可以设置 OpenAI response_2 = graph.invoke({"messages": [input_message]}, context={"model_provider": "openai"})["messages"][-1] print(response_1.response_metadata["model_name"]) print(response_2.response_metadata["model_name"]) ``` ``` claude-haiku-4-5-20251001 gpt-5.4-mini ``` 下面我们演示一个实际示例，我们在运行时配置两个参数：LLM 和要使用的系统消息。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from dataclasses import dataclass from langchain.chat_models import init_chat_model from langchain.messages import SystemMessage from langgraph.graph import END, MessagesState, StateGraph, START from langgraph.runtime import Runtime from typing_extensions import TypedDict @dataclass class ContextSchema: model_provider: str = "anthropic" system_message: str | None = None MODELS = { "anthropic": init_chat_model("claude-haiku-4-5-20251001"), "openai": init_chat_model("gpt-5.4-mini"), } def call_model(state: MessagesState, runtime: Runtime[ContextSchema]): model = MODELS[runtime.context.model_provider] messages = state["messages"] if (system_message := runtime.context.system_message): messages = [SystemMessage(system_message)] + messages response = model.invoke(messages) return {"messages": [response]} builder = StateGraph(MessagesState, context_schema=ContextSchema) builder.add_node("model", call_model) builder.add_edge(START, "model") builder.add_edge("model", END) graph = builder.compile() # 用法 input_message = {"role": "user", "content": "hi"} response = graph.invoke({"messages": [input_message]}, context={"model_provider": "openai", "system_message": "Respond in Italian."}) for message in response["messages"]: message.pretty_print() ``` ``` ================================ Human Message ================================ hi ================================== Ai Message ================================== Ciao! Come posso aiutarti oggi? ``` ## 添加重试策略 有许多用例你可能希望节点具有自定义的重试策略，例如，如果你正在调用 API、查询数据库或调用 LLM 等。LangGraph 允许你为节点添加重试策略。 要配置重试策略，请将 `retry_policy` 参数传递给 [`add_node`](https://reference.langchain.com/python/langgraph/graph/state/StateGraph/add_node)。`retry_policy` 参数接受一个 `RetryPolicy` 命名元组对象。下面我们使用默认参数实例化一个 `RetryPolicy` 对象并将其与节点关联： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.types import RetryPolicy builder.add_node( "node_name", node_function, retry_policy=RetryPolicy(), ) ``` 默认情况下，`retry_on` 参数使用 `default_retry_on` 函数，它会重试除以下异常之外的任何异常： * `ValueError` * `TypeError` * `ArithmeticError` * `ImportError` * `LookupError` * `NameError` * `SyntaxError` * `RuntimeError` * `ReferenceError` * `StopIteration` * `StopAsyncIteration` * `OSError` 此外，对于来自流行 HTTP 请求库（如 `requests` 和 `httpx`）的异常，它仅在 5xx 状态码时重试。 考虑一个我们从 SQL 数据库读取的示例。下面我们向节点传递两个不同的重试策略： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} import sqlite3 from typing_extensions import TypedDict from langchain.chat_models import init_chat_model from langgraph.graph import END, MessagesState, StateGraph, START from langgraph.types import RetryPolicy from langchain_community.utilities import SQLDatabase from langchain.messages import AIMessage db = SQLDatabase.from_uri("sqlite:///:memory:") model = init_chat_model("claude-haiku-4-5-20251001") def query_database(state: MessagesState): query_result = db.run("SELECT * FROM Artist LIMIT 10;") return {"messages": [AIMessage(content=query_result)]} def call_model(state: MessagesState): response = model.invoke(state["messages"]) return {"messages": [response]} # 定义一个新图 builder = StateGraph(MessagesState) builder.add_node( "query_database", query_database, retry_policy=RetryPolicy(retry_on=sqlite3.OperationalError), ) builder.add_node("model", call_model, retry_policy=RetryPolicy(max_attempts=5)) builder.add_edge(START, "model") builder.add_edge("model", "query_database") builder.add_edge("query_database", END) graph = builder.compile() ``` ## 配置节点超时 [`add_node`](https://reference.langchain.com/python/langgraph/graph/state/StateGraph/add_node) 上的 `timeout=` 参数限制单个异步节点尝试可能运行的时间。传递一个数字（秒）、一个 `timedelta` 或一个 [`TimeoutPolicy`](https://reference.langchain.com/python/langgraph/types/TimeoutPolicy) 以对运行和空闲超时进行更精细的控制。当超过限制时，LangGraph 会引发 [`NodeTimeoutError`](https://reference.langchain.com/python/langgraph/errors/NodeTimeoutError)，并让重试策略决定是否重试。 每个节点的超时需要 `langgraph>=1.2`，目前处于 alpha 阶段。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.types import TimeoutPolicy builder.add_node( "call_model", call_model, timeout=TimeoutPolicy(run_timeout=120, idle_timeout=30), ) ``` 有关完整的超时生命周期、空闲超时刷新源和 `runtime.heartbeat()`，请参阅[容错](/oss/python/langgraph/fault-tolerance#timeouts)。 ## 处理节点错误 [`add_node`](https://reference.langchain.com/python/langgraph/graph/state/StateGraph/add_node) 上的 `error_handler=` 参数注册一个函数，该函数在节点失败且所有重试耗尽后运行。处理程序接收当前状态和一个带有失败上下文的类型化 [`NodeError`](https://reference.langchain.com/python/langgraph/errors/NodeError)，并且可以通过 [`Command`](https://reference.langchain.com/python/langgraph/types/Command) 路由到恢复分支： 节点级错误处理程序需要 `langgraph>=1.2`，目前处于 alpha 阶段。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.errors import NodeError from langgraph.types import Command, RetryPolicy def payment_error_handler(state: State, error: NodeError) -> Command: return Command( update={"status": f"compensated: {error.error}"}, goto="finalize", ) builder.add_node( "charge_payment", charge_payment, retry_policy=RetryPolicy(max_attempts=3, retry_on=ConnectionError), error_handler=payment_error_handler, ) ``` 有关补偿模式和 `Command` 路由，请参阅[容错](/oss/python/langgraph/fault-tolerance#error-handling)。 ### 在节点内访问执行信息 你可以通过 `runtime.execution_info` 访问执行标识和重试信息。这提供了线程、运行和检查点标识符以及重试状态，而无需直接从 `config` 读取。 | 属性 | 类型 | 描述 | | ------------------------- | --------------- | -------------------------------------------- | | `thread_id` | `str \| None` | 当前线程的线程 ID。没有检查点器时为 `None`。 | | `run_id` | `str \| None` | 当前运行的运行 ID。未在配置中提供时为 `None`。 | | `checkpoint_id` | `str` | 当前执行的检查点 ID。 | | `checkpoint_ns` | `str` | 当前执行的检查点命名空间。 | | `task_id` | `str` | 当前执行的任务 ID。 | | `node_attempt` | `int` | 当前执行尝试次数（从 1 开始）。第一次尝试为 `1`，第一次重试为 `2`，依此类推。 | | `node_first_attempt_time` | `float \| None` | 第一次尝试开始时的 Unix 时间戳（秒）。在重试期间保持不变。 | #### 访问线程和运行 ID 使用 `execution_info` 在节点内访问线程 ID、运行 ID 和其他标识字段： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.graph import StateGraph, START, END from langgraph.runtime import Runtime from typing_extensions import TypedDict class State(TypedDict): result: str def my_node(state: State, runtime: Runtime): info = runtime.execution_info print(f"Thread: {info.thread_id}, Run: {info.run_id}") # [!code highlight] return {"result": "done"} builder = StateGraph(State) builder.add_node("my_node", my_node) builder.add_edge(START, "my_node") builder.add_edge("my_node", END) graph = builder.compile() ``` #### 根据重试状态调整行为 当节点具有重试策略时，使用 `execution_info` 检查当前尝试次数，并在第一次尝试失败后切换到回退： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.graph import StateGraph, START, END from langgraph.runtime import Runtime from langgraph.types import RetryPolicy from typing_extensions import TypedDict class State(TypedDict): result: str def my_node(state: State, runtime: Runtime): info = runtime.execution_info if info.node_attempt > 1: # [!code highlight] # 在重试时使用回退 return {"result": call_fallback_api()} return {"result": call_primary_api()} builder = StateGraph(State) builder.add_node("my_node", my_node, retry_policy=RetryPolicy(max_attempts=3)) builder.add_edge(START, "my_node") builder.add_edge("my_node", END) graph = builder.compile() ``` 即使没有重试策略，`execution_info` 也可在 `Runtime` 对象上使用——`node_attempt` 默认为 `1`，`node_first_attempt_time` 设置为节点开始执行的时间。 ### 在节点内访问服务器信息 当你的图在 LangGraph Server 上运行时，你可以通过 `runtime.server_info` 访问服务器特定的元数据。这提供了助手 ID、图 ID 和经过身份验证的用户，而无需直接从配置元数据或可配置键读取。 | 属性 | 类型 | 描述 | | -------------- | ------------------ | -------------------------------------------------- | | `assistant_id` | `str` | 当前部署的助手 ID。 | | `graph_id` | `str` | 当前部署的图 ID。 | | `user` | `BaseUser \| None` | 经过身份验证的用户（如果配置了[自定义身份验证](/langsmith/custom-auth)）。 | ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.graph import StateGraph, START, END from langgraph.runtime import Runtime from typing_extensions import TypedDict class State(TypedDict): result: str def my_node(state: State, runtime: Runtime): server = runtime.server_info if server is not None: print(f"Assistant: {server.assistant_id}, Graph: {server.graph_id}") # [!code highlight] if server.user is not None: print(f"User: {server.user.identity}") return {"result": "done"} builder = StateGraph(State) builder.add_node("my_node", my_node) builder.add_edge(START, "my_node") builder.add_edge("my_node", END) graph = builder.compile() ``` 当图未在 LangGraph Server 上运行时（例如，在本地开发或测试期间），`server_info` 为 `None`。 需要 `deepagents>=0.5.0`（或 `langgraph>=1.1.5`）才能使用 `runtime.execution_info` 和 `runtime.server_info`。 ### 在节点内访问排空状态 当请求[优雅关闭](/oss/python/langgraph/durable-execution#graceful-shutdown)时，`runtime.drain_requested` 为 `True`。在节点内读取此值，以便在下一个超级步骤边界之前跳过昂贵的工作： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.runtime import Runtime def my_node(state: State, runtime: Runtime) -> State: if runtime.drain_requested: # [!code highlight] return {"status": "skipped", "reason": runtime.drain_reason} return {"status": do_work()} ``` | 属性 | 类型 | 描述 | | ----------------- | ------------- | ------------------------------------------------- | | `drain_requested` | `bool` | 如果已为此运行调用 `RunControl.request_drain()`，则为 `True`。 | | `drain_reason` | `str \| None` | 传递给 `request_drain()` 的原因字符串，如果未请求排空则为 `None`。 | 需要 `langgraph>=1.2`，目前处于 alpha 阶段。有关完整的 `RunControl` API，请参阅[优雅关闭](/oss/python/langgraph/durable-execution#graceful-shutdown)。 ## 添加节点缓存 节点缓存对于希望避免重复操作的情况很有用，例如在执行昂贵操作（无论是时间还是成本方面）时。LangGraph 允许你为图中的节点添加个性化的缓存策略。 要配置缓存策略，请将 `cache_policy` 参数传递给 [`add_node`](https://reference.langchain.com/python/langgraph/graph/state/StateGraph/add_node) 函数。在以下示例中，一个 [`CachePolicy`](https://reference.langchain.com/python/langgraph/types/CachePolicy) 对象被实例化，其生存时间为 120 秒，并使用默认的 `key_func` 生成器。然后将其与节点关联： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.types import CachePolicy builder.add_node( "node_name", node_function, cache_policy=CachePolicy(ttl=120), ) ``` 然后，要为图启用节点级缓存，请在编译图时设置 `cache` 参数。下面的示例使用 `InMemoryCache` 来设置具有内存缓存的图，但 `SqliteCache` 也可用。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.cache.memory import InMemoryCache graph = builder.compile(cache=InMemoryCache()) ``` ## 创建步骤序列 **前提条件** 本指南假设你熟悉上面关于[状态](#定义和更新状态)的部分。 这里我们演示如何构建一个简单的步骤序列。我们将展示： 1. 如何构建顺序图 2. 用于构建类似图的内置简写。 要添加一系列节点，我们使用 [graph](/oss/python/langgraph/graph-api#stategraph) 的 [`add_node`](https://reference.langchain.com/python/langgraph/graph/state/StateGraph/add_node) 和 [`add_edge`](https://reference.langchain.com/python/langgraph/pregel/_draw/add_edge) 方法： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.graph import START, StateGraph builder = StateGraph(State) # 添加节点 builder.add_node(step_1) builder.add_node(step_2) builder.add_node(step_3) # 添加边 builder.add_edge(START, "step_1") builder.add_edge("step_1", "step_2") builder.add_edge("step_2", "step_3") ``` 我们也可以使用内置的简写 `.add_sequence`： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} builder = StateGraph(State).add_sequence([step_1, step_2, step_3]) builder.add_edge(START, "step_1") ``` LangGraph 使得为你的应用程序添加底层持久化层变得容易。 这允许在节点执行之间对状态进行检查点，因此你的 LangGraph 节点控制： * 状态更新如何被[检查点化](/oss/python/langgraph/persistence) * [Human in the Loop](/oss/python/langgraph/interrupts)工作流中的中断如何恢复 * 我们如何使用 LangGraph 的[时间旅行](/oss/python/langgraph/use-time-travel)功能“回退”和分支执行 它们还决定了执行步骤如何被[流式传输](/oss/python/langgraph/streaming)，以及你的应用程序如何使用 [Studio](/langsmith/studio) 进行可视化和调试。 让我们演示一个端到端的示例。我们将创建三个步骤的序列： 1. 在状态的一个键中填充一个值 2. 更新相同的值 3. 填充一个不同的值 让我们首先定义我们的[状态](/oss/python/langgraph/graph-api#state)。这控制着[图的模式](/oss/python/langgraph/graph-api#schema)，也可以指定如何应用更新。有关更多详细信息，请参阅[使用归约器处理状态更新](#使用归约器处理状态更新)。 在我们的例子中，我们将只跟踪两个值： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from typing_extensions import TypedDict class State(TypedDict): value_1: str value_2: int ``` 我们的[节点](/oss/python/langgraph/graph-api#nodes)只是读取图状态并对其进行更新的 Python 函数。此函数的第一个参数始终是状态： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} def step_1(state: State): return {"value_1": "a"} def step_2(state: State): current_value_1 = state["value_1"] return {"value_1": f"{current_value_1} b"} def step_3(state: State): return {"value_2": 10} ``` 请注意，当向状态发出更新时，每个节点只需指定它希望更新的键的值。 默认情况下，这将**覆盖**相应键的值。你也可以使用[归约器](/oss/python/langgraph/graph-api#reducers)来控制更新的处理方式——例如，你可以将连续的更新追加到一个键中。有关更多详细信息，请参阅[使用归约器处理状态更新](#使用归约器处理状态更新)。 最后，我们定义图。我们使用 [StateGraph](/oss/python/langgraph/graph-api#stategraph) 来定义一个操作此状态的图。 然后我们将使用 [`add_node`](/oss/python/langgraph/graph-api#messagesstate) 和 [`add_edge`](/oss/python/langgraph/graph-api#edges) 来填充我们的图并定义其控制流。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.graph import START, StateGraph builder = StateGraph(State) # 添加节点 builder.add_node(step_1) builder.add_node(step_2) builder.add_node(step_3) # 添加边 builder.add_edge(START, "step_1") builder.add_edge("step_1", "step_2") builder.add_edge("step_2", "step_3") ``` **指定自定义名称** 你可以使用 [`add_node`](https://reference.langchain.com/python/langgraph/graph/state/StateGraph/add_node) 为节点指定自定义名称： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} builder.add_node("my_node", step_1) ``` 请注意： * [`add_edge`](https://reference.langchain.com/python/langgraph/pregel/_draw/add_edge) 接受节点的名称，对于函数，默认为 `node.__name__`。 * 我们必须指定图的入口点。为此，我们添加一条带有 [START 节点](/oss/python/langgraph/graph-api#start-node)的边。 * 当没有更多节点可执行时，图停止。 接下来我们[编译](/oss/python/langgraph/graph-api#compiling-your-graph)我们的图。这会对图的结构进行一些基本检查（例如，识别孤立节点）。如果我们通过[检查点器](/oss/python/langgraph/persistence)为应用程序添加持久化，它也会在这里传入。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} graph = builder.compile() ``` LangGraph 提供了内置的可视化工具来查看你的图。让我们检查一下我们的序列。有关可视化的详细信息，请参阅[可视化你的图](#可视化你的图)。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from IPython.display import Image, display display(Image(graph.get_graph().draw_mermaid_png())) ``` 步骤序列图 让我们进行一个简单的调用： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} graph.invoke({"value_1": "c"}) ``` ``` {'value_1': 'a b', 'value_2': 10} ``` 请注意： * 我们通过为单个状态键提供值来启动调用。我们必须始终至少为一个键提供值。 * 我们传入的值被第一个节点覆盖。 * 第二个节点更新了该值。 * 第三个节点填充了一个不同的值。 **内置简写** `langgraph>=0.2.46` 包含一个内置的简写 `add_sequence`，用于添加节点序列。你可以如下编译相同的图： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} builder = StateGraph(State).add_sequence([step_1, step_2, step_3]) # [!code highlight] builder.add_edge(START, "step_1") graph = builder.compile() graph.invoke({"value_1": "c"}) ``` ## 创建分支 节点的并行执行对于加速整体图操作至关重要。LangGraph 提供了对节点并行执行的原生支持，这可以显著提高基于图的工作流的性能。这种并行化是通过扇出和扇入机制实现的，利用标准边和 [conditional\_edges](https://langchain-ai.github.io/langgraph/reference/graphs.md#langgraph.graph.MessageGraph.add_conditional_edges)。以下是一些示例，展示如何添加创建适合你的分支数据流。 ### 并行运行图节点 在这个示例中，我们从 `Node A` 扇出到 `B and C`，然后扇入到 `D`。对于我们的状态，[我们指定了归约器 add 操作](/oss/python/langgraph/graph-api#reducers)。这将组合或累积状态中特定键的值，而不是简单地覆盖现有值。对于列表，这意味着将新列表与现有列表连接起来。有关使用归约器更新状态的更多详细信息，请参阅上面的[状态归约器](#使用归约器处理状态更新)部分。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} import operator from typing import Annotated, Any from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): # operator.add 归约器函数使其仅追加 aggregate: Annotated[list, operator.add] def a(state: State): print(f'Adding "A" to {state["aggregate"]}') return {"aggregate": ["A"]} def b(state: State): print(f'Adding "B" to {state["aggregate"]}') return {"aggregate": ["B"]} def c(state: State): print(f'Adding "C" to {state["aggregate"]}') return {"aggregate": ["C"]} def d(state: State): print(f'Adding "D" to {state["aggregate"]}') return {"aggregate": ["D"]} builder = StateGraph(State) builder.add_node(a) builder.add_node(b) builder.add_node(c) builder.add_node(d) builder.add_edge(START, "a") builder.add_edge("a", "b") builder.add_edge("a", "c") builder.add_edge("b", "d") builder.add_edge("c", "d") builder.add_edge("d", END) graph = builder.compile() ``` ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from IPython.display import Image, display display(Image(graph.get_graph().draw_mermaid_png())) ``` 并行执行图 使用归约器，你可以看到在每个节点中添加的值被累积了。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} graph.invoke({"aggregate": []}, {"configurable": {"thread_id": "foo"}}) ``` ``` Adding "A" to [] Adding "B" to ['A'] Adding "C" to ['A'] Adding "D" to ['A', 'B', 'C'] ``` 在上面的示例中，节点 `"b"` 和 `"c"` 在同一个[超级步骤](/oss/python/langgraph/graph-api#graphs)中并发执行。因为它们在同一个步骤中，所以节点 `"d"` 在 `"b"` 和 `"c"` 都完成后执行。 重要的是，并行超级步骤的更新可能不会按一致的顺序排列。如果你需要并行超级步骤的更新具有一致的、预定的顺序，你应该将输出与用于排序的值一起写入状态中的单独字段。 LangGraph 在[超级步骤](/oss/python/langgraph/graph-api#graphs)内执行节点，这意味着虽然并行分支是并行执行的，但整个超级步骤是**事务性的**。如果这些分支中的任何一个引发异常，则**不会**有任何更新应用到状态（整个超级步骤出错）。 重要的是，当使用[检查点器](/oss/python/langgraph/persistence)时，超级步骤内成功节点的结果会被保存，并且在恢复时不会重复。 如果你有容易出错的（可能想处理不稳定的 API 调用），LangGraph 提供了两种解决方法： 1. 你可以在节点内编写常规的 Python 代码来捕获和处理异常。 2. 你可以设置一个 **[retry\_policy](https://langchain-ai.github.io/langgraph/reference/types/#langgraph.types.RetryPolicy)** 来指示图重试引发某些类型异常的节点。只有失败的分支会被重试，因此你不必担心执行冗余工作。 这些结合在一起，让你可以执行并行执行并完全控制异常处理。 **设置最大并发数** 你可以在调用图时通过在[配置](https://reference.langchain.com/python/langchain-core/runnables/config/RunnableConfig)中设置 `max_concurrency` 来控制最大并发任务数。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} graph.invoke({"value_1": "c"}, {"configurable": {"max_concurrency": 10}}) ``` ### 延迟节点执行 当你想要延迟节点的执行直到所有其他待处理任务完成时，延迟节点执行很有用。当分支具有不同的长度时，这在 Map-Reduce 流程等工作流中尤其相关。 上面的示例展示了当每个路径只有一步时如何扇出和扇入。但如果一个分支有多个步骤呢？让我们在 `"b"` 分支中添加一个节点 `"b_2"`： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} import operator from typing import Annotated, Any from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): # operator.add 归约器函数使其仅追加 aggregate: Annotated[list, operator.add] def a(state: State): print(f'Adding "A" to {state["aggregate"]}') return {"aggregate": ["A"]} def b(state: State): print(f'Adding "B" to {state["aggregate"]}') return {"aggregate": ["B"]} def b_2(state: State): print(f'Adding "B_2" to {state["aggregate"]}') return {"aggregate": ["B_2"]} def c(state: State): print(f'Adding "C" to {state["aggregate"]}') return {"aggregate": ["C"]} def d(state: State): print(f'Adding "D" to {state["aggregate"]}') return {"aggregate": ["D"]} builder = StateGraph(State) builder.add_node(a) builder.add_node(b) builder.add_node(b_2) builder.add_node(c) builder.add_node(d, defer=True) # [!code highlight] builder.add_edge(START, "a") builder.add_edge("a", "b") builder.add_edge("a", "c") builder.add_edge("b", "b_2") builder.add_edge("b_2", "d") builder.add_edge("c", "d") builder.add_edge("d", END) graph = builder.compile() ``` ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from IPython.display import Image, display display(Image(graph.get_graph().draw_mermaid_png())) ``` 延迟执行图 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} graph.invoke({"aggregate": []}) ``` ``` Adding "A" to [] Adding "B" to ['A'] Adding "C" to ['A'] Adding "B_2" to ['A', 'B', 'C'] Adding "D" to ['A', 'B', 'C', 'B_2'] ``` 在上面的示例中，节点 `"b"` 和 `"c"` 在同一个超级步骤中并发执行。我们在节点 `d` 上设置了 `defer=True`，因此它将不会执行，直到所有待处理任务完成。在这种情况下，这意味着 `"d"` 等待执行，直到整个 `"b"` 分支完成。 ### 条件分支 如果你的扇出应该在运行时根据状态变化，你可以使用 [`add_conditional_edges`](https://reference.langchain.com/python/langgraph/graph/state/StateGraph/add_conditional_edges) 来使用图状态选择一个或多个路径。请参见下面的示例，其中节点 `a` 生成一个状态更新，该更新决定下一个节点。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} import operator from typing import Annotated, Literal, Sequence from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): aggregate: Annotated[list, operator.add] # 向状态添加一个键。我们将设置此键来确定 # 如何分支。 which: str def a(state: State): print(f'Adding "A" to {state["aggregate"]}') return {"aggregate": ["A"], "which": "c"} # [!code highlight] def b(state: State): print(f'Adding "B" to {state["aggregate"]}') return {"aggregate": ["B"]} def c(state: State): print(f'Adding "C" to {state["aggregate"]}') return {"aggregate": ["C"]} builder = StateGraph(State) builder.add_node(a) builder.add_node(b) builder.add_node(c) builder.add_edge(START, "a") builder.add_edge("b", END) builder.add_edge("c", END) def conditional_edge(state: State) -> Literal["b", "c"]: # 在这里填写使用状态的任意逻辑 # 来确定下一个节点 return state["which"] builder.add_conditional_edges("a", conditional_edge) # [!code highlight] graph = builder.compile() ``` ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from IPython.display import Image, display display(Image(graph.get_graph().draw_mermaid_png())) ``` 条件分支图 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} result = graph.invoke({"aggregate": []}) print(result) ``` ``` Adding "A" to [] Adding "C" to ['A'] {'aggregate': ['A', 'C'], 'which': 'c'} ``` 你的条件边可以路由到多个目标节点。例如： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} def route_bc_or_cd(state: State) -> Sequence[str]: if state["which"] == "cd": return ["c", "d"] return ["b", "c"] ``` ## Map-Reduce 和 send API LangGraph 使用 Send API 支持 Map-Reduce 和其他高级分支模式。以下是使用它的一个示例： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.graph import StateGraph, START, END from langgraph.types import Send from typing_extensions import TypedDict, Annotated import operator class OverallState(TypedDict): topic: str subjects: list[str] jokes: Annotated[list[str], operator.add] best_selected_joke: str def generate_topics(state: OverallState): return {"subjects": ["lions", "elephants", "penguins"]} def generate_joke(state: OverallState): joke_map = { "lions": "Why don't lions like fast food? Because they can't catch it!", "elephants": "Why don't elephants use computers? They're afraid of the mouse!", "penguins": "Why don't penguins like talking to strangers at parties? Because they find it hard to break the ice." } return {"jokes": [joke_map[state["subject"]]]} def continue_to_jokes(state: OverallState): return [Send("generate_joke", {"subject": s}) for s in state["subjects"]] def best_joke(state: OverallState): return {"best_selected_joke": "penguins"} builder = StateGraph(OverallState) builder.add_node("generate_topics", generate_topics) builder.add_node("generate_joke", generate_joke) builder.add_node("best_joke", best_joke) builder.add_edge(START, "generate_topics") builder.add_conditional_edges("generate_topics", continue_to_jokes, ["generate_joke"]) builder.add_edge("generate_joke", "best_joke") builder.add_edge("best_joke", END) graph = builder.compile() ``` ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from IPython.display import Image, display display(Image(graph.get_graph().draw_mermaid_png())) ``` 带有扇出的 Map-Reduce 图 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} # 调用图：这里我们调用它来生成一个笑话列表 for step in graph.stream({"topic": "animals"}): print(step) ``` ``` {'generate_topics': {'subjects': ['lions', 'elephants', 'penguins']}} {'generate_joke': {'jokes': ["Why don't lions like fast food? Because they can't catch it!"]}} {'generate_joke': {'jokes': ["Why don't elephants use computers? They're afraid of the mouse!"]}} {'generate_joke': {'jokes': ['Why don't penguins like talking to strangers at parties? Because they find it hard to break the ice.']}} {'best_joke': {'best_selected_joke': 'penguins'}} ``` ## 创建和控制循环 在创建带有循环的图时，我们需要一种终止执行的机制。最常见的是通过添加一个[条件边](/oss/python/langgraph/graph-api#conditional-edges)来实现，该边在达到某个终止条件时路由到 [END](/oss/python/langgraph/graph-api#end-node) 节点。 你也可以在调用或流式传输图时设置图的递归限制。递归限制设置了图在引发错误之前允许执行的[超级步骤](/oss/python/langgraph/graph-api#graphs)数量。阅读更多关于[递归限制概念](/oss/python/langgraph/graph-api#recursion-limit)的信息。 让我们考虑一个带有循环的简单图，以更好地理解这些机制的工作原理。 要返回状态的最后一个值而不是收到递归限制错误，请参阅[下一节](#施加递归限制)。 创建循环时，你可以包含一个指定终止条件的条件边： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} builder = StateGraph(State) builder.add_node(a) builder.add_node(b) def route(state: State) -> Literal["b", END]: if termination_condition(state): return END else: return "b" builder.add_edge(START, "a") builder.add_conditional_edges("a", route) builder.add_edge("b", "a") graph = builder.compile() ``` 要控制递归限制，请在配置中指定 `"recursion_limit"`。这将引发一个 `GraphRecursionError`，你可以捕获并处理它： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.errors import GraphRecursionError try: graph.invoke(inputs, {"recursion_limit": 3}) except GraphRecursionError: print("Recursion Error") ``` 让我们定义一个带有简单循环的图。请注意，我们使用条件边来实现终止条件。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} import operator from typing import Annotated, Literal from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): # operator.add 归约器函数使其仅追加 aggregate: Annotated[list, operator.add] def a(state: State): print(f'Node A sees {state["aggregate"]}') return {"aggregate": ["A"]} def b(state: State): print(f'Node B sees {state["aggregate"]}') return {"aggregate": ["B"]} # 定义节点 builder = StateGraph(State) builder.add_node(a) builder.add_node(b) # 定义边 def route(state: State) -> Literal["b", END]: if len(state["aggregate"]) < 7: return "b" else: return END builder.add_edge(START, "a") builder.add_conditional_edges("a", route) builder.add_edge("b", "a") graph = builder.compile() ``` ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from IPython.display import Image, display display(Image(graph.get_graph().draw_mermaid_png())) ``` 简单循环图 这种架构类似于 [ReAct 代理](/oss/python/langgraph/workflows-agents)，其中节点 `"a"` 是一个工具调用模型，节点 `"b"` 代表工具。 在我们的 `route` 条件边中，我们指定当状态中的 `"aggregate"` 列表超过阈值长度后应结束。 调用图，我们看到在达到终止条件之前，我们在节点 `"a"` 和 `"b"` 之间交替。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} graph.invoke({"aggregate": []}) ``` ``` Node A sees [] Node B sees ['A'] Node A sees ['A', 'B'] Node B sees ['A', 'B', 'A'] Node A sees ['A', 'B', 'A', 'B'] Node B sees ['A', 'B', 'A', 'B', 'A'] Node A sees ['A', 'B', 'A', 'B', 'A', 'B'] ``` ### 施加递归限制 在某些应用程序中，我们可能无法保证会达到给定的终止条件。在这些情况下，我们可以设置图的[递归限制](/oss/python/langgraph/graph-api#recursion-limit)。这将在给定数量的[超级步骤](/oss/python/langgraph/graph-api#graphs)后引发 `GraphRecursionError`。然后我们可以捕获并处理此异常： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.errors import GraphRecursionError try: graph.invoke({"aggregate": []}, {"recursion_limit": 4}) except GraphRecursionError: print("Recursion Error") ``` ``` Node A sees [] Node B sees ['A'] Node C sees ['A', 'B'] Node D sees ['A', 'B'] Node A sees ['A', 'B', 'C', 'D'] Recursion Error ``` 与其引发 `GraphRecursionError`，我们可以向状态添加一个新键来跟踪达到递归限制之前剩余的步骤数。然后我们可以使用此键来确定是否应结束运行。 LangGraph 实现了一个特殊的 `RemainingSteps` 注解。在底层，它创建了一个 `ManagedValue` 通道——一个在图运行期间存在而之后不再存在的状态通道。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} import operator from typing import Annotated, Literal from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.managed.is_last_step import RemainingSteps class State(TypedDict): aggregate: Annotated[list, operator.add] remaining_steps: RemainingSteps def a(state: State): print(f'Node A sees {state["aggregate"]}') return {"aggregate": ["A"]} def b(state: State): print(f'Node B sees {state["aggregate"]}') return {"aggregate": ["B"]} # 定义节点 builder = StateGraph(State) builder.add_node(a) builder.add_node(b) # 定义边 def route(state: State) -> Literal["b", END]: if state["remaining_steps"] <= 2: return END else: return "b" builder.add_edge(START, "a") builder.add_conditional_edges("a", route) builder.add_edge("b", "a") graph = builder.compile() # 测试一下 result = graph.invoke({"aggregate": []}, {"recursion_limit": 4}) print(result) ``` ``` Node A sees [] Node B sees ['A'] Node A sees ['A', 'B'] {'aggregate': ['A', 'B', 'A']} ``` 为了更好地理解递归限制的工作原理，让我们考虑一个更复杂的例子。下面我们实现一个循环，但其中一个步骤扇出到两个节点： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} import operator from typing import Annotated, Literal from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): aggregate: Annotated[list, operator.add] def a(state: State): print(f'Node A sees {state["aggregate"]}') return {"aggregate": ["A"]} def b(state: State): print(f'Node B sees {state["aggregate"]}') return {"aggregate": ["B"]} def c(state: State): print(f'Node C sees {state["aggregate"]}') return {"aggregate": ["C"]} def d(state: State): print(f'Node D sees {state["aggregate"]}') return {"aggregate": ["D"]} # 定义节点 builder = StateGraph(State) builder.add_node(a) builder.add_node(b) builder.add_node(c) builder.add_node(d) # 定义边 def route(state: State) -> Literal["b", END]: if len(state["aggregate"]) < 7: return "b" else: return END builder.add_edge(START, "a") builder.add_conditional_edges("a", route) builder.add_edge("b", "c") builder.add_edge("b", "d") builder.add_edge(["c", "d"], "a") graph = builder.compile() ``` ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from IPython.display import Image, display display(Image(graph.get_graph().draw_mermaid_png())) ``` 带有分支的复杂循环图 这个图看起来很复杂，但可以概念化为[超级步骤](/oss/python/langgraph/graph-api#graphs)的循环： 1. 节点 A 2. 节点 B 3. 节点 C 和 D 4. 节点 A 5. ... 我们有一个四个超级步骤的循环，其中节点 C 和 D 并发执行。 像以前一样调用图，我们看到在达到终止条件之前，我们完成了两个完整的“圈”： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} result = graph.invoke({"aggregate": []}) ``` ``` Node A sees [] Node B sees ['A'] Node D sees ['A', 'B'] Node C sees ['A', 'B'] Node A sees ['A', 'B', 'C', 'D'] Node B sees ['A', 'B', 'C', 'D', 'A'] Node D sees ['A', 'B', 'C', 'D', 'A', 'B'] Node C sees ['A', 'B', 'C', 'D', 'A', 'B'] Node A sees ['A', 'B', 'C', 'D', 'A', 'B', 'C', 'D'] ``` 然而，如果我们设置递归限制为四，我们只完成一圈，因为每圈是四个超级步骤： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.errors import GraphRecursionError try: result = graph.invoke({"aggregate": []}, {"recursion_limit": 4}) except GraphRecursionError: print("Recursion Error") ``` ``` Node A sees [] Node B sees ['A'] Node C sees ['A', 'B'] Node D sees ['A', 'B'] Node A sees ['A', 'B', 'C', 'D'] Recursion Error ``` ## 异步 使用异步编程范式可以在并发运行 [IO 密集型](https://en.wikipedia.org/wiki/I/O_bound)代码时（例如，向聊天模型提供者发出并发 API 请求）产生显著的性能改进。 要将图的 `sync` 实现转换为 `async` 实现，你需要： 1. 更新 `nodes` 使用 `async def` 而不是 `def`。 2. 更新内部代码以适当地使用 `await`。 3. 根据需要使用 `.ainvoke` 或 `.astream` 调用图。 因为许多 LangChain 对象实现了 [Runnable Protocol](https://python.langchain.com/docs/expression_language/interface/)，该协议具有所有 `sync` 方法的 `async` 变体，所以通常将 `sync` 图升级为 `async` 图相当快。 请参见下面的示例。为了演示底层 LLM 的异步调用，我们将包含一个聊天模型： 👉 阅读 [OpenAI 聊天模型集成文档](/oss/python/integrations/chat/openai/) ```shell theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} pip install -U "langchain[openai]" ``` ```python init_chat_model theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} import os from langchain.chat_models import init_chat_model os.environ["OPENAI_API_KEY"] = "sk-..." model = init_chat_model("gpt-5.4") ``` ```python Model Class theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} import os from langchain_openai import ChatOpenAI os.environ["OPENAI_API_KEY"] = "sk-..." model = ChatOpenAI(model="gpt-5.4") ``` 👉 阅读 [Anthropic 聊天模型集成文档](/oss/python/integrations/chat/anthropic/) ```shell theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} pip install -U "langchain[anthropic]" ``` ```python init_chat_model theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} import os from langchain.chat_models import init_chat_model os.environ["ANTHROPIC_API_KEY"] = "sk-..." model = init_chat_model("claude-sonnet-4-6") ``` ```python Model Class theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} import os from langchain_anthropic import ChatAnthropic os.environ["ANTHROPIC_API_KEY"] = "sk-..." model = ChatAnthropic(model="claude-sonnet-4-6") ``` 👉 阅读 [Azure 聊天模型集成文档](/oss/python/integrations/chat/azure_chat_openai/) ```shell theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} pip install -U "langchain[openai]" ``` ```python init_chat_model theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} import os from langchain.chat_models import init_chat_model os.environ["AZURE_OPENAI_API_KEY"] = "..." os.environ["AZURE_OPENAI_ENDPOINT"] = "..." os.environ["OPENAI_API_VERSION"] = "2025-03-01-preview" model = init_chat_model( "azure_openai:gpt-5.4", azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"], ) ``` ```python Model Class theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} import os from langchain_openai import AzureChatOpenAI os.environ["AZURE_OPENAI_API_KEY"] = "..." os.environ["AZURE_OPENAI_ENDPOINT"] = "..." os.environ["OPENAI_API_VERSION"] = "2025-03-01-preview" model = AzureChatOpenAI( model="gpt-5.4", azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"] ) ``` 👉 阅读 [Google GenAI 聊天模型集成文档](/oss/python/integrations/chat/google_generative_ai/) ```shell theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} pip install -U "langchain[google-genai]" ``` ```python init_chat_model theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} import os from langchain.chat_models import init_chat_model os.environ["GOOGLE_API_KEY"] = "..." model = init_chat_model("google_genai:gemini-2.5-flash-lite") ``` ```python Model Class theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} import os from langchain_google_genai import ChatGoogleGenerativeAI os.environ["GOOGLE_API_KEY"] = "..." model = ChatGoogleGenerativeAI(model="gemini-2.5-flash-lite") ``` 👉 阅读 [AWS Bedrock 聊天模型集成文档](/oss/python/integrations/chat/bedrock/) ```shell theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} pip install -U "langchain[aws]" ``` ```python init_chat_model theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langchain.chat_models import init_chat_model # 按照此处步骤配置您的凭证： # https://docs.aws.amazon.com/bedrock/latest/userguide/getting-started.html model = init_chat_model( "anthropic.claude-3-5-sonnet-20240620-v1:0", model_provider="bedrock_converse", ) ``` ```python Model Class theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langchain_aws import ChatBedrock model = ChatBedrock(model="anthropic.claude-3-5-sonnet-20240620-v1:0") ``` 👉 阅读 [HuggingFace 聊天模型集成文档](/oss/python/integrations/chat/huggingface/) ```shell theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} pip install -U "langchain[huggingface]" ``` ```python init_chat_model theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} import os from langchain.chat_models import init_chat_model os.environ["HUGGINGFACEHUB_API_TOKEN"] = "hf_..." model = init_chat_model( "microsoft/Phi-3-mini-4k-instruct", model_provider="huggingface", temperature=0.7, max_tokens=1024, ) ``` ```python Model Class theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} import os from langchain_huggingface import ChatHuggingFace, HuggingFaceEndpoint os.environ["HUGGINGFACEHUB_API_TOKEN"] = "hf_..." llm = HuggingFaceEndpoint( repo_id="microsoft/Phi-3-mini-4k-instruct", temperature=0.7, max_length=1024, ) model = ChatHuggingFace(llm=llm) ``` 👉 阅读 [OpenRouter 聊天模型集成文档](/oss/python/integrations/chat/openrouter/) ```shell theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} pip install -U "langchain-openrouter" ``` ```python init_chat_model theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} import os from langchain.chat_models import init_chat_model os.environ["OPENROUTER_API_KEY"] = "sk-..." model = init_chat_model( "auto", model_provider="openrouter", ) ``` ```python Model Class theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} import os from langchain_openrouter import ChatOpenRouter os.environ["OPENROUTER_API_KEY"] = "sk-..." model = ChatOpenRouter(model="auto") ``` ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langchain.chat_models import init_chat_model from langgraph.graph import MessagesState, StateGraph async def node(state: MessagesState): # [!code highlight] new_message = await llm.ainvoke(state["messages"]) # [!code highlight] return {"messages": [new_message]} builder = StateGraph(MessagesState).add_node(node).set_entry_point("node") graph = builder.compile() input_message = {"role": "user", "content": "Hello"} result = await graph.ainvoke({"messages": [input_message]}) # [!code highlight] ``` **异步流式传输** 有关异步流式传输的示例，请参阅[流式传输指南](/oss/python/langgraph/streaming)。 ## 使用 `Command` 结合控制流和状态更新 结合控制流（边）和状态更新（节点）可能很有用。例如，你可能希望在同一个节点中同时执行状态更新并决定下一个节点。LangGraph 提供了一种通过从节点函数返回 [Command](https://langchain-ai.github.io/langgraph/reference/types/#langgraph.types.Command) 对象来实现此目的的方法： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} def my_node(state: State) -> Command[Literal["my_other_node"]]: return Command( # 状态更新 update={"foo": "bar"}, # 控制流 goto="my_other_node" ) ``` 我们在下面展示一个端到端的示例。让我们创建一个包含 3 个节点的简单图：A、B 和 C。我们将首先执行节点 A，然后根据节点 A 的输出决定接下来是转到节点 B 还是节点 C。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} import random from typing_extensions import TypedDict, Literal from langgraph.graph import StateGraph, START from langgraph.types import Command # 定义图状态 class State(TypedDict): foo: str # 定义节点 def node_a(state: State) -> Command[Literal["node_b", "node_c"]]: print("Called A") value = random.choice(["b", "c"]) # 这是条件边函数的替代 if value == "b": goto = "node_b" else: goto = "node_c" # 注意 Command 如何允许你同时更新图状态并路由到下一个节点 return Command( # 这是状态更新 update={"foo": value}, # 这是边的替代 goto=goto, ) def node_b(state: State): print("Called B") return {"foo": state["foo"] + "b"} def node_c(state: State): print("Called C") return {"foo": state["foo"] + "c"} ``` 我们现在可以使用上面的节点创建 [`StateGraph`](https://reference.langchain.com/python/langgraph/graph/state/StateGraph)。请注意，图没有用于路由的[条件边](/oss/python/langgraph/graph-api#conditional-edges)！这是因为控制流是在 `node_a` 内部使用 [`Command`](https://reference.langchain.com/python/langgraph/types/Command) 定义的。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} builder = StateGraph(State) builder.add_edge(START, "node_a") builder.add_node(node_a) builder.add_node(node_b) builder.add_node(node_c) # 注意：节点 A、B 和 C 之间没有边！ graph = builder.compile() ``` 你可能已经注意到我们使用 [`Command`](https://reference.langchain.com/python/langgraph/types/Command) 作为返回类型注解，例如 `Command[Literal["node_b", "node_c"]]`。这对于图渲染是必要的，并告诉 LangGraph `node_a` 可以导航到 `node_b` 和 `node_c`。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from IPython.display import display, Image display(Image(graph.get_graph().draw_mermaid_png())) ``` 基于 Command 的图导航 如果我们多次运行图，我们会看到它根据节点 A 中的随机选择采取不同的路径（A -> B 或 A -> C）。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} graph.invoke({"foo": ""}) ``` ``` Called A Called C ``` ### 导航到父图中的节点 如果你正在使用[子图](/oss/python/langgraph/use-subgraphs)，你可能希望从子图中的节点导航到不同的子图（即父图中的不同节点）。为此，你可以在 `Command` 中指定 `graph=Command.PARENT`： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} def my_node(state: State) -> Command[Literal["my_other_node"]]: return Command( update={"foo": "bar"}, goto="other_subgraph", # 其中 `other_subgraph` 是父图中的一个节点 graph=Command.PARENT ) ``` 让我们使用上面的示例来演示这一点。我们将通过将上面示例中的 `nodeA` 更改为一个单节点图来实现，该图将作为子图添加到我们的父图中。 **使用 `Command.PARENT` 进行状态更新** 当你从子图节点向父图节点发送更新时，对于父图和子图[状态模式](/oss/python/langgraph/graph-api#schema)共享的键，你**必须**在父图状态中为要更新的键定义一个[归约器](/oss/python/langgraph/graph-api#reducers)。请参见下面的示例。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} import operator from typing_extensions import Annotated class State(TypedDict): # 注意：我们在这里定义了一个归约器 foo: Annotated[str, operator.add] # [!code highlight] def node_a(state: State): print("Called A") value = random.choice(["a", "b"]) # 这是条件边函数的替代 if value == "a": goto = "node_b" else: goto = "node_c" # 注意 Command 如何允许你同时更新图状态并路由到下一个节点 return Command( update={"foo": value}, goto=goto, # 这告诉 LangGraph 导航到父图中的 node_b 或 node_c # 注意：这将导航到相对于子图的最近父图 graph=Command.PARENT, # [!code highlight] ) subgraph = StateGraph(State).add_node(node_a).add_edge(START, "node_a").compile() def node_b(state: State): print("Called B") # 注意：由于我们已经定义了归约器，我们不需要手动追加 # 新字符到现有的 'foo' 值。相反，归约器将自动追加这些 # （通过 operator.add） return {"foo": "b"} # [!code highlight] def node_c(state: State): print("Called C") return {"foo": "c"} # [!code highlight] builder = StateGraph(State) builder.add_edge(START, "subgraph") builder.add_node("subgraph", subgraph) builder.add_node(node_b) builder.add_node(node_c) graph = builder.compile() ``` ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} graph.invoke({"foo": ""}) ``` ``` Called A Called C ``` ### 在工具内使用 一个常见的用例是从工具内部更新图状态。例如，在客户支持应用程序中，你可能希望在对话开始时根据客户的帐户号或 ID 查找客户信息。要从工具更新图状态，你可以从工具返回 `Command(update={"my_custom_key": "foo", "messages": [...]})`： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langchain.tools import ToolRuntime @tool def lookup_user_info(runtime: ToolRuntime): """Use this to look up user information to better assist them with their questions.""" user_info = get_user_info(runtime.server_info.user.identity) # [!code highlight] return Command( update={ # 更新状态键 "user_info": user_info, # 更新消息历史 "messages": [ToolMessage("Successfully looked up user information", tool_call_id=runtime.tool_call_id)] } ) ``` 当你从工具返回 [`Command`](https://reference.langchain.com/python/langgraph/types/Command) 时，你**必须**在 `Command.update` 中包含 `messages`（或用于消息历史的任何状态键），并且 `messages` 中的消息列表**必须**包含一个 `ToolMessage`。这对于生成的消息历史有效是必要的（LLM 提供者要求带有工具调用的 AI 消息后面必须跟有工具结果消息）。 如果你正在使用通过 [`Command`](https://reference.langchain.com/python/langgraph/types/Command) 更新状态的工具，我们建议使用预构建的 [`ToolNode`](https://reference.langchain.com/python/langgraph/agents/#langgraph.prebuilt.tool_node.ToolNode)，它自动处理返回 [`Command`](https://reference.langchain.com/python/langgraph/types/Command) 对象的工具并将其传播到图状态。如果你正在编写一个调用工具的自定义节点，你需要手动将工具返回的 [`Command`](https://reference.langchain.com/python/langgraph/types/Command) 对象作为节点的更新进行传播。 ## 可视化你的图 这里我们演示如何可视化你创建的图。 你可以可视化任何任意的 [Graph](https://langchain-ai.github.io/langgraph/reference/graphs/)，包括 [StateGraph](https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.state.StateGraph)。 让我们通过绘制分形来获得一些乐趣 :)。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} import random from typing import Annotated, Literal from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages class State(TypedDict): messages: Annotated[list, add_messages] class MyNode: def __init__(self, name: str): self.name = name def __call__(self, state: State): return {"messages": [("assistant", f"Called node {self.name}")]} def route(state) -> Literal["entry_node", END]: if len(state["messages"]) > 10: return END return "entry_node" def add_fractal_nodes(builder, current_node, level, max_level): if level > max_level: return # 此级别要创建的节点数 num_nodes = random.randint(1, 3) # 根据需要调整随机性 for i in range(num_nodes): nm = ["A", "B", "C"][i] node_name = f"node_{current_node}_{nm}" builder.add_node(node_name, MyNode(node_name)) builder.add_edge(current_node, node_name) # 递归添加更多节点 r = random.random() if r > 0.2 and level + 1 < max_level: add_fractal_nodes(builder, node_name, level + 1, max_level) elif r > 0.05: builder.add_conditional_edges(node_name, route, node_name) else: # 结束 builder.add_edge(node_name, END) def build_fractal_graph(max_level: int): builder = StateGraph(State) entry_point = "entry_node" builder.add_node(entry_point, MyNode(entry_point)) builder.add_edge(START, entry_point) add_fractal_nodes(builder, entry_point, 1, max_level) # 可选：如果需要，设置一个结束点 builder.add_edge(entry_point, END) # 或任何特定节点 return builder.compile() app = build_fractal_graph(3) ``` ### Mermaid 我们也可以将图类转换为 Mermaid 语法。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} print(app.get_graph().draw_mermaid()) ``` ``` %%{init: {'flowchart': {'curve': 'linear'}}}%% graph TD; tart__([

__start__

]):::first ry_node(entry_node) e_entry_node_A(node_entry_node_A) e_entry_node_B(node_entry_node_B) e_node_entry_node_B_A(node_node_entry_node_B_A) e_node_entry_node_B_B(node_node_entry_node_B_B) e_node_entry_node_B_C(node_node_entry_node_B_C) nd__([

__end__

]):::last tart__ --> entry_node; ry_node --> __end__; ry_node --> node_entry_node_A; ry_node --> node_entry_node_B; e_entry_node_B --> node_node_entry_node_B_A; e_entry_node_B --> node_node_entry_node_B_B; e_entry_node_B --> node_node_entry_node_B_C; e_entry_node_A -.-> entry_node; e_entry_node_A -.-> __end__; e_node_entry_node_B_A -.-> entry_node; e_node_entry_node_B_A -.-> __end__; e_node_entry_node_B_B -.-> entry_node; e_node_entry_node_B_B -.-> __end__; e_node_entry_node_B_C -.-> entry_node; e_node_entry_node_B_C -.-> __end__; ssDef default fill:#f2f0ff,line-height:1.2 ssDef first fill-opacity:0 ssDef last fill:#bfb6fc ``` ### PNG 如果需要，我们可以将图渲染为 `.png`。这里我们可以使用三个选项： * 使用 Mermaid.ink API（不需要额外的包） * 使用 Mermaid + Pyppeteer（需要 `pip install pyppeteer`） * 使用 graphviz（需要 `pip install graphviz`） **使用 Mermaid.Ink** 默认情况下，`draw_mermaid_png()` 使用 Mermaid.Ink 的 API 来生成图表。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from IPython.display import Image, display from langchain_core.runnables.graph import CurveStyle, MermaidDrawMethod, NodeStyles display(Image(app.get_graph().draw_mermaid_png())) ``` 分形图可视化 **使用 Mermaid + Pyppeteer** ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} import nest_asyncio nest_asyncio.apply() # Jupyter Notebook 运行异步函数所需 display( Image( app.get_graph().draw_mermaid_png( curve_style=CurveStyle.LINEAR, node_colors=NodeStyles(first="#ffdfba", last="#baffc9", default="#fad7de"), wrap_label_n_words=9, output_file_path=None, draw_method=MermaidDrawMethod.PYPPETEER, background_color="white", padding=10, ) ) ) ``` **使用 Graphviz** ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} try: display(Image(app.get_graph().draw_png())) except ImportError: print( "You likely need to install dependencies for pygraphviz, see more here https://github.com/pygraphviz/pygraphviz/blob/main/INSTALL.txt" ) ``` ***
[将这些文档连接](/use-these-docs)到 Claude、VSCode 等，通过 MCP 获得实时答案。 [在 GitHub 上编辑此页面](https://github.com/langchain-ai/docs/edit/main/src/oss/langgraph/use-graph-api.mdx)或[提交问题](https://github.com/langchain-ai/docs/issues/new/choose)。