一、为什么需要 LangGraph？

1. Chain 只能“一条线”：当业务出现“循环、分支、并行、回滚”时，SequentialChain 立刻崩溃。
2. Agent 黑盒不可控：ReAct 一旦进入死循环，只能干瞪眼。
3. LangGraph 把 LLM 工作流变成“有向图”：节点（Node）= 函数，边（Edge）= 条件，支持循环、分支、人机在环，天然可可视化、可打断点、可单步重试。

一句话：Chain 是公路，LangGraph 是立交桥 + 红绿灯 + GPS 导航。

---

二、LangGraph 核心概念

Graph
├─ State(状态)     ：整个图的“共享内存”，Python TypedDict
├─ Node(节点)      ：普通函数或 LCEL 链，输入 State → 输出 State
├─ Edge(边)        ：固定边 | 条件边（函数返回字符串决定下一步）
├─ Entry / END     ：入口与终止
├─ Breakpoint      ：人机在环，暂停等待用户输入
└─ Visualization   ：一键生成 png，调试神器

---

三、5 分钟安装 & 第一条“图”

pip install langgraph langchain openai python-dotenv

import os
from langgraph.graph import StateGraph, END
from typing import TypedDict

class State(TypedDict):
    question: str
    answer: str

# 1. 节点函数
def human_node(state: State):
    print("👤 用户问题：", state["question"])
    return state

def llm_node(state: State):
    # 这里先用伪代码，后面替换成真实 LLM
    state["answer"] = f"模拟答案：{state['question']} 的答案是 42"
    return state

# 2. 建图
builder = StateGraph(State)
builder.add_node("human", human_node)
builder.add_node("llm", llm_node)
builder.add_edge("human", "llm")
builder.add_edge("llm", END)
builder.set_entry_point("human")

graph = builder.compile()

# 3. 运行
result = graph.invoke({"question": "地球终极答案"})
print(result)

输出：

👤 用户问题： 地球终极答案
{'question': '地球终极答案', 'answer': '模拟答案：地球终极答案 的答案是 42'}

🎉 你的第一张图跑通了！

---

四、6 大实战案例，由浅入深

案例 1：带条件分支的“智能客服”

需求：用户情绪负面→转人工；中性/正面→LLM 自动回复。

from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

llm = ChatOpenAI(temperature=0)

class State(TypedDict):
    query: str
    sentiment: str  # positive | negative | neutral
    reply: str

def sentiment_node(state: State):
    prompt = ChatPromptTemplate.from_template(
        "判断下列用户情感属于 positive negative neutral 之一，只返回小写单词：\n{query}"
    )
    chain = prompt | llm
    state["sentiment"] = chain.invoke({"query": state["query"]}).content.strip()
    return state

def auto_reply_node(state: State):
    prompt = ChatPromptTemplate.from_template("你是客服，请礼貌回复：\n{query}")
    chain = prompt | llm
    state["reply"] = chain.invoke({"query": state["query"]}).content
    return state

def human_handoff_node(state: State):
    state["reply"] = "[系统] 已转接人工客服，请稍等..."
    return state

def routing_edge(state: State) -> str:
    return "negative" if state["sentiment"] == "negative" else "auto"

builder = StateGraph(State)
builder.add_node("sentiment", sentiment_node)
builder.add_node("auto_reply", auto_reply_node)
builder.add_node("human", human_handoff_node)
builder.add_edge("sentiment", "auto_reply", routing_edge)
builder.add_edge("sentiment", "human", lambda s: "negative" if s["sentiment"]=="negative" else None)
builder.add_edge("auto_reply", END)
builder.add_edge("human", END)
builder.set_entry_point("sentiment")

graph = builder.compile()
print(graph.invoke({"query": "你们这破系统怎么又崩了！"}))

运行结果：

{'query': '你们这破系统怎么又崩了！', 'sentiment': 'negative', 'reply': '[系统] 已转接人工客服，请稍等...'}

