1. 本地知识库 RAG 到底解决什么问题本地知识库 RAG检索增强生成说白了就是给大模型配一个随身资料柜你把自己的文档、手册、工单、产品说明塞进向量库用户提问时先从库里捞出最相关的几段再连同问题一起喂给模型让它基于这些真实材料回答而不是凭空编。它适合三类人一是手里有大量内部文档但不想上传到公有云的团队二是想用本地显卡跑推理、对数据出境敏感的企业三是想搞明白 RAG 每个环节怎么串起来的开发者。我这次要跑通的链路是LangChain 做编排Hugging Face 三件套Transformers、Datasets、Tokenizers负责模型与数据本地大模型ChatGLM3-6B 或千问系列做推理Chroma 做向量检索MCP 协议接入外部工具最后用 Agent 把检索链和工具调度起来。整套东西跑在本地问答闭环不依赖外部接口。下面按装依赖 → 配向量库 → 接模型 → 注册 MCP → 验证的顺序走一遍配置和命令都能直接复制。2. 前置准备环境、依赖与 TaoToken 接入2.1 基础环境与依赖清单先确认 Python 版本在 3.10 以上显卡驱动和 CUDA 能正常识别。依赖分两块向量与编排用 LangChain 系列模型与嵌入用 Hugging Face 系列。建议单独建虚拟环境避免和系统包打架。python -m venv rag_env source rag_env/bin/activate # Windows 用 rag_env\Scripts\activate pip install -U langchain langchain-community langchain-chroma langchain-huggingface pip install -U langchain-milvus pymilvus sentence-transformers pip install -U transformers datasets tokenizers accelerate pip install -U chromadb mcp如果你打算先用云端模型快速验证链路再切回本地模型可以准备一个兼容 OpenAI 协议的接入点。TaoToken 提供统一的 API 入口模型对话、编码计划、控制台和密钥管理都有对应页面接入时把 base_url 指向https://taotoken.net/api即可密钥在控制台生成。2.2 目录结构约定为了让后面的配置不混乱先固定一套目录。向量库按环境分目录模型权重单独放配置文件和代码分离。rag_project/ ├── config/ │ ├── config.toml │ └── settings.json ├── models/ │ └── chatglm3-6b/ ├── chroma_db/ │ └── prod/ ├── milvus_lite/ │ └── prod/ └── rag_agent.py2.3 config.toml 与 settings.json 骨架把易变的参数抽到配置文件里切换环境时不用改代码。config.toml管模型和路径settings.json管向量库客户端行为。# config/config.toml [model] name chatglm3-6b path ./models/chatglm3-6b max_new_tokens 512 repetition_penalty 1.1 device cuda [embedding] model_name BAAI/bge-large-zh-v1.5 cache_folder ./models device cuda [vectorstore] backend chroma # chroma 或 milvus collection wiki_docs persist_dir ./chroma_db/prod milvus_uri ./milvus_lite/prod/lite.db [splitter] chunk_size 512 chunk_overlap 64 [api] base_url https://taotoken.net/api model gpt-4o-mini{ chroma: { is_persistent: true, anonymized_telemetry: false }, retriever: { k: 3 }, mcp: { server_url: http://127.0.0.1:8765/mcp, tool_name: weather } }注意persist_dir和milvus_uri里的环境名建议用环境变量ENV_NAME控制dev/stg/prod 各存一份避免测试数据污染生产库。3. 可复制配置向量库、嵌入与本地模型3.1 用 Hugging Face 三件套生成嵌入并写入 Chroma嵌入模型选BAAI/bge-large-zh-v1.5中文语义检索效果稳。切块用RecursiveCharacterTextSplitter512 字符一块、64 字符重叠能保住上下文。写入前用 MD5 指纹做去重重复跑不会产生脏数据。import os, hashlib, tomllib from datetime import datetime from langchain_huggingface.embeddings import HuggingFaceEmbeddings from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.schema import Document from langchain_chroma import Chroma from chromadb.config import Settings with open(config/config.toml, rb) as f: cfg tomllib.load(f) env_name os.getenv(ENV_NAME, prod) persist_dir cfg[vectorstore][persist_dir] os.makedirs(persist_dir, exist_okTrue) embed_model HuggingFaceEmbeddings( model_namecfg[embedding][model_name], cache_foldercfg[embedding][cache_folder], model_kwargs{device: cfg[embedding][device]}, ) splitter RecursiveCharacterTextSplitter( chunk_sizecfg[splitter][chunk_size], chunk_overlapcfg[splitter][chunk_overlap], ) texts_source [ 上海市市东中学前身是1916年创办的聂中丞华童公学后多次更名。, CSDN 成立于1999年是中国领先的 IT 技术社区和开发者服务平台。, ] docs, ids [], [] for idx, text in enumerate(texts_source, start1): meta {biz_id: fdoc_{idx}, inserted_at: int(datetime.utcnow().timestamp())} for chunk in splitter.split_text(text): docs.append(Document(page_contentchunk, metadatameta)) ids.append(hashlib.md5(chunk.encode(utf-8)).hexdigest()) client_settings Settings( is_persistentTrue, persist_directorypersist_dir, anonymized_telemetryFalse, ) vectorstore Chroma( persist_directorypersist_dir, embedding_functionembed_model, client_settingsclient_settings, collection_namecfg[vectorstore][collection], create_collection_if_not_existsTrue, ) existing vectorstore.get(idsids) existing_ids set(existing[ids]) new_docs, new_ids [], [] for doc, doc_id in zip(docs, ids): if doc_id not in existing_ids: new_docs.append(doc) new_ids.append(doc_id) if new_docs: vectorstore.add_documents(documentsnew_docs, idsnew_ids) print(f已有 {len(existing_ids)} 条计划 {len(docs)} 条 f重复 {len(docs) - len(new_docs)} 条实际新增 {len(new_ids)} 条)跑完会打印新增条数chroma_db/prod下能看到持久化文件。第二次运行同一批数据新增应为 0说明去重生效。3.2 切换到 Milvus Lite单机版数据量上到几十万条、或者想要集群能力时把后端换成 Milvus。单机版用本地文件即可集群版只需改 URI 和端口。import pathlib, time, hashlib from langchain_milvus import Milvus from langchain.schema import Document BASE pathlib.Path.cwd() / milvus_lite / prod BASE.mkdir(parentsTrue, exist_okTrue) URI str(BASE / lite.db) COL wiki_docs docs, ids [], [] for i, raw in enumerate(texts_source, 1): for chunk in splitter.split_text(str(raw)): pk hashlib.md5(chunk.encode()).hexdigest() meta {biz_id: fdoc_{i}, inserted_at: int(time.time())} docs.append(Document(page_contentchunk, metadatameta)) ids.append(pk) if not pathlib.Path(URI).exists(): Milvus.from_documents( documentsdocs, idsids, embeddingembed_model, collection_nameCOL, connection_args{uri: URI}, index_params{index_type: IVF_FLAT, metric_type: L2}, auto_idFalse, ) vs Milvus( embedding_functionembed_model, collection_nameCOL, connection_args{uri: URI}, auto_idFalse, ) vs.delete(idsids) # 幂等先删后插等价 Upsert vs.add_documents(docs, idsids) print(fSYNC 完成 {len(ids)} 条)3.3 本地大模型封装成 LangChain LLMChatGLM3-6B 用AutoModel加载注意不是AutoModelForCausalLM。封装成 LangChain 的 LLM 类同时实现_call和_stream流式输出时取差异量避免前缀重复。import os, warnings, logging, torch from typing import List, Iterator, Optional, Any from transformers import AutoTokenizer, AutoModel from langchain_core.language_models.llms import LLM warnings.filterwarnings(ignore, message.*max_new_tokens.*max_length.*) for lib in [transformers, huggingface_hub, torch]: logging.getLogger(lib).setLevel(logging.ERROR) MODEL_DIR ./models/chatglm3-6b tok AutoTokenizer.from_pretrained(MODEL_DIR, trust_remote_codeTrue) glm (AutoModel.from_pretrained(MODEL_DIR, trust_remote_codeTrue) .half().cuda().eval()) class GLM3StreamLLM(LLM): model: Any glm tokenizer: Any tok max_new_tokens: int 512 property def _llm_type(self) - str: return chatglm3 def _call(self, prompt: str, stop: Optional[List[str]] None) - str: full, _ self.model.chat( self.tokenizer, prompt, history[], max_new_tokensself.max_new_tokens, repetition_penalty1.1) return full def _stream(self, prompt: str, stop: Optional[List[str]] None) - Iterator[str]: prev for cur, _ in self.model.stream_chat( self.tokenizer, prompt, history[], max_new_tokensself.max_new_tokens, repetition_penalty1.1): delta cur[len(prev):] prev cur if delta: yield delta3.4 组装检索链与流式 RAG把向量库封成 retriever配上 Prompt 模板再用RetrievalQA串起来。流式版本手动先检索、再拼 Prompt、最后调_stream比等官方 streaming 稳定。from langchain.prompts import PromptTemplate from langchain.chains import RetrievalQA retriever vectorstore.as_retriever(search_kwargs{k: 3}) PROMPT PromptTemplate.from_template( 你是一个问答助手。请根据以下上下文回答问题答不上请说“我不知道”。\n\n 上下文:\n{context}\n\n问题: {question}\n回答:) llm GLM3StreamLLM() rag_chain RetrievalQA.from_llm( llmllm, retrieverretriever, promptPROMPT, return_source_documentsTrue) def rag_stream(question: str) - Iterator[str]: docs retriever.invoke(question) ctx \n\n.join(d.page_content for d in docs) prompt PROMPT.format(contextctx, questionquestion) yield from llm._stream(prompt)3.5 MCP 服务注册片段MCP 协议的价值在于工具在 Server 上注册一次客户端只认工具名和参数不用管底层请求细节。下面是一个最小 MCP Server 注册示例把天气查询注册成工具。# mcp_server.py from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent app Server(local-tools) app.list_tools() async def list_tools(): return [Tool( nameweather, description查询指定城市的天气, inputSchema{ type: object, properties: {city: {type: string}}, required: [city], }, )] app.call_tool() async def call_tool(name: str, arguments: dict): if name weather: city arguments.get(city, ) return [TextContent(typetext, textf{city} 今日晴气温 22 摄氏度)] raise ValueError(f未知工具: {name}) async def main(): async with stdio_server() as (read, write): await app.run(read, write, app.create_initialization_options()) if __name__ __main__: import asyncio asyncio.run(main())客户端侧用streamablehttp_client或 stdio 连上 Server把工具包成 LangChain 的Tool再交给 Agent 调度。from langchain.tools import Tool from mcp import ClientSession, streamablehttp_client async def weather_tool(params: dict) - str: async with streamablehttp_client(http://127.0.0.1:8765/mcp) as (read, write, _, _): async with ClientSession(read, write) as session: await session.initialize() resp await session.call_tool(weather, params) return resp[result] mcp_tool Tool(nameweather, funcweather_tool, description查询城市天气参数为 city)3.6 Agent 调度检索链与 MCP 工具Agent 用ZERO_SHOT_REACT_DESCRIPTION把检索链包成一个工具和 MCP 工具一起注册。模型会自己决定先检索还是先调工具。from langchain.agents import initialize_agent, AgentType from langchain.tools import Tool def qa_tool(query: str) - str: return rag_chain.invoke({query: query})[result] tools [ Tool(nameknowledge_base, funcqa_tool, description查询本地知识库输入自然语言问题), mcp_tool, ] agent initialize_agent( toolstools, llmllm, agentAgentType.ZERO_SHOT_REACT_DESCRIPTION, verboseTrue, )4. 验证请求与成功结果4.1 检索链验证先单独验证检索链确认向量库能召回正确文档。q 上海市市东中学是什么时候创办的 out rag_chain.invoke({query: q}) print(Answer:, out[result]) for d in out[source_documents]: print(ID:, d.metadata.get(biz_id), |, d.page_content[:60], …)预期输出答案里出现1916 年来源文档命中市东中学那条。如果答案说我不知道多半是检索没召回检查嵌入模型是否和写入时一致。4.2 流式输出验证print(【流式回答】, end, flushTrue) for chunk in rag_stream(请简单介绍一下 CSDN): print(chunk, end, flushTrue) print()预期逐字吐出且没有重复前缀。如果出现整段重复说明_stream里没做差异截取。4.3 Agent 多工具验证answer agent.run(北京今天天气怎么样顺便查一下知识库里 CSDN 成立时间。) print(answer)预期 Agent 先调weather工具拿到天气再调knowledge_base拿到成立年份最后汇总。verboseTrue时能看到完整的 Thought/Action/Observation 轨迹。4.4 用 TaoToken 快速对照验证如果本地模型还没下完想先确认链路通不通可以把 LLM 换成兼容 OpenAI 协议的接入点。密钥在控制台生成模型对话页面可以直接试问答效果。from langchain.chat_models import init_chat_model import os os.environ[OPENAI_API_KEY] 你的密钥 os.environ[OPENAI_BASE_URL] https://taotoken.net/api llm_api init_chat_model(gpt-4o-mini, model_provideropenai) qa_api RetrievalQA.from_llm( llmllm_api, retrieverretriever, promptPROMPT, return_source_documentsTrue) print(qa_api.invoke({query: CSDN 成立于什么时候})[result])同一套 retriever 和 Prompt只换 LLM能快速判断问题出在检索还是模型。长期跑编码和 Agent 任务的话可以看下 Coding Plan额度更划算。5. 本篇常见错排查5.1 嵌入维度不匹配报错Collection expecting embedding with dimension of X, got Y说明写入和查询用了不同嵌入模型。检查config.toml里embedding.model_name是否被改过。换模型必须重建集合旧向量维度对不上。5.2 Chroma persist 报 AttributeError新版 Chroma≥0.4自动落盘手动调persist()会抛异常。用 try/except 包住或者直接不调。老版本才需要显式持久化。5.3 ChatGLM3 加载报错用AutoModelForCausalLM加载 ChatGLM3 会报缺lm_head。必须用AutoModel并加trust_remote_codeTrue。显存不够时把.half()换成 4bit 量化。5.4 MCP 连接超时streamablehttp_client连不上先确认 Server 已启动且端口一致。stdio 模式不需要 URL直接传命令。工具名大小写敏感call_tool里的名字必须和list_tools注册的一致。5.5 检索召回为空retriever.invoke返回空列表常见原因有三个集合名写错、数据没写进去、k设太小。先用vectorstore.get()看集合里到底有多少条再调k。5.6 流式输出重复stream_chat每步返回完整上下文直接 yield 会重复。必须记录prev取差异这是 ChatGLM 系列特有的坑。6. 下一步把链路接进你的业务跑通之后把texts_source换成你自己的文档加载逻辑就行。PDF 用PyPDFLoaderMarkdown 用UnstructuredMarkdownLoader数据库导出直接拼成 list。切块参数按文档类型调技术文档 512 够用法律合同建议 1024 保完整条款。MCP 工具可以继续扩把内部工单系统、监控查询、部署脚本都注册成工具Agent 就能在问答之外执行动作。密钥和接入点统一在控制台管理接入文档里有各语言的示例。本地模型和云端模型可以按数据敏感度分流公开知识走云端内部资料走本地一套 retriever 两种 LLM 切换即可。