使用图 API - 以编程方式使用文档

本指南演示了 LangGraph 图 API 的基础知识。它逐步讲解状态，以及组合常见的图结构，如序列, 分支和循环。还涵盖 LangGraph 的控制功能，包括用于 map-reduce 工作流的 Send API 以及用于将状态更新与跨节点"跳转"相结合的 Command API 。

设置

安装 langgraph:

pip install -U langgraph

uv add langgraph

定义和更新状态

这里展示如何在 LangGraph 中定义和更新状态。我们将演示：

如何使用状态来定义图的模式
如何使用归约器来控制状态更新的处理方式。

定义状态

状态在 LangGraph 中可以是 TypedDict, Pydantic 模型或数据类。下面我们将使用 TypedDict。参见使用 Pydantic 模型作为图状态了解使用 Pydantic 的详细信息。

默认情况下，图的输入和输出模式相同，状态决定该模式。参见定义输入和输出模式了解如何定义不同的输入和输出模式。

让我们考虑一个使用消息的简单示例。这代表了适用于许多 LLM 应用的通用状态公式。更多信息请参阅我们的概念页面。

from langchain.messages import AnyMessage
from typing_extensions import TypedDict

class State(TypedDict):
    messages: list[AnyMessage]
    extra_field: int

此状态跟踪一个消息对象列表，以及一个额外的整数字段。

更新状态

让我们构建一个包含单个节点的示例图。我们的节点只是一个读取我们图状态并对其进行更新的 Python 函数。此函数的第一个参数始终是状态：

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 来定义一个在此状态上运行的图。然后我们使用 add_node 来填充我们的图。

from langgraph.graph import StateGraph

builder = StateGraph(State)
builder.add_node(node)
builder.set_entry_point("node")
graph = builder.compile()

LangGraph 提供了内置的图形可视化工具。让我们检查一下我们的图。请参阅可视化您的图了解可视化的详细信息。

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

!包含单个节点的简单图

在这种情况下，我们的图只是执行一个节点。让我们继续进行一个简单的调用：

from langchain.messages import HumanMessage

result = graph.invoke({"messages": [HumanMessage("Hi")]})
result

{'messages': [HumanMessage(content='Hi'), AIMessage(content='Hello!')], 'extra_field': 10}

请注意：

* 我们通过更新状态的一个键来启动调用。
* 我们在调用结果中收到整个状态。

为方便起见，我们经常检查消息对象的内容通过格式化输出：

for message in result["messages"]:
    message.pretty_print()

================================ Human Message ================================

Hi
================================== Ai Message ==================================

Hello!

使用归约器处理状态更新

状态中的每个键都可以有自己独立的归约器函数，该函数控制如何应用来自节点的更新。如果未明确指定归约器函数，则假定对该键的所有更新都应覆盖它。

对于 TypedDict 状态模式，我们可以通过使用归约器函数注解状态的相应字段来定义归约器。

在之前的示例中，我们的节点更新了 "messages" 通过向其中追加消息来更新状态中的键。下面，我们向此键添加一个 reducer，这样更新会自动追加：

from typing_extensions import Annotated

def add(left, right):
    """Can also import `add` from the `operator` built-in."""
    return left + right

class State(TypedDict):
    messages: Annotated[list[AnyMessage], add]  # [!code highlight]
    extra_field: int

现在我们的节点可以简化：

def node(state: State):
    new_message = AIMessage("Hello!")
    return {"messages": [new_message], "extra_field": 10}  # [!code highlight]

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

在实践中，更新消息列表还有额外的考虑事项：

* 我们可能希望更新状态中现有的消息。
* 我们可能希望接受消息格式的简写，例如 OpenAI 格式.

LangGraph 包含一个内置的 reducer add_messages 来处理这些考虑事项：

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()

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!

这是一种用于涉及聊天模型应用的通用状态表示。LangGraph 包含一个预构建的 MessagesState 以方便使用，我们可以这样：

from langgraph.graph import MessagesState

class State(MessagesState):
    extra_field: int

使用 `Overwrite`

绕过 reducer 在某些情况下，你可能想绕过 reducer 并直接覆盖状态值。LangGraph 提供了 Overwrite 类型来实现此目的。当节点返回用 Overwrite包装的值时，reducer 被绕过，通道直接设置为此值。

当你想要重置或替换累积的状态而不是与现有值合并时，这很有用。

