企业官网建设流程全解析

2. 项目概述与选型逻辑

“Milvus + LangChain + Ollama 搭建生产级 RAG（含 Tag / Metadata 解析）”，这句话本身信息量很大。它不是单纯跑通一个 Hello World 级别的问答 Demo，而是冲着“生产级”和“Metadata 解析”这两个核心关键词去的。我把这个项目的真实需求拆解成三层来理解：

第一层，是搭一套基于本地大模型的完整 RAG 应用，用 Ollama 承载推理，用 Milvus 做向量检索，用 LangChain 把整个流程串起来，实现“文档进、答案出”的效果。第二层，是精细化处理：不满足于把整篇文档切块塞进向量库，而是要对文档本身的结构化信息（Tags、Metadata，比如文档来源、作者、分类、时间等）进行解析，让这些信息参与过滤和检索，从而显著提升答案的准确率。第三层，是生产级指标：涉及并发、持久化、检索效果调优、可观测性和故障恢复，而不是停留在本地脚本的直接调用。

我为什么认为这套组合在当前很有代表性？因为“大模型私有化 + 私有知识库”几乎是所有有数据安全要求场景的共同需求。Ollama 能跑 Llama 3、Qwen 等开源模型，Milvus 是开源向量库中分布式和生态支持比较好的一档，LangChain 是目前做应用编排最主流的一层。三者的组合既能脱离对外部 API 的依赖，又能把公司内部的多源文档统一纳管起来做智能问答。

我在实际调研和搭建中确认了一个关键认知：这套技术栈不是靠堆功能就能上生产的。大量精力会消耗在处理“文档怎么切、Metadata 怎么提取、过滤条件怎么写、检索结果怎么做重排、并发怎么处理延峰”这些听起来很基础、做起来很现实的问题上。所以这篇内容，我按照我自己落地这套系统的顺序来写，重点讲思路和排坑，给出一套可以直接照着改的工程骨架。

这个项目适用的读者画像非常明确：已经跑通过一条最简单的 LangChain + 向量库链路，现在需要处理真实业务文档、需要多标签筛选、需要做多人使用场景的开发者。如果你是纯零基础，建议你先跑一遍官方 Quickstart，再回头参考这里的设计会更顺手。

3. 整体链路设计与选型深挖

任何 RAG 系统，本质都是“索引 + 检索 + 生成”三段式的工程化。但在生产环境里，这三段每一步都可能变成瓶颈。我画一下我在这个项目里的整体链路设计：

文档加载 -> 文档解析（Metdata/Tag 提取） -> 切块 -> Embedding -> 向量入库 ↓ 用户问题 -> Query 改写/意图识别 -> Embedding -> 向量检索 + Metadata 预过滤 ↓ 重排 -> 上下文拼装 -> LLM 生成

这一版链路比普通的 RAG Demo 多出两个关键环节：一是文档解析阶段的 Tag/Metadata 提取，二是查询阶段的 Metadata 过滤与重排。这两个环节恰恰是回答质量大幅度提升的关键。

3.1 为什么选这三件套

LangChain 版本这个问题我必须单拎出来说。我建项目时用了当时的0.2.x版本，等到搭建完的时候已经出现了0.3.x的更新，部分 API 有细微变化。代码里我会注明版本，避免你直接复制却发现方法不存在。

3.2 Metadata 与 Tag 的用途拆解

我见过不少人做的 RAG 是彻头彻尾的“裸奔”方案：Loader 读入文档，随便切分一下，就一股脑塞进向量库。这样的结果是什么？用户搜索返回的是整库范围内的相似内容，结果没有业务边界、没有时间过滤、没有来源优先级，回答的准确率自然飘。其实只要在查询阶段加一条“Metadata 等于某部门”或者“发布时间大于某日期”的过滤条件，精度立刻上一个台阶。

具体到这个项目，我设计了四类 Metadata：

