图 (Graphs)

langgraph.graph.state.StateGraph

Bases: Generic[StateT, ContextT, InputT, OutputT]

一种图，其节点通过读取和写入共享状态进行通信。每个节点的签名为 State -> Partial.

每个状态键都可以选择性地使用一个归约函数进行注解，该函数将用于聚合从多个节点接收到的该键的值。归约函数的签名为 (Value, Value) -> Value。

参数	描述
state_schema	定义状态的模式类。 类型: type[StateT]
context_schema	定义运行时上下文的模式类。使用它向您的节点暴露不可变的上下文数据，例如 user_id、db_conn 等。 类型： type[ContextT] | None 默认值： None
input_schema	定义图输入的模式类。 类型: type[InputT] | None 默认值: None
output_schema	定义图输出的模式类。 类型: type[OutputT] | None 默认值: None

config_schema 已弃用

config_schema 参数在 v0.6.0 中已弃用，并将在 v2.0.0 中移除支持。请改用 context_schema 来指定运行范围内的上下文模式。

示例

from langchain_core.runnables import RunnableConfig
from typing_extensions import Annotated, TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph
from langgraph.runtime import Runtime


def reducer(a: list, b: int | None) -> list:
    if b is not None:
        return a + [b]
    return a


class State(TypedDict):
    x: Annotated[list, reducer]


class Context(TypedDict):
    r: float


graph = StateGraph(state_schema=State, context_schema=Context)


def node(state: State, runtime: Runtime[Context]) -> dict:
    r = runtime.context.get("r", 1.0)
    x = state["x"][-1]
    next_value = x * r * (1 - x)
    return {"x": next_value}


graph.add_node("A", node)
graph.set_entry_point("A")
graph.set_finish_point("A")
compiled = graph.compile()

step1 = compiled.invoke({"x": 0.5}, context={"r": 3.0})
# {'x': [0.5, 0.75]}

方法	描述
add_node	向状态图中添加一个新节点。
add_edge	添加一条从起始节点（或起始节点列表）到结束节点的有向边。
add_conditional_edges	添加一条从起始节点到任意数量目标节点的条件边。
add_sequence	添加一个将按所提供顺序执行的节点序列。
compile	将状态图编译成一个 CompiledStateGraph 对象。

add_node

add_node(
    node: str | StateNode[NodeInputT, ContextT],
    action: StateNode[NodeInputT, ContextT] | None = None,
    *,
    defer: bool = False,
    metadata: dict[str, Any] | None = None,
    input_schema: type[NodeInputT] | None = None,
    retry_policy: RetryPolicy | Sequence[RetryPolicy] | None = None,
    cache_policy: CachePolicy | None = None,
    destinations: dict[str, str] | tuple[str, ...] | None = None,
    **kwargs: Unpack[DeprecatedKwargs],
) -> Self

向状态图中添加一个新节点。

参数	描述
node	该节点将运行的函数或 runnable。如果提供的是字符串，它将被用作节点名称，而 action 将被用作函数或 runnable。 类型: str | StateNode[NodeInputT, ContextT]
action	与节点关联的动作。如果 node 是一个字符串（节点名称），它将被用作节点函数或 runnable。 类型: StateNode[NodeInputT, ContextT] | None 默认值: None
defer	是否将节点的执行推迟到运行即将结束时。 类型： bool 默认值： False
metadata	与节点关联的元数据。 类型: dict[str, Any] | None 默认值: None
input_schema	节点的输入模式。（默认：图的状态模式） 类型: type[NodeInputT] | None 默认值: None
retry_policy	节点的重试策略。如果提供了一个序列，将应用第一个匹配的策略。 类型: RetryPolicy | Sequence[RetryPolicy] | None 默认值: None
cache_policy	节点的缓存策略。 类型: CachePolicy | None 默认值: None
destinations	指示节点可以路由到的目标。这对于返回 Command 对象的无边图很有用。如果提供了一个 dict，键将用作目标节点名称，值将用作边的标签。如果提供了一个 tuple，值将用作目标节点名称。 注意 这仅用于图的渲染，对图的执行没有任何影响。 类型: dict[str, str] | tuple[str, ...] | None 默认值: None

