第 15 期 | 手动修图：通过 update_state 人工修正 Agent 幻觉

🎯 本期学习目标

各位架构师、同学们，欢迎回到《LangGraph 多智能体专家课》！上期我们深入探讨了如何让 Agent 们在复杂工作流中灵活决策。这期，我们不聊决策，我们聊“纠错”——而且是“人工纠错”。

想象一下，你的 Researcher Agent 辛辛苦苦查了一堆资料，结果它“幻觉”了，把一个关键数据搞错了。Writer Agent 傻傻地基于这个错误信息开始创作，最终产出了一个“事实性错误”的稿件。这简直是内容创作机构的噩梦！我们能怎么办？让 Editor Agent 发现？太晚了！我们得在错误蔓延之前，甚至是在错误发生之后，能有办法“倒流时光”，手动修正这个 Agent 的输出。

本期，我们将深入学习 LangGraph 中一个强大且非常实用的功能：update_state。它就像是你的内容机构里，那个手握“红笔”的终极审查员，能在任何时候介入，修正 Agent 的错误输出，确保整个流程沿着正确的轨道前进。

通过本期学习，你将能够：

1. 理解 update_state 的核心机制： 掌握如何在 LangGraph 中精准定位并修改特定线程的运行时状态。
2. 实现人工干预 Agent 工作流： 学会通过 update_state 模拟人工审查和修正 Agent 输出，有效应对“幻觉”问题。
3. 构建更健壮的 AI 应用： 为你的 AI Content Agency 引入一个关键的人机协作（Human-In-The-Loop, HITL）环节，提升内容质量和可靠性。
4. 掌握调试和回溯技巧： 将 update_state 作为一种强大的调试工具，在开发阶段快速修正 Agent 行为，加速迭代。

📖 原理解析

同学们，我们都知道 LangGraph 的核心是“状态”（State）。Agent 们在节点间传递的，就是这个不断演进的全局状态。通常，状态的修改是由节点执行后返回新的状态值来完成的。但今天，我们要介绍的 update_state 方法，它跳出了这个“节点执行 -> 返回新状态”的常规流程，允许我们从外部，以一种“上帝视角”，直接修改指定线程的当前状态。

这就像什么呢？你的 Researcher Agent 在研究某篇文章时，把“AI 市场规模是 1000 亿”写成了“1 万亿”。这个错误已经进入了你的 LangGraph 状态，并且即将传递给 Writer。如果按照正常流程，Writer 会基于“1 万亿”这个错误数据开始创作。而 update_state 的作用，就是让你能像一个超级编辑一样，在 Writer 还没动笔之前，或者在 Writer 已经写了一半，你发现问题之后，直接冲进状态库，把那个错误的 research_output 字段，从“1 万亿”手动改成“1000 亿”。然后，Writer Agent 就能基于这个被修正的正确数据继续工作了。

是不是很像在 Photoshop 里，发现照片某个地方没修好，直接打开图层，局部调整，而不必从头再拍一张？所以，我把它形象地称为“手动修图”。

update_state 的核心参数与机制

update_state 方法通常在 CompiledGraph 实例上调用，它需要以下几个关键信息：

1. thread_id (或 config): 这是告诉 LangGraph 你要修改哪个具体的对话或任务线程的状态。在 LangGraph 中，每个独立的 invoke 或 stream 调用都对应一个唯一的 thread_id。这个 thread_id 是 LangGraph 用来跟踪和持久化每个会话状态的关键。
2. state: 这是一个字典，包含了你希望更新的状态键值对。传入的字典会与目标线程的当前状态进行合并（通常是浅合并，即如果键已存在则覆盖，否则新增）。
3. as_node (可选，默认为 False): 这个参数比较高级。如果设为 True，LangGraph 会将这次状态更新视为一个“节点”的输出。这意味着这次更新可能会触发图中的条件边，导致图的进一步执行。但在我们今天“人工纠错”的场景下，我们通常只是想修改状态，然后可能手动决定下一步是重新执行某个节点，还是让流程继续，所以通常保持 False。

