How to use the graph API¶
如何使用图 API¶

This guide demonstrates the basics of LangGraph's Graph API. It walks through state, as well as composing common graph structures such as sequences, branches, and loops. It also covers LangGraph's control features, including the Send API for map-reduce workflows and the Command API for combining state updates with "hops" across nodes.
本指南演示了 LangGraph 的 Graph API 的基础知识。它遍历状态 ，以及组成常见的图形结构，如序列 、 分支和循环 。它还涵盖了 LangGraph 的控制功能，包括用于映射缩减工作流的 Send API 和用于将状态更新与节点间的“跳”相结合的 Command API。

Setup¶  设置 ¶

Install langgraph:
安装 langgraph：

pip install -U langgraph

Define and update state¶
定义和更新状态 ¶

Here we show how to define and update state in LangGraph. We will demonstrate:
在这里，我们展示了如何在 LangGraph 中定义和更新状态 。我们将演示：

1. How to use state to define a graph's schema
   如何使用状态定义图的模式
2. How to use reducers to control how state updates are processed.
   如何使用 reducer 来控制状态更新的处理方式。

Define state¶
定义状态 ¶

State in LangGraph can be a TypedDict, Pydantic model, or dataclass. Below we will use TypedDict. See this section for detail on using Pydantic.
LangGraph 中的状态可以是 TypedDict、Pydantic 模型或类。下面我们将使用 TypedDict。有关使用 Pydantic 的详细信息，请参阅本节 。

By default, graphs will have the same input and output schema, and the state determines that schema. See this section for how to define distinct input and output schemas.
默认情况下，图形将具有相同的输入和输出模式，并且状态决定该模式。请参阅本节了解如何定义不同的输入和输出模式。

Let's consider a simple example using messages. This represents a versatile formulation of state for many LLM applications. See our concepts page for more detail.
让我们考虑一个使用消息的简单示例。这代表了许多 LLM 应用程序的通用状态公式。请参阅我们的概念页面了解更多详情。

^{API Reference: AnyMessage}
API 参考：AnyMessage

from langchain_core.messages import AnyMessage
from typing_extensions import TypedDict

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

This state tracks a list of message objects, as well as an extra integer field.
此状态跟踪消息对象列表以及一个额外的整数字段。

Update state¶
更新状态 ¶

Let's build an example graph with a single node. Our node is just a Python function that reads our graph's state and makes updates to it. The first argument to this function will always be the state:
让我们构建一个具有单个节点的示例图。我们的节点只是一个 Python 函数，它读取我们的图的状态并对其进行更新。这个函数的第一个参数将始终是状态：

^{API Reference: AIMessage}  API 参考：AIMessage

from langchain_core.messages import AIMessage

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

This node simply appends a message to our message list, and populates an extra field.
该节点只是将消息添加到我们的消息列表中，并填充一个额外的字段。

Let's next define a simple graph containing this node. We use StateGraph to define a graph that operates on this state. We then use add_node populate our graph.
接下来，让我们定义一个包含此节点的简单图。我们使用 StateGraph 来定义一个对这个状态进行操作的图。然后我们使用 add_node 填充我们的图。

^{API Reference: StateGraph}
API 参考：StateGraph

from langgraph.graph import StateGraph

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

LangGraph provides built-in utilities for visualizing your graph. Let's inspect our graph. See this section for detail on visualization.
LangGraph 提供了内置的实用程序来可视化您的图形。让我们看看我们的图表。有关可视化的详细信息，请参阅本节 。

from IPython.display import Image, display

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

In this case, our graph just executes a single node. Let's proceed with a simple invocation:
在这种情况下，我们的图只执行一个节点。让我们继续一个简单的调用：

^{API Reference: HumanMessage}
API 参考：HumanMessage

from langchain_core.messages import HumanMessage

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

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

Note that:  请注意：

• We kicked off invocation by updating a single key of the state.
  我们通过更新状态的一个键来启动调用。
• We receive the entire state in the invocation result.
  我们在调用结果中接收整个状态。