示例

from typing_extensions import TypedDict

from langchain_core.runnables import RunnableConfig
from langgraph.graph import START, StateGraph


class State(TypedDict):
    x: int


def my_node(state: State, config: RunnableConfig) -> State:
    return {"x": state["x"] + 1}


builder = StateGraph(State)
builder.add_node(my_node)  # node name will be 'my_node'
builder.add_edge(START, "my_node")
graph = builder.compile()
graph.invoke({"x": 1})
# {'x': 2}

自定义名称

builder = StateGraph(State)
builder.add_node("my_fair_node", my_node)
builder.add_edge(START, "my_fair_node")
graph = builder.compile()
graph.invoke({"x": 1})
# {'x': 2}

返回	描述
Self	状态图的实例，允许方法链式调用。 类型: Self

add_edge

add_edge(start_key: str | list[str], end_key: str) -> Self

添加一条从起始节点（或起始节点列表）到结束节点的有向边。

当提供单个起始节点时，图将等待该节点完成后再执行结束节点。当提供多个起始节点时，图将等待所有起始节点完成后再执行结束节点。

参数	描述
start_key	边的起始节点的键。 类型: str | list[str]
end_key	边的结束节点的键。 类型： str

引发	描述
ValueError	如果起始键是 'END'，或者如果起始键或结束键在图中不存在。

返回	描述
Self	状态图的实例，允许方法链式调用。 类型: Self

add_conditional_edges

add_conditional_edges(
    source: str,
    path: Callable[..., Hashable | Sequence[Hashable]]
    | Callable[..., Awaitable[Hashable | Sequence[Hashable]]]
    | Runnable[Any, Hashable | Sequence[Hashable]],
    path_map: dict[Hashable, str] | list[str] | None = None,
) -> Self

添加一条从起始节点到任意数量目标节点的条件边。

参数	描述
source	起始节点。此条件边将在退出此节点时运行。 类型： str
path	确定下一个节点或多个节点的可调用对象。如果未指定 path_map，它应返回一个或多个节点。如果返回 'END'，图将停止执行。 类型: Callable[..., Hashable | Sequence[Hashable]] | Callable[..., Awaitable[Hashable | Sequence[Hashable]]] | Runnable[Any, Hashable | Sequence[Hashable]]
path_map	可选的路径到节点名称的映射。如果省略，path 返回的路径应为节点名称。 类型: dict[Hashable, str] | list[str] | None 默认值: None

返回	描述
Self	图的实例，允许方法链式调用。 类型: Self

警告

如果在 path 函数的返回值上没有类型提示（例如，-> Literal["foo", "__end__"]:）或没有 path_map，图的可视化将假定该边可以转换到图中的任何节点。

add_sequence

add_sequence(
    nodes: Sequence[
        StateNode[NodeInputT, ContextT] | tuple[str, StateNode[NodeInputT, ContextT]]
    ],
) -> Self

添加一个将按所提供顺序执行的节点序列。

参数	描述
nodes	一个由 StateNode（接受 state 参数的可调用对象）或 (name, StateNode) 元组组成的序列。如果未提供名称，将从节点对象（例如 Runnable 或 Callable 名称）推断名称。每个节点将按提供的顺序执行。 类型: Sequence[StateNode[NodeInputT, ContextT] | tuple[str, StateNode[NodeInputT, ContextT]]]

引发	描述
ValueError	如果序列为空。
ValueError	如果序列包含重复的节点名称。

返回	描述
Self	状态图的实例，允许方法链式调用。 类型: Self

compile

compile(
    checkpointer: Checkpointer = None,
    *,
    cache: BaseCache | None = None,
    store: BaseStore | None = None,
    interrupt_before: All | list[str] | None = None,
    interrupt_after: All | list[str] | None = None,
    debug: bool = False,
    name: str | None = None,
) -> CompiledStateGraph[StateT, ContextT, InputT, OutputT]

将状态图编译成一个 CompiledStateGraph 对象。

编译后的图实现了 Runnable 接口，可以被调用、流式传输、批处理和异步运行。