from langgraph.graph import StateGraph, START, END
from langgraph.types import Overwrite
from typing_extensions import Annotated, TypedDict


class State(TypedDict):
    messages: Annotated[list, operator.add]

def add_message(state: State):
    return {"messages": ["first message"]}

def replace_messages(state: State):
    # Bypass the reducer and replace the entire messages list
    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']

你也可以使用带有特殊键的 JSON 格式 "__overwrite__":

def replace_messages(state: State):
    return {"messages": {"__overwrite__": ["replacement message"]}}

定义输入和输出模式

默认情况下， StateGraph 使用单一模式运行，所有节点都期望使用该模式进行通信。但是，也可以为图定义不同的输入和输出模式。

当指定了不同的模式时，节点之间的通信仍会使用内部模式。输入模式确保提供的输入符合预期结构，而输出模式则根据定义的输出模式过滤内部数据，仅返回相关信息。

下面，我们将了解如何定义不同的输入和输出模式。

from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict

# Define the schema for the input
class InputState(TypedDict):
    question: str

# Define the schema for the output
class OutputState(TypedDict):
    answer: str

# Define the overall schema, combining both input and output
class OverallState(InputState, OutputState):
    pass

# Define the node that processes the input and generates an answer
def answer_node(state: InputState):
    # Example answer and an extra key
    return {"answer": "bye", "question": state["question"]}

# Build the graph with input and output schemas specified
builder = StateGraph(OverallState, input_schema=InputState, output_schema=OutputState)
builder.add_node(answer_node)  # Add the answer node
builder.add_edge(START, "answer_node")  # Define the starting edge
builder.add_edge("answer_node", END)  # Define the ending edge
graph = builder.compile()  # Compile the graph

# Invoke the graph with an input and print the result
print(graph.invoke({"question": "hi"}))

{'answer': 'bye'}

请注意，invoke 的输出仅包含输出模式。

在节点之间传递私有状态

In some cases, you may want nodes to exchange information that is crucial for intermediate logic but doesn't need to be part of the main schema of the graph. This private data is not relevant to the overall input/output of the graph and should only be shared between certain nodes.

下面，我们将创建一个由三个节点（节点_1、节点_2 和节点_3）组成的有序图示例，其中私有数据在前两个步骤（节点_1 和节点_2）之间传递，而第三个步骤（节点_3）只能访问公开的总体状态。

from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict

# The overall state of the graph (this is the public state shared across nodes)
class OverallState(TypedDict):
    a: str

# Output from node_1 contains private data that is not part of the overall state
class Node1Output(TypedDict):
    private_data: str

# The private data is only shared between node_1 and 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 input only requests the private data available after 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 only has access to the overall state (no access to private data from 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

# Connect nodes in a sequence
# node_2 accepts private data from node_1, whereas
# node_3 does not see the private data.
builder = StateGraph(OverallState).add_sequence([node_1, node_2, node_3])
builder.add_edge(START, "node_1")
graph = builder.compile()

# Invoke the graph with the initial state
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 模型定义图状态

A StateGraph 在初始化时接受一个 state_schema 参数，用于指定图中节点可以访问和更新的状态"形状"。

在我们的示例中，通常使用 Python 原生的 TypedDict or dataclass 作为 state_schema，但 state_schema 可以是任何类型.

这里，我们将了解如何使用 Pydantic BaseModel 作为 state_schema 以对 **输入**.

from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
from pydantic import BaseModel

# The overall state of the graph (this is the public state shared across nodes)
class OverallState(BaseModel):
    a: str

def node(state: OverallState):
    return {"a": "goodbye"}

# Build the state graph
builder = StateGraph(OverallState)
builder.add_node(node)  # node_1 is the first node
builder.add_edge(START, "node")  # Start the graph with node_1
builder.add_edge("node", END)  # End the graph after node_1
graph = builder.compile()

# Test the graph with a valid input
graph.invoke({"a": "hello"})

使用 **无效** 输入调用图

try:
    graph.invoke({"a": 123})  # Should be a string
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 模型状态的更多特性：

Serialization Behavior

当使用 Pydantic 模型作为状态模式时，理解序列化的工作方式非常重要，尤其是当：

* 将 Pydantic 对象作为输入传递
* 从图接收输出
* 使用嵌套的 Pydantic 模型