---

案例 2：循环图——“Multi-Step 研究助手”

需求：LLM 先生成 3 个子问题→依次搜索→汇总成最终报告；支持“搜不到回滚再提问”。

from langchain.tools import DuckDuckGoSearchRun

search = DuckDuckGoSearchRun()

class State(TypedDict):
    topic: str
    sub_questions: list
    results: list
    report: str
    step: int

def generate_subquestions_node(state: State):
    prompt = ChatPromptTemplate.from_template(
        "针对主题'{topic}'，生成 3 个用于研究的子问题，每行一个，不要编号。"
    )
    chain = prompt | llm
    text = chain.invoke({"topic": state["topic"]}).content
    state["sub_questions"] = [q for q in text.split("\n") if q.strip()]
    state["step"] = 0
    return state

def search_node(state: State):
    q = state["sub_questions"][state["step"]]
    ret = search.run(q)
    state["results"].append({"question": q, "answer": ret})
    return state

def should_continue(state: State) -> str:
    return "search" if state["step"] < len(state["sub_questions"]) - 1 else "report"

def inc_step_node(state: State):
    state["step"] += 1
    return state

def report_node(state: State):
    prompt = ChatPromptTemplate.from_template(
        "根据以下搜索结果写一段 200 字报告：\n{results}"
    )
    chain = prompt | llm
    state["report"] = chain.invoke({"results": state["results"]}).content
    return state

builder = StateGraph(State)
builder.add_node("gen", generate_subquestions_node)
builder.add_node("search", search_node)
builder.add_node("inc", inc_step_node)
builder.add_node("report", report_node)
builder.add_edge("gen", "search")
builder.add_edge("search", "inc")
builder.add_conditional_edges("inc", should_continue, {"search": "search", "report": "report"})
builder.add_edge("report", END)
builder.set_entry_point("gen")

graph = builder.compile()
result = graph.invoke({"topic": "AGI 安全", "results": [], "step": 0})
print(result["report"])

运行后得到 200 字 AGI 安全小报告，且循环 3 次搜索自动结束。

---

案例 3：并行节点——“多路召回 + 重排”

需求：同时用向量库、Google、百科三路召回，再让 LLM 重排最佳答案。

class State(TypedDict):
    q: str
    vec_docs: list
    google_docs: list
    baike_docs: list
    final: str

def vec_retrieve(state: State):
    # 伪代码：FAISS 召回
    state["vec_docs"] = ["vec1", "vec2"]
    return state

def google_retrieve(state: State):
    state["google_docs"] = ["google1", "google2"]
    return state

def baike_retrieve(state: State):
    state["baike_docs"] = ["baike1"]
    return state

def rerank(state: State):
    docs = state["vec_docs"] + state["google_docs"] + state["baike_docs"]
    prompt = f"根据以下片段回答问题'{state['q']}'：\n{docs}"
    state["final"] = llm.invoke(prompt).content
    return state

builder = StateGraph(State)
# 三个节点共享起点，并行执行
for name, node in [("vec", vec_retrieve), ("google", google_retrieve), ("baike", baike_retrieve)]:
    builder.add_node(name, node)
    builder.add_edge(name, "rerank")
builder.add_node("rerank", rerank)
builder.add_edge("rerank", END)
builder.set_entry_point("vec")  # 任意选一个入口即可，其余并行
graph = builder.compile()
print(graph.invoke({"q": "哪吒 2 上映时间"}))

---

案例 4：人机在环——“审批流”

需求：LLM 生成 SQL → 先让人确认 → 再执行。

class State(TypedDict):
    question: str
    sql: str
    human_review: str   # approved | rejected
    result: str

def generate_sql_node(state: State):
    prompt = ChatPromptTemplate.from_template(
        "根据问题生成 SQLite SQL，只返回语句：\n{question}"
    )
    state["sql"] = (prompt | llm).invoke({"question": state["question"]}).content
    return state

