06-LangGraph代码编排RAG：状态机与图编排实现可定制化RAG流程

第 6 篇：LangGraph 代码编排 RAG——状态机与图编排实现可定制化 RAG 流程

本文为「从零到落地：RAG 检索增强生成实战系列」第 6 篇，完整系列持续更新中。

---

前言

本篇基于 LangGraph 的状态机 + 图编排能力，搭建可深度定制的 RAG 流程：从基础检索-生成双节点，到多轮对话记忆、条件路由、Query 改写、RAG + Agent 融合，完整代码逐行讲解。

在第 5 篇里，你用 Dify 社区版零代码搭建了一个完整的 RAG 知识库，点几下鼠标就能跑通。但你一定也感受到了 Dify 的边界——想加一个”检索结果不满意就自动改写 Query 重试”的分支逻辑？想做多路并行检索再融合排序？想把 RAG 和 Agent 工具调用深度结合？这些在可视化界面上很难直接实现。

这就是 LangGraph 的价值所在。

LangGraph 是 LangChain 团队推出的状态机图编排框架，核心理念是把 AI 应用拆解为一个个节点（Node），用边（Edge） 定义节点之间的流转逻辑，所有节点共享一个状态（State）。你可以像搭积木一样，把检索、生成、改写、重排、工具调用等能力自由组合，构建出任意复杂度的 RAG 流程——而且每一步的输入输出都清晰可见，出了问题能精准定位。

本篇学完你将掌握：

• LangGraph 核心概念：State、Node、Edge、条件路由
• 用 LangGraph 搭建基础 RAG（检索节点 + 生成节点）
• 接入对话记忆，实现多轮连续对话 RAG
• 进阶改造：条件路由 + Query 改写 + 检索质量自动判断
• RAG + Agent 融合：让 Agent 决定是否调用外部工具
• LangGraph 与 Dify 的取舍对比

---

一、LangGraph 是什么

1.1 核心定位

LangGraph 是一个基于状态机的图编排框架，专门用于构建有状态的、多步骤的 AI 应用。和 LangChain 的关系是：LangChain 提供基础组件（LLM 调用、文档加载、向量检索等），LangGraph 在这些组件之上提供流程编排和状态管理能力。

一个直观的类比：

• LangChain = 一盒乐高积木（各种零件）
• LangGraph = 乐高图纸 + 拼装逻辑（把零件按特定流程组合起来）
• Dify = 已经拼好的乐高成品（开箱即用，但改造受限）

1.2 四个核心概念

LangGraph 的核心只有四个概念，搞懂这四个就能开始写代码：

概念	说明	类比
State（状态）	所有节点共享的数据容器，用 TypedDict 定义	流水线上的”工件”，每个工序加工后放回工件
Node（节点）	一个处理函数，接收 State，返回 State 的更新	流水线上的”工位”，每个工位做一件事
Edge（边）	节点之间的连接，定义执行顺序	流水线的”传送带”，把工件从一个工位送到下一个
Conditional Edge（条件边）	根据状态动态决定下一个节点	流水线上的”质检站”，合格走 A 路，不合格走 B 路

用户提问
    ↓
┌─────────┐     ┌─────────┐     ┌─────────┐
│ 检索节点  │────▶│ 生成节点  │────▶│  输出    │
│(retrieve)│     │(generate)│     │ (END)   │
└─────────┘     └─────────┘     └─────────┘
    ↑                               │
    └───────── 下一轮提问 ────────────┘

核心价值：LangGraph 让你把 RAG 流程拆成独立的、可测试的节点，然后用代码精确定义它们之间的流转逻辑。每一步做了什么、输入是什么、输出是什么，全部透明可控——这是 Dify 这类低代码平台做不到的。

1.3 LangGraph vs Dify：一句话总结

• Dify：可视化拖拽，快速上线，适合”够用就行”的场景
• LangGraph：代码编排，深度定制，适合”需要精确控制每一步”的场景

详细对比在第八节。

---

二、环境准备

2.1 技术栈