让我们看看这些行为的具体表现。

  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):
      # Node receives a validated Pydantic object
      print(f"Input state type: {type(state)}")
      print(f"Nested type: {type(state.nested)}")
      # Return a dictionary update
      return {"text": state.text + " processed", "count": state.count + 1}

  # Build the graph
  builder = StateGraph(ComplexState)
  builder.add_node("process", process_node)
  builder.add_edge(START, "process")
  builder.add_edge("process", END)
  graph = builder.compile()

  # Create a Pydantic instance for input
  input_state = ComplexState(text="hello", count=0, nested=NestedModel(value="test"))
  print(f"Input object type: {type(input_state)}")

  # Invoke graph with a Pydantic instance
  result = graph.invoke(input_state)
  print(f"Output type: {type(result)}")
  print(f"Output content: {result}")

  # Convert back to Pydantic model if needed
  output_model = ComplexState(**result)
  print(f"Converted back to Pydantic: {type(output_model)}")

Runtime Type Coercion

Pydantic 对某些数据类型执行运行时类型强制转换。这可能有帮助，但如果不注意也可能导致意外行为。

  from langgraph.graph import StateGraph, START, END
  from pydantic import BaseModel

  class CoercionExample(BaseModel):
      # Pydantic will coerce string numbers to integers
      number: int
      # Pydantic will parse string booleans to 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()

  # Demonstrate coercion with string inputs that will be converted
  result = graph.invoke({"number": "42", "flag": "true"})

  # This would fail with a validation error
  try:
      graph.invoke({"number": "not-a-number", "flag": "true"})
  except Exception as e:
      print(f"\nExpected validation error: {e}")

Working with Message Models

在状态模式中使用 LangChain 消息类型时，序列化有重要的注意事项。您应该使用 AnyMessage （而不是 BaseMessage) for proper serialization/deserialization when using message objects over the wire.

  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()

  # Create input with a message
  initial_state = ChatState(
      messages=[HumanMessage(content="Hi")], context="Customer support chat"
  )

  result = graph.invoke(initial_state)
  print(f"Output: {result}")

  # Convert back to Pydantic model to see message types
  output_model = ChatState(**result)
  for i, msg in enumerate(output_model.messages):
      print(f"Message {i}: {type(msg).__name__} - {msg.content}")

添加运行时配置

有时你可能希望能够在调用图时对其进行配置。例如，你可能希望在运行时指定要使用的 LLM 或系统提示， _而不污染图状态中的这些参数_.

添加运行时配置的方法：

为配置指定 schema
将配置添加到节点或条件边的函数签名中
将配置传入图中。

下面是简单示例：

from langgraph.graph import END, StateGraph, START
from langgraph.runtime import Runtime
from typing_extensions import TypedDict

# 1. Specify config schema
class ContextSchema(TypedDict):
    my_runtime_value: str

# 2. Define a graph that accesses the config in a node
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. Pass in configuration at runtime:
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}

Extended example: specifying LLM at runtime

下面我们演示一个实际示例，其中配置在运行时使用的 LLM。我们将使用 OpenAI 和 Anthropic 模型。

  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()

  # Usage
  input_message = {"role": "user", "content": "hi"}
  # With no configuration, uses default (Anthropic)
  response_1 = graph.invoke({"messages": [input_message]}, context=ContextSchema())["messages"][-1]
  # Or, can set 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

Extended example: specifying model and system message at runtime

下面我们演示一个实际示例，其中配置两个参数：运行时使用的 LLM 和系统消息。

  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()

  # Usage
  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。 retry_policy 参数接受一个 RetryPolicy 命名元组对象。下面我们实例化一个 RetryPolicy 对象及其默认参数，并将其与一个节点关联：

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 状态码时重试。

Extended example: customizing retry policies

考虑一个从 SQL 数据库读取的示例。下面我们向节点传递两个不同的重试策略：

  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.messages import AIMessage

  con = sqlite3.connect(":memory:")
  model = init_chat_model("claude-haiku-4-5-20251001")

  def query_database(state: MessagesState):
      cursor = con.cursor()
      cursor.execute("SELECT * FROM Artist LIMIT 10;")
      query_result = str(cursor.fetchall())
      return {"messages": [AIMessage(content=query_result)]}

  def call_model(state: MessagesState):
      response = model.invoke(state["messages"])
      return {"messages": [response]}

  # Define a new graph
  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()