def human_review_node(state: State):
    print("=== 请审核下方 SQL ===")
    print(state["sql"])
    state["human_review"] = input("输入 approved 或 rejected：")
    return state

def exec_sql_node(state: State):
    import sqlite3
    conn = sqlite3.connect(":memory:")
    conn.execute("CREATE TABLE sales(id int, amount int)")
    conn.execute("INSERT INTO sales VALUES (1,100), (2,200)")
    cursor = conn.execute(state["sql"])
    state["result"] = str(cursor.fetchall())
    return state

def should_exec(state: State) -> str:
    return "exec" if state["human_review"] == "approved" else END

builder = StateGraph(State)
builder.add_node("gen_sql", generate_sql_node)
builder.add_node("review", human_review_node)
builder.add_node("exec", exec_sql_node)
builder.add_edge("gen_sql", "review")
builder.add_conditional_edges("review", should_exec, {"exec": "exec", END: END})
builder.add_edge("exec", END)
builder.set_entry_point("gen_sql")

graph = builder.compile()
print(graph.invoke({"question": "总共销售额是多少？"}))

终端会暂停等你输入 approved，实现真正“人在回路”。

---

案例 5：可视化 + 断点调试

# 生成流程图
graph.get_graph().draw_mermaid_png(output_file_path="workflow.png")

打开 workflow.png 就是节点-边结构图。
调试时可在任意节点后加 Breakpoint：

from langgraph.checkpoint import MemorySaver
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)
# 运行后
state = graph.get_state("节点名")
print(state)  # 查看中间变量

---

案例 6：接入真实 LLM + 异步加速

from langchain.chat_models import ChatOpenAI
from langgraph.graph import StateGraph, END
import asyncio

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

class State(TypedDict):
    ask: str
    answer: str

async def ainvoke_node(state: State):
    state["answer"] = (await llm.ainvoke(state["ask"])).content
    return state

builder = StateGraph(State)
builder.add_node("llm", ainvoke_node)
builder.add_edge("llm", END)
builder.set_entry_point("llm")
graph = builder.compile()

async def main():
    result = await graph.ainvoke({"ask": "用一句话介绍 LangGraph"})
    print(result)

asyncio.run(main())

利用 graph.ainvoke 原生支持异步，并发量直接拉满。

---

五、LangGraph vs 传统 Chain/Agent 速查表

能力	Chain	Agent	LangGraph
顺序执行	✅	✅	✅
条件分支	❌	弱	✅
循环/回滚	❌	可能死循环	✅ 显式控制
并行节点	❌	❌	✅
人机在环	❌	❌	✅ Breakpoint
可视化	❌	❌	✅ 一键 PNG
状态持久化	❌	❌	✅ Checkpoint

---

六、生产级最佳实践 checklist

1. 状态设计

   • 用 TypedDict 严格声明字段，避免“魔法 key”。
   • 大文件/图片放指针，别直接塞 State。

2. 循环安全

   • 显式计数器 step += 1，超阈值强制 END。

3. 异步加速

   • 节点函数直接定义 async def，graph.ainvoke 自动并行。

4. 错误重试

   • 节点内 try/except + 状态标记 retry_times，条件边决定重试或回滚。

5. 可视化 + 观测

   • 每次 CI 自动生成 workflow.png 插入文档；
   • 接入 LangSmith，自动追踪每个节点耗时、Token 耗用。

6. 状态持久化

   • 用 RedisSaver 或 SQLiteSaver 替代 MemorySaver，重启服务也能断点续跑。

---

七、总结

LangGraph 把 LLM 应用从“单行道”升级为“立交桥”：

• 节点即函数，想写什么逻辑就写什么；
• 条件边让循环/分支一目了然；
• 人机在环让关键步骤永远可干预；
• 一键可视化，调试不再靠猜。
  AI大模型学习代码仓库