LangGraph 完整实战手册

导航目录

• 一、前言：简介、价值、场景、安装、依赖、导入验证
• 二、入门基础：核心概念、开发流程、基础组件、常见误区
• 三、核心功能与基础技巧：节点、边、状态、基础图结构
• 四、进阶技巧：复杂图、Agent集成、工具联动、调试可视化
• 五、实战场景：对话机器人、任务规划、本地部署、FastAPI、容错
• 六、高级进阶：LangChain适配、性能优化、自定义组件、部署上线
• 七、核心工具与资源：参数手册、实战模板、学习资源
• 八、常见问题与避坑指南：高频问题修复与高级避坑
• 九、实战案例汇总（3-5个完整案例）
• 十、运行环境与依赖清单
• 十一、总结

一、前言：简介、价值、场景、安装、依赖、导入验证

1.1 LangGraph 是什么

核心说明
LangGraph 是一个“图结构驱动”的 LLM 应用编排框架。你可以把复杂任务拆成多个节点（Node），再用边（Edge）定义执行顺序、分支和循环。它特别适合多步骤任务、复杂对话和 Agent 工作流。

1.2 核心价值

• 可控：显式定义流程，不再依赖“黑盒式”链路。
• 可扩展：节点可接 LLM、工具、检索、数据库。
• 可恢复：支持 checkpoint、interrupt、人类审批。
• 可观测：便于调试节点状态和执行路径。

1.3 应用场景

• 复杂多轮对话机器人（意图切换 + 状态记忆）
• 智能 Agent 规划执行（Plan -> Execute -> Reflect）
• 工作流自动化（文档处理、工单流转、审批流）
• 多步骤任务拆解（分析、检索、汇总、输出）
• 知识库联动（Embedding + VectorStore + 工具）

1.4 安装方法（pip + 镜像）

pip install langgraph langchain langchain-openai langchain-community chromadb fastapi uvicorn

pip install -i https://pypi.tuna.tsinghua.edu.cn/simple langgraph langchain langchain-openai langchain-community chromadb fastapi uvicorn

1.5 核心依赖与版本建议

• Python：3.10+
• langgraph：0.2+
• langchain：0.2+
• 大模型 SDK：如 langchain-openai

1.6 标准导入与验证

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

print("LangGraph import success")

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

class State(TypedDict):
    msg: str

def echo_node(state: State) -> State:
    return {"msg": f"echo: {state['msg']}"}

graph = StateGraph(State)
graph.add_node("echo", echo_node)
graph.add_edge(START, "echo")
graph.add_edge("echo", END)
app = graph.compile()

result = app.invoke({"msg": "hello"})
print(result)
# 效果说明：输出 {'msg': 'echo: hello'}，表示图可正常运行

---

二、入门基础：核心概念、开发流程、基础组件、常见误区

2.1 LangGraph 核心概念

核心说明

• Node（节点）：执行一个动作（调用 LLM、工具、规则函数）
• Edge（边）：定义节点间流转关系
• Graph（图）：由节点和边组成的任务流
• State（状态）：图运行过程中的共享数据容器
• 条件分支：根据状态动态选择下一条边
• 循环逻辑：节点可回到前节点，直到满足退出条件
• START / END：启动和结束锚点

极简代码示例

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

class S(TypedDict):
    n: int

def add_one(state: S) -> S:
    return {"n": state["n"] + 1}

g = StateGraph(S)
g.add_node("add_one", add_one)
g.add_edge(START, "add_one")
g.add_edge("add_one", END)
app = g.compile()
print(app.invoke({"n": 1}))
# 效果说明：输出 {'n': 2}

2.2 基础开发流程

核心说明

1. 环境搭建
2. 导入依赖
3. 定义节点函数
4. 创建图结构
5. 添加节点与边
6. 设置启动/结束节点
7. 运行图
8. 调试优化

完整可运行基础代码示例

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

class WorkflowState(TypedDict):
    user_input: str
    intent: str
    answer: str

def classify_intent(state: WorkflowState) -> WorkflowState:
    text = state["user_input"].lower()
    intent = "weather" if "天气" in text or "weather" in text else "general"
    return {"intent": intent}

