Skip to content

LangGraph 完整实战手册 ​

如果说 LangChain 是「把 AI 能力串成一条流水线」,那 LangGraph 就是「给流水线装上红绿灯、岔路口和返工工位」——它用图结构来编排复杂任务,让流程能分支、能循环、能中断、能恢复。本手册按「建立认知 → 上手三件套 → 图编排能力 → 进阶集成 → 实战落地 → 生产上线」循序渐进,每章都配生活类比帮你秒懂概念,适合从入门到工程落地全程查阅。

导航目录 ​

前言篇(先建立整体认知)

入门篇(掌握基础三件套)

核心功能篇(图编排能力)

进阶篇(复杂编排与集成)

实战篇(真实场景落地)

高级篇(生产上线)

案例总结篇(巩固与查阅)

一、LangGraph 是什么与核心价值 ​

核心概念

LangGraph 是一个「图结构驱动」的 LLM 应用编排框架。把复杂任务拆成多个节点(Node),再用**边(Edge)**定义执行顺序、分支和循环,特别适合多步骤任务、复杂对话和 Agent 工作流。

生活类比

如果把 AI 应用比作工厂流水线:LangChain 是一条笔直的传送带(原料进 → 加工 → 成品出);而 LangGraph 是一座带岔路口、返工工位和质检回环的智能车间——遇到不合格品能退回重做(循环),遇到不同订单能走不同产线(分支),还能随时暂停等主管审批(中断/恢复)。

1.1 LangGraph 是什么 ​

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

1.2 核心价值 ​

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

1.3 应用场景 ​

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

二、安装、依赖与环境验证 ​

核心概念

安装 LangGraph 全家桶(含 LangChain、模型 SDK、向量库、Web 框架),锁定 Python 3.10+ 与主版本,最后用最小可运行图验证环境是否就绪,避免“昨天能跑今天报错”。

生活类比

装环境就像搬进新家通水电:先确认水电规格对不对(Python/依赖版本),再逐个开关试一遍(导入验证),最后烧壶水确认真能用(跑通最小图)。别嫌验证麻烦——不然等你搬完家具才发现没通电,返工成本更高。

2.1 安装方法(pip + 镜像) ​

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

2.2 核心依赖与版本建议 ​

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

2.3 标准导入与验证 ​

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

print("LangGraph import success")
python
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.4 新手常见误区与避坑 ​

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

避坑建议

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

三、核心概念:节点/边/图/状态 ​

核心概念

LangGraph 的五大基石:Node(节点) 执行动作、Edge(边) 定义流转、Graph(图) 组织整体任务流、State(状态) 是贯穿全图的共享数据容器,再加上 START/END 起止锚点。分支与循环都建立在这五者之上。

生活类比

把一次图运行想象成地铁出行:车站是节点(每站办一件事),轨道是边(决定往哪开),整条线路图是 Graph,而你手里那张一路刷到底的交通卡就是 State——余额、乘车记录随行程不断更新,每个闸机(节点)都读它、改它。

3.1 LangGraph 核心概念 ​

核心说明

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

极简代码示例

python
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}

四、开发流程与基础组件 ​

核心概念

标准开发流程八步走:环境搭建 → 导入依赖 → 定义节点 → 创建图 → 加节点与边 → 设起止节点 → 运行 → 调试。基础组件即 Graph 创建、节点定义、边添加、图运行四件套。

生活类比

搭一张图就像用乐高拼轨道:先倒出零件(导入依赖)、造出一块块轨道段(定义节点)、拼接成完整线路(加边)、装上起点终点站牌(START/END),最后放上小车跑一圈看通不通(invoke 运行)。零件标准、接口统一,才能随时拆改扩建。

4.1 基础开发流程 ​

核心说明

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

完整可运行基础代码示例

python
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:路由键 -> 下一个节点

4.2 基础组件使用 ​

4.2.1 Graph 创建 ​

python
from typing import TypedDict
from langgraph.graph import StateGraph

class S(TypedDict):
    text: str

graph = StateGraph(S)

4.2.2 节点定义(基础节点 + 条件节点) ​

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

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

4.2.3 边添加(普通边 + 条件边) ​

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

4.2.4 图运行与状态查看 ​

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

五、节点进阶:工具/LLM/分支/循环节点 ​

核心概念

节点是图里真正「干活」的单元。常见有四类:工具调用节点(执行计算/查询)、LLM 节点(调用大模型生成)、分支节点(按状态决定走向)、循环节点(回环重试直到满足条件)。

生活类比

节点就像餐厅后厨的不同工位:切配工位(工具节点,处理原料)、炒锅工位(LLM 节点,核心加工)、传菜口的分流员(分支节点,看订单送不同桌)、以及重做工位(循环节点,菜品不合格退回重炒)。搭好工位分工,整个后厨才能高效运转。

