LangGraph 能力详解 · 学习笔记

认识智能服务（Agent Server）#

先分清两类东西，我之前一直混着叫。

我见过的典型 Agent Server 形态：说一句「帮我规划周末旅行」，它自己往下走 —— 查天气 → 找景点 → 订酒店 → 规划路线 → 给打包清单，每一步都能看见中间结果。

一句话概括差异：它不只是回答问题，而是把一件事从头跟到尾。

构建 Agent Server 时遇到的四大难题#

思考一下：如果要做 7×24 小时的 AI 客服，我认为最头疼的是「无法干预」——出错时没有刹车，比出错本身更可怕。这四个难题正好是后面各章的主线。

解决方案：LangGraph —— Agent Server 的「操作系统」#

它是给 Agent Server 用的操作系统内核：不关心我用什么模型、提示词怎么写，专心解决状态管理、流程编排、持久化、人工监督这些底层工程问题。

把 Agent Server 比作一家公司：

• AI 模型 = 干活的人（员工）
• 工具 = 办公装备（电脑、电话）
• LangGraph = 既当项目经理，又当流程系统

LangGraph 生态系统#

• LangSmith：追踪、评估、监控；调试图的时候几乎离不开。
• LangGraph 部署平台：把 Agent Server 部署和扩起来。
• LangChain：提供大量集成与组件。

注意：三者拼起来是完整的「开发 → 调试 → 部署」链路，但并不是绑死的。LangGraph 可以只身使用。

术语速查（这份笔记反复出现的名字）#

常见概念区分#

工作流#

工作流的代码路径是提前定死的，按固定次序跑。做复杂度更高的 AI 应用时，它能给边界清晰的任务带来可预测与一致。

书里的例子是研发一款城市通勤用的智能电动自行车（导航、社交、防盗），动手前先做多维分析：

• 市场分析：用户看重能跑多远、车身沉不沉、防盗行不行，还冒出个新兴趣 —— 「骑行社交」（组队、晒路线）
• 竞品分析：传统品牌智能化不足；互联网品牌续航和线下售后是短板。
• 技术分析：更轻的电池材料与车身设计提升续航和便携，用 GPS + 移动网络做防盗，并与社交 App 打通。
• 最后汇总。

形状就是：START 同时分出三条分析分支，三条都汇入「汇总报告」，再指向 END。

在 LangGraph 里，工作流是搭在 LLM 之上的：LLM 出入两端之外，还挂着 Retrieval（检索）、Memory（记忆）、Tool（工具）三类增强能力 —— 也就是检索、短期记忆、工具调用这些。

再次强调：它不关心我用什么模型和提示词，解决的是状态管理、流程编排、持久化、人工监督这类工程问题。

理解图计算#

用节点和边表示复杂系统；在 AI 场景里，它特别适合搭多步骤、有状态的工作流。

用快递配送网络类比包裹从揽收站到配送站的过程：

对应关系（这张对照表是理解全篇的钥匙）：

这三件套里有几个名字得先对齐。

实现工作流的核心概念#

1
from typing import TypedDict
2

3
class PackageState(TypedDict):
4
    # 包裹基本信息
5
    package_id: str      # 包裹 id
6
    origin: str          # 始发站
7
    destination: str     # 目的地
8

9
    # 配送状态
10
    status: str          # "待揽收" / "运输中" / "派送中" / "已签收"
11
    history: list[str]   # 流转历史
12
    total_distance: int  # 总里程
13

14
    # 配送详情
15
    priority: str        # "普通" / "加急"

1
def receive_package(state: PackageState):
2
    """揽收站"""
3
    return {
4
        "status": "已揽收",
5
        "history": [f"在{state['origin']}揽收"],
6
    }
7

8
def sort_package(state: PackageState):
9
    """分拣中心：按目的地分拣"""
10
    destination = state["destination"]
11
    if "北京" in destination:
12
        next_station = "北京分拣中心"
13
    elif "上海" in destination:
14
        next_station = "上海分拣中心"
15
    else:
