首页 / 文章

langgraphpythonagentsworkflowshuman-in-the-loop

Python中的LangGraph：状态、并行节点、检查点与人工审核

用 Python 实践 LangGraph：TypedDict 状态与归约器、边、检查点机制、中断处理，以及需人工审批的邮件分类工作流程。

发布者：REACTAPP.TOP阅读约 25 分钟5723 词

来源: LangGraph Essentials in Python: Build AI Agent Workflows with State, Routing, and Human-in-the-Loop

任何在大型语言模型基础上开发过非玩具级应用的人都会遇到三种反复出现的痛点：连续发起的请求会累积延迟；复杂的流程一旦崩溃就会丢失所有进展；最糟糕的情况是，模型会在最不合时宜的时刻给出完全错误的、却显得很有把握的答案。

LangGraph解决了这三大问题。它是一个基于图的编排框架，由几个简单组件构成：状态用TypedDict表示，节点为普通函数，边则负责路由逻辑。这些基本元素组合在一起，能够实现节点的并行运行、自动保存状态以便从崩溃中恢复，以及在关键环节暂停流程供人工审核。

接下来的章节将逐步讲解。首先是一个单独的节点，接着是并行执行、条件路由和持久内存，随后是人工干预中断。最后，一个完整的邮件分类工作流会将所有这些内容整合起来。每个示例都应保持原样运行。TypeScript SDK 提供的 API 几乎相同，因此可以运用相同的思维模型。

设置

首先需要两个包：

pip install langgraph langchain-openai

接着是 OpenAI 的环境变量，以及可选的 LangSmith 跟踪参数：

export OPENAI_API_KEY="your-key-here"
# Optional: enable LangSmith tracing for debugging
export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY="your-langsmith-key"

这些示例是使用 langgraph 1.1.3（2026年3月版本）和 langchain-openai 1.1.10 进行测试的，需要Python 3.10或更高版本。旧版指南中有时会使用 LANGCHAIN_TRACING_V2 和 LANGCHAIN_API_KEY；这两种方式仍然有效，但目前更推荐使用 LANGSMITH_TRACING 和 LANGSMITH_API_KEY。

LangGraph的整个设计基于两个核心理念：状态用于存储数据，节点则是用于处理这些数据的函数。一旦理解了这两个概念，框架的其余部分就会迎刃而解。

使用TypedDict定义状态

在LangGraph中，状态其实就是普通的Python TypedDict，没有特殊的基类，也没有框架专用的装饰器，仅仅是带类型的字典而已：

from typing import Annotated
from typing_extensions import TypedDict
import operator

class State(TypedDict):
    messages: Annotated[list[str], operator.add]
    summary: str

该类定义了在图中流动的数据结构，每个节点都从这一共享结构中读取数据并向其写入数据。

为何选择TypedDict而非普通字典或数据类？因为它能实现静态类型检查，因为mypy和Pyright都能识别它以及其中的类型提示，同时还能避免数据类或Pydantic模型带来的运行时开销。LangGraph还能通过检查这些注解来了解数据结构，这对于结构化输出和验证非常重要。

需要理解的是Annotated[list[str], operator.add]这行代码，它用于定义归约器。这是LangGraph状态中最核心的概念，几乎每个人在初次接触时都会被它难住。

如果没有归约器，返回 {"summary": "new value"} 的节点会直接替换存储的值。对于 summary 这类标量字段，这正是所需的行为。但当有两个并行节点同时向 messages 写入数据时，问题就出现了：只有最后完成写入的节点的内容会被保留。而通过声明 operator.add，则可以告诉 LangGraph 进行追加操作，这样所有对 messages 的写入内容都能安全地累积起来。

更复杂的合并逻辑则需要自定义归约器：

def keep_latest_n(existing: list[str], new: list[str]) -> list[str]:
    """Keep only the last 10 messages."""
    combined = existing + new
    return combined[-10:]

class State(TypedDict):
    messages: Annotated[list[str], keep_latest_n]

该 reducer 会将现有列表与新消息合并，然后只保留最新的十条消息，这有助于处理那些否则会导致令牌数量激增的长时间对话。operator.add 用于简单的追加操作；而要实现去重、限制列表大小或进行超出简单追加的合并操作，则需要自定义 reducer。

常见的错误是在并行节点写入数据的字段上忘记使用 reducer。如果没有 reducer，各个分支会默默地互相覆盖，最后完成写入的那个分支的内容会保留下来，且不会出现任何错误提示。

TypedDict 与 Pydantic

两者都会出现在 LangGraph 中，且它们之间的区分是刻意为之的。TypedDict 用于表示图的状态，即在图中移动的容器。LangGraph 是围绕它设计的，其字段支持 Annotated 约简器语法，能够清晰地序列化以便保存检查点，且不会增加运行时开销。

Pydantic 用于处理面向大语言模型的架构。调用 .with_structured_output(SomeModel) 可以为大语言模型提供 JSON 架构作为参考，并验证回复是否符合该架构。若回复不匹配，会立即引发错误，防止格式错误的数据在图中传播。

