LangGraph 实战：构建一个研究报告生成 Agent

前几篇文章分别讲了状态机、条件分支、人机协作、子图、持久化、流式输出、Multi-Agent 和错误处理。本篇把这些知识综合起来，构建一个完整的实战项目：研究报告生成 Agent。

项目需求：用户输入一个研究主题，Agent 自动完成搜索资料、生成大纲、逐节写作、质量审核，最终输出结构化报告。大纲生成后，需要用户确认再继续（Human-in-the-loop）。

1.1 项目架构设计

研究报告生成 Agent 工作流——规划子问题、Send API 并行搜索、汇总、撰写、审阅的完整循环

1.1.1 State 设计

State 是整个项目的核心数据结构，需要提前规划清楚每个字段的职责：

python

# research_agent/state.py

from typing import TypedDict, List, Optional, Annotated
import operator


class ReportState(TypedDict):
    # 用户输入
    topic: str                              # 研究主题

    # 搜索阶段
    search_results: List[str]               # 搜索到的资料列表

    # 规划阶段
    outline: List[str]                      # 报告大纲（章节列表）
    outline_approved: bool                  # 用户是否确认了大纲

    # 写作阶段
    sections: Annotated[List[str], operator.add]   # 各章节内容（追加模式）
    current_section_index: int              # 当前正在写第几节

    # 审核阶段
    review_feedback: str                    # 审核意见
    revision_count: int                     # 修改次数（防止无限循环）

    # 最终输出
    final_report: str                       # 完整报告

    # 错误处理
    error_message: Optional[str]
    error_history: Annotated[List[str], operator.add]

字段设计要点：

sections 使用 Annotated + operator.add，每次写完一节就追加，避免覆盖
outline_approved 是 Human-in-the-loop 的标志位，False 时图会在大纲节点暂停等待用户
revision_count 限制审核-修改循环的次数

1.1.2 节点设计

1.2 完整代码

python

# research_agent.py
# 完整的研究报告生成 Agent
# 功能：搜索 → 规划大纲 → 人工确认 → 分节写作 → 质量审核 → 输出报告

import operator
from typing import TypedDict, List, Optional, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver


# ============================================================
# State 定义
# ============================================================

class ReportState(TypedDict):
    topic: str
    search_results: List[str]
    outline: List[str]
    outline_approved: bool
    sections: Annotated[List[str], operator.add]
    current_section_index: int
    review_feedback: str
    revision_count: int
    final_report: str
    error_message: Optional[str]
    error_history: Annotated[List[str], operator.add]


# ============================================================
# 节点一：search_node（信息搜索）
# ============================================================

def search_node(state: ReportState) -> dict:
    """
    搜索节点：收集研究主题相关的背景资料。
    实际场景中替换为真实搜索 API（Tavily、SerpAPI 等）。
    这里用模拟数据演示完整流程。
    """
    topic = state["topic"]

    # 模拟搜索结果，实际替换为：
    # from tavily import TavilyClient
    # client = TavilyClient(api_key="YOUR_KEY")
    # results = client.search(query=topic, max_results=5)
    search_results = [
        f"【定义与背景】{topic} 是指...（来源：维基百科）",
        f"【技术原理】{topic} 的核心机制包括：1. 数据处理层 2. 推理层 3. 输出层（来源：arXiv 2025）",
        f"【应用场景】{topic} 在以下领域有广泛应用：金融、医疗、教育（来源：行业报告）",
        f"【挑战与限制】{topic} 面临的主要挑战：可解释性差、数据需求大、计算成本高（来源：学术论文）",
        f"【最新进展】2026 年，{topic} 领域取得的最新突破...（来源：技术博客）",
    ]

    return {
        "search_results": search_results,
        "error_history": [f"[search] 搜索完成，获取 {len(search_results)} 条资料"],
    }


# ============================================================
# 节点二：plan_node（规划大纲）
# ============================================================

def plan_node(state: ReportState) -> dict:
    """
    规划节点：基于搜索结果生成报告大纲。
    实际场景中用 LLM 生成大纲，让 LLM 读取 search_results 后规划章节结构。
    """
    topic = state["topic"]
    search_results = state.get("search_results", [])

    # 模拟基于搜索结果生成大纲（实际替换为 LLM 调用）
    # from langchain_openai import ChatOpenAI
    # llm = ChatOpenAI(model="gpt-4o")
    # prompt = f"基于以下资料，为「{topic}」生成一个5节的报告大纲...\n{search_results}"
    # outline = llm.invoke(prompt).content.split("\n")

    outline = [
        f"1. {topic} 概述与背景",
        f"2. {topic} 的核心技术原理",
        f"3. {topic} 的主要应用场景",
        f"4. {topic} 面临的挑战与局限",
        f"5. {topic} 的未来发展趋势",
    ]

    return {
        "outline": outline,
        "outline_approved": False,      # 重置为未批准，等待用户确认
        "current_section_index": 0,     # 重置写作进度
        "sections": [],                 # sections 是 operator.add，这里传空列表不会有问题
        # 注意：这里不能传 sections: [] 来"清空"，Annotated 字段会追加
        # 如果需要重置 sections，应该在 State 中设计 sections_draft 等中间字段
        "error_history": [f"[plan] 大纲生成完成，共 {len(outline)} 节，等待用户确认"],
    }