设置节点超时

使用 timeout 参数和 add_node 来限制单个异步节点调用的运行时长。以秒为单位或以 datetime.timedelta.

from typing_extensions import TypedDict

from langgraph.errors import NodeTimeoutError
from langgraph.graph import END, START, StateGraph


class State(TypedDict):
    value: str


async def call_model(state: State) -> State:
    await asyncio.sleep(2)
    return {"value": "done"}


builder = StateGraph(State)
builder.add_node("model", call_model, timeout=1.0)
builder.add_edge(START, "model")
builder.add_edge("model", END)
graph = builder.compile()

try:
    await graph.ainvoke({"value": "start"})
except NodeTimeoutError:
    print("Node timed out")

节点超时仅适用于异步节点。如果设置 timeout 在同步节点上，LangGraph 会在编译图时引发错误，因为同步 Python 执行无法在进程内安全取消。

当节点超过其超时时，LangGraph 会引发 NodeTimeoutError，它继承自 Python 的内置 TimeoutError。 retry_policy 如果节点有 TimeoutError or NodeTimeoutError重试的超时尝试，则该超时会重试。超时独立应用于每次尝试，因此每次重试时计时器都会重置。

超时的尝试不会提交其缓冲写入。这可以防止状态更新或子任务调度在超时边界后泄漏。

配置节点超时

@[ timeout= ] 上的add_node参数限制了单个异步节点尝试可以运行的时间。传递一个数字（秒）或一个 timedelta, or a @TimeoutPolicy对象以更精细地控制运行和空闲超时。当超过限制时，LangGraph 会引发 [NodeTimeoutError，并让重试策略决定是否重试。

from langgraph.types import TimeoutPolicy

builder.add_node(
    "call_model",
    call_model,
    timeout=TimeoutPolicy(run_timeout=120, idle_timeout=30),
)

参见容错以了解完整的超时生命周期、空闲超时刷新源以及 runtime.heartbeat().

处理节点错误

@[ error_handler= ] 上的add_node参数注册了一个函数，该函数在节点失败且所有重试耗尽后运行。处理程序接收当前状态和一个类型化的 NodeError 带有失败上下文，并可通过 @[路由到恢复分支Command]:

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 routing.

在节点内访问执行信息

您可以通过以下方式访问执行标识和重试信息 runtime.execution_info。这会公开线程、运行和检查点标识符以及重试状态，无需读取 config directly.

属性	类型	描述
`thread_id`	`str \	None`
`run_id`	`str \	None`
`checkpoint_id`	`str`	当前执行的检查点 ID。
`checkpoint_ns`	`str`	当前执行的检查点命名空间。
`task_id`	`str`	当前执行的任务 ID。
`node_attempt`	`int`	当前执行尝试次数（从 1 开始）。
`node_first_attempt_time`	`float \	None`

访问线程和运行 ID

使用 execution_info 在节点内访问线程 ID、运行 ID 和其他身份字段：

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 检查当前尝试次数并在首次尝试失败后切换到后备方案：

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]
        # use a fallback on retries
        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`

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()

server_info is None 当图未在 LangGraph Server 上运行时（例如，在本地开发或测试期间）。

在节点内访问排除状态

当优雅关闭已被请求时， runtime.drain_requested is True。在节点内读取此值以在下一个超步边界之前跳过昂贵的计算工作：

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`	`True` if `RunControl.request_drain()` 已为此运行调用。
`drain_reason`	`str \	None`

添加节点缓存

节点缓存在需要避免重复操作的情况下很有用，例如执行耗时或成本高昂的操作时。LangGraph 允许您向图中的节点添加个性化缓存策略。

要配置缓存策略，请将 cache_policy 参数传递给 add_node 函数。在以下示例中，CachePolicy 对象实例化时设置了 120 秒的生存时间和默认 key_func 生成器。然后它与一个节点关联：

from langgraph.types import CachePolicy

builder.add_node(
    "node_name",
    node_function,
    cache_policy=CachePolicy(ttl=120),
)

然后，要为图启用节点级缓存，请设置 cache 编译图时的参数。下面的示例使用 InMemoryCache 来设置带有内存缓存的图，但 SqliteCache 也是可用的。

from langgraph.cache.memory import InMemoryCache

graph = builder.compile(cache=InMemoryCache())

