LangGraph 03：消息与多 Schema

发表于 2026-08-18 更新于 2026-08-20 分类于 ai

讲解 add_messages、Overwrite、input_schema / output_schema 与私有状态，并用 MessagesState 对接 Coding Plan。

前言

《LangGraph 02：StateGraph 基础》里，operator.add 适合普通列表追加。
聊天 Agent 的 messages 还要按 消息 id 更新/替换，不能简单拼接。
本文覆盖 add_messages、用 Overwrite 强行覆盖带 reducer 的通道、图的输入输出 Schema 与节点间私有状态，最后用 MessagesState 接一次 Coding Plan。
模型示例继续用 Coding Plan，与 LangChain 系列共用 .env。

依赖

在已有工程上确保下列依赖可用（与《LangGraph 01：生态与环境》一致）。

1	uv add langgraph langchain-openai langchain-core python-dotenv rich

项目根目录 .env：

1 2	OPENAI_API_KEY=你的火山方舟 API Key OPENAI_BASE_URL=https://ark.cn-beijing.volces.com/api/coding/v3

实现

add_messages

add_messages 是消息专用 reducer：默认把右侧新消息追加到左侧；若新消息带有与已有消息相同的 id，则替换旧消息而不是再插一条。
下面用纯函数调用演示合并规则（无需构图）。

from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
from langgraph.graph.message import add_messages
from rich import print as rprint

left = [
    SystemMessage(content="你是助手", id="1"),
    HumanMessage(content="你好", id="2"),
    AIMessage(content="你好，有什么可以帮你？", id="3"),
]

right = [
    HumanMessage(content="我叫小明", id="2"),  # 同 id → 替换
    AIMessage(content="好的，我记住了", id="3"),  # 同 id → 替换
    HumanMessage(content="我是谁？", id="4"),
]

merged = add_messages(left, right)
rprint(merged)

观察结果：id="2" / "3" 被右侧内容替换，id="4" 为新增。
在 State 里应写成 messages: Annotated[list, add_messages]，而不是 Annotated[list, operator.add]。

Overwrite

带 reducer 的通道默认走合并；若某节点要 整表重置，可用 Overwrite(新值) 绕过 reducer。
下面 node_2 清空并重写 logs，后续节点再追加。

from operator import add
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import Overwrite
from rich import print as rprint


class OverAllState(TypedDict):
    logs: Annotated[list[str], add]
    cur_id: str


def node_1(state: OverAllState) -> dict:
    return {"logs": ["node_1 执行完成"]}


def node_2(state: OverAllState) -> dict:
    return {"logs": Overwrite(["node_2 重置后的唯一日志"])}


def node_3(state: OverAllState) -> dict:
    return {"logs": ["node_3 执行完成"]}


builder = StateGraph(OverAllState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_2", "node_3")
builder.add_edge("node_3", END)

graph = builder.compile()
rprint(graph.invoke({"logs": ["start"], "cur_id": "s"}))

最终 logs 大致为 ['node_2 重置后的唯一日志', 'node_3 执行完成']，初始与 node_1 的条目在重置后消失。
同一超步内多个并行节点对同一 key 使用 Overwrite 会冲突，需避免。

多 Schema

默认情况下，图的输入、内部状态、输出共用同一套 state_schema。
可以拆成：

input_schema：调用方 invoke 必须提供的字段。
state_schema（全局）：节点之间流转的完整内部状态。
output_schema：对外返回时过滤后的字段。
私有状态：仅部分节点读写、不必出现在最终输出里的通道（通过节点注解的独立 TypedDict 表达）。

下面示例把用户名加工成问候语，最终只输出 graph_output。

from typing import TypedDict

from langgraph.graph import END, START, StateGraph
from rich import print as rprint


class InputState(TypedDict):
    username: str


class OutputState(TypedDict):
    graph_output: str


class OverAllState(TypedDict):
    username: str
    graph_output: str
    nickname: str


class PrivateState(TypedDict):
    greeting: str


def node_1(state: InputState) -> dict:
    return {"nickname": "Dear " + state["username"]}


def node_2(state: OverAllState) -> PrivateState:
    return {"greeting": "Hello, " + state["nickname"]}


def node_3(state: PrivateState) -> OutputState:
    return {"graph_output": state["greeting"] + " 很高兴认识你！"}


builder = StateGraph(
    state_schema=OverAllState,
    input_schema=InputState,
    output_schema=OutputState,
)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_2", "node_3")
builder.add_edge("node_3", END)

graph = builder.compile()
rprint(graph.invoke({"username": "Ada"}))

invoke 只需 username；返回值通常只有 graph_output，中间 nickname / greeting 被输出 Schema 过滤。
各 Schema 之间应通过共享键名衔接；私有通道用于节点间临时传递，不必写进对外输出。

MessagesState 接模型

MessagesState 预置了 messages: Annotated[..., add_messages]。
业务字段可在子类中追加；节点往 messages 里追加 HumanMessage / 模型回复即可自动合并。

下面用 Coding Plan（OpenAI 兼容）跑一个最小对话节点。
配置方式与本站 LangChain 系列一致：ChatOpenAI + .env 中的 OPENAI_API_KEY / OPENAI_BASE_URL。

import os

from dotenv import load_dotenv
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, MessagesState, StateGraph
from rich import print as rprint

load_dotenv()

model = ChatOpenAI(
    model="ark-code-latest",
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
    temperature=0.2,
)


class OverAllState(MessagesState):
    username: str
    output: str


def prepare(state: OverAllState) -> dict:
    return {
        "messages": [HumanMessage(content=f"你好，我是{state['username']}，用一句话介绍你自己。")],
    }


def call_llm(state: OverAllState) -> dict:
    res = model.invoke(state["messages"])
    return {
        "messages": [res],
        "output": res.content,
    }


builder = StateGraph(OverAllState)
builder.add_node("prepare", prepare)
builder.add_node("call_llm", call_llm)
builder.add_edge(START, "prepare")
builder.add_edge("prepare", "call_llm")
builder.add_edge("call_llm", END)

graph = builder.compile()
result = graph.invoke({"username": "老王", "messages": []})
rprint(result["output"])
rprint(result["messages"])

也可用 init_chat_model("openai:ark-code-latest", api_key=..., base_url=...)，效果等价。
messages 中应看到用户消息与 AI 回复两条（或更多，取决于模型是否返回额外结构）。

验证

先跑 add_messages 与 Overwrite 脚本，确认 id 替换与列表重置。
跑多 Schema 示例，确认输出只有 graph_output。
配置好 .env 后执行 MessagesState 脚本：uv run python messages_llm.py，终端应打印模型回复与完整 messages。

总结

对话列表用 add_messages / MessagesState，不要用裸 operator.add。
需要无视 reducer 时用 Overwrite；并行同 key 慎用。
input_schema / output_schema 控制对外契约，私有 TypedDict 传递中间量。
下一篇讲边与路由：《LangGraph 04：边与条件路由》。