Tags 我则用了一个更灵活的方式，在 Metadata 中用tags字段保存一个列表，例如["故障处理", "稳定版", "适用v2.3"]。在 Milvus 中，这种VarChar数组可以通过JSON_CONTAINS实现精确匹配过滤。相比哈希 Tag，这样可以做到“多个标签组合筛选”，业务语义更强。

4. 环境准备与工具链安装

生产级应用的第一步，一定要把环境当作可控的基础设施来对待，而不是“装个 Python 库就能跑”。我在正式动手前，把以下工具统一锁定在测试机（Ubuntu 22.04，32G 内存，无独显）上验证：

4.1 Ollama 模型准备

Ollama 在这套链路由两个作用：一是提供 Embedding 模型，二是提供对话模型。我在实际验证时发现，模型选择对检索质量和生成质量的影响完全不亚于工程配置。我这里用两类常用模型做演示：

# 拉起对话模型（测试选用 Llama 3.1 8B 版） ollama pull llama3.1:8b # 拉起 Embedding 模型（用于文本向量化） ollama pull nomic-embed-text # 启动/验证 ollama list

关于 Embedding 模型，有个细节特别值得注意：nomic-embed-text默认输出 768 维向量，生成向量的质量和中文支持度中规中矩，作为测试起步完全没有问题。如果你要在正式的中文业务场景上做高精度检索，建议多对比bge-m3这类专门面向中文的模型，形如“建立索引时用什么维度，查询时就对应什么维度”的规则一定要守住。

4.2 Milvus 单机部署与持久化配置

我选用单机 Standalone 模式起步，它完整保留了 Milvus 的全部核心能力，又不至于一上来就接触分布式协调。用 Docker Compose 启动是最稳妥的。实际环境中我建议你为 Milvus 单独规划持久化目录：

services: etcd: image: quay.io/coreos/etcd:v3.5.5 environment: - ETCD_AUTO_COMPACTION_MODE=revision - ETCD_AUTO_COMPACTION_RETENTION=1000 - ETCD_QUOTA_BACKEND_BYTES=4294967296 volumes: - ${DOCKER_VOLUME_DIRECTORY:-.}/etcd:/etcd minio: image: minio/minio:RELEASE.2023-03-20T20-16-18Z environment: MINIO_ACCESS_KEY: minioadmin MINIO_SECRET_KEY: minioadmin volumes: - ${DOCKER_VOLUME_DIRECTORY:-.}/minio:/minio_data command: minio server /minio_data milvus: image: milvusdb/milvus:v2.4.1 command: ["milvus", "run", "standalone"] depends_on: - etcd - minio ports: - "19530:19530" - "9091:9091" volumes: - ${DOCKER_VOLUME_DIRECTORY:-.}/milvus:/var/lib/milvus

为什么 Milvus 要依赖 etcd 和 MinIO？etcd 存的是元数据和协调信息，MinIO 存的是向量索引文件与数据文件。这种“计算、元数据、存储分离”的设计保证了它在单机上和在分布式集群上，存储行为一致。

启动命令：

docker-compose up -d docker-compose ps

拉完镜像后，用docker-compose ps看到etcd、minio、milvus三者都处于 healthy 状态，才算环境就绪。

5. 核心链路实现：从文档解析到向量入库

我在文档加载这一层遇到的最大问题不是代码写不出来，而是不同来源的文档结构差异巨大。有 Markdown、有 PDF、有 Excel 导出的 CSV、还有 HTML。因此我专门设计了一个流程：先按文件类型分发到不同的解析器，再统一提取 Metadata 和标签，最后进入切分和向量化。

5.1 文档解析与 Metadata 提取

这个环节就是标题里所谓“Tag / Metadata 解析”落地的核心地带。

我实际的解析流程是这样的：对于 Markdown 文档，我会先用UnstructuredMarkdownLoader或TextLoader读入纯文本，然后用正则和规则抽取文档头部的元信息（比如 YAML front matter 中的category、department、date字段）。对于 PDF 文档，先用PyMuPDFLoader读取全文，再利用文本中出现的固定位置（如“摘要”“关键词”段落）做标签抽出。

这里给出一段我认为通用性很强的解析骨架：