组件	选择	理由
Python	3.11	系列统一版本
图编排框架	LangGraph	LangChain 官方出品，状态机 + 图编排
LLM 接口	langchain-openai	兼容 DeepSeek API，也支持 OpenAI
Embedding 模型	bge-large-zh-v1.5（本地）	中文效果最好的开源模型
向量数据库	Chroma	与第 4 篇一致，零配置启动

2.2 安装依赖

# 创建虚拟环境（推荐）
python -m venv langgraph-rag
langgraph-rag\Scripts\activate    # Windows
# source langgraph-rag/bin/activate  # macOS/Linux

# 安装依赖包
pip install langgraph langchain-core langchain-openai sentence-transformers chromadb

💡 提示：langchain-openai 提供了兼容 OpenAI 接口的 ChatModel，DeepSeek、通义千问等有 OpenAI 兼容 API 的模型都可以通过修改 base_url 接入。

2.3 准备测试文档

复用第 4 篇的测试文档 docs/company_faq.md：

# 公司常见问题

## 年假制度
入职满一年后享受 5 天带薪年假，满三年后 8 天，满五年后 10 天。年假需提前 3 天在 OA 系统提交申请，由直属领导审批。未休完的年假可在次年 3 月底前补休，过期作废。

## 报销流程
差旅报销需在出差结束后 5 个工作日内提交，附上发票原件和出差审批单。单笔报销金额超过 5000 元需要部门总监审批。报销款一般在提交后 10 个工作日内打到工资卡。

## 技术栈
公司后端主要使用 Python 和 Go，前端使用 React + TypeScript。数据库以 PostgreSQL 为主，缓存用 Redis，消息队列用 RabbitMQ。部署在阿里云 K8s 集群上，CI/CD 用 GitLab CI。

## 产品架构
核心产品分为三个模块：数据采集层（支持 API、SDK、文件上传三种接入方式）、数据处理层（实时流处理 + 离线批处理双引擎）、数据展示层（可视化看板 + 报表导出）。

2.4 构建向量索引

在开始写 LangGraph 之前，先把文档向量化并存入 Chroma。这段代码和第 4 篇类似：

"""
构建向量索引：加载文档 → 分块 → 向量化 → 存入 Chroma
后续 LangGraph 的检索节点会直接使用这个索引
"""
import os
from sentence_transformers import SentenceTransformer
import chromadb

# 加载 Embedding 模型
embedding_model = SentenceTransformer("BAAI/bge-large-zh-v1.5")

# 加载文档并分块
def load_and_chunk(docs_dir: str, chunk_size: int = 300, overlap: int = 50) -> list[str]:
    """加载目录下所有 .md 文件并切分为文本块"""
    all_chunks = []
    for filename in os.listdir(docs_dir):
        if filename.endswith(".md"):
            with open(os.path.join(docs_dir, filename), "r", encoding="utf-8") as f:
                text = f.read()
            start = 0
            while start < len(text):
                all_chunks.append(text[start:start + chunk_size])
                start += chunk_size - overlap
    return all_chunks

chunks = load_and_chunk("docs")

# 存入 Chroma
client = chromadb.Client()
collection = client.create_collection(name="company_docs")
embeddings = embedding_model.encode(chunks, normalize_embeddings=True)
collection.add(
    documents=chunks,
    embeddings=embeddings.tolist(),
    ids=[f"chunk_{i}" for i in range(len(chunks))]
)
print(f"已索引 {collection.count()} 个文本块")

---

三、核心概念详解

在写 RAG 图之前，先用一个最简示例帮你理解 LangGraph 的运行机制。

3.1 定义 State

State 是所有节点共享的数据容器，用 TypedDict 定义：

from typing import TypedDict

class SimpleState(TypedDict):
    input: str       # 用户输入
    output: str      # 处理结果
    step_count: int  # 经过了多少个节点

3.2 定义 Node

每个 Node 是一个函数，接收当前 State，返回要更新的字段：

def node_a(state: SimpleState) -> dict:
    """节点 A：把输入转为大写"""
    return {"output": state["input"].upper(), "step_count": 1}

