0. 为什么需要 LangGraph？

传统 LangChain 用“链”描述线性流程，一到循环、分支、并行、回溯就捉襟见肘。
LangGraph 把整条链路升级成有向图：

• 节点 = 任意函数 / LLM / Tool
• 边 = 流转规则（可带条件）
• 状态 = 全局共享“内存”

结果：多轮对话、工具调用、循环反思都能一张图表达，代码量反而更少。

---

1. 核心概念 4 件套（5 分钟速记）

组件	解释	类比
State	全局字典，节点间传递数据	内存条
Node	处理函数，接收 State → 返回新 State	CPU 指令
Edge	连接节点，可条件分支	总线
Graph	整张有向图，编译后可调用	主板电路

所有节点只读写同一份 State，天然解耦，又自带“记忆”。

---

2. 安装 & 最小可运行 Demo

pip install langgraph langchain-core

from typing import TypedDict, List
from langgraph.graph import StateGraph, START, END

# ① 定义 State
class AgentState(TypedDict):
    messages: List[str]
    current_step: str

# ② 写节点函数
def hello_node(state: AgentState) -> AgentState:
    state["messages"].append("Hello from LangGraph!")
    state["current_step"] = "hello"
    return state

def bye_node(state: AgentState) -> AgentState:
    state["messages"].append("Bye ~")
    state["current_step"] = "bye"
    return state

# ③ 建图并连边
builder = StateGraph(AgentState)
builder.add_node("hello", hello_node)
builder.add_node("bye",  bye_node)
builder.add_edge(START, "hello")
builder.add_edge("hello", "bye")
builder.add_edge("bye",  END)

graph = builder.compile()

# ④ 运行
result = graph.invoke({"messages": [], "current_step": ""})
print(result["messages"])
# 输出：['Hello from LangGraph!', 'Bye ~']

这就是最小图：START → hello → bye → END。

---

3. 实战：带条件分支的“智能客服”

场景

用户输入 → 识别意图 → 根据意图

• 咨询 → 检索知识库
• 投诉 → 生成工单
• 其他 → 直接回答

3.1 定义共享 State

from typing import TypedDict, List, Optional

class State(TypedDict):
    user_input: str
    intent: Optional[str] = None      # 意图
    answer: Optional[str] = None      # 最终回答
    ticket_id: Optional[str] = None   # 工单号

3.2 实现节点函数

from langchain_community.chat_models import ChatOpenAI

llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

def classify_intent(state: State) -> State:
    prompt = f"用户输入：{state['user_input']}\n\n请返回意图类别：咨询/投诉/其他"
    state["intent"] = llm.invoke(prompt).content.strip()
    return state

def handle_consult(state: State) -> State:
    # 模拟 RAG
    state["answer"] = "根据知识库，咨询答案是 ..."
    return state

def handle_complaint(state: State) -> State:
    # 模拟生成工单
    state["ticket_id"] = "T10086"
    state["answer"] = "已为您生成投诉工单，编号 T10086，专员将在 24h 内联系您。"
    return state

def handle_other(state: State) -> State:
    state["answer"] = llm.invoke(state["user_input"]).content
    return state

3.3 条件边：让 LLM 决定走向

def route_intent(state: State) -> str:
    return state["intent"].lower()

3.4 建图 & 编译

from langgraph.graph import StateGraph, START, END

workflow = StateGraph(State)
workflow.add_node("classify", classify_intent)
workflow.add_node("consult",  handle_consult)
workflow.add_node("complaint", handle_complaint)
workflow.add_node("other",     handle_other)

workflow.add_edge(START, "classify")
workflow.add_conditional_edges(
    "classify",
    route_intent,
    {"咨询": "consult", "投诉": "complaint", "其他": "other"}
)
workflow.add_edge("consult",  END)
workflow.add_edge("complaint", END)
workflow.add_edge("other",    END)

app = workflow.compile()

3.5 运行

out = app.invoke({"user_input": "我要投诉你们产品！"})
print(out["answer"])
# 已为您生成投诉工单，编号 T10086...

意图分类 → 条件分支 → 不同处理，一张图解决。

---

4. 高级技巧：循环 + 反思（Self-Correct）

场景

LLM 先写代码 → 自动执行 → 失败则 反思+重试（最多 3 次）。

import subprocess, re
from langgraph.graph import StateGraph, START, END

class CodeState(TypedDict):
    task: str
    code: str
    error: str | None
    attempts: int

def write_code(state: CodeState) -> CodeState:
    state["code"] = llm.invoke(f"用 Python 实现：{state['task']}，只返回代码块").content
    state["attempts"] += 1
    return state

def run_code(state: CodeState) -> CodeState:
    try:
        out = subprocess.check_output(["python", "-c", state["code"]], stderr=subprocess.STDOUT, text=True)
        state["error"] = None
    except subprocess.CalledProcessError as e:
        state["error"] = e.output
    return state

def should_continue(state: CodeState) -> str:
    if state["error"] is None or state["attempts"] >= 3:
        return "end"
    return "reflect"

def reflect(state: CodeState) -> CodeState:
    prompt = f"代码执行报错：{state['error']}\n\n请修正代码，只返回修正后的完整代码块。"
    state["code"] = llm.invoke(prompt).content
    return state