后续的核心工作流程正是遵循这种划分：EmailState 是作为图容器的 TypedDict，而 EmailClassification 是用于描述大语言模型输出的 Pydantic 模型。

编写第一个节点

每个节点以完整状态作为输入，返回一个部分更新字典。LangGraph会将该更新整合到状态中，并应用已声明的任何归约函数：

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END

class ChatState(TypedDict):
    messages: Annotated[list[str], operator.add]

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def chatbot(state: ChatState) -> dict:
    """Call the LLM with the conversation history."""
    response = llm.invoke(state["messages"])
    return {"messages": [response.content]}

# Build the graph
graph = StateGraph(ChatState)
graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)
app = graph.compile()
# Run it
result = app.invoke({"messages": ["What is LangGraph?"]})
print(result["messages"][-1])

逐行解析代码：

  • StateGraph(ChatState) 会创建一个以 ChatState 作为架构的图结构，LangGraph 会读取相关注释以了解还原函数及字段类型。
  • graph.add_node("chatbot", chatbot) 会以 "chatbot" 为名称注册该函数，这也是边定义所引用的名称。
  • graph.add_edge(START, "chatbot") 会指定 "chatbot" 作为第一个执行的节点。START 是一个内置的哨兵节点。
  • graph.add_edge("chatbot", END) 会在 chatbot 的执行完成后终止整个流程。END 是另一个内置的哨兵节点。
  • graph.compile() 会验证图的结构，检查是否存在悬空边或无法访问的节点，最终返回一个可运行的 CompiledGraph。
  • app.invoke({"messages": [...]}) 会同步执行该图计算并返回最终状态。

    • 由于 messages 使用 operator.add 作为归约函数，该节点会将字符串包裹在列表中。返回 {"messages": [response.content]} 可以在现有列表基础上追加一个只包含一个元素的列表，而直接返回字符串则会在运行时出错，因为 operator.add 无法将字符串添加到列表中。

    规则是让节点的返回值与字段的类型相匹配，包括归约函数在内。使用 operator.add 的 list[str] 类型字段需要接收 list[str] 类型的值，而没有归约函数的普通 str 类型字段则需要接收 str 类型的值。

    这就是一个仅约20行代码的完整可运行 LangGraph 应用。接下来要探讨的是 invoke() 函数内部的实现机制。

    超级步骤：执行机制

    LangGraph以超级步骤的形式运行节点。同一步骤中安排的节点会同时执行，且所有节点必须全部完成之后才能进入下一步。单节点示例中，一个超级步骤仅包含一个节点；添加并行边后，多个节点会被置于同一个超级步骤中。

    这一点很重要，原因有二：首先，模型从一开始就内置了并行执行机制，而非后期添加的功能；其次，借助检查点机制，LangGraph会在每个超级步骤完成后保存完整的状态快照。这正是实现崩溃恢复的关键：如果进程在运行过程中中断，执行可以从最后一个已完成的高级步骤处继续。

    在了解了状态如何在单个节点间传递之后，下一个问题就是如何控制执行顺序或让节点同时运行。这就是边的作用。

    边：串行、并行与条件执行

    边控制着执行流程在节点之间的流转方式。LangGraph提供了三种类型的边，它们共同涵盖了所有可能出现的路由模式。最简单的形式已在聊天机器人示例中出现过。

    串行边

    最简单的情况是一个节点之后紧接着是另一个节点。

    class DraftState(TypedDict):
    topic: str
    draft: str
    review: str

def generate_draft(state: DraftState) -> dict:
    response = llm.invoke(f"Write a short paragraph about: {state['topic']}")
    return {"draft": response.content}

def review_draft(state: DraftState) -> dict:
    response = llm.invoke(f"Review this draft and suggest improvements:\\n{state['draft']}")
    return {"review": response.content}

graph = StateGraph(DraftState)
graph.add_node("generate", generate_draft)
graph.add_node("review", review_draft)
graph.add_edge(START, "generate")
graph.add_edge("generate", "review")
graph.add_edge("review", END)

app = graph.compile()
result = app.invoke({"topic": "Why Python developers love type hints"})
print(result["review"])

    generate_draft会首先执行。一旦它完成，LangGraph会将draft字段合并到状态中，随后review_draft才会执行，并能从state["draft"]中读取草稿内容。由于边关系强制规定了执行顺序，因此无需额外的排序代码，这种串行执行方式就能得到保证。只要节点B依赖于节点A的输出，就可以采用串行边关系：先生成再审核，先获取再处理，先分类再路由。

    意外形成的循环是一个常见的问题。默认情况下LangGraph期望的是有向无环图，因此除非明确将图设置为循环结构，否则从review节点向generate节点添加边会在编译时引发错误。迭代循环通常是通过可路由回 earlier节点的条件边来构建的。

    并行边

    当两条边从一个节点发出时，它们的目标会在同一个超级步骤中同时处理。此时归约器就显得至关重要：

    class ResearchState(TypedDict):
    topic: str
    findings: Annotated[list[str], operator.add]
    outline: str
    final_summary: str

def research(state: ResearchState) -> dict:
    response = llm.invoke(f"List 3 key facts about: {state['topic']}")
    return {"findings": [f"Research: {response.content}"]}