创建步骤序列

这里我们演示如何构建一个简单的步骤序列。我们将展示：

如何构建顺序图
用于构造类似图的内置简写方法。

要添加一系列节点，我们使用 add_node 和 add_edge 方法，我们的图:

from langgraph.graph import START, StateGraph

builder = StateGraph(State)

# Add nodes
builder.add_node(step_1)
builder.add_node(step_2)
builder.add_node(step_3)

# Add edges
builder.add_edge(START, "step_1")
builder.add_edge("step_1", "step_2")
builder.add_edge("step_2", "step_3")

我们也可以使用内置简写方法 .add_sequence:

builder = StateGraph(State).add_sequence([step_1, step_2, step_3])
builder.add_edge(START, "step_1")

Why split application steps into a sequence with LangGraph?

LangGraph 让为您的应用程序添加底层持久层变得很容易。这允许在节点执行之间对状态进行检查点保存，因此您的 LangGraph 节点可以控制：

* 状态更新如何进行检查点保存
* 如何在 human-in-the-loop 工作流
* 中恢复中断我们如何使用 LangGraph 的时间旅行功能特性

它们还决定了执行步骤的流式传输方式，以及如何使用 Studio.

让我们演示一个端到端的示例。我们将创建一个包含三个步骤的序列：

在状态的键中填充一个值
更新相同的值
填充一个不同的值

让我们首先定义我们的状态。这控制着图的模式，也可以指定如何应用更新。请参阅使用 reducer 处理状态更新了解更多详情。

在我们的例子中，我们只跟踪两个值：

from typing_extensions import TypedDict

class State(TypedDict):
    value_1: str
    value_2: int

我们的节点是读取图状态并更新它的 Python 函数。该函数的第一个参数始终是状态：

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}

最后，我们定义图。我们使用 StateGraph 来定义一个在此状态上操作的图。

然后我们将使用 add_node 和 add_edge 来填充我们的图并定义其控制流。

from langgraph.graph import START, StateGraph

builder = StateGraph(State)

# Add nodes
builder.add_node(step_1)
builder.add_node(step_2)
builder.add_node(step_3)

# Add edges
builder.add_edge(START, "step_1")
builder.add_edge("step_1", "step_2")
builder.add_edge("step_2", "step_3")

建议

指定自定义名称 您可以使用 @[ 指定节点的自定义名称add_node]:

builder.add_node("my_node", step_1)

请注意：

