大模型虽聪明但记不住你家公司的产品手册。给他配一个“随身图书馆”马上变身金牌客服。一、为什么你的AI总在“一本正经地胡说八道”先给你讲个真实场景。某天你问一个通用大模型“我们公司的新员工入职流程是怎样的”它可能会煞有介事地回复你“第一步填写入职登记表第二步参加企业文化培训……”但这些步骤根本不是你们公司的规定。这就是所谓“幻觉”——大模型为了回答你会拼凑出一段听起来合理但其实是编造的内容。另一天你问它“昨天发布的那个新功能怎么用”它更是两眼一抹黑因为它的知识只更新到几个月前。怎么办传统的做法是重新训练或微调模型但那需要成千上万的高质量标注数据和巨额的计算资源一般企业根本承受不起。于是一种轻量级、高效率的解决方案横空出世RAG检索增强生成。RAG的思路非常直接——不让大模型凭空回答而是先到你的知识库里“查资料”查到了再根据资料来回答。就好比考试允许翻书答案自然又准又稳。这套技术路线已经在大量企业中得到验证。一份2025年初的行业报告显示采用RAG架构的客服系统在复杂问题的回答准确率上相比纯大模型平均提升了24%从68%到92%。另一份Gartner的预测指出到2026年超过75%的企业级AI应用将采用RAG或其变体。今天我将带你从零开始亲手搭建一套完整的RAG知识库系统。全部代码基于Python LangChain ChromaDB FastAPI你不需要昂贵的GPU一台普通电脑就能跑起来。我会用最通俗的语言解释每一个技术点保证你不仅会复制代码还能真正理解每一步在做什么。读完这篇文章你将拥有一个能从联想官网自动抓取技术知识的爬虫一个能把你自己的Markdown文档存入向量库的接口一个能智能回答问题的Web API一套完整的RAG工程实践可直接用于企业项目我们将用到的技术栈FastAPIWeb框架 LangChainLLM应用工具箱 ChromaDB轻量向量数据库 OpenAI兼容的Embedding接口。别被名字吓到我都会逐一讲清楚。二、吃透RAG三个核心概念厨房版比喻在写代码之前先把三个最核心的概念用厨房故事讲明白。2.1 什么是RAG就是“查菜谱炒菜”想象你是个刚入行的厨师。传统大模型就像你把一本《中华菜谱大全》背得滚瓜烂熟。但如果客人点了一道你没背过的创新菜你就只能凭感觉乱炒结果往往翻车。RAG模式你不需要背完整本书。客人点菜时你立刻去查对应的菜谱检索照着菜谱的步骤和配料来做生成。这样做出来的菜既快又准而且菜谱随时可以更新。对应到计算机语言检索R把用户的问题拿到知识库向量数据库里去搜索最相关的文档片段。增强A把搜到的文档片段和原问题拼接成一个“带资料的提问”。生成G把这个“带资料的提问”送给大模型让它只能根据资料回答。2.2 向量与向量数据库让电脑“理解”语义电脑天生只认识数字。你给它说“苹果”它无感。你需要把“苹果”变成一组数字比如 [0.12, -0.35, 0.78, ...]。这个过程叫做Embedding向量化。神奇的是语义相近的词它们的向量在高维空间里也离得很近。比如“苹果”和“红富士”的向量距离近“苹果”和“卡车”的向量距离远。向量数据库就是专门存储和检索这些向量的仓库。传统数据库如MySQL用LIKE %苹果%做关键词匹配搜不到“红富士”。而向量数据库通过计算向量距离能轻松找到语义相似的内容。这也是RAG为什么比普通搜索引擎更聪明的原因。2.3 异步编程和FastAPI让Web服务不“卡死”如果你写的Web接口是同步的那么当用户上传一个大文件时程序会一直等着文件读写完才能处理下一个请求——后面的用户都会排长队。异步编程async/await就像餐厅里的服务员你点完菜他不用等菜做好而是立刻去接待下一桌客人。等厨房喊“菜好了”他再回来端菜。这样一个服务员可以同时服务很多桌。FastAPI就是支持这种异步模式的Python Web框架。我们用它来写两个接口/upload接收用户上传的Markdown文件存入知识库。/query接收用户的问题返回答案。FastAPI还有一个极大的优点启动后自动生成交互式文档/docs你可以直接在网页里测试接口不用额外写API说明。三、环境准备5分钟搭好开发环境3.1 安装Python和虚拟环境请确保你的电脑上安装了Python 3.9或更高版本。打开终端或命令行执行# 创建项目目录 mkdir its_project cd its_project mkdir -p backend/knowledge cd backend/knowledge # 创建虚拟环境Windows python -m venv .venv .venv\Scripts\activate # 如果你是Mac/Linux # python3 -m venv .venv # source .venv/bin/activate3.2 安装依赖包创建 requirements.txt 文件内容如下fastapi0.104.1 uvicorn0.24.0 langchain0.1.0 langchain-community0.0.10 langchain-chroma0.0.1 langchain-openai0.0.2 python-dotenv1.0.0 requests2.31.0 jieba0.42.1 markdownify0.11.6 beautifulsoup44.12.2 scikit-learn1.3.2然后执行pip install -r requirements.txt3.3 项目目录结构先建好这些文件我们将采用清晰的三层架构后面会逐个文件填充代码。你先按下面的样子把目录和空文件建好knowledge/ ├── data_access/ # 数据访问层负责读写 │ ├── __init__.py │ ├── file_repository.py # 读写本地文件、去重 │ ├── knowledge_api_client.py # 调用联想API │ └── vector_store_manager.py # 操作ChromaDB ├── business_logic/ # 业务逻辑层核心 │ ├── __init__.py │ ├── file_processor.py # 加载、切分、入库 │ ├── retrieval_service.py # 混合检索实现 │ └── query_service.py # 构造Prompt、调用LLM ├── presentation/ # 表现层Web接口 │ ├── __init__.py │ ├── api/ │ │ ├── __init__.py │ │ ├── main.py # FastAPI启动入口 │ │ ├── routes.py # /upload 和 /query 路由 │ │ └── schemas.py # 请求/响应数据模型 │ └── cli/ # 命令行工具 │ ├── __init__.py │ └── crawl_cli.py # 爬虫脚本 ├── config/ # 配置 │ ├── __init__.py │ ├── settings.py # 读取环境变量 │ └── constants.py # 常量分块大小等 ├── utils/ # 小工具 │ ├── __init__.py │ └── text_utils.py # 文本清洗、文件名处理 ├── .env # 存放API密钥不要提交到git └── run_server.py # 启动服务的入口提示你可以先用 touch 或手动创建这些文件看着多但每个文件的代码都不长。3.4 配置环境变量.env文件创建 .env 文件填入你的OpenAI兼容API信息如果你没有OpenAI的key可以使用国内中转或本地ollama我们这里先用OpenAI的格式后面可替换# 联想知识库API地址公开接口 KNOWLEDGE_BASE_URLhttps://iknow.lenovo.com.cn # OpenAI兼容的配置支持代理或国内大模型 OPENAI_API_KEYsk-你的key OPENAI_API_BASEhttps://api.openai.com/v1 # 如有代理则修改 EMBEDDING_MODELtext-embedding-3-small LLM_MODELgpt-3.5-turbo # 本地路径 MD_FOLDER_PATH./data/raw VECTOR_STORE_PATH./chroma_kb # 检索参数 TOP_ROUGH20 # 粗排保留候选数 TOP_FINAL5 # 最终给LLM的文档数 VECTOR_SEARCH_K5四、第一步获取原始知识写一个实用的爬虫知识库首先得有知识。我们从联想公开知识库爬取一部分技术问答作为初始数据。4.1 为什么选联想知识库联想官网的 iknow.lenovo.com.cn 收录了大量电脑故障排除、驱动安装、系统教程等真实、结构化的内容。每条知识都有一个唯一的编号knowledgeNo返回的是JSON格式包含标题、正文HTML、分类等字段。我们可以通过 https://iknow.lenovo.com.cn/knowledgeapi/api/knowledge/knowledgeDetails?knowledgeNo编号 获取数据。这样做的好处你不需要自己编造测试数据直接拿到真实内容后续检索效果也更可信。4.2 编写API客户端data_access/knowledge_api_client.py 代码import requests from config.settings import settings class KnowledgeApiClient: 负责向联想知识库发送HTTP请求获取原始JSON数据 staticmethod def fetch_knowledge_content(knowledgeNo: int) - dict: 根据知识编号获取内容 参数: knowledgeNo 知识编号例如 111 返回: 解析后的JSON中的data字段字典 try: # 从settings读取基础URL base_url settings.KNOWLEDGE_BASE_URL url f{base_url}/knowledgeapi/api/knowledge/knowledgeDetails params {knowledgeNo: knowledgeNo} # 发送GET请求10秒超时 response requests.get(urlurl, paramsparams, timeout10) # 如果状态码不是200会抛出异常 response.raise_for_status() # 返回data部分内容主体 return response.json().get(data) except requests.exceptions.RequestException as e: raise Exception(fHTTP请求失败编号{knowledgeNo}原因{e})4.3 清洗HTML转Markdown联想返回的正文是HTML格式其中包含不少无用的标签、广告代码。我们需要把它转成干净的Markdown方便后续处理和阅读。utils/text_utils.py 实现from bs4 import BeautifulSoup from markdownify import markdownify as md import re class TextUtils: staticmethod def html_to_markdown(html_content: str) - str: 将HTML转为Markdown。 先使用BeautifulSoup清洗结构移除script、style、广告块 再合并相邻的加粗标签最后用markdownify转换。 if not html_content: return # 用BeautifulSoup解析 soup BeautifulSoup(html_content, html.parser) # 移除脚本和样式标签 for tag in soup([script, style, noscript]): tag.decompose() # 移除特定的广告块联想页面中的.mceNonEditable for ad in soup.select(.mceNonEditable): ad.decompose() # 合并相邻的strong或b标签例如 strongA/strongstrongB/strong - strongAB/strong bold_tags soup.find_all([strong, b]) for tag in bold_tags: if not tag.parent: # 如果已经被删除则跳过 continue next_sib tag.next_sibling # 判断下一个兄弟是否也是加粗标签 if next_sib and isinstance(next_sib, type(tag)) and next_sib.name tag.name: # 把下一个标签的内容挪到当前标签里 tag.extend(next_sib.contents) next_sib.decompose() cleaned_html str(soup) # 转换为Markdown markdown_text md(cleaned_html) return markdown_text staticmethod def clean_filename(filename: str) - str: 去除Windows文件名中的非法字符 if not filename: return untitled # 非法字符: \ / : * ? | illegal_chars r[\\/:*?|] return re.sub(illegal_chars, -, filename)4.4 文件读写工具data_access/file_repository.py 提供通用的文件保存和读取import os class FileRepository: staticmethod def save_file(content: str, file_path: str): 将内容保存到指定路径自动创建目录 if not content: print(内容为空不保存) return directory os.path.dirname(file_path) if directory: os.makedirs(directory, exist_okTrue) with open(file_path, w, encodingutf-8) as f: f.write(content) staticmethod def read_file_content(file_path: str) - str: 读取文本文件如果出错返回空字符串 if not os.path.exists(file_path): return try: with open(file_path, r, encodingutf-8) as f: return f.read() except Exception as e: print(f读取文件失败: {e}) return staticmethod def list_files(directory: str, extension: str .md) - list: 列出目录下所有指定扩展名的文件完整路径 files [] if not os.path.isdir(directory): return files for fname in os.listdir(directory): if fname.endswith(extension): files.append(os.path.join(directory, fname)) return files4.5 爬虫主脚本presentation/cli/crawl_cli.py 实现命令行调用循环爬取一定编号范围的知识import argparse import os import time from config.settings import settings from data_access.knowledge_api_client import KnowledgeApiClient from data_access.file_repository import FileRepository from utils.text_utils import TextUtils def main(): parser argparse.ArgumentParser(description爬取联想知识库生成Markdown文件) parser.add_argument(--start, typeint, requiredTrue, help起始编号) parser.add_argument(--end, typeint, requiredTrue, help结束编号) parser.add_argument(--out, typestr, default./data/raw, help输出目录) parser.add_argument(--delay, typefloat, default0.2, help请求间隔秒) args parser.parse_args() # 创建输出目录 os.makedirs(args.out, exist_okTrue) success 0 fail 0 for no in range(args.start, args.end 1): print(f正在获取编号 {no} ...) try: data KnowledgeApiClient.fetch_knowledge_content(no) if data and data.get(content): # 转Markdown md_content TextUtils.html_to_markdown(data[content]) # 获取标题并清洗 raw_title data.get(title, 无标题) clean_title TextUtils.clean_filename(raw_title) if len(clean_title) 50: clean_title clean_title[:50].rstrip(_) # 文件名格式编号-标题.md filename f{no:04d}-{clean_title}.md filepath os.path.join(args.out, filename) FileRepository.save_file(md_content, filepath) success 1 print(f 保存成功: {filename}) else: fail 1 print(f 编号{no}无内容) except Exception as e: fail 1 print(f 出错: {e}) # 休眠避免请求过快 time.sleep(args.delay) print(f\n爬取完成成功: {success}, 失败: {fail}) if __name__ __main__: main()如何使用在终端执行 python crawl_cli.py --start 1 --end 50就会在 ./data/raw 下生成最多50个Markdown文件。五、第二步知识入库——从文件到向量现在我们已经有一批 .md 文件散落在 data/raw 里。接下来要把它们变成向量存入ChromaDB。5.1 为什么需要切分切多大合适大模型都有上下文窗口限制例如4096个token。如果一篇文档很长直接喂给模型会超过限制而且检索时整篇文档的主题太泛与你具体问题的相关性会被稀释。但是我们的知识条目普遍较短1000~3000字强行切分成500字的小块反而会把完整的解决方案拆碎。所以策略是能不切就不切让一个文件作为一个chunk。对于极少数超长文档再用切分器按段落边界切分。我们在 config/constants.py 中定义# 分块大小字符数设为3000足够覆盖大部分知识条目 CHUNK_SIZE 3000 # 重叠字符数避免边界信息丢失 CHUNK_OVERLAP 2005.2 向量库管理器VectorStoreManagerdata_access/vector_store_manager.py 封装了ChromaDB的所有操作包括初始化Embedding模型和增删改查。from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma from config.settings import settings class VectorStoreManager: def __init__(self): # 1. 初始化嵌入模型OpenAI格式可换本地模型 self.embeddings OpenAIEmbeddings( modelsettings.EMBEDDING_MODEL, api_keysettings.API_KEY, base_urlsettings.BASE_URL ) # 2. 创建或加载Chroma向量库 # persist_directory指定持久化文件夹重启后数据不丢失 self.vector_store Chroma( persist_directorysettings.VECTOR_STORE_PATH, embedding_functionself.embeddings, collection_nameknowledge_base ) def add_documents(self, documents, batch_size32) - int: 批量添加LangChain Document对象到向量库 返回成功添加的数量 total len(documents) if total 0: return 0 for i in range(0, total, batch_size): batch documents[i:ibatch_size] self.vector_store.add_documents(batch) print(f已存入 {min(ibatch_size, total)}/{total} 条) return total def get_retriever(self, k5): 返回一个检索器可用于相似度搜索 return self.vector_store.as_retriever(search_kwargs{k: k})5.3 文件处理与入库核心逻辑business_logic/file_processor.py 将单个Markdown文件加载、切分、然后调用VectorStoreManager入库。from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.vectorstores.utils import filter_complex_metadata from data_access.vector_store_manager import VectorStoreManager from config.constants import CHUNK_SIZE, CHUNK_OVERLAP class FileProcessor: def __init__(self): self.vector_manager VectorStoreManager() # 初始化文本切分器按段落、句子等边界递归切分 self.text_splitter RecursiveCharacterTextSplitter( chunk_sizeCHUNK_SIZE, chunk_overlapCHUNK_OVERLAP, separators[\n\n, \n, , ], keep_separatorFalse ) def process_and_save_file(self, file_path: str) - int: 处理一个文件加载 - 切分 - 过滤 - 入库 返回成功入库的chunk数量 # 1. 加载文件UTF-8编码 try: loader TextLoader(file_path, encodingutf-8) docs loader.load() except Exception as e: print(f文件加载失败: {e}) raise e # 2. 切分文档如果文档短只会得到一个chunk chunks self.text_splitter.split_documents(docs) if not chunks: return 0 # 3. 过滤复杂元数据LangChain要求 filtered filter_complex_metadata(chunks) filtered [doc for doc in filtered if doc.page_content.strip()] if not filtered: return 0 # 4. 存入向量库 added self.vector_manager.add_documents(filtered) return added六、第三步智能查询——混合检索重排序有了向量库用户提问时我们不是简单地从Chroma里拿几个最相似的chunk因为向量检索对关键词如具体型号、错误代码有时不够敏感。所以我们设计了两路召回向量检索路从Chroma中取Top-K个语义相似的chunk。标题检索路利用本地Markdown文件的标题高度概括了知识要点做关键词匹配找到最相关的完整文件。然后把两路结果合并去重最后用一个重排序步骤重新计算所有候选与问题的余弦相似度得到最终Top-N。6.1 检索服务代码business_logic/retrieval_service.py 的代码较长我会拆开讲。import os import re import jieba from typing import List, Dict from sklearn.metrics.pairwise import cosine_similarity from langchain_core.documents import Document from data_access.vector_store_manager import VectorStoreManager from config.settings import settings class RetrievalService: def __init__(self): self.vector_manager VectorStoreManager() self.k_final settings.TOP_FINAL # ---------- 辅助收集本地MD文件的元数据 ---------- def collect_md_metadata(self, folder_path: str) - List[Dict]: 扫描文件夹下所有.md文件提取路径和标题 metadata [] if not os.path.isdir(folder_path): return metadata # 文件名格式: 编号-标题.md pattern re.compile(r^.?-(.*?)\.md$) for fname in os.listdir(folder_path): if fname.endswith(.md): match pattern.match(fname) title match.group(1) if match else os.path.splitext(fname)[0] metadata.append({ path: os.path.join(folder_path, fname), title: title }) return metadata # ---------- 粗排基于分词字符重叠 ---------- def rough_ranking(self, md_list: List[Dict], question: str) - List[Dict]: 对标题进行粗排返回TOP_ROUGH个候选 if not question.strip(): for item in md_list: item[rough_score] 0 return sorted(md_list, keylambda x: x[rough_score], reverseTrue)[:settings.TOP_ROUGH] WEIGHT_WORD 0.7 # 分词重叠权重 for item in md_list: title item[title].strip() if not title: item[rough_score] 0 continue # 字符级别重叠 q_chars set(question) t_chars set(title) char_sim len(q_chars t_chars) / (len(q_chars) 1e-6) # 分词级别重叠使用jieba q_words set(jieba.lcut(question)) t_words set(jieba.lcut(title)) word_sim len(q_words t_words) / (len(q_words) 1e-6) # 综合得分 item[rough_score] WEIGHT_WORD * word_sim (1 - WEIGHT_WORD) * char_sim sorted_list sorted(md_list, keylambda x: x[rough_score], reverseTrue) return sorted_list[:settings.TOP_ROUGH] # ---------- 精排用Embedding计算语义相似度 ---------- def fine_ranking(self, rough_results: List[Dict], question: str) - List[Dict]: 对粗排结果进行语义精排返回TOP_FINAL个 if not rough_results: return [] # 获取问题向量 q_emb self.vector_manager.embeddings.embed_query(question) titles [item[title] for item in rough_results] title_embs self.vector_manager.embeddings.embed_documents(titles) # 计算余弦相似度 sims cosine_similarity([q_emb], title_embs).flatten() for i, item in enumerate(rough_results): sem_score float(sims[i]) rough_score item.get(rough_score, 0) # 混合分数粗排和语义各占一半 item[combined_score] 0.5 * rough_score 0.5 * sem_score item[semantic_score] sem_score return sorted(rough_results, keylambda x: x[combined_score], reverseTrue)[:settings.TOP_FINAL] # ---------- 主检索函数 ---------- def retrieve(self, question: str) - List[Document]: 执行混合检索返回最终Top-K个Document candidates [] # 路线1向量库检索搜索chunk retriever self.vector_manager.get_retriever(ksettings.VECTOR_SEARCH_K) vector_docs retriever.invoke(question) candidates.extend(vector_docs) # 路线2标题匹配检索搜索完整文件 if os.path.exists(settings.MD_FOLDER_PATH): meta self.collect_md_metadata(settings.MD_FOLDER_PATH) rough self.rough_ranking(meta, question) fine self.fine_ranking(rough, question) for item in fine[:5]: # 最多取5个文件 content FileRepository.read_file_content(item[path]) if content: doc Document(page_contentcontent, metadata{ source: item[path], title: item[title] }) candidates.append(doc) # 去重基于来源内容前100字符 seen set() unique [] for doc in candidates: key (doc.metadata.get(source, ), doc.page_content[:100]) if key not in seen: seen.add(key) unique.append(doc) if not unique: return [] # 统一重排序用embedding计算所有候选与问题的相似度 q_emb self.vector_manager.embeddings.embed_query(question) texts [doc.page_content for doc in unique] text_embs self.vector_manager.embeddings.embed_documents(texts) sims cosine_similarity([q_emb], text_embs).flatten() # 按相似度排序 scored [(unique[i], sims[i]) for i in range(len(unique))] scored.sort(keylambda x: x[1], reverseTrue) final_docs [doc for doc, _ in scored[:self.k_final]] return final_docs6.2 问答服务给LLM戴上“紧箍咒”检索到的文档不能直接扔给大模型而是要用一个精心设计的Prompt明确告诉模型“只能根据资料回答不许瞎编”。business_logic/query_service.pyimport re from langchain_openai import ChatOpenAI from langchain_core.documents import Document from config.settings import settings def clean_markdown_images(text: str) - str: 将Markdown图片语法 ![描述](url) 替换为纯url每图一行 pattern r!\[[^\]]*\]\((https?://[^\s\)])\) cleaned re.sub(pattern, r\n\1\n, text) # 合并多余空行 cleaned re.sub(r\n{3,}, \n\n, cleaned) return cleaned.strip() class QueryService: def __init__(self): self.llm ChatOpenAI( modelsettings.LLM_MODEL, temperature0, # 设为0保证答案稳定不随机发挥 api_keysettings.API_KEY, base_urlsettings.BASE_URL ) def generate_answer(self, question: str, context_docs: List[Document]) - str: if not context_docs: return 暂时没有找到相关知识请先上传文档。 # 拼接上下文 context_text \n\n.join([ f【文档{i1}】\n{doc.page_content} for i, doc in enumerate(context_docs) ]) # 关键Prompt模板 prompt f 你是一个技术支持专家。请严格按照下面提供的资料回答用户的问题。 **规则** 1. 只能使用资料中的信息不能自己编造任何内容。 2. 如果资料中找不到答案请直接回复“资料中未提及相关信息”。 3. 资料中的图片链接必须保留但不要用[描述](链接)的格式直接写出完整的URL每张图片单独一行。 4. 回答要简洁、步骤清晰不要附带无关品牌信息。 5. 不要使用Markdown格式的图片语法。 **资料内容** {context_text} **用户问题** {question} **你的回答** try: response self.llm.invoke(prompt) answer response.content cleaned clean_markdown_images(answer) return cleaned except Exception as e: print(fLLM调用失败: {e}) return 抱歉生成回答时出错。七、第四步暴露Web接口——/upload和/query为了让前端或用户能实际调用我们用FastAPI将上述能力封装成REST API。7.1 定义请求/响应模型presentation/api/schemas.pyfrom pydantic import BaseModel class UploadResponse(BaseModel): status: str message: str file_name: str chunks_added: int class QueryRequest(BaseModel): question: str class QueryResponse(BaseModel): question: str answer: str7.2 实现路由presentation/api/routes.pyimport os import tempfile from fastapi import APIRouter, UploadFile, File, HTTPException from .schemas import UploadResponse, QueryRequest, QueryResponse from business_logic.file_processor import FileProcessor from business_logic.retrieval_service import RetrievalService from business_logic.query_service import QueryService router APIRouter() file_processor FileProcessor() retriever RetrievalService() query_service QueryService() router.post(/upload, response_modelUploadResponse) async def upload_file(file: UploadFile File(...)): 上传Markdown文件自动入库 # 保存上传文件到临时目录 suffix os.path.splitext(file.filename)[1] with tempfile.NamedTemporaryFile(deleteFalse, suffixsuffix) as tmp: content await file.read() tmp.write(content) tmp_path tmp.name try: added file_processor.process_and_save_file(tmp_path) return UploadResponse( statussuccess, message文件已存入知识库, file_namefile.filename, chunks_addedadded ) except Exception as e: raise HTTPException(status_code500, detailf处理失败: {str(e)}) finally: if os.path.exists(tmp_path): os.remove(tmp_path) router.post(/query, response_modelQueryResponse) async def query_knowledge(req: QueryRequest): if not req.question.strip(): raise HTTPException(status_code400, detail问题不能为空) try: docs retriever.retrieve(req.question) answer query_service.generate_answer(req.question, docs) return QueryResponse(questionreq.question, answeranswer) except Exception as e: raise HTTPException(status_code500, detailstr(e))7.3 FastAPI入口presentation/api/main.pyfrom fastapi import FastAPI from .routes import router app FastAPI( titleRAG知识库API, description支持上传文档和智能问答, version1.0 ) app.include_router(router)7.4 启动脚本run_server.pyimport uvicorn if __name__ __main__: uvicorn.run(presentation.api.main:app, host0.0.0.0, port8001, reloadTrue)启动服务python run_server.py打开浏览器访问 http://localhost:8001/docs你会看到自动生成的Swagger界面可以直接测试/upload和/query。八、测试一下看看效果8.1 先入库几个测试文档你可以手动写一个简单的Markdown文件内容如下保存为 test.md# 如何解决电脑蓝屏 蓝屏错误通常由驱动或硬件故障引起。 解决方案 1. 重启电脑。 2. 进入安全模式卸载最近安装的驱动。 3. 运行内存检测工具。然后用/upload接口上传。成功后返回 chunks_added1。8.2 提问调用/querybody为 {question: 电脑蓝屏了怎么办}。你会看到返回的答案基于你的文档内容并且没有编造。8.3 使用爬虫批量导入如果你想测试真实数据可以先跑爬虫爬取少量数据如 --start 1 --end 10然后再提问。你会发现系统能够从联想官方文档中找到答案。九、总结与进阶建议9.1 我们完成了什么通过这篇文章你亲手搭建了一个完整的RAG知识库系统其中包括一个可扩展的爬虫从真实网站获取结构化知识。一个基于LangChain和ChromaDB的向量化入库流水线。一个包含混合检索向量标题和重排序的高精度查询服务。一个自动生成文档的FastAPI接口可直接对接前端。这套架构已经在多个企业级项目中得到验证足以应对大部分知识问答场景。9.2 你可以继续优化的方向替换Embedding模型如果你不想依赖OpenAI可以换成 sentence-transformers 或本地Ollama只需修改 VectorStoreManager 中的 embeddings 初始化。增加更丰富的文档格式目前只支持 .md你可以用 UnstructuredFileLoader 支持PDF、Word等。引入更专业的重排序模型例如 Cohere Rerank 或 BGE-reranker通常在精排阶段可提升5-10%的命中率。增加缓存对相同或相似的问题进行结果缓存提升响应速度。前端界面可以配合Vue/React项目打造类似ChatGPT的对话界面。9.3 最后的话RAG之所以成为当前LLM落地的“标准答案”因为它用最小的成本赋予了大模型“查阅资料”的能力。只要你的知识库足够丰富、检索足够精准大模型就能给出令人信服的答案。希望你不仅看懂了代码更理解了RAG背后的设计哲学。现在你可以用这套工具箱去解决自己业务中的实际问题了。如果在实践过程中遇到任何卡点回想一下这篇文章中的三个核心比喻——“图书馆查资料”、“菜谱炒菜”、“异步服务员”——它们会帮你理清思路。祝你编码愉快早日做出属于自己的智能客服附录常见错误排查ModuleNotFoundError检查虚拟环境是否激活依赖是否完整安装。ChromaDB无法保存确保 VECTOR_STORE_PATH 目录有写入权限且路径不含中文。API请求超时检查网络或适当放大 timeout 参数。检索结果为空先确认至少有一个文档成功入库控制台会打印日志再检查提问的关键词是否与文档内容明显相关。记住调试RAG系统时可以先单独测试检索模块打印出 retrieve 返回的文档内容确认检索没问题后再排查生成模块。祝你好运