import re import uuid from datetime import datetime from langchain_community.document_loaders import UnstructuredMarkdownLoader from langchain_core.documents import Document def parse_markdown_with_metadata(file_path: str) -> Document: loader = UnstructuredMarkdownLoader(file_path) raw_docs = loader.load() metadata = { "doc_id": uuid.uuid4().hex, "source": file_path, "category": "未知分类", "department": "未指定部门", "publish_date": datetime.now().strftime("%Y-%m-%d"), "tags": [] } page_content = raw_docs[0].page_content if raw_docs else "" # 抽取 Markdown 的 Front Matter 中的元信息 fm_match = re.search(r"^---\s*\n(.*?)\n---", page_content, re.DOTALL) if fm_match: fm_text = fm_match.group(1) for line in fm_text.splitlines(): if ":" in line: key, value = line.split(":", 1) value = value.strip().strip('"') if key.strip() in metadata: metadata[key.strip()] = value # 剥离 front matter，得到纯正文 page_content = re.sub(r"^---\s*\n.*?\n---", "", page_content, flags=re.DOTALL) # 简单关键词标签抽取 for keyword in ["故障处理", "安装指南", "API参考", "版本升级"]: if keyword in page_content: metadata["tags"].append(keyword) return Document(page_content=page_content, metadata=metadata)

这段代码背后有一个容易忽略的坑：原始的Document对象中metadata如果是来自 Loader 自带的，那么它往往只包含source等文件路径信息，没有业务字段。如果要增加字段，必须先把page_content和metadata都变成独立副本，再合并，否则可能出现数据污染。

5.2 文本切分策略与参数选择

文本切分的参数直接决定检索粒度。我见过不少同学直接把每块设为chunk_size=1000, chunk_overlap=200就开跑，其实这个取值在真实文档下并不一定合理。切分的目标是：块与块之间有足够的语义独立性，同时又不能在查询时丢失上下文。

我对不同文档类型分别处理：

• 对操作手册、指南类文档，按 Markdown 章节标题（#、## 等）作为天然断点切开，再将每个章节内部按RecursiveCharacterTextSplitter二次切分。
• 对长篇技术文档，设置chunk_size=800，chunk_overlap=200，每个块保证语义尽量自洽。
• 对代码片段较多的接口说明，采用“先按代码块整体保留，再按正文段落找回”的策略，避免把代码逻辑拦腰切断。

具体实现：

from langchain.text_splitter import RecursiveCharacterTextSplitter text_splitter = RecursiveCharacterTextSplitter( chunk_size=800, chunk_overlap=200, length_function=len, separators=["\n\n", "\n", "。", "！", "？", " ", ""], ) split_docs = text_splitter.split_documents([parsed_doc]) # 切分后，每个子块必须继承父级 doc 的 metadata for i, chunk in enumerate(split_docs): chunk.metadata["chunk_index"] = i

切分粒度为什么这么重要？想象一下：如果整份文档都是一个大块，那么检索出来以后上下文很长，LLM 需要花费大量 token 去理解无关内容，回答容易跑偏；如果切得太碎，检索出来的几个块彼此之间又可能没有连贯的信息，LLM 看到的上下文是碎片化的。实践下来，技术文档用 800-1000 这个范围最稳妥。

5.3 向量化与 Milvus Collection 设计

向量化我基于 Ollama 提供 Embedding 接口，完全本地化处理。这里给出一个统一的 Embedding 函数：

import requests from langchain_core.embeddings import Embeddings class OllamaEmbeddingsWrapper(Embeddings): def __init__(self, base_url: str = "http://localhost:11434", model: str = "nomic-embed-text"): self.base_url = base_url self.model = model def embed_documents(self, texts): embeddings = [] for text in texts: embeddings.append(self._embed_text(text)) return embeddings def embed_query(self, text): return self._embed_text(text) def _embed_text(self, text): resp = requests.post(f"{self.base_url}/api/embed", json={"model": self.model, "input": text}) resp.raise_for_status() return resp.json()["embeddings"][0]