为什么 update_state 是必要的？

• 应对 Agent 幻觉（Hallucination）： 这是最直接的应用。LLM 并非完美，它们会犯错。当 Agent 基于 LLM 构建时，这些错误就会被带入工作流。update_state 提供了一个人工干预的“安全阀”。
• 人机协作（Human-In-The-Loop, HITL）： 在某些关键环节，例如内容审核、重要决策，需要人类专家进行最终确认或修正。update_state 是实现这种协作模式的基石。
• 调试与开发： 在开发复杂的 Agent 工作流时，Agent 行为不总是符合预期。通过 update_state，你可以快速修正中间状态，测试不同场景，而无需从头运行整个流程，大大加速开发效率。
• 迭代优化： 当你发现 Agent 的某个特定输出模式需要微调时，可以在不修改 Agent 本身代码的情况下，通过 update_state 进行临时修正，为后续的 Agent 优化提供缓冲。

Mermaid 图解：人工干预下的内容创作工作流

让我们通过一个 Mermaid 图来直观地理解 update_state 在我们的 AI Content Agency 项目中的作用。

graph TD
    A[用户输入: 内容创作请求] --> B(Planner Agent: 规划内容大纲)
    B --> C(Researcher Agent: 收集资料)
    C -- research_output --> D{人工审查点: 发现错误?}
    D -- 是 (幻觉!) --> E[人工修正: 调用 update_state]
    E -- 更新 state.research_output --> F(Writer Agent: 撰写初稿)
    D -- 否 (正确) --> F
    F --> G(Editor Agent: 润色审校)
    G --> H[输出: 最终内容]

    style A fill:#f9f,stroke:#333,stroke-width:2px
    style E fill:#ffc,stroke:#a00,stroke-width:2px,color:#a00
    style D fill:#fcf,stroke:#333,stroke-width:2px

图解说明：

1. 用户输入：一切的开始，一个内容创作请求。
2. Planner Agent：负责根据请求生成内容大纲。
3. Researcher Agent：根据大纲去收集资料，并将 research_output 写入状态。
4. 人工审查点 (发现错误?)：这是我们引入的关键环节。在这个点，我们可以选择性地暂停或检查 Researcher 的输出。
5. 是 (幻觉!)：如果发现 research_output 中存在幻觉或错误，我们将进入人工修正流程。
6. 人工修正 (调用 update_state)：在这个步骤，我们手动调用 graph.update_state()，传入当前线程的 thread_id 和一个包含修正后 research_output 的字典，直接覆盖掉状态中的错误信息。
7. Writer Agent：无论 Researcher 的输出是直接通过，还是经过人工修正，Writer 都会基于当前状态中最新的、正确的 research_output 来撰写初稿。
8. Editor Agent：对初稿进行润色和审校。
9. 输出：最终高质量的内容。

通过这个流程，update_state 就像一个强大的“修正带”，在内容生产链条的任何环节，都能精准地修正之前的错误，确保下游 Agent 接收到的是最准确的信息。

💻 实战代码演练 (Agency项目中的具体应用)

好了，理论讲得够多了，让我们把手弄脏，看看如何在我们的 AI Content Agency 项目中实际应用 update_state。

我们的场景是：Researcher Agent 在查询某个热门话题（比如“生成式 AI 市场规模”）时，不小心给出了一个错误的或过时的数据。我们作为“AI 内容机构的 CTO 兼总编”，要亲自介入，修正这个错误，然后让 Writer Agent 基于修正后的数据继续创作。

import operator
from typing import Annotated, TypedDict, List
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.sqlite import SqliteSaver
import os

# 确保设置了 OpenAI API 密钥
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY" # 请替换为你的实际密钥