16
        next_station = "其他地区分拣中心"
17
    return {"status": "已分拣", "history": [f"分拣至{next_station}"]}
18

19
def final_delivery(state: PackageState):
20
    """派送站"""
21
    return {"status": "已签收", "history": [f"已送达{state['destination']}"]}

State 更新机制：Reducers#

这是我认为最容易踩坑、也最该先搞懂的一点。

节点返回的不是整个新状态，而只是一份「要改哪些字段」的增量。多个节点都要往同一个字段写时，怎么合并？答案靠 reducer。示例里用 Annotated 给 history、total_distance 指定了 reducer 函数 operator.add，语义就从「替换」变成了「追加 / 累加」。

1
from typing import Annotated
2
from operator import add
3

4
class PackageState(TypedDict):
5
    # 覆盖更新：每次新状态替换旧状态
6
    status: str
7

8
    # 追加更新：新的流转记录添加到历史列表
9
    history: Annotated[list[str], add]
10

11
    # 数值累加：里程数累加
12
    total_distance: Annotated[int, add]

知识点小结：status 走覆盖更新；history 走追加更新；total_distance 走数值累加。后两者之所以能累起来，全靠 Annotated[..., add] 里那个 reducer。

我现在的写法：默认字段用覆盖语义就够了（状态机式的「当前值」）；只有列表类字段（消息、历史、日志）才加 operator.add，否则每轮都会把前面的记录冲掉。

边与特殊节点的 API 细节#

1
from langgraph.graph import StateGraph, START, END
2

3
# 等价写法：显式走 state 模块
4
from langgraph.graph.state import StateGraph
5
# 完整限定名就是 langgraph.graph.state.StateGraph

编译到底做了什么#

第 2 步只是构建出 StateGraph，它还不能直接执行，必须先编译。compile() 会把它变成 CompiledStateGraph；产物实现了 Runnable，所以它能异步调、能流式传、也能批处理。

编译会顺手做个体检，主要看三件事：

• 每个节点从 START 出发都够得着
• 每个节点最终都能走到 END
• 不存在被孤立的节点，也没有死循环

注意这里有个认知坑：LangGraph 的「编译」和 C++/Java 那种语言编译完全不是一回事。它本质是运行时动态构建 + 校验一个图对象，不是把代码翻译成别的形态。

另外，默认情况下图的输入和输出结构相同；invoke() 接受单个输入（字典或任意类型），默认返回最新的 State。

编码前的版本说明#

• LangGraph 已经不再支持 Python 3.9，相关包都要求 Python 3.10 及以上。
• 用 python --version 先看一眼版本；低于 3.10 就直接换环境，别折腾。

【案例一】智能快递配送系统#

顺着步骤走一遍。

编码思路：Graph API 的三步#

建图顺序是固定的三步：① 定义状态 → ② 定义并添加节点和边 → ③ 编译。编译会顺手做结构检查（比如有没有孤立节点）。

写代码时的六步落地：

1. 定义 State（包裹信息）
2. 定义 StateGraph 图（成立快递公司）
3. 定义 Nodes（定义各个配送站点）
4. 添加 Nodes（把站点建进图里）
5. 添加 Edges（规划运输路线，含条件边）
6. 编译 + 测试

步骤 1：定义 State#

1
from typing import TypedDict, Annotated
2
from operator import add
3

4
# 1. 定义包裹状态
5
class PackageState(TypedDict):
6
    package_id: str             # 包裹 id
7
    origin: str                 # 始发站
8
    destination: str            # 目的地
9

10
    status: str                          # "待揽收" / "运输中" / "派送中" / "已签收"
11
    history: Annotated[list[str], add]   # 流转历史
12
    total_distance: Annotated[int, add]  # 总里程
13

14
    priority: str               # "普通" / "加急"

注意：状态既可以用 TypedDict，也可以用 Pydantic 模型；但 Pydantic 的性能不如 TypedDict，我一般只在需要校验时用它。

步骤 2：定义 StateGraph 图#