def create_outline(state: ResearchState) -> dict:
    response = llm.invoke(f"Create a brief outline for an article about: {state['topic']}")
    return {"outline": response.content}

def combine(state: ResearchState) -> dict:
    research_text = "\\n".join(state["findings"])
    response = llm.invoke(
        f"Combine this research and outline into a summary:\\n"
        f"Research:\\n{research_text}\\n\\n"
        f"Outline:\\n{state['outline']}"
    )
    return {"final_summary": response.content}

graph = StateGraph(ResearchState)
graph.add_node("research", research)
graph.add_node("outline", create_outline)
graph.add_node("combine", combine)
# Fan-out: both start after START
graph.add_edge(START, "research")
graph.add_edge(START, "outline")
# Fan-in: both feed into combine
graph.add_edge("research", "combine")
graph.add_edge("outline", "combine")
graph.add_edge("combine", END)
app = graph.compile()
result = app.invoke({"topic": "LangGraph parallel execution"})
print(result["final_summary"])

    START发出的两条边将research和outline安排在同一个超级步骤中，LangGraph会同时执行它们。这两者也都有一条边指向combine，在所有指向它的节点都完成之前，LangGraph不会安排combine的执行，这起到了自动同步屏障的作用。

    尽管此处只有research会向findings写入数据，但它仍有一个归约器，因为实际系统中可能会有四个并行运行的研究节点向同一个列表添加内容，如果没有归约器的话，最终只会保留一个结果。

    延迟的计算很简单。如果research需要2秒，outline需要3秒，串行执行需要5秒，而并行执行则需要较长的那个任务所需的3秒。对于独立的LLM调用而言，并行执行几乎可以实现无延迟的效率提升。

    适合并行化的节点对是指那些彼此的输出互不依赖的节点对。即便将两个耗时一秒的调用合并到同一个超级步骤中，经过数千次请求后也会产生显著效果。需注意那些看似独立但实际上以不太明显的方式共享状态的节点。如果两个并行节点在没有归约器的情况下同时写入同一个字段，就会出现竞态条件，最终由完成得较晚的节点决定该字段的值，而LangGraph不会发出任何警告。

    这种分发与汇聚模式是邮件处理功能的核心，它在起草回复之前会并行执行知识库搜索和错误追踪查询。

    条件边

    当下一个节点依赖于当前节点的输出时，就会应用add_conditional_edges函数：

    class ClassifyState(TypedDict):
    text: str
    sentiment: str
    result: str

def classify(state: ClassifyState) -> dict:
    response = llm.invoke(
        f"Classify the sentiment of this text as 'positive' or 'negative'. "
        f"Reply with just the word.\\n\\n{state['text']}"
    )
    return {"sentiment": response.content.strip().lower()}

def handle_positive(state: ClassifyState) -> dict:
    return {"result": f"Great feedback! Text was: {state['text']}"}

def handle_negative(state: ClassifyState) -> dict:
    return {"result": f"Escalating concern. Text was: {state['text']}"}

def route_by_sentiment(state: ClassifyState) -> str:
    """Routing function: inspect state, return a string key."""
    if state["sentiment"] == "positive":
        return "positive"
    return "negative"

graph = StateGraph(ClassifyState)
graph.add_node("classify", classify)
graph.add_node("handle_positive", handle_positive)
graph.add_node("handle_negative", handle_negative)
graph.add_edge(START, "classify")
graph.add_conditional_edges(
    "classify",
    route_by_sentiment,
    {"positive": "handle_positive", "negative": "handle_negative"}
)
graph.add_edge("handle_positive", END)
graph.add_edge("handle_negative", END)
app = graph.compile()
result = app.invoke({"text": "This product is absolutely fantastic!"})
print(result["result"])
# Great feedback! Text was: This product is absolutely fantastic!

    route_by_sentiment函数会分析状态并返回一个字符串键，而传递给add_conditional_edges的映射字典会将该键转换为节点名称。纯函数式的路由函数（输入为状态，输出为字符串，无副作用）易于测试，因为可以不借助大型语言模型即可进行单元测试。

    add_conditional_edges接受三个参数：源节点、路由函数，以及一个将字符串键映射到目标节点名称的字典。路由函数可能返回的每一个键都必须存在于该字典中，否则LangGraph会在运行时抛出错误。

    Command替代方案

    当一个节点需要在同一步骤中同时更新状态和进行路由时，它可以直接返回一个Command，从而跳过add_conditional_edges：

    from langgraph.types import Command

