LangChain： LEDVR工作流详解

LangChain 1.0+ 官方文档和路线图中没有直接定义或描述“LEDVR” 这个术语。

目前 LEDVR 这个词主要出现在一些中文社区博客和技术文章中，是对RAG（检索增强生成）流程节点的概述。而官方的英文术语中，一般用 RAG（检索增强生成）来描述从文档加载到向量检索的端到端数据预处理流程。

核心概念

在LangChain开发体系中，”LEDVR”实际上指向两个不同的核心工作流概念，它们服务于不同的开发层次：

数据处理工作流

数据处理工作流：L→E→D→V→R，这是最常被提及的LEDVR定义——一个用于构建RAG（检索增强生成）知识库的标准化数据处理流程：

步骤	字母	含义	LangChain 1.0+ 核心组件
1	L	Load（加载数据）	PyPDFLoader、WebBaseLoader、TextLoader等
2	E	Extract（提取内容）	由加载器内部完成， 从原始文件中提取纯文本
3	D	Divide（分割文档）	RecursiveCharacterTextSplitter （LangChain 1.0+仍广泛使用）
4	V	Vectorize（向量化）	OpenAIEmbeddings、 HuggingFaceEmbeddings等嵌入模型
5	R	Retrieve（检索）	Chroma、FAISS、Pinecone 等向量数据库 + as_retriever()

数据增强检索

数据增强检索：Loader → Embedding → Document Transformers → VectorStore → Retriever

此视角下的LEDVR同样是LangChain数据增强模块的功能集成工具，各字母对应：

步骤	字母	含义	功能说明
1	L	Loader（加载器）	从PDF、网页、数据库 等20+种来源加载原始数据
2	E	Embedding Model（嵌入模型）	将文本转换为向量表示， 如OpenAI、HuggingFace模型
3	D	Document Transformers（文档转换器）	文档分割、过滤、格式化等转换操作
4	V	VectorStore（向量存储器）	存储嵌入向量，支持相似度检索
5	R	Retriever（检索器）	针对查询匹配最相关的文档片段并返回

理解这两个概念的关系：

• 前者（L-E-D-V-R）是从业务逻辑角度描述构建RAG应用的5步操作流程。
• 后者（Loader-Embedding-Document-VectorStore-Retriever）是从框架组件角度描述数据增强模块的5个功能组件。

两者虽字母对应略有不同，核心目标一致——将外部文档转化为可供LLM检索和生成的知识资源。实践中开发者通常将二者混用，重点在于掌握各组件在RAG流水线中的具体用法。

官方术语

在阅读官方文档时，若想获得更权威的设计思路和最佳实践，仍建议使用：

Loader → Transform → Embed → Store → Retrieve 等官方术语。

在 LangChain 官方文档和 RAG 指南中，Loader → Transform → Embed → Store → Retrieve 描述了构建检索增强生成（RAG）系统的核心数据流水线。以下是每个步骤的精确含义及对应的 LangChain 组件：

步骤	官方术语	含义	对应 LangChain 组件	作用
1	Loader	从不同数据源（文件、网页、数据库等）加载原始文档	BaseLoader 及其子类：PyPDFLoader, TextLoader, WebBaseLoader, DirectoryLoader	将外部数据读取为 Document 对象列表
2	Transform	对加载后的文档进行转换处理，最常见的是文本分割（splitting），也可包括过滤、清洗、元数据增强等	TextSplitter：RecursiveCharacterTextSplitter, TokenTextSplitter 以及 DocumentTransformer	将长文档切分成适合嵌入模型的短块（chunk），保证语义完整性和检索精度
3	Embed	将文本块（chunk）转换为稠密向量（embedding vector）	Embeddings 接口：OpenAIEmbeddings, HuggingFaceEmbeddings, CohereEmbeddings	将文本映射到向量空间，用于语义相似度计算
4	Store	将生成的向量连同原始文本块存入向量数据库，支持高效检索	VectorStore：Chroma, FAISS, Pinecone, Milvus, Weaviate	持久化存储向量，提供 similarity_search、max_marginal_relevance_search 等接口
5	Retrieve	根据用户查询，从向量存储中检索最相关的文档块	VectorStoreRetriever：通过 vectorstore.as_retriever() 获得 也可用 ParentDocumentRetriever, MultiVectorRetriever 等	封装检索逻辑，返回相关文档列表供 LLM 生成最终回答

官方推荐流程图：

[Data Source] → Loader → [Document] → Transform → [Chunks] → Embed → [Vectors] → Store → [Vector DB] → Retrieve → [Relevant Chunks] → LLM → Answer