StateGraph 是一套有状态图计算框架，走**有向图（Directed Graph）**模型，天生适合多步骤、带状态的工作流。它把复杂流程变得可视、可拆 —— 像设计快递配送网一样设计软件系统。

1
from langgraph.graph import StateGraph, START, END
2

3
# 2. 成立快递公司
4
delivery = StateGraph(PackageState)

注意：到这一步只是有个构建器，还不能执行，必须编译。

步骤 3：定义 Nodes#

节点就是 Python 函数（同步或异步都行）。两条铁律：

• 节点把状态当入参收下
• 节点不需要返回整个状态，给一份增量补丁就够了

1
# 3. 定义各个配送站点
2
def receive_package(state: PackageState):
3
    """揽收站"""
4
    return {"status": "已揽收", "history": [f"在{state['origin']}揽收"]}
5

6
def sort_package(state: PackageState):
7
    """分拣中心：根据目的地分拣"""
8
    destination = state["destination"]
9
    if "北京" in destination:
10
        next_station = "北京分拣中心"
11
    elif "上海" in destination:
12
        next_station = "上海分拣中心"
13
    else:
14
        next_station = "其他地区分拣中心"
15
    return {"status": "已分拣", "history": [f"分拣至{next_station}"]}
16

17
def final_delivery(state: PackageState):
18
    """派送站"""
19
    return {"status": "已签收", "history": [f"已送达{state['destination']}"]}

步骤 4：添加 Nodes#

1
# 4. 添加配送站点
2
delivery.add_node("揽收站", receive_package)
3
delivery.add_node("分拣中心", sort_package)
4
delivery.add_node("派送站", final_delivery)

步骤 5：添加 Edges#

固定边最简路线：揽收站 → 分拣中心 → 派送站。

条件边：加急件走空运、普通件走陆运。

先补两个分支节点，再连线：

1
def standard_delivery(state: PackageState):
2
    """标准配送"""
3
    return {"status": "运输中", "history": ["标准陆运"], "total_distance": 500}
4

5
def express_delivery(state: PackageState):
6
    """加急配送"""
7
    return {"status": "加急运输", "history": ["空运加急"], "total_distance": 800}
8

9
delivery.add_node("标准配送", standard_delivery)
10
delivery.add_node("加急配送", express_delivery)
11

12
# 固定边
13
delivery.add_edge(START, "揽收站")
14
delivery.add_edge("揽收站", "分拣中心")
15

16
# 智能路由：根据优先级选择配送方式
17
def select_delivery(state: PackageState):
18
    """智能路由决策 - 根据包裹特性选择路线"""
19
    if state["priority"] == "加急":
20
        return "加急配送"
21
    return "标准配送"
22

23
delivery.add_conditional_edges(
24
    "分拣中心",               # source：起始节点
25
    select_delivery,          # path：确定下一个节点的可调用对象
26
    ["加急配送", "标准配送"],  # path_map
27
)
28

29
delivery.add_edge("标准配送", "派送站")
30
delivery.add_edge("加急配送", "派送站")
31
delivery.add_edge("派送站", END)

path_map 也可以用字典做「返回值 → 节点名」的映射：

1
def select_delivery(state: PackageState):
2
    if state["priority"] == "加急":
3
        return "备注加急"
4
    return "无备注"
5

6
delivery.add_conditional_edges(
7
    "分拣中心",
8
    select_delivery,
9
    {"备注加急": "加急配送", "无备注": "标准配送"},
10
)

步骤 6：编译与测试#

1
# 7. 编译系统
2
delivery_system = delivery.compile()
3

4
# 8. 测试配送
5
test_packages = [
6
    {"package_id": "P001", "origin": "北京", "destination": "上海",
7
     "priority": "普通", "history": [], "total_distance": 0},
8
    {"package_id": "P002", "origin": "广州", "destination": "乌鲁木齐",
9
     "priority": "加急", "history": [], "total_distance": 0},
10
]
11

12
for package in test_packages:
13
    result = delivery_system.invoke(package)
14
    print(package["package_id"], result["status"], result["history"], result["total_distance"])

