Day 11 / 共 60 天 · 阶段 2 状态与数据流

输入/输出 schema:让图对外和对内长得不一样

到目前为止,图的输入、内部状态、输出用的是同一个 State。但真实项目里你常常希望:调用方只需要传两三个字段(干净的输入契约)、内部却有一堆中间字段(草稿、计数器、检索结果)、最后只吐一个结果字段。今天就看 input_schema / output_schema 怎么实现这种"对外精简、对内丰富",以及那些既不在输入也不在输出的字段——私有中间字段——是怎么自然产生的。

📍 阶段 2「状态与数据流」(D07-12) · 你在第 11 天
D07 reducers D08 add_messages D09 累加通道 D10 并行写 D11 输入/输出 D12 Pydantic
💡 一句话锚点 StateGraph 有三个 schema:input_schema(图接受什么输入)、state_schema(内部完整状态)、output_schema(图返回什么)。不指定时三者相等(前十天一直如此)。一旦分开指定,图就有了"过滤器":进来时只保留 input 声明的键,出去时只保留 output 声明的键。凡是只在 state 里、不在 input/output 里的字段,就是外界看不见、只在内部流转的私有字段。
L01

痛点:为什么要区分对外和对内的状态?

🤔 痛点假设你的图内部有 8 个字段:question(输入)、docs(检索到的文档)、draft(草稿)、critique(自我批评)、retry_count(重试计数)、answer(最终答案)……如果输入输出都用这个完整 State:① 调用方得知道所有 8 个字段、还可能误传本该内部计算的 docs;② 返回时把 retry_countcritique 这些内部噪音也吐出去,调用方还得自己挑。这就是"内部实现细节泄漏给了调用方"。
💡 本质:给图定义清晰的"公共接口",把实现藏起来就像一个函数:调用方只关心参数和返回值,不关心函数体里的局部变量。input_schema 是"参数表",output_schema 是"返回类型",state_schema 里多出来的字段就是"局部变量"。LangGraph 在入口处过滤输入、出口处过滤输出,从机制上保证内部字段不外泄、也不被外部乱塞。
大白话输入 schema = 前台只收这几样东西;输出 schema = 出口只发这几样东西;中间那一大堆,是后厨自己用的,客人看不到也碰不到。
L02

三个 schema:构造时就分别登记

回顾 D03 的 __init__——三个 schema 在这里定下并各自解析:

libs/langgraph/langgraph/graph/state.py:260-269
        self.state_schema = state_schema
        self.input_schema = cast(type[InputT], input_schema or state_schema)   # 没给→=state
        self.output_schema = cast(type[OutputT], output_schema or state_schema)
        self.context_schema = context_schema
        self._node_defaults = _NodeDefaults()

        self._add_schema(self.state_schema)                       # 完整状态
        self._add_schema(self.input_schema, allow_managed=False)  # 输入(禁托管字段)
        self._add_schema(self.output_schema, allow_managed=False) # 输出(禁托管字段)
input_schema or state_schema核心一行:不单独传 input_schema,它就默认等于 state_schema。这解释了前十天"输入=内部=输出"的现象。
三次 _add_schema三个 schema 都走 D05 学过的解析。它们的字段通道汇总进同一个 self.channels——也就是说图的通道总集 = 三者字段的并集。
allow_managed=False★输入/输出 schema 里不允许出现"引擎托管字段"(如 RemainingSteps)。因为托管值是引擎内部算的,让调用方输入或接收它没有意义。

这个 allow_managed=False_add_schema 里真会拦人:

libs/langgraph/langgraph/graph/state.py:346-352
            if managed and not allow_managed:
                names = ", ".join(managed)
                raise ValueError(
                    f"Invalid managed channels detected in {schema_name}: {names}."
                    " Managed channels are not permitted in Input/Output schema.")
为什么通道要取并集? 因为输入字段进来后要能被内部节点读、内部字段又要能流到输出。三者共享同一套通道,数据才能贯通。schema 只是"每个阶段允许看见哪些通道"的视图。
L03

私有字段:不在 input/output 里就自动私有

LangGraph 没有 private 关键字。私有字段是差集算出来的——在 state 里、但不在 input 也不在 output 里的字段,天然只在内部流转。看个例子:

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

class InputS(TypedDict):
    question: str                 # 调用方只需给这个