def classify_and_route(state: ClassifyState) -> Command:
    response = llm.invoke(
        f"Classify sentiment as 'positive' or 'negative'.\\n\\n{state['text']}"
    )
    sentiment = response.content.strip().lower()
    if sentiment == "positive":
        return Command(update={"sentiment": sentiment}, goto="handle_positive")
    else:
        return Command(update={"sentiment": sentiment}, goto="handle_negative")

    一个 Command 返回值同时包含状态更新（update=）和路由选择（goto=），因此节点可以在一次操作中完成数据写入并选定下一个目标。

    当路由逻辑仅读取状态而不写入新数据、路由处理应与节点逻辑分离，或多个源节点使用相同的路由功能时，add_conditional_edges 是最佳选择。

    当节点生成会影响其自身路由的新数据、一次性场景无需单独的路由功能，或目标取决于节点内部计算的结果而非现有状态时，Command 是更合适的选择。

    在人工审核环节，email capstone依赖于add_conditional_edges，因为该环节的路由逻辑非常简单：它仅检查状态中已存储的优先级和置信度。

    内存与检查点机制

    在控制了排序和分支逻辑之后，下一个问题就是持久性：当进程在对话进行到一半时重新启动，会发生什么？

    到目前为止，每次调用都是从空状态开始的。而对话和多步骤工作流需要能够在多次调用之间保留状态。LangGraph通过检查点机制来解决这一问题，其API设计十分简洁，因此从开发环境过渡到生产环境时的持久化处理只需修改一行代码即可。

    用于开发的MemorySaver

    MemorySaver会将状态保存在内存中，这非常适合开发和测试场景：

    from langgraph.checkpoint.memory import MemorySaver

class ConversationState(TypedDict):
    messages: Annotated[list[str], operator.add]

def chatbot(state: ConversationState) -> dict:
    history = "\\n".join(state["messages"])
    response = llm.invoke(f"Conversation so far:\\n{history}\\n\\nRespond to the latest message.")
    return {"messages": [response.content]}

graph = StateGraph(ConversationState)
graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)

# Compile WITH a checkpointer
memory = MemorySaver()
app = graph.compile(checkpointer=memory)

# Thread 1: a conversation about Python
config_1 = {"configurable": {"thread_id": "user-alice"}}
app.invoke({"messages": ["Hi, I'm learning Python"]}, config=config_1)
result = app.invoke({"messages": ["What should I learn first?"]}, config=config_1)
print(result["messages"][-1])

# The response knows about the previous message because state persisted
# Thread 2: a completely independent conversation
config_2 = {"configurable": {"thread_id": "user-bob"}}
result = app.invoke({"messages": ["Tell me about Rust"]}, config=config_2)
print(result["messages"][-1])
# This conversation has no knowledge of Alice's conversation

    使用 checkpointer=memory 时，编译后的图结构会在每个超级步骤之后保存完整的快照。使用相同的 thread_id 再次调用 invoke() 时，LangGraph 会首先加载该线程的最新检查点。thread_id 的作用类似于会话 ID：重复使用它可继续处理，更改它则可重新开始。

    由于检查点已存储了之前的消息，第二次调用时可以省略这些消息。如果 Alice 使用 {"messages": ["What should I learn first?"]} 调用，LangGraph 会加载 "user-alice" 检查点，找到已存储的消息列表，然后通过 operator.add 减算器添加新消息。一旦 chatbot 开始运行，state["messages"] 中就会包含所有的历史消息。

    将同一个 thread_id 用于无关的对话是错误的。如果 "user-alice" 被重复用于不同主题，聊天机器人会带上艾丽丝之前的所有对话内容，因此每个新会话都应拥有全新的 thread_id。在开发过程中，app.get_state_history(config) 可以显示某个对话线程的完整检查点历史，LangSmith 中的时间回溯调试功能也是基于同一机制实现的。

    适用于生产环境的持久化检查点工具

    MemorySaver 存在于进程内存中，因此当进程终止时，相关状态也会随之丢失。生产环境需要一种持久化的后端机制：

    # Install: pip install langgraph-checkpoint-postgres
from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:pass@localhost:5432/mydb"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # Creates tables on first run
    app = graph.compile(checkpointer=checkpointer)
    # Same API, durable state
    result = app.invoke({"messages": ["Hello"]}, config=config_1)

    checkpointer.setup()会在首次运行时创建所需的数据库表，之后则无需执行任何操作，因此只需在初始化时调用一次。由于检查点机制位于共享的抽象层之后，不同后端之间的API保持一致。MemorySaver、PostgresSaver和RedisSaver遵循相同的协议，因此图处理代码无需知道它正在与哪一个组件通信。

    选择方案时需考虑：

    • MemorySaver适用于开发、测试和演示场景，因为这些场景并不要求在重启后数据依然存在。
    • PostgresSaver适用于需要数据持久性以及能够查询检查点历史的生产环境。
    • 来自langgraph-redis的RedisSaver适用于高吞吐量场景，这类场景需要快速读取检查点数据，并且能够接受Redis在持久性方面的妥协。

    会保存什么

    每个超级步骤都会保存完整的快照。这实现了时间回溯功能：可以查看历史中的任意状态，从选定的检查点重新执行操作，或进行调试以确定问题出在何处。中断功能也基于同样的机制。当节点调用interrupt()时，LangGraph需要一个地方来保存暂停后的状态，以便日后恢复执行。如果没有检查点，就无法保存状态，中断机制就会失效。

    结合中断机制的人机协同

    有些决策过于重要，不能完全自动化处理。在智能体发送客户邮件、批准退款或部署代码之前，都应由人工进行审核和批准。LangGraph通过interrupt()函数来实现这一功能。

    interrupt()函数

    在节点内部调用interrupt()会暂停执行，将状态写入检查指针，并将有效载荷传回调用方：

    from langgraph.types import interrupt, Command