参数	描述
checkpointer	一个检查点保存器对象或标志。如果提供，此 Checkpointer 将作为图的完全版本化的“短期记忆”，允许它从任何点暂停、恢复和重播。如果为 None，当用作子图时，它可能会继承父图的检查点器。如果为 False，它将不会使用或继承任何检查点器。 类型: Checkpointer 默认值: None
interrupt_before	一个可选的在执行前中断的节点名称列表。 类型: All | list[str] | None 默认值: None
interrupt_after	一个可选的在执行后中断的节点名称列表。 类型: All | list[str] | None 默认值: None
debug	一个指示是否启用调试模式的标志。 类型： bool 默认值： False
name	用于已编译图的名称。 类型: str | None 默认值: None

返回	描述
CompiledStateGraph	已编译的状态图。 类型: CompiledStateGraph[StateT, ContextT, InputT, OutputT]

langgraph.graph.state.CompiledStateGraph

Bases: Pregel[StateT, ContextT, InputT, OutputT], Generic[StateT, ContextT, InputT, OutputT]

方法	描述
stream	为单个输入流式传输图的步骤。
astream	为单个输入异步流式传输图的步骤。
invoke	使用单个输入和配置运行图。
ainvoke	在单个输入上异步调用图。
get_state	获取图的当前状态。
aget_state	获取图的当前状态。
get_state_history	获取图的状态历史记录。
aget_state_history	异步获取图的状态历史记录。
update_state	使用给定值更新图的状态，就像它们来自
aupdate_state	异步使用给定值更新图的状态，就像它们来自
bulk_update_state	批量应用对图状态的更新。需要设置检查点器。
abulk_update_state	异步批量应用对图状态的更新。需要设置检查点器。
get_graph	返回计算图的可绘制表示。
aget_graph	返回计算图的可绘制表示。
get_subgraphs	获取图的子图。
aget_subgraphs	获取图的子图。
with_config	创建具有更新配置的 Pregel 对象的副本。

stream

stream(
    input: InputT | Command | None,
    config: RunnableConfig | None = None,
    *,
    context: ContextT | None = None,
    stream_mode: StreamMode | Sequence[StreamMode] | None = None,
    print_mode: StreamMode | Sequence[StreamMode] = (),
    output_keys: str | Sequence[str] | None = None,
    interrupt_before: All | Sequence[str] | None = None,
    interrupt_after: All | Sequence[str] | None = None,
    durability: Durability | None = None,
    subgraphs: bool = False,
    debug: bool | None = None,
    **kwargs: Unpack[DeprecatedKwargs],
) -> Iterator[dict[str, Any] | Any]

为单个输入流式传输图的步骤。

参数	描述
输入	图的输入。 类型: InputT | Command | None
配置	用于运行的配置。 类型： RunnableConfig | None 默认值： None
context	用于运行的静态上下文。 在 0.6.0 版本中添加 类型: ContextT | None 默认值: None
stream_mode	流式输出的模式，默认为 self.stream_mode。选项有 "values"：在每个步骤后发出状态中的所有值，包括中断。与函数式 API 一起使用时，值在工作流结束时发出一次。 "updates"：仅在每个步骤后发出节点或任务返回的节点或任务名称及更新。如果在同一步骤中进行了多次更新（例如运行了多个节点），则这些更新将分别发出。 "custom": 使用 StreamWriter 从节点或任务内部发出自定义数据。 "messages": 逐个 token 地发出 LLM 消息，并附带节点或任务内任何 LLM 调用的元数据。将以 2 元组 (LLM token, metadata) 的形式发出。 "checkpoints"：在创建检查点时发出一个事件，格式与 get_state() 返回的格式相同。 "tasks"：在任务开始和结束时发出事件，包括其结果和错误。 您可以将一个列表作为 stream_mode 参数传递，以同时流式传输多种模式。流式输出将是 (mode, data) 的元组。 更多详情请参阅 LangGraph 流式处理指南。 类型: StreamMode | Sequence[StreamMode] | None 默认值: None
print_mode	接受与 stream_mode 相同的值，但仅将输出打印到控制台，用于调试目的。不以任何方式影响图的输出。 类型: StreamMode | Sequence[StreamMode] 默认值: ()
output_keys	要流式传输的键，默认为所有非上下文通道。 类型: str | Sequence[str] | None 默认值: None
interrupt_before	在这些节点前中断，默认为图中的所有节点。 类型: All | Sequence[str] | None 默认值: None
interrupt_after	在这些节点后中断，默认为图中的所有节点。 类型: All | Sequence[str] | None 默认值: None
durability	图执行的持久性模式，默认为 "async"。选项有 "sync": 更改在下一步开始前同步持久化。 "async": 更改在下一步执行时异步持久化。 "exit": 更改仅在图退出时持久化。 类型: Durability | None 默认值: None
subgraphs	是否从子图内部流式传输事件，默认为 False。如果为 True，事件将以 (namespace, data) 的元组形式发出，或者如果 stream_mode 是一个列表，则以 (namespace, mode, data) 的形式发出，其中 namespace 是一个包含调用子图的节点路径的元组，例如 ("parent_node:<task_id>", "child_node:<task_id>")。 更多详情请参阅 LangGraph 流式处理指南。 类型： bool 默认值： False

