这篇不是"hello world"教程——你会跟着搭一个能自动爬取论文、翻译摘要、生成中文综述报告的完整 Agent，全程可运行。

一、先搞清楚：为什么不是 LangChain，而是 LangGraph？

LangChain 解决的是"把 LLM 和各种工具串起来"的问题——一条线性的 Chain。但真实场景的 Agent 工作流从来不是线性的：

• 翻译失败 → 重试还是换引擎？
• 爬虫没抓到数据 → 跳过还是报错？
• 中间结果质量太低 → 让 LLM 自己 review 一遍？

LangChain 的 Chain 写分支逻辑很痛苦。LangGraph 用状态图（State Graph）解决——每一步都是一个节点，节点间用条件边连接，Agent 自己决定下一步走哪。

┌─────────────┐ │ 开始 │ └──────┬──────┘ │ ┌──────▼──────┐ │ 爬取论文 │ └──────┬──────┘ │ ┌──────▼──────┐ ┌───────│ 判断结果 │───────┐ │ 失败 └─────────────┘ 成功 │ │ │ ┌───────▼──────┐ ┌───────▼──────┐ │ 重试/跳过 │ │ 翻译摘要 │ └───────┬──────┘ └───────┬──────┘ │ │ └──────────┬───────────────────┘ │ ┌──────▼──────┐ │ 生成综述 │ └──────┬──────┘ │ ┌──────▼──────┐ │ 输出结果 │ └─────────────┘

LangGraph 三个核心概念（5 分钟搞懂）

这三个概念就够搭出 90% 的 Agent。

二、实战：搭一个论文综述 Agent

场景：用户输入一个研究领域关键词 → Agent 自动找相关论文 → 翻译摘要 → 生成中文综述报告。

2.1 环境准备

pipinstalllanggraph langchain-openai langchain-community arxiv httpx

2.2 定义 State：Agent 的共享记忆

fromtypingimportTypedDict,List,AnnotatedimportoperatorclassResearchState(TypedDict):# 用户输入query:str# 搜索关键词，如 "large language model agent"max_papers:int# 最多爬几篇论文# 中间结果papers:List[dict]# 爬到的论文列表errors:List[str]# 出错记录# 翻译结果translated_abstracts:List[str]# 翻译后的中文摘要# 最终产出final_report:str# 生成的中文综述# 控制流retry_count:int# 重试次数next_step:str# 下一步走哪

2.3 写节点函数：每个节点做一件事

节点 1：爬取论文（带重试）

importarxivdeffetch_papers(state:ResearchState)->ResearchState:"""从 arxiv 搜索论文，支持重试"""print(f"🔍 正在搜索 arxiv:{state['query']}")try:client=arxiv.Client()search=arxiv.Search(query=state["query"],max_results=state["max_papers"],sort_by=arxiv.SortCriterion.Relevance)papers=[]forresultinclient.results(search):papers.append({"title":result.title,"authors":[a.nameforainresult.authors],"published":result.published.strftime("%Y-%m-%d"),"summary":result.summary.replace("\n"," "),"url":result.entry_id,"pdf_url":result.pdf_url})ifnotpapers:state["errors"].append(f"未找到与 '{state['query']}' 相关的论文")else:print(f"✅ 找到{len(papers)}篇论文")state["papers"]=papers state["retry_count"]=0exceptExceptionase:state["errors"].append(f"爬取失败:{str(e)}")state["retry_count"]+=1returnstate

节点 2：翻译摘要

fromlangchain_openaiimportChatOpenAI llm=ChatOpenAI(model="gpt-4o",temperature=0.3)deftranslate_abstracts(state:ResearchState)->ResearchState:"""用 LLM 翻译每篇论文的摘要为中文"""print(f"🌐 正在翻译{len(state['papers'])}篇论文摘要...")translated=[]fori,paperinenumerate(state["papers"]):prompt=f"""请将以下英文学术论文摘要翻译为中文，要求： 1. 保留专业术语的英文原文并用括号标注中文 2. 保持学术风格，不添加解释 3. 如果摘要过长，恰当分段 英文摘要：{paper['summary']}"""response=llm.invoke(prompt)translated.append(response.content)print(f" [{i+1}/{len(state['papers'])}] ✅{paper['title'][:50]}...")state["translated_abstracts"]=translatedreturnstate