5.1 节点进阶(工具调用节点、LLM 节点、分支节点、循环节点) ​

5.1.1 工具调用节点 ​

python
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"]}

5.1.2 LLM 节点(OpenAI 示例) ​

python
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),越低越稳定

5.1.3 分支节点与循环节点(对比) ​

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

六、边的进阶与状态管理 ​

核心概念

边决定「下一步去哪」:普通边固定流转、条件边按路由函数动态选择、循环边构成回环。而状态(State) 是全图的数据总线,节点通过返回 dict 增量更新它,设计时要分清「输入字段 / 中间字段 / 输出字段」。

生活类比

边就像导航软件的路线规划:普通边是「直行」,条件边是「前方拥堵,为你切换到备选路线」,循环边是「你已偏航,正在重新规划」。而状态就是车上的行车记录仪 + 油表——一路记录、实时更新,谁上车都能看到当前进度和剩余资源。

6.1 边的进阶用法 ​

6.1.1 条件边配置 ​

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

6.1.2 动态边思路 ​

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

6.1.3 多节点联动与循环边 ​

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

6.2 状态管理(核心重点) ​

核心说明

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

完整状态传递示例

python
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": ""}))

七、基础图结构:线性/分支/循环 ​

核心概念

三种最常用的图拓扑:线性图(固定顺序执行)、分支图(按条件走不同路径)、循环图(回环重试直到满足退出条件)。绝大多数复杂图都是这三者的组合。

生活类比

这三种图就像三种做事方式:线性图是「按菜谱一步步做」(洗菜 → 切菜 → 下锅),分支图是「看客人口味决定放不放辣」,循环图是「尝一口不够味就继续加盐,直到刚好」。先想清楚任务属于哪种,再动手搭图就不会乱。

三种图结构一图看懂

text
线性图:  START ──▶ A ──▶ B ──▶ END        固定顺序

分支图:            ┌──▶ A ──┐
          START ──▶ Router   ├──▶ END       条件路由
                    └──▶ B ──┘

循环图:  START ──▶ Check ──┬─(不满足)─▶ Retry ──┐
                            │                     │
                            │◀────────────────────┘
                            └─(满足)──▶ END        回环重试

7.1 基础图结构搭建(线性 / 分支 / 简单循环) ​

7.1.1 线性图 ​

text
START -> A -> B -> END

7.1.2 分支图 ​

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

7.1.3 简单循环图 ​

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

适用场景对比

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

八、复杂图结构与 Agent 集成 ​

核心概念

当业务不再是「一条直线」,就需要多条件分支、并行汇总、循环终止等复杂图结构;把「规划-执行-反馈」三个节点串起来,就构成了一个能自我推进的智能 Agent。

生活类比

复杂图就像项目管理流程:先由「组长」拆解任务(规划),再让「执行者」干活(执行),最后「验收员」检查(反馈)——每一环都是一个节点,环环相扣才能交付。

规划-执行-反馈一图看懂

text
           ┌──────────┐   ┌──────────┐   ┌──────────┐
START ──▶  │ planner  │─▶ │ executor │─▶ │ reviewer │ ──▶ END
           │ 拆解目标 │   │ 执行步骤 │   │ 验收交付 │
           └──────────┘   └──────────┘   └──────────┘

8.1 复杂图结构开发 ​

8.1.1 多条件分支图 ​

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

8.1.2 并行节点思路 ​

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

8.1.3 循环终止逻辑 ​

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

8.2 智能 Agent 与 LangGraph 结合(规划-执行-反馈) ​

python
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": ""}))

九、工具联动与调试可视化 ​

核心概念

节点不仅能跑 LLM,还能挂载「工具」执行真实动作(计算、查询、向量检索);当图跑不动时,用流式运行 + 状态打印,就能像看监控回放一样定位卡点。

生活类比

工具联动像给厨师配齐厨房家电:会算账的(Tool)、能翻菜谱的(VectorStore)各司其职;调试可视化则像行车记录仪,把每一步执行都录下来,出问题时倒回去看哪一帧出了错。

9.1 工具联动与节点集成(Tool + Embedding + VectorStore) ​

python
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 适合知识检索节点

9.2 图调试与可视化 ​

调试要点

  • 打印每个节点输入输出状态
  • 给关键节点加 try/except 和错误字段写回 state
  • 使用流式运行观察执行路径
python
events = app.stream({"goal": "测试目标", "plan": "", "output": "", "review": ""})
for event in events:
    print(event)
# 效果说明:逐步输出节点执行事件,便于定位卡点

十、实战场景:对话机器人与任务规划 ​

核心概念

把前面学的分支、状态、条件边组合起来,就能做出真实业务:能识别意图的多轮对话机器人、能拆解目标的任务规划流水线。

生活类比