class ReviewState(TypedDict):
    messages: Annotated[list[str], operator.add]
    draft: str
    approved: bool

def generate(state: ReviewState) -> dict:
    response = llm.invoke("Write a short thank-you email to a customer.")
    return {"draft": response.content}

def review(state: ReviewState) -> dict:
    """Pause here for human review."""
    decision = interrupt(f"Please review this draft:\\n\\n{state['draft']}")
    if decision == "approve":
        return {"approved": True}
    return {"approved": False}

def send(state: ReviewState) -> dict:
    if state["approved"]:
        return {"messages": ["Email sent successfully."]}
    return {"messages": ["Email was rejected. Not sent."]}

    该调用依次执行四项操作：

    1. 在调用发生的确切行处暂停执行。
    2. 将完整状态保存到检查指针中，这也是需要检查指针的原因。
    3. 将传入的有效载荷，即草稿文本，返回给调用 invoke() 的方。
    4. 进入等待状态，使图结构保持挂起状态。

    最初 decision 变量不会接收任何内容，因为 interrupt() 会在通过 Command(resume=value) 恢复图结构时才会返回值。此时 decision 会获得传入的 resume= 参数所指定的内容。

    有效载荷用于告知人工审核者当前正在查看的内容，因此它可以包含任何所需信息：草稿、摘要，或是带有上下文的结构化字典。调用方会在中断通知中收到该内容。

    使用 Command 恢复

    在中断之后继续执行意味着需要使用 Command(resume=value) 再次调用该图形处理程序：

    memory = MemorySaver()

graph = StateGraph(ReviewState)
graph.add_node("generate", generate)
graph.add_node("review", review)
graph.add_node("send", send)
graph.add_edge(START, "generate")
graph.add_edge("generate", "review")
graph.add_edge("review", "send")
graph.add_edge("send", END)
app = graph.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "email-review-1"}}

# Step 1: Invoke the graph. It will run generate, then pause at review.
result = app.invoke({"messages": ["Start email workflow"]}, config=config)
# result contains the interrupt payload with the draft

# Step 2: The human reviews the draft and approves.
result = app.invoke(Command(resume="approve"), config=config)
print(result["messages"][-1])
# "Email sent successfully."

    第一步中，LangGraph会执行generate操作，将草稿存储在状态中，然后进入review节点的interrupt()函数。它会保存当前状态并返回，而这次首次invoke()操作得到的result中会包含中断信息以及供审核者查看的草稿。第二步中，Command(resume="approve")会让LangGraph加载"email-review-1"的保存状态，从interrupt()调用处继续执行，并将"approve"作为返回值传回。此时review节点的decision值变为"approve"，系统会设置approved=True，随后执行流程将进入send步骤。

    两次调用必须使用相同的 config，因为 LangGraph 会通过 thread_id 来定位正确的检查点。如果第二次调用使用不同的 thread_id，则会指向不存在的检查点，从而导致调用失败。

    中断操作必须使用检查点。没有检查点的话，暂停状态就无处存放，因此此处 app.compile(checkpointer=memory) 并非可选项。

    从 LangGraph 1.1.0 版本开始，invoke() 函数支持传入 version="v2" 参数，并返回一个包含 .value 和 .interrupts 属性的 GraphOutput 对象，这种更安全的类型处理方式在熟悉基本用法后值得尝试。中断载荷可以是任何可序列化的值，无论是字符串、字典还是列表。在实际应用中，通常会传递一个结构化的字典，其中包含审核者所需的所有信息，如草稿内容、分类信息、发送者以及审核链接，应用层可将这些信息展示在用户界面中，或通过 Slack 或邮件转发出去。

    还有一个隐患。用完整的初始状态再次调用 invoke() 代替 Command(resume=value) 是一个错误：在相同的 thread_id 上执行 app.invoke({"messages": [...]}, config=config) 会重新启动该线程的图结构，并丢弃暂停状态。被中断的图结构应始终通过 Command(resume=value) 来继续处理。

    总结：电子邮件分类工作流

    当所有组件都准备就绪后，就可以将它们整合成一个类似生产环境的电子邮件分类系统。该系统能够读取收到的邮件，通过结构化输出对邮件意图进行分类，同时查询知识库和缺陷跟踪系统，起草回复内容，并在发送前暂停以供人工审核。

    之前的每个概念都对应着此工作流中的某个环节：

    • 带有还原器的TypedDict状态显示为EmailState，其search_results上应用了operator.add操作。
    • 串行边将read_email与classify_intent相连。
    • 并行边使得classify_intent可以分支为search_docs和bug_tracking两个路径。
    • 汇聚式同步意味着两个并行节点都会向write_response提供数据。
    • 条件路由方式为route_by_priority，用于在人工审核与自动发送之间做出选择。
    • Pydantic结构化输出为EmailClassification，用于约束大语言模型的回复内容。
    • 中断机制为human_review，用于暂停流程以等待人工确认。
    • 检查点机制为MemorySaver，它实现了中断与恢复的循环功能。

    状态与模式

    首先是需要结构化分类的状态以及 Pydantic 模式：

    from typing import Annotated
