BigGraph：RAG + Agent 混合检索平台

项目背景

BigGraph 是我投入精力最多的一个项目，目标是从零构建一个生产级的 RAG（检索增强生成）+ Agent 混合检索平台。传统的 RAG 系统往往只依赖单一的向量检索（如纯 Qdrant 或 Pinecone），但实际场景中用户的查询需求差异很大——有的需要语义匹配（"类似文档"），有的需要精确关键词匹配（"2024年Q3财报"），有的需要复杂的多步推理（"分析A和B的差异并生成报告"）。

为了解决这些场景，BigGraph 定位为一个"AI 知识库操作系统"：能接入多种格式的文档（PDF/Word/Excel/MD/CSV/HTML/TXT），做结构化解析和智能分块，同时提供向量检索 + 全文检索两道通道，并在上层用 LangGraph 编排 Agent 实现多步推理和工具调用。

技术选型

组件	选择	理由
后端框架	FastAPI	异步原生，类型安全，自动生成 OpenAPI 文档
向量数据库	Qdrant	支持 payload 过滤、高性能、可自托管
全文检索	Elasticsearch	成熟的 BM25 算法、高亮和聚合分析
Agent 引擎	LangGraph	有状态图编排，天然支持多步推理和 Checkpoint
嵌入模型	OpenAI Embeddings API	兼容第三方代理，支持 text-embedding-3-small 等
关系存储	PostgreSQL + SQLAlchemy	知识库元数据、用户、文档关系管理
缓存/会话	Redis	聊天历史缓存、会话管理
认证	JWT（PyJWT）	无状态认证，支持多租户

架构设计

BigGraph 的整体架构分为四层：

1. 接入层（API Layer）
FastAPI 承载所有 RESTful 接口，包括知识库 CRUD、文档上传、检索查询、聊天会话管理。使用 JWT 中间件做认证，RequestID 中间件做日志追踪。

2. 解析层（Ingestion Pipeline）
上传文件后进入后台处理流水线：

• ParserRegistry 根据文件类型自动选择解析器（PDF/DOCX/CSV/Excel/HTML/MD/TXT）
• StructureAwareChunker 对解析后的文本做智能分块，保留标题层级、段落结构和表格信息
• 分块后同时写入 Qdrant（向量）+ Elasticsearch（全文）+ PostgreSQL（元数据）

3. 检索层（Retrieval Layer）
采用 Hybrid 检索策略：

• Qdrant 做稠密向量检索（语义匹配）
• Elasticsearch 做稀疏 BM25 检索（关键词匹配）
• RRF（Reciprocal Rank Fusion）算法融合两个结果集，输出最终 Top-K

4. 推理层（Agent Layer）
基于 LangGraph StateGraph 构建 Agent，支持多轮对话中的上下文记忆和工具调用。Agent 可以调用检索工具、生成报告、分析对比文档等。

核心实现

混合检索服务

class HybridRetrievalService:
    def __init__(self) -> None:
        self.qdrant = QdrantService()
        self.es = ElasticsearchService()
        self.embedder = EmbeddingService()

    def search(self, query_text: str, tenant_id: str,
               knowledge_base_ids: list[str] | None = None,
               top_k: int = 10) -> list[dict]:
        # 1. 向量检索
        query_vector = self.embedder.embed_query(query_text)
        dense = self.qdrant.search(query_vector=query_vector,
                                   tenant_id=tenant_id,
                                   knowledge_base_ids=knowledge_base_ids,
                                   top_k=top_k * 3)
        for r in dense:
            r["_source"], r["bm25_score"] = "qdrant", 0.0

        # 2. 全文检索
        bm25 = self.es.search_bm25(query=query_text,
                                   tenant_id=tenant_id,
                                   knowledge_base_ids=knowledge_base_ids,
                                   top_k=top_k * 3)
        for r in bm25:
            r["_source"], r["bm25_score"] = "es", r.get("score", 0.0)

        # 3. RRF 融合
        fused = reciprocal_rank_fusion(dense, bm25, k=60, top_k=top_k)

        # 4. 补充完整内容
        chunk_ids = [r["chunk_id"] for r in fused if r.get("chunk_id")]
        full_contents = self.es.get_full_contents(chunk_ids)
        for r in fused:
            cid = r.get("chunk_id", "")
            if cid in full_contents:
                r["full_content"] = full_contents[cid]
        return fused