For convenience, we frequently inspect the content of message objects via pretty-print:
为了方便起见，我们经常通过 pretty-print 检查消息对象的内容：

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

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

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

Hello!

Process state updates with reducers¶
使用 reducer 处理状态更新 ¶

Each key in the state can have its own independent reducer function, which controls how updates from nodes are applied. If no reducer function is explicitly specified then it is assumed that all updates to the key should override it.
状态中的每个关键点都可以有自己独立的 reducer 函数，该函数控制如何应用来自节点的更新。如果没有明确指定 reducer 函数，则假定对键的所有更新都应该覆盖它。

For TypedDict state schemas, we can define reducers by annotating the corresponding field of the state with a reducer function.
对于 TypedDict 状态模式，我们可以通过使用 reducer 函数注释状态的相应字段来定义 reducer。

In the earlier example, our node updated the "messages" key in the state by appending a message to it. Below, we add a reducer to this key, such that updates are automatically appended:
在前面的例子中，我们的节点通过向状态中的“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]
    extra_field: int

Now our node can be simplified:
现在我们的节点可以简化：

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

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¶
MessagesState 状态 ¶

In practice, there are additional considerations for updating lists of messages:
在实践中，更新消息列表还有其他考虑因素：

• We may wish to update an existing message in the state.
  我们可能希望更新状态中的现有消息。
• We may want to accept short-hands for message formats, such as OpenAI format.
  我们可能希望接受消息格式的简写，例如 OpenAI 格式 。

LangGraph includes a built-in reducer add_messages that handles these considerations:
LangGraph 包含一个内置的 reduceradd_messages，它可以处理这些注意事项：

^{API Reference: add_messages}
API 参考：add_messages

from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    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"}

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

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

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

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

Hello!

This is a versatile representation of state for applications involving chat models. LangGraph includes a pre-built MessagesState for convenience, so that we can have:
这是一种通用的状态表示，适用于涉及聊天模型的应用程序。LangGraph 包括一个预先构建的 MessagesState，以便我们可以：

from langgraph.graph import MessagesState

class State(MessagesState):
    extra_field: int

Define input and output schemas¶
定义输入和输出模式 ¶

By default, StateGraph operates with a single schema, and all nodes are expected to communicate using that schema. However, it's also possible to define distinct input and output schemas for a graph.
默认情况下，StateGraph 使用单个模式运行，所有节点都应该使用该模式进行通信。 但是，也可以为图定义不同的输入和输出模式。

When distinct schemas are specified, an internal schema will still be used for communication between nodes.
当指定了不同的模式时，节点之间的通信仍将使用内部模式。
The input schema ensures that the provided input matches the expected structure, while the output schema filters the internal data to return only the relevant information according to the defined output schema.
输入模式确保提供的输入与预期的结构相匹配，而输出模式则根据定义的输出模式过滤内部数据，以仅返回相关信息。

Below, we'll see how to define distinct input and output schema.
下面，我们将看到如何定义不同的输入和输出模式。

^{API Reference: StateGraph | START | END}
API 参考：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'}

Notice that the output of invoke only includes the output schema.
请注意，invoke 的输出仅包括输出模式。

Pass private state between nodes¶
在节点之间传递私有状态 ¶

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.
这些私有数据与图的整体输入/输出无关，只应在某些节点之间共享。

Below, we'll create an example sequential graph consisting of three nodes (node_1, node_2 and node_3), where private data is passed between the first two steps (node_1 and node_2), while the third step (node_3) only has access to the public overall state.
下面，我们将创建一个由三个节点（node_1，node_2 和 node_3）组成的示例顺序图，其中私有数据在前两个步骤（node_1 和 node_2）之间传递，而第三个步骤（node_3）只能访问公共的整体状态。

^{API Reference: StateGraph | START | END}
API 参考：StateGraph|开始|端

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

Use Pydantic models for graph state¶
对图状态使用 Pydantic 模型 ¶

A StateGraph accepts a state_schema argument on initialization that specifies the "shape" of the state that the nodes in the graph can access and update.
StateGraph 在初始化时接受 state_schema 参数，该参数指定图中的节点可以访问和更新的状态的“形状”。