def weather_node(state: WorkflowState) -> WorkflowState:
    return {"answer": "今天天气晴，温度 26°C（示例数据）"}

def general_node(state: WorkflowState) -> WorkflowState:
    return {"answer": f"你问的是：{state['user_input']}，这是通用回复。"}

def route(state: WorkflowState) -> str:
    return state["intent"]

graph = StateGraph(WorkflowState)
graph.add_node("classify", classify_intent)
graph.add_node("weather", weather_node)
graph.add_node("general", general_node)

graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route, {"weather": "weather", "general": "general"})
graph.add_edge("weather", END)
graph.add_edge("general", END)

app = graph.compile()
print(app.invoke({"user_input": "北京天气怎么样？", "intent": "", "answer": ""}))

参数说明

• add_conditional_edges(node, router, mapping)
  • node：当前节点名
  • router：路由函数，返回路由键
  • mapping：路由键 -> 下一个节点

2.3 基础组件使用

2.3.1 Graph 创建

from typing import TypedDict
from langgraph.graph import StateGraph

class S(TypedDict):
    text: str

graph = StateGraph(S)

2.3.2 节点定义（基础节点 + 条件节点）

def basic_node(state):
    return {"text": state["text"] + " [processed]"}

def route_node(state):
    return "a" if len(state["text"]) < 10 else "b"

2.3.3 边添加（普通边 + 条件边）

graph.add_edge("node1", "node2")
graph.add_conditional_edges("router", route_node, {"a": "short_path", "b": "long_path"})

2.3.4 图运行与状态查看

app = graph.compile()
result = app.invoke({"text": "hello"})
print(result)

2.4 新手常见误区与避坑

• 节点函数返回格式错误：必须返回 dict（状态增量）
• 边关联混乱：节点名拼写不一致会报找不到节点
• 状态字段未初始化：访问不存在 key 触发异常
• 循环无终止条件：图会无限执行
• 版本冲突：LangChain/LangGraph 主版本不匹配

避坑建议

• 统一定义 TypedDict 状态字段
• 每个循环都加计数器或终止条件
• 锁定依赖版本，避免“昨天能跑今天报错”

---

三、核心功能与基础技巧：节点、边、状态、基础图结构

3.1 节点进阶（工具调用节点、LLM节点、分支节点、循环节点）

3.1.1 工具调用节点

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

class S(TypedDict):
    x: int
    y: int
    result: int

def calc_tool_node(state: S) -> S:
    return {"result": state["x"] + state["y"]}

3.1.2 LLM 节点（OpenAI 示例）

import os
from typing import TypedDict
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", api_key=os.getenv("OPENAI_API_KEY"), temperature=0.2)

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

def llm_node(state: LlmState) -> LlmState:
    resp = llm.invoke(f"请简洁回答：{state['question']}")
    return {"answer": resp.content}

参数说明

• model：模型名
• temperature：创造性（0~1），越低越稳定

3.1.3 分支节点与循环节点（对比）

def branch_router(state):
    return "retry" if state["retry_count"] < 2 else "done"

3.2 边的进阶用法

3.2.1 条件边配置

graph.add_conditional_edges(
    "check",
    lambda s: "ok" if s["score"] >= 60 else "fail",
    {"ok": "pass_node", "fail": "retry_node"}
)

3.2.2 动态边思路

动态边常通过“路由函数 + mapping 字典”实现，运行时由 state 决定下一跳。

3.2.3 多节点联动与循环边

graph.add_edge("retry_node", "check")  # 回环

3.3 状态管理（核心重点）

核心说明

• 状态是图的“数据总线”
• 节点通过返回 dict 更新状态
• 设计时要区分“输入字段 / 中间字段 / 输出字段”

完整状态传递示例

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

class State(TypedDict):
    task: str
    plan: str
    result: str

def plan_node(state: State) -> State:
    return {"plan": f"执行计划：先分析任务 '{state['task']}'，再输出结果。"}

def exec_node(state: State) -> State:
    return {"result": f"已按计划执行：{state['plan']}"}

g = StateGraph(State)
g.add_node("plan", plan_node)
g.add_node("exec", exec_node)
g.add_edge(START, "plan")
g.add_edge("plan", "exec")
g.add_edge("exec", END)
app = g.compile()

