摘要：传统的 RAG 只能回答简单问题，想要实现像 OpenAI O1 那样具备“深度思考”能力的 Agent 怎么办？本文将带你深入源码，使用 LangGraph + Python 构建一个生产级的 Deep Research Agent。它是目前市面上最接近人类研究员工作流的开源实现。
关键词：Python, LangChain, LangGraph, AI Agent, Deep Research, 源码分析

---

1. 什么是 Deep Research Agent？

Deep Research Agent 是一个模仿人类研究员工作流程的 AI 智能体。与普通 RAG 不同，它不会只搜一次就瞎编，而是会：

1. 自主规划 (Plan)：把大问题拆解为多个子查询。
2. 并行搜索 (Parallel Search)：同时抓取多个网页的信息。
3. 自我反思 (Reflect)：阅读搜索结果，发现资料不够时自动补充搜索。
4. 最终汇总 (Report)：生成长篇深度报告。

本文复刻的完整源码已开源：
👉 GitHub 地址：https://github.com/changflow/deep-research-agent

---

2. 核心代码实现（手把手）

2.1 定义状态（State）

在 LangGraph 中，一切从 State 开始。我们需要定义 Agent 在思考过程中存储的数据结构：

# src/deep_research_agent/core/state.py

from typing import TypedDict, List, Annotated
import operator

class AgentState(TypedDict):
    topic: str                    # 用户输入的原始话题
    plan: List[str]               # 拆解后的研究计划
    search_results: Annotated[List[dict], operator.add] # 搜集到的所有资料（增量更新）
    report: str                   # 最终生成的报告
    iterations: int               # 当前迭代次数（防止死循环）

2.2 构建思考图（Graph Construction）

这是整个 Agent 的“大脑回路”。我们使用 StateGraph 将各个节点串联起来：

# src/deep_research_agent/graph.py

from langgraph.graph import StateGraph, END

def build_graph():
    workflow = StateGraph(AgentState)

    # 1.以此添加节点
    workflow.add_node("plan_node", plan_node)       # 规划师
    workflow.add_node("search_node", search_node)   # 搜查员
    workflow.add_node("report_node", report_node)   # 撰稿人

    # 2. 定义流转逻辑
    workflow.set_entry_point("plan_node")
    
    workflow.add_edge("plan_node", "search_node")
    workflow.add_edge("search_node", "report_node")
    
    # 3. 编译成可运行的 App
    app = workflow.compile()
    return app

2.3 进阶技巧：添加 AOP 中间件

在生产环境中，我们不能让 Agent “裸奔”。为了监控每个节点的耗时和异常，我引入了一套简单的 AOP 装饰器：

# src/deep_research_agent/middleware/base.py

class MiddlewareManager:
    def wrap_node(self, node_func):
        async def wrapped(state):
            print(f"🚀 [Start] {node_func.__name__}...")
            try:
                result = await node_func(state)
                print(f"✅ [Success] {node_func.__name__}")
                return result
            except Exception as e:
                print(f"❌ [Error] {node_func.__name__}: {e}")
                raise e
        return wrapped

2.4 杀手级功能：Human-in-the-Loop (人机协同)

真正的 Deep Research 不是 Agent 闷头干活，而是允许人类在关键节点介入。LangGraph 对 HITL (Human-in-the-Loop) 的支持非常优雅。
我们可以在图编译时，指定在某些节点前“暂停”：

# 编译时增加 interrupt_before
# 注：本项目采用了生产级的 Event-Driven 模式（详见下文对比），
# 因此我们配置在专门的 'wait_node' 前中断，而不是直接卡在业务节点前。
app = workflow.compile(
    checkpointer=memory,           # 需要有记忆才能暂停
    interrupt_before=["wait_node"] # 在等待节点前暂停，等待前端的用户指令
)

这样，当 Agent 生成完计划后，程序会挂起。你可以：

1. 从 State 中读取 pending_hitl_event，获取待审核内容。
2. 前端展示弹窗，让人类修改或确认计划。
3. 调用 API 更新 State (批准/拒绝)，触发 Agent 继续运行。

这对于长时间运行的任务（Long-running tasks）至关重要。

2.5 深度对比：为什么不直接断点？(Event-Driven vs Native)

有些读者可能会问：“官方文档里不是直接 interrupt_before=['search_node'] 就可以了吗？为什么要专门搞个 wait_node 和 Event？”

这里有一个工程化落地的细节差异：

特性	官方原生模式 (Native)	本项目模式 (Event-Driven)
中断方式	直接在目标节点前“急刹车”	先生成 Event 写入 State，再在 wait_node 处停车
前端交互	前端只能看到“停了”，不知道为什么停	State 里有明确的 pending_hitl_event 字段，前端可以直接根据这个字段弹出审核弹窗
适用场景	CLI 命令行工具，开发者调试	Web 应用、生产级 SaaS 平台

在本项目中，我们选择了更复杂的 Event-Driven 模式，是为了让前端开发更爽。这也体现了“从 Demo 到 Production”的转换思维。

3. 如何运行本项目？

环境要求

• Python 3.10+
• OpenAI API Key (或其他 LLM Key)

步骤 1：克隆仓库

git clone https://github.com/changflow/deep-research-agent.git
cd deep-research-agent

步骤 2：安装依赖

pip install -r requirements.txt

步骤 3：配置环境

复制 .env.example 为 .env，并填入你的 key：

OPENAI_API_KEY=sk-xxxx
TAVILY_API_KEY=tvly-xxxx  # 用于搜索

步骤 4：一键启动

python app.py

---

4. 总结

通过 LangGraph，我们将原本复杂的“慢思考”过程具象化为了一张清晰的有向无环图（DAG）。如果你正在寻找一个可扩展、生产级的 Agent 模板，本项目绝对是你的最佳起点。

觉得有用的话，请在 GitHub 上点个 Star ⭐️ 支持一下！
👉 https://github.com/changflow/deep-research-agent

版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。