RRF 排序融合算法

def reciprocal_rank_fusion(qdrant_results: list[dict],
                           es_results: list[dict],
                           k: int = 60, top_k: int = 10) -> list[dict]:
    rrf: dict[str, float] = {}
    records: dict[str, dict] = {}
    for rank, item in enumerate(qdrant_results, start=1):
        cid = item.get("chunk_id", "")
        if cid:
            rrf[cid] = rrf.get(cid, 0.0) + 1.0 / (k + rank)
            records[cid] = item
    for rank, item in enumerate(es_results, start=1):
        cid = item.get("chunk_id", "")
        if cid:
            rrf[cid] = rrf.get(cid, 0.0) + 1.0 / (k + rank)
            if cid not in records:
                records[cid] = item
    sorted_ids = sorted(rrf, key=lambda cid: rrf[cid], reverse=True)[:top_k]
    results = []
    for cid in sorted_ids:
        item = dict(records[cid])
        item["rrf_score"] = rrf[cid]
        results.append(item)
    return results

智能文档分块器（StructureAwareChunker）

解析管线中最关键的是分块策略。传统的固定窗口分块会破坏语义边界。BigGraph 实现了一个结构感知分块器：

class StructureAwareChunker:
    def __init__(self, max_chunk_size: int = 1024, overlap: int = 128):
        self.max_chunk_size = max_chunk_size
        self.overlap = overlap

    def chunk_document(self, parsed_doc: ParsedDocument) -> list[Chunk]:
        chunks = []
        current_chunk = []
        current_size = 0
        for element in parsed_doc.elements:
            elem_size = len(element.text)
            if current_size + elem_size > self.max_chunk_size and current_chunk:
                chunks.append(self._build_chunk(current_chunk, parsed_doc))
                # 保留 overlap 元素用于上下文连贯
                overlap_elements = self._find_overlap(current_chunk, self.overlap)
                current_chunk = list(overlap_elements)
                current_size = sum(len(e.text) for e in current_chunk)
            current_chunk.append(element)
            current_size += elem_size
        if current_chunk:
            chunks.append(self._build_chunk(current_chunk, parsed_doc))
        return chunks

踩坑记录

1. Qdrant payload 大小限制
Qdrant 对每个 point 的 payload 有大小限制，最初我们把完整的分块内容全部写入 payload，导致大文档存储失败。解决方案：payload 只存前 200 字符作为摘要，完整内容存储在 ES 和 PG 中，检索时通过 chunk_id 回查。

2. Embedding 批量请求限速
OpenAI Embedding API 有 RPM（每分钟请求数）限制。在批量处理大量文档时，直接并行发送所有请求会触发 429。解决方案：加入分批处理逻辑，每批 32 条，逐批提交。

3. 混合检索中的分数不可比
Qdrant 返回 cosine 相似度（0~1），ES 返回 BM25 分数（无上界），两者直接相加或比较没有意义。RRF 算法通过排名位置而非分数来融合，完美解决了这个问题。

4. LangGraph Checkpoint 序列化
LangGraph 的 State 中如果包含自定义对象（如 Pydantic 模型），Checkpoint 持久化时可能失败。解决方案：确保所有 State 字段都是 JSON 可序列化的基本类型。

总结

BigGraph 是我目前做过最完整的 RAG 项目。它不只是一个 demo，而是一个可以真正部署使用的企业级知识库系统。混合检索策略显著提升了检索质量——向量检索找回语义相似但关键词不同的结果，BM25 补回精确匹配但语义距离较远的结果。LangGraph 的引入让系统具备了 Agent 能力，可以执行多步推理任务。

项目仍在迭代中，下一步计划加入更多的 Agent 工具（如代码执行、图表生成）和完善的监控告警系统。总体来说，BigGraph 是 RAG 从实验走向生产的一次完整实践。