print(app.invoke({"task": "写周报", "plan": "", "result": ""}))

3.4 基础图结构搭建（线性 / 分支 / 简单循环）

3.4.1 线性图

START -> A -> B -> END

3.4.2 分支图

START -> Router -> (A or B) -> END

3.4.3 简单循环图

START -> Check -> Retry -> Check -> ... -> END

适用场景对比

• 线性图：固定流程（数据清洗 -> 分析 -> 汇总）
• 分支图：条件路由（按意图选择回复策略）
• 循环图：重试机制（校验失败后回环）

---

四、进阶技巧：复杂图、Agent集成、工具联动、调试可视化

4.1 复杂图结构开发

4.1.1 多条件分支图

def route_by_task(state):
    task_type = state["task_type"]
    if task_type == "analysis":
        return "analysis"
    if task_type == "write":
        return "write"
    return "default"

4.1.2 并行节点思路

LangGraph 可通过分支执行多个节点，再在汇总节点聚合结果（实现“伪并行编排”）。

4.1.3 循环终止逻辑

def should_stop(state):
    return "end" if state["iteration"] >= state["max_iteration"] else "continue"

4.2 智能 Agent 与 LangGraph 结合（规划-执行-反馈）

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

class AgentState(TypedDict):
    goal: str
    plan: str
    output: str
    review: str

def planner(state: AgentState):
    return {"plan": f"计划：先拆解目标[{state['goal']}]，再执行关键步骤。"}

def executor(state: AgentState):
    return {"output": f"执行结果：根据计划完成目标 -> {state['plan']}"}

def reviewer(state: AgentState):
    return {"review": "检查通过，可交付。"}

g = StateGraph(AgentState)
g.add_node("planner", planner)
g.add_node("executor", executor)
g.add_node("reviewer", reviewer)
g.add_edge(START, "planner")
g.add_edge("planner", "executor")
g.add_edge("executor", "reviewer")
g.add_edge("reviewer", END)
app = g.compile()

print(app.invoke({"goal": "完成数据分析报告", "plan": "", "output": "", "review": ""}))

4.3 工具联动与节点集成（Tool + Embedding + VectorStore）

from langchain.tools import tool
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

@tool
def add(a: int, b: int) -> int:
    """返回两个整数之和"""
    return a + b

# 向量库示例（需 OPENAI_API_KEY）
# embeddings = OpenAIEmbeddings()
# vectordb = Chroma(collection_name="demo", embedding_function=embeddings)

场景说明

• Tool 适合可执行动作（计算、查询）
• Embedding + VectorStore 适合知识检索节点

4.4 图调试与可视化

调试要点

• 打印每个节点输入输出状态
• 给关键节点加 try/except 和错误字段写回 state
• 使用流式运行观察执行路径

events = app.stream({"goal": "测试目标", "plan": "", "output": "", "review": ""})
for event in events:
    print(event)
# 效果说明：逐步输出节点执行事件，便于定位卡点

---

五、实战场景：对话机器人、任务规划、本地部署、FastAPI、容错

5.1 复杂多轮对话机器人（分支 + 记忆）

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

class ChatState(TypedDict):
    user_msg: str
    history: List[str]
    intent: str
    reply: str

def detect_intent(state: ChatState):
    text = state["user_msg"]
    if "订单" in text:
        return {"intent": "order"}
    if "退款" in text:
        return {"intent": "refund"}
    return {"intent": "general"}

def order_node(state: ChatState):
    return {"reply": "请提供订单号，我来帮你查询。", "history": state["history"] + [state["user_msg"]]}

def refund_node(state: ChatState):
    return {"reply": "退款已进入审核流程，预计 1-3 天。", "history": state["history"] + [state["user_msg"]]}

def general_node(state: ChatState):
    return {"reply": "我可以帮你处理订单、退款、物流问题。", "history": state["history"] + [state["user_msg"]]}

def route(state: ChatState):
    return state["intent"]

g = StateGraph(ChatState)
g.add_node("detect", detect_intent)
g.add_node("order", order_node)
g.add_node("refund", refund_node)
g.add_node("general", general_node)
g.add_edge(START, "detect")
g.add_conditional_edges("detect", route, {"order": "order", "refund": "refund", "general": "general"})
g.add_edge("order", END)
g.add_edge("refund", END)
g.add_edge("general", END)
app = g.compile()