理解这五个官方术语后，就能在阅读 LangChain 1.0+ 文档时快速定位到对应的 API 和最佳实践。例如：

• Transform 阶段的 RecursiveCharacterTextSplitter 参数调优是 RAG 性能的关键
• Store 阶段可选择支持元数据过滤的向量库（如 Chroma 的 where 参数）
• Retrieve 阶段可结合 as_retriever(search_type="mmr") 提升结果多样性

LEDVR工作流示例

要为一家电商公司开发一个智能客服机器人，它能够自动回答用户关于产品、订单、退货政策等问题。这些答案都存在于公司的产品手册、FAQ 文档和内部知识库中。

LEDVR 工作流就是为这个机器人“武装”知识库的完美方案。

场景拆解

1. 目标: 构建一个能基于公司私有文档回答用户问题的智能客服。
2. 数据源: 产品PDF手册、FAQ网页、包含退货政策的Word文档。
3. 流程: 使用 LEDVR 工作流处理所有文档，构建一个专属的向量知识库。当用户提问时，机器人从这个知识库中检索相关信息，再生成准确的回答。

代码示例

下面是一个简化的代码示例(LangChain 1.0+ 风格)，展示如何使用 LangChain 1.0+ 的模块化组件来实现这个智能客服的知识库构建。

# 1. Load (加载)
# 从不同来源加载文档
from langchain_community.document_loaders import PyPDFLoader, WebBaseLoader, Docx2txtLoader

pdf_loader = PyPDFLoader("data/product_manual.pdf")
web_loader = WebBaseLoader("https://example-company.com/faq")
docx_loader = Docx2txtLoader("data/return_policy.docx")

# 加载所有文档
documents = []
documents.extend(pdf_loader.load())
documents.extend(web_loader.load())
documents.extend(docx_loader.load())

# 2. Extract & Divide (提取与分割)
# 将长文档分割成小块
from langchain_text_splitters import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,  # 每个文本块的大小
    chunk_overlap=50 # 块与块之间的重叠部分，以保持上下文连贯
)
splits = text_splitter.split_documents(documents)

# 3. Vectorize (向量化)
# 使用嵌入模型将文本块转换为向量
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 4. Vector Store (向量存储)
# 将向量存入本地向量数据库 (以 FAISS 为例)
from langchain_community.vectorstores import FAISS

vectorstore = FAISS.from_documents(documents=splits, embedding=embeddings)

# --- 知识库构建完成！以下是在线问答阶段 ---

# 5. Retrieve (检索)
# 将向量库转换为检索器
retriever = vectorstore.as_retriever(search_kwargs={"k": 3}) # 检索最相关的3个片段

# 模拟用户提问
query = "我买的商品有质量问题，怎么退货？"

# 检索相关文档片段
relevant_docs = retriever.invoke(query)

print("检索到的相关上下文:")
for i, doc in enumerate(relevant_docs):
    print(f"--- 片段 {i+1} ---")
    print(doc.page_content)

# 最后，将这些 relevant_docs 的内容作为上下文，连同用户的问题 query 一起
# 发送给大语言模型（如 ChatGPT），让它生成最终的回答。

LangChain 1.0 Agent工作流

从Chain到ReAct的架构跃迁

LangChain 1.0的发布标志着框架定位的根本转变——从大语言模型”调用工具集”升级为”生产级Agent系统开发框架”。

团队通过创造性地放弃早期的Chain设计，全面拥抱ReAct（Reasoning + Acting）模式，解决了0.x版本的核心缺陷：

0.x版本Chain问题	1.0版本ReAct解决方案
僵化的线性流程，偏离模板时陷入两难	灵活的推理→工具调用→观察→判断闭环，动态决策
生产级控制缺失，上下文溢出、敏感信息泄露	中间件机制注入PII检测、人工审批、自动重试
模型切换时需重写适配代码	Content Block统一内容规范，跨模型兼容

ReAct核心执行循环

ReAct模式是LangChain 1.0 Agent工作流的底层运行时引擎，其核心逻辑为一个标准化闭环：

模型推理（Thought） → 工具选择（Action） → 工具执行（Observation） → 任务判断（Decision） → 未完成则循环执行

LangChain 1.0将ReAct循环设为核心架构，使其成为构建可靠、可解释和可投入生产的智能体的默认架构。LangGraph 作为底层执行引擎，提供显式状态管理、动态分支控制和可恢复执行等能力。

分层策略：从简单到复杂

LangChain 1.0采用**”标准—扩展—复杂”**三级分层策略，实现从简单到复杂的开发路径，覆盖80%的通用场景与20%的定制化需求：