# 1. 定义图的状态 (Graph State)
# -----------------------------------------------------------
class AgentState(TypedDict):
    """
    定义 LangGraph 的全局状态。
    这个状态将会在各个 Agent 节点之间传递和修改。
    """
    user_query: str  # 用户的原始内容创作请求
    research_output: Annotated[str, operator.add]  # 研究员的输出，可能包含多条信息，使用 operator.add 聚合
    writing_draft: Annotated[str, operator.add]  # 作家撰写的初稿，使用 operator.add 聚合
    final_content: str  # 最终产出的内容
    messages: Annotated[List[BaseMessage], operator.add] # 对话历史，方便 Agent 之间传递上下文

# 2. 定义 Agent 节点函数
# -----------------------------------------------------------

# 模拟一个 LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

def planner_node(state: AgentState) -> AgentState:
    """
    Planner Agent 节点：根据用户查询规划内容大纲。
    """
    print("\n--- Planner Agent 正在规划内容 ---")
    user_query = state["user_query"]
    messages = state.get("messages", [])
    
    # 构建规划提示
    prompt = f"""
    你是一个经验丰富的内容规划师。
    用户的请求是：'{user_query}'。
    请为这篇文章提供一个详细的内容大纲，包括主要章节和关键点。
    """
    
    # 调用 LLM 进行规划
    response = llm.invoke([HumanMessage(content=prompt)])
    plan = response.content
    
    print(f"规划结果:\n{plan}")
    
    # 更新状态
    return {
        "messages": [AIMessage(content=f"规划结果:\n{plan}")],
        "research_output": f"规划大纲：\n{plan}\n", # 将规划结果也放入research_output，方便后续Agent参考
        "writing_draft": "", # 清空写作草稿，为Writer准备
    }

def researcher_node(state: AgentState) -> AgentState:
    """
    Researcher Agent 节点：根据规划大纲收集资料，模拟幻觉。
    """
    print("\n--- Researcher Agent 正在收集资料 ---")
    current_messages = state["messages"]
    
    # 模拟研究过程，并故意引入一个“幻觉”数据
    # 假设用户查询是关于“生成式 AI 市场规模”
    if "生成式 AI 市场规模" in state["user_query"]:
        research_info = """
        根据最新研究，生成式 AI 市场规模预计将在 **2025 年达到 500 亿美元**。
        主要驱动因素包括：技术进步、企业数字化转型需求、以及新兴应用场景的爆发。
        （注意：此处故意设置了一个较低的、可能错误的或过时的数据，方便演示 update_state）
        """
        print("研究员故意犯了个错误：市场规模数据可能有误！")
    else:
        research_info = "这是关于其他主题的研究信息..."

    # 更新状态
    return {
        "messages": current_messages + [AIMessage(content=f"研究结果:\n{research_info}")],
        "research_output": research_info,
    }

def writer_node(state: AgentState) -> AgentState:
    """
    Writer Agent 节点：根据研究结果撰写初稿。
    """
    print("\n--- Writer Agent 正在撰写初稿 ---")
    user_query = state["user_query"]
    research_output = state["research_output"]
    current_messages = state["messages"]
    
    # 构建写作提示
    prompt = f"""
    你是一个专业的文章撰写者。
    用户请求：'{user_query}'
    研究员提供了以下资料：
    {research_output}
    
    请根据这些资料，撰写一篇关于 '{user_query}' 的文章初稿。
    务必引用研究员提供的数据和关键信息。
    """
    
    # 调用 LLM 进行写作
    response = llm.invoke([HumanMessage(content=prompt)])
    draft = response.content
    
    print(f"撰写初稿:\n{draft[:200]}...") # 打印部分内容
    
    # 更新状态
    return {
        "messages": current_messages + [AIMessage(content=f"撰写初稿:\n{draft}")],
        "writing_draft": draft,
    }