class OutputS(TypedDict):
    answer: str                   # 调用方只拿到这个

class OverallS(TypedDict):
    question: str
    docs: list                    # ← 私有:检索中间结果
    draft: str                    # ← 私有:草稿
    answer: str

g = StateGraph(OverallS, input_schema=InputS, output_schema=OutputS)
数据结构:三个 schema 的字段范围(同心圈) state_schema (OverallS):question / docs / draft / answer input_schema question 调用方能传的键 output_schema answer 调用方能拿的键 docs / draft 私有字段 私有字段 = state − input − output(既不能被传入,也不会被返回)
图注:docs / draft 只在中间节点之间读写,从图外完全不可见。
💡 私有字段是"约束"不是"新类型"它和普通字段在 self.channels 里没区别(都是通道)。"私有"只体现在两处过滤:入口不接收它、出口不返回它。理解这点,下面 L04/L05 看过滤逻辑就通了。
L04

输入过滤:START 节点只写 input_schema 的键

图的入口是一个特殊的 START 节点。编译时它的"能写哪些通道"只取 input_schema 的字段

libs/langgraph/langgraph/graph/state.py:1431-1441
    def attach_node(self, key, node):
        if key == START:
            output_keys = [
                k
                for k, v in self.builder.schemas[self.builder.input_schema].items()
                if not is_managed_value(v)                # ★只要 input_schema 的字段
            ]
        else:
            output_keys = list(self.builder.channels) + [...]   # 普通节点能写全部字段

而这个 output_keys 会被用来过滤写入——多余的键直接丢弃:

libs/langgraph/langgraph/graph/state.py:1443-1449
        def _get_updates(input):
            if input is None: return None
            elif isinstance(input, dict):
                return [(k, v) for k, v in input.items() if k in output_keys]   # ★过滤
key == STARTSTART 是"把用户输入灌进状态"的虚拟节点。它的 output_keys 限定为 input_schema 的字段。
k in output_keys★关键过滤:用户输入的 dict 里,只有 key 属于 input_schema 的才被写入通道。你多传了 docs?START 直接把它丢掉——私有字段无法被外部注入。
普通节点 output_keys=全部通道对比:内部节点能写所有字段(包括私有的 docs/draft)。可见"私有"只卡在 START 这一关。
📝 真实值走一遍 你调 app.invoke({"question": "...", "docs": ["偷偷塞的"]})。START 的 output_keys=["question"],过滤后只有 question 进状态,docs静默丢弃(不报错)。想初始化 docs?只能让内部节点去写。
L05

输出过滤:结束时只吐 output_schema 的键

出口对称。编译时算出 output_channels——只取 output_schema 的非托管字段:

libs/langgraph/langgraph/graph/state.py:1256-1266
        output_channels = (
            "__root__"
            if len(self.schemas[self.output_schema]) == 1
            and "__root__" in self.schemas[self.output_schema]
            else [
                key
                for key, val in self.schemas[self.output_schema].items()
                if not is_managed_value(val)              # ★只要 output_schema 的字段
            ]
        )

对比同处的 stream_channels——流式默认吐的是全部通道:

libs/langgraph/langgraph/graph/state.py:1267-1273
        stream_channels = (
            "__root__"
            if len(self.channels) == 1 and "__root__" in self.channels
            else [key for key, val in self.channels.items() if not is_managed_value(val)]
        )
output_channels 来自 output_schemainvoke 最终返回的 dict,只包含这些键。私有字段 docs/draft 即使状态里有值,也不会出现在返回里。
stream_channels 来自全部 channels★微妙差异:stream() 默认能看到所有字段(含私有)。因为流式是"调试/观察"用途,看全一点合理。要严格隐藏私有字段,得留意流式模式。
__root__ 特判裸类型状态(D05)时输出就是那个单值本身,不包成 dict。
控制流:入口过滤 → 内部全见 → 出口过滤 用户输入question,docs(偷塞) START滤:仅input键 内部节点可读写全部字段含 docs/draft END滤:仅output键 answer docs 被丢弃 docs/draft 不返回 私有字段只在中间那段"全见区"存在,两头都被过滤门挡住
图注:过滤只发生在 START(入口)和 END(出口)两处,中间节点看得到全部字段。
⚠️ 边界:invoke 藏得住私有字段,stream 不一定正因为 output_channels(invoke 用)取自 output_schema、而 stream_channels(stream 用)取自全部通道——同一个图,invoke 的返回不含私有字段,但 stream() 默认会流出私有字段。如果你靠 output_schema 隐藏敏感中间数据,别忘了流式路径需要额外用 stream_mode / output_keys 控制,否则会从流里漏出去。
L06