print(app.invoke({"user_msg": "我想查订单", "history": [], "intent": "", "reply": ""}))

5.2 任务规划与执行（论文写作/数据分析拆解）

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

class TaskState(TypedDict):
    topic: str
    outline: str
    draft: str
    final: str

def make_outline(state: TaskState):
    return {"outline": f"{state['topic']} 大纲：背景-方法-结果-结论"}

def write_draft(state: TaskState):
    return {"draft": f"根据大纲生成初稿：{state['outline']}"}

def polish(state: TaskState):
    return {"final": f"润色后的终稿：{state['draft']}"}

g = StateGraph(TaskState)
g.add_node("outline", make_outline)
g.add_node("draft", write_draft)
g.add_node("polish", polish)
g.add_edge(START, "outline")
g.add_edge("outline", "draft")
g.add_edge("draft", "polish")
g.add_edge("polish", END)
app = g.compile()
print(app.invoke({"topic": "AI 在教育中的应用", "outline": "", "draft": "", "final": ""}))

5.3 本地化部署（本地 LLM 示例思路）

# 以本地模型网关为例（如 Ollama 的 OpenAI 兼容接口）
import os
from langchain_openai import ChatOpenAI

local_llm = ChatOpenAI(
    model="qwen2.5:7b",
    base_url="http://localhost:11434/v1",
    api_key=os.getenv("LOCAL_LLM_KEY", "dummy"),
)

场景解读

• 适合内网、隐私数据、本地调试，降低外网依赖。

5.4 FastAPI 接口封装

from fastapi import FastAPI
from pydantic import BaseModel

app_api = FastAPI()

class RequestBody(BaseModel):
    text: str

@app_api.post("/run")
def run_graph(body: RequestBody):
    result = app.invoke({"user_msg": body.text, "history": [], "intent": "", "reply": ""})
    return {"result": result}

# 启动命令：
# uvicorn your_file:app_api --reload

5.5 错误处理与容错机制（重试 + fallback）

from typing import TypedDict

class SafeState(TypedDict):
    query: str
    answer: str
    error: str

def risky_node(state: SafeState):
    try:
        if "error" in state["query"]:
            raise ValueError("模拟节点异常")
        return {"answer": f"处理成功: {state['query']}", "error": ""}
    except Exception as e:
        return {"answer": "fallback: 系统繁忙，请稍后再试", "error": str(e)}

注意事项

• 异常信息写入状态，便于后续节点做补偿或告警。

---

六、高级进阶：LangChain适配、性能优化、自定义组件、部署上线

6.1 LangGraph 与 LangChain 生态适配

核心说明

• LangChain 负责“能力组件”（LLM/Tool/Retriever）
• LangGraph 负责“流程编排”（分支/循环/状态/中断）

对比代码示例（简化）

# LangChain 常见是线性链式调用
# LangGraph 更适合有条件分支和回环的复杂流程

6.2 性能优化

优化策略

• 减少不必要节点和回环
• 对稳定结果做缓存（如检索结果、工具结果）
• 控制 LLM 调用次数和 token 长度
• 对可独立任务做并发执行（应用层并行）

批量处理示例

inputs = ["任务A", "任务B", "任务C"]
for item in inputs:
    print(app.invoke({"user_msg": item, "history": [], "intent": "", "reply": ""}))

6.3 自定义组件开发（节点 / 边 / 状态管理器）

from dataclasses import dataclass

@dataclass
class CustomConfig:
    max_retry: int = 3
    timeout_sec: int = 10

config = CustomConfig()
print(config)

场景说明

• 可把“重试次数、超时、路由策略”抽象成统一配置，复用到多个图。

6.4 部署与上线（Docker + 监控）

FROM python:3.11-slim
WORKDIR /app
COPY . /app
RUN pip install -r requirements.txt
CMD ["uvicorn", "main:app_api", "--host", "0.0.0.0", "--port", "8000"]

上线建议

• 日志：记录每次图运行的 trace_id
• 监控：节点失败率、平均时延、成本指标
• 回滚：保留稳定版图配置，支持快速切换