from typing_extensions import TypedDict
import operator
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
from langgraph.checkpoint.memory import MemorySaver

# Pydantic model for structured classification
class EmailClassification(BaseModel):
    intent: str = Field(description="The intent: 'bug_report', 'question', 'feedback', or 'other'")
    priority: str = Field(description="Priority level: 'high', 'medium', or 'low'")
    confidence: float = Field(description="Confidence score between 0.0 and 1.0")

# Graph state
class EmailState(TypedDict):
    email_subject: str
    email_body: str
    email_sender: str
    classification: dict
    search_results: Annotated[list[str], operator.add]
    bug_info: str
    draft_response: str
    approved: bool
    status: str
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

    EmailClassification 是一个 Pydantic 模型，不属于 EmailState，这体现了 TypedDict 与 Pydantic 的分工。它作为 JSON 规范的目标结构传递给大型语言模型，Pydantic 会验证返回结果，若不匹配则会立即抛出异常。EmailState 则是用于存储数据的容器，采用 TypedDict 形式，因为 LangGraph 需要还原器语法以及高效的序列化功能。

    classification 以 dict 的形式存储在 EmailState 中，因为在 Pydantic 验证完响应后，.model_dump() 会先将其转换为普通字典再存入状态中。这样一来，状态数据就能被任何检查点工具序列化，且在恢复时无需依赖 Pydantic。

    最重要的字段是 search_results: Annotated[list[str], operator.add]。由于 search_docs 和 bug_tracking 是并行运行的，且两者都能写入结果，因此还原器会将它们的输出合并在一起；而如果没有还原器，其中一个分支就会悄悄覆盖另一个分支的成果。

    构建节点

    节点保持小巧且功能单一。每个节点仅从状态中获取所需的值，完成相应操作后返回部分更新结果：

    def read_email(state: EmailState) -> dict:
    """Parse the incoming email into structured fields."""
    return {
        "email_subject": state["email_subject"],
        "email_body": state["email_body"],
        "email_sender": state["email_sender"],
    }


def classify_intent(state: EmailState) -> dict:
    """Use structured output to classify the email."""
    structured_llm = llm.with_structured_output(EmailClassification)
    classification = structured_llm.invoke(
        f"Classify this email.\\n\\n"
        f"Subject: {state['email_subject']}\\n"
        f"Body: {state['email_body']}"
    )
    return {"classification": classification.model_dump()}

def search_docs(state: EmailState) -> dict:
    """Search knowledge base for relevant documentation."""
    intent = state["classification"]["intent"]
    # In production, this would hit a real vector store or search API
    response = llm.invoke(
        f"Suggest 2 relevant documentation links for a {intent} "
        f"about: {state['email_subject']}"
    )
    return {"search_results": [f"[Docs] {response.content}"]}

def bug_tracking(state: EmailState) -> dict:
    """Look up related bugs in the tracking system."""
    # In production, this would query Jira, Linear, GitHub Issues, etc.
    response = llm.invoke(
        f"Summarize any known bugs related to: {state['email_subject']}"
    )
    return {"bug_info": response.content}