跑出来的效果：P001 走标准陆运、总里程 500；P002 走空运加急、总里程 800；两者最终都是「已签收」。这一步很关键 —— 它验证了 reducer 真的在起作用：如果 history 没加 add，我只会看到最后一条记录。

案例一回顾#

【案例二】会联网搜索的智能代理系统#

先说要做成什么样。

案例介绍#

一个「模型自己判断要不要搜、搜完再答」的循环式代理。流程：

这个图要覆盖的能力清单：

• 智能对话与工具调用：模型自己决定调不调工具
• 自动搜索整合：把搜索结果回灌给模型
• 循环决策机制：搜完再判断，可以来回多轮
• 延伸场景：实时信息查询、事实核查、深度研究、知识扩展

编码思路：设置 Nodes / State / Edges#

设置 Nodes（靠单一职责原则拆）

• 节点 1 tool_node：专门负责搜索，拿搜索结果
• 节点 2 llm_call：专门负责调模型，拿最终结果

设置 State（靠共享性原则）

把历史对话存进 State 的 messages 列表，是构建对话系统的关键：

1
from typing import Annotated, TypedDict
2
from langchain_core.messages import AnyMessage
3
import operator
4

5
class AgentState(TypedDict):
6
    # 类型: list[AnyMessage]；合并策略: operator.add —— 新消息追加而非替换
7
    messages: Annotated[list[AnyMessage], operator.add]
8
    # 类型: int —— 跟踪 LLM 调用次数
9
    llm_call_count: int

注意 Annotated[list[AnyMessage], operator.add] 这个注解特别重要：operator.add 让新消息追加到列表而不是替换，这保证了对话历史的连续性。

messages 的三重作用：

1. 对话记忆：存储完整的对话流程
2. 上下文维护：llm_call 节点调用模型时，把整个 messages 当上下文传进去，模型基于完整历史生成回复
3. 状态持久化：messages 字段在节点之间传递，对话状态不会丢

没有这个字段，系统每次都像第一次对话。

代码实现#

步骤 1：先把模型和搜索工具备好#

1
# 步骤 1: 定义工具和模型
2
tools = [search_tool]              # 例如 TavilySearch 之类的搜索工具
3
model = init_chat_model("gpt-4o-mini", temperature=0)
4
model_with_tools = model.bind_tools(tools)   # 绑定工具
5
# 模型这一次的回复就是 new_ai_message = AIMessage(...)，随后被塞进 messages

步骤 2：定义状态（同上面的 AgentState）

步骤 3：模型节点#

1
# 步骤 3: 定义模型节点
2
def llm_call(state: AgentState):
3
    response = model_with_tools.invoke(
4
        [SystemMessage(content="你是一个乐于助人的助手")] + state["messages"]
5
    )
6
    # 只返回一个更新，reducer 会把它追加进 messages
7
    return {"messages": [response], "llm_call_count": state.get("llm_call_count", 0) + 1}

步骤 4：工具节点#

三个要点，我认为这段是案例二的核心。

• 要点 1：回顾 AIMessage 消息结构。 模型要调工具时，输出的 AIMessage 里带一个 tool_calls 属性，包含工具名和输入参数 —— 有这些就能发起工具调用。
• 要点 2：构造 ToolMessage。 光调用成功不够，必须把工具返回值包成 ToolMessage 再送回模型，才能拿到真正想要的答案。完整链路是：HumanMessage → AIMessage（含工具调用）→ ToolMessage → 模型再输出最终 AIMessage。

1
from langchain_core.messages import ToolMessage
2

3
tool_message = ToolMessage(
4
    content=str(tool_output),
5
    tool_call_id=ai_message.tool_calls[0]["id"],
6
)

• 要点 3：在 State 中访问 messages 用点表示法。 虽然可以传「字典形态」的消息，但 State 里的 message 更新总是会被反序列化为 LangChain Message 格式。所以要访问 AIMessage 的 tool_calls，得用 state["messages"][-1].tool_calls 这种点表示法，而不是下标取字典。