层级	适用场景	核心API	特点
标准场景（~90%的任务）	常规推理与工具调用	create_agent()	10行代码完成Agent构建，内置工具选择、状态管理、异常处理
扩展场景	需PII检测、人工审批、自动重试等生产级控制	Middleware	洋葱模型，在模型调用前/工具执行时/输出后插入自定义逻辑
复杂场景	多Agent协作、复杂状态机、长周期执行	LangGraph + 自定义工作流	图的节点和边，完全控制执行流程

构建智能体方式

# v1.0（官方推荐写法）
agent = create_agent(
    model="openai:gpt-4o",
    tools=tools + [retriever_tool],   # 直接注册工具列表
    system_prompt=system_prompt,
    middleware=[piiRedactionMiddleware(), summarizationMiddleware()],
    checkpointer=MemorySaver()        # 持久化
)
result = agent.invoke({"input": user_input})

典型应用场景

电商客服智能体

以下示例将完整演示：LEDVR数据处理工作流构建RAG知识库 + LangChain 1.0 Agent工作流实现自主推理与工具调用，两者结合搭建一个可投入生产的电商客服智能体。

场景背景

电商客服需同时处理三类任务：

• 知识查询：物流政策、退换货流程（需要RAG检索）
• 订单操作：查询订单状态、取消订单、修改地址（需要工具调用）
• 综合规划：面对”这个订单还能取消吗？取消后我重新下单有什么影响？”等问题时，需要多步推理和任务拆解

代码示例