然后设计 Milvus 的 Collection Schema。这里有一个关键点：Milvus 2.4 中向量的维度必须在建 Collection 时就固定下来，且embedding的向量类型要定义成FLOAT_VECTOR，维度和 Embedding 模型的输出一致。我会额外添加一个标量字段metadata_json，用来存储整份 Metadata 的 JSON 字符串，这样在必要的时候还能反查原始信息。

from pymilvus import ( connections, Collection, CollectionSchema, FieldSchema, DataType, utility ) connections.connect(host="localhost", port="19530") fields = [ FieldSchema(name="id", dtype=DataType.VARCHAR, is_primary=True, max_length=64), FieldSchema(name="doc_id", dtype=DataType.VARCHAR, max_length=64), FieldSchema(name="category", dtype=DataType.VARCHAR, max_length=128), FieldSchema(name="department", dtype=DataType.VARCHAR, max_length=128), FieldSchema(name="publish_date", dtype=DataType.VARCHAR, max_length=32), FieldSchema(name="tags", dtype=DataType.JSON), FieldSchema(name="metadata_json", dtype=DataType.JSON), FieldSchema(name="content", dtype=DataType.VARCHAR, max_length=65535), FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=768), ] schema = CollectionSchema(fields, description="RAG 知识库") if utility.has_collection("rag_kb"): collection = Collection("rag_kb") else: collection = Collection(name="rag_kb", schema=schema)

这里我把category、department、publish_date、tags做成独立字段，而不是全部塞进 JSON。为什么要这样？因为 Milvus 的标量过滤（过滤表达式expr）对结构化字段的支持会更直接、性能更好。JSON 字段适合存一些不被频繁过滤但是需要展示的数据，比如原始来源路径、自定义扩展参数等。

5.4 数据入库的幂等与批量写入

写库前必须处理一个现实问题：重复导入。同一份文档如果被导入两次，检索时就会出现冗余结果。所以我在插入前做了一次幂等控制，用doc_id查询是否已存在：

from pymilvus import utility def upsert_documents(docs, collection_name="rag_kb"): collection = Collection(collection_name) collection.load() # 已存在 doc_id 列表 existing_docs = set() res = collection.query( expr=f"doc_id != ''", output_fields=["doc_id"] ) for r in res: existing_docs.add(r["doc_id"]) # 批量向量化 new_docs = [] for doc in docs: if doc.metadata["doc_id"] in existing_docs: continue embedding = embed_docs([doc.page_content])[0] new_docs.append({ "id": uuid.uuid4().hex, "doc_id": doc.metadata["doc_id"], "category": doc.metadata.get("category", ""), "department": doc.metadata.get("department", ""), "publish_date": doc.metadata.get("publish_date", ""), "tags": doc.metadata.get("tags", []), "metadata_json": doc.metadata, "content": doc.page_content, "embedding": embedding, }) if new_docs: collection.insert(new_docs) collection.flush()

为什么最后要加一个flush()？元数据落盘需要时间，如果刚写完就立刻查询，有可能查到的是内存中的未落盘数据。在测试环境和生产环境，flush()都是为了确保写入可见性必须存在的一步。

6. 检索链路的精细设计：过滤、召回与重排

向量检索只是整个 RAG 的中间环节，但它的质量上限决定了最终答案的质量上限。到这一步，Milvus 的查询能力开始真正体现价值。

6.1 Metadata 预过滤时的表达式坑位设计

Milvus 支持在搜索时传入expr表达式，结合标量字段过滤，然后再在过滤结果中做向量检索。这比“先向量召回再内存过滤”的性能高得多，因为向量检索的范围被大大缩小了。

from pymilvus import Collection collection = Collection("rag_kb") collection.load() query_text = "Milvus 数据导入延迟高的原因？" query_embedding = embed_query(query_text) # 组装过滤条件 filters = [] filters.append("category == '故障处理'") filters.append("department == '研发中心'") filters.append("publish_date >= '2024-01-01'") # 对 JSON 字段中的数组做包含判断 filters.append("JSON_CONTAINS(tags, '稳定版')") filter_expr = " and ".join(filters) results = collection.search( data=[query_embedding], anns_field="embedding", param={"metric_type": "IP", "params": {"nprobe": 16}}, limit=20, expr=filter_expr, output_fields=["content", "category", "department", "publish_date", "doc_id"] )

