基于pgvector与Milvus的RAG可追溯报告生成：DocResearch架构与Agent并发实践

1. 一条命令背后到底发生了什么

DocResearch 这个名字听起来像是个普通的文档检索工具，但真正让我感兴趣的，是它那句描述——“从一条命令到一份有出处的报告”。这句话里藏着两个关键信息：第一，入口极简，一条命令；第二，出口可信，报告里每个结论都能追溯到原始文档。这两点恰恰是当前大多数 RAG（检索增强生成）项目最容易翻车的地方——要么用起来太复杂，要么生成的内容全是“幻觉”，你根本不知道它从哪编出来的。

我自己做过好几个知识库问答项目，踩过最深的坑就是：模型回答得头头是道，但你一查原文，发现它把三段不相干的内容缝合在了一起。DocResearch 想解决的就是这个问题。它的核心思路是把“检索”和“生成”拆成两个独立但可追溯的阶段，检索阶段负责找到原文片段并记录来源，生成阶段只允许基于这些片段组织语言，最后输出时把引用标注带上。

这篇文章适合谁看？如果你正在做 AI Agent 开发、知识库系统、或者任何需要“让模型说人话且说真话”的项目，那这篇内容应该能帮你省下不少试错时间。我会从架构设计、向量库选型、Agent 编排、并发处理、以及实际部署中遇到的坑这几个角度，把 DocResearch 这类项目的完整实现链路拆开讲清楚。关键词里提到的 pgvector、Milvus、Python、Agent 这些都会涉及到，但我不会只讲概念，而是讲它们在真实项目里怎么配合、怎么选、怎么调。

先说结论：DocResearch 这类项目的技术栈并不复杂，难的是把“可追溯”这个要求贯穿到每一行代码里。下面我按实际搭建顺序来展开。

2. 为什么“有出处”比“答得好”更难做到

2.1 检索增强生成的根本矛盾

RAG 的基本流程大家都熟：用户提问 → 向量检索 → 拼接上下文 → 模型生成。问题出在第三步和第四步之间。当你把检索到的 5 个片段拼成一个 prompt 丢给模型时，模型并不知道哪个片段对应哪个来源，它只是把所有文本当成一个整体来理解。结果就是，它可能用片段 A 的信息回答了片段 B 的问题，或者把多个片段的结论混合成一个新结论——而这个新结论在任何单个片段里都不存在。

这就是“有出处”难做的根本原因：检索阶段丢失了来源与内容的绑定关系。大多数实现只把文本内容拼进 prompt，来源信息（文件名、页码、段落编号）要么被丢弃，要么只作为元数据存在数据库里，生成时根本用不上。

DocResearch 的做法是在检索阶段就给每个片段打上唯一标识，生成时要求模型在输出中引用这些标识。具体来说，检索返回的不是纯文本列表，而是带 ID 的结构化片段：

# 检索结果的结构示例 retrieved_chunks = [ { "chunk_id": "doc_001_p3_c2", "source": "产品需求文档_v2.pdf", "page": 3, "content": "系统需要支持多租户隔离...", "score": 0.87 }, { "chunk_id": "doc_007_p1_c1", "source": "技术架构说明.md", "page": 1, "content": "租户数据通过 schema 级别隔离...", "score": 0.82 } ]

生成阶段的 prompt 里明确要求：每个结论后面必须用[chunk_id]标注来源。这样最终报告里的每一句话都能对应回原始文档的具体位置。

2.2 分块策略决定了可追溯性的上限

很多人做 RAG 时对分块（chunking）很随意，固定 512 个 token 切一刀就完事。但分块方式直接决定了你能不能准确定位来源。如果一刀切在句子中间，检索出来的片段本身就是残缺的，模型理解都会出问题，更别说追溯了。

我在实际项目里总结了几条分块原则：

• 按语义边界切：优先在段落、标题、列表项之间切分，而不是按固定长度硬切。可以用简单的规则（如按\n\n分段）配合长度限制。
• 保留上下文头：每个 chunk 前面带上所属章节的标题路径，比如[第三章 系统架构 > 3.2 数据隔离]，这样即使片段被单独检索出来，模型也能知道它在原文中的位置。
• 重叠窗口：相邻 chunk 之间保留 10%–15% 的重叠内容，避免关键信息刚好落在切割线上被截断。
• chunk_id 编码位置信息：像doc_001_p3_c2这样的 ID 本身就包含了文档编号、页码、块序号，追溯时不需要额外查询。