节点 3：生成综述报告

defgenerate_report(state:ResearchState)->ResearchState:"""基于翻译后的摘要生成中文综述报告"""print("📝 正在生成综述报告...")# 组装上下文papers_context=""fori,paperinenumerate(state["papers"]):papers_context+=f""" ### 论文{i+1}：{paper['title']}- 作者：{', '.join(paper['authors'])}- 发表时间：{paper['published']}- 链接：{paper['url']}**中文摘要**：{state['translated_abstracts'][i]}--- """prompt=f"""你是一位资深学术研究者。请基于以下{len(state['papers'])}篇关于"{state['query']}"的论文，撰写一份中文综述报告。{papers_context}要求： 1. 开头概述该领域的研究现状和趋势（200字左右） 2. 按主题或方法对论文进行分类介绍，每组2-3篇 3. 指出各方法的优缺点和适用场景 4. 结尾给出未来研究方向建议 5. 每篇论文至少引用一次，标注为 [论文N] 6. 总字数控制在 1500-2500 字 请直接输出 Markdown 格式的综述报告。"""response=llm.invoke(prompt)state["final_report"]=response.contentprint(f"✅ 综述报告生成完成（{len(response.content)}字符）")returnstate

三、组装 Graph：让节点连起来

3.1 定义路由逻辑：Agent 自己决定走哪条路

defshould_retry_fetch(state:ResearchState)->str:"""爬虫节点之后的路由决策"""ifstate["papers"]:return"translate"# 有数据 → 去翻译elifstate["retry_count"]<3:return"fetch_papers"# 没数据 + 还没重试够 → 再爬一次else:return"end"# 重试 3 次全败 → 放弃defshould_review(state:ResearchState)->str:"""翻译之后：要不要让 LLM 自审？"""# 策略：如果翻译结果中有明显的"未翻译"标记，触发自审forabstractinstate["translated_abstracts"]:if"Error"inabstractor"failed"inabstract.lower():return"retry_translate"return"generate_report"

3.2 构建 Graph

fromlanggraph.graphimportStateGraph,END# 创建状态图workflow=StateGraph(ResearchState)# 添加节点workflow.add_node("fetch_papers",fetch_papers)workflow.add_node("translate_abstracts",translate_abstracts)workflow.add_node("generate_report",generate_report)# 设置入口workflow.set_entry_point("fetch_papers")# 添加条件边：爬虫 → 翻译 or 重试 or 结束workflow.add_conditional_edges("fetch_papers",should_retry_fetch,{"translate":"translate_abstracts","fetch_papers":"fetch_papers",# 重试回到自己"end":END})# 翻译 → 生成报告（直连）workflow.add_edge("translate_abstracts","generate_report")# 报告 → 结束workflow.add_edge("generate_report",END)# 编译app=workflow.compile()

3.3 运行

result=app.invoke({"query":"AI agent tool use LLM","max_papers":5,"papers":[],"errors":[],"translated_abstracts":[],"final_report":"","retry_count":0,"next_step":"fetch_papers"})# 结果在 result["final_report"] 里print(result["final_report"])

四、进阶：给 Agent 装上外部工具

纯 LLM 推理的 Agent 能力有限。LangGraph 的杀手锏是把工具调用也建模为节点。

4.1 定义工具节点

fromlangchain.toolsimporttoolfromlanggraph.prebuiltimportToolNode@tooldefsearch_web(query:str)->str:"""搜索网络获取最新信息"""# 实际项目中接入 SerpAPI / Tavily / Brave Searchimporthttpx response=httpx.get("https://api.duckduckgo.com/",params={"q":query,"format":"json"})data=response.json()snippets=[item.get("Text","")foritemindata.get("RelatedTopics",[])[:3]]return"\n---\n".join(snippets)ifsnippetselse"无搜索结果"@tooldefsave_to_file(content:str,filename:str)->str:"""将内容保存到本地文件"""withopen(filename,"w",encoding="utf-8")asf:f.write(content)returnf"✅ 已保存到{filename}（{len(content)}字符）"tools=[search_web,save_to_file]