YIELDS	描述
dict[str, Any] | Any	图中每一步的输出。输出形状取决于 stream_mode。

astream async

astream(
    input: InputT | Command | None,
    config: RunnableConfig | None = None,
    *,
    context: ContextT | None = None,
    stream_mode: StreamMode | Sequence[StreamMode] | None = None,
    print_mode: StreamMode | Sequence[StreamMode] = (),
    output_keys: str | Sequence[str] | None = None,
    interrupt_before: All | Sequence[str] | None = None,
    interrupt_after: All | Sequence[str] | None = None,
    durability: Durability | None = None,
    subgraphs: bool = False,
    debug: bool | None = None,
    **kwargs: Unpack[DeprecatedKwargs],
) -> AsyncIterator[dict[str, Any] | Any]

为单个输入异步流式传输图的步骤。

参数	描述
输入	图的输入。 类型: InputT | Command | None
配置	用于运行的配置。 类型： RunnableConfig | None 默认值： None
context	用于运行的静态上下文。 在 0.6.0 版本中添加 类型: ContextT | None 默认值: None
stream_mode	流式输出的模式，默认为 self.stream_mode。选项有 "values"：在每个步骤后发出状态中的所有值，包括中断。与函数式 API 一起使用时，值在工作流结束时发出一次。 "updates"：仅在每个步骤后发出节点或任务返回的节点或任务名称及更新。如果在同一步骤中进行了多次更新（例如运行了多个节点），则这些更新将分别发出。 "custom": 使用 StreamWriter 从节点或任务内部发出自定义数据。 "messages": 逐个 token 地发出 LLM 消息，并附带节点或任务内任何 LLM 调用的元数据。将以 2 元组 (LLM token, metadata) 的形式发出。 "debug": 为每一步发出包含尽可能多信息的调试事件。 您可以将一个列表作为 stream_mode 参数传递，以同时流式传输多种模式。流式输出将是 (mode, data) 的元组。 更多详情请参阅 LangGraph 流式处理指南。 类型: StreamMode | Sequence[StreamMode] | None 默认值: None
print_mode	接受与 stream_mode 相同的值，但仅将输出打印到控制台，用于调试目的。不以任何方式影响图的输出。 类型: StreamMode | Sequence[StreamMode] 默认值: ()
output_keys	要流式传输的键，默认为所有非上下文通道。 类型: str | Sequence[str] | None 默认值: None
interrupt_before	在这些节点前中断，默认为图中的所有节点。 类型: All | Sequence[str] | None 默认值: None
interrupt_after	在这些节点后中断，默认为图中的所有节点。 类型: All | Sequence[str] | None 默认值: None
durability	图执行的持久性模式，默认为 "async"。选项有 "sync": 更改在下一步开始前同步持久化。 "async": 更改在下一步执行时异步持久化。 "exit": 更改仅在图退出时持久化。 类型: Durability | None 默认值: None
subgraphs	是否从子图内部流式传输事件，默认为 False。如果为 True，事件将以 (namespace, data) 的元组形式发出，或者如果 stream_mode 是一个列表，则以 (namespace, mode, data) 的形式发出，其中 namespace 是一个包含调用子图的节点路径的元组，例如 ("parent_node:<task_id>", "child_node:<task_id>")。 更多详情请参阅 LangGraph 流式处理指南。 类型： bool 默认值： False