def editor_node(state: AgentState) -> AgentState:
    """
    Editor Agent 节点：对初稿进行润色和审校。
    """
    print("\n--- Editor Agent 正在审校和润色 ---")
    user_query = state["user_query"]
    writing_draft = state["writing_draft"]
    current_messages = state["messages"]
    
    # 构建编辑提示
    prompt = f"""
    你是一个严谨的内容编辑。
    这是关于 '{user_query}' 的文章初稿：
    {writing_draft}
    
    请对这篇文章进行以下修改：
    1. 修正语法错误和拼写错误。
    2. 优化句式，使其更流畅和专业。
    3. 确保内容逻辑清晰，论点有力。
    4. 检查事实准确性（虽然我们这里主要模拟润色，但实际中编辑会进行事实核查）。
    
    请返回最终润色后的文章。
    """
    
    # 调用 LLM 进行编辑
    response = llm.invoke([HumanMessage(content=prompt)])
    final_content = response.content
    
    print(f"最终内容:\n{final_content[:200]}...") # 打印部分内容
    
    # 更新状态
    return {
        "messages": current_messages + [AIMessage(content=f"最终内容:\n{final_content}")],
        "final_content": final_content,
    }

# 3. 构建 LangGraph 工作流
# -----------------------------------------------------------
def build_graph():
    workflow = StateGraph(AgentState)

    # 添加节点
    workflow.add_node("planner", planner_node)
    workflow.add_node("researcher", researcher_node)
    workflow.add_node("writer", writer_node)
    workflow.add_node("editor", editor_node)

    # 设置边
    workflow.add_edge(START, "planner")
    workflow.add_edge("planner", "researcher")
    workflow.add_edge("researcher", "writer")
    workflow.add_edge("writer", "editor")
    workflow.add_edge("editor", END)
    
    # 使用 SqliteSaver 持久化状态，方便我们获取 thread_id 和修改状态
    memory = SqliteSaver.from_conn_string(":memory:")
    
    # 编译图
    app = workflow.compile(checkpointer=memory)
    return app