# ============================================================
# 节点三：write_node（分节写作）
# ============================================================

def write_node(state: ReportState) -> dict:
    """
    写作节点：写当前章节的内容。
    每次调用只写一节，通过 current_section_index 追踪进度。
    写完后 index 递增，条件边判断是否还有未写的章节。
    """
    topic = state["topic"]
    outline = state["outline"]
    search_results = state.get("search_results", [])
    current_idx = state.get("current_section_index", 0)
    revision_count = state.get("revision_count", 0)

    if current_idx >= len(outline):
        # 防御：不应该走到这里，但加上保护更稳健
        return {"error_history": ["[write] 警告：写作节点被调用但所有章节已完成"]}

    current_section_title = outline[current_idx]

    # 模拟写作（实际替换为 LLM 调用，传入章节标题和相关搜索结果）
    section_content = f"""## {current_section_title}

{topic} 在本章节涵盖的核心内容如下：

基于收集的研究资料，{current_section_title.split('. ', 1)[-1]} 方面的分析显示：
相关文献表明，该领域具有重要的研究价值和实践意义。
{'（修订版：已根据审核意见补充了更多细节）' if revision_count > 0 else ''}

**关键要点：**
- 要点一：理论基础与研究现状
- 要点二：实践应用与案例分析
- 要点三：未来方向与开放问题

"""

    return {
        "sections": [section_content],          # Annotated 列表，自动追加到现有内容
        "current_section_index": current_idx + 1,  # 推进到下一节
        "error_history": [f"[write] 完成第 {current_idx + 1}/{len(outline)} 节：{current_section_title}"],
    }


# ============================================================
# 节点四：review_node（质量审核）
# ============================================================

def review_node(state: ReportState) -> dict:
    """
    审核节点：检查报告质量。
    实际场景可以用 LLM 扮演审稿人，检查：
    - 内容完整性
    - 章节间逻辑连贯性
    - 事实准确性
    - 格式规范性
    """
    sections = state.get("sections", [])
    revision_count = state.get("revision_count", 0)

    # 模拟审核逻辑：第一版总是要求修改，修改一次后通过
    # 实际替换为 LLM 审稿调用
    if revision_count == 0 and len(sections) > 0:
        feedback = (
            "报告初稿基本合格，但存在以下问题需要修改：\n"
            "1. 各章节内容较为简略，建议每节增加具体案例\n"
            "2. 第3节的应用场景缺乏数据支撑\n"
            "请修订后重新提交。"
        )
        approved = False
    else:
        feedback = "报告质量良好，内容完整，逻辑清晰，符合发布标准。"
        approved = True

    return {
        "review_feedback": feedback,
        "revision_count": revision_count + 1,  # 每次审核都递增，包括通过的那次
        "error_history": [f"[review] 审核{'通过' if approved else '不通过'}，第 {revision_count + 1} 次审核"],
        # 把审核结果存入临时字段供路由函数读取
        "_review_approved": approved,   # 注意：这个字段需要在 State 中声明
    }


# ============================================================
# 节点五：finalize_node（最终整合）
# ============================================================

def finalize_node(state: ReportState) -> dict:
    """
    最终整合节点：把所有章节拼成完整报告，添加封面和结语。
    """
    topic = state["topic"]
    outline = state["outline"]
    sections = state.get("sections", [])
    revision_count = state.get("revision_count", 0)

    # 构建完整报告
    outline_text = "\n".join(f"  {item}" for item in outline)

    # sections 列表里包含了所有章节内容（可能有重复，因为 revision 导致重写）
    # 取最后 len(outline) 个，即最新版本的章节
    latest_sections = sections[-len(outline):] if len(sections) >= len(outline) else sections
    sections_text = "\n".join(latest_sections)

    final_report = f"""# {topic} 研究报告

**生成时间：** 自动生成
**修订次数：** {revision_count - 1} 次
**报告章节：** {len(outline)} 节

---

## 目录

{outline_text}

---

{sections_text}

---

## 参考资料

本报告基于以下 {len(state.get('search_results', []))} 条资料综合整理而成。

*本报告由 LangGraph 研究报告 Agent 自动生成。*
"""

    return {
        "final_report": final_report,
        "error_history": [f"[finalize] 报告整合完成，共 {len(final_report)} 字符"],
    }