def node_b(state: SimpleState) -> dict:
    """节点 B：在结果后面加上感叹号"""
    return {"output": state["output"] + "!", "step_count": state["step_count"] + 1}

3.3 构建图并运行

from langgraph.graph import StateGraph, START, END

# 创建图，指定 State 类型
graph = StateGraph(SimpleState)

# 添加节点
graph.add_node("node_a", node_a)
graph.add_node("node_b", node_b)

# 添加边：START → node_a → node_b → END
graph.add_edge(START, "node_a")
graph.add_edge("node_a", "node_b")
graph.add_edge("node_b", END)

# 编译图
app = graph.compile()

# 运行
result = app.invoke({"input": "hello world", "output": "", "step_count": 0})
print(result["output"])   # 输出: HELLO WORLD!
print(result["step_count"])  # 输出: 2

执行过程：

START → node_a（"hello world" → "HELLO WORLD"）→ node_b（"HELLO WORLD" → "HELLO WORLD!"）→ END

💡 提示：LangGraph 的每个节点只返回它要更新的字段，而不是完整的 State。框架会自动把返回值合并到当前 State 中。这个设计让节点之间的数据传递非常清晰——你只需要关心”我这个节点改了什么”。

---

四、搭建基础 RAG 图

现在用 LangGraph 搭建一个完整的 RAG 流程，核心是两个节点：检索节点和生成节点。

START → retrieve（检索）→ generate（生成）→ END

4.1 定义 RAG State

from typing import TypedDict

class RAGState(TypedDict):
    question: str          # 用户问题
    context: list[str]     # 检索到的文本块
    answer: str            # LLM 生成的回答

4.2 检索节点

def retrieve(state: RAGState) -> dict:
    """
    检索节点：把用户问题转为向量，在 Chroma 中检索最相关的文本块
    """
    question = state["question"]

    # 问题向量化（必须用和索引时同一个 Embedding 模型）
    query_embedding = embedding_model.encode([question], normalize_embeddings=True)

    # 在 Chroma 中检索 Top-3
    results = collection.query(
        query_embeddings=query_embedding.tolist(),
        n_results=3,
        include=["documents", "distances"]
    )

    # 提取检索到的文本块
    retrieved_docs = results["documents"][0]
    distances = results["distances"][0]

    print(f"  [检索] 问题: {question}")
    for i, (doc, dist) in enumerate(zip(retrieved_docs, distances)):
        print(f"  [Top-{i+1}] 距离: {dist:.4f} | {doc[:60]}...")

    return {"context": retrieved_docs}

4.3 生成节点

from langchain_openai import ChatOpenAI

# 初始化 LLM（DeepSeek 兼容 OpenAI 接口）
llm = ChatOpenAI(
    model="deepseek-chat",
    api_key="sk-xxx",                    # 替换为你的 API Key
    base_url="https://api.deepseek.com",
    temperature=0.3
)

def generate(state: RAGState) -> dict:
    """
    生成节点：把检索到的上下文和用户问题拼接成 Prompt，调用 LLM 生成回答
    """
    from langchain_core.messages import SystemMessage, HumanMessage

    # 拼接检索结果为上下文
    context_text = "\n\n".join([f"[{i+1}] {doc}" for i, doc in enumerate(state["context"])])

    # 构造消息
    messages = [
        SystemMessage(content=(
            "你是一个专业的企业知识库助手。请严格根据提供的参考资料回答用户问题。"
            "如果资料中没有相关内容，请明确告知'根据现有资料无法回答该问题'。"
            "回答简洁准确，必要时引用参考资料编号。"
        )),
        HumanMessage(content=f"## 参考资料\n{context_text}\n\n## 用户问题\n{state['question']}")
    ]

    # 调用 LLM
    response = llm.invoke(messages)

    print(f"  [生成] 回答: {response.content[:80]}...")

    return {"answer": response.content}

4.4 构建并运行 RAG 图

from langgraph.graph import StateGraph, START, END

# 构建图
rag_graph = StateGraph(RAGState)