提示：分块长度没有万能值。技术文档建议 300–500 token，对话记录可以更短（200–300），法律合同这类需要精确引用的可以到 800–1000。关键是保证每个 chunk 是一个完整的语义单元。

2.3 生成阶段的引用约束怎么落地

光在 prompt 里写“请标注来源”是不够的，模型经常会偷懒或者编造 ID。我的做法是加一层后处理校验：生成完成后，用正则提取所有[chunk_id]引用，检查每个 ID 是否真的存在于本次检索结果中。如果发现不存在的 ID，要么重新生成，要么把该引用标记为“未验证”。

更严格的做法是要求模型输出结构化格式（如 JSON），每个结论作为一个对象，包含text和citations两个字段。这样校验起来更可靠：

import json import re def validate_citations(report_text, valid_chunk_ids): """校验报告中引用的 chunk_id 是否都来自检索结果""" cited_ids = set(re.findall(r'\[([a-zA-Z0-9_]+)\]', report_text)) invalid = cited_ids - set(valid_chunk_ids) if invalid: return False, f"发现无效引用: {invalid}" return True, "所有引用均有效"

这个校验步骤看起来简单，但它把“可追溯”从一个软要求变成了硬约束。模型知道后面会被检查，生成时也会更规矩。

3. 向量库选型：pgvector 和 Milvus 各自的主场

3.1 两个库的定位差异

关键词里同时出现了 pgvector 和 Milvus，这两个不是竞争关系，而是适用场景不同。我在不同规模的项目里都用过，感受很明确：

DocResearch 这种项目，如果文档量在几万到十几万这个量级，pgvector 完全够用，而且省去了单独维护一个向量数据库的麻烦。你的文档元数据、用户权限、检索日志本来就要存关系型数据库，向量也放同一个库里，查询时一个 SQL 就能搞定“向量相似度 + 元数据过滤”的混合检索。

3.2 pgvector 的实际配置要点

在 PostgreSQL 里启用 pgvector 很简单，但有几个参数直接影响检索效果和性能：

-- 创建扩展 CREATE EXTENSION IF NOT EXISTS vector; -- 建表，向量维度根据你的 embedding 模型定 CREATE TABLE doc_chunks ( id SERIAL PRIMARY KEY, chunk_id VARCHAR(64) UNIQUE NOT NULL, source_file VARCHAR(255), page_num INT, content TEXT, embedding vector(1536) -- 对应 text-embedding-3-small ); -- 创建 HNSW 索引，比 IVFFlat 更适合动态数据 CREATE INDEX ON doc_chunks USING hnsw (embedding vector_cosine_ops) WITH (m = 16, ef_construction = 64);

这里有几个经验值：m=16是 HNSW 图的连接数，越大召回率越高但内存占用也越大，16 是大多数场景的平衡点。ef_construction=64控制建索引时的搜索范围，64 对于十万级数据足够。查询时的ef_search参数可以动态调整，设大一点召回率高但慢，我一般设 40–100 之间。

注意：pgvector 的余弦距离用<=>操作符，不是<->（那是欧氏距离）。如果你用错了操作符，检索结果会完全不对，而且不会报错，非常隐蔽。

3.3 Milvus 什么时候该上场

当你的文档量超过百万级，或者需要支持高并发检索（比如几十个 Agent 同时查询），pgvector 就开始吃力了。Milvus 的分布式架构就是为这个场景设计的。standalone 模式适合单机部署，数据量在千万级以内性能很好；再往上就要上集群模式。

Milvus 的安装用 Docker 是最省事的：

# 下载 standalone 的 docker-compose 配置 wget https://github.com/milvus-io/milvus/releases/download/v2.4.0/milvus-standalone-docker-compose.yml -O docker-compose.yml # 启动 docker compose up -d

