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_count、critique 这些内部噪音也吐出去,调用方还得自己挑。这就是"内部实现细节泄漏给了调用方"。💡 本质:给图定义清晰的"公共接口",把实现藏起来就像一个函数:调用方只关心参数和返回值,不关心函数体里的局部变量。
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)
图注: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。图注:过滤只发生在 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 一次(性能与安全的权衡)。