In our examples, we typically use a python-native TypedDict or dataclass for state_schema, but state_schema can be any type.
在我们的示例中，我们通常使用 Python 原生的 TypedDict 或 dataclass 作为 state_schema，但 state_schema 可以是任何类型 。

Here, we'll see how a Pydantic BaseModel can be used for state_schema to add run-time validation on inputs.
在这里，我们将看到如何将 Pydantic BaseModel 用于 state_schema，以在输入上添加运行时验证。

^{API Reference: StateGraph | START | END}
API 参考：StateGraph|开始|端

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

Invoke the graph with an invalid input
使用无效输入删除图表

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

See below for additional features of Pydantic model state:
下面是 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)}")

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

from langgraph.graph import StateGraph, START, END
from pydantic import BaseModel
from langchain_core.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}")

Add runtime configuration¶
添加运行时配置 ¶

Sometimes you want to be able to configure your graph when calling it. For example, you might want to be able to specify what LLM or system prompt to use at runtime, without polluting the graph state with these parameters.
有时候，你希望在调用图时能够配置它。例如，你可能希望能够指定在运行时使用什么 LLM 或系统提示符，而不会用这些参数污染图的状态 。

To add runtime configuration:
要添加运行时配置，请执行以下操作：

1. Specify a schema for your configuration
   为您的配置指定架构
2. Add the configuration to the function signature for nodes or conditional edges
   将配置添加到节点或条件边的函数签名
3. Pass the configuration into the graph.
   将配置传递到图表中。

See below for a simple example:
下面是一个简单的例子：

^{API Reference: END | StateGraph | START}
API 参考：END|状态图|开始

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]):
    if runtime.context["my_runtime_value"] == "a":
        return {"my_state_value": 1}
    elif runtime.context["my_runtime_value"] == "b":
        return {"my_state_value": 2}
    else:
        raise ValueError("Unknown values.")

builder = StateGraph(State, context_schema=ContextSchema)
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"}))
print(graph.invoke({}, context={"my_runtime_value": "b"}))