这里有几个需要重点标记的地方：

第一，metric_type的选择。我建库时用的向量是nomic-embed-text，这类模型在语义相似度上常配合余弦相似度（COSINE）。但我实际实验下来，在维度不高、数据量可控时，IP内积与归一化后的向量效果接近，不过还是在建索引时尽可能保持一致性。如果你在建索引时指定了COSINE，查询时的metric_type也必须是对应参数，不匹配会出现查询异常或者结果失真。

第二，JSON_CONTAINS(tags, '稳定版')这个是 Milvus 对 JSON 字段提供的能力扩展，但需要注意标签字段的类型是JSON而非ARRAY。我在初始化时如果误用ARRAY类型，查询时会遇到字段类型不匹配的问题。这一点在官方 Schema 定义中很隐蔽，值得反复确认。

第三，过滤表达式不是万能的。像publish_date >= '2024-01-01'这样的日期比较，由于我把日期存成了VARCHAR，比较时是按照字符串字典序比较的。如果日期统一是YYYY-MM-DD格式，字典序和真实日期序一致，没有问题；如果存在2024-1-1这种格式，那么排序和比较都会出错。所以日期字段入库前必须做格式化。

6.2 提升精度的重排模块设计

纯向量检索返回 TopK 结果，如果直接塞给大模型，通常会因为“语义最相近”但实际并不重要的问题导致回答失真。我在项目里引入了简单的**重排（Rerank）**策略。重排本身不一定要引入一个巨大的重排模型，通过规则与模型结合的方式，性价比最高。

我的重排思路是：在获得 Milvus 的 Top50 结果后，基于以下加权因子做一次排序：

这个规则型重排虽然朴素，但在实际业务数据上带来的改善非常明显。以“数据导入延迟”为例，用户问的其实更偏向“运维问题”，搜索出的结果如果都是“API 参考”类的说明，直接送到 LLM 生成的内容就会偏。重排可以把“故障处理”“运维指南”类的文档推到前面去。

重排实现可以直接写一个评分函数：

from datetime import datetime def rerank(query: str, candidates: list, current_date="2025-01-01", top_k=8): query_terms = set(query.lower().split()) scored = [] for r in candidates: content = r["entity"]["content"].lower() term_hits = sum(1 for term in query_terms if term in content) term_coverage = term_hits / len(query_terms) if query_terms else 0 pub_date = r["entity"].get("publish_date", "1900-01-01") days_diff = (datetime.strptime(current_date, "%Y-%m-%d") - datetime.strptime(pub_date, "%Y-%m-%d")).days freshness = max(0, 1 - days_diff / 3650) score = 0.5 * r["distance"] + 0.2 * term_coverage + 0.2 * freshness scored.append((score, r)) scored.sort(key=lambda x: x[0], reverse=True) return [r for _, r in scored[:top_k]]

这里的r["distance"]来自 Milvus 搜索结果的分数，需要先做归一化，否则不同指标下分数尺度完全不同。

6.3 上下文拼装与提示词工程

重排后，我要把命中的文档块拼装成给 LLM 的上下文。这里的关键是：不能直接把所有内容堆给模型，要把每条内容的来源和关键 Metadata 一并提供，让模型知道这段内容出自哪里、何时发布、属于什么分类。

我在提示词模板里做了如下设计：

from langchain_core.prompts import ChatPromptTemplate prompt_template = ChatPromptTemplate.from_messages([ ("system", """你是企业内部的知识库助手。请根据提供的参考资料，用中文回答用户问题。请遵循以下规则： 1. 只依据参考资料回答，不要编造没有依据的内容。 2. 如果资料中有明显冲突，指出冲突并说明各自的出处。 3. 在回答末尾，用一句话注明参考文档的分类和部门信息。 4. 如果资料与问题无关，直接表示未找到相关信息。"""), ("human", """参考资料如下： {context} 用户问题： {question} """) ])