# 添加节点
rag_graph.add_node("retrieve", retrieve)
rag_graph.add_node("generate", generate)

# 添加边：START → retrieve → generate → END
rag_graph.add_edge(START, "retrieve")
rag_graph.add_edge("retrieve", "generate")
rag_graph.add_edge("generate", END)

# 编译
rag_app = rag_graph.compile()

# 运行
result = rag_app.invoke({
    "question": "公司年假有几天？",
    "context": [],
    "answer": ""
})

print(f"\n最终回答: {result['answer']}")

运行结果：

  [检索] 问题: 公司年假有几天？
  [Top-1] 距离: 0.2103 | 入职满一年后享受 5 天带薪年假，满三年后 8 天...
  [Top-2] 距离: 0.5841 | 差旅报销需在出差结束后 5 个工作日内提交...
  [Top-3] 距离: 0.6723 | 核心产品分为三个模块：数据采集层...
  [生成] 回答: 根据公司年假制度：入职满一年享受5天带薪年假，满三年8天，满五年10天...

最终回答: 根据公司年假制度：
- 入职满一年：5 天带薪年假
- 入职满三年：8 天带薪年假
- 入职满五年：10 天带薪年假

核心价值：到这里你已经用 LangGraph 跑通了基础 RAG。和原生手写相比，检索和生成被拆成了独立的节点——你可以单独测试检索节点的效果，单独优化生成节点的 Prompt，甚至替换其中一个节点而不影响另一个。这就是图编排的价值。

---

五、接入对话记忆：多轮 RAG

基础 RAG 每次提问都是独立的，不支持连续追问。LangGraph 内置了 Checkpointer 机制，可以自动保存和恢复对话状态，实现多轮对话。

5.1 核心思路

• 用 MemorySaver 作为 Checkpointer，把每轮对话的 State 保存到内存
• 每次调用时传入 thread_id，LangGraph 会自动恢复该线程的历史状态
• 在 Prompt 中加入历史对话，让 LLM 理解上下文

5.2 改造 State 和节点

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, BaseMessage
import operator

class ChatRAGState(TypedDict):
    question: str                              # 当前用户问题
    context: list[str]                         # 检索到的文本块
    messages: Annotated[list[BaseMessage], operator.add]  # 对话历史（自动追加）

💡 提示：Annotated[list, operator.add] 是 LangGraph 的关键语法——它告诉框架”这个字段是列表类型，每次节点返回的值应该追加到现有列表，而不是替换”。这就是多轮对话记忆的核心机制。

5.3 改造生成节点（加入对话历史）

def chat_generate(state: ChatRAGState) -> dict:
    """生成节点：结合对话历史和检索上下文生成回答"""
    # 拼接检索结果
    context_text = "\n\n".join([f"[{i+1}] {doc}" for i, doc in enumerate(state["context"])])

    # 构造系统消息
    system_msg = SystemMessage(content=(
        "你是一个专业的企业知识库助手。请严格根据提供的参考资料回答用户问题。"
        "如果资料中没有相关内容，请告知'根据现有资料无法回答'。回答简洁准确。"
    ))

    # 把检索上下文作为一条系统消息注入
    context_msg = HumanMessage(content=f"[参考资料]\n{context_text}")

    # 当前用户问题
    user_msg = HumanMessage(content=state["question"])

    # 组合消息：系统消息 + 历史对话 + 上下文 + 当前问题
    all_messages = [system_msg] + state["messages"] + [context_msg, user_msg]

    response = llm.invoke(all_messages)

    # 返回新消息（会自动追加到 messages 列表）
    return {
        "messages": [user_msg, AIMessage(content=response.content)],
        "context": []  # 清空上下文，避免下一轮重复
    }

5.4 构建带记忆的 RAG 图

# 构建图
chat_graph = StateGraph(ChatRAGState)
chat_graph.add_node("retrieve", retrieve)     # 复用之前的检索节点
chat_graph.add_node("generate", chat_generate)
chat_graph.add_edge(START, "retrieve")
chat_graph.add_edge("retrieve", "generate")
chat_graph.add_edge("generate", END)