def write_response(state: EmailState) -> dict:
    """Draft an email response using all gathered context."""
    context_parts = [
        f"Original email from {state['email_sender']}:",
        f"Subject: {state['email_subject']}",
        f"Body: {state['email_body']}",
        f"\\nClassification: {state['classification']}",
        f"\\nRelevant docs: {chr(10).join(state['search_results'])}",
        f"\\nBug info: {state['bug_info']}",
    ]
    context = "\\n".join(context_parts)
    response = llm.invoke(
        f"Draft a helpful, professional reply to this customer email.\\n\\n

    乍看之下，read_email似乎毫无意义，因为它只返回状态中已存在的字段，但实际上它起到了明确的解析作用。在正式版本中，会直接处理原始的 webhook 数据并提取所需字段，因此这里的函数便承担了该提取任务。

    classify_intent正是 Pydantic 发挥作用的地方。在 structured_llm = llm.with_structured_output(EmailClassification) 中，LLM 被封装起来，要求其回复必须符合指定的结构规范。它会接收到从 EmailClassification 导出的 JSON 结构规范，并必须返回符合要求的有效 JSON；如果未能做到，LangChain 会自动重试，直到成功或抛出错误。.model_dump()则将经过验证的对象转换为普通的字典，以便存储在状态中。

    这两个并行节点分别向search_results和bug_info这两个字段写入数据，因此它们的LLM调用可以同时执行。对于search_results，由于operator.add需要列表作为输入，search_docs会返回一个列表（[f"[Docs] {response.content}"]）。而bug_info没有对应的归约函数，bug_tracking直接返回普通字符串，因此使用常规赋值即可。

    入度为1的节点是write_response。当它开始运行时，两个并行节点已经完成处理，其结果存储在状态中，于是它会利用这些数据构建提示语并拟写回复内容。

    human_review向interrupt()传递的是结构化的字典而非普通字符串，这属于良好的生产实践。该字典包含了草稿内容、分类信息以及给审核人员的提示，因此无论另一端是网页界面、Slack机器人还是邮件通知，都能获取完整的上下文信息。

    send_reply采用state.get("approved", True)作为默认值以确保安全性，即便状态中不存在approved字段，它也会发送消息。在风险等级要求更为谨慎的情况下，该默认值可被设置为False。

    在实际应用中，search_docs和bug_tracking会调用向量数据库、Jira或GitHub Issues等真实系统，而LangChain的@tool装饰器让这些调用的封装变得十分简单。节点结构保持不变，只是用工具调用替代了LLM调用。还有一个需要注意的问题：write_response会读取state[“search_results”]，如果该值仍是空列表，则访问可能会失败，为避免此问题，应在初始调用参数中将search_results: []设为初始值，这样reducer就能从该位置开始追加数据。

    连接节点间的边

    串行边、并行边和条件边用于连接这些节点：

    def route_by_priority(state: EmailState) -> str:
    """Route based on priority and confidence."""
    classification = state["classification"]
    if classification["priority"] == "high" or classification["confidence"] < 0.8:
        return "needs_review"
    return "auto_send"


graph = StateGraph(EmailState)

# Add all nodes
graph.add_node("read_email", read_email)
graph.add_node("classify_intent", classify_intent)
graph.add_node("search_docs", search_docs)
graph.add_node("bug_tracking", bug_tracking)
graph.add_node("write_response", write_response)
graph.add_node("human_review", human_review)
graph.add_node("send_reply", send_reply)

# Serial edges
graph.add_edge(START, "read_email")
graph.add_edge("read_email", "classify_intent")

# Parallel fan-out from classify_intent
graph.add_edge("classify_intent", "search_docs")
graph.add_edge("classify_intent", "bug_tracking")

# Parallel fan-in to write_response
graph.add_edge("search_docs", "write_response")
graph.add_edge("bug_tracking", "write_response")

# Conditional routing after write_response
graph.add_conditional_edges(
    "write_response",
    route_by_priority,
    {"needs_review": "human_review", "auto_send": "send_reply"}
)

# After human review, send the reply
graph.add_edge("human_review", "send_reply")
graph.add_edge("send_reply", END)

# Compile with memory for interrupt support
memory = MemorySaver()
app = graph.compile(checkpointer=memory)

    route_by_priority 会将高优先级的邮件发送给人工审核，而不考虑其置信度。置信度低于 0.8 的邮件也会被送去审核，因为如果模型不确定邮件的内容，就需要由人工来决定。其余邮件则会自动处理，这是合理的默认设置，可依据实际错误率进行调整。

    这种路由机制是通过 add_conditional_edges 而非 Command 来实现的，因为 route_by_priority 是一个纯函数：它仅检查状态并返回一个键值，不会创建新数据，而这正是 add_conditional_edges 的设计目的。

    以下是图中的所有连接：

    • 串行处理流程从 START 开始，经过 read_email，最终进入 classify_intent。
    • classify_intent 的输出会同时启动 search_docs 和 bug_tracking。
    • search_docs 和 bug_tracking 的结果会被传递给 write_response。
    • 在 write_response 之后，一条条件分支会选择 human_review 或 send_reply。
    • 审核完成后，human_review 的结果会传递给 send_reply，最终流程结束于 END。

    端到端运行工作流

    首先输入一封示例邮件，第一次调用会一直运行直到出现中断：

    config = {"configurable": {"thread_id": "email-001"}}

# Invoke with a sample email
result = app.invoke({
    "email_subject": "Login page broken after update",
    "email_body": (
        "Hi, since yesterday's update I can't log into my account. "
        "I get a 500 error every time. This is blocking my team. "
        "Please fix ASAP."
    ),
    "email_sender": "frustrated-customer@example.com",
    "search_results": [],
    "bug_info": "",
    "draft_response": "",
    "approved": False,
    "status": "",
}, config=config)

# The graph will classify this as high-priority, so it pauses at human_review.
# The interrupt payload contains the draft for review.
print("Interrupt payload received. Draft response ready for review.")

    到那时，处理流程已经完成了许多工作：

    1. read_email 已解析该邮件内容。
    2. classify_intent 使用结构化的 Pydantic 输出将其标记为高优先级的错误报告。
    3. 文档搜索与错误追踪查询在同一个超级步骤中并行执行。
    4. write_response 根据所有这些信息拟定了回复草稿。
    5. human_review 节点调用了 interrupt() 函数并暂停了流程。

    整个处理流程，包括两次并行调用的大型语言模型，都在暂停之前完成。当流程停止时，所有状态都会被完整保存，因此审核人员可以花几分钟甚至几小时来回复，而不会丢失任何信息。

    随后审核人员会予以批准：

    # Resume after human approval
final = app.invoke(Command(resume="approve"), config=config)
print(final["status"])
# "Reply sent to frustrated-customer@example.com"

    LangGraph加载了"email-001"的检查点，从之前的interrupt()调用处继续执行human_review，将"approve"作为返回值传回，同时在状态中设置approved=True，随后继续执行send_reply，该步骤返回了确认状态。

    所有早期的设计思路都体现在最终版本中：基于reducer的TypedDict状态管理、串行与并行边、条件路由、Pydantic结构化输出、通过中断实现人工审核，以及由检查点保存的记忆数据。

    对于置信度很高但优先级较低的邮件，route_by_priority会返回"auto_send"，从而跳过人工审核步骤。整个流程会同步进行：读取、分类、搜索、草拟和发送，只有在需要时才会出现中断。这就是条件路由的优势所在。

    最后需要提醒的是初始状态的问题。与Pydantic不同，TypedDict不会为缺失的字段设置默认值。如果write_response尝试访问state["search_results"]，而该键并不存在于初始状态中，就会引发KeyError异常。所有字段都应包含在初始状态字典中，即便它们初始值为空字符串、空列表或False。

    下一步学习方向

    此处采用的路径始于一个节点和一个TypedDict，最终形成一种生产级工作流，能够并行执行大语言模型调用、验证结构化输出、根据条件进行分支处理、通过检查点避免崩溃，并等待审核人员。从这里还有几条发展路径：

    • LangGraph Platform负责部署、托管式检查点管理以及水平扩展，当目标是处理真实流量时，可免除基础设施运维的负担。
    • LangSmith能够追踪并调试每一次图执行过程，当某个节点出现异常时还能显示每个超级步骤的精确状态。一旦图包含五个或更多带有条件路径的节点，这一点就显得尤为重要。
    • 子图可将较小的图组合成更大的多智能体系统。每个子图管理自己的状态，并通过父图进行通信，这是从单图工作流发展而来的自然步骤。
  • 通过 LangChain 的 @tool 和 ToolNode 实现工具调用，可使智能体在图结构内部使用外部 API、数据库和服务，而 search_docs 和 bug_tracking 这类核心节点正是尝试此功能的理想位置。
  • 为保证生产环境中的数据持久性，需安装 langgraph-checkpoint-postgres 以存储可在重启后依然保留的状态，而 API 交换操作仅需一行代码即可完成。

    • 简要回顾

    • 状态以 TypedDict 的形式存在。凡是类型定义为 Annotated[list, operator.add] 的字段会在多次写入时累积值，而没有定义合并函数的字段则会在每次写入时被替换。
    • 节点实际上就是一个函数：它接收完整状态并返回部分更新后的字典，由 LangGraph 负责合并这些更新。
    • 边决定了数据流的方向。在顺序重要的场景使用串行边，用于同时执行相互独立的任务时使用并行边，而在需要根据数据选择分支时则使用条件边。
    • 使用 TypedDict 来表示图的状态，并结合 Pydantic 的 .with_structured_output() 方法来限定大语言模型应返回的内容。
    • 检查点机制可实现数据持久化并支持中断操作：开发阶段可使用 MemorySaver，生产环境则可用 PostgresSaver。没有检查点机制就无法实现中断。
    • 中断实际上只是保存下来的暂停状态。interrupt(payload) 会暂停当前操作并创建检查点，而 Command(resume=value) 可以从该检查点精确位置继续执行。
    • 并行处理功能是内置的：从一个节点出发的两条边会同时向目标节点发送数据，而这些节点写入的每个字段都需要对应的归约器。

    需要思考的问题

    1. MemorySaver适用于开发环境，而PostgresSaver则适合生产环境。对于高吞吐量系统，还有哪些存储后端值得考虑？在做出选择时需要权衡哪些因素？
    2. route_by_priority会将高优先级的邮件发送给相关人员审核，其余邮件则自动发送。在真实系统中如何调整这一阈值？哪些指标可以表明阈值设置是否正确？
    3. 最终路由是在write_response之后通过add_conditional_edges实现的。在邮件处理流程中，是否有某种场景更适合使用Command模式而非单独的路由函数？

    相关阅读

  • 用文件夹和 Markdown 文件构建本地 LLM Agent 工作流 — 构建一个分阶段的本地 Agent 流水线，每个阶段对应一个文件夹，提示词为可编辑的 Markdown 格式，且在执行昂贵操作之前由人工审核输出结果。
  • 实用笔记：Python 中的文本转 SQL Agent：LLM 工具调用教程 — 针对采用该模式的团队提供的实用笔记操作指南，包括契约设计、校验机制以及可直接插入的代码片段。
  • 实用笔记：为何没人谈论 LangChain、LangGraph 和 AutoAgent — 该实用笔记的详细操作指南：为采用此类架构的团队提供合同模板、检查清单以及可直接使用的代码片段。
  • LangGraph 中的自愈式多智能体团队：架构师、测试员与调试器 — 使用 Python 构建 LangGraph 状态机，让架构师、开发人员、pytest 沙箱环境、调试器及审核员协同工作并循环迭代，直至测试通过。
  • 实用笔记：使用 A2A、LangGraph 构建双智能体代码审查循环 — 该实用笔记提供了详细操作指南，包括适用于采用此模式的团队的合同条款、检查项以及可直接插入的代码模板。
    • ← 全部文章发送简报