说实话我刚开始折腾个人知识库的时候市面上大部分教程都在讲概念、讲架构图真正能从零跑起来、把私有文档喂进去、然后让大模型给出靠谱答案的完整链路反而很难找到。我自己的资料散落在 Markdown、PDF、网页剪藏里真到用的时候翻半天也搜不出一句有用的话。后来花了一个周末把 RAG检索增强生成这套东西从原理到代码完整捋了一遍基于 FastAPI LangChain LangGraph pgvector 搭了一个本地私有知识库现在把整套实战过程、代码、还有踩过的坑一起写出来。这篇文章不是那种只讲概念的水文我会把需求分析、选型理由、每一步代码怎么来的、参数为什么这么定都交代清楚。适合有一点点 Python 基础、想搭一个真正能用的本地知识库、又不想被各种模糊说法绕晕的开发者。1. 先想清楚你的私有知识库到底要解决什么问题动手之前我建议你先花十分钟回答一个问题现有工具到底哪里不够用很多人对知识库的想象是把文件丢进去问什么它答什么。但真上手之后你会发现文件格式、问题类型、答案要求、数据量级、是否需要多轮追问每一个变量都会改变技术方案。拿我自己举例我的需求很明确资料散落在本地和私有服务器不能传到公网服务商那里文档以 Markdown、TXT、PDF 为主总量在几千个文件级别提问方式是自然语言且经常需要连续追问回答里要能指出依据来自哪篇文档方便我溯源这其实是一个 私有化部署 文档问答 可追溯 的组合需求。传统方案里用 grep 或全文搜索只能做关键词匹配搜出几百条结果还得自己一条条翻把文档全部丢给大模型做长文本输入成本高、超长上下文又撑不住。RAG 恰好是折中路线先把文档切块、向量化、存进数据库提问时先召回最相关的片段再让大模型基于这些片段生成答案。这里有两个容易被误解的点我一开始也没绕明白第一RAG 不是让大模型记住你的文档而是给大模型一份开卷考试的参考书。模型本身参数不变变的是每次提问时检索出来的上下文。所以它的好处是更新文档不需要重新训练模型换一批资料只需要重新灌库。第二私有知识库的核心价值其实不在生成而在检索。如果你检索出来的片段本身就不相关后面大模型再怎么生成都是瞎编。这也决定了整个项目的重心切块策略和检索质量远比模型大小重要。基于上面这些分析我把技术目标定成了这样支持本地部署数据不出内网提供 HTTP 接口方便后续接到其他内部工具上有基础的多轮对话能力而不是每次提问都失忆回复能带上来源片段方便我核对目标定完之后再来看技术方案的选择思路就清楚很多了。2. RAG 和 MCP 的区别以及RAG 必须用 API 吗这个经典问题最近社区里 RAG 和 MCP 这两个词经常被放在一起讨论不少新人被绕晕了。我先用一个简单类比说清楚RAG 是给模型准备参考书让它在作答前先查资料MCP 是给模型接上手脚让它能调用外部工具去执行操作。前者解决知识不够新、不够全的问题后者解决模型只能聊天、不能干活的问题。打个比方你问模型帮我查一下上个月的销售数据并画成图表。RAG 负责把数据库文档里的销售数据片段喂给模型供它作答MCP 则让模型主动去连接销售数据库、唤起绘图工具。二者不是替代关系而是可以叠加使用的关系。很多 Agentic RAG 项目里RAG 负责检索知识MCP 负责调用工具执行后续动作。另一个高频疑问是RAG 必须用 API 吗答案是不必须。RAG 的完整链路包括向量化、存储、检索、生成四步其中只有生成这一步必须有大模型参与但大模型可以是自托管的开源模型如 Qwen、Llama 系列向量化也可以用本地 Embedding 模型完成。也就是说整套 RAG 完全可以离线跑通数据不出内网。如果你想彻底本地化选一个能在自己机器上跑起来的 Embedding 模型和 LLM 就行如果你对数据不敏感生成环节接云端 API 也可以但那样就不算严格意义的私有了。我这次选择的是混合方案向量化和检索全部本地化生成环节用本地 LLM 的 OpenAI 兼容接口这样既保住了隐私又不影响后续接入更强大的模型。3. 技术选型FastAPI LangChain LangGraph pgvector每一步都有理由选型这事最怕上来就抄一堆网红框架结果部署的时候发现复杂度远超实际需求。我的选择标准只有一条在个人数据量级 私有化部署 可维护性这个约束下选最省心的组合。3.1 为什么用 pgvector 而不是 Milvus、Chroma向量数据库是 RAG 的存储底座市面上选择很多我把主流的几个放在一起过了一遍方案部署复杂度适合场景说明pgvector低PostgreSQL 插件个人/中小团队数据量在百万级向量内完全够用还能直接用 SQL 做业务字段过滤Chroma低原型验证纯本地文件式存储跑 demo 快但生产级能力弱Milvus中高大规模生产分布式能力强但需要额外维护一套集群Elasticsearch中高全文检索向量混合检索功能全但资源占用大个人场景偏重我最终选了 pgvector核心原因有三个。第一我本来就在用 PostgreSQL装一个插件就能把业务数据和向量数据存在同一个库里不用再维护一套新系统。第二pgvector 支持 SQL 里直接写 WHERE 条件过滤文档来源、时间、权限等字段这对私有知识库来说非常实用——你可以轻松做到只搜索某个月份的文档或只搜索某个目录下的文档。第三备份和恢复直接用 PostgreSQL 的机制省掉了向量库单独备份的麻烦。Milvus 确实更强大但它是为亿级向量、高并发、分布式部署设计的个人知识库这个量级属于杀鸡用牛刀。Chroma 更适合快速验证真到要长期维护的时候文件式存储还是不如 PostgreSQL 稳。3.2 LangChain 和 LangGraph 的分工LangChain 在 RAG 社区里几乎是标配它把文档加载、切块、向量化、检索、提示词组装这些步骤都封装成了标准化组件省掉大量样板代码。但只用 LangChain 的话多轮对话和条件分支这类逻辑会写得比较绕。这次项目里我引入了 LangGraph因为设计目标里包含了多轮对话和后续 Agent 化扩展LangGraph 把整个问答流程定义成一张图节点和节点之间的状态流转一目了然调试的时候能清楚地看到数据走到了哪一步。如果你的需求只是一个单轮问答 demoLangGraph 确实可以不用但要做多轮会话管理或者更复杂的 Agentic RAG用 LangGraph 的状态图会让代码的可维护性提升一个档次。3.3 大模型和 Embedding 模型的选型这次用的是 Qwen 系列开源模型跑本地推理通过 Ollama 暴露 OpenAI 兼容接口。Embedding 模型用了开源的 bge-m3中英文效果都不错而且支持 8192 长度的输入对大多数文档切块场景足够。这两个模型都能完全本地运行数据不会出网。如果你机器配置一般Embedding 可以用更轻量的 bge-small-zh生成模型可以用 Qwen 的 7B 量化版。个人知识库场景下Embedding 的质量对检索效果的影响比生成模型更大所以 Embedding 尽量别用太小的生成模型倒是可以适当取舍。4. 完整代码实现从建库到跑通问答下面进入正题。整个项目我拆成了几个模块依赖安装与数据库准备、文档处理加载与切块、向量化与入库、FastAPI 接口层、LangGraph 问答链路。我会把关键代码贴出来并解释每一段的意图最后给出启动方式和验证方法。4.1 环境准备依赖、数据库、目录结构# 创建虚拟环境 python -m venv .venv source .venv/bin/activate # 安装依赖 pip install fastapi uvicorn langchain langchain-community langgraph \ pgvector psycopg2-binary python-multipart \ sentence-transformers openai这里我用 sentence-transformers 直接加载本地 Embedding 模型openai 库用来统一适配 OpenAI 兼容接口既可以用本地 Ollama也可以平滑切换到云端。PostgreSQL 需要先安装并启动然后启用 pgvector 插件并创建数据库# 进入 PostgreSQL 后执行 CREATE EXTENSION IF NOT EXISTS vector; CREATE DATABASE knowledge_db;工程目录结构如下rag-knowledge-base/ ├── app/ │ ├── main.py # FastAPI 入口 │ ├── ingest.py # 文档加载、切块、入库 │ ├── retriever.py # 检索逻辑 │ ├── graph.py # LangGraph 问答链路 │ └── config.py # 配置项 ├── docs/ # 放你需要灌入的文档 ├── requirements.txt └── .env4.2 文档加载与切块最影响效果的一步切块是整个 RAG 链路里最脏活累活但影响最大的环节。切得太碎语义不完整切得太长向量化时会被无关信息稀释检索精度下降。我的经验是先按文档结构切再按大小兜底。from langchain_community.document_loaders import DirectoryLoader, TextLoader, PyPDFLoader from langchain.text_splitter import RecursiveCharacterTextSplitter def load_documents(docs_dir: str): loaders [ DirectoryLoader(docs_dir, glob**/*.md, loader_clsTextLoader, loader_kwargs{encoding: utf-8}), DirectoryLoader(docs_dir, glob**/*.txt, loader_clsTextLoader, loader_kwargs{encoding: utf-8}), DirectoryLoader(docs_dir, glob**/*.pdf, loader_clsPyPDFLoader), ] docs [] for loader in loaders: docs.extend(loader.load()) return docs def split_documents(docs): text_splitter RecursiveCharacterTextSplitter( chunk_size512, chunk_overlap64, separators[\n\n, \n, 。, , , ., , ], ) chunks text_splitter.split_documents(docs) return chunks几个参数说明一下都是我实际调出来的经验值chunk_size 取 512 字符。这个值对中文比较友好短了语义容易断长了向量表征会被稀释。如果你用的是英文文档可以适当放大到 800-1000。chunk_overlap 取 64。overlap 的作用是防止切块把一句话从中间截断让上下文有一定重叠。一般取 chunk_size 的 10%-20% 比较合理。separators 的顺序有讲究。RecursiveCharacterTextSplitter 会按这个顺序优先找分隔符切分我把换行和中英文句号放在前面就是为了尽量在语义完整的位置切断。PDF 加载这里要提醒一句PyPDFLoader 对扫描版 PDF 无能为力那种情况需要 OCR 预处理本文不展开。如果你的 PDF 都是非扫描版用它就够了。4.3 向量化入库pgvector 表结构与写入逻辑向量化部分用 sentence-transformers 加载 bge-m3将切好的文档块转成向量后写入 pgvector。from sentence_transformers import SentenceTransformer import psycopg2 from pgvector.psycopg2 import register_vector embedder SentenceTransformer(BAAI/bge-m3) def get_embedding(text: str): # bge 模型建议加上查询指令前缀来区分检索和入库 return embedder.encode(text, normalize_embeddingsTrue) def create_table_if_not_exists(conn): with conn.cursor() as cur: cur.execute( CREATE TABLE IF NOT EXISTS document_chunks ( id BIGSERIAL PRIMARY KEY, content TEXT NOT NULL, source TEXT NOT NULL, embedding vector(1024) NOT NULL, created_at TIMESTAMP DEFAULT now() ); ) conn.commit() def insert_chunks(chunks): conn psycopg2.connect(dsn) register_vector(conn) create_table_if_not_exists(conn) with conn.cursor() as cur: for chunk in chunks: vec get_embedding(chunk.page_content) cur.execute( INSERT INTO document_chunks (content, source, embedding) VALUES (%s, %s, %s), (chunk.page_content, chunk.metadata.get(source, unknown), vec) ) conn.commit()注意 bge-m3 输出维度是 1024建表时 vector 维度必须匹配否则插入报错。如果换成 bge-small-zh维度是 512需要改表结构。4.4 FastAPI 接口层文件上传和问答接口层我做了两个最基础也最核心的端点一个负责上传文档并触发入库一个负责接收问题并返回答案。生成环节我留了一个抽象层方便切换本地模型和远端模型。from fastapi import FastAPI, UploadFile, File, Form from pydantic import BaseModel app FastAPI(titlePrivate RAG Knowledge Base) app.post(/upload) async def upload_doc(file: UploadFile File(...)): # 实际项目中建议先落盘再加载避免大文件占用内存过多 content await file.read() save_path fdocs/{file.filename} with open(save_path, wb) as f: f.write(content) # 重新加载文档目录并增量入库本文简化为全量重建 chunks load_and_split() insert_chunks(chunks) return {message: ingest done, chunks: len(chunks)} class QueryBody(BaseModel): question: str history: list [] # [{role: user, content: ...}, ...] app.post(/query) async def query(body: QueryBody): answer, sources run_rag(body.question, body.history) return {answer: answer, sources: sources}这里我把 UploadFile 直接读了内存数据量大时建议先存临时文件再处理。个人知识库场景够用但如果上传的 PDF 超过几十 MB建议改成先落盘再加工。4.5 检索逻辑向量相似度 元数据过滤检索这一步是 RAG 的核心。我用 pgvector 的余弦距离做相似度排序同时支持按来源过滤。top_k 我一般取 4-6太少容易漏太多会塞入大量无关片段干扰生成。def retrieve(query: str, top_k: int 5, source_filter: str None): conn psycopg2.connect(dsn) register_vector(conn) query_vec get_embedding(query) sql SELECT content, source, 1 - (embedding %s) AS similarity FROM document_chunks params [query_vec] if source_filter: sql WHERE source %s params.append(source_filter) sql ORDER BY similarity DESC LIMIT %s params.append(top_k) with conn.cursor() as cur: cur.execute(sql, params) rows cur.fetchall() return [{content: r[0], source: r[1], similarity: r[2]} for r in rows]pgvector 的 运算符计算余弦距离1 - 距离就是相似度。如果有权限或目录过滤需求在 WHERE 后面追加条件就行这也是我选 pgvector 的一个重要理由。4.6 LangGraph 问答链路把检索和生成串成状态图我用 LangGraph 把整个问答流程定义成三个节点改写问题、检索、生成。多轮对话时先把历史对话和当前问题合并让模型生成一个独立的检索问题避免它多少钱这种指代性问题在检索阶段失效。from langgraph.graph import StateGraph, END from typing import TypedDict class RAGState(TypedDict): question: str history: list rewritten_question: str context_docs: list answer: str def rewrite_node(state: RAGState) - dict: if not state.get(history): return {rewritten_question: state[question]} prompt ( 根据对话历史和当前问题生成一个独立且完整的检索问题。\n f历史{state[history]}\n f当前问题{state[question]}\n 只输出改写后的问题 ) rewritten llm.invoke(prompt).content.strip() return {rewritten_question: rewritten} def retrieve_node(state: RAGState) - dict: docs retrieve(state[rewritten_question], top_k5) return {context_docs: docs} def generate_node(state: RAGState) - dict: context \n\n.join([f[来源 {d[source]}]\n{d[content]} for d in state[context_docs]]) prompt ( 你是一个严谨的问答助手。请基于以下参考资料回答问题。\n f参考资料\n{context}\n\n f问题{state[question]}\n\n 如果资料中没有可靠依据请明确说资料库中没有找到相关内容不要编造。 ) answer llm.invoke(prompt).content.strip() return {answer: answer} graph StateGraph(RAGState) graph.add_node(rewrite, rewrite_node) graph.add_node(retrieve, retrieve_node) graph.add_node(generate, generate_node) graph.set_entry_point(rewrite) graph.add_edge(rewrite, retrieve) graph.add_edge(retrieve, generate) graph.add_edge(generate, END) rag_app graph.compile()这一段是 LangGraph 的核心用法StateGraph 里的每个节点接收当前状态返回需要更新的字段边定义流转方向。这样做的最大好处是后续想在检索之后加一个判断是否还需要再检索一次的节点只需要在 generate 之前插入一个条件边不用改动其他代码。4.7 启动项目并验证效果uvicorn app.main:app --host 0.0.0.0 --port 8000先用 FastAPI 自带的文档页面/docs上传几个测试文档然后在 /query 接口发一个问题试试。我第一次跑通时用了一篇自己写的项目方案文档问这个方案的架构分几个模块模型不仅答对了还指出来源文件名那一刻还是很有成就感的。5. 切块与召回参数调试我在实际踩坑中总结出来的经验前面代码给的是最终参数但实际调试过程并不是一帆风顺的我把几个影响最大的坑单独拎出来说。5.1 切块大小512 不是万能答案但适合大多数场景我一开始图省事chunk_size 设成了 1024想着上下文长一点模型总能找到答案。结果发现检索返回的片段里经常混着大段无关内容一篇 5000 字的文档被切成了 5 块每块内部话题跨度大导致向量表征被平均化跟问题的相关性反而下降。后来改成 512并且强制优先在段落边界切分效果立竿见影。经验是如果你的文档以技术方案、说明文档为主512-768 是个甜蜜区间如果是碎片化的笔记每条几十到一两百字切块反而没必要直接每条作为一个 chunk 更好。5.2 overlap 少了会断句多了会冗余overlap 的作用可以用一句话说清给相邻切块之间搭一座桥。如果 overlap 是 0一个句子可能在上一块的末尾被切断一半下一块又从句子中间开始两块的向量都会被污染。overlap 太大又会让相邻块大量重复检索回来的 top_k 里可能出现内容高度重叠的片段。我的建议是固定间隔测试64、128、256 各跑一轮观察同一问题的召回结果。最后 64 胜出因为我把中文句号作为分隔符放在较高优先级后大多数切块已经能在完整句子处断开overlap 只是兜底。5.3 召回数量 top_k不是越多越好top_k 这个参数很多人会忽略。我刚开始设 10想让模型有更多参考结果答案反而变散因为它把低相关的片段也塞进了提示词。降到 5 之后回答质量明显提升。一个可行的方法是先跑召回测试打印出相似度分数观察从第几个开始相似度明显下跌。断崖式下跌的位置基本就是 top_k 的上限。另外如果你在 pgvector 里用了相似度阈值比如低于 0.5 直接丢弃能避免模型拿到一堆硬凑的片段。5.4 中文文档的分隔符顺序值得单独调LangChain 默认分隔符列表对中文不够友好它优先按英文句点和换行切分中文长文本经常在奇怪的位置被切断。我调整后的顺序是段落换行、单个换行、中文句号、中文问号叹号、英文句号、空格。这样切出来的块基本都能保证一个完整语义单元。6. 多轮对话与 Agentic RAG从单次问答向会追问演进很多人的知识库跑通单轮问答后就想加多轮对话我也是这个路径。但多轮对话不是简单地把历史消息全塞给大模型就行它会给 RAG 链路带来两个问题检索条件被历史稀释。如果用户连续问这个方案有哪几个模块性能怎么样第二问如果不做改写直接拿性能怎么样去检索召回结果会飘上下文膨胀。历史越长提示词越长很多本地小模型会被超长上下文拖垮响应变慢甚至报错我的方案是在 LangGraph 里加一个改写节点就是上面代码里的 rewrite_node先把当前问题 最近几轮历史压缩成一个独立的检索问题再做向量检索。这里有一个取舍别把全部历史都塞进去只保留最近 2-3 轮即可。再进一步就是 Agentic RAG。传统 RAG 是固定的检索-生成两步Agentic RAG 让模型自己决定要不要检索、要不要追问、要不要换一种检索方式。LangGraph 的条件边就是为这个设计的def should_retrieve_again(state: RAGState) - str: if state.get(need_more_context): return retrieve return generate graph.add_conditional_edges(generate, should_retrieve_again, {retrieve: retrieve, generate: END})这个条件边的意思是生成节点输出后如果模型自评认为资料不足可以再回检索节点重新检索一次。这种能力在文档特别多、一次检索很难命中时很实用。不过要提醒一句Agentic RAG 不是银弹它会让调用次数和延迟成倍增加个人知识库场景下先把单轮 RAG 调到靠谱再考虑 Agent 化步子不要迈太大。7. 跑通之后我复盘了几个早知道就好了的经验项目能跑起来是一回事跑得稳、可维护是另一回事。最后分享几个我在实际使用中积累的判断。关于RAG 要不要上重框架这件事我的结论是个人知识库场景下技术选型应该往少维护方向走而不是往热门方向走。pgvector FastAPI LangChain 这套组合看起来不炫但胜在稳定、社区大、出问题容易搜到解决方案。等数据量真到了几百万向量再迁移到 Milvus 也不迟。关于回答质量我日常使用中最有效的优化手段不是换更强的生成模型而是定期检查召回结果。FastAPI 接口里返回的 sources 字段要经常看如果答案不好先别怪模型看看召回的前 5 条是不是真的相关。大部分时候问题出在切块策略上改完切块答案立刻变好。最后是一条扩展建议如果你觉得关键词检索不够准可以试试在图谱 RAG 方向上做文章把文档里的实体和关系抽出来存成知识图谱再结合向量检索做混合召回。我在一个 200 篇文档的测试集上对比过加了图谱之后涉及多个实体交叉的问题准确率提升明显。不过那已经是更高阶的话题了这篇先写到这里建议你先把基础链路跑通、把切块和召回调好再考虑要不要往那个方向深入。