# 编译时传入 MemorySaver（关键！）
memory = MemorySaver()
chat_app = chat_graph.compile(checkpointer=memory)

5.5 测试多轮对话

# 定义线程配置（同一个 thread_id = 同一个对话）
config = {"configurable": {"thread_id": "test-001"}}

# 第一轮
result1 = chat_app.invoke(
    {"question": "公司年假有几天？", "context": [], "messages": []},
    config=config
)
print(f"AI: {result1['messages'][-1].content}\n")

# 第二轮（追问，AI 知道"那"指的是年假）
result2 = chat_app.invoke(
    {"question": "那怎么申请？", "context": [], "messages": []},
    config=config
)
print(f"AI: {result2['messages'][-1].content}\n")

# 第三轮
result3 = chat_app.invoke(
    {"question": "没休完怎么办？", "context": [], "messages": []},
    config=config
)
print(f"AI: {result3['messages'][-1].content}")

运行结果：

AI: 根据公司年假制度：入职满一年5天，满三年8天，满五年10天。

AI: 年假需提前3天在OA系统提交申请，由直属领导审批。

AI: 未休完的年假可在次年3月底前补休，过期作废。

三轮对话共享同一个 thread_id，LangGraph 自动维护了对话历史，AI 能理解每轮追问的上下文。

💡 提示：如果需要持久化记忆（重启不丢失），可以把 MemorySaver() 替换为 SqliteSaver 或 PostgresSaver，用法完全一样，只是存储后端不同。

---

六、进阶改造：条件路由与 Query 改写

这是 LangGraph 真正发光的场景——你可以用条件边（Conditional Edge） 实现动态分支逻辑，让 RAG 流程具备”判断”和”自我修正”的能力。

6.1 目标流程

START → retrieve → grade（质量评估）
                      │
                      ├─ 检索质量好 → generate → END
                      │
                      └─ 检索质量差 → rewrite（改写问题）→ retrieve（重新检索）→ ...

当检索结果不够相关时，系统会自动改写用户问题，用新的问题重新检索，而不是一股脑把噪声喂给 LLM。

6.2 改造 State

class AdvancedRAGState(TypedDict):
    question: str              # 当前问题（可能被改写）
    original_question: str     # 原始问题（保持不变）
    context: list[str]         # 检索到的文本块
    answer: str                # 最终回答
    retry_count: int           # 重试次数（防止无限循环）

6.3 质量评估节点

def grade_retrieval(state: AdvancedRAGState) -> dict:
    """
    评估检索质量：用 LLM 判断检索结果是否与问题相关
    返回 quality: "good" 或 "bad"
    """
    context_text = "\n".join(state["context"][:2])  # 只看前两个结果

    messages = [
        SystemMessage(content=(
            "你是一个检索质量评估专家。判断以下参考资料是否与用户问题相关。"
            "只回答 'good'（相关）或 'bad'（不相关），不要解释。"
        )),
        HumanMessage(content=f"用户问题：{state['question']}\n\n参考资料：\n{context_text}")
    ]

    response = llm.invoke(messages)
    quality = "good" if "good" in response.content.lower() else "bad"

    print(f"  [评估] 检索质量: {quality}")
    return {"quality": quality}  # 临时存入 state

6.4 Query 改写节点

def rewrite_query(state: AdvancedRAGState) -> dict:
    """
    改写用户问题：让 LLM 重新表述问题，提升检索命中率
    """
    messages = [
        SystemMessage(content=(
            "你是一个搜索查询优化专家。将用户问题改写为更适合知识库检索的形式。"
            "只输出改写后的问题，不要解释。"
        )),
        HumanMessage(content=f"原始问题：{state['original_question']}\n当前问题：{state['question']}")
    ]

    response = llm.invoke(messages)
    new_question = response.content.strip()

    print(f"  [改写] {state['question']} → {new_question}")
    return {
        "question": new_question,
        "retry_count": state["retry_count"] + 1
    }

6.5 条件路由函数