* @[add_edge接受节点名称，对于函数默认为 node.__name__.
* 我们必须指定图的入口点。为此，我们使用 START 节点.
* 当没有更多节点可执行时，图停止运行。

接下来我们编译我们的图。这会对图的结构进行一些基本检查（例如，识别孤立的节点）。如果我们要通过 checkpointer为应用程序添加持久化功能，也会在此处传入。

graph = builder.compile()

LangGraph 提供了内置的图形可视化工具。让我们检查一下我们的序列。参见可视化您的图了解可视化详情。

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

!步骤序列图

让我们进行一个简单的调用：

graph.invoke({"value_1": "c"})

{'value_1': 'a b', 'value_2': 10}

请注意：

* 我们通过为单个状态键提供值来启动调用。我们必须始终至少为其中一个键提供值。
* 我们传入的值被第一个节点覆盖了。
* 第二个节点更新了该值。
* 第三个节点填充了一个不同的值。

建议

内置简写 langgraph>=0.2.46 包含一个内置简写 add_sequence 用于添加节点序列。您可以按如下方式编译相同的图：

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 原生支持节点的并行执行，这可以显著增强基于图的工作流程的性能。这种并行化通过扇出和扇入机制实现，利用标准边和条件_边。以下是一些示例，展示如何创建适合您的分支数据流。

并行运行图节点

在此示例中，我们从 Node A to B and C 扇出，然后扇入到 D。使用我们的状态，我们指定 reducer add 操作。这将组合或累加 State 中特定键的值，而不是简单覆盖现有值。对于列表，这意味着将新列表与现有列表连接起来。请参阅上面关于状态 reducer 的更多详情，了解如何使用 reducer 更新状态。

from typing import Annotated, Any
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    # The operator.add reducer fn makes this append-only
    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()

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

!并行执行图

使用 reducer，您可以看到每个节点添加的值都被累加了。

graph.invoke({"aggregate": []}, {"configurable": {"thread_id": "foo"}})

Adding "A" to []
Adding "B" to ['A']
Adding "C" to ['A']
Adding "D" to ['A', 'B', 'C']

Exception handling?

LangGraph 在 superstep中执行节点，这意味着虽然并行分支是并行执行的，但整个 superstep 是 **事务性的**。如果这些分支中的任何一个抛出异常， **所有** 更新都不会应用到状态（整个 superstep 报错）。

重要的是，当使用 checkpointer时，superstep 内成功节点的结果会被保存，恢复时不会重复执行。

如果您有容易出错的节点（也许想处理不稳定的 API 调用），LangGraph 提供了两种方式来解决这个问题：

您可以在节点内编写常规 Python 代码来捕获和处理异常。
您可以设置 **retry_策略** 来指示图重试抛出某些类型异常的节点。只有失败的分支会被重试，因此您无需担心执行冗余工作。

结合使用，您可以进行并行执行并完全控制异常处理。

建议

设置最大并发数 您可以通过设置来控制最大并发任务数 max_concurrency 在配置调用图时。

graph.invoke({"value_1": "c"}, {"configurable": {"max_concurrency": 10}})

延迟节点执行

延迟节点执行在您希望将节点的执行推迟到所有其他待处理任务完成后的情况下非常有用。这在分支具有不同长度时尤其重要，这在 map-reduce 等工作流中很常见。

上面的示例展示了当每个路径只有一个步骤时如何进行扇出和扇入。但如果一个分支有多个步骤怎么办？让我们添加一个节点 "b_2" 在 "b" branch:

from typing import Annotated, Any
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    # The operator.add reducer fn makes this append-only
    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()

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

!延迟执行图

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" 在同一超步中并发执行。我们设置 defer=True 在节点上 d 以便它不会执行，直到所有待处理任务完成。在这种情况下，这意味着 "d" 等待执行，直到整个 "b" 分支完成。

条件分支

如果您的扇出应根据状态在运行时变化，您可以使用 add_conditional_edges 使用图状态选择一个或多个路径。请参阅下面的示例，其中节点 a 生成一个决定后续节点的状态更新。

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]
    # Add a key to the state. We will set this key to determine
    # how we branch.
    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"]:
    # Fill in arbitrary logic here that uses the state
    # to determine the next node
    return state["which"]

builder.add_conditional_edges("a", conditional_edge)  # [!code highlight]

graph = builder.compile()

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

!条件分支图

result = graph.invoke({"aggregate": []})
print(result)

Adding "A" to []
Adding "C" to ['A']
{'aggregate': ['A', 'C'], 'which': 'c'}

建议

您的条件边可以路由到多个目标节点。例如：

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 和其他高级分支模式。下面是一个如何使用它的示例：

from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
from typing_extensions import TypedDict, Annotated


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()

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

!带扇出的 Map-reduce 图

# Call the graph: here we call it to generate a list of jokes
stream = graph.stream_events({"topic": "animals"}, version="v3")
for message in stream.messages:
    for token in message.text:
        print(token, end="", flush=True)

{'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'}}

创建和控制循环

在创建带有循环的图时，我们需要一种终止执行的机制。这通常通过添加一个条件边来实现，一旦达到某个终止条件就路由到 END 节点。

您还可以在调用或流式传输图时设置图的递归限制。递归限制设置了图在引发错误之前允许执行的 super-steps 数量。了解更多关于递归限制概念.

让我们考虑一个带有循环的简单图，以便更好地理解这些机制是如何工作的。

创建循环时，您可以包含一个指定终止条件的条件边：

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，您可以捕获并处理它：

from langgraph.errors import GraphRecursionError

try:
    graph.invoke(inputs, {"recursion_limit": 3})
except GraphRecursionError:
    print("Recursion Error")

让我们定义一个带有简单循环的图。请注意，我们使用条件边来实现终止条件。

from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    # The operator.add reducer fn makes this append-only
    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"]}

# Define nodes
builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)

# Define edges
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()

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

!简单循环图

这种架构类似于 ReAct 智能体，其中节点 "a" 是工具调用模型，而节点 "b" 代表工具。

在我们的 route 条件边中，我们指定应在 "aggregate" 列表在状态中超过阈值长度后结束。

调用图时，我们看到在节点 "a" 和 "b" 之间交替，直到达到终止条件后终止。

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']