1
# 步骤 4: 定义工具节点
2
def tool_node(state: AgentState):
3
    outputs = []
4
    for tool_call in state["messages"][-1].tool_calls:
5
        result = tools_by_name[tool_call["name"]].invoke(tool_call["args"])
6
        outputs.append(ToolMessage(content=str(result), tool_call_id=tool_call["id"]))
7
    return {"messages": outputs}

步骤 5：构建图#

1
# 步骤 5: 构建图
2
from langgraph.graph import StateGraph, START, END
3
from langgraph.prebuilt import ToolNode, tools_condition
4

5
builder = StateGraph(AgentState)
6
builder.add_node("llm_call", llm_call)
7
builder.add_node("tool_node", ToolNode(tools))   # ToolNode 直接吃工具列表
8
builder.add_edge(START, "llm_call")
9

10
# 定义结束逻辑：手写 should_continue
11
def should_continue(state: AgentState):
12
    last = state["messages"][-1]
13
    if last.tool_calls:            # 如果 LLM 调用工具，则执行操作
14
        return "tool_node"
15
    return END
16

17
builder.add_conditional_edges("llm_call", should_continue, ["tool_node", END])
18
builder.add_edge("tool_node", "llm_call")
19

20
# 编译图
21
agent_builder = builder.compile()

我的习惯：能用手写的 should_continue 就手写，逻辑一眼能看懂；只判断「有没有工具调用」这种标准场景，直接用框架给的 tools_condition 更省事（案例三就用它）。

步骤 6：可视化图#

编译出来的图带一个 get_graph()，传 xray=True（或 xray=1）就是详细视图；接着 draw_mermaid_png() 会吐出 PNG 的二进制内容；再用 matplotlib 显示即可。

1
import matplotlib.pyplot as plt
2

3
png_bytes = agent_builder.get_graph(xray=1).draw_mermaid_png()
4
with open("graph.png", "wb") as f:      # 保存文件
5
    f.write(png_bytes)
6
plt.imshow(plt.imread("graph.png"))     # 使用 matplotlib 显示图像
7
plt.show()

Mermaid 本身是一种用文本描述流程图的标记语言，画甘特图、时序图、饼图都行；不装 matplotlib 也可以直接把文本贴到在线编辑器里看。

步骤 7：执行（非流式与流式）#

1
# 非流式
2
result = agent_builder.invoke({"messages": [HumanMessage(content="今天天气怎么样？")]})
3
print(result["messages"][-1].content)
4

5
# 流式：看图的执行步骤
6
for chunk in agent_builder.stream({"messages": [HumanMessage(content="今天天气怎么样？")]}):
7
    print(chunk)

注意这里的「流式」指的是图执行步骤的流式输出（经过节点的过程），不是 token 流。每个节点对 State 的更新值（注意：是将要更新的值，不是更新后的值）都能这样追踪。所以 chunk 类型是 dict：key 是节点名，value 是待更新的 State 片段。

【案例三】代理式 RAG：会自我纠错的检索增强生成系统#

先看它跟链式写法差在哪。

案例介绍#

链式 RAG 是一条直线，好处是简单：

• 概念少：只要理解 Runnable、StrOutputParser 这类基础东西
• 逻辑简单：线性流程，好理解好调试
• 代码直观：一眼能看出数据怎么流
• 快速上手：几分钟就能跑通

但它的形状是写死的。这个案例要做的是按检索质量动态调整查询策略的文档问答系统：

举个我实际跑过的例子：知识库放几篇课程宣传文档，然后问「这些课都包含哪些方向」「Java 方向的课程安排是什么」「Redis 课程讲了什么」。当用户问「总共提供了哪些课程」时，链路是：

1. 模型决定调用检索工具，用「课程」相关关键词去搜
2. 检索到文档后做相关性评估
3. 搭得上就直接答；搭不上就把问题改写一遍再搜

鲁棒性是这个案例的关键词。它指的是系统在各种内外部干扰下仍能保持性能稳定：