def should_continue(state: AdvancedRAGState) -> str:
    """
    条件路由：根据检索质量和重试次数决定下一步
    """
    # 最多重试 1 次，防止无限循环
    if state.get("retry_count", 0) >= 1:
        return "generate"

    # 根据评估结果决定走哪条路
    if state.get("quality") == "good":
        return "generate"
    else:
        return "rewrite"

6.6 构建高级 RAG 图

# 构建图
adv_graph = StateGraph(AdvancedRAGState)

# 添加节点
adv_graph.add_node("retrieve", retrieve)        # 检索节点（复用）
adv_graph.add_node("grade", grade_retrieval)     # 质量评估
adv_graph.add_node("rewrite", rewrite_query)     # Query 改写
adv_graph.add_node("generate", generate)         # 生成节点（复用）

# 添加边
adv_graph.add_edge(START, "retrieve")
adv_graph.add_edge("retrieve", "grade")

# 条件边：根据 grade 的结果路由到不同节点
adv_graph.add_conditional_edges(
    "grade",
    should_continue,
    {"generate": "generate", "rewrite": "rewrite"}
)

# 改写后重新检索
adv_graph.add_edge("rewrite", "retrieve")
adv_graph.add_edge("generate", END)

adv_app = adv_graph.compile()

运行测试：

# 测试一个模糊的问题，观察系统是否会自动改写
result = adv_app.invoke({
    "question": "请假",
    "original_question": "请假",
    "context": [],
    "answer": "",
    "retry_count": 0
})
print(f"\n最终回答: {result['answer']}")

可能的输出：

[检索] 问题: 请假
[评估] 检索质量: bad
[改写] 请假 → 公司员工请假制度和年假规定是什么？
[检索] 问题: 公司员工请假制度和年假规定是什么？
[评估] 检索质量: good
[生成] 回答: 根据公司年假制度...

系统检测到”请假”太模糊、检索质量差，自动改写为更具体的问题后重新检索，最终获得了高质量的回答。 这种”自我修正”的能力在 Dify 的可视化界面中很难直接实现。

---

七、RAG + Agent 融合：工具调用

LangGraph 还支持把 RAG 和 Agent 能力结合起来——让 LLM 自己决定是”查知识库”还是”调用外部工具”。

7.1 思路

给 LLM 绑定两个工具：一个”查知识库”工具（就是我们的 RAG 检索），一个”查天气”工具。LLM 会根据用户问题自动选择调用哪个工具。

7.2 定义工具

from langchain_core.tools import tool

@tool
def search_knowledge_base(query: str) -> str:
    """从公司知识库中检索信息。适用于公司内部制度、技术、产品相关的问题。"""
    query_embedding = embedding_model.encode([query], normalize_embeddings=True)
    results = collection.query(
        query_embeddings=query_embedding.tolist(),
        n_results=3,
        include=["documents"]
    )
    docs = results["documents"][0]
    return "\n\n".join(docs)