代码示例：LangChain 1.0+ API

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
电商客服智能体完整示例（LangChain 1.0+）
包含：LEDVR数据工作流 + ReAct Agent工作流 + Middleware中间件控制
新增：将 RAG 检索器包装为 Agent 可调用的工具
"""

import os
from typing import TypedDict, List, Dict, Any
from pydantic import BaseModel, Field

# ============================================================
# 第一部分：LEDVR数据工作流 —— 构建RAG知识库
# ============================================================
from langchain.document_loaders import TextLoader, DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain_core.vectorstores import VectorStoreRetriever

# -------------------- 1. L - Load（加载数据）--------------------
loader = DirectoryLoader(
    path="./knowledge_base/",
    glob="**/*.txt",
    loader_cls=TextLoader,
    show_progress=True
)
raw_docs = loader.load()

# -------------------- 2. E & D - Extract & Divide（提取与分割）--------------------
splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", "。", "，", " ", ""]
)
chunks = splitter.split_documents(raw_docs)

# -------------------- 3. V - Vectorize（向量化）& Store--------------------
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./ecommerce_db"
)

# -------------------- 4. R - Retrieve（检索器）--------------------
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 3}
)


# ============================================================
# 第二部分：工具定义 —— Agent可调用的业务函数
# ============================================================
from langchain.tools import tool   # 新增：用于将普通函数或检索器包装为工具

# -------------------- 重点：将 retriever 包装成 Agent 可调用的工具 --------------------
@tool
def retrieve_knowledge(query: str) -> str:
    """
    从电商客服知识库中检索相关信息。
    当用户询问退货政策、物流时效、售后保障、发票制度、优惠券使用等问题时，必须调用此工具获取准确答案。
    输入：用户的问题或关键词（如“退货政策”“发货时间”）。
    输出：知识库中最相关的三个文档片段，以换行分隔。
    """
    docs = retriever.invoke(query)
    if not docs:
        return "未在知识库中找到相关信息，请尝试换个方式提问或联系人工客服。"
    # 将检索到的文档内容拼接成字符串，供LLM参考
    return "\n\n".join([doc.page_content for doc in docs])

# -------------------- 原有的业务工具 --------------------
def get_order_status(order_id: str) -> str:
    """获取订单状态：'pending'（待发货）、'shipped'（已发货）、'delivered'（已送达）、'cancelled'（已取消）"""
    orders_db = {
        "ORD001": {"status": "pending", "items": "T恤×2", "cost": 299},
        "ORD002": {"status": "shipped", "items": "牛仔裤×1", "cost": 599},
        "ORD003": {"status": "delivered", "items": "运动鞋×1", "cost": 899}
    }
    if order_id in orders_db:
        order = orders_db[order_id]
        return f"订单 {order_id} 状态: {order['status']}，商品: {order['items']}，金额: ¥{order['cost']}"
    return f"未找到订单 {order_id}，请确认订单号是否正确。"


def cancel_order(order_id: str) -> str:
    """取消订单。仅当订单状态为'pending'（待发货）时可取消。"""
    orders_db = {
        "ORD001": {"status": "pending", "items": "T恤×2", "cost": 299},
        "ORD002": {"status": "shipped", "items": "牛仔裤×1", "cost": 599},
        "ORD003": {"status": "delivered", "items": "运动鞋×1", "cost": 899}
    }
    if order_id not in orders_db:
        return f"未找到订单 {order_id}，取消失败。"
    
    if orders_db[order_id]["status"] != "pending":
        return f"无法取消订单 {order_id}：当前状态为 {orders_db[order_id]['status']}，仅'pending'状态可取消。"
    
    orders_db[order_id]["status"] = "cancelled"
    return f"订单 {order_id} 已成功取消。退款将在3个工作日内原路返回。"


def update_shipping_address(order_id: str, new_address: str) -> str:
    """更新订单收货地址。订单未发货时（pending状态）可修改。"""
    orders_db = {
        "ORD001": {"status": "pending", "address": "北京市朝阳区xxx"},
        "ORD002": {"status": "shipped", "address": "上海市浦东新区xxx"},
    }
    if order_id not in orders_db:
        return f"未找到订单 {order_id}，地址更新失败。"
    
    if orders_db[order_id]["status"] != "pending":
        return f"订单 {order_id} 已发货（状态: {orders_db[order_id]['status']}），无法修改地址。如需帮助请联系客服。"
    
    orders_db[order_id]["address"] = new_address
    return f"订单 {order_id} 收货地址已更新为: {new_address}"


def get_product_info(product_name: str) -> str:
    """查询商品信息，包括价格、库存和简要描述。"""
    products_db = {
        "夏季T恤": {"price": 149, "stock": 234, "description": "纯棉透气夏季短袖T恤"},
        "牛仔裤": {"price": 399, "stock": 89, "description": "经典修身牛仔裤"},
        "运动鞋": {"price": 599, "stock": 156, "description": "轻便透气运动跑鞋"}
    }
    for name, info in products_db.items():
        if name in product_name or product_name in name:
            return f"商品: {name}，价格: ¥{info['price']}，库存: {info['stock']}件，说明: {info['description']}"
    return f"未找到商品 '{product_name}'，请尝试更准确的产品名称。"


# ============================================================
# 第三部分：中间件 —— 生产级控制能力
# ============================================================
from langchain.agents.middleware import AgentMiddleware, BaseMiddleware
from langchain_core.messages import HumanMessage, AIMessage

class PIIRedactionMiddleware(BaseMiddleware):
    """PII检测与脱敏中间件：在模型调用前自动识别并遮蔽敏感信息"""
    
    async def before_model(self, state: Dict[str, Any]) -> Dict[str, Any]:
        sensitive_patterns = {
            r"\d{11}": "[REDACTED_PHONE]",
            r"\d{17}[\dXx]": "[REDACTED_ID_CARD]",
            r"[\w\.-]+@[\w\.-]+\.\w+": "[REDACTED_EMAIL]"
        }
        import re
        messages = state.get("messages", [])
        for msg in messages:
            if isinstance(msg, HumanMessage) and hasattr(msg, "content"):
                content = msg.content
                for pattern, replacement in sensitive_patterns.items():
                    content = re.sub(pattern, replacement, content)
                msg.content = content
        return state


class ApprovalMiddleware(BaseMiddleware):
    """人工审批中间件：高风险操作（取消订单、退款）需要人工确认"""
    
    async def before_tool(self, tool_name: str, tool_args: Dict[str, Any]) -> Dict[str, Any]:
        if tool_name in ["cancel_order", "update_shipping_address"]:
            print(f"\n⚠️【待审批】检测到高风险操作: {tool_name}")
            print(f"   操作参数: {tool_args}")
            user_input = input("   是否批准执行？(yes/no): ").strip().lower()
            if user_input != "yes":
                raise Exception(f"操作 {tool_name} 已被用户拒绝")
        return tool_args


# ============================================================
# 第四部分：Agent构建 —— 整合所有能力
# ============================================================
from langchain.agents import create_agent
from langgraph.checkpoint.memory import MemorySaver

# 1. 构建系统提示词，明确要求使用 retrieve_knowledge 工具进行知识检索
SYSTEM_PROMPT = """你是一个专业的电商客服助手，名为"小助手"。你的职责是为用户提供优质的购物支持和问题解答。

## 核心能力
1. **知识检索**：当用户询问退货政策、物流时效、售后保障、发票制度、优惠券使用等问题时，**必须**使用 retrieve_knowledge 工具从知识库中获取准确信息。不要凭自己的知识回答这些政策类问题。
2. **订单处理**：支持查询订单状态(get_order_status)、取消订单(cancel_order)、更新地址(update_shipping_address)。
3. **商品咨询**：回答商品相关问题，查询价格和库存信息(get_product_info)。