• 说的是有不确定性、有噪声、有误差时它表现如何
• 它和稳定性不同：稳定性关注扰动消失后能否恢复，鲁棒性关注持续扰动下还稳不稳
• 放到模型上：环境陌生或带噪声时，它照样能把预期任务做完

一句话对比：链式 RAG 解决「怎么让 RAG 跑起来」，这个案例解决「怎么让它跑得更好、更智能」。

编码思路#

• 模块化设计：一个节点只担一件事
• 质量闭环：查 → 评 → 改 → 再查
• 智能路由：下一步走哪儿由 AI 自己定，不靠人工干预

四步走：

1. 把「知识库」备齐（加载 + 切分）
2. 创建「检索工具」
3. 把流程里的节点设计出来
   • 节点 1：generate_query_or_respond，负责决策
   • 节点 2：retrieve，检索工具
   • 节点 3：rewrite_question，改写提问
   • 节点 4：generate_answer，产出答案
4. 组装「工作流水线」
   • 条件边 1：让 LLM 决定要不要去知识库里搜
   • 条件边 2：判断捞回来的文档跟问题搭不搭

代码实现#

1
# 聊天模型与嵌入模型
2
model = init_chat_model("gpt-4o-mini", temperature=0)
3
embeddings = init_embeddings("openai:text-embedding-3-small")
4

5
# 加载文档列表 → 切分 → 建向量库
6
docs = loader.load()
7
splits = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
8
    chunk_size=100, chunk_overlap=50
9
).split_documents(docs)
10

11
# 使用内存中向量存储和 OpenAI 嵌入
12
vectorstore = InMemoryVectorStore.from_documents(splits, embedding=embeddings)
13
retriever = vectorstore.as_retriever()
14

15
# 使用 LangChain 的预构建 create_retriever_tool 创建检索器工具
16
retriever_tool = create_retriever_tool(
17
    retriever,
18
    "retrieve_blog_posts",
19
    "搜索并返回关于课程内容的文档片段。",
20
)

1
def generate_query_or_respond(state: MessagesState):
2
    response = model.bind_tools([retriever_tool]).invoke(state["messages"])
3
    return {"messages": [response]}

1
retrieve = ToolNode([retriever_tool])

1
def rewrite_question(state: MessagesState):
2
    question = state["messages"][0].content
3
    response = model.invoke([
4
        HumanMessage(content=f"请把这个问题改写得更利于检索：\n\n{question}")
5
    ])
6
    return {"messages": [HumanMessage(content=response.content)]}

1
def generate_answer(state: MessagesState):
2
    question = state["messages"][0].content                                  # 最原始问题
3
    context = state["messages"][-1].content                                  # 最新问题的检索结果
4
    response = model.invoke([
5
        HumanMessage(content=f"根据以下文档回答问题：\n\n{context}\n\n问题：{question}")
6
    ])
7
    return {"messages": [response]}

1
workflow = StateGraph(MessagesState)
2

3
# 添加节点与入口点
4
workflow.add_node("generate_query_or_respond", generate_query_or_respond)
5
workflow.add_node("retrieve", retrieve)
6
workflow.add_node("rewrite_question", rewrite_question)
7
workflow.add_node("generate_answer", generate_answer)
8
workflow.add_edge(START, "generate_query_or_respond")
9

10
# 条件边1：LLM 决策是否需要进行知识库检索
11
workflow.add_conditional_edges(
12
    "generate_query_or_respond",
13
    tools_condition,          # 等价于手写 should_continue
14
)
15

16
# 条件边2：检测检索到的文档是否与问题相关
17
def grade_documents(state: MessagesState) -> Literal["generate_answer", "rewrite_question"]:
18
    question = state["messages"][0].content
19
    context = state["messages"][-1].content
20
    prompt = f"判断这段检索内容与问题是否相关，只回答 yes 或 no。\n\n{context}\n\n问题：{question}"
21
    response = model.with_structured_output(GradeDocuments).invoke(
22
        [HumanMessage(content=prompt)]
23
    )
24
    score = response.binary_score       # yes / no