# 建图
g = StateGraph(CodeState)
g.add_node("write",  write_code)
g.add_node("run",   run_code)
g.add_node("reflect", reflect)
g.add_edge(START, "write")
g.add_edge("write", "run")
g.add_conditional_edges("run", should_continue, {"reflect": "reflect", "end": END})
g.add_edge("reflect", "run")
g.add_edge("run", END, when=lambda s: s["error"] is None)

bot = g.compile()
final = bot.invoke({"task": "快速排序", "code": "", "error": None, "attempts": 0})
print(final["code"])

节点 run → 条件边判断 → reflect 节点循环，天然 while 语法。

---

5. 彩蛋：条件边返回 "human" → 弹出审批框

场景

高风险操作（转账 > 10k、删除数据库）必须经过人工审批才能继续。

5.1 定义状态

class PaymentState(TypedDict):
    amount: float
    currency: str
    to_account: str
    approved: bool | None  # None=待审批，True=通过，False=拒绝

5.2 节点函数

def check_amount(state: PaymentState) -> PaymentState:
    if state["amount"] > 10_000:
        state["approved"] = None   # 标记需审批
    else:
        state["approved"] = True   # 小额自动通过
    return state

def wait_human(state: PaymentState) -> PaymentState:
    # 这里只是占位，真正部署时会阻塞等待外部审批系统
    print(f"[系统] 等待人工审批：向 {state['to_account']} 转账 {state['amount']} {state['currency']}")
    # 模拟审批结果（实际从外部系统读取）
    state["approved"] = input("审批结果（y/n）：").lower() == "y"
    return state

def execute(state: PaymentState) -> PaymentState:
    print(f"[执行] 转账成功 ✅")
    return state

def reject(state: PaymentState) -> PaymentState:
    print(f"[拒绝] 转账已取消 ❌")
    return state

5.3 条件边：返回 "human" 或 "exec" 或 "reject"

def route_after_check(state: PaymentState) -> str:
    if state["approved"] is None:
        return "human"      # 需要人工审批
    return "exec" if state["approved"] else "reject"

5.4 建图 & 编译

from langgraph.graph import StateGraph, START, END

workflow = StateGraph(PaymentState)
workflow.add_node("check",   check_amount)
workflow.add_node("human",   wait_human)   # 人工审批节点
workflow.add_node("exec",    execute)
workflow.add_node("reject",  reject)

workflow.add_edge(START, "check")
workflow.add_conditional_edges("check", route_after_check,
                               {"human": "human", "exec": "exec", "reject": "reject"})
workflow.add_edge("human", "exec")   # 审批通过后执行
workflow.add_edge("human", "reject", when=lambda s: s["approved"] is False)
workflow.add_edge("exec", END)
workflow.add_edge("reject", END)

app = workflow.compile()

5.5 运行：大额触发审批

# 小额自动通过
app.invoke({"amount": 100, "currency": "CNY", "to_account": "A123", "approved": None})
# 输出：[执行] 转账成功 ✅

# 大额等待人工输入
app.invoke({"amount": 50000, "currency": "CNY", "to_account": "B456", "approved": None})
# 控制台弹出：
# [系统] 等待人工审批：向 B456 转账 50000.0 CNY
# 审批结果（y/n）：y
# [执行] 转账成功 ✅

5.6 生产级扩展

• 前端弹窗：把 wait_human 改成 REST 阻塞调用 → 企业微信 / 飞书审批
• 超时拒绝：加 asyncio.wait_for(human_approval(), timeout=300)
• 审计日志：任何 approved=None 状态写入 LangSmith / 自建审计表

---

6. 与 LangChain 无缝集成

任何 langchain 的 LLM、Tool、Retriever 都能直接当节点用：

from langchain.tools import ShellTool
from langchain import hub
from langgraph.graph import StateGraph

prompt = hub.pull("hwchase17/react")  # 官方 ReAct prompt
tool = ShellTool()
agent_executor = create_react_agent(model, [tool], prompt)  # LangGraph 内置

graph = agent_executor  # 本身就是一张图
graph.invoke({"input": "列出当前目录下文件并按大小排序"})

一行代码把传统 AgentExecutor 升级成有状态图。

---

7. 可视化调试：随手出图

from IPython.display import Image
Image(app.get_graph().draw_mermaid_png())

自动生成 Mermaid 流程图，节点-边-条件一目了然，调试效率 ×3。

---

8. 生产级 Tips

1. 断点续跑
   用 CheckpointSaver 把 State 快照到 Redis/SQLite，重启后从断点继续。

2. 人机协同（本文彩蛋）
   条件边返回 "human" → 弹出审批框 → 人工确认后再继续，关键操作双保险。

3. 并行化
   Send API 一键 Map-Reduce，同时生成故事+翻译+摘要，总耗时由串行 9s → 并行 3s。

---

9. 总结：一张图记住 LangGraph

State（共享内存）
   ↑
Node（函数/LLM/Tool） ←→ Edge（条件或顺序）
   ↓
START / END / 循环 / 并行 / 人机

把业务画成图，把图跑成代码——LangGraph 让复杂代理从“玄学”变“工程”。

---

下一步

• 阅读官方 LangGraph 文档
• 把你的 RAG 脚本改成“图” → 支持 循环优化查询
• 尝试 多智能体 子图：客服 + 技术 + 财务 三方协同

现在就 pip install langgraph，把今天的示例跑通，你就拥有了第一张“会循环的 AI 工作流”！
学习代码