# 4. 模拟运行与人工修正
# -----------------------------------------------------------
if __name__ == "__main__":
    app = build_graph()
    
    user_input = "请撰写一篇关于生成式 AI 市场规模及其未来趋势的文章。"
    
    print("--- 第一次运行：Researcher Agent 犯错 ---")
    
    # 运行图，并获取 thread_id
    # config 是一个字典，可以包含 thread_id 和 thread_ts (时间戳)
    # 我们这里只传入 thread_id，LangGraph 会自动生成一个 UUID 作为 thread_id
    # 或者我们可以显式指定一个
    config = {"configurable": {"thread_id": "gen_ai_market_report_1"}}
    
    # 第一次运行，让 Researcher Agent 犯错
    initial_output = app.invoke(
        {"user_query": user_input, "messages": [HumanMessage(content=user_input)]},
        config=config
    )
    
    print("\n--- 第一次运行结果（Researcher 犯错版）---")
    print(f"研究输出: {initial_output['research_output']}")
    print(f"写作草稿（基于错误数据）:\n{initial_output['writing_draft'][:200]}...")
    print(f"最终内容（基于错误数据）:\n{initial_output['final_content'][:200]}...")
    
    # 假设我们人工审查发现了 Researcher 的错误
    print("\n--- 人工审查发现 Researcher 幻觉：市场规模数据错误！ ---")
    wrong_data = "2025 年达到 500 亿美元"
    correct_data = "2030 年预计将达到 1.1 万亿美元（根据 Grand View Research 最新报告）"
    
    print(f"原始错误数据: '{wrong_data}'")
    print(f"修正为正确数据: '{correct_data}'")
    
    # 关键步骤：使用 update_state 修正状态
    # 我们直接修改 'research_output' 字段
    # 注意：这里我们只更新了 'research_output'，但也可以更新其他字段，甚至添加新字段
    app.update_state(
        config, # 使用之前运行的 thread_id
        {"research_output": f"规划大纲：\n{initial_output['research_output']}\n修正后的研究结果：\n{correct_data}\n"}
        # 这里为了演示方便，我们直接覆盖了research_output，实际中可能需要更精细的合并逻辑
        # 或者在 Researcher 节点返回时就将规划和研究结果分开存储
    )
    
    print("\n--- 状态已人工修正。现在重新运行 Writer 和 Editor Agent ---")
    
    # 重新从 Writer 节点开始运行（或者从 Researcher 之后继续，取决于你想回滚到哪一步）
    # 为了演示 update_state 的效果，我们让图从 Writer 节点继续执行
    # LangGraph 会从上次停止的地方继续，但由于我们修改了状态，Writer 将会看到新的状态
    
    # 这里我们模拟的是，Writer 已经跑过了，但我们修改了 Researcher 的输出，
    # 希望 Writer 基于新数据重新生成。所以我们再次调用 invoke，LangGraph 会从最新状态继续。
    # 实际上，如果想确保 Writer 重新执行，可能需要更复杂的控制流，
    # 比如在 Editor 发现错误后，将状态引导回 Researcher 或 Writer 节点。
    # 但对于 update_state 的直接效果演示，我们只需要修改状态，然后让后续节点继续即可。
    
    # 为了更清晰地演示，我们模拟“回滚”到 writer 节点之前的状态（即修改了 research_output 之后）
    # 实际上，`invoke` 总是从当前最新状态开始执行，直到 END
    # 如果要精确地从某个节点开始，需要用到 `stream` 或更底层的控制
    # 在这里，我们修改了状态，再次 invoke 会让所有后续节点基于新状态执行。
    
    # 重新获取状态，确认 state.research_output 已经被更新
    current_state_after_update = app.get_state(config).values
    print(f"\n修正后的研究输出 (从状态中读取): {current_state_after_update['research_output']}")
    
    # 再次运行整个图，Writer 和 Editor 会使用修正后的数据
    # 注意：这里我们没有显式地“回退”到某个节点，而是让图从当前状态继续运行
    # 由于我们修改了 research_output，Writer 会在下一次执行时看到这个新值。
    # LangGraph 的 invoke 默认会从上次执行结束的地方（或第一个未完成的节点）继续，
    # 但如果整个图已经执行到 END，再次 invoke 可能会重新开始。
    # 为了演示 `update_state` 影响后续节点，我们直接让它继续。
    
    # 更严谨的做法是：
    # 1. 在 Researcher 之后引入一个条件边，判断是否需要人工审查。
    # 2. 如果需要，进入一个“人工修正”节点，该节点内部调用 update_state。
    # 3. 修正完毕后，再引导回 Writer 节点。
    # 但为了本期 `update_state` 的核心演示，我们直接在外部调用。
    
    # 简单粗暴地再次调用 invoke，让 Writer 和 Editor 基于新状态重新跑一遍
    # 实际上，如果图已经结束，invoke 会启动一个新的执行，但由于 thread_id 相同，
    # 它会加载旧状态并在此基础上继续。
    # 这也意味着 Writer 和 Editor 会重新计算。
    print("\n--- 第二次运行：基于修正后的 Researcher 输出 ---")
    final_output_corrected = app.invoke(
        {"user_query": user_input, "messages": [HumanMessage(content=user_input)]},
        config=config
    )
    
    print("\n--- 第二次运行结果（修正后版）---")
    print(f"研究输出: {final_output_corrected['research_output']}")
    print(f"写作草稿（基于正确数据）:\n{final_output_corrected['writing_draft'][:200]}...")
    print(f"最终内容（基于正确数据）:\n{final_output_corrected['final_content'][:200]}...")
    
    # 验证是否使用了修正后的数据
    assert correct_data in final_output_corrected['research_output']
    assert wrong_data not in final_output_corrected['writing_draft'] # 确保 Writer 没有使用旧数据
    assert correct_data in final_output_corrected['writing_draft'] # 确保 Writer 使用了新数据
    
    print("\n--- 验证成功：Writer 和 Editor Agent 已经基于修正后的数据继续工作！ ---")

代码解析：