启动后默认端口是 19530。Python 端用pymilvus连接：

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType connections.connect(host='localhost', port='19530') # 定义 schema fields = [ FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True), FieldSchema(name="chunk_id", dtype=DataType.VARCHAR, max_length=64), FieldSchema(name="content", dtype=DataType.VARCHAR, max_length=65535), FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=1536) ] schema = CollectionSchema(fields, description="文档块") collection = Collection(name="doc_chunks", schema=schema) # 创建索引 index_params = { "metric_type": "COSINE", "index_type": "HNSW", "params": {"M": 16, "efConstruction": 64} } collection.create_index(field_name="embedding", index_params=index_params)

Milvus 的余弦相似度用COSINE作为 metric_type，这点和 pgvector 的逻辑一致。但要注意 Milvus 返回的是相似度分数（越大越相似），而 pgvector 的<=>返回的是距离（越小越相似），写代码时别搞混了。

3.4 我的选型建议

如果你刚开始做 DocResearch 这类项目，我的建议是：先用 pgvector 跑通全流程，等数据量或并发真的成为瓶颈了再迁 Milvus。迁移成本主要在 embedding 的重新导入和索引重建，业务代码改动不大，因为检索接口可以抽象成统一的一层。

抽象接口大概长这样：

class VectorStore: def search(self, query_embedding, top_k=5, filters=None): raise NotImplementedError class PgVectorStore(VectorStore): def search(self, query_embedding, top_k=5, filters=None): # SQL 查询实现 pass class MilvusStore(VectorStore): def search(self, query_embedding, top_k=5, filters=None): # Milvus 查询实现 pass

这样上层 Agent 逻辑完全不用改，换存储只是换一个实现类的事。

4. Agent 编排：从命令解析到报告生成的全链路

4.1 为什么这里需要 Agent 而不是简单脚本

一条命令到一份报告，中间其实有好几个决策点：用户的问题需要拆成几个子查询？检索到的片段够不够回答问题？需不需要追加检索？报告的结构怎么组织？这些决策如果全用 if-else 写死，维护起来会非常痛苦。Agent 的价值在于把这些决策交给模型来做，代码只负责提供工具和约束。

但这里有个常见的误解：Agent 不是越自主越好。DocResearch 这种场景，我倾向于用“有限自主”的设计——Agent 可以决定检索什么、检索几次，但生成报告的格式和引用规则是硬约束，不允许它自由发挥。

4.2 工具定义与调用链路

Agent 需要几个核心工具：

tools = [ { "name": "search_documents", "description": "根据查询语句检索相关文档片段，返回带 chunk_id 的结果", "parameters": { "query": "检索查询语句", "top_k": "返回结果数量，默认5" } }, { "name": "get_chunk_detail", "description": "根据 chunk_id 获取完整片段内容及上下文", "parameters": { "chunk_id": "片段唯一标识" } }, { "name": "generate_report", "description": "基于已收集的片段生成带引用的报告", "parameters": { "question": "原始问题", "chunks": "已检索的片段列表" } } ]

实际运行时的链路是这样的：用户输入问题 → Agent 调用search_documents→ 拿到片段后判断信息是否充分 → 不充分则调整查询词再次检索 → 充分则调用generate_report→ 后处理校验引用 → 输出最终报告。

这个链路里最关键的是“判断信息是否充分”这一步。我的做法是让模型输出一个结构化的判断：

{ "sufficient": false, "missing": "缺少关于数据隔离具体实现方案的信息", "next_query": "多租户数据隔离 实现方案 schema" }

这样 Agent 就知道下一步该检索什么，而不是盲目地重复检索。

4.3 并发场景下的 Agent 设计

关键词里有个“ai agent 怎么扛并发”，这是实际部署时绕不开的问题。单个 Agent 处理一个请求没问题，但同时来几十个请求时，瓶颈通常不在模型调用，而在向量检索和数据库连接。

我的处理方式分三层：

第一层是连接池。无论是 pgvector 还是 Milvus，都要用连接池管理连接。PostgreSQL 用psycopg2.pool或 SQLAlchemy 的 pool，Milvus 的connections本身支持多连接，但要注意每个线程用独立连接。

第二层是检索缓存。相同或相似的查询没必要每次都打向量库。我用 Redis 做了一层查询缓存，key 是查询语句的 embedding 哈希，value 是检索结果。缓存命中率在重复问题较多的场景下能到 40% 以上。