YIELDS	描述
AsyncIterator[dict[str, Any] | Any]	图中每一步的输出。输出形状取决于 stream_mode。

invoke

invoke(
    input: InputT | Command | None,
    config: RunnableConfig | None = None,
    *,
    context: ContextT | None = None,
    stream_mode: StreamMode = "values",
    print_mode: StreamMode | Sequence[StreamMode] = (),
    output_keys: str | Sequence[str] | None = None,
    interrupt_before: All | Sequence[str] | None = None,
    interrupt_after: All | Sequence[str] | None = None,
    durability: Durability | None = None,
    **kwargs: Any,
) -> dict[str, Any] | Any

使用单个输入和配置运行图。

参数	描述
输入	图的输入数据。它可以是字典或任何其他类型。 类型: InputT | Command | None
配置	图运行的配置。 类型： RunnableConfig | None 默认值： None
context	用于运行的静态上下文。 在 0.6.0 版本中添加 类型: ContextT | None 默认值: None
stream_mode	图运行的流模式。 类型: StreamMode 默认值: 'values'
print_mode	接受与 stream_mode 相同的值，但仅将输出打印到控制台，用于调试目的。不以任何方式影响图的输出。 类型: StreamMode | Sequence[StreamMode] 默认值: ()
output_keys	从图运行中检索的输出键。 类型: str | Sequence[str] | None 默认值: None
interrupt_before	在这些节点之前中断图运行。 类型: All | Sequence[str] | None 默认值: None
interrupt_after	在这些节点之后中断图运行。 类型: All | Sequence[str] | None 默认值: None
durability	图执行的持久性模式，默认为 "async"。选项有 "sync": 更改在下一步开始前同步持久化。 "async": 更改在下一步执行时异步持久化。 "exit": 更改仅在图退出时持久化。 类型: Durability | None 默认值: None
**kwargs	传递给图运行的其他关键字参数。 类型： Any 默认值： {}

返回	描述
dict[str, Any] | Any	图运行的输出。如果 stream_mode 是 "values"，则返回最新的输出。
dict[str, Any] | Any	如果 stream_mode 不是 "values"，则返回一个输出块列表。

ainvoke async

ainvoke(
    input: InputT | Command | None,
    config: RunnableConfig | None = None,
    *,
    context: ContextT | None = None,
    stream_mode: StreamMode = "values",
    print_mode: StreamMode | Sequence[StreamMode] = (),
    output_keys: str | Sequence[str] | None = None,
    interrupt_before: All | Sequence[str] | None = None,
    interrupt_after: All | Sequence[str] | None = None,
    durability: Durability | None = None,
    **kwargs: Any,
) -> dict[str, Any] | Any

在单个输入上异步调用图。

参数	描述
输入	计算的输入数据。它可以是字典或任何其他类型。 类型: InputT | Command | None
配置	计算的配置。 类型： RunnableConfig | None 默认值： None
context	用于运行的静态上下文。 在 0.6.0 版本中添加 类型: ContextT | None 默认值: None
stream_mode	计算的流模式。 类型: StreamMode 默认值: 'values'
print_mode	接受与 stream_mode 相同的值，但仅将输出打印到控制台，用于调试目的。不以任何方式影响图的输出。 类型: StreamMode | Sequence[StreamMode] 默认值: ()
output_keys	结果中要包含的输出键。 类型: str | Sequence[str] | None 默认值: None
interrupt_before	在这些节点之前中断。 类型: All | Sequence[str] | None 默认值: None
interrupt_after	在这些节点之后中断。 类型: All | Sequence[str] | None 默认值: None
durability	图执行的持久性模式，默认为 "async"。选项有 "sync": 更改在下一步开始前同步持久化。 "async": 更改在下一步执行时异步持久化。 "exit": 更改仅在图退出时持久化。 类型: Durability | None 默认值: None
**kwargs	附加的关键字参数。 类型： Any 默认值： {}