4.2 扩展现有 Agent：加上联网搜索

fromlanggraph.prebuiltimportToolNode# 在原有 workflow 上加一个新节点tool_node=ToolNode(tools)workflow.add_node("search_web",tool_node)# 修改论文爬虫后的路由：如果 arxiv 没数据，自动切到 web 搜索defsmart_fetch_router(state:ResearchState)->str:ifstate["papers"]:return"translate"elifstate["retry_count"]>=2:# arxiv 两轮都没数据，切 web 搜索return"search_web"else:return"fetch_papers"workflow.add_conditional_edges("fetch_papers",smart_fetch_router,{"translate":"translate_abstracts","fetch_papers":"fetch_papers","search_web":"search_web"})

4.3 加一个"最终输出"节点：自动保存报告

defsave_report(state:ResearchState)->ResearchState:filename=f"综述_{state['query'].replace(' ','_')}.md"withopen(filename,"w",encoding="utf-8")asf:f.write(state["final_report"])print(f"📄 报告已保存至{filename}")returnstate workflow.add_node("save_report",save_report)workflow.add_edge("generate_report","save_report")workflow.add_edge("save_report",END)

五、可视化你的 Agent Graph

LangGraph 内置了 Mermaid 可视化——尤其当你开始写几十个节点的复杂 Agent 时，这个救命：

fromlanggraph.graphimportENDfromIPython.displayimportImage,display# 生成 Mermaid 图mermaid_png=app.get_graph().draw_mermaid_png()withopen("agent_graph.png","wb")asf:f.write(mermaid_png)display(Image(mermaid_png))

输出就是一张完整的流程图，所有节点和条件边一目了然。

六、生产环境三个踩坑经验

1. State 不要太大

Agent 每走一步，整个 State 都会传给 LLM。你的translated_abstracts如果包含 10 篇论文的中文摘要，一次请求可能吃掉几万 token。解决：节点间传递摘要而非全文，最终生成报告时才用完整数据。

2. 加 human-in-the-loop 断点

LangGraph 支持在任何节点前暂停，等你人工确认：

# 编译时加 interrupt_beforeapp=workflow.compile(interrupt_before=["generate_report"])# 运行时会在 generate_report 前暂停result=app.invoke(input_data)# 你看完中间结果后，手动批准继续result=app.invoke(None,config)# None = 从断点继续

这个功能在"Agent 要发邮件 / 要花钱调 API"的场景下是硬需求。

3. 用 Checkpointer 持久化状态

默认 State 在内存里，App 挂了全丢。加一个 SQLite checkpointer：

fromlanggraph.checkpoint.sqliteimportSqliteSaverwithSqliteSaver.from_conn_string("agent_state.db")ascheckpointer:app=workflow.compile(checkpointer=checkpointer)# 每次 invoke 传入 thread_idresult=app.invoke(input_data,{"configurable":{"thread_id":"session-1"}})

中断后恢复、长时间运行的批量任务，靠它续命。

七、LangGraph vs 其他 Agent 框架

LangGraph 的不可替代性：你要的是一个稳定的、可测试的、能精确控制每一步决策的 Agent，而不是让 LLM 在"黑盒里自由发挥"。当你需要"第 3 步失败就回退到第 1 步"这种精确控制时，其他框架都会让你抓狂。

八、总结

这篇你 get 了：

• LangGraph 三个核心概念（State / Node / Edge）就能搭 Agent
• 一个真实可运行的论文综述 Agent 完整代码
• 工具节点（ToolNode）的集成方式
• 三个生产环境必须注意的坑

下一步建议：把这两个连起来——用 LangGraph 编排一个 Agent，用 MCP 协议暴露文件系统和数据库工具给这个 Agent。爬虫 Agent 从 arxiv 抓论文 → 翻译 Agent 处理摘要 → 报告 Agent 生成综述 → 最后通过 MCP Server 把报告写入本地文件。全链路自动化，一篇新的硬核文章就有了。