## 执行原则
- 先理解用户意图，再选择合适的工具
- 对于复杂问题（如同时涉及订单和政策的综合问题），可以分步调用多个工具
- 在调用高风险工具（取消订单、修改地址）前，向用户确认操作风险

## 回答规范
- 答案需清晰、准确、专业
- 引用知识库内容时，可适当总结，但不要编造信息
- 遇到无法解决的问题时，主动建议转人工客服
"""

# 2. 构建Agent（整合工具列表，包含新增的 retrieve_knowledge 工具）
agent = create_agent(
    model="openai:gpt-4o",
    tools=[
        retrieve_knowledge,        # ← 新增的RAG检索工具
        get_order_status,
        cancel_order,
        update_shipping_address,
        get_product_info,
    ],
    system_prompt=SYSTEM_PROMPT,
    middleware=[PIIRedactionMiddleware(), ApprovalMiddleware()],
    checkpointer=MemorySaver()
)


# ============================================================
# 第五部分：运行与测试
# ============================================================

def run_agent_example():
    """运行Agent，逐一测试各典型场景"""
    
    # 场景1：知识检索（现在会正确调用 retrieve_knowledge 工具）
    print("=== 场景1：知识检索（使用RAG工具）===")
    response1 = agent.invoke({
        "input": "你们的退货政策是什么？我买了东西不合适可以退吗？"
    })
    print("Agent回答:", response1["messages"][-1].content)
    
    # 场景2：工具调用 + 中间件控制
    print("\n=== 场景2：订单查询与取消（含中间件审批） ===")
    response2 = agent.invoke({
        "input": "我的订单号ORD001什么时候发货？我想取消这个订单。"
    })
    print("Agent回答:", response2["messages"][-1].content)
    
    # 场景3：多步推理（拆解复杂任务）
    print("\n=== 场景3：多步推理 ===")
    response3 = agent.invoke({
        "input": "帮我查一下夏季T恤的库存情况，如果够的话我要买2件。我的地址在上海市浦东新区，之前下单时默认填的北京地址，能帮我改过来吗？"
    })
    print("Agent回答:", response3["messages"][-1].content)


if __name__ == "__main__":
    run_agent_example()

代码说明

1. 环境准备：安装依赖，pip install langchain langchain-openai langchain-community chromadb pypdf python-dotenv。
2. 配置API密钥：在项目根目录创建.env文件，填入OPENAI_API_KEY="your-key"。
3. 知识库准备：在./knowledge_base/目录放置客服政策文档（如return_policy.txt、shipping_guide.txt等）。
4. RAG 检索工具未完整接入：代码中构建了 retriever并将它包装成 Agent 可调用的工具，然后将 retrieve_knowledge 加入 tools 列表。
5. 中间件执行顺序：声明顺序为 [PIIRedactionMiddleware, ApprovalMiddleware]，执行顺序为：
   • before_model（脱敏）→ before_tool（审批）→ 实际工具调用 → after_tool → after_model
6. 状态持久化：MemorySaver 仅用于演示，生产环境建议使用 RedisSaver 或 PostgresSaver 支持跨进程恢复。
7. 异步支持：中间件中的 before_model / before_tool 方法是异步的，Agent 的 ainvoke 方法支持异步调用。

工作流总结

两条工作流的协同关系

在完整的电商客服智能体中，两条工作流层次分明地协同运作：

工作流类型	执行角色	核心机制	运行位置
LEDVR数据工作流	离线/定时	文档加载→分割→向量化→入库	构建阶段运行， 知识库可定期更新
Agent工作流（ReAct）	在线/实时	推理→工具调用→观察→判断	每次用户请求触发， 实时执行

实际运行中：当用户提出请求时，Agent通过ReAct循环自行判断——若为知识类问题，通过检索器访问已构建的向量知识库；若为订单/商品操作，直接调用工具函数；若问题复杂，则拆解为多个子任务逐一处理，最后整合答案。

实践总结

1. 系统提示词设计至关重要：它决定了Agent的行为边界和决策质量，需清晰定义角色、能力和约束条件。
2. 中间件逐层递进：从基础安全控制（PII检测）开始，逐步叠加审计日志、人工审批等企业级能力。
3. 分层策略是开发路径：从create_agent()快速构建MVP，通过中间件解决生产级控制需求，遇到复杂多Agent场景再转向LangGraph定制工作流。