Multiple Schema Design
Private state · Input filtering · Output filtering

1Why multiple schemas are needed? ——Limitations of single Schema

In the previous lessons, we learned the basics of LangGraph:All nodes share the same State (OverallState), the input and output of the graph are also the same State. This works well in simple scenarios, but as the complexity of the graph increases, several obvious problems are exposed.

Problem 1: Intermediate calculation data pollutes the public interface

Imagine a question and answer system: the user only needs to provide a "question" and get an "answer". However, during the calculation process inside the graph, a lot of temporary data will be generated - search keywords, drafts of intermediate reasoning, scoring values, etc. If using a single Schema, all of this data is exposed to the caller:

The caller only wantsanswer, but got all the internal fields, the interface is bloated and easy to misuse.

Problem 2: "Temporary transfer" data between nodes should not be exposed to the outside world

Sometimes,node_1 needs to pass an intermediate value to node_2, but this intermediate value has nothing to do with the final input/output of the graph. If you put it into OverallState, you have to expose this implementation detail in the "public interface" of the entire graph.

Global perspective of multiple schemas

Note: Private State is a concept from another dimension. It exists on the "channel" between nodes and does not even appear in OverallState. We will explain them one by one next.

2Private State - "internal notes" between nodes

core concepts

Private State refers to:Some data only circulates between specific nodes and is neither the input nor the output of the graph, nor does it belong to the global OverallState. It is a note for "private chat" between nodes and cannot be seen by other nodes or external callers.

Illustration: Data flow of private state

Comparison of private state vs global state

3Complete code analysis of private state

Below is the complete example code for private state in the notebook. Let’s break it down piece by piece:

Step 1: Define two TypedDict Schemas

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

Step 2: Define node functions (key: type annotation)

# 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}

Step Three: Build and Run the Graph

# 构建图——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 从未出现

Complete data flow tracing

4Single Schema input and output issues

Before introducing Input/Output Schema, we first use the example in the notebook to intuitively feel the limitations of the single Schema approach.

Scenario: A simple question and answer graph

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()

Problem exposed: invoke() returns all fields

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

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

These two pain points may seem insignificant in a simple example, but in a real production scenario - where the graph has dozens of fields and external systems rely on its output - they become serious design issues.Input/Output Schema is the solution to these two pain points.

5Input / Output Schema - precise interface filtering

core mechanism

StateGraphThree construction parameters are supported to define multiple Schemas:

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

The functions of these two parameters:

• input_schema: Perform on the dictionary passed in by invoke()Validation and filtering. Only fields declared in InputState are accepted, extra fields are rejected.
• output_schema: Perform on the OverallState after the graph is executed.filter, only the fields declared in OutputState are returned to the caller.

Hierarchical relationship between three Schemas

Type annotations for node functions (optional but recommended)

When the graph uses multiple Schemas, the type annotation of the node function can clearly declare which Schema it uses, increasing code readability:

# 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"}

6Input / Output Schema complete code analysis

The following is the complete implementation code of Input/Output Schema in the notebook. Let’s analyze it line by line:

Step 1: Define the three-layer 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 中

Step 2: Define node functions

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"}

Step 3: Build the graph and pass in the input/output Schema parameters

# 关键！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()

Step 4: Run to verify the effect

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

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

Complete data flow tracking

7Comparison and comprehensive application of two mechanisms

Private state vs Input/Output Schema: core differences

Both mechanisms can be used simultaneously

In a complex graph, you can use private state and Input/Output Schema at the same time:

# 同时使用两种机制的设计示例
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 ...

8Real-life scenarios: When to use multiple schemas?

Scenarios suitable for using private state

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

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

Scenarios suitable for using Input/Output Schema

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

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

Decision suggestions

• 1

  Start with a single Schema

  In the prototype stage, an OverallState is used to cover all fields and quickly verify the logic. Don't optimize prematurely.

• 2

  When the interface is exposed to the outside world, Input/Output Schema is introduced

  The graph needs to be called by external systems (API, front-end, other microservices), or you find that the return value field of invoke() has too many fields, affecting usability.

• 3

  When there is temporary intermediate data between nodes, Private State is introduced

  You find that some fields in OverallState are "only used by two adjacent nodes, and are never read or written elsewhere." This is a candidate for private state.

• 4

  Keep OverallState as the "source of truth"

  No matter how many Schemas are used, OverallState is always the complete data structure for communication between nodes within the graph. Make sure it contains all the fields that the node actually needs to read and write.

Knowledge map of this lesson