设置递归限制

在某些应用中，我们可能无法保证达到给定的终止条件。在这种情况下，我们可以设置图的递归限制。这将在给定的 GraphRecursionError 超级步骤数后引发。我们随后可以捕获并处理此异常：

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

Extended example: return state on hitting recursion limit

不引发 GraphRecursionError，我们可以引入一个新的状态键来追踪距离达到递归限制前的剩余步数。然后我们可以使用此键来确定是否应该结束运行。

LangGraph 实现了一个特殊的 RemainingSteps 注解。在底层，它创建一个 ManagedValue 通道——一个状态通道，它只存在于图运行的持续时间内，之后就不存在了。

  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"]}

  # Define nodes
  builder = StateGraph(State)
  builder.add_node(a)
  builder.add_node(b)

  # Define edges
  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()

  # Test it out
  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']}

Extended example: loops with branches

为了更好地理解递归限制的工作原理，让我们考虑一个更复杂的示例。下面我们实现一个循环，但其中一个步骤会分叉到两个节点：

  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"]}

  # Define nodes
  builder = StateGraph(State)
  builder.add_node(a)
  builder.add_node(b)
  builder.add_node(c)
  builder.add_node(d)

  # Define edges
  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()

  from IPython.display import Image, display

  display(Image(graph.get_graph().draw_mermaid_png()))

!带分支的复杂循环图

这个图看起来很复杂，但可以概念化为超级步骤:

的循环
节点 A
节点 B
节点 C 和 D
...

我们有一个四个超级步骤的循环，其中节点 C 和 D 并发执行。

像之前一样调用图，我们看到在达到终止条件之前我们完成了两整"圈"：

  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']

但是，如果我们把递归限制设置为四，我们只完成一圈，因为每一圈是四个超级步骤：

  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-bound 代码时产生显著的性能提升（例如，向聊天模型提供商发出并发 API 请求）。

要将图的 sync 实现转换为 async 实现，您需要：

更新 nodes 使用 async def 而不是 def.
更新内部代码以使用 await appropriately.
使用以下方式调用图 .ainvoke or .astream 根据需要。

因为许多 LangChain 对象实现了 Runnable 协议它具有 async 所有 sync 方法的变体，通常相当快速地升级 sync 图为 async graph.

请参阅下面的示例。为了演示底层 LLM 的异步调用，我们将包含一个聊天模型：

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]

使用以下方式结合控制流和状态更新 `Command`

结合控制流（边）和状态更新（节点）可能很有用。例如，您可能希望在同一节点中同时执行状态更新并决定下一步进入哪个节点。LangGraph 通过从节点函数返回 Command 对象来提供一种方式：

def my_node(state: State) -> Command[Literal["my_other_node"]]:
    return Command(
        # state update
        update={"foo": "bar"},
        # control flow
        goto="my_other_node"
    )

我们在下面展示一个端到端示例。让我们创建一个包含 3 个节点的简单图：A、B 和 C。我们将首先执行节点 A，然后根据节点 A 的输出决定下一步是去节点 B 还是节点 C。

from typing_extensions import TypedDict, Literal
from langgraph.graph import StateGraph, START
from langgraph.types import Command

# Define graph state
class State(TypedDict):
    foo: str

# Define the nodes

