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

---

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

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

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

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

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

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

概念	你的理解	代码对应
State	Agent 的"记忆"——每一步都能读写	一个 TypedDict，记录当前数据、决策、错误
Node	一个执行单元——爬虫、翻译、审核	一个 Python 函数，输入 State 输出更新后的 State
Edge	节点间的路——正常走还是岔路走	add_edge 是直走，add_conditional_edges 是分岔

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

---

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

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

2.1 环境准备

pip install langgraph langchain-openai langchain-community arxiv httpx

2.2 定义 State：Agent 的共享记忆

from typing import TypedDict, List, Annotated
import operator

class ResearchState(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：爬取论文（带重试）

import arxiv

def fetch_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 = []
        for result in client.results(search):
            papers.append({
                "title": result.title,
                "authors": [a.name for a in result.authors],
                "published": result.published.strftime("%Y-%m-%d"),
                "summary": result.summary.replace("\n", " "),
                "url": result.entry_id,
                "pdf_url": result.pdf_url
            })
        
        if not papers:
            state["errors"].append(f"未找到与 '{state['query']}' 相关的论文")
        else:
            print(f"✅ 找到 {len(papers)} 篇论文")
        
        state["papers"] = papers
        state["retry_count"] = 0
        
    except Exception as e:
        state["errors"].append(f"爬取失败: {str(e)}")
        state["retry_count"] += 1
    
    return state

节点 2：翻译摘要

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", temperature=0.3)

def translate_abstracts(state: ResearchState) -> ResearchState:
    """用 LLM 翻译每篇论文的摘要为中文"""
    print(f"🌐 正在翻译 {len(state['papers'])} 篇论文摘要...")
    
    translated = []
    for i, paper in enumerate(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"] = translated
    return state

节点 3：生成综述报告

def generate_report(state: ResearchState) -> ResearchState:
    """基于翻译后的摘要生成中文综述报告"""
    print("📝 正在生成综述报告...")
    
    # 组装上下文
    papers_context = ""
    for i, paper in enumerate(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.content
    print(f"✅ 综述报告生成完成（{len(response.content)} 字符）")
    
    return state

---

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

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

def should_retry_fetch(state: ResearchState) -> str:
    """爬虫节点之后的路由决策"""
    if state["papers"]:
        return "translate"         # 有数据 → 去翻译
    elif state["retry_count"] < 3:
        return "fetch_papers"      # 没数据 + 还没重试够 → 再爬一次
    else:
        return "end"               # 重试 3 次全败 → 放弃

def should_review(state: ResearchState) -> str:
    """翻译之后：要不要让 LLM 自审？"""
    # 策略：如果翻译结果中有明显的"未翻译"标记，触发自审
    for abstract in state["translated_abstracts"]:
        if "Error" in abstract or "failed" in abstract.lower():
            return "retry_translate"
    return "generate_report"

3.2 构建 Graph

from langgraph.graph import StateGraph, 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 定义工具节点

from langchain.tools import tool
from langgraph.prebuilt import ToolNode

@tool
def search_web(query: str) -> str:
    """搜索网络获取最新信息"""
    # 实际项目中接入 SerpAPI / Tavily / Brave Search
    import httpx
    response = httpx.get(
        "https://api.duckduckgo.com/",
        params={"q": query, "format": "json"}
    )
    data = response.json()
    snippets = [item.get("Text", "") for item in data.get("RelatedTopics", [])[:3]]
    return "\n---\n".join(snippets) if snippets else "无搜索结果"

@tool
def save_to_file(content: str, filename: str) -> str:
    """将内容保存到本地文件"""
    with open(filename, "w", encoding="utf-8") as f:
        f.write(content)
    return f"✅ 已保存到 {filename}（{len(content)} 字符）"

tools = [search_web, save_to_file]

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

from langgraph.prebuilt import ToolNode

# 在原有 workflow 上加一个新节点
tool_node = ToolNode(tools)
workflow.add_node("search_web", tool_node)

# 修改论文爬虫后的路由：如果 arxiv 没数据，自动切到 web 搜索
def smart_fetch_router(state: ResearchState) -> str:
    if state["papers"]:
        return "translate"
    elif state["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 加一个"最终输出"节点：自动保存报告

def save_report(state: ResearchState) -> ResearchState:
    filename = f"综述_{state['query'].replace(' ', '_')}.md"
    with open(filename, "w", encoding="utf-8") as f:
        f.write(state["final_report"])
    print(f"📄 报告已保存至 {filename}")
    return state

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 时，这个救命：

from langgraph.graph import END
from IPython.display import Image, display

# 生成 Mermaid 图
mermaid_png = app.get_graph().draw_mermaid_png()
with open("agent_graph.png", "wb") as f:
    f.write(mermaid_png)

display(Image(mermaid_png))

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

---

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

1. State 不要太大

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

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

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

# 编译时加 interrupt_before
app = 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：

from langgraph.checkpoint.sqlite import SqliteSaver

with SqliteSaver.from_conn_string("agent_state.db") as checkpointer:
    app = workflow.compile(checkpointer=checkpointer)
    # 每次 invoke 传入 thread_id
    result = app.invoke(input_data, {"configurable": {"thread_id": "session-1"}})

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

---

七、LangGraph vs 其他 Agent 框架

框架	设计哲学	适合场景	学习曲线
LangGraph	显式状态图，节点+边	复杂的多步推理、需要精确流程控制	🟡 中
CrewAI	角色扮演，多 Agent 对话	团队协作模拟	🟢 低
AutoGen	对话驱动，消息传递	多 Agent 对话式协作	🟡 中
OpenAI Swarm	轻量级，handoff 模式	简单的任务分发	🟢 低
手写 Agent	自己写所有逻辑	简单场景、有特殊需求	🔴 高

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

---

八、总结

这篇你 get 了：

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

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