返回	描述
dict[str, Any] | Any	计算的结果。如果 stream_mode 是 "values"，则返回最新的值。
dict[str, Any] | Any	如果 stream_mode 是 "chunks"，则返回一个块列表。

get_state

get_state(config: RunnableConfig, *, subgraphs: bool = False) -> StateSnapshot

获取图的当前状态。

aget_state async

aget_state(config: RunnableConfig, *, subgraphs: bool = False) -> StateSnapshot

获取图的当前状态。

get_state_history

get_state_history(
    config: RunnableConfig,
    *,
    filter: dict[str, Any] | None = None,
    before: RunnableConfig | None = None,
    limit: int | None = None,
) -> Iterator[StateSnapshot]

获取图的状态历史记录。

aget_state_history async

aget_state_history(
    config: RunnableConfig,
    *,
    filter: dict[str, Any] | None = None,
    before: RunnableConfig | None = None,
    limit: int | None = None,
) -> AsyncIterator[StateSnapshot]

异步获取图的状态历史记录。

update_state

update_state(
    config: RunnableConfig,
    values: dict[str, Any] | Any | None,
    as_node: str | None = None,
    task_id: str | None = None,
) -> RunnableConfig

使用给定值更新图的状态，就像它们来自节点 as_node 一样。如果未提供 as_node，它将被设置为最后一个更新状态的节点（如果不明确）。

aupdate_state async

aupdate_state(
    config: RunnableConfig,
    values: dict[str, Any] | Any,
    as_node: str | None = None,
    task_id: str | None = None,
) -> RunnableConfig

异步使用给定值更新图的状态，就像它们来自节点 as_node 一样。如果未提供 as_node，它将被设置为最后一个更新状态的节点（如果不明确）。

bulk_update_state

bulk_update_state(
    config: RunnableConfig, supersteps: Sequence[Sequence[StateUpdate]]
) -> RunnableConfig

批量应用对图状态的更新。需要设置检查点器。

参数	描述
配置	要应用更新的配置。 类型: RunnableConfig
supersteps	一个超步列表，每个超步包含一个要顺序应用于图状态的更新列表。每个更新都是 (values, as_node, task_id) 形式的元组，其中 task_id 是可选的。 类型: Sequence[Sequence[StateUpdate]]

引发	描述
ValueError	如果未设置检查点器或未提供更新。
InvalidUpdateError	如果提供了无效的更新。

返回	描述
RunnableConfig	更新后的配置。 类型: RunnableConfig

abulk_update_state async

abulk_update_state(
    config: RunnableConfig, supersteps: Sequence[Sequence[StateUpdate]]
) -> RunnableConfig

异步批量应用对图状态的更新。需要设置检查点器。

参数	描述
配置	要应用更新的配置。 类型: RunnableConfig
supersteps	一个超步列表，每个超步包含一个要顺序应用于图状态的更新列表。每个更新都是 (values, as_node, task_id) 形式的元组，其中 task_id 是可选的。 类型: Sequence[Sequence[StateUpdate]]

引发	描述
ValueError	如果未设置检查点器或未提供更新。
InvalidUpdateError	如果提供了无效的更新。

返回	描述
RunnableConfig	更新后的配置。 类型: RunnableConfig

get_graph

get_graph(config: RunnableConfig | None = None, *, xray: int | bool = False) -> Graph

返回计算图的可绘制表示。

aget_graph async

aget_graph(config: RunnableConfig | None = None, *, xray: int | bool = False) -> Graph

返回计算图的可绘制表示。

get_subgraphs

get_subgraphs(
    *, namespace: str | None = None, recurse: bool = False
) -> Iterator[tuple[str, PregelProtocol]]

获取图的子图。

参数	描述
namespace	用于过滤子图的命名空间。 类型: str | None 默认值: None
recurse	是否递归进入子图。如果为 False，则只返回直接子图。 类型： bool 默认值： False