{'my_state_value': 1}
{'my_state_value': 2}

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("anthropic:claude-3-5-haiku-latest"),
    "openai": init_chat_model("openai:gpt-4.1-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]})["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-3-5-haiku-20241022
gpt-4.1-mini-2025-04-14

from dataclasses import dataclass
from typing import Optional
from langchain.chat_models import init_chat_model
from langchain_core.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("anthropic:claude-3-5-haiku-latest"),
    "openai": init_chat_model("openai:gpt-4.1-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?

Add retry policies¶
添加重试策略 ¶

There are many use cases where you may wish for your node to have a custom retry policy, for example if you are calling an API, querying a database, or calling an LLM, etc. LangGraph lets you add retry policies to nodes.
在许多用例中，您可能希望您的节点具有自定义重试策略，例如，如果您正在调用 API，查询数据库或调用 LLM 等。LangGraph 允许您向节点添加重试策略。

To configure a retry policy, pass the retry_policy parameter to the add_node. The retry_policy parameter takes in a RetryPolicy named tuple object. Below we instantiate a RetryPolicy object with the default parameters and associate it with a node:
要配置重试策略，请将 retry_policy 参数传递给 add_node。retry_policy 参数接受一个名为 RetryPolicy 的元组对象。下面我们使用默认参数实例化 RetryPolicy 对象，并将其与节点关联：

from langgraph.pregel import RetryPolicy

builder.add_node(
    "node_name",
    node_function,
    retry_policy=RetryPolicy(),
)

By default, the retry_on parameter uses the default_retry_on function, which retries on any exception except for the following:
默认情况下，retry_on 参数使用 default_retry_on 函数，该函数会对除以下情况以外的任何异常进行重试：

• ValueError
• TypeError
• ArithmeticError
• ImportError
• LookupError
• NameError
• SyntaxError
• RuntimeError
• ReferenceError
• StopIteration
• StopAsyncIteration
• OSError

In addition, for exceptions from popular http request libraries such as requests and httpx it only retries on 5xx status codes.
此外，对于常见的 http 请求库（如 requests 和 httpx） 的异常，它只会在 5xx 状态码上重试。

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.pregel import RetryPolicy
from langchain_community.utilities import SQLDatabase
from langchain_core.messages import AIMessage

db = SQLDatabase.from_uri("sqlite:///:memory:")
model = init_chat_model("anthropic:claude-3-5-haiku-latest")

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

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

Add node caching¶
添加节点缓存 ¶

Node caching is useful in cases where you want to avoid repeating operations, like when doing something expensive (either in terms of time or cost). LangGraph lets you add individualized caching policies to nodes in a graph.
在您希望避免重复操作的情况下，节点缓存非常有用，例如在执行一些昂贵的操作（无论是在时间还是成本方面）时。LangGraph 允许您将个性化的缓存策略添加到图中的节点。

To configure a cache policy, pass the cache_policy parameter to the add_node function. In the following example, a CachePolicy object is instantiated with a time to live of 120 seconds and the default key_func generator. Then it is associated with a node:
要配置缓存策略，请将该高速缓存_policy 参数传递给 add_node 函数。在下面的示例中，CachePolicy 对象被实例化，生存时间为 120 秒，并使用默认的 key_func 生成器。然后它与一个节点关联：

from langgraph.types import CachePolicy

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

Then, to enable node-level caching for a graph, set the cache argument when compiling the graph. The example below uses InMemoryCache to set up a graph with in-memory cache, but SqliteCache is also available.
然后，要为图形启用节点级缓存，请在编译图形时设置该高速缓存参数。下面的示例使用 InMemoryCache 来设置一个带有内存缓存的图，但 SqliteCache 也可用。

from langgraph.cache.memory import InMemoryCache

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

Create a sequence of steps¶
创建一系列步骤 ¶

Here we demonstrate how to construct a simple sequence of steps. We will show:
在这里，我们演示如何构建一个简单的步骤序列。我们将展示：

1. How to build a sequential graph
   如何构建序列图
2. Built-in short-hand for constructing similar graphs.
   内置的速记用于构建类似的图形。

To add a sequence of nodes, we use the .add_node and .add_edge methods of our graph:
为了添加一系列节点，我们使用图的 .add_node 和 .add_edge 方法：

^{API Reference: START | StateGraph}
API 参考：START|状态图

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

We can also use the built-in shorthand .add_sequence:
我们也可以使用内置的简写 .add_sequence：

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

Let's demonstrate an end-to-end example. We will create a sequence of three steps:
让我们演示一个端到端的示例。我们将创建一个包含三个步骤的序列：

1. Populate a value in a key of the state
   在状态的键中填充值
2. Update the same value
   更新相同的值
3. Populate a different value
   填充不同的值

Let's first define our state. This governs the schema of the graph, and can also specify how to apply updates. See this section for more detail.
首先，让我们来定义我们的国家 。这控制了图的模式 ，还可以指定如何应用更新。有关详细信息，请参阅本节 。

In our case, we will just keep track of two values:
在我们的例子中，我们将只跟踪两个值：

from typing_extensions import TypedDict

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

Our nodes are just Python functions that read our graph's state and make updates to it. The first argument to this function will always be the state:
我们的节点只是读取我们的图的状态并对其进行更新的 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}

Finally, we define the graph. We use StateGraph to define a graph that operates on this state.
最后，我们定义了图。我们使用 StateGraph 来定义一个对这个状态进行操作的图。

We will then use add_node and add_edge to populate our graph and define its control flow.
然后我们将使用 add_node 和 add_edge 来填充我们的图并定义其控制流。

^{API Reference: START | StateGraph}
API 参考：START|状态图

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

builder.add_node("my_node", step_1)

Note that:  请注意：

• .add_edge takes the names of nodes, which for functions defaults to node.__name__.
  . add_edge 接受节点的名称，对于函数，默认为 node。__ 名称__。
• We must specify the entry point of the graph. For this we add an edge with the START node.
  我们必须指定图的入口点。为此，我们添加一个带有 START 节点的边。
• The graph halts when there are no more nodes to execute.
  当没有更多的节点要执行时，图形停止。

We next compile our graph. This provides a few basic checks on the structure of the graph (e.g., identifying orphaned nodes). If we were adding persistence to our application via a checkpointer, it would also be passed in here.
接下来，我们编译图表。这提供了对图的结构的一些基本检查（例如，识别孤立节点）。如果我们通过检查指针向应用程序添加持久性，它也会在这里传递。

graph = builder.compile()

LangGraph provides built-in utilities for visualizing your graph. Let's inspect our sequence. See this guide for detail on visualization.
LangGraph 提供了内置的实用程序来可视化您的图形。让我们检查我们的序列。有关可视化的详细信息，请参阅本指南 。

from IPython.display import Image, display

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

Let's proceed with a simple invocation:
让我们继续一个简单的调用：

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

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

Note that:  请注意：

• We kicked off invocation by providing a value for a single state key. We must always provide a value for at least one key.
  我们通过为单个状态键提供一个值来启动调用。我们必须始终为至少一个键提供值。
• The value we passed in was overwritten by the first node.
  我们传入的值被第一个节点覆盖了。
• The second node updated the value.
  第二个节点更新了该值。
• The third node populated a different value.
  第三个节点填充了不同的值。

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

graph = builder.compile()

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

Create branches¶
创建分支 ¶

Parallel execution of nodes is essential to speed up overall graph operation. LangGraph offers native support for parallel execution of nodes, which can significantly enhance the performance of graph-based workflows.
节点的并行执行对于加速整个图形操作至关重要。LangGraph 为节点的并行执行提供了原生支持，这可以显着提高基于图的工作流的性能。
This parallelization is achieved through fan-out and fan-in mechanisms, utilizing both standard edges and conditional_edges. Below are some examples showing how to add create branching dataflows that work for you.
这种并行化是通过扇出和扇入机制实现的，同时利用标准边缘和条件边缘 。下面是一些示例，展示了如何添加适合您的创建分支流程。

Run graph nodes in parallel¶
并行运行图节点 ¶

In this example, we fan out from Node A to B and C and then fan in to D. With our state, we specify the reducer add operation. This will combine or accumulate values for the specific key in the State, rather than simply overwriting the existing value. For lists, this means concatenating the new list with the existing list. See the above section on state reducers for more detail on updating state with reducers.
在本例中，我们从节点 A 扇出到 B 和 C，然后扇入到 D。对于我们的状态， 我们指定 reducer add 操作 。这将为 State 中的特定键组合联合收割机或累积值，而不是简单地将现有值重新组合。对于列表，这意味着将新列表与现有列表连接起来。 有关使用 reducer 更新状态的更多详细信息，请参见上面关于状态 reducer 的部分。

^{API Reference: StateGraph | START | END}
API 参考：StateGraph|开始|端

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

With the reducer, you can see that the values added in each node are accumulated.
使用 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']

Defer node execution¶
延迟节点执行 ¶

Deferring node execution is useful when you want to delay the execution of a node until all other pending tasks are completed. This is particularly relevant when branches have different lengths, which is common in workflows like map-reduce flows.
当您希望延迟节点的执行直到所有其他挂起任务完成时，延迟节点执行非常有用。当分支具有不同的长度时，这一点尤其重要，这在诸如 map-reduce 流之类的工作流中很常见。

The above example showed how to fan-out and fan-in when each path was only one step. But what if one branch had more than one step? Let's add a node "b_2" in the "b" branch:
上面的例子展示了当每条路径只有一步时如何扇出和扇入。但如果一个分支有多个步骤呢？让我们在 “B” 分支中添加一个节点 “b_2”：

^{API Reference: StateGraph | START | END}
API 参考：StateGraph|开始|端

import operator
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)
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']

