多重 Schema 设计
私有状态 · 输入过滤 · 输出过滤

1 为什么需要多重 Schema？——单 Schema 的局限

在前面的课程中，我们学到了 LangGraph 的基础：所有节点共享同一个 State（OverallState），图的输入和输出也都是这同一个 State。这在简单场景下工作良好，但随着图的复杂度增加，会暴露出几个明显的问题。

问题一：中间计算数据污染了公开接口

想象一个问答系统：用户只需要提供一个"问题"，得到一个"答案"。但图内部在计算过程中，会产生很多临时数据——搜索的关键词、中间推理的草稿、评分数值等。如果使用单 Schema，这些数据全部暴露给调用者：

调用者只想要 answer，却拿到了所有内部字段，接口臃肿、容易误用。

问题二：节点间的"临时传递"数据不应对外暴露

有时候，node_1 需要传递一个中间值给 node_2，但这个中间值与图的最终输入/输出毫无关系。如果你把它放进 OverallState，就不得不在整个图的"公共接口"里暴露这个实现细节。

多重 Schema 的全局视角

注意：私有状态（PrivateState）是另一个维度的概念，它存在于节点与节点之间的"通道"上，甚至不出现在 OverallState 里。我们接下来逐一讲解。

2 私有状态（Private State）——节点间的"内部便签"

核心概念

私有状态（Private State）是指：某些数据只在特定的节点之间流通，既不是图的输入，也不是图的输出，更不属于全局 OverallState。它是节点间"私聊"的便签条，其他节点和外部调用者都看不到。

图示：私有状态的数据流

私有状态 vs 全局状态的对比

3 私有状态完整代码解析

下面是 notebook 中私有状态的完整示例代码。我们逐段拆解分析：

第一步：定义两个 TypedDict Schema

from typing_extensions import TypedDict
from IPython.display import Image, display
from langgraph.graph import StateGraph, START, END

# ① 全局状态：图的"公开面板"，输入输出都通过它
class OverallState(TypedDict):
    foo: int

# ② 私有状态：只在 node_1 和 node_2 之间传递
class PrivateState(TypedDict):
    baz: int

第二步：定义节点函数（关键：类型注解）

# node_1：接受 OverallState，返回 PrivateState
def node_1(state: OverallState) -> PrivateState:
    print("---Node 1---")
    # 从 OverallState 读取 foo，加 1 后写入 PrivateState 的 baz
    return {"baz": state['foo'] + 1}

# node_2：接受 PrivateState，返回 OverallState
def node_2(state: PrivateState) -> OverallState:
    print("---Node 2---")
    # 从 PrivateState 读取 baz，加 1 后写入 OverallState 的 foo
    return {"foo": state['baz'] + 1}

第三步：构建并运行图

# 构建图——StateGraph 的主 Schema 是 OverallState
builder = StateGraph(OverallState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)

builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_2", END)

graph = builder.compile()

# 运行：传入 OverallState，foo=1
result = graph.invoke({"foo": 1})
# 输出：
# ---Node 1---
# ---Node 2---
# {'foo': 3}   ← 只有 foo，baz 从未出现

完整的数据流追踪

4 单 Schema 的输入输出问题

在引入 Input/Output Schema 之前，我们先用 notebook 里的例子，直观感受单 Schema 方式的局限。

场景：一个简单的问答图

class OverallState(TypedDict):
    question: str   # 用户的问题
    answer: str     # 最终答案
    notes: str      # 内部推理笔记（中间数据）

def thinking_node(state: OverallState):
    # 思考节点：产出中间笔记和初步答案
    return {"answer": "bye", "notes": "... his name is Lance"}

def answer_node(state: OverallState):
    # 答案节点：产出最终答案
    return {"answer": "bye Lance"}

graph = StateGraph(OverallState)
graph.add_node("answer_node", answer_node)
graph.add_node("thinking_node", thinking_node)
graph.add_edge(START, "thinking_node")
graph.add_edge("thinking_node", "answer_node")
graph.add_edge("answer_node", END)

graph = graph.compile()

问题暴露：invoke() 返回了所有字段

graph.invoke({"question": "hi"})

# 实际输出（包含用户不需要的字段）：
# {
#   'question': 'hi',        ← 用户自己知道，不需要再返回
#   'answer': 'bye Lance',   ← 用户真正需要的
#   'notes': '... his name is Lance'  ← 内部中间数据，泄露给外部！
# }

这两个痛点在简单示例里看似无关紧要，但在真实生产场景中——图有数十个字段、外部系统依赖其输出——就会成为严重的设计问题。Input/Output Schema 就是针对这两个痛点的解决方案。

5 Input / Output Schema——精确的接口过滤

核心机制

StateGraph 支持三个构造参数来定义多重 Schema：

graph = StateGraph(
    OverallState,            # 第一个参数：内部全局状态（节点间通信用）
    input_schema=InputState,  # 关键字参数：定义 invoke() 接受的输入结构
    output_schema=OutputState # 关键字参数：定义 invoke() 返回的输出结构
)

这两个参数的作用：

• input_schema：对 invoke() 传入的字典进行验证和过滤。只有 InputState 中声明的字段才被接受，多余的字段会被拒绝。
• output_schema：对图执行完毕后的 OverallState 进行过滤，只返回 OutputState 中声明的字段给调用者。

三个 Schema 的层次关系

节点函数的类型注解（可选但推荐）

当图使用多重 Schema 时，节点函数的类型注解可以明确声明它使用的是哪个 Schema，增加代码可读性：

