直接开工，咱们聊聊用LangGraph搭一个类似DeepResearch的AI Agent这件事。
我最早接触LangGraph是某次做复杂Agent项目，项目里要处理十几步动态决策，用LangChain的老写法写Chain怎么都别扭——条件分支、循环、人工确认、断点恢复，逻辑全堆一起，代码烂到不想看第二遍。后来换LangGraph，把Agent的每一步定义成节点，把流转规则画成图，整个项目结构瞬间清楚了。今天这篇就是把这段经验拆开揉碎，从核心原理到代码实操，按我实际跑通一套DeepResearch流程的路径来讲。
1. 为什么DeepResearch这类Agent必须用图来编排
1.1 传统Chain方案的根本痛点
先想一下DeepResearch这个场景到底在做什么。给一个复杂问题，Agent要先理解意图，拆成子问题，然后逐个调研、搜集信息、交叉验证，最后汇总成长文报告。这里面有几个绕不开的麻烦：
- 调研深度没法预设。根据问题难度，可能需要查3个来源，也可能需要查10个来源，Agent得自己判断什么时候信息够了。
- 分支逻辑复杂。某个子问题查不到答案，是换个关键词重查，还是直接跳过？答案需要追问时，要不要启动第二轮询问？
- 流程会中断。API超时、单次查询结果异常、人类要中途补充信息，这些情况都需要流程能暂停、恢复。
老式的LangChain链式写法，本质是把步骤A到步骤F严格执行一遍，遇到"查不到就换个策略"这种动态分支，就得在代码里堆if-else，步骤一多整个逻辑就炸了。我见过一个项目写了十几个嵌套回调，中间改一次流程，等于把代码推倒重写。这不是写法问题，是抽象方式根本不匹配。DeepResearch的流程天然是一张图，不是一条链。
1.2 图结构如何让动态决策变成"搭积木"
LangGraph的核心抽象非常直接：把Agent的每一步定义成节点，把"下一步走哪"定义成边。节点之间通过共享状态传递数据，边可以是固定的，也可以是带条件判断的动态边。这样一来，流程的复杂度被分摊到一个个节点内部，节点之间的连接规则单独管理。
这个思路和电路图很像。每个节点像一个独立模块，输入进来，处理完，吐出去。你要加一个新功能，就增加一个节点，连两条边，不用去动老节点的代码。要改调研策略，就改一条条件边的判断逻辑。整个流程的复杂度从"代码里的逻辑"转移到了"图里的结构"，而图是可视化的，一眼就能看懂。
1.3 状态管理与可控性：LangGraph的护城河
光有节点和边还不够，Agent系统最容易被忽略的是状态管理。DeepResearch跑起来可能持续几分钟，中间Agent的记忆、已查过的信息、待办任务、最终报告草稿，这些状态散落各处的话，调试起来会让人崩溃。
LangGraph用共享State对象管理所有数据，Any类型、TypedDict都行，灵活度很高。再加上LangGraph的检查点机制，每步执行完可以自动保存状态，进程挂了能从上次节点恢复，这个能力对长时间运行的Agent太关键了。更别提它还内置了"人在回路"的支持——在某个节点暂停，等用户确认再继续。这些不是锦上添花的功能，是做生产级Agent的刚需。
所以我的结论是，只要Agent流程里存在条件分支、循环、人工介入，图就是比链更合适的抽象方式。DeepResearch这种场景，几乎把图的所有优势都用上了。
2. LangGraph核心概念拆解：小白视角的通俗翻译
2.1 状态（State）：Agent的共享笔记本
LangGraph里State是所有节点共享的数据容器，理解成团队协作时的共享笔记本就行。每个节点可以在笔记本上读内容、写新内容，下一个节点接着读。
代码里最常见的定义方式是用TypedDict，指定每个字段的类型：
from typing_extensions import TypedDict class ResearchState(TypedDict): question: str # 原始问题 plan: list # 拆分后的子问题列表 research_data: dict # 收集到的调研信息 report: str # 最终报告 iteration: int # 当前循环轮次State的更新规则有一个关键点，默认是"覆盖式"的：节点返回 {"plan": [...]}，会直接覆盖旧的plan。如果你想让多个字段累加，比如不断追加research_data，需要在定义State时用Annotated加reducer函数。
from typing import Annotated from langgraph.graph import add_messages class ResearchState(TypedDict): messages: Annotated[list, add_messages]我用add_messages的示例比较多，因为它能自动处理消息追加，正好适合多轮调研的数据积累。其他自定义字段用什么reducer，根据业务逻辑定，初期用覆盖式足够了。
2.2 节点（Node）：一次只做一件事的工人
节点就是普通的Python函数，输入整个State，输出要更新的字段。设计原则只有一个：职责单一。比如"生成调研计划"的节点只负责拆解问题，"执行搜索"的节点只负责调用搜索接口，"撰写报告"的节点只负责汇总输出。
每个节点就是一个工人，用笔记本（State）交流，各干各的活儿：
def generate_plan(state: ResearchState): question = state["question"] sub_questions = [f"{question}的最新进展", f"{question}的主要挑战"] return {"plan": sub_questions} def search_node(state: ResearchState): plan = state["plan"] results = [] for sq in plan: results.extend(search_api(sq, top_k=5)) return {"research_data": results}2.3 边（Edge）与条件边（Conditional Edge）：红绿灯
边就是"这个节点跑完，下一步跑谁"。固定边用add_edge直接连，但真正体现LangGraph能力的，是条件边。条件边是一个判断函数，输入State，输出下一个节点的名字。
def should_continue(state: ResearchState): if len(state["research_data"]) < 5: return "search_node" else: return "write_report"这就像一个红绿灯，根据当前情况自动选择路径。DeepResearch的循环、提前终止、任务切换，都是靠条件边实现的。掌握条件边，等于掌握了图编排的灵魂。
2.4 Checkpointer：Agent的存档系统
Checkpointer负责在每步执行后保存State快照。有了它才有"断点续跑"、"随时暂停"、"流程回溯"这些能力。
from langgraph.checkpoint.memory import MemorySaver checkpointer = MemorySaver() graph = workflow.compile(checkpointer=checkpointer)MemorySaver适合开发和调试，数据放在内存里。生产环境建议用SQLite或PostgreSQL适配器持久化到磁盘或数据库，防止重启丢状态。这个细节，很多人一开始不重视，等系统真跑起来就知道有多重要了。
3. DeepResearch实战：从零搭一套多阶段调研Agent
3.1 项目需求与整体设计
目标很明确：输入任意问题，Agent自动完成"拆解问题-收集资料-补充验证-输出报告"全流程。技术栈选LangGraph做编排核心，搜索用通用的搜索API（任何你手头能用的封装都行），LLM调用先用OpenAI兼容接口示例。
整体流程图设计思路如下：
- 入口：接收问题，生成调研计划。
- 主循环：执行每轮搜索、提炼信息。
- 判断：轮次不足或信息不够，继续循环；超过最大轮次，强制收尾。
- 报告生成：所有信息汇总，写最终报告。
- 人工确认（可选）：报告前加一道确认闸门。
3.2 候选方案选型与理由
我之前想过用更多节点把流程做细，比如拆出"意图识别节点"、"关键词扩展节点"、"去重节点"，但后来砍掉不少。原因是：图节点越多，流程越灵活，但每一跳都有函数调用开销和LLM调用开销，节点过多会让单轮流程变慢。DeepResearch的核心瓶颈在于LLM调用数量，所以节点划分要顺着"一次LLM调用"来切，而不是把函数拆得过碎。
搜索API我优先用了支持异步的版本，后面聊并行加速时你会看到异步的好处。LLM接口统一封装在util里，方便快速切换模型供应商。
3.3 环境准备与依赖安装
先搞定依赖。我推荐新建干净环境，避免依赖冲突：
python -m venv venv source venv/bin/activate # Windows上执行 venv\Scripts\activate pip install langgraph langchain-openai版本说明：LangGraph更新很快，API有小幅变动，跑示例时如果报错先看模块名字是不是变了。搜索服务的具体封装各家不一样，核心是把网络请求的返回结果统一成字典结构保存进State。
3.4 从0到1实现：每一步的完整代码
先定义搜索工具。为了示例完整，我写一个模拟工具，你的实现换成真实搜索API就行：
import requests import os SEARCH_API_KEY = os.getenv("SEARCH_API_KEY") def search_web(query: str, top_k: int = 5) -> list: # 这里接任意搜索服务 # 返回: [{"title": ..., "link": ..., "snippet": ...}] return search_api_call(query=query, top_k=top_k)再定义LLM调用函数，假设用OpenAI兼容接口：
from langchain_openai import ChatOpenAI llm = ChatOpenAI( model="gpt-4o-mini", temperature=0.2, api_key=os.getenv("OPENAI_API_KEY"), )接下来按节点实现编排。把每个节点定义成普通函数，加上清晰注释：
from typing_extensions import TypedDict class AgentState(TypedDict): question: str plan: list findings: dict iteration: int max_iterations: int report: str def plan_node(state: AgentState): """基于问题生成子问题列表""" prompt = f"针对问题【{state['question']}】，列出3-5个需要调研的子问题，输出JSON数组。" resp = llm.invoke(prompt) try: plan = json.loads(resp.content) except Exception: plan = [state["question"]] return {"plan": plan} def search_node(state: AgentState): """遍历计划，执行搜索，把结果累加到findings""" queries = state["plan"] new_findings = {} for q in queries: new_findings[q] = search_web(q, top_k=3) return {"findings": new_findings}到这里，同步的搜索是慢的，每条搜索都要等网络。这还只是五六个查询，如果扩展到几十条，整套流程能跑几分钟。所以我实际项目里，搜索节点用的是异步实现，一会儿单独讲。
主循环的"提炼信息"节点，把原始搜索结果交给LLM提取要点，省得报告阶段处理海量噪音：
def analyze_node(state: AgentState): """把搜索到的原始信息压缩成要点""" findings = state["findings"] summary = {} for q, results in findings.items(): prompt = f"基于以下搜索结果，提炼关于问题的要点，不超过150字：\n{results}" summary[q] = llm.invoke(prompt).content return {"summaries": summary}最后是"判断是否继续"的条件边逻辑：
def should_keep_searching(state: AgentState): """条件边：根据迭代次数和已收集信息决定下一步""" if state["iteration"] >= state["max_iterations"]: return "report" # 还可以加：关键信息缺失判断 if len(state.get("findings", {})) < 2: return "search_improved" return "analyze"完整编译图的操作如下：
from langgraph.graph import StateGraph, START, END builder = StateGraph(AgentState) builder.add_node("plan", plan_node) builder.add_node("search", search_node) builder.add_node("analyze", analyze_node) builder.add_node("report", report_node) builder.add_edge(START, "plan") builder.add_edge("plan", "search") builder.add_edge("search", "analyze") builder.add_conditional_edges( "analyze", should_keep_searching, { "report": "report", "search_improved": "search", "analyze": "analyze", }, ) builder.add_edge("report", END) graph = builder.compile(checkpointer=MemorySaver())运行整个流程：
config = {"configurable": {"thread_id": "research-001"}} initial_state = {"question": "如何优化本地大模型的推理延迟？", "plan": [], "findings": {}, "iteration": 0, "max_iterations": 3} result = graph.invoke(initial_state, config) print(result["report"])这里的thread_id就是一个会话ID，checkpointer靠它区分不同会话的状态。同一个thread_id多次调用，流程可以接着上次的状态继续跑，这对长任务非常重要。
3.5 状态累积与循环控制的细节
我在搭建这个流程时遇到一个隐蔽问题：每次search节点返回 {"findings": new_findings}，默认覆盖了之前的findings，导致上一轮的搜索成果直接被清空。如果你希望多轮调研的结果累积起来，就得靠reducer。
这里放一个实际可用的reducer写法：
from typing import Annotated def merge_dict(left: dict, right: dict) -> dict: if not left: return right merged = left.copy() for k, v in right.items(): if k in merged: # 如果同一问题的结果重复调研，合并列表并去重 merged[k] = merged[k] + v else: merged[k] = v return merged class AgentState(TypedDict): question: str plan: list findings: Annotated[dict, merge_dict] # 关键在这行 iteration: int max_iterations: int report: str加了Annotated后，每次search节点返回的findings会自动和旧值合并，不再是覆盖。这个细节，新手十有八九会踩坑。
循环计数也有讲究。iteration字段在节点里负责自增，初始值0，plan_revision节点每次自增1，条件边根据iteration判断是否超限。记住：LangGraph自身没有内置循环限制，能不能正常退出全看你条件边的终止判断写没写对。为了防止意外死循环，我在启动参数里固定max_iterations，默认3轮，够用且安全。
4. 进阶能力：并行执行、检查点恢复与人在回路
4.1 异步节点实现并行搜索调用
前面提到，多个搜索之间互不依赖，串行跑太浪费时间。LangGraph原生支持异步节点，把节点函数定义成async def，invoke改成ainvoke即可。
import asyncio async def search_node_async(state: AgentState): queries = state["plan"] new_findings = await asyncio.gather( *(smart_search(q) for q in queries) ) return {"findings": dict(zip(queries, new_findings))}smart_search里面用你选型的异步客户端发请求，结果一次性收集。我实测过，把8个搜索任务并行化后，整轮耗时从40秒压到10秒左右，用户体验完全不一样。DeepResearch这种任务，搜索越密集，收益越大。
异步节点在图里使用方式和同步节点完全相同，add_node的逻辑一致，编译阶段会帮你处理好事件循环。有一条要注意：如果你的LLM调用用了同步客户端，而搜索用了异步，建议把LLM调用也切成异步版本，或者干脆用asyncio.to_thread包装，避免事件循环被阻塞。
4.2 设置断点恢复：让人工确认Agent的中间结果
在生产环境里，让Agent完全自主跑完整个流程风险很大。最起码的保障，是在关键节点插入"人工确认"。
LangGraph实现这个非常直接，用interrupt_before参数让图在指定节点前暂停：
config = { "configurable": {"thread_id": "research-002"}, } graph.invoke(initial_state, config, interrupt_before=["report"])流程执行到report节点前会暂停，返回当前状态。用户检查收集的信息够不够，不够就通过update_state补充指令让它继续，够了再resume：
user_feedback = "重点补充一下模型量化方案的对比信息" graph.update_state(config, {"user_feedback": user_feedback}) graph.invoke(None, config) # 继续执行这个能力用来做"人工确认最终报告质量"、"确认是否追加预算查询"这类闸门，非常合适。生产级Agent很难一步到位全自动，"人在回路"是多数真实系统上线的必要条件。
4.3 用持久化Checkpointer实现任务断点续跑
MemorySaver适合调试，但生产环境进程一重启就全丢了。LangGraph官方提供了SQLite和Postgres等持久化适配器，我分享一个SQLite的简单接法：
from langgraph.checkpoint.sqlite import SqliteSaver import sqlite3 conn = sqlite3.connect("agent_state.db", check_same_thread=False) checkpointer = SqliteSaver(conn) graph = workflow.compile(checkpointer=checkpointer)这样即便跑了一半由于网络错误崩溃，重启之后拿到相同的thread_id，直接从最后一个完成的节点继续，不用从头跑一遍。调研类任务动不动几十步，断点续跑的价值不用多说。
4.4 Stream模式实时观测Agent内部步骤
调试时我最常开的功能是stream模式。在LangGraph里，用stream代替invoke，就能像看直播一样看到每一步的输入输出：
for event in graph.stream(initial_state, config=config, stream_mode="updates"): for node_name, node_output in event.items(): print(f"节点 [{node_name}] 输出: {node_output}")初期调试Agent，别急着看最终报告，先看每个节点喂进去什么、吐出来什么。很多问题就是这一步暴露出来的，比如搜索结果解析失败、LLM返回格式不对、字段没更新。stream模式就是Agent系统的调试器，学会用它，排查问题能省一半时间。
5. 常见坑与排查技巧：实测汇总
5.1 字段覆盖与状态合并异常
这个我前面说过了，是最高频的坑。症状：循环了几轮，findings始终只有最后一轮的数据。原因就是默认覆盖式更新。解决办法就是上Annotated+reducer。排查方法很简单，在流程中打印节点返回，发现返回里字段值不是累积的，基本就是没加reducer。
5.2 条件边路由名字不匹配
LangGraph的条件边映射表，键是条件函数的返回值，值是节点的名字。拼写不一致或者引用了不存在的节点名，编译时能通过但运行时会报错。我建议把节点名抽成常量：
NODE_PLAN = "plan" NODE_SEARCH = "search" NODE_ANALYZE = "analyze" NODE_REPORT = "report"所有add_node、add_edge、映射表全用常量引用。字符串一多，人眼根本盯不住，常量至少能让IDE帮你查错。
5.3 LLM返回格式不稳定导致节点崩掉
LLM输出JSON时偶尔会有额外的解释文本，导致json.loads直接抛异常。我的习惯是在LLM调用时强制指定输出格式，或者做好异常兜底。前面plan_node里那个try/except就是兜底，解析失败就退回用原始问题当单一调研项。生产环境建议加个"重试一次"的逻辑，效果更好。
5.4 长流程超时与重试策略
DeepResearch跑几十步，任何一步网络抖动都会中断。我的建议分三层：
- 给每次LLM调用设置合理的timeout参数。
- 对搜索请求做重试包装，最多重试2次，间隔1秒。
- 检查点设置好，任务中断后从最近节点恢复。
5.5 并行节点操作共享状态的注意点
当一个节点里用asyncio.gather并行执行多个任务时，注意这些任务内部不要直接修改State对象。并行任务只负责返回自己的数据，最终由节点统一组装成返回字典。多个写操作同时进行，LangGraph的Reducer也不是线程安全的，最好只在主节点函数里更新状态。并行改造前，先把功能跑通，再加并行。
5.6 Token消耗控制：降低生产成本的实战技巧
DeepResearch最烧钱的地方是每轮循环都要多次调用LLM。我的优化套路：主要流程只用轻量模型（如带mini后缀的），报告生成阶段才换强模型。搜索结果的"提炼要点"用便宜模型足够了，最终写报告再用贵模型，这样效果和成本平衡得很理想。再比如限制每轮最大搜索数、控制上下文的温度参数，这些细节叠加起来成本差异非常明显。我实际项目里，单次调研成本从几块降到几毛，就是因为做了轻量模型前置过滤。
6. 面向未来的扩展：让DeepResearch更强
6.1 接入RAG与本地知识库
搜索外部网络不是唯一的信息来源。企业内部的文档、本地知识库，同样应该成为DeepResearch的调研范围。在搜索节点里加一个判断：如果查询命中了本地知识库，先把本地结果和网络结果合并，再发给LLM提炼。LangGraph的节点是普通函数，加这样一道逻辑几乎不费事。
6.2 引入反射与自我修正
目前这套结构的自我纠错能力有限——信息不够只是多搜几轮，但不会主动反思"搜索策略是否有问题"。可以加一个reflection节点，每轮调研后让LLM评估当前信息是否能回答问题，输出评估意见和改进建议。条件边根据评估结果决定进入下一轮搜索还是写报告。这种反思机制能显著减少无效搜索。DeepResearch的核心目标不是多搜几轮，而是"给出的报告真的能解决疑问"，有了反射节点，这个目标更容易逼近。
6.3 多Agent协作架构演进
复杂项目里，拆成多个专职协作的Agent可能更合适：一个Researcher Agent负责全网搜索，一个Reviewer Agent负责交叉验证信息，一个Writer Agent负责最终撰写。LangGraph对多Agent也很友好——每个Agent本身是一个节点，它们之间通过消息通信，在同一个图里协作，共享同一个State。这个方案比单体Step实现更容易维护，也更贴近真实团队分工。如果你的流程复杂到单Agent代码膨胀严重，这就是下一步该走的方向。
6.4 长期记忆与用户画像整合
对话型DeepResearch要保留用户的使用记录、偏好领域、历史关注话题。结合我们前面用的checkpointer机制，在State里增加memory字段，让Agent在调研前先读历史交互，提供更精准的调研方向。长期记忆引入还能让后续轮次的多轮对话更自然，这也是很多生产级Agent系统的标配能力。
写在最后：我的实际体会
如果非让我说这条技术路线里最重要的一句话，我会说：先用图想清楚Agent的流程，再动笔写代码。LangGraph最大的价值不是性能优化，也不是某个API有多好用，而是逼着你在动手之前把Agent的行为边界、分支逻辑、终止条件都想清楚。流程设计好了，代码只是顺着图填肉而已，小白也能顺畅写完；流程没想清楚，换什么框架都救不了你。
DeepResearch只是Agent应用的一种场景，但我强烈建议你把这个实战流程跑通一遍。这个项目麻雀虽小、五脏俱全——有条件边、有循环、有状态聚合、有断点恢复、有人工干预。这套组合拳打一遍，你对Agent编排的理解会上一个台阶。之后遇到任何复杂的Agent需求，你自然知道该从哪里下手。我个人踩过最深的坑就是早期过度追求node数量，最后维护成本压垮了项目；返璞归真，把节点按"一次任务、一次决策、一次LLM调用"划分之后，代码清爽度直线上升。希望这篇实战笔记能让你少走这些弯路，把绝大多数踩坑时间省下来，直接去做真正有价值的事情。