In the above example, nodes "b" and "c" are executed concurrently in the same superstep. We set defer=True on node d so it will not execute until all pending tasks are finished. In this case, this means that "d" waits to execute until the entire "b" branch is finished.
在上述示例中，节点 “B” 和 “c” 在同一超步中同时执行。我们在节点 d 上设置 defer=True，这样它就不会执行，直到所有挂起的任务都完成。在这种情况下，这意味着 “d” 等待执行，直到整个 “B” 分支完成。

Conditional branching¶
条件分支 ¶

If your fan-out should vary at runtime based on the state, you can use add_conditional_edges to select one or more paths using the graph state. See example below, where node a generates a state update that determines the following node.
如果您的扇出在运行时应该根据状态而变化，则可以使用 add_conditional_edges 来使用图形状态选择一个或多个路径。参见下面的示例，其中节点 a 生成状态更新，该状态更新确定以下节点。

^{API Reference: StateGraph | START | END}
API 参考：StateGraph|开始|端

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

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)

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 and the Send API¶
Map-Reduce 和发送 API¶

LangGraph supports map-reduce and other advanced branching patterns using the Send API. Here is an example of how to use it:
LangGraph 支持使用 Send API 的 map-reduce 和其他高级分支模式。下面是一个如何使用它的例子：

