learn-langgraph · 02
State、Node、Graph 三件套
LangGraph 应用由三个东西构成：状态（State）、节点（Node）、图（Graph）。这一篇把每个细节讲清楚，后续所有示例都建在这个基础上。
安装
pip install langgraph langchain-core
如果要用 OpenAI：
pip install langchain-openai
State：图的内存
State 是整张图的共享数据容器。所有节点都从这里读数据，修改后写回来。
用 TypedDict 定义
from typing import TypedDict, List, Optional
class MyState(TypedDict):
# 用户输入
user_input: str
# 中间结果
analysis: str
# 最终输出
result: str
# 可选字段
error: Optional[str]
TypedDict 是 Python 标准库里的类型，让字典有了类型检查。LangGraph 用它来追踪状态的结构。
节点返回 dict，不是完整 State
节点只需要返回要修改的字段：
def my_node(state: MyState) -> dict:
# 只返回要更新的字段
return {"analysis": "positive"}
# 不需要返回完整的 state
LangGraph 会把返回的 dict 合并进当前 state，未修改的字段保持不变。
用 Annotated + operator.add 追加列表
如果某个字段是列表，想追加而不是覆盖：
from typing import Annotated
import operator
class ChatState(TypedDict):
messages: Annotated[List[str], operator.add]
summary: str
这样每个节点返回的 {"messages": ["新消息"]} 会被追加到列表里，而不是替换整个列表。
Node：图的操作单元
节点就是普通 Python 函数，签名固定：
def node_name(state: YourState) -> dict:
# 读取状态
value = state["some_field"]
# 做处理
result = process(value)
# 返回要更新的字段
return {"another_field": result}
节点的几个原则
原则一：节点只做一件事
每个节点聚焦一个职责。不要把分析、调用 API、格式化输出都塞进一个节点。
原则二：节点是纯的（尽量）
理想情况下，节点的输出只依赖 state 输入，没有隐藏的全局状态。这样更容易测试和调试。
原则三：返回 dict，不是 State 对象
# 正确
return {"field_a": value_a, "field_b": value_b}
# 错误——不要返回完整 state
return state
Graph：把节点连起来
from langgraph.graph import StateGraph, END, START
# 1. 创建图，指定 State 类型
graph = StateGraph(MyState)
# 2. 添加节点
graph.add_node("node_a", function_a)
graph.add_node("node_b", function_b)
graph.add_node("node_c", function_c)
# 3. 设置入口
graph.set_entry_point("node_a")
# 等价于：graph.add_edge(START, "node_a")
# 4. 添加边
graph.add_edge("node_a", "node_b")
graph.add_edge("node_b", "node_c")
graph.add_edge("node_c", END)
# 5. 编译
app = graph.compile()
运行图
# invoke：同步运行，返回最终 state
result = app.invoke({"user_input": "hello"})
print(result)
# stream：流式运行，每个节点完成后返回一次
for chunk in app.stream({"user_input": "hello"}):
print(chunk)
invoke 返回的是最终的完整 state 字典。stream 每次 yield 一个 {node_name: state_update} 的字典。
完整示例：文本处理管道
把这三件套组合起来，写一个简单的文本处理管道：
from typing import TypedDict
from langgraph.graph import StateGraph, END
# 1. 定义 State
class TextState(TypedDict):
raw_text: str
cleaned: str
word_count: int
summary: str
# 2. 定义节点
def clean_node(state: TextState) -> dict:
"""清理文本：去掉多余空格"""
text = state["raw_text"].strip()
return {"cleaned": text}
def count_node(state: TextState) -> dict:
"""统计词数"""
count = len(state["cleaned"].split())
return {"word_count": count}
def summary_node(state: TextState) -> dict:
"""生成摘要（这里简化为截断）"""
text = state["cleaned"]
summary = text[:50] + "..." if len(text) > 50 else text
return {"summary": summary}
# 3. 组装图
graph = StateGraph(TextState)
graph.add_node("clean", clean_node)
graph.add_node("count", count_node)
graph.add_node("summarize", summary_node)
graph.set_entry_point("clean")
graph.add_edge("clean", "count")
graph.add_edge("count", "summarize")
graph.add_edge("summarize", END)
app = graph.compile()
# 4. 运行
result = app.invoke({
"raw_text": " LangGraph 是一个用于构建有状态 Agent 应用的框架，基于图结构描述执行流。 ",
"cleaned": "",
"word_count": 0,
"summary": ""
})
print("清理后:", result["cleaned"])
print("词数:", result["word_count"])
print("摘要:", result["summary"])
输出：
清理后: LangGraph 是一个用于构建有状态 Agent 应用的框架，基于图结构描述执行流。
词数: 17
摘要: LangGraph 是一个用于构建有状态 Agent 应用的框架，基于图结构描述执行流。
可视化图结构
LangGraph 可以把图渲染成 ASCII 或图片：
# ASCII 可视化
print(app.get_graph().draw_ascii())
# 输出大致如下：
# +-----------+
# | __start__ |
# +-----------+
# |
# clean
# |
# count
# |
# summarize
# |
# +---------+
# | __end__ |
# +---------+
小结
三件套的关系：
- State 是数据，所有节点共享，TypedDict 定义结构
- Node 是操作，读 State 写 State，普通函数
- Graph 是流程，把节点用边连起来，
compile()后可运行
记住这三条规则：
- 节点只返回要修改的字段（dict），不返回完整 state
- 边决定执行顺序，
END是终止信号 invoke传入的是初始 state（dict），返回的是最终 state（dict）
下一篇我们写第一个真实可运行的工作流：顺序图：第一个可运行的 Workflow。