返回	描述
Iterator[tuple[str, PregelProtocol]]	一个 (namespace, subgraph) 对的迭代器。

aget_subgraphs async

aget_subgraphs(
    *, namespace: str | None = None, recurse: bool = False
) -> AsyncIterator[tuple[str, PregelProtocol]]

获取图的子图。

参数	描述
namespace	用于过滤子图的命名空间。 类型: str | None 默认值: None
recurse	是否递归进入子图。如果为 False，则只返回直接子图。 类型： bool 默认值： False

返回	描述
AsyncIterator[tuple[str, PregelProtocol]]	一个 (namespace, subgraph) 对的迭代器。

with_config

with_config(config: RunnableConfig | None = None, **kwargs: Any) -> Self

创建具有更新配置的 Pregel 对象的副本。

langgraph.graph.message

函数	描述
add_messages	通过 ID 合并两个消息列表，更新现有消息。

add_messages

add_messages(
    left: Messages,
    right: Messages,
    *,
    format: Literal["langchain-openai"] | None = None,
) -> Messages

通过 ID 合并两个消息列表，更新现有消息。

默认情况下，这确保状态是“仅追加”的，除非新消息与现有消息具有相同的 ID。

参数	描述
left	Messages 的基础列表。 类型: Messages
right	要合并到基础列表中的 Messages 列表（或单个 Message）。 类型: Messages
format	返回消息的格式。如果为 None，则 Messages 将按原样返回。如果为 langchain-openai，则 Messages 将作为 BaseMessage 对象返回，其内容格式化以匹配 OpenAI 消息格式，这意味着内容可以是字符串、'text' 块或 'image_url' 块，工具响应将作为其自己的 ToolMessage 对象返回。 要求 必须安装 langchain-core>=0.3.11 才能使用此功能。 类型: Literal['langchain-openai'] | None 默认值: None

返回	描述
消息	一个新的消息列表，其中来自 right 的消息已合并到 left 中。
消息	如果 right 中的消息与 left 中的消息具有相同的 ID，则来自 right 的消息将替换来自 left 的消息。

示例

基本用法

from langchain_core.messages import AIMessage, HumanMessage

msgs1 = [HumanMessage(content="Hello", id="1")]
msgs2 = [AIMessage(content="Hi there!", id="2")]
add_messages(msgs1, msgs2)
# [HumanMessage(content='Hello', id='1'), AIMessage(content='Hi there!', id='2')]

覆盖现有消息

msgs1 = [HumanMessage(content="Hello", id="1")]
msgs2 = [HumanMessage(content="Hello again", id="1")]
add_messages(msgs1, msgs2)
# [HumanMessage(content='Hello again', id='1')]

在 StateGraph 中使用

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph


class State(TypedDict):
    messages: Annotated[list, add_messages]


builder = StateGraph(State)
builder.add_node("chatbot", lambda state: {"messages": [("assistant", "Hello")]})
builder.set_entry_point("chatbot")
builder.set_finish_point("chatbot")
graph = builder.compile()
graph.invoke({})
# {'messages': [AIMessage(content='Hello', id=...)]}

使用 OpenAI 消息格式

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, add_messages


class State(TypedDict):
    messages: Annotated[list, add_messages(format="langchain-openai")]


def chatbot_node(state: State) -> list:
    return {
        "messages": [
            {
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": "Here's an image:",
                        "cache_control": {"type": "ephemeral"},
                    },
                    {
                        "type": "image",
                        "source": {
                            "type": "base64",
                            "media_type": "image/jpeg",
                            "data": "1234",
                        },
                    },
                ],
            },
        ]
    }


builder = StateGraph(State)
builder.add_node("chatbot", chatbot_node)
builder.set_entry_point("chatbot")
builder.set_finish_point("chatbot")
graph = builder.compile()
graph.invoke({"messages": []})
# {
#     'messages': [
#         HumanMessage(
#             content=[
#                 {"type": "text", "text": "Here's an image:"},
#                 {
#                     "type": "image_url",
#                     "image_url": {"url": "data:image/jpeg;base64,1234"},
#                 },
#             ],
#         ),
#     ]
# }