为什么要让系统“注明参考文档的分类和部门信息”？这不仅能提升回答的可追溯性，还能让用户直观知道答案的适用边界。例如回答末尾指出“本答案参考了研发中心发布的《故障处理指南》”，使用者能立刻判断这个答案是否适用于自己当前的环境。

拼装 context 的代码：

def build_context(docs): context_parts = [] for doc in docs: meta = doc["entity"] content = meta.get("content", "") source_info = f"[来源: {meta.get('doc_id')}, 分类: {meta.get('category')}, 部门: {meta.get('department')}, 日期: {meta.get('publish_date')}]" context_parts.append(f"{source_info}\n{content}") return "\n\n".join(context_parts)

6.4 调用 Ollama 完成最终生成

构建好上下文后，调用 Ollama。这一步比较常规，但我仍然建议用 LangChain 的ChatOllama组件而不是直接 HTTP 请求，这样后续如果要切换模型服务商，只需换一个类，不用改链路代码。

from langchain_ollama import ChatOllama llm = ChatOllama(model="llama3.1:8b", base_url="http://localhost:11434", temperature=0.1) chain = prompt_template | llm response = chain.invoke({ "context": build_context(reranked_docs), "question": query_text }) print(response.content)

温度设置为 0.1 是为了让答案更具确定性，减少编造和随意发挥。这对有正确答案的题库类场景特别有效，但对需要生成创意的场景就要适当调高。

7. 实战排坑：我在真实调试中遇到的 7 个典型问题

这一部分是我自己最重要、也最想写给后来者看的。这 7 个问题，我在多个环境里反复撞过，几乎每一个都会卡住你半小时到一整天。

7.1 向量维度不一致导致查询报错

这是我遇到的第一个“硬坑”。在建立 Collection 时，我把维度写成了 512，但nomic-embed-text实际输出 768 维。写入时不会立刻报错，而是在写入第一个向量时就触发维度不匹配。关键是错误信息比较隐蔽，进入日志后才会看到 "dimension mismatch"。

解决方式：建立 Collection 前先验证维度：

resp = requests.post("http://localhost:11434/api/embed", json={"model": "nomic-embed-text", "input": "test"}) dim = len(resp.json()["embeddings"][0]) print(dim)

不要让 dim 和模型输出维度不一致。一旦定错了维度，重新建 Collection 往往比重建数据更省事。

7.2 JSON 类型字段过滤失效

在实际测试中，我发现tags字段如果定义成JSON类型，并且插入的值是 Python 的 list，在过滤时使用JSON_CONTAINS(tags, '稳定版')是可行的。但有些人会习惯性写成tags IN ['稳定版']，这种写法在 Milvus 中可能不会报错，但语义完全不一样，过滤结果直接为空。

建议每次过滤前先做一个最小集验证：

collection.query( expr="JSON_CONTAINS(tags, '稳定版')", output_fields=["tags", "doc_id"] )

如果返回空，再检查字段是否真的插入了数据。我之前遇到过“字段存量数据还没有 flush”的情况，插完库后没有 flush 就立刻查询，导致部分数据不可见。

7.3 文档切碎后 Metadata 丢失

这是 LangChain 概念中的常见问题：很多自定义切分器或 Loader 返回的 Document 对象里，metadata会随page_content一同带入split_documents方法，但是如果你在 Loader 阶段就把 metadata 清空或重置了，那么切分后的所有块都变成无源之水。

我从实践中学到的经验是：尽量在 Loader 阶段就保证 metadata 完整，后续所有操作只做增加、不做重置。在切分前，把Document(metadata=...)的 metadata 复制成字典并传给每个新块。

7.4 并发写入导致 Milvus 连接耗尽

在尝试做压测时，我开了多线程往 Milvus 写入数据，结果出现连接数暴涨、主进程报connection pool exhausted。原因是 Milvus 客户端默认连接池有限，多线程写入时每个线程都需要独立连接。