---

七、核心工具与资源：参数手册、实战模板、学习资源

7.1 LangGraph 常用参数手册

StateGraph(state_schema)

• state_schema：状态定义（如 TypedDict）

add_node(name, func)

• name：节点名（唯一）
• func：节点函数（输入 state，输出 dict）

add_edge(from_node, to_node)

• 普通固定流转边

add_conditional_edges(node, router, mapping)

• router：返回路由 key 的函数
• mapping：路由 key -> 目标节点

compile()

• 编译图得到可执行 app

invoke(input_state)

• 同步执行并返回最终状态

stream(input_state)

• 流式返回节点执行事件

7.2 优质实战模板（可直接复用）

模板1：线性处理模板

START -> preprocess -> execute -> summarize -> END

模板2：意图分发模板

START -> classify -> (faq | order | refund) -> END

模板3：重试回环模板

START -> call_api -> check_result -> (retry -> call_api) / (success -> END)

7.3 学习资源推荐

• 官方文档：LangGraph Docs
• LangChain 文档：LangChain Docs
• 社区案例：GitHub 搜索 langgraph 示例仓库

---

八、常见问题与避坑指南：高频问题修复与高级避坑

8.1 高频问题与解决方案

问题1：图运行卡死（循环不退出）

错误代码

graph.add_edge("retry", "check")  # 但没有终止条件

修正代码

def router(state):
    return "end" if state["retry_count"] >= 3 else "retry"

原因说明：循环必须有退出分支。

问题2：节点执行失败（返回类型错误）

错误代码

def node(state):
    return "just string"  # 错

修正代码

def node(state):
    return {"result": "ok"}  # 对

原因说明：节点应返回状态增量 dict。

问题3：条件分支不生效

常见原因

• router 返回 key 与 mapping 不一致
• mapping 少配置某个分支

8.2 高级避坑技巧

• 避免节点冗余：相近逻辑可合并为可配置节点
• 优化效率：先做规则分流，再调用 LLM，减少 token 成本
• 降低成本：固定模板 + 低温度 + 控制 max tokens
• 模型适配：不同模型对工具调用格式要求不同，需 A/B 验证

---

九、实战案例汇总（3-5个完整案例）

案例1：客服多意图路由机器人

需求分析：识别订单/退款/通用问题并回复。
环境搭建：pip install langgraph
图结构：START -> detect -> (order/refund/general) -> END
效果说明：支持意图切换与基础记忆。

案例2：论文写作任务拆解 Agent

需求分析：把“写论文”拆成大纲、初稿、润色。
图结构设计：planner -> writer -> reviewer
部署测试：可封装 FastAPI 后提供外部调用。

案例3：检索增强问答工作流（RAG）

需求分析：用户提问 -> 检索 -> 生成答案。
核心节点：query_rewrite、retrieve、answer
效果说明：减少幻觉，提高回答可追溯性。

案例4：容错重试型工具执行流

需求分析：外部 API 不稳定时自动重试与降级。
图结构：call_api -> check -> retry loop / fallback / END
效果说明：稳定性明显提升，失败可追踪。

案例5：本地 LLM 离线 Agent

需求分析：内网环境不能访问公网模型。
方案：本地模型网关 + LangGraph 编排 + 本地工具节点
效果说明：满足隐私和离线运行要求。

---

十、运行环境与依赖清单

10.1 推荐环境

• Python 3.10+
• langgraph 0.2+
• langchain 0.2+
• langchain-openai / langchain-community
• fastapi + uvicorn（API 场景）
• chromadb（RAG 场景）

10.2 一键安装命令

pip install langgraph langchain langchain-openai langchain-community fastapi uvicorn chromadb

10.3 API 配置方法

# Windows PowerShell
$env:OPENAI_API_KEY="your_api_key"

# Linux / macOS
export OPENAI_API_KEY="your_api_key"

---

十一、总结

关键结论

• LangGraph 的核心是：用图结构管理复杂任务流。
• 真正稳定的 Agent 应用，离不开：清晰状态、可控分支、可恢复机制、可观测日志。
• 工程落地优先级建议：先跑通基础图 -> 再做分支循环 -> 最后做容错与部署。