25
    return "generate_answer" if score == "yes" else "rewrite_question"
26

27
workflow.add_conditional_edges("retrieve", grade_documents)
28
workflow.add_edge("generate_answer", END)
29
workflow.add_edge("rewrite_question", "generate_query_or_respond")
30

31
rag_agent = workflow.compile()

LangGraph 的其他特性#

使用 Overwrite 绕过 reducer#

reducer 决定状态更新怎么合并。但总有想「直接覆盖」的时候：聊天应用里新消息通常是追加，可我要清空聊天记录重开，就得绕过追加逻辑。

1
from langgraph.types import Overwrite
2

3
def reset_messages(state: MessagesState):
4
    # 绕过 reducer 并替换整个消息列表
5
    return {"messages": Overwrite(["replacement message"])}

输出只剩 ["replacement message"]，前面那几条不会留下。

注意：Overwrite 要从 langgraph.types 导入；用法就是把新值包进 Overwrite(...) 再返回。

适用场景：

• 重置对话：把聊天记录清空重开
• 状态重置：出错恢复之后把应用状态拨回去
• 数据清理：把坏掉或过期的那份换掉

定义输入输出模式#

默认只用一套状态模式。可以拆成三份：

• 输入模式独立：管进来的数据合不合规
• 输出模式独立：该露的才露，其余滤掉
• 内部模式：节点之间传的那份完整状态

为什么要拆？考虑问答系统：输入是用户问题（字符串），输出是 AI 答案（字符串），但内部要存中间结果、上下文 —— 我不想把这些内部状态暴露给用户。

StateGraph 的初始化参数：

1
# 1. 定义输入模式 - 只包含用户问题
2
class InputState(TypedDict):
3
    question: str
4

5
# 2. 定义输出模式 - 只包含 AI 答案
6
class OutputState(TypedDict):
7
    answer: str
8

9
# 3. 定义完整状态模式（内部使用）
10
class OverallState(InputState, OutputState):
11
    pass
12

13
def answer_node(state: OverallState):
14
    # 这里可以访问 question，生成 answer
15
    return {"answer": "..."}
16

17
# 构建图时指定输入输出模式
18
builder = StateGraph(OverallState, input_schema=InputState, output_schema=OutputState)

跑一下会发现 question 被滤掉了，输出里看不到。适用场景：写 API（请求与响应边界清楚）、微服务（服务之间约定好字段）、数据管道（进出规格写明白）。

在节点间传递私有状态#

有些临时数据对中间逻辑重要，但不该出现在最终输出里，只在特定节点之间共享 —— 这就是私有状态。

场景：根据数据库信息生成报告，三个节点 ——

1. 节点 1：捞原始数据（里面有敏感字段）
2. 节点 2：加工一遍，把敏感字段剔掉
3. 节点 3：出报告

节点 1 和 2 要共享原始数据，节点 3 不该看到敏感信息。

1
class OverallState(TypedDict):
2
    final_result: str            # 公共状态（最终输出中可见）
3

4
class Node1Output(TypedDict):
5
    sensitive_data: str          # 节点1的私有输出
6

7
class Node2Input(OverallState):
8
    sensitive_data: str          # 节点2需要的输入（包含私有数据）
9

10
def node_1(state: OverallState) -> Node1Output:
11
    return {"sensitive_data": "原始敏感数据"}
12

13
def node_2(state: Node2Input) -> OverallState:
14
    # 处理数据，返回清理后的结果
15
    return {"final_result": "已清理的报告"}
16

17
def node_3(state: OverallState) -> OverallState:
18
    return {"final_result": state["final_result"] + "（终稿）"}
19

20
# 构建图
21
builder = StateGraph(OverallState)
22
# add_sequence：支持添加一系列节点，按所给的顺序执行。
23
builder.add_sequence([node_1, node_2, node_3])

要点：私有状态靠函数签名上的类型标注声明 —— 入参写 Node2Input，返回值写 Node1Output，图本身用 OverallState 当公共状态。Node1Output 里的 sensitive_data 只在相邻两步之间用，落不到最终状态里；节点 3 拿到的只有 final_result。