第三层是异步化。Agent 的多个检索步骤之间如果是独立的，可以用asyncio并发执行。比如同时检索三个不同的子查询，而不是串行等三个结果。

import asyncio async def parallel_search(queries, vector_store): tasks = [vector_store.async_search(q) for q in queries] results = await asyncio.gather(*tasks) return results

这样原本串行需要 3 秒的三个检索，并发后可能只要 1.2 秒。

4.4 Agent 安全边界

Agent 自主决策带来便利的同时也带来风险。DocResearch 场景下我关注两个安全点：一是检索范围不能越权，用户只能检索自己有权限的文档；二是生成内容不能包含敏感信息，即使检索到了也要过滤。

权限控制的做法是在向量检索时加元数据过滤条件。pgvector 的查询可以这样写：

SELECT chunk_id, content, 1 - (embedding <=> %s) AS similarity FROM doc_chunks WHERE source_file = ANY(%s) -- 用户有权限的文件列表 ORDER BY embedding <=> %s LIMIT %s;

这样从检索层面就保证了不会返回越权内容，比在生成后再过滤要可靠得多。

5. 从零跑通 DocResearch 的实操步骤

5.1 环境准备与依赖安装

先把基础环境搭起来。Python 版本建议 3.10 以上，因为要用到一些新的类型注解语法。

# 创建虚拟环境 python -m venv docresearch-env source docresearch-env/bin/activate # Windows 用 docresearch-env\Scripts\activate # 安装核心依赖 pip install openai psycopg2-binary pgvector python-dotenv pip install pymilvus # 如果用 Milvus pip install numpy tiktoken

如果你在 Mac 上用 Docker 装 Milvus，注意 Docker Desktop 的内存分配至少给 4GB，否则 Milvus 启动时会因为内存不足反复重启。这个坑我踩过，日志里只显示连接超时，排查了半天才发现是 Docker 内存不够。

5.2 文档入库的完整流程

文档入库分四步：读取 → 分块 → 向量化 → 存储。

import os from pathlib import Path def load_documents(doc_dir): """读取目录下所有文档""" docs = [] for path in Path(doc_dir).rglob("*"): if path.suffix in [".txt", ".md", ".pdf"]: # PDF 需要用 pypdf 等库提取文本 if path.suffix == ".pdf": from pypdf import PdfReader reader = PdfReader(path) text = "\n".join(page.extract_text() for page in reader.pages) else: text = path.read_text(encoding="utf-8") docs.append({"source": path.name, "text": text}) return docs def chunk_document(doc, max_tokens=400, overlap=50): """按段落分块，保留章节标题""" paragraphs = doc["text"].split("\n\n") chunks = [] current = [] current_len = 0 for para in paragraphs: para_len = len(para) // 4 # 粗略估算 token 数 if current_len + para_len > max_tokens and current: chunk_text = "\n\n".join(current) chunks.append({ "source": doc["source"], "content": chunk_text, "chunk_id": f"{doc['source']}_{len(chunks)}" }) # 保留重叠部分 current = current[-1:] if overlap > 0 else [] current_len = len(current[0]) // 4 if current else 0 current.append(para) current_len += para_len if current: chunks.append({ "source": doc["source"], "content": "\n\n".join(current), "chunk_id": f"{doc['source']}_{len(chunks)}" }) return chunks

分块完成后调用 embedding 接口批量向量化，然后写入向量库。批量写入时注意控制每批的大小，pgvector 建议每批 100–500 条，Milvus 可以到 1000 条。

5.3 检索与生成的串联

检索环节我建议做一次“粗排 + 精排”。粗排用向量相似度取 top 20，精排可以用一个交叉编码器（cross-encoder）或者简单的关键词匹配来重新排序，取 top 5 送给模型。这样能显著提升检索质量。

def retrieve_and_rerank(query, vector_store, top_k=5): # 粗排：向量检索取 top 20 candidates = vector_store.search(query, top_k=20) # 精排：用关键词覆盖率重新打分 query_terms = set(query.lower().split()) for c in candidates: content_terms = set(c["content"].lower().split()) keyword_score = len(query_terms & content_terms) / max(len(query_terms), 1) c["final_score"] = 0.7 * c["score"] + 0.3 * keyword_score candidates.sort(key=lambda x: x["final_score"], reverse=True) return candidates[:top_k]