# thinking_node：只需要读 question（InputState 的字段）
# 使用 InputState 类型注解，表明它只关心输入字段
def thinking_node(state: InputState):
    return {"answer": "bye", "notes": "... his name is Lance"}

# answer_node：需要读取所有字段，最终输出 OutputState
# -> OutputState 类型注解表明它的返回值会被过滤为 OutputState
def answer_node(state: OverallState) -> OutputState:
    return {"answer": "bye Lance"}

6 Input / Output Schema 完整代码解析

下面是 notebook 中 Input/Output Schema 的完整实现代码，我们逐行深入解析：

第一步：定义三层 Schema

# ① 输入 Schema：定义调用者可以/必须传入的字段
class InputState(TypedDict):
    question: str

# ② 输出 Schema：定义 invoke() 返回给调用者的字段
class OutputState(TypedDict):
    answer: str

# ③ 内部全局状态：图内部节点间通信的完整数据结构
class OverallState(TypedDict):
    question: str   # 来自 InputState
    answer: str     # 最终写入 OutputState
    notes: str      # 纯内部字段，不在 Input/Output Schema 中

第二步：定义节点函数

def thinking_node(state: InputState):
    # state 只包含 question，因为此节点声明了 InputState 类型
    # 但返回值会被合并到 OverallState 中
    return {"answer": "bye", "notes": "... his is name is Lance"}

def answer_node(state: OverallState) -> OutputState:
    # state 包含完整的 OverallState（question + answer + notes）
    # -> OutputState 声明返回值语义上对应 OutputState
    return {"answer": "bye Lance"}

第三步：构建图，传入 input/output Schema 参数

# 关键！StateGraph 第一个参数是内部 Schema
# input_schema 和 output_schema 是过滤器参数
graph = StateGraph(
    OverallState,
    input_schema=InputState,
    output_schema=OutputState
)

graph.add_node("answer_node", answer_node)
graph.add_node("thinking_node", thinking_node)
graph.add_edge(START, "thinking_node")
graph.add_edge("thinking_node", "answer_node")
graph.add_edge("answer_node", END)

graph = graph.compile()

第四步：运行验证效果

# 调用者只需传 InputState 声明的字段
result = graph.invoke({"question": "hi"})

# 实际输出（经过 OutputState 过滤）：
# {'answer': 'bye Lance'}
#
# 注意：question 和 notes 都不见了！
# 图内部确实有这两个字段，只是输出时被过滤掉了。

完整数据流追踪

7 两种机制对比与综合应用

私有状态 vs Input/Output Schema：核心区别

两种机制可以同时使用

在一个复杂的图中，你完全可以同时使用私有状态和 Input/Output Schema：

# 同时使用两种机制的设计示例
class InputState(TypedDict):
    user_query: str                   # 用户输入

class OutputState(TypedDict):
    final_answer: str                 # 最终答案

class OverallState(TypedDict):
    user_query: str                   # 来自 InputState
    search_results: list              # 搜索结果
    draft_answer: str                 # 草稿答案
    final_answer: str                 # 写入 OutputState

class PrivateState(TypedDict):
    relevance_score: float            # 私有：节点间传递评分
    refined_query: str               # 私有：精炼后的查询词

# 节点 A：接受 InputState，内部处理后传递 PrivateState
def query_refiner(state: InputState) -> PrivateState:
    refined = state["user_query"].strip().lower()
    return {"refined_query": refined, "relevance_score": 0.0}

# 节点 B：接受 PrivateState，写回 OverallState
def searcher(state: PrivateState) -> OverallState:
    results = do_search(state["refined_query"])
    return {"search_results": results, "draft_answer": ""}

# 节点 C：接受完整 OverallState，产出 final_answer
def synthesizer(state: OverallState) -> OutputState:
    answer = synthesize(state["search_results"], state["user_query"])
    return {"final_answer": answer}

# 构建图，同时配置 Input/Output Schema
graph = StateGraph(
    OverallState,
    input_schema=InputState,
    output_schema=OutputState
)
# ... add_node, add_edge, compile ...

8 真实场景：什么时候用多重 Schema？

适合使用私有状态的场景

class ScoredDocsState(TypedDict):
    # 私有：带评分的文档，只在检索和排序节点之间流通
    scored_docs: list[tuple[str, float]]

class ReasoningState(TypedDict):
    # 私有：详细推理过程，最终用户不需要看到
    chain_of_thought: str
    confidence: float

适合使用 Input/Output Schema 的场景

# 前端友好的接口
class ChatInput(TypedDict):
    message: str     # 用户发送的消息
    session_id: str  # 会话 ID

class ChatOutput(TypedDict):
    reply: str        # AI 回复
    tokens_used: int  # 消耗的 token 数（计费用）

决策建议

• 1

  先从单 Schema 开始

  原型阶段用一个 OverallState 涵盖所有字段，快速验证逻辑。不要过早优化。

• 2

  当接口对外暴露时，引入 Input/Output Schema

  图需要被外部系统调用（API、前端、其他微服务），或者你发现 invoke() 的返回值字段过多，影响可用性。

• 3

  当节点间有临时中间数据时，引入 Private State

  你发现 OverallState 里有些字段"只被相邻两个节点使用，其他地方从不读写"，这正是私有状态的候选。

• 4

  保持 OverallState 作为"事实来源"

  无论使用多少 Schema，OverallState 始终是图内部节点间通信的完整数据结构。确保它包含所有节点实际需要读写的字段。

本节课知识地图