在节点间传递私有状态（续）#

跑起来会看到：sensitive_data 不出现在最终状态里，只剩 final_result。

适用场景：数据处理（步骤间的临时量）、鉴权流程（令牌这类敏感值）、复杂计算（中间结果）、错误处理（细节留在内部，对外只给友好提示）。

1. 提示链模式（Prompt Chaining）#

概念#

提示链就是串糖葫芦：上一步的输出就是下一步的输入。内容创作最典型 —— 大纲 → 初稿 → 润色 → 最终稿，每一步的产出都往下传，质量才逐级往上抬。

模式实践#

四个节点，每个只干一件事：

• generate_outline：只管写大纲
• generate_draft：只管写初稿
• polish_content：只管润色
• finalize_content：只管出终稿

这个模式里我会顺手用上「独立输入输出模式」：

1
# 1. 定义输入模式 - 只包含用户输入
2
class InputState(TypedDict):
3
    topic: str
4

5
# 2. 定义输出模式 - 只包含最终结果
6
class OutputState(TypedDict):
7
    final_article: str
8

9
# 3. 定义完整状态模式（内部使用）
10
class OverallState(InputState, OutputState):
11
    outline: str
12
    draft: str
13
    polished: str
14

15
def generate_outline(state: OverallState):      # 第一步：生成大纲
16
    return {"outline": llm.invoke(f"给这个主题写大纲：{state['topic']}").content}
17

18
def generate_draft(state: OverallState):        # 第二步：基于大纲生成初稿
19
    return {"draft": llm.invoke(f"按大纲写初稿：{state['outline']}").content}
20

21
def polish_content(state: OverallState):        # 第三步：润色稿件
22
    return {"polished": llm.invoke(f"润色这段文字：{state['draft']}").content}
23

24
def finalize_content(state: OverallState):      # 第四步：生成最终稿
25
    return {"final_article": llm.invoke(f"输出最终稿：{state['polished']}").content}
26

27
# 构建工作流
28
workflow = StateGraph(OverallState, input_schema=InputState, output_schema=OutputState)
29
workflow.add_node("generate_outline", generate_outline)   # 添加节点
30
workflow.add_node("generate_draft", generate_draft)
31
workflow.add_node("polish_content", polish_content)
32
workflow.add_node("finalize_content", finalize_content)
33

34
# 连接节点（直线流程）
35
workflow.add_edge(START, "generate_outline")
36
workflow.add_edge("generate_outline", "generate_draft")
37
workflow.add_edge("generate_draft", "polish_content")
38
workflow.add_edge("polish_content", "finalize_content")
39
workflow.add_edge("finalize_content", END)
40

41
app = workflow.compile()      # 编译工作流
42
app.invoke({"topic": "人工智能的未来发展"})   # 使用工作流

这个例子跑出来的提纲长这样（模型自己发挥的标题层级）：

• 技术趋势与创新方向
  • 自然语言处理的进步
  • 机器学习与深度学习的演变
  • 强人工智能的可能性
• 应用领域与社会影响
  • 医疗行业的变革
  • 自动化与就业市场的关系
  • 伦理与监管的挑战

我认识的优点：逻辑简单、代码直观、快速上手，调试时每个节点的日志一眼能对上是哪一步。

2. 并行化模式（Parallelization）#

概念#

并行化：多个任务同时跑、最后汇总，效率高。适合从多个角度处理同一个问题。

还是那个智能电动自行车的例子（城市通勤，带导航、社交、防盗）。开工前先做多维分析：

• 市场分析：用户看重能跑多远、车重、防盗，还冒出个新兴趣 —— 「骑行社交」（组队、晒路线）
• 竞品分析：老牌子智能化做得不够；互联网牌子短板在续航和线下售后
• 技术分析：用更轻的电池与车身方案把续航和便携提上去，再基于 GPS + 移动网络做防盗，并和社交 App 打通

最后汇总。并行分析不只是省时间，决策质量也更好。