^{API Reference: StateGraph | START | END | Send}
API 参考：StateGraph|开始|端|发送

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)
builder.add_edge("generate_topics", END)
graph = builder.compile()

from IPython.display import Image, display

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

# Call the graph: here we call it to generate a list of jokes
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'}}

Create and control loops¶
创建和控制循环 ¶

When creating a graph with a loop, we require a mechanism for terminating execution. This is most commonly done by adding a conditional edge that routes to the END node once we reach some termination condition.
当创建一个带有循环的图时，我们需要一个终止执行的机制。这通常是通过添加一个条件边来完成的，一旦我们达到某个终止条件，该条件边就会路由到 END 节点。

You can also set the graph recursion limit when invoking or streaming the graph. The recursion limit sets the number of supersteps that the graph is allowed to execute before it raises an error. Read more about the concept of recursion limits here.
您还可以在调用或流式传输图形时设置图形递归限制。递归限制设置在引发错误之前允许图形执行的超步数。 在这里阅读更多关于递归极限的概念。

Let's consider a simple graph with a loop to better understand how these mechanisms work.
让我们考虑一个带有循环的简单图，以更好地理解这些机制是如何工作的。

When creating a loop, you can include a conditional edge that specifies a termination condition:
创建循环时，可以包含指定终止条件的条件边：

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

To control the recursion limit, specify "recursion_limit" in the config. This will raise a GraphRecursionError, which you can catch and handle:
要控制递归限制，请在配置中指定 “recursion_limit”。这将引发一个 GraphRecursionError，您可以捕获并处理它：

from langgraph.errors import GraphRecursionError

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

Let's define a graph with a simple loop. Note that we use a conditional edge to implement a termination condition.
让我们用一个简单的循环定义一个图。请注意，我们使用条件边来实现终止条件。

^{API Reference: StateGraph | START | END}
API 参考：StateGraph|开始|端

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

This architecture is similar to a ReAct agent in which node "a" is a tool-calling model, and node "b" represents the tools.
此架构类似于 ReAct 代理 ，其中节点 “a” 是工具调用模型，节点 “B” 表示工具。

In our route conditional edge, we specify that we should end after the "aggregate" list in the state passes a threshold length.
在我们的路由条件边中，我们指定在状态中的 “聚合” 列表超过阈值长度后结束。

Invoking the graph, we see that we alternate between nodes "a" and "b" before terminating once we reach the termination condition.
通过绘制图，我们可以看到，一旦达到终止条件，就在终止之前在节点 “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']

Impose a recursion limit¶
强加一个递归限制 ¶

In some applications, we may not have a guarantee that we will reach a given termination condition. In these cases, we can set the graph's recursion limit. This will raise a GraphRecursionError after a given number of supersteps. We can then catch and handle this exception:
在某些应用程序中，我们可能无法保证我们将达到给定的终止条件。在这些情况下，我们可以设置图的递归限制 。这将在给定的超级步数后引发 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

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

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

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

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