1. AgentState 定义： 我们定义了一个 AgentState 来承载整个工作流中的数据，包括 user_query, research_output, writing_draft, final_content 和 messages。Annotated 结合 operator.add 确保了某些字段（如 research_output 和 writing_draft）可以被追加内容。
2. Agent 节点函数：
   • planner_node：负责生成内容大纲。
   • researcher_node：这是我们故意引入“幻觉”的地方。当用户查询涉及“生成式 AI 市场规模”时，它会返回一个过时或错误的数据。
   • writer_node：基于 research_output 撰写初稿。
   • editor_node：对初稿进行润色。
3. 构建 LangGraph： 我们像往常一样构建了 StateGraph，定义了节点和边。
4. SqliteSaver： 引入 SqliteSaver 是为了能够持久化每个线程的状态，这样我们就可以通过 thread_id 准确地获取和修改状态。
5. 模拟运行：
   • 第一次运行： 我们首先正常运行一次图。researcher_node 会故意输出错误的市场规模数据。你会看到 writer_node 和 editor_node 都基于这个错误数据进行了创作。
   • 人工修正： 这是核心！我们模拟人工审查发现 research_output 中的错误数据。然后，我们调用 app.update_state()。
     • config 参数指定了我们希望修改哪个 thread_id 的状态。
     • 第二个参数是一个字典 {"research_output": "..."}，它告诉 LangGraph，请将当前线程的 research_output 字段更新为我们提供的新值。
   • 第二次运行： 状态被修正后，我们再次调用 app.invoke()。由于 thread_id 相同，LangGraph 会加载我们刚刚修正过的状态。此时，当 writer_node 再次执行时（或者说，如果它还没有执行，它将看到新的状态），它将看到的是被我们人工修正后的 research_output，从而生成正确的文章。

通过这个实战演练，你清晰地看到了 update_state 如何在 LangGraph 的执行过程中，以一种“插队”的方式，改变了数据流向，纠正了 Agent 的错误。

坑与避坑指南

update_state 功能强大，但如果使用不当，也可能挖坑。作为你的高级导师，我必须得给你打打预防针。

1. 线程 ID 混淆：

   • 坑： update_state 必须指定正确的 thread_id。如果你在开发过程中，不小心修改了错误的 thread_id，那么你修改的将是另一个会话的状态，导致难以追踪的问题。
   • 避坑： 始终明确你正在操作的是哪个 thread_id。在实际应用中，thread_id 通常会与用户会话 ID 绑定。在调试时，可以打印 config 中的 thread_id 来确认。
   • 进阶： 对于 stream 模式，每个 chunk 都会返回 config，其中包含当前的 thread_id。

2. 状态覆盖与合并的理解：

   • 坑： update_state 传入的字典会与现有状态进行合并。如果你传入的键是 LangGraph 状态中已经存在的，那么新值会覆盖旧值。如果你期望的是追加而不是覆盖（例如 research_output 字段可能需要追加多条研究结果），但却直接覆盖了，那么你可能会丢失历史信息。
   • 避坑： 仔细理解你的 AgentState 中每个字段的 TypedDict 定义和 Annotated 的聚合操作（如 operator.add）。如果需要追加，确保你传入 update_state 的值是与现有值合并后的结果，或者你的状态定义本身就是支持追加的。在我们的例子中，research_output 是 Annotated[str, operator.add]，所以每次节点返回都会追加。但 update_state 默认是覆盖，所以你需要手动获取旧值并合并。

     # 错误示范：直接覆盖，可能丢失 Planner 的规划结果
     # app.update_state(config, {"research_output": correct_data})
     
     # 正确示范：获取旧状态，然后合并新修正的数据
     current_state = app.get_state(config).values
     updated_research_output = current_state.get("research_output", "") + f"\n[人工修正]: {correct_data}\n"
     app.update_state(config, {"research_output": updated_research_output})
     

     在我们的示例代码中，为了简化，我直接演示了覆盖，但注释中也提醒了实际应用中可能需要更精细的合并逻辑。

3. as_node=True 的副作用：

   • 坑： 当 as_node=True 时，update_state 会被视为