两处设计取舍 + 一处边界

🎨 设计取舍①:多传的输入键为什么"静默丢弃"而不是报错? _get_updates 对不在 input_schema 的键直接过滤掉,不抛异常。好处:调用方可以复用一个更大的字典喂给多个图,各图只取自己认识的键,互不干扰;也方便"多传几个键做实验"不至于崩。代价:拼错字段名(quesion 少个 t)不会报错,你会得到一个"输入没生效"的隐形 bug。缓解:图跑出来结果不对时,先核对输入键名是否精确匹配 input_schema。这是"宽容输入"换来的调试成本。
🎨 设计取舍②:为什么私有字段用"差集"实现,而不加显式标记? LangGraph 没有 Private[...] 标记,私有 = state 有、input/output 没有。好处:零新语法——你只要把不想暴露的字段"从 input/output schema 里省掉"即可,规则简单到可以口算。代价:私有性散落在三个类的定义里,看单个类看不出哪些字段私有,得三个对照着看。对于字段很多的大状态,建议用注释标一下"这些是私有中间字段",弥补语言层没有标记的缺失。
⚠️ 边界:input_schema 里放了 state 没有的字段?会进 channels 但没人读因为三个 schema 的通道取并集(L02),你在 input_schema 声明一个 state_schema 里没有的字段,它会被加进 self.channels——但没有任何内部节点的 schema 包含它,于是它进得来、却没节点读得到,成了"死字段"。正确做法:input_schema 的字段应是 state_schema 字段的子集(output 同理)。它们是 state 的"视图",不该引入 state 没有的新字段。
L07

今日小结 + 动手 + 明日预告

🧠 今天你应该能回答

  • StateGraph 的三个 schema 分别管什么?(input=接受的输入 / state=完整内部 / output=返回值)
  • 不单独指定时三者什么关系?(input/output 默认等于 state)
  • 私有字段怎么定义出来的?(在 state 里、但不在 input 也不在 output 里)
  • 输入过滤发生在哪?(START 节点,output_keys 只取 input_schema 字段,多传的键丢弃)
  • 输出过滤发生在哪?(output_channels 只取 output_schema 字段)
  • 为什么 input/output schema 不许有托管字段?(allow_managed=False,托管值是引擎内部算的)
  • invoke 和 stream 对私有字段的可见性有何不同?(invoke 用 output_channels 会隐藏;stream 默认用全部通道会漏出)

✋ 10 分钟动手

# 1. 看三个 schema 的初始化
sed -n '260,269p' libs/langgraph/langgraph/graph/state.py

# 2. 看输入过滤(START 只写 input_schema 的键)
sed -n '1431,1449p' libs/langgraph/langgraph/graph/state.py

# 3. 看输出/流式通道的差异
sed -n '1256,1273p' libs/langgraph/langgraph/graph/state.py

# 4. 亲手验证"多传的键被丢弃 + 私有字段不返回"
python3 -c "
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
class I(TypedDict): question:str
class O(TypedDict): answer:str
class S(TypedDict):
    question:str; docs:list; answer:str
def work(s): return {'docs':['d1'], 'answer':'42'}
g=StateGraph(S, input_schema=I, output_schema=O)
g.add_node('w', work); g.add_edge(START,'w'); g.add_edge('w',END)
app=g.compile()
print(app.invoke({'question':'q','docs':['偷塞']}))  # docs 被丢弃, 返回只含 answer
"
💡 明日预告 · Day 12今天的 schema 都是 TypedDict(无校验)。明天 D12 换成 Pydantic BaseModel 做状态:看 _internal/_pydantic.py 怎么处理"字段名和 pydantic 内部保留名冲突"、bench/pydantic_state.py 这个真实压测状态长啥样,以及为什么用 Pydantic 状态每读一次状态就 validate 一次(性能与安全的权衡)。
← Day 10 · 并行写 Day 12 · Pydantic State →