生成阶段的 prompt 模板要明确引用格式：

你是一个文档分析助手。基于以下文档片段回答问题。 每个结论后面必须用 [chunk_id] 标注来源。 如果片段中没有相关信息，明确说"文档中未提及"，不要编造。 文档片段： {chunks} 问题：{question} 请输出带引用的报告：

5.4 报告输出与引用校验

生成完成后，除了前面说的引用校验，我还会做一步“引用密度检查”——如果报告里超过 30% 的句子没有引用，说明模型可能在自由发挥，需要重新生成或者标记为低置信度。

def check_citation_density(report_text): sentences = [s for s in report_text.split("。") if len(s.strip()) > 10] cited = [s for s in sentences if re.search(r'\[[a-zA-Z0-9_]+\]', s)] density = len(cited) / max(len(sentences), 1) return density # 密度低于 0.7 时触发重新生成

这个阈值可以根据文档类型调整，技术文档要求高一些（0.8），综述类可以低一些（0.5）。

6. 实际部署中踩过的坑与应对

6.1 向量维度不匹配的隐蔽错误

有一次我换了 embedding 模型，从 1536 维换到 3072 维，但忘了改数据库里的 vector 维度定义。结果插入数据时 PostgreSQL 直接报错，还算好排查。更隐蔽的是 Milvus，它会在插入时静默截断或填充维度，检索结果完全不对但没有任何报错。所以换模型时一定要同步检查向量库的维度定义。

6.2 分块边界导致的关键信息丢失

有个文档里有一张对比表格，被分块逻辑从中间切开了，前半部分在一个 chunk 里，后半部分在另一个 chunk 里。用户问表格里的对比结论时，检索只召回了前半部分，模型基于残缺信息给出了错误答案。后来我在分块时加了规则：检测到 Markdown 表格或代码块时，尽量保持完整，如果超过长度限制就单独成块并标记。

6.3 并发下的连接耗尽

用 pgvector 时，每个 Agent 请求都开一个新数据库连接，并发一上来连接数就爆了。PostgreSQL 默认最大连接数是 100，几十个并发请求就能打满。解决方案是用连接池，并且把 Agent 的数据库操作改成异步的。如果用的是同步驱动，至少要用ThreadedConnectionPool限制最大连接数。

6.4 引用 ID 冲突

多个文档的 chunk_id 如果只用序号生成，不同文档之间会冲突。比如两个文档都有chunk_0，追溯时就分不清是哪个文档的。我的做法是在 chunk_id 里带上文档的哈希前缀，比如a3f2_0、b7c1_0，保证全局唯一。

6.5 模型不遵守引用格式

即使 prompt 里写得很清楚，模型有时还是会忘记加引用，或者用错误的格式（比如用圆括号而不是方括号）。我的应对是在 prompt 里给一个具体的输出示例，并且用 few-shot 的方式展示正确和错误的引用格式。实测下来，给了示例之后格式遵守率从 70% 提升到了 95% 以上。

7. 一些关于扩展方向的个人想法

这套架构跑通之后，扩展空间其实很大。比如可以加一个“多轮追问”的能力，用户对报告里的某个结论不满意，可以针对那个 chunk_id 追问细节，Agent 只需要重新检索相关片段并生成补充说明。再比如可以做“报告模板化”，不同场景用不同的输出结构，技术调研报告、竞品分析、会议纪要各有各的模板，Agent 根据问题类型自动选择。

另外，如果你想把 DocResearch 做成一个长期运行的服务，建议把检索日志和用户反馈都存下来。哪些查询经常被检索、哪些引用被用户标记为不准确，这些数据可以用来优化分块策略和检索排序，形成一个正向循环。

我在实际使用中最大的体会是：可追溯性不是加一个功能就能实现的，它需要从分块、检索、生成到后处理的每一个环节都保持一致的设计。任何一环偷懒，最终报告的可信度都会打折扣。DocResearch 这个项目名起得很准确，它做的就是把“研究”这个动作变得有据可查。