对话机器人像前台接待员:先听你说啥(识别意图),再把你转接到对应窗口(订单/退款/咨询);任务规划像写论文流程:先列大纲,再写初稿,最后润色成终稿,一步都不能跳。

10.1 复杂多轮对话机器人(分支 + 记忆) ​

python
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": ""}))

10.2 任务规划与执行(论文写作/数据分析拆解) ​

python
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": ""}))

十一、本地部署、FastAPI 与容错 ​

核心概念

写好的图要能上线:可以接本地 LLM 保护隐私、用 FastAPI 包成 HTTP 接口对外服务,再加上重试与 fallback 保证系统稳定不崩。

生活类比

本地部署像自建厨房(数据不出门),FastAPI 像开了个外卖窗口(对外提供服务),容错机制像备用电源:主流程一出故障,立刻切到降级方案,客人感知不到后厨着火了。

11.1 本地化部署(本地 LLM 示例思路) ​

python
# 以本地模型网关为例(如 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"),
)

场景解读

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

11.2 FastAPI 接口封装 ​

python
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

11.3 错误处理与容错机制(重试 + fallback) ​

python
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)}

注意事项

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

十二、生态适配与性能优化 ​

核心概念

LangGraph 不是要取代 LangChain,而是分工协作:LangChain 提供「能力组件」,LangGraph 负责「流程编排」;上线前还要做性能优化,减少多余节点、缓存稳定结果、控制 LLM 调用。

生活类比

LangChain 像乐高零件盒(提供各种积木),LangGraph 像拼装说明书(决定怎么拼);性能优化则像给流水线提速:去掉多余工序、把重复劳动缓存下来、别让贵的机器(LLM)空转。

12.1 LangGraph 与 LangChain 生态适配 ​

核心说明

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

对比代码示例(简化)

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

12.2 性能优化 ​

优化策略

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

批量处理示例

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

十三、自定义组件与部署上线 ​

核心概念

把重试次数、超时、路由策略抽象成统一配置,就能复用到多个图;再用 Docker 打包 + 日志监控 + 版本回滚,让服务稳稳跑在生产环境。

生活类比

自定义组件像标准化零件:一次做好、到处复用,不用每次重造轮子;部署上线像开一家连锁店:装修标准(Docker)、装监控摄像头(日志)、留好上一版菜单(回滚),出问题随时切回去。

13.1 自定义组件开发(节点 / 边 / 状态管理器) ​

python
from dataclasses import dataclass

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

config = CustomConfig()
print(config)

场景说明

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

13.2 部署与上线(Docker + 监控) ​

dockerfile
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
  • 监控:节点失败率、平均时延、成本指标
  • 回滚:保留稳定版图配置,支持快速切换

十四、核心工具与资源:参数手册、实战模板、学习资源 ​

核心概念

本章是「速查手册」:常用 API 参数一览、三套可直接复用的图模板、以及权威学习资源,需要时翻这里即可。

生活类比

这一章就像工具箱里的说明卡:每个扳手(API)的用法都印在旁边,还配了三张常见拼装图纸(模板),拿来即用,不用每次都从头琢磨。

14.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)

  • 流式返回节点执行事件

14.2 优质实战模板(可直接复用) ​

模板 1:线性处理模板

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

模板 2:意图分发模板

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

模板 3:重试回环模板

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

14.3 学习资源推荐 ​


十五、常见问题与实战案例汇总 ​

核心概念

本章分两部分:先列出高频报错与避坑技巧(帮你少踩坑),再汇总 5 个完整实战案例(帮你把知识串成体系)。

生活类比

前半部分像驾校教练的错题本:把新手最容易翻车的地方一条条标出来;后半部分像样板间:把家具都摆好给你看,照着装修就行。

15.1 高频问题与解决方案 ​

问题 1:图运行卡死(循环不退出) ​

错误代码

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

修正代码

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

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

问题 2:节点执行失败(返回类型错误) ​

错误代码

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

修正代码

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

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

问题 3:条件分支不生效 ​

常见原因

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

15.2 高级避坑技巧 ​

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

15.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 编排 + 本地工具节点
效果说明:满足隐私和离线运行要求。


十六、运行环境、依赖与总结 ​

核心概念

最后一章给出推荐运行环境、一键安装命令与 API 配置方法,并用「关键结论」把全篇的核心思想收束成一句话:用图结构管理复杂任务流。

生活类比

这一章像产品说明书的最后一页:先列清楚「运行需要什么配件」(环境依赖),再用一句话点题「这台机器到底是干嘛的」(总结),让你合上手册时心里有底。

16.1 推荐环境 ​

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

16.2 一键安装命令 ​

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

16.3 API 配置方法 ​

bash
# Windows PowerShell
$env:OPENAI_API_KEY="your_api_key"
bash
# Linux / macOS
export OPENAI_API_KEY="your_api_key"

16.4 总结 ​

关键结论

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