def node_a(state: State) -> Command[Literal["node_b", "node_c"]]:
    print("Called A")
    value = random.choice(["b", "c"])
    # this is a replacement for a conditional edge function
    if value == "b":
        goto = "node_b"
    else:
        goto = "node_c"

    # note how Command allows you to BOTH update the graph state AND route to the next node
    return Command(
        # this is the state update
        update={"foo": value},
        # this is a replacement for an edge
        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。请注意，该图没有条件边用于路由！这是因为控制流是通过 Command 在内部定义的 node_a.

builder = StateGraph(State)
builder.add_edge(START, "node_a")
builder.add_node(node_a)
builder.add_node(node_b)
builder.add_node(node_c)
# NOTE: there are no edges between nodes A, B and C!

graph = builder.compile()

from IPython.display import display, Image

display(Image(graph.get_graph().draw_mermaid_png()))

!基于 Command 的图导航

如果我们多次运行该图，我们会看到它根据节点 A 中的随机选择采取不同的路径（A -> B 或 A -> C）。

graph.invoke({"foo": ""})

Called A
Called C

导航到父图中的节点

如果您正在使用子图，您可能想从子图中的一个节点导航到不同的子图（即父图中的不同节点）。为此，您可以指定 graph=Command.PARENT in Command:

def my_node(state: State) -> Command[Literal["my_other_node"]]:
    return Command(
        update={"foo": "bar"},
        goto="other_subgraph",  # where `other_subgraph` is a node in the parent graph
        graph=Command.PARENT
    )

让我们使用上面的例子来演示。我们将通过更改 nodeA 在上面的例子中，将其转换为一个单节点图，然后作为子图添加到我们的父图中。

from typing_extensions import Annotated

class State(TypedDict):
    # NOTE: we define a reducer here
    foo: Annotated[str, operator.add]  # [!code highlight]

def node_a(state: State):
    print("Called A")
    value = random.choice(["a", "b"])
    # this is a replacement for a conditional edge function
    if value == "a":
        goto = "node_b"
    else:
        goto = "node_c"

    # note how Command allows you to BOTH update the graph state AND route to the next node
    return Command(
        update={"foo": value},
        goto=goto,
        # this tells LangGraph to navigate to node_b or node_c in the parent graph
        # NOTE: this will navigate to the closest parent graph relative to the subgraph
        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")
    # NOTE: since we've defined a reducer, we don't need to manually append
    # new characters to existing 'foo' value. instead, reducer will append these
    # automatically (via 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()

graph.invoke({"foo": ""})

Called A
Called C

在工具内部使用

一个常见的用例是从工具内部更新图状态。例如，在客户支持应用程序中，您可能希望在对话开始时根据客户的帐号或ID查找客户信息。要从工具更新图状态，您可以返回 Command(update={"my_custom_key": "foo", "messages": [...]}) 从工具中返回：

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={
            # update the state keys
            "user_info": user_info,
            # update the message history
            "messages": [ToolMessage("Successfully looked up user information", tool_call_id=runtime.tool_call_id)]
        }
    )

如果您使用的是通过Command更新状态的工具，我们建议使用预构建的ToolNode，它会自动处理工具返回的Command对象并将其传播到图状态。如果您正在编写调用工具的自定义节点，您需要手动将工具返回的Command对象作为节点更新传播。

可视化您的图

这里我们演示如何可视化您创建的图。

您可以可视化任意图，包括状态图.

让我们通过绘制分形来玩点有趣的 :)。

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
    # Number of nodes to create at this level
    num_nodes = random.randint(1, 3)  # Adjust randomness as needed
    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)
        # Recursively add more nodes
        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:
            # End
            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)
    # Optional: set a finish point if required
    builder.add_edge(entry_point, END)  # or any specific node
    return builder.compile()

app = build_fractal_graph(3)

Mermaid

我们还可以将图类转换为 Mermaid 语法。

print(app.get_graph().draw_mermaid())

%%{init: {'flowchart': {'curve': 'linear'}}}%%
graph TD;
    tart__([<p>__start__</p>]):::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__([<p>__end__</p>]):::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 生成图表。

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

nest_asyncio.apply()  # Required for Jupyter Notebook to run async functions

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

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"
    )

设置

定义和更新状态

定义状态

更新状态

使用归约器处理状态更新

MessagesState

使用 Overwrite

定义输入和输出模式

在节点之间传递私有状态

使用 Pydantic 模型定义图状态

Serialization Behavior

Runtime Type Coercion

Working with Message Models

添加运行时配置

Extended example: specifying LLM at runtime

Extended example: specifying model and system message at runtime

添加重试策略

Extended example: customizing retry policies

设置节点超时

配置节点超时

处理节点错误

在节点内访问执行信息

访问线程和运行 ID

根据重试状态调整行为

在节点内访问服务器信息

在节点内访问排除状态

添加节点缓存

创建步骤序列

Why split application steps into a sequence with LangGraph?

创建分支

并行运行图节点

Exception handling?

延迟节点执行

条件分支

Map-Reduce 和 Send API

创建和控制循环

设置递归限制

Extended example: return state on hitting recursion limit

Extended example: loops with branches

异步

使用以下方式结合控制流和状态更新 Command

导航到父图中的节点

在工具内部使用

可视化您的图

Mermaid

PNG

使用 `Overwrite`

使用以下方式结合控制流和状态更新 `Command`