从零手搓一个劳动法问答 RAG：LangChain + 本地向量库实战（零基础可跟）

一、先搞清楚RAG 到底解决什么问题直接问大模型试用期最长多久它可能答对也可能一本正经地编造条文幻觉。根源在于大模型的知识是训练时背下来的具体到某部法律的某一条它既可能记错也没法给你准确引用出处。RAGRetrieval-Augmented Generation检索增强生成的思路很朴素提前把权威文档比如法律原文切成小块算出每块的语义向量存起来——这叫建库预处理用户提问时先算问题的向量去库里找出语义最接近的几块原文——这叫检索Retrieval把检索到的原文连同问题一起塞给大模型让它看着资料答题——这叫增强生成Augmented Generation。类比开卷考试大模型是那个聪明但记不全的考生RAG 给他发了一本可以现场翻的教材。本文的最终效果问试用期最长多久 答根据《中华人民共和国劳动合同法》第十九条劳动合同期限三个月以上不满一年的 试用期不得超过一个月……三年以上固定期限和无固定期限的劳动合同 试用期不得超过六个月。……本回答仅供参考不构成正式法律意见。整体流程一张图看懂上半是离线预处理下半是在线问答二、技术选型每一层用什么、为什么环节选型理由文档解析python-docx法律文件是 docx按段落读取即可Embedding 模型bge-small-zh-v1.5ONNX 量化版约 25MB中文检索效果好、体积小、CPU 就能跑不依赖 torch向量存储FAISSfaiss-cpulangchain-community本地嵌入式索引自带save_local/load_local持久化和元数据过滤ChromaDB 的 Rust 后端在本机崩溃Agent 框架deepagentslanggraph自动处理模型决策调工具 → 拿到结果再作答的循环自带多轮对话记忆大模型火山引擎方舟OpenAI 兼容接口模型deepseek-v4.1-flash便宜、快通过langchain-openai接入接口服务FastAPI标配自带/docs调试页面一个值得记住的架构取舍检索侧全部本地化ONNX 模型 FAISS 向量库零网络调用、零费用只有最终生成走云端大模型。embedding 是高频计算本地化之后整个系统又快又省钱。三、项目结构总览lg-learn/ ├── doc/ # 法律 docx 原文知识库来源 │ ├── 中华人民共和国劳动法_20181229.docx │ └── 中华人民共和国劳动合同法_20121228.docx ├── models/ # 本地 embedding 模型首次运行自动下载 ├── vector_store/ # 向量库持久化自动生成可安全删除 └── src/lg_learn/ ├── legal_kb.py # ★ docx 按条切分 向量库存取预处理核心 ├── local_embeddings.py # ★ 本地中文 embeddingONNX ├── labor_legal_agent.py # ★ agent 装配检索工具 提示词 ├── api.py # FastAPI 接口 └── model.py # 大模型接入火山引擎环境准备Python 3.13用 uv 管理依赖uv sync核心依赖deepagents、langchain、langchain-openai、langgraph、faiss-cpu、langchain-community、onnxruntime、tokenizers、python-docx、numpy、fastapi、huggingface-hub。四、预处理第一步按条切分法律文档切分chunking是 RAG 质量的生命线切太碎上下文不完整切太大检索不精准。法律条文有个天然优势——第X条就是最完美的切分边界。每一条是独立的语义单元长度适中还自带引用编号。切分逻辑legal_kb.pyimportrefrompathlibimportPathfromdocximportDocument# 匹配段首的条文标题如第一条第一百零七条ARTICLE_HEADER_REre.compile(r^(第[一二三四五六七八九十百千零]条))defload_docx_articles(doc_dir:Path)-list[dict]:chunks:list[dict][]fordocx_pathinsorted(doc_dir.glob(*.docx)):paragraphs[p.text.strip()forpinDocument(docx_path).paragraphs]# 法律名取自文件名第一个下划线前如中华人民共和国劳动合同法law_namedocx_path.stem.split(_)[0]current_article:str|NoneNone# 当前条号current_lines:list[str][]# 当前条的所有段落defflush():遇到新条号或文件结束时把已累积的条文存为一个 chunkifcurrent_articleandcurrent_lines:chunks.append({text:\n.join(current_lines),metadata:{source:law_name,article:current_article},})forparainparagraphs:ifnotpara:continuemARTICLE_HEADER_RE.match(para)ifm:flush()# 先闭合上一条current_articlem.group(1)# 再开启新一条current_lines[para]elifcurrent_article:current_lines.append(para)# 正文段落款、项归入当前条flush()returnchunks两个细节值得注意正则只匹配段首^。法条正文里到处是依照本法第X条规定这类交叉引用但它们不会出现在段落开头所以不会误切分。metadata 里存了source法律名和article条号。这不是可有可无的——它是回答末尾准确引用的来源也是 RAG 可解释性的关键。切完两部法律共得 205 个 chunk每个长这样{text:第一条 为了完善劳动合同制度……制定本法。,metadata:{source:中华人民共和国劳动合同法,article:第一条}}五、预处理第二步本地 Embedding把文字变成向量Embedding嵌入就是把一段文字变成一串数字向量语义越接近的文字向量方向越接近。这里用bge-small-zh-v1.5的ONNX 量化版Xenova/bge-small-zh-v1.5直接用onnxruntime推理不需要装几个 G 的 PyTorch。首次运行自动从 hf-mirror 镜像下载模型约 25MBfromhuggingface_hubimportsnapshot_downloaddefensure_model(model_dir:Path)-Path:ifall((model_dir/f).exists()forfinMODEL_FILES):returnmodel_dir os.environ.setdefault(HF_ENDPOINT,https://hf-mirror.com)# 国内镜像returnPath(snapshot_download(REPO_ID,allow_patternsMODEL_FILES,local_dirstr(model_dir)))实现 LangChain 的Embeddings接口local_embeddings.pyfromlangchain_core.embeddingsimportEmbeddings# bge 系列模型要求只在“查询侧”加这条指令前缀文档侧不加非对称检索QUERY_INSTRUCTION为这个句子生成表示以用于检索相关文章classBgeOnnxEmbeddings(Embeddings):def__init__(self,model_dirNone):importonnxruntimefromtokenizersimportTokenizer model_dirensure_model(model_dirorMODEL_DIR)self._tokenizerTokenizer.from_file(str(model_dir/tokenizer.json))self._tokenizer.enable_truncation(max_length512)# 超长截断self._tokenizer.enable_padding()# 补齐以便批量推理self._sessiononnxruntime.InferenceSession(str(model_dir/onnx/model_quantized.onnx),providers[CPUExecutionProvider],)def_embed(self,texts:list[str])-list[list[float]]:encodingsself._tokenizer.encode_batch(texts)inputs{input_ids:np.array([e.idsforeinencodings],dtypenp.int64),attention_mask:np.array([e.attention_maskforeinencodings],dtypenp.int64),token_type_ids:np.array([e.type_idsforeinencodings],dtypenp.int64),}hiddenself._session.run(None,inputs)[0]# CLS pooling L2 归一化bge 官方推荐的检索用法clshidden[:,0]return(cls/np.linalg.norm(cls,axis1,keepdimsTrue)).tolist()defembed_documents(self,texts):# 文档侧原文直接编码returnself._embed(texts)defembed_query(self,text):# 查询侧加指令前缀再编码returnself._embed([QUERY_INSTRUCTIONtext])[0]这段代码里藏着三个 bge 模型的标准动作新手最容易漏CLS pooling不取所有 token 的平均而是取第 0 个 token[CLS]的向量L2 归一化把向量长度统一为 1之后点积就等于余弦相似度查询加前缀、文档不加bge 在训练时就是这么做的非对称检索加了对中文检索效果提升明显。六、预处理第三步FAISS 向量库为什么不用 ChromaDB205 条法条、每条一个 512 维向量——总共不到 1MB 的数据。这个量级下服务器式的向量数据库是负担本项目实测 ChromaDB 的 Rust 后端在 Windows 本机直接崩溃。最终选型是FAISSFacebook 开源的向量检索库纯 CPU wheel、嵌入式运行、零服务还自带save_local/load_local磁盘持久化。项目第一版其实用 numpy 手写了 30 行向量库后来迁移到 FAISS——代码更少还白捡了元数据过滤和标准的 LangChainVectorStore接口演变过程见第十节踩坑记录。入口仍然是有缓存就用、没有就重建legal_kb.pyfromlangchain_community.vectorstoresimportFAISSfromlangchain_core.documentsimportDocumentasLCDocumentdefget_vectorstore(embeddings,persist_dirPERSIST_DIR,rebuildFalse)-FAISS:has_dataall((persist_dir/f).exists()forfin(index.faiss,index.pkl))ifhas_dataandnotrebuild:# index.pkl 是 pickle 格式官方要求显式确认来源可信returnFAISS.load_local(str(persist_dir),embeddings,allow_dangerous_deserializationTrue)chunksload_docx_articles()# 否则从头构建docs[LCDocument(page_contentc[text],metadatac[metadata])forcinchunks]storeFAISS.from_documents(docs,embeddings)# 内部自动批量 embeddingstore.save_local(str(persist_dir))# 持久化为 index.faiss index.pklreturnstore为行文简洁省略了旧文件清理和空库报错完整代码见仓库。两个细节值得注意相似度语义不变bge 输出已做 L2 归一化而 FAISS 默认的 L2 距离在归一化向量上与余弦相似度单调等价检索结果和手写版完全一致。白捡的元数据过滤FAISS 的similarity_search原生支持filter参数按 metadata 精确匹配比如只在《劳动合同法》范围内检索store.similarity_search(公司可以随意变更工作地点吗,k2,filter{source:中华人民共和国劳动合同法})以后更新了 docx删掉vector_store/目录或以rebuildTrue调用就会自动重建。七、把检索能力交给大模型Agent 装配有了向量库下一步是把它包装成工具tool让大模型自己决定何时调用。这就是 Agent 模式比写死先检索再提问的朴素 RAG 更灵活的地方——多轮对话中模型可以按需多次检索。检索工具labor_legal_agent.py_vectorstoreget_vectorstore(BgeOnnxEmbeddings())# 模块级单例启动时加载defsearch_labor_law(query:str)-str:检索《劳动法》《劳动合同法》中与问题相关的法条。 Args: query: 用中文描述的法律问题或关键词。 docs_vectorstore.similarity_search(query,k4)ifnotdocs:return未检索到相关法条。# 每条结果带上法律名称和条号方便模型在回答中准确引用return\n\n.join(f《{d.metadata[source]}》{d.metadata[article]}\n{d.page_content}fordindocs)函数的 docstring 就是给大模型看的工具说明书——写清楚这个工具干什么、参数是什么模型才能正确调用所以 docstring 一定要认真写。系统提示词明确约束模型只依据检索结果作答SYSTEM_PROMPT你是一名专业的中国劳动法律咨询助手。 回答问题时必须遵守 1. 先调用 search_labor_law 工具检索相关法条仅依据检索到的条文作答不要凭记忆编造法条。 2. 引用条文时注明具体法律名称和条号如《中华人民共和国劳动合同法》第十条。 3. 如果检索不到相关条文如实告知用户不要编造。 4. 回答末尾附上提示本回答仅供参考不构成正式法律意见。 用deepagents一行装配自带多轮对话记忆fromdeepagentsimportcreate_deep_agentfromlanggraph.checkpoint.memoryimportInMemorySaverdefcreate_labor_agent():returncreate_deep_agent(modelget_huoshan_model_object(),# 火山引擎 deepseek-v4.1-flashtools[search_labor_law],system_promptSYSTEM_PROMPT,checkpointerInMemorySaver(),# 对话历史存内存重启即清空)大模型接入用的是火山引擎方舟的 OpenAI 兼容接口model.pyimportosfromlangchain_openaiimportChatOpenAIdefget_huoshan_model_object(model_name:strdeepseek-v4.1-flash):returnChatOpenAI(modelmodel_name,api_keyos.environ[ARK_API_KEY],# 密钥放环境变量不要硬编码进代码base_urlhttps://ark.cn-beijing.volces.com/api/plan/v3,)注意API Key 务必通过环境变量注入如$env:ARK_API_KEY你的key硬编码在源码里一旦随仓库泄露损失的不只是额度。八、包一层 FastAPI支持多轮对话appFastAPI(titleLabor Legal Agent)_agentcreate_labor_agent()# 模块级单例所有请求复用classChatRequest(BaseModel):question:strthread_id:str|NoneNone# 用于多轮对话不传则视为新会话app.post(/chat,response_modelChatResponse)defchat(req:ChatRequest):# 没有 thread_id 就生成新的同一 thread_id 的历史会被保留thread_idreq.thread_idoruuid.uuid4().hexresult_agent.invoke({messages:[{role:user,content:req.question}]},config{configurable:{thread_id:thread_id}},)returnChatResponse(answerresult[messages][-1].content,thread_idthread_id)多轮对话的原理thread_id是会话的钥匙InMemorySaver按它存历史消息。客户端保存第一次返回的thread_id下一轮带上模型就知道那超过上限有什么后果里的那指什么。九、跑起来验证启动服务cd src\lg_learn..\..\.venv\Scripts\python.exe-m uvicorn api:app--port 8000首次启动会自动下载 embedding 模型并构建向量库之后启动都是秒级。然后# 单轮提问curl.exe-X POST http://127.0.0.1:8000/chat -HContent-Type: application/json-d{question:试用期最长多久}# 多轮对话带上第一次返回的 thread_idcurl.exe-X POST http://127.0.0.1:8000/chat -HContent-Type: application/json-d{question:那超过上限有什么后果,thread_id:上一步返回的 thread_id}也可以打开http://127.0.0.1:8000/docs在 Swagger 页面里直接调试。十、踩坑记录都是真金白银的教训chromadb 在 Windows 本机崩溃Rust 后端兼容性问题。先用 numpy 自建过渡最终迁移到 FAISS——纯 CPU wheel 没有崩溃问题save_local/load_local持久化开箱即用代码比手写版还少。PyPI 上的langchain-faiss是空壳包名字唬人装完发现__init__.py是空的作者也非 LangChain 官方。FAISS 的官方 LangChain 集成目前在langchain-community里import 时会提示该包进入 sunset 维护期但目前它是唯一正经来源作为模块导入时警告不会显示直接运行脚本才会看到一次。onnxruntime 新版崩溃固定在1.20.1不要随意升级。忘记加 bge 查询前缀检索质量肉眼可见地下降。用 bge 系列一定记得查询加前缀、文档不加。Windows 控制台中文乱码脚本入口加sys.stdout.reconfigure(encodingutf-8)。对话历史存内存InMemorySaver重启即丢。生产环境应换SqliteSaver或 Redis 等持久化 checkpointer。十一、总结与可扩展方向回顾整条链路RAG 的本质就三件事预处理按语义边界切块本文按法律条文→ 本地模型算向量 → 存入 FAISS 索引并持久化检索FAISS 索引检索 top-k归一化后 L2 距离与余弦相似度等价可按 metadata 过滤生成检索结果 约束性提示词交给大模型作答并要求引用出处。在这个骨架上可以继续演进换更大的 embedding 模型如 bge-large-zh提升检索精度接口完全不变换真正的向量数据库数据量上到十万级以上后从 FAISS 迁移到 Milvus / Qdrant / pgvector因为都实现了 LangChainVectorStore接口检索调用代码基本不用改加混合检索向量检索 BM25 关键词检索取并集改善对专有名词的命中加重排序rerank初检取 20 条用 bge-reranker 精排后再给模型持久化对话历史换SqliteSaver支持服务重启不丢会话。如果这篇文章帮你跑通了第一个 RAG欢迎在评论区交流告诉我你用它接了什么知识库免责声明本系统输出仅供参考不构成正式法律意见。