LangChain 1.x RAG完全指南：从传统检索到Agentic RAG，一篇搞定！

写在前面大家好随着大模型技术的爆发式发展如何让大模型“掌握”私有知识、如何有效遏制幻觉问题已成为企业落地AI应用的核心痛点。RAG检索增强生成正是目前公认的最有效的解决方案之一而LangChain作为大模型应用开发的事实标准框架其1.x版本带来了革命性的变化。本篇文章将带你从零开始系统掌握基于LangChain 1.x的RAG开发技术。全文约5000字涵盖理论原理、组件详解、完整项目实战和前沿的Agentic RAG建议收藏后慢慢阅读。你将学到传统信息检索的演进历程与核心痛点RAG核心原理数据入库流与用户检索流LangChain 1.x的RAG组件全家桶详解从零搭建生产级传统RAG系统含配置管理、日志、异常处理Agentic RAG原理与实战工具封装、记忆管理、自动决策一、 追本溯源传统信息检索的“三板斧”在深入RAG之前我们有必要回顾传统信息检索方案的发展历程。理解它们的优缺点能帮助我们更深刻地认识RAG的价值所在。1. 数据库检索结构化数据的精确匹配数据库正式开启了大规模数据检索的历史篇章。以关系型数据库为代表我们可以通过SQL语句实现精确、高效的条件筛选。典型示例-- 精确查询 SELECT city, fx_date, wind_dir_day FROM weather_data WHERE city北京 AND fx_date2025-12-14 -- 模糊匹配 SELECT city, fx_date, text_day FROM weather_data WHERE city北京 AND text_day LIKE %雨%优势支持复杂条件组合查询事务一致性保证成熟的索引优化机制局限性仅适用于结构化数据无法处理大段文本、图片、音视频等非结构化数据需要预先设计严格的库表结构除了关系型数据库还有NoSQL、时序数据库、图数据库等各有所长的专业化方案但它们共同的问题是依赖精确的结构和查询语法无法理解语义。2. 全文检索引擎关键词匹配的规模化方案为了解决海量文本的检索问题全文检索引擎应运而生。其核心技术是倒排索引。原理简述正常文章库文档1、文档2、文档3...倒排索引词语A → [文档1, 文档3]、词语B → [文档2, 文档3]...当用户搜索某个词语时系统可以直接从倒排索引中找到包含该词语的所有文档速度极快。代表技术TF-IDF词频-逆文档频率BM25目前最经典的排序算法之一优势海量文本中关键词检索速度极快局限性语义鸿沟搜索电脑无法匹配笔记本电脑无法理解同义词、上下文含义排序质量依赖词频统计而非语义相关性3. 个性化推荐从被动检索到主动推送除了用户主动输入的条件还有大量隐含信息未被利用性别、年龄、职业、历史行为、人群偏好等。个性化推荐系统正是为此而生。技术手段协同过滤基于用户/基于物品矩阵分解深度学习推荐模型优势实现千人千面发现用户潜在兴趣适用场景电商推荐、内容流、广告投放等局限性冷启动问题严重依赖大量用户行为数据可能造成信息茧房4. 小结传统方案的共同局限上述三种方案本质上都是半自动化系统——它们能够快速筛选出候选内容但最终哪些内容有用、哪一部分是答案需要人工二次判断。我们理想中的信息检索系统应该是用户用自然语言提出问题系统直接返回正确答案。大模型的出现让这个梦想有了实现的可能但幻觉问题和知识时效性又带来了新的挑战。RAG正是当前解决这些问题的最佳实践。二、 RAG核心原理两条腿走路RAGRetrieval-Augmented Generation由Facebook AI团队在2020年的论文《Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks》中首次提出。其核心思想简单而强大在让大模型回答问题之前先从外部知识库中检索相关信息作为参考资料。RAG包含两条主要工作流1. 数据入库流构建知识库这是离线阶段目标是构建可检索的向量知识库。步骤详解步骤说明关键技术点文档加载读取各类格式文件PDF、Word、CSV、HTML、Markdown等文本切块将长文档切分成语义完整的短文本chunk_size、chunk_overlap、分隔符选择文本向量化用Embedding模型将文本转为向量模型选型、维度确定、语义指纹向量入库存储到向量数据库索引类型、持久化、标量过滤为什么要切块块太大一个向量融合过多信息语义模糊难以精准召回块太小破坏完整语义切碎的信息无法被正确召回最佳实践根据文档类型调整通常256-512 token并设置20-50的overlap2. 用户检索流回答问题这是在线阶段实时响应用户请求。步骤详解用户提问 → 问题向量化 → 向量相似度检索 → 组织Prompt → 大模型生成 → 输出答案关键要点问题向量化必须使用与入库完全相同的Embedding模型相似度算法常用余弦相似度Cosine SimilarityPrompt工程至关重要明确要求如果参考资料中没有答案就说不知道三、 LangChain 1.x RAG组件全家桶LangChain从0.x到1.x经历了重大架构升级。0.x采用链式结构API混乱且仅支持传统RAG1.x底层基于LangGraph支持复杂控制流可构建Agentic RAG。核心组件详解1. Document对象统一的数据抽象from langchain_core.documents import Document doc Document( page_contentDogs are great companions..., metadata{source: mammal-pets-doc, author: John}, id0015 )2. 文档加载器Document LoadersLangChain 1.x不生产加载器而是集成了几十种现有工具。官方文档IntegrationsPDF加载示例from langchain_community.document_loaders import PyPDFLoader loader PyPDFLoader(物流信息.pdf) docs loader.load() print(docs[0].metadata) # 包含来源、页码等元数据3. 文本分割器Text Splittersfrom langchain_text_splitters import RecursiveCharacterTextSplitter text_splitter RecursiveCharacterTextSplitter( separators[\n\n, \n, , ], # 优先级从高到低 chunk_size500, # 块大小字符数或token数 chunk_overlap50, # 重叠长度 add_start_indexTrue # 记录切分位置 ) chunks text_splitter.split_documents(docs)4. 嵌入模型Embedding Models将文本转化为向量。LangChain集成了OpenAI、Cohere、HuggingFace、Ollama等几十家提供商。使用Ollama本地模型from langchain.embeddings import init_embeddings embedding_model init_embeddings( modelollama:quentinz/bge-large-zh-v1.5, # 中文优化模型 base_urlhttp://localhost:11434/ ) vector embedding_model.embed_query(物流公司速达物流) print(len(vector)) # 输出1024表示1024维向量5. 向量存储Vector Stores支持的向量库Chroma轻量级、FAISS学术常用、Milvus大规模分布式、Pinecone云服务等。Chroma使用示例from langchain_chroma import Chroma vector_store Chroma( collection_namelogistics_kb, embedding_functionembedding_model, persist_directory./chroma_db # 持久化目录 ) # 入库 ids vector_store.add_documents(documentschunks) # 检索 results vector_store.similarity_search( 快递是哪一天发出的, k3 # 返回top3 )6. 聊天模型Chat Modelsfrom langchain.chat_models import init_chat_model chat_model init_chat_model( modelqwen2.5:7b, model_providerollama, base_urlhttp://localhost:11434/ ) response chat_model.invoke(你好) print(response.content)四、 实战从零搭建生产级传统RAG系统理论讲完我们来动手搭建一个完整的RAG系统。这个系统将具备多环境配置管理、多格式文档加载、日志记录、异常处理等生产级特性。项目结构 traditional_rag/ ├── data/ # 待入库的文档 │ ├── 物流信息.pdf │ └── 常见问题.csv ├── logs/ # 日志目录 ├── vector_store/ # 向量库持久化 ├── document_loaders/ # 加载器模块 │ ├── __init__.py │ ├── pdf_loader.py │ ├── csv_loader.py │ └── load_all.py ├── text_splitters/ # 分割器 │ └── splitter.py ├── embeddings/ # 嵌入模型 │ └── embedding_model.py ├── utils/ # 工具模块 │ ├── config.py # 配置管理核心 │ └── logger.py # 日志封装 └── main.py # 主入口入库查询 核心模块代码1. 配置管理utils/config.py- 生产级参数管理import sys import logging from pathlib import Path # 环境识别启动时必须传入 if len(sys.argv) ! 2 or sys.argv[1] not in [development, pre_production, production]: print(Usage: python main.py [development|pre_production|production]) sys.exit(1) ENVIRONMENT sys.argv[1] print(fCurrent environment: {ENVIRONMENT}) # 基础路径根据环境可能不同 BASE_DIR Path(__file__).parent.parent # 开发环境配置 if ENVIRONMENT development: # 数据路径 DATA_PATH str(BASE_DIR / data/) VECTOR_STORE_PATH str(BASE_DIR / vector_store/) LOG_PATH str(BASE_DIR / logs/) # 日志配置 LOG_LEVEL logging.DEBUG # 开发环境用DEBUG LOG_FILE LOG_PATH rag_app.log # Embedding模型配置 EMBEDDING_MODEL ollama:quentinz/bge-large-zh-v1.5 EMBEDDING_BASE_URL http://localhost:11434/ # 向量库配置 COLLECTION_NAME logistics_dev # Chat模型配置 CHAT_MODEL_PROVIDER ollama CHAT_MODEL_NAME qwen2.5:7b CHAT_BASE_URL http://localhost:11434/ # Prompt模板防幻觉设计 PROMPT_TEMPLATE 你是一位专业的物流助手。请根据【参考资料】回答【用户问题】。 重要规则 1. 如果参考资料中有明确答案直接回答 2. 如果参考资料中没有答案请回答根据现有资料无法回答这个问题 3. 不要编造任何信息 4. 只输出答案不要输出其他内容 【参考资料】 {context} 【用户问题】 {question} 【你的回答】 # 预生产环境配置 elif ENVIRONMENT pre_production: DATA_PATH /app/data/ VECTOR_STORE_PATH /app/vector_store/ LOG_PATH /app/logs/ LOG_LEVEL logging.INFO # ... 其他配置类似但指向不同资源 # 生产环境配置 elif ENVIRONMENT production: DATA_PATH /prod/data/ VECTOR_STORE_PATH /prod/vector_store/ LOG_PATH /prod/logs/ LOG_LEVEL logging.WARNING # 生产环境只记录警告及以上 # ... 可能使用云端模型服务2. 日志封装utils/logger.pyimport logging from logging.handlers import RotatingFileHandler import os def get_logger(name, level, log_file): 获取配置好的logger支持日志轮转 logger logging.getLogger(name) logger.setLevel(level) # 避免重复添加handler if logger.handlers: return logger # 确保日志目录存在 log_dir os.path.dirname(log_file) if log_dir and not os.path.exists(log_dir): os.makedirs(log_dir) # 文件处理器10MB轮转保留5个备份 file_handler RotatingFileHandler( log_file, maxBytes10*1024*1024, backupCount5, encodingutf-8 ) file_handler.setLevel(level) # 控制台处理器 console_handler logging.StreamHandler() console_handler.setLevel(level) # 格式化器 formatter logging.Formatter( %(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s ) file_handler.setFormatter(formatter) console_handler.setFormatter(formatter) logger.addHandler(file_handler) logger.addHandler(console_handler) return logger3. 多格式文档加载器PDF加载器document_loaders/pdf_loader.pyfrom langchain_community.document_loaders import PyPDFLoader def load_pdf(file_path: str): 加载PDF文件返回Document列表 loader PyPDFLoader(file_path) docs loader.load() return docsCSV加载器document_loaders/csv_loader.pyfrom langchain_community.document_loaders import CSVLoader def load_csv(file_path: str): 加载CSV文件 loader CSVLoader(file_path, encodingutf-8) docs loader.load() return docs统一加载器document_loaders/load_all.pyimport os from . import pdf_loader, csv_loader from ..utils import config def load_documents_from_folder(folder_path: str): 递归加载文件夹下所有支持的文档 all_docs [] supported_extensions { .pdf: pdf_loader.load_pdf, .csv: csv_loader.load_csv, } for root, dirs, files in os.walk(folder_path): for file in files: ext os.path.splitext(file)[1].lower() if ext in supported_extensions: file_path os.path.join(root, file) try: docs supported_extensions[ext](file_path) all_docs.extend(docs) config.logger.info(fLoaded {len(docs)} documents from {file_path}) except Exception as e: config.logger.error(fFailed to load {file_path}: {e}) else: config.logger.warning(fUnsupported file type: {file}) return all_docs4. 主程序main.py入库流程from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain.embeddings import init_embeddings from langchain_chroma import Chroma from utils import config from document_loaders.load_all import load_documents_from_folder import traceback def build_vector_store(): 构建向量库主流程 config.logger.info( * 50) config.logger.info(Starting vector store building...) try: # 1. 加载文档 config.logger.info(fLoading documents from {config.DATA_PATH}) docs load_documents_from_folder(config.DATA_PATH) config.logger.info(fLoaded {len(docs)} raw documents) # 2. 文本切块 config.logger.info(Splitting documents...) text_splitter RecursiveCharacterTextSplitter( separators[\n\n, \n, 。, , , , , , ], chunk_size300, chunk_overlap50, add_start_indexTrue ) chunks text_splitter.split_documents(docs) config.logger.info(fCreated {len(chunks)} chunks) # 3. 加载Embedding模型 config.logger.info(fLoading embedding model: {config.EMBEDDING_MODEL}) embedding_model init_embeddings( modelconfig.EMBEDDING_MODEL, base_urlconfig.EMBEDDING_BASE_URL ) # 4. 创建向量库并入库 config.logger.info(fCreating vector store at {config.VECTOR_STORE_PATH}) vector_store Chroma( collection_nameconfig.COLLECTION_NAME, embedding_functionembedding_model, persist_directoryconfig.VECTOR_STORE_PATH ) # 分批入库避免内存爆炸 batch_size 100 for i in range(0, len(chunks), batch_size): batch chunks[i:ibatch_size] vector_store.add_documents(batch) config.logger.info(fIndexed batch {i//batch_size 1}/{(len(chunks)-1)//batch_size 1}) config.logger.info(Vector store building completed successfully!) return vector_store except Exception as e: config.logger.error(fFailed to build vector store: {traceback.format_exc()}) raise查询流程支持多轮对话from langchain.chat_models import init_chat_model from langchain.messages import HumanMessage, AIMessage from langchain_chroma import Chroma from utils import config def load_models_and_store(): 加载已构建的向量库和模型 config.logger.info(Loading embedding model...) embedding_model init_embeddings( modelconfig.EMBEDDING_MODEL, base_urlconfig.EMBEDDING_BASE_URL ) config.logger.info(Loading vector store...) vector_store Chroma( collection_nameconfig.COLLECTION_NAME, embedding_functionembedding_model, persist_directoryconfig.VECTOR_STORE_PATH ) config.logger.info(Loading chat model...) chat_model init_chat_model( modelconfig.CHAT_MODEL_NAME, model_providerconfig.CHAT_MODEL_PROVIDER, base_urlconfig.CHAT_BASE_URL ) return chat_model, vector_store def chat_loop(chat_model, vector_store): 多轮对话主循环 messages [] config.logger.info(Chat loop started. Type exit to quit.) while True: user_input input(\n 请提问: ).strip() if user_input.lower() in [exit, quit, 退出]: print(再见) break if not user_input: continue config.logger.info(fUser query: {user_input}) try: # 1. 检索相关文档 relevant_docs vector_store.similarity_search(user_input, k3) context \n---\n.join([doc.page_content for doc in relevant_docs]) config.logger.debug(fRetrieved {len(relevant_docs)} documents) # 2. 构建Prompt prompt config.PROMPT_TEMPLATE.format( contextcontext, questionuser_input ) config.logger.debug(fPrompt: {prompt[:200]}...) # 3. 调用大模型 response chat_model.invoke(prompt) answer response.content print(f 回答: {answer}) config.logger.info(fAnswer: {answer[:100]}...) except Exception as e: error_msg f处理问题时发生错误: {e} print(f❌ {error_msg}) config.logger.error(f{error_msg}\n{traceback.format_exc()})程序入口if __name__ __main__: import sys import argparse # 命令行参数解析 parser argparse.ArgumentParser(descriptionRAG System) parser.add_argument(env, choices[development, pre_production, production]) parser.add_argument(--build-only, actionstore_true, help仅构建向量库) parser.add_argument(--query-only, actionstore_true, help仅启动查询需已构建库) args parser.parse_args() # 初始化配置代码中已根据args.env加载 # ... 配置初始化代码 if args.build_only: build_vector_store() elif args.query_only: chat_model, vector_store load_models_and_store() chat_loop(chat_model, vector_store) else: # 默认先构建再查询 build_vector_store() chat_model, vector_store load_models_and_store() chat_loop(chat_model, vector_store)运行方式# 开发环境完整流程 python main.py development # 仅入库 python main.py development --build-only # 仅查询假设已入库 python main.py development --query-only五、 进阶Agentic RAG - 让系统拥有“行动力”传统RAG是被动的问答系统而Agentic RAG将RAG与AI Agent结合让大模型能够自主决策何时检索、何时调用外部工具发邮件、查API、写数据库等。Agentic RAG的核心架构用户输入 → Agent大脑 → 决策├─→ 调用检索工具 → 获取知识 → 生成回答└─→ 调用行动工具 → 执行操作 → 返回结果实战智能物流助手我们将构建一个能自动查询物流信息、并能发送邮件的智能助手。1. 封装工具检索工具from langchain_core.tools import tool from langchain_chroma import Chroma tool def search_logistics(query: str) - str: 查询物流相关信息。 当用户询问快递状态、发货日期、运输方式、理赔流程等问题时使用。 Args: query: 用户的具体问题 Returns: 从知识库中检索到的相关内容 # 获取全局的vector_store实例 vector_store get_vector_store() # 单例模式获取 docs vector_store.similarity_search(query, k2) return \n\n.join([doc.page_content for doc in docs])邮件发送工具import smtplib from email.mime.text import MIMEText from email.header import Header tool def send_complaint_email(tracking_number: str, issue_description: str) - str: 向快递公司发送投诉或查询邮件。 Args: tracking_number: 货物追踪编号必填 issue_description: 问题描述如包裹丢失、破损严重等 Returns: 发送结果说明 # 邮件配置应从配置读取 smtp_server smtp.163.com sender your_email163.com password your_authorization_code receiver courier_serviceexample.com # 构建邮件内容 content f 尊敬的客服团队 我的货物追踪号{tracking_number}出现问题{issue_description} 请尽快核实并回复处理方案。 此致 敬礼 try: msg MIMEText(content, plain, utf-8) msg[From] Header(sender) msg[To] Header(receiver) msg[Subject] Header(f快递问题反馈 - {tracking_number}, utf-8) with smtplib.SMTP_SSL(smtp_server, 465) as server: server.login(sender, password) server.sendmail(sender, [receiver], msg.as_string()) return f邮件发送成功已通知快递公司处理追踪号 {tracking_number} 的问题。 except Exception as e: return f邮件发送失败{str(e)}2. 创建Agent带记忆from langgraph.prebuilt import create_react_agent from langgraph.checkpoint.memory import InMemorySaver from langchain.chat_models import init_chat_model # 系统提示词关键定义Agent的行为规范 AGENT_SYSTEM_PROMPT 你是一个智能物流客服助手。你有以下能力 1. 查询物流信息当用户询问快递状态、发货时间、运输方式等问题时使用 search_logistics 工具 2. 发送投诉邮件当用户要求投诉、反馈问题、或包裹出现异常时使用 send_complaint_email 工具 注意事项 - 发送邮件前必须确认用户提供了货物追踪号 - 如果用户没有提供追踪号请先询问 - 回答要简洁、专业、有帮助 - 不要编造信息如果不知道就说不确定 def create_logistics_agent(): # 初始化模型 model init_chat_model( modelconfig.CHAT_MODEL_NAME, model_providerconfig.CHAT_MODEL_PROVIDER, base_urlconfig.CHAT_BASE_URL, temperature0.1 # 降低随机性提高确定性 ) # 创建Agent带记忆 checkpointer InMemorySaver() agent create_react_agent( modelmodel, tools[search_logistics, send_complaint_email], checkpointercheckpointer, promptAGENT_SYSTEM_PROMPT ) return agent3. 运行Agentdef run_agent(): agent create_logistics_agent() # 会话配置thread_id用于区分不同用户/会话 config {configurable: {thread_id: user_123_session_1}} print(智能物流助手已启动输入exit退出) while True: user_input input(\n用户: ).strip() if user_input.lower() in [exit, 退出]: print(感谢使用再见) break # 调用Agent自动决策和工具调用 response agent.invoke( {messages: [{role: user, content: user_input}]}, configconfig ) # 提取最终回答 final_message response[messages][-1] print(f助手: {final_message.content}) # 可选打印工具调用过程调试用 for msg in response[messages]: if hasattr(msg, tool_calls) and msg.tool_calls: print(f [调用工具: {msg.tool_calls[0][name]}])运行效果演示 用户: 帮我查一下货物ABC123456什么时候发的 助手: [调用工具: search_logistics] 根据查询结果货物ABC123456的发货日期是2025年12月15日。 用户: 包裹好像丢了怎么办 助手: [调用工具: search_logistics] 如果包裹丢失建议1) 联系快递公司查询 2) 提供订单号和联系方式 3) 如需赔偿保存好相关证据。 请问需要我帮您向快递公司发送投诉邮件吗 用户: 是的帮我发邮件追踪号是ABC123456 助手: [调用工具: send_complaint_email] 邮件发送成功已通知快递公司处理追踪号ABC123456的问题。他们会在1-2个工作日内回复。 Agentic RAG的优缺点维度传统RAGAgentic RAG决策能力固定流程自主决策工具调用不支持支持多工具组合适用场景简单问答复杂多步任务可控性高中依赖模型能力成本低高多次模型调用最佳实践80%场景20%复杂场景实际项目中推荐混合模式简单问题走传统RAG快、便宜、可控复杂任务走Agentic RAG灵活、智能。六、 总结与展望本文核心要点回顾阶段核心技术LangChain组件适用场景传统检索SQL、倒排索引不适用结构化数据、关键词搜索传统RAG向量检索 生成Loader, Splitter, Embeddings, VectorStore知识库问答、客服机器人Agentic RAG工具调用 自主决策LangGraph, create_react_agent, tool自动化业务处理、智能助理关键经验总结RAG不是万能药对于实时性要求极高的场景仍需结合微调或实时APIEmbedding模型选型至关重要建议用领域数据做A/B测试Prompt工程是灵魂防幻觉设计不知道就说不知道必须要有日志和监控是生产底线没有日志的RAG系统就是黑盒传统RAG和Agentic RAG各有所长不要盲目追求Agent下一步学习方向LangGraph深度探索构建更复杂的状态机、多智能体协作RAG评估体系RAGAS、ARES等框架进行准确率评估高级RAG技术HyDE假设性文档嵌入、RAPTOR递归摘要树、Self-RAG多模态RAG图文混合检索、PDF表格/图表解析写在最后LangChain 1.x的最大变化是底层基于LangGraph这使得构建有状态、可循环、可中断恢复的智能应用成为可能。RAG只是其中一种应用形态未来还会有更多创新模式涌现。希望这篇文章能帮助你真正掌握LangChain 1.x的RAG开发。如果觉得有用欢迎点赞、收藏、转发有问题请在评论区留言我会尽力解答。下期预告《LangGraph实战手写一个能自动修复代码的Bug解决Agent》参考资料LangChain 1.x 官方文档RAG原始论文 - Facebook AI, 2020LangGraph 文档