@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气信息。"""
    # 这里用模拟数据，实际可接入真实天气 API
    weather_data = {
        "北京": "晴，25°C",
        "上海": "多云，22°C",
        "广州": "小雨，28°C"
    }
    return weather_data.get(city, f"{city}：暂无天气数据")

7.3 构建 Agent RAG 图

from langchain_core.messages import SystemMessage, HumanMessage

# 给 LLM 绑定工具
llm_with_tools = llm.bind_tools([search_knowledge_base, get_weather])

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], operator.add]

def agent_node(state: AgentState) -> dict:
    """Agent 节点：LLM 决定下一步动作"""
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

def tool_node(state: AgentState) -> dict:
    """工具执行节点：执行 LLM 选择的工具"""
    from langchain_core.messages import ToolMessage

    last_message = state["messages"][-1]
    results = []

    for tool_call in last_message.tool_calls:
        # 根据工具名执行对应函数
        if tool_call["name"] == "search_knowledge_base":
            result = search_knowledge_base.invoke(tool_call["args"])
        elif tool_call["name"] == "get_weather":
            result = get_weather.invoke(tool_call["args"])
        else:
            result = "未知工具"

        results.append(ToolMessage(content=str(result), tool_call_id=tool_call["id"]))

    return {"messages": results}

def should_use_tools(state: AgentState) -> str:
    """判断 LLM 是否要调用工具"""
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    return "end"

# 构建 Agent 图
agent_graph = StateGraph(AgentState)
agent_graph.add_node("agent", agent_node)
agent_graph.add_node("tools", tool_node)

agent_graph.add_edge(START, "agent")
agent_graph.add_conditional_edges("agent", should_use_tools, {"tools": "tools", "end": END})
agent_graph.add_edge("tools", "agent")  # 工具执行后回到 Agent 继续推理

agent_app = agent_graph.compile()

7.4 测试 Agent RAG

# 测试 1：知识库问题
result = agent_app.invoke({
    "messages": [
        SystemMessage(content="你是公司助手，可以查询知识库或查天气。"),
        HumanMessage(content="公司用什么技术栈？")
    ]
})
print(f"AI: {result['messages'][-1].content}\n")

# 测试 2：天气问题
result = agent_app.invoke({
    "messages": [
        SystemMessage(content="你是公司助手，可以查询知识库或查天气。"),
        HumanMessage(content="北京天气怎么样？")
    ]
})
print(f"AI: {result['messages'][-1].content}")

运行结果：

AI: 公司后端主要使用 Python 和 Go，前端使用 React + TypeScript。数据库以 PostgreSQL 为主...

AI: 北京今天的天气是晴，25°C。

Agent 自动判断了问题类型——技术问题走知识库检索，天气问题走天气工具——整个过程不需要手动写条件分支。

💡 提示：这里只是最简单的 Agent + RAG 融合示例。实际场景中，你可以给 Agent 绑定更多工具（数据库查询、API 调用、代码执行等），让它成为一个”全能助手”——知识库能回答的就查知识库，不能回答的就调用其他工具，灵活度远超 Dify 的固定流程。

踩坑记录：LangGraph 版本兼容性

现象：

安装 langgraph 后运行报错：

ImportError: cannot import name 'StateGraph' from 'langgraph.graph'

或者 add_conditional_edges 的参数格式报错。

原因：

LangGraph 在 0.x 到 1.x 之间有较大的 API 变更。StateGraph、START、END 的导入路径和 add_conditional_edges 的参数格式在不同版本之间可能不一致。

解决方案：

# 锁定兼容版本
pip install langgraph>=0.2.0 langchain-core>=0.3.0 langchain-openai>=0.2.0

避坑建议：

安装时锁定版本号，在 requirements.txt 中明确记录。LangGraph 更新频率很高，升级前先查看 changelog，确认 API 兼容性。

---

八、LangGraph vs Dify：怎么选

对比维度	LangGraph	Dify
核心定位	代码编排框架，深度定制 RAG 流程	低代码 AI 应用平台，快速搭建
上手难度	需要 Python 基础 + 理解状态机概念	零代码，可视化操作
定制灵活度	⭐⭐⭐⭐⭐ 完全自由	⭐⭐⭐ 受限于平台能力
条件分支	任意复杂的条件路由、循环、自我修正	基础条件分支（Chatflow 有 IF/ELSE 节点）
多路检索融合	自由实现（并行检索、RRF 融合排序等）	不直接支持，需借助工具节点间接实现
Agent 融合	原生支持，工具绑定即用	支持 Agent 节点，但深度定制受限
对话记忆	Checkpointer 机制，支持多种存储后端	内置自动管理，配置简单
可视化调试	LangSmith 付费平台，或自己打日志	内置 Web 界面，实时预览
部署运维	自己管理代码和依赖	Docker 一键部署，运维成本低
团队协作	代码仓库协作，适合开发团队	Web 界面协作，非技术人员也能参与
适用人群	后端开发、AI 工程师	产品经理、运维、全栈开发

选型建议：

场景	推荐方案	理由
快速上线企业内部知识库	Dify	零代码部署，30 分钟搞定
需要深度定制检索策略	LangGraph	条件路由、多路检索、自我修正随便写
非技术团队也要参与配置	Dify	Web 界面对非技术人员友好
RAG + Agent 复杂业务	LangGraph	原生工具调用，灵活编排
快速原型验证	Dify	先验证想法，再用 LangGraph 深度开发
两者结合	Dify 做前端 + LangGraph 做后端	Dify 管界面和基础流程，LangGraph 管复杂逻辑

核心价值：Dify 和 LangGraph 不是替代关系，而是互补关系。Dify 解决”快速上线”的问题，LangGraph 解决”深度定制”的问题。 很多团队的最佳实践是：先用 Dify 快速验证想法和效果，确认 RAG 方案可行后，再用 LangGraph 重写核心流程，获得更高的定制自由度。

---

常见问题踩坑汇总

Q：LangGraph 的节点执行顺序怎么调试？
A：编译图后调用 app.get_graph().print_ascii() 可以打印图的拓扑结构。每个节点内部加 print 日志是最直接的调试方式。如果需要更专业的可视化，可以使用 LangSmith 平台（付费）。

Q：多轮对话的 MemorySaver 重启后丢失怎么办？
A：MemorySaver 是内存存储，进程结束即丢失。生产环境替换为持久化存储：from langgraph.checkpoint.sqlite import SqliteSaver，用法完全一样，只是构造函数传入数据库路径。

Q：条件路由出现无限循环怎么排查？
A：在 State 中加一个计数器字段（如 retry_count），条件路由函数中检查计数器上限。本篇的 should_continue 函数就是这样做的——最多重试 1 次，超过就强制走生成节点。

Q：Annotated[list, operator.add] 是什么意思？
A：这是 LangGraph 的 Reducer 语法。默认情况下，节点返回的字典会覆盖 State 中同名字段的值。但如果字段类型是 Annotated[list, operator.add]，节点返回的列表会追加到现有列表而不是覆盖。对话历史就需要这个特性——每轮新消息追加到历史列表，而不是替换。

Q：LangGraph 和 LangChain 是什么关系？必须一起用吗？
A：LangGraph 是 LangChain 团队出品的独立框架，可以单独使用。本篇用了 langchain-openai 和 langchain-core 是因为它们提供了方便的 LLM 调用和消息类型定义，但 LangGraph 的图编排能力本身不依赖 LangChain。你也可以用原生 OpenAI SDK 替代，只是代码量会多一些。

---

总结与回顾

知识点	关键要点
LangGraph 定位	状态机图编排框架，把 RAG 拆成独立节点自由组合
核心概念	State（共享状态）、Node（处理函数）、Edge（连接）、Conditional Edge（条件路由）
基础 RAG	retrieve 节点 + generate 节点，START → 检索 → 生成 → END
多轮对话	MemorySaver + thread_id，Annotated[list, operator.add] 实现消息追加
条件路由	质量评估节点 → 好则生成、差则改写重试，实现”自我修正”
Query 改写	用 LLM 改写模糊问题为具体查询，提升检索命中率
Agent 融合	bind_tools 绑定知识库和外部工具，LLM 自主选择调用策略
与 Dify 对比	Dify 快速上线，LangGraph 深度定制，两者互补而非替代

---

下篇预告

第 7 篇：LlamaIndex 实现 RAG —— LlamaIndex 是专为 RAG 设计的框架，API 比 LangGraph 更简洁，开箱即用。如果你不需要 LangGraph 的复杂编排能力，LlamaIndex 可能是更高效的选择。下一篇完整实战 LlamaIndex 的文档加载、节点切分、索引构建、查询引擎，末尾与 LangGraph 做框架选型对比。

---

参考资料

资源	链接
LangGraph 官方文档	langchain-ai.github.io/langgraph
LangGraph GitHub	langchain-ai/langgraph
LangChain 官方文档	python.langchain.com
DeepSeek 开放平台	platform.deepseek.com
Chroma 向量数据库	trychroma.com
bge-large-zh-v1.5	HuggingFace 模型页

---

本文为「从零到落地：RAG 检索增强生成实战系列」第 6 篇，完整系列持续更新中。