解决方案是引入连接池复用机制，或者把写入任务合并成批量写。批量写不仅性能高，还能避免连接反复创建销毁。单次插入建议控制在 1000~5000 条向量以内，如果超过这个量，分批次插入并保持flush()间隔适中。

7.5 Ollama 模型加载的显存与内存问题

我这个测试机没有独立显卡，用的是 CPU 推理。加载llama3.1:8b模型后，内存占用直接飙升到 12GB+，再加上 Embedding 模型，整体资源直接吃紧。如果同机部署 Milvus，很容易内存溢出。

一个经验是：在配置文件中限制 Ollama 的并发请求数。Ollama 默认情况下对同一模型的请求会串行排队，但如果你一次发多个请求，它会尝试加载多个模型实例，内存急剧膨胀。建议在启动服务时显式设置环境变量来控制并发。

OLLAMA_NUM_PARALLEL=1 OLLAMA_MAX_LOADED_MODELS=1 ollama serve

7.6 Metadata 过滤后召回结果过少

测试时我加了department == '研发中心'的过滤条件，结果返回为空。原因是有一批文档在写入时department字段是空字符串，而查询条件里写了== '研发中心'，空字符串自然不匹配。

这里的处理策略是：写入前做字段归一化，确保所有字段都存在并且有明确默认值。不要留空字符串给过滤链路去猜。

7.7 生成阶段上下文超长

当 TopK 调得太大时，context 直接超过 LLM 的上下文窗口会报错（Llama 3.1 8B 的上下文是 128K，但实际塞给它的内容要减去提示词本身占用的空间）。经验建议是：先用小模型做测试时把top_k设置到 4~6 即可，在正式环境再根据上下文窗口适当增加。

8. 生产化增强：基于 Tag 的权限过滤与检索隔离

真正能称之为“生产级”的 RAG，不能只满足于“能回答问题”，还要回答得可控、可追溯、有边界。基于 Metadata 的隔离能力，可以在检索层就实现“不同部门看到不同文档”的效果。

我实现了一个比较简单的多租户过滤方案：

实现上，只需要在查询入口注入对应的filter_expr，并把当前用户的角色映射成过滤条件即可。这个方案不额外增加 Milvus 复杂操作，却在业务权限语义上非常清晰。

MTLS（多租户）实践中有一个特别注意点：过滤条件中的字段一定要走索引。像category、department、publish_date这类高频过滤字段，在建表时就要设置为indexed，否则过滤操作会退化成全表扫描，查询性能急剧下降。Milvus 提供自动索引，但只有将标量字段设置为indexed=True才会走内存索引加速，否则默认是普通字段。

9. 接下来还能怎么扩展

如果你照着上面的链路搭完了第一版，那么后续值得投入的方向，按性价比排序：

1. 引入混合检索。向量检索负责语义，BM25 负责关键词精确匹配，用 RRF（Reciprocal Rank Fusion）做结果融合。现在 LangChain 里已经有ensemble_retriever可以直接组合。
2. 引入真实重排模型。规则重排能覆盖大部分基础场景，但到了需要精细排序的业务场景，就需要像 BGE-reranker 这样的专门模型。
3. 为每个知识库建立版本管理。文档会更新，旧的向量要标记为“过期版本”，通过 Metadata 中is_current_version字段屏蔽旧内容。
4. 做基于流式输出的交互体验。Ollama 支持流式输出，LangChain 也支持stream方法，在生产界面中非常重要。

我个人在实际操作中的体会是：RAG 不是一个“搭完就完”的项目，它更像一个持续迭代的数据工程问题。你的数据形态在变、业务标签体系在变、用户问题分布也在变。真正常态化起作用的，恰恰是那套把 Metadata 解析、过滤、追踪做扎实的基本功。最后再分享一个小技巧：任何一次检索效果变差，先不要急着调模型或者换提示词，先去看召回的前 20 条结果里，到底有多少条是真正相关的。这一看，绝大多数问题的原因立刻浮出水面。