# ============================================================
# 路由函数
# ============================================================

def route_after_outline(state: ReportState) -> Literal["human_feedback", "write"]:
    """
    大纲生成后的路由：
    - 未批准 → 去 human_feedback 等待用户输入
    - 已批准 → 直接开始写作
    """
    if state.get("outline_approved"):
        return "write"
    return "human_feedback"


def route_after_write(state: ReportState) -> Literal["write", "review"]:
    """
    写作节点完成后的路由：
    - 还有章节未写 → 继续写
    - 所有章节写完 → 进入审核
    """
    current_idx = state.get("current_section_index", 0)
    outline = state.get("outline", [])

    if current_idx < len(outline):
        return "write"   # 还有章节，继续写
    return "review"      # 全部写完，进入审核


def route_after_review(state: ReportState) -> Literal["write", "finalize"]:
    """
    审核后的路由：
    - 不通过且修改次数未超限 → 重新写作
    - 通过或修改次数超限 → 最终整合
    """
    revision_count = state.get("revision_count", 0)
    MAX_REVISIONS = 3  # 最多修改 3 次，防止无限循环

    # 简化判断：第一次审核（revision_count == 1）要求修改，之后通过
    # 实际场景读取 review_feedback 中的通过/不通过标志
    if revision_count <= 1 and revision_count < MAX_REVISIONS:
        # 重置写作进度，重新写所有章节
        return "write"
    return "finalize"


# ============================================================
# 构建图
# ============================================================

def build_report_agent(use_checkpointer: bool = True):
    """
    构建研究报告 Agent。
    use_checkpointer=True 时启用持久化，支持断点续跑和 Human-in-the-loop。
    """
    graph = StateGraph(ReportState)

    # 注册节点
    graph.add_node("search", search_node)
    graph.add_node("plan", plan_node)
    graph.add_node("write", write_node)
    graph.add_node("review", review_node)
    graph.add_node("finalize", finalize_node)

    # Human-in-the-loop 节点（空节点，图在这里暂停等待外部输入）
    # 用户通过 update_state 修改 outline_approved=True 来继续执行
    graph.add_node("human_feedback", lambda state: {})

    # 添加边
    graph.add_edge(START, "search")
    graph.add_edge("search", "plan")

    # 大纲生成后：条件路由（等待用户确认 或 直接写作）
    graph.add_conditional_edges(
        "plan",
        route_after_outline,
        {"human_feedback": "human_feedback", "write": "write"}
    )

    # human_feedback 节点完成后回到 plan（用户可能要求修改大纲）
    # 也可以路由到 write（如果用户批准了）
    # 这里简化：human_feedback 后直接去 write（用户通过 update_state 批准大纲）
    graph.add_conditional_edges(
        "human_feedback",
        route_after_outline,  # 再次检查 outline_approved 状态
        {"human_feedback": "human_feedback", "write": "write"}
    )

    # 写作节点：完成一节后条件路由（继续写 或 进入审核）
    graph.add_conditional_edges(
        "write",
        route_after_write,
        {"write": "write", "review": "review"}
    )

    # 审核节点：通过则整合，不通过则重写
    graph.add_conditional_edges(
        "review",
        route_after_review,
        {"write": "write", "finalize": "finalize"}
    )

    graph.add_edge("finalize", END)

    # 配置 checkpointer 支持持久化和断点
    if use_checkpointer:
        memory = MemorySaver()
        return graph.compile(
            checkpointer=memory,
            # 在 human_feedback 节点前暂停，等待用户输入
            interrupt_before=["human_feedback"]
        )
    else:
        return graph.compile()


# ============================================================
# 运行示例一：无人工介入的自动模式
# ============================================================

def run_automatic_mode():
    """自动模式：跳过大纲确认，直接运行完整流程。"""
    print("\n" + "=" * 60)
    print("模式：自动运行（无人工确认）")
    print("=" * 60)

    # 不使用 checkpointer，自动跳过 human-in-the-loop
    app = build_report_agent(use_checkpointer=False)

    initial_state = {
        "topic": "大型语言模型的 Agent 框架",
        "search_results": [],
        "outline": [],
        "outline_approved": True,   # 预先批准，跳过人工确认
        "sections": [],
        "current_section_index": 0,
        "review_feedback": "",
        "revision_count": 0,
        "final_report": "",
        "error_message": None,
        "error_history": [],
    }

    # 流式观察每个节点的执行
    print("\n执行过程：")
    for event in app.stream(initial_state, stream_mode="updates"):
        for node_name, updates in event.items():
            if node_name.startswith("__"):
                continue
            logs = updates.get("error_history", [])
            for log in logs:
                print(f"  {log}")

    final_state = app.invoke(initial_state)
    print("\n最终报告（前 500 字）：")
    print(final_state["final_report"][:500])
    print("...")


# ============================================================
# 运行示例二：Human-in-the-loop 模式
# ============================================================

def run_human_in_the_loop_mode():
    """
    Human-in-the-loop 模式：大纲生成后暂停，等待用户确认。
    演示完整的中断-恢复交互流程。
    """
    print("\n" + "=" * 60)
    print("模式：Human-in-the-loop（大纲需要人工确认）")
    print("=" * 60)

    app = build_report_agent(use_checkpointer=True)

    # 每次运行需要一个唯一的 thread_id，用于恢复状态
    thread_config = {"configurable": {"thread_id": "report-session-001"}}

    initial_state = {
        "topic": "强化学习在机器人控制中的应用",
        "search_results": [],
        "outline": [],
        "outline_approved": False,  # 需要人工确认
        "sections": [],
        "current_section_index": 0,
        "review_feedback": "",
        "revision_count": 0,
        "final_report": "",
        "error_message": None,
        "error_history": [],
    }

    # 第一次运行：会在 human_feedback 节点前暂停
    print("\n[第一阶段] 运行至大纲生成，等待用户确认...")
    for event in app.stream(initial_state, config=thread_config, stream_mode="updates"):
        for node_name, updates in event.items():
            if node_name.startswith("__"):
                continue
            logs = updates.get("error_history", [])
            for log in logs:
                print(f"  {log}")

    # 检查当前状态（图已暂停）
    current_state = app.get_state(thread_config)
    outline = current_state.values.get("outline", [])

    print("\n[等待用户输入] 生成的大纲如下：")
    for item in outline:
        print(f"  {item}")

    # 模拟用户确认（实际场景通过 API 接收用户输入）
    user_approved = True  # 模拟用户批准
    print(f"\n[用户操作] 用户{'批准' if user_approved else '拒绝'}了大纲")

    if user_approved:
        # 通过 update_state 修改 State，然后继续执行
        app.update_state(
            thread_config,
            {"outline_approved": True},
            as_node="human_feedback"   # 以 human_feedback 节点的身份更新
        )

        # 第二次运行：从暂停处继续（注意：不传 initial_state，传 None）
        print("\n[第二阶段] 用户批准大纲，继续执行写作流程...")
        for event in app.stream(None, config=thread_config, stream_mode="updates"):
            for node_name, updates in event.items():
                if node_name.startswith("__"):
                    continue
                logs = updates.get("error_history", [])
                for log in logs:
                    print(f"  {log}")

        # 获取最终结果
        final_state = app.get_state(thread_config)
        report = final_state.values.get("final_report", "")
        if report:
            print("\n报告生成完成！")
            print(f"报告长度：{len(report)} 字符")
            print("\n报告预览（前 300 字）：")
            print(report[:300])
            print("...")
        else:
            print("\n报告未能生成，请检查执行日志。")


# ============================================================
# 入口
# ============================================================

if __name__ == "__main__":
    # 演示自动模式（不需要持久化配置）
    run_automatic_mode()

    # 演示人机协作模式
    # run_human_in_the_loop_mode()

1.5 关键设计决策解析

1.5.1 为什么 sections 用追加模式

sections: Annotated[List[str], operator.add] 的追加语义，是写作节点每次调用只写一节设计的基础。如果用普通列表字段，每个节点更新会覆盖前一节的内容。追加模式保证每节写完后自动累积。

审核-修改循环时，sections 会累积两版内容（原版+修订版）。finalize_node 用 sections[-len(outline):] 取最新版本：

python

latest_sections = sections[-len(outline):] if len(sections) >= len(outline) else sections

1.5.2 interrupt_before 的位置

interrupt_before=["human_feedback"] 让图在进入 human_feedback 节点之前暂停。用户调用 update_state 更新 State 后，再次调用 stream(None, config=...) 从暂停处恢复。注意第二次调用传的是 None 而不是原始 initial_state。

1.5.3 revision_count 防止无限循环

review_node 每次执行都递增 revision_count，route_after_review 用它判断是否达到最大修改次数。这是 Multi-Agent 系统中防止审核-修改死循环的标准做法。

1.6 节点间的数据流

1.7 小结

这个研究报告生成 Agent 综合了 LangGraph 的核心特性：

State 设计：区分追加字段（sections、error_history）和覆盖字段（current_section_index、outline_approved）
条件边路由：基于 State 字段的值做路由，而非硬编码顺序
Human-in-the-loop：interrupt_before + update_state + 恢复执行
循环控制：revision_count 防止审核-修改无限循环
错误记录：error_history 追踪完整执行路径

从这个项目继续扩展有两个方向：接入真实 LLM 和搜索 API 替换模拟数据，或者增加更多专业 Agent（事实核查、引用格式化）——后者就是 Multi-Agent 架构的延伸。