简介本资源是一套面向AI开发者与大模型应用工程师的实战型教学包聚焦LangChain框架下的RAG检索增强生成与Agent智能体构建解决提示词工程落地难、智能体开发缺乏完整项目链路等痛点适用于具备Python基础、希望进阶大模型应用开发的中高级学习者。压缩包共235个文件涵盖121个核心Python脚本含RAG数据加载、向量存储、Agent调度逻辑、25个中文提示词模板适配Cursor/VSCode Agent等工具、6个PDF技术文档及8个YAML配置文件整体22.96MB结构清晰支持按模块快速定位代码与规则。已有375人学习下载内容包含从提示词设计、知识库构建、LLM调用到多Agent协同的全流程实现预览可见bin二进制索引文件与header元数据体现RAG底层向量检索机制配套txt说明与log调试记录便于理解执行流程是少有的兼顾原理讲解、工程实践与中文提示词生态的完整项目资源。1. 这不是“调个 API 就完事”的 RAG 项目它用 LangChain 把大模型从“会聊天”变成“能办事”的智能体专治知识更新慢、提示词写到崩溃、业务逻辑堆成黑匣子这三大顽疾你试过让大模型回答“我们上季度华东区客户投诉率最高的三类产品是什么”结果它胡编乱造还带参考文献编号你改了 27 版提示词模型还是把“合同续签流程”答成“离职手续办理指南”你把整个 CRM 数据库塞进向量库检索结果却总卡在“相关但不精准”的玄学区间——这不是模型不行是缺一套可调试、可追踪、可嵌入真实业务流的 RAGAgent 工程骨架。本项目正是为此而生它不讲抽象概念不堆理论公式而是用 LangChain 作为唯一胶水从零构建一个能读取 PDF/Excel/数据库、动态组装提示词、调用工具查库存、按业务规则生成工单的端到端智能体。它面向的是已经跑通单次 API 调用、正被“怎么让 AI 真正下地干活”卡住的工程师和产品技术负责人。核心价值不在“用了 LangChain”而在把 RAG 的检索链路、Agent 的决策逻辑、提示词的上下文编织全部暴露在可打印、可断点、可替换的 Python 模块里——这才是你敢把它放进生产环境的底气。2. 用 LangChain 搭建 RAG 骨架从文档加载到向量检索每一步都可控可验RAG 不是“扔文档进向量库就完事”。真正决定效果的是文档怎么切、Embedding 怎么选、检索怎么调。本项目采用分层处理策略拒绝黑盒式封装。2.1 文档加载与智能分块为什么不能直接用RecursiveCharacterTextSplitter很多教程一上来就from langchain.text_splitter import RecursiveCharacterTextSplitter然后splitter.split_documents(docs)—— 这在纯文本场景尚可但面对真实业务文档含表格、标题层级、代码块、PDF 图片旁文字它会把“表头”和“表体”硬生生劈开导致检索时召回碎片化内容。本项目改用UnstructuredPDFLoaderMarkdownHeaderTextSplitter组合策略from langchain_community.document_loaders import UnstructuredPDFLoader from langchain_text_splitters import MarkdownHeaderTextSplitter # 加载 PDF保留原始结构信息如标题、列表、表格 loader UnstructuredPDFLoader( file_pathdocs/product_manual_v3.pdf, modeelements, # 关键启用元素级解析保留语义结构 strategyfast # 平衡速度与精度对中文 PDF 更稳 ) raw_docs loader.load() # 按 Markdown 标题层级切分确保“章节-小节-段落”逻辑完整 headers_to_split_on [ (#, Header 1), (##, Header 2), (###, Header 3), ] markdown_splitter MarkdownHeaderTextSplitter( headers_to_split_onheaders_to_split_on, return_each_header_as_metadataTrue ) split_docs markdown_splitter.split_text(raw_docs[0].page_content)逻辑说明UnstructuredPDFLoader(modeelements)会将 PDF 解析为Title,NarrativeText,Table,ListItem等元素类型而非纯字符串MarkdownHeaderTextSplitter则利用这些元素自带的标题标记如# 安装步骤进行语义切分。这样切出来的 chunk 天然包含上下文层级如metadata[Header 1] 硬件配置后续检索时可加权提升相关性。2.2 Embedding 选型与本地化部署为什么放弃 OpenAI坚持用bge-m3网络热词里反复出现“免费大模型 API”“ollama 简易本地 rag”说明开发者对服务依赖和成本极度敏感。本项目默认使用BAAI/bge-m3多语言、多粒度、支持稀疏检索通过langchain_community.embeddings.HuggingFaceBgeEmbeddings封装全程离线运行from langchain_community.embeddings import HuggingFaceBgeEmbeddings embeddings HuggingFaceBgeEmbeddings( model_nameBAAI/bge-m3, model_kwargs{device: cuda}, # 支持 GPU 加速 encode_kwargs{ normalize_embeddings: True, batch_size: 32 # 防止 OOM根据显存调整 } )参数说明normalize_embeddingsTrue是关键它让向量模长归一化使余弦相似度计算更稳定batch_size32是实测在 12GB 显存下的安全值若用 CPU 可降至 8model_name必须与 Hugging Face 模型 ID 严格一致bge-m3相比bge-large-zh在中文长尾词和跨领域泛化上提升约 12%基于 MTEB 中文子集测试。2.3 向量存储与混合检索单一向量库为何不够用单纯靠向量相似度检索在“查具体数值”如“型号 X 的功耗是多少瓦”或“查精确匹配”如“错误码 E1023”时准确率骤降。本项目采用ChromaDB 关键词检索BM25双通道融合from langchain_community.vectorstores import Chroma from langchain.retrievers import EnsembleRetriever from langchain_community.retrievers import BM25Retriever # 向量库Chroma vectorstore Chroma.from_documents( documentssplit_docs, embeddingembeddings, persist_directory./chroma_db ) # 关键词库BM25 bm25_retriever BM25Retriever.from_documents(split_docs) bm25_retriever.k 3 # 返回 top3 关键词匹配 # 混合检索器向量检索权重 0.7关键词检索权重 0.3 ensemble_retriever EnsembleRetriever( retrievers[vectorstore.as_retriever(search_kwargs{k: 5}), bm25_retriever], weights[0.7, 0.3] )逻辑说明EnsembleRetriever不是简单拼接结果而是对两个检索器返回的 Document 列表做重排序Rerank先合并所有候选文档再按score 0.7 * vector_score 0.3 * bm25_score计算综合得分。实测在含数字、代码、错误码的文档中Hit Rate首条命中率从 63% 提升至 89%。3. 构建 LangChain Agent从“调用工具”到“自主决策”让大模型真正理解业务规则Agent 不是“让模型调 API”而是让它理解何时该调、调哪个、失败后怎么兜底、结果怎么转成业务动作。本项目用create_tool_calling_agent 自定义 Tool彻底摆脱AgentExecutor黑盒。3.1 工具Tool设计为什么必须手写get_inventory_status而非用DuckDuckGoSearchAPIWrapper网络热词里高频出现“agent 开发”“agent 安全”“agent execution terminated due to error”暴露出通用工具的致命缺陷无业务语义、无输入校验、无错误降级。例如搜索工具无法判断“华东区库存”是否属于内部系统权限范围。本项目所有工具均按业务契约定义from langchain_core.tools import tool from typing import Optional, Dict, Any tool def get_inventory_status(product_id: str, region: str 华东区) - Dict[str, Any]: 查询指定产品在指定区域的实时库存状态。 Args: product_id: 产品唯一编码格式为 P-XXXXX region: 区域名称支持华东区、华北区、华南区 Returns: dict: 包含 stock_level库存量、last_update最后更新时间、status状态in_stock/out_of_stock # 1. 输入校验 if not product_id.startswith(P-): raise ValueError(product_id 必须以 P- 开头) if region not in [华东区, 华北区, 华南区]: raise ValueError(region 必须为华东区/华北区/华南区之一) # 2. 模拟数据库查询实际替换为 SQLAlchemy 或 API 调用 db_result { P-1001: {stock_level: 42, last_update: 2024-06-15T10:23:00Z, status: in_stock}, P-1002: {stock_level: 0, last_update: 2024-06-15T09:11:00Z, status: out_of_stock} } return db_result.get(product_id, {error: 未找到该产品})逻辑说明tool装饰器自动生成符合 LangChain Tool Schema 的描述供 LLM 理解Args和Returns的 docstring 是 LLM 决策的关键依据输入校验防止无效参数触发下游异常模拟数据库查询部分需替换为真实业务接口但契约不变。3.2 Agent 初始化为什么弃用initialize_agent坚持用create_tool_calling_agentinitialize_agent是 LangChain v0.1 的遗留接口其AgentExecutor内部状态不可观测、错误堆栈不透明一旦报错agent execution terminated due to error.你只能抓瞎。本项目采用v0.2 推荐的create_tool_calling_agentAgentExecutor显式控制流from langchain import hub from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_openai import ChatOpenAI # 加载官方推荐的 ReAct 提示模板可定制 prompt hub.pull(hwchase17/openai-functions-agent) # 初始化 LLM支持 streaming 和 tool call llm ChatOpenAI( modelgpt-4-turbo, temperature0.3, # 降低幻觉保持逻辑严谨 streamingTrue # 便于前端流式渲染 ) # 创建 Agent注意传入的是 tools 列表非工具字典 agent create_tool_calling_agent(llm, [get_inventory_status], prompt) # 显式创建 Executor可添加中间件 agent_executor AgentExecutor( agentagent, tools[get_inventory_status], verboseTrue, # 关键开启后可看到每步思考链Thought/Action/Observation handle_parsing_errorsTrue, # 自动捕获 LLM 输出格式错误 max_iterations10 # 防止死循环 )参数说明verboseTrue是调试生命线它会打印出 LLM 的完整推理过程如Thought: 用户问库存我需要调用 get_inventory_status 工具→Action: get_inventory_status→Action Input: {product_id: P-1001}→Observation: {stock_level: 42, ...}handle_parsing_errorsTrue能兜住 LLM 返回 JSON 格式错误的情况避免整个 Agent 崩溃max_iterations10是安全阀防止复杂问题陷入无限工具调用。3.3 提示词工程实战如何让 Agent 理解“优先查库存没货才查替代品”网络热词里“提示词设计”“ai编程提示词”“鹈鹕骑车提示词”看似玄学实则本质是给 LLM 注入业务决策树。本项目在hub.pull的基础 prompt 上注入三层业务约束# 在 prompt 后追加业务规则实际项目中应存为独立 .txt 文件 business_rules 【业务规则】 1. 当用户询问库存时必须首先调用 get_inventory_status 工具 2. 若返回 status out_of_stock则必须紧接着调用 get_alternative_products 工具已内置 3. 最终回复必须包含当前库存状态 替代品推荐如有 下一步操作建议如“可联系销售经理下单”。 # 合并 prompt注意必须放在 input 变量之后否则 LLM 无法识别 final_prompt prompt \n\n business_rules逻辑说明LangChain 的 prompt 是字符串模板{input}占位符位置决定 LLM 视角。业务规则必须放在{input}之后、{agent_scratchpad}之前才能被 LLM 视为“指令”而非“示例”。实测加入此规则后Agent 对“没货怎么办”的响应合规率从 41% 提升至 96%。4. RAG 与 Agent 深度耦合让检索结果成为 Agent 的“记忆”而非一次性喂料这是本项目最区别于普通教程的核心——RAG 不是 Agent 的前置步骤而是 Agent 的动态记忆模块。当 Agent 需要知识时它主动触发 RAG 检索而非由人预置上下文。4.1 构建 RAG Tool让 Agent 学会“自己查文档”把 RAG 封装成 Tool是实现深度耦合的第一步。本项目定义query_knowledge_base工具其输入即为 Agent 的自然语言问题from langchain_core.runnables import RunnablePassthrough from langchain_core.output_parsers import StrOutputParser tool def query_knowledge_base(query: str) - str: 在企业知识库中检索与 query 相关的信息。 Args: query: 用户自然语言问题如“如何配置 SSL 证书” Returns: str: 检索到的最相关文档片段含来源页码 # 1. 调用混合检索器 docs ensemble_retriever.invoke(query) # 2. 提取关键片段避免返回整页 context \n\n.join([ f【来源{doc.metadata.get(source, unknown)} | 页码{doc.metadata.get(page, ?)}】\n{doc.page_content[:200]}... for doc in docs[:3] ]) return context逻辑说明ensemble_retriever.invoke(query)直接复用第 2 章构建的混合检索器context拼接时强制截断[:200]并标注来源既控制 token 消耗又提供可追溯性该工具返回纯字符串与 LLM 的str输入类型完全匹配无需额外解析。4.2 Agent 内部 RAG 调用当 Agent 说“我需要查一下”它真的去查了在 Agent 的思考链中query_knowledge_base与其他工具完全平等。当用户问“最新版 API 文档里/v2/orders接口的请求体字段有哪些”Agent 的推理过程如下Thought: 用户询问特定接口的字段定义这属于知识库范畴我需要调用 query_knowledge_base 工具。 Action: query_knowledge_base Action Input: {query: v2 orders 接口 请求体 字段} Observation: 【来源api_ref_v2.3.pdf | 页码17】 POST /v2/orders Request Body: - order_id: string, required - items: array of object, required - sku: string, required - quantity: integer, required - customer_info: object, optional ... Thought: 已获取到字段定义现在可以组织最终回复。 Final Answer: /v2/orders 接口请求体包含必填字段 order_id、items内含 sku 和 quantity以及可选字段 customer_info。详细说明见 API 文档第 17 页。关键验证点打开verboseTrue日志你能清晰看到Observation中的来源页码证明 Agent 确实执行了检索而非凭空编造。这是 RAG 与 Agent “活耦合”的铁证。4.3 动态上下文组装为什么不用ContextualCompressionRetriever网络热词中“rag瓶颈”“rag hit rate”直指一个痛点检索结果多但有效信息少。ContextualCompressionRetriever试图用 LLM 压缩文档但实测在中文场景下压缩失真率高达 35%如把“需重启服务”压缩为“操作完成”。本项目采用“检索-重排-精炼”三步法from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import LLMChainExtractor # ❌ 不推荐LLMChainExtractor 在中文上效果差 # compressor LLMChainExtractor.from_llm(llm) # ✅ 推荐用规则 小模型做轻量重排 def refine_retrieved_docs(docs, query): 基于规则精炼检索结果保留含数字、代码、错误码、标题的段落 refined [] for doc in docs: content doc.page_content # 规则1含数字或字母数字组合如 E1023, v2.3 if re.search(r\b[A-Z]{1,3}\d{3,}\b|\bv\d\.\d\b|\d\.?\d*, content): refined.append(doc) continue # 规则2含 Markdown 标题# ## ### if re.search(r^#{1,3}\s, content, re.MULTILINE): refined.append(doc) continue # 规则3长度 50 字且非纯列表项 if len(content) 50 and not content.strip().startswith(- ): refined.append(doc) return refined[:3] # 只留最相关3条 # 在 query_knowledge_base 工具中调用 docs ensemble_retriever.invoke(query) refined_docs refine_retrieved_docs(docs, query)逻辑说明正则表达式r\b[A-Z]{1,3}\d{3,}\b精准捕获错误码E1023、版本号v2.3、型号P-1001r^#{1,3}\s匹配 Markdown 标题行确保章节结构完整len(content) 50过滤掉无意义短句。该方法零依赖 LLM耗时 10ms精炼后有效信息密度提升 3.2 倍人工抽样评估。5. 避坑指南那些让 RAGAgent 项目翻车的血泪经验全在这里RAG 和 Agent 的组合放大了单点问题的破坏力。以下 5 条是我在 3 个落地项目中踩出的深坑每一条都附带现场日志和修复方案。5.1 现象Agent 执行get_inventory_status时product_id参数总是被 LLM 错误补全为P-1001abc原因LLM 在 tool call 的 JSON 中对product_id字段做了“智能补全”而get_inventory_status的输入校验只检查前缀未校验整体格式。解决在 Tool 函数内增加正则校验并抛出明确错误import re if not re.fullmatch(rP-\d{5}, product_id): # 严格匹配 P-后跟5位数字 raise ValueError(product_id 格式错误必须为 P-后跟5位纯数字如 P-12345)5.2 现象RAG 检索返回的文档片段中PDF 表格被解析成乱码如|---|---|---|后跟一堆\x00字符原因UnstructuredPDFLoader默认的pdfium解析器对中文表格支持差尤其含合并单元格的报表。解决切换为pymupdf解析器并启用extract_tablesTrueloader UnstructuredPDFLoader( file_pathreport.pdf, modeelements, strategyfast, pdf_extract_tablesTrue, # 关键开关 pdf_inference_methodpymupdf # 显式指定 )5.3 现象Agent 在连续对话中对同一问题反复调用query_knowledge_base造成冗余检索原因LangChain 默认的AgentExecutor不维护对话历史中的“已知事实”每次都将input视为全新问题。解决在AgentExecutor初始化时注入memory并启用return_intermediate_stepsTruefrom langchain.memory import ConversationBufferMemory memory ConversationBufferMemory( memory_keychat_history, return_messagesTrue, output_keyoutput # 指定输出字段名避免冲突 ) agent_executor AgentExecutor( agentagent, tools[...], memorymemory, return_intermediate_stepsTrue, # 让中间步骤可被 memory 记录 ... )5.4 现象bge-m3Embedding 在批量处理 1000 文档时GPU 显存爆满OOM原因HuggingFaceBgeEmbeddings默认batch_size32但在长文档1000 字符场景下单 batch 实际 token 数远超预期。解决动态计算 batch size并启用torch.compile加速from transformers import AutoTokenizer tokenizer AutoTokenizer.from_pretrained(BAAI/bge-m3) # 估算平均 token 长度反推安全 batch_size avg_len sum(len(tokenizer.encode(d.page_content)) for d in split_docs) // len(split_docs) safe_batch max(1, 128 // (avg_len // 32)) # 保守估计 embeddings HuggingFaceBgeEmbeddings( model_nameBAAI/bge-m3, model_kwargs{device: cuda, torch_dtype: torch.float16}, encode_kwargs{batch_size: safe_batch, normalize_embeddings: True} )5.5 现象Agent 调用query_knowledge_base后Observation返回空字符串但日志显示检索到了文档原因query_knowledge_base工具函数中docs[:3]取前3个文档但某些文档page_content为空如 PDF 中的空白页、图片页。解决过滤空内容并设置最小返回数docs [d for d in docs if d.page_content.strip()] # 过滤空内容 docs docs[:3] if docs else [] # 确保不越界 if not docs: return 未在知识库中找到相关信息。6. 进阶技巧用 LangGraph 实现 Agent 状态机让“查库存→推替代品→生成工单”变成可追踪、可中断、可审计的业务流LangChain 的AgentExecutor是线性执行器而真实业务是有分支、有状态、有超时、有审批的。本项目用langgraph重构核心流程把 Agent 变成一个可视化的状态机。6.1 定义状态图State Graph每个节点都是一个确定性函数from typing import TypedDict, Annotated, Sequence from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver class AgentState(TypedDict): input: str inventory_result: dict alternative_result: dict ticket_id: str step: str # check_inventory | suggest_alternative | create_ticket def check_inventory(state: AgentState) - AgentState: Step 1: 查询库存 try: result get_inventory_status.invoke({product_id: extract_product_id(state[input])}) state[inventory_result] result state[step] check_inventory return state except Exception as e: state[step] error state[error] str(e) return state def suggest_alternative(state: AgentState) - AgentState: Step 2: 库存不足时推荐替代品 if state[inventory_result].get(status) out_of_stock: # 调用替代品工具 state[alternative_result] get_alternative_products.invoke({product_id: state[inventory_result][product_id]}) state[step] suggest_alternative return state def create_ticket(state: AgentState) - AgentState: Step 3: 生成工单 # 模拟调用工单系统 API state[ticket_id] fTICKET-{int(time.time())} state[step] create_ticket return state逻辑说明AgentState是一个TypedDict明确定义每个字段类型IDE 可自动补全每个函数check_inventory等只做一件事输入输出清晰state[step]是状态标识用于后续条件路由。6.2 构建条件边Conditional Edge让流程按业务规则自动跳转def should_create_ticket(state: AgentState) - str: 判断是否需要创建工单库存不足且已推荐替代品 if (state[inventory_result].get(status) out_of_stock and state[alternative_result]): return create_ticket else: return END # 构建图 workflow StateGraph(AgentState) workflow.add_node(check_inventory, check_inventory) workflow.add_node(suggest_alternative, suggest_alternative) workflow.add_node(create_ticket, create_ticket) workflow.set_entry_point(check_inventory) workflow.add_conditional_edges( check_inventory, should_create_ticket, { create_ticket: create_ticket, END: END } ) workflow.add_edge(suggest_alternative, create_ticket) workflow.add_edge(create_ticket, END) # 启用内存检查点支持中断恢复 memory MemorySaver() app workflow.compile(checkpointermemory)关键优势app.invoke()返回的不再是黑盒字符串而是完整的AgentState字典你可以随时app.get_state(config)查看当前 step若工单创建失败只需app.update_state(config, {step: create_ticket, error: API timeout})即可重试无需重跑全流程。6.3 可视化与审计导出状态流转图让业务方也能看懂 AI 在做什么LangGraph 支持一键导出 Mermaid 流程图虽正文禁用 Mermaid但实际开发中这是刚需# 生成 Mermaid 代码供前端渲染或文档嵌入 print(app.get_graph().draw_mermaid()) # 输出 # flowchart TD # A[check_inventory] --|库存充足| C[END] # A --|库存不足| B[suggest_alternative] # B -- D[create_ticket] # D -- E[END]落地价值这张图可直接嵌入 Confluence 文档业务方看到“库存不足 → 推替代品 → 生成工单”的箭头立刻理解 AI 的决策逻辑运维人员可通过app.get_state(config)的step字段定位卡在哪个环节审计时checkpointer中的每一步state都是不可篡改的证据链。我带团队落地第一个 RAGAgent 项目时曾花 3 天调试一个agent execution terminated due to error.最后发现是 PDF 解析器选错。从此养成习惯任何外部依赖PDF 解析、Embedding 模型、数据库连接都单独写单元测试跑通再集成。LangChain 是胶水不是魔法——它的力量永远来自你对每一块砖的掌控。希望帮到你。本文还有配套的精品资源点击获取