git clone <repo-url>
cd Modular-RAG-MCP-Server
python -m venv .venv
.venv\Scripts\activate      # Windows
# source .venv/bin/activate # macOS/Linux
pip install -r requirements.txt

llm:
  provider: "deepseek"          # 或 openai / azure / ollama
  model: "deepseek-v4-pro"
  api_key: "sk-xxx"             # 你的 API Key

embedding:
  provider: "openai"
  model: "text-embedding-v3"
  api_key: "sk-xxx"
  base_url: "https://dashscope.aliyuncs.com/compatible-mode/v1"

vector_store:
  provider: "chroma"
  persist_directory: "./data/db/chroma"

python scripts/ingest.py --file data/documents/your_doc.pdf --collection default

python scripts/query.py --query "你的问题" --top-k 5

streamlit run src/observability/dashboard/app.py

python -m src.mcp_server.server

PDF → Load → Chunk → Transform → Dual Encode → Store
                                                    ↓
                                               ChromaDB + BM25
                                                    ↓
                          Query → IntentRoute → HybridSearch → Rerank → Response
                                                    ↓
                                          Dashboard + Trace 可观测

LLMFactory._PROVIDERS = {openai, azure, deepseek, ollama}
EmbeddingFactory._PROVIDERS = {openai, azure, ollama}
VectorStoreFactory._PROVIDERS = {chroma, ...}
→ 改一行 YAML 即可切 Provider

MODULAR-RAG-MCP-SERVER/
├── config/
│   └── settings.yaml              # 全局配置（LLM/Embedding/VectorStore/Ingestion/Rerank）
├── scripts/
│   ├── query.py                   # CLI 查询脚本（含 IntentRouter）
│   ├── ingest.py                  # CLI 摄入脚本
│   └── evaluate.py                # CLI 评估脚本
├── tests/
│   ├── unit/                      # 单元测试（test_intent_router 等）
│   ├── integration/               # 集成测试
│   └── fixtures/                  # 测试数据（golden_test_set.json）
├── src/
│   ├── core/                      # 核心层：类型定义 + 配置 + 检索引擎 + 响应构建
│   │   ├── types.py               # Document / Chunk / RetrievalResult 等核心数据结构
│   │   ├── settings.py            # 配置加载与解析
│   │   ├── trace/                 # Trace 上下文与收集器
│   │   ├── query_engine/          # 检索引擎
│   │   │   ├── intent_router.py   # 🆕 查询意图路由器
│   │   │   ├── hybrid_search.py   # Dense + Sparse + RRF 融合
│   │   │   ├── dense_retriever.py # 向量检索
│   │   │   ├── sparse_retriever.py# BM25 检索
│   │   │   ├── fusion.py          # RRF 融合算法
│   │   │   ├── reranker.py        # 重排序
│   │   │   └── query_processor.py # 查询预处理（分词/关键词提取）
│   │   └── response/              # MCP 响应构建
│   │       └── response_builder.py
│   ├── libs/                      # 可插拔库层：工厂模式 + 抽象基类 + 各 Provider 实现
│   │   ├── llm/                   # LLM Provider（OpenAI/Azure/DeepSeek/Ollama + Vision）
│   │   │   ├── base_llm.py        # BaseLLM 抽象类
│   │   │   ├── llm_factory.py     # LLMFactory（双注册表：文本 + 视觉）
│   │   │   ├── deepseek_llm.py
│   │   │   ├── openai_llm.py
│   │   │   ├── azure_llm.py
│   │   │   └── ollama_llm.py
│   │   ├── embedding/             # Embedding Provider
│   │   │   ├── base_embedding.py
│   │   │   └── embedding_factory.py
│   │   ├── vector_store/          # VectorStore Provider（ChromaDB 等）
│   │   ├── splitter/              # 文本分割器（Recursive 等）
│   │   ├── loader/                # 文档加载器 + 文件完整性检查
│   │   ├── reranker/              # 重排序器工厂
│   │   └── evaluator/             # 评估器工厂
│   ├── ingestion/                 # 摄入管道
│   │   ├── pipeline.py            # 🎯 6 阶段管道编排器
│   │   ├── chunking/              # 文档分块
│   │   ├── transform/             # Chunk 精炼/元数据增强/图片描述
│   │   ├── embedding/             # Dense/Sparse 编码 + 批处理
│   │   └── storage/               # BM25 索引 / ChromaDB Upsert / 图片存储
│   ├── mcp_server/                # MCP Server
│   │   ├── server.py              # stdio 启动入口
│   │   ├── protocol_handler.py    # JSON-RPC 协议处理 + Tool 注册
│   │   └── tools/                 # MCP Tool 实现
│   │       ├── query_knowledge_hub.py  # 🎯 核心检索 Tool
│   │       ├── list_collections.py
│   │       └── get_document_summary.py
│   └── observability/             # 可观测性
│       ├── dashboard/             # Streamlit Dashboard
│       │   ├── app.py             # 导航入口（6 页面注册）
│       │   ├── pages/             # 6 个页面实现
│       │   └── services/          # ConfigService / DataService / TraceService
│       └── evaluation/            # Ragas 评估器 + EvalRunner
└── data/                          # 运行时数据（gitignore）
    ├── db/chroma/                 # ChromaDB 持久化
    ├── db/bm25/                   # BM25 索引文件
    └── images/                    # 提取的图片

PDF 文件
  │
  ▼
Stage 1: File Integrity Check    ← SHA256 hash → SQLite 查重，已处理过的跳过
  │
  ▼
Stage 2: Document Loading        ← PDF → Markdown（markitdown），提取图片
  │                                 输出: Document(text + metadata.images)
  ▼
Stage 3: Chunking                ← Recursive 分块，800 chars + 150 overlap
  │                                 输出: List[Chunk]（ID 格式: doc_hash_index_contenthash）
  ▼
Stage 4: Transform Pipeline      ← 3 个子步骤串联
  │  4a: Chunk Refinement        ← LLM 优化文本 / Rule 规则清理
  │  4b: Metadata Enrichment     ← LLM 提取标题/标签/摘要 / Rule 正则提取
  │  4c: Image Captioning        ← Vision LLM 生成图片描述，缝合进 Chunk.text
  ▼
Stage 5: Encoding                ← Dense（Embedding API → float vector）
  │                                + Sparse（本地 BM25 词频统计）
  ▼
Stage 6: Storage                 ← 3 路并行写入
     6a: ChromaDB Upsert         ← 向量 + metadata
     6b: BM25 Index              ← Whoosh 倒排索引（chunk_id 与 Chroma 对齐）
     6c: Image Storage Index     ← 图片元数据注册

用户查询: "Azure OpenAI 怎么配置？"
  │
  ▼
IntentRouter.classify(query)      ← 🆕 LLM 零样本分类（chat/search/filter/compare）
  │                                  chat → 短路，LLM 直接回答
  │                                  filter → 提取 metadata 过滤条件
  │                                  compare → 拆分子查询
  ▼
QueryProcessor                    ← jieba 分词 + 关键词提取 + 停用词过滤
  │
  ├── Dense Retrieval             ← Embedding API → ChromaDB 向量相似度查询（top 20）
  │
  └── Sparse Retrieval            ← BM25 关键词查询 → Whoosh 倒排索引（top 20）
  │
  ▼
RRF Fusion                        ← k=60，合并排序 → top 10
  │
  ▼
Reranker（可选，默认关闭）         ← Cross-Encoder 或 LLM 精排 → top 5
  │
  ▼
ResponseBuilder                   ← 格式化结果 + citations + MCP 协议包装

query → _pre_check(query)         ← 规则层（< 1ms）
  ├── 命中（问候/元问题）         → chat, confidence=0.99
  └── 未命中                      → _llm_classify(query)  ← LLM 零样本（~200ms）
                                       ├── 成功              → 返回意图 + 置信度
                                       └── 失败              → fallback search

{
  "name": "query_knowledge_hub",
  "parameters": {
    "query":       { "type": "string",  "required": true },
    "top_k":       { "type": "integer", "default": 5, "minimum": 1, "maximum": 20 },
    "collection":  { "type": "string",  "default": "default" }
  }
}

MCP Client 调用 query_knowledge_hub
  │
  ▼
1. 参数校验
   └── query 不能为空，top_k 限制在 1-20，collection 默认 "default"

2. 懒加载组件初始化（首次查询，后续缓存命中跳过）
   └── asyncio.to_thread(_ensure_initialized, collection)
       ├── EmbeddingFactory.create(settings)        ← 永久缓存，不重建
       ├── create_core_reranker(settings)           ← 永久缓存，不重建
       ├── VectorStoreFactory.create(collection)    ← 每次重建（读写入进程数据）
       ├── create_dense_retriever(embedding, store)
       ├── BM25Indexer + create_sparse_retriever    ← 每次 reload Whoosh 索引
       ├── QueryProcessor                          ← jieba 分词 + 停用词过滤
       └── create_hybrid_search(dense, sparse, processor)

3. 🆕 意图路由（IntentRouter.classify）
   └── asyncio.to_thread(router.classify, query)
       ├── 规则预检（_pre_check）→ 问候/元问题 → chat，< 1ms
       ├── LLM 零样本分类（_llm_classify）→ 结构化 JSON 输出，~200ms
       └── 路由分支：
           ├── chat    → 短路，直接返回 build_chat_response()
           ├── filter  → 提取 metadata 过滤条件，注入 query 上下文
           ├── compare → 拆分子查询，拼接增强 query
           └── search  → 不做干预，走原有检索流程

4. 混合搜索（HybridSearch.search）
   └── asyncio.to_thread(_perform_search, query, top_k*2)
       ├── Dense Retrieval（并行）
       │   ├── QueryProcessor 分词 + 提取关键词
       │   ├── Embedding API 编码 query → float vector
       │   └── ChromaDB 向量相似度查询 → top 20 结果
       ├── Sparse Retrieval（并行）
       │   ├── BM25Indexer._ensure_index_loaded() → 从磁盘加载 Whoosh 索引
       │   ├── BM25 词频匹配 → top 20 结果
       │   └── 通过 chunk_id 从 ChromaDB 回填完整 metadata
       └── RRF Fusion (k=60)
           └── 合并两个排名列表，按 1/(k+rank) 加权求和，取 top 10

5. 重排序（可选，config.enable_rerank 控制）
   └── asyncio.to_thread(_apply_rerank, query, results, top_k)
       ├── Cross-Encoder 逐对打分（query, chunk_text）→ 精排
       ├── LLM Rerank（备选，按 settings.rerank.provider 切换）
       └── 失败降级：取原始排序的前 top_k 结果

6. 响应构建（ResponseBuilder.build）
   └── 遍历 results → 格式化 Markdown 文本
       ├── 每个 chunk: 分数 + 来源文件 + chunk 索引 + 文本预览
       ├── citations 列表: {title, source_path, chunk_id, score}
       └── 图片 chunk → ImageContent 块

7. Trace 收集
   └── trace.metadata["final_results"] = [每个结果的关键字段]
   └── TraceCollector().collect(trace) → 写入 logs/traces.jsonl

8. MCP 协议包装
   └── response.to_mcp_content() → 分离 TextContent + ImageContent
   └── types.CallToolResult(content=content_blocks, isError=False)

{
  "name": "list_collections",
  "parameters": {}
}

MCP Client 调用 list_collections
  │
  ▼
1. 获取所有 collection
   └── VectorStoreFactory.create(settings) → 获取 ChromaDB PersistentClient
   └── client.list_collections() → [(name, count), ...]

2. 收集每个 collection 的统计
   ├── collection.count() → chunk 总数
   └── 查询 BM25 索引目录 data/db/bm25/{name}/ → 是否有索引

3. 格式化响应
   └── Markdown 表格: | Name | Chunks | BM25 Index |
   └── types.CallToolResult(content=[TextContent])

{
  "name": "get_document_summary",
  "parameters": {
    "source_hash": { "type": "string", "required": true },
    "collection":  { "type": "string", "default": "default" }
  }
}

MCP Client 调用 get_document_summary
  │
  ▼
1. 参数解析
   └── source_hash: 文件 SHA256，来自 Ingestion Pipeline Stage 1

2. 查 ChromaDB
   └── collection.get(where={"doc_hash": source_hash})
       ├── 返回该文档所有 chunk（ids + documents + metadatas）
       └── 汇总统计: chunk_count, total_chars, avg_chunk_size

3. 查 Ingestion 历史
   └── SQLiteIntegrityChecker.get_record(file_hash)
       ├── source_path（原始文件路径）
       ├── processed_at（摄入时间）
       └── status（success / failed）

4. 查图片记录
   └── ImageStorage.list_images(doc_hash=source_hash)
       └── 返回图片列表: [{image_id, file_path, page_num}, ...]

5. 格式化响应
   └── Markdown: 文档来源 + 摄入时间 + chunk 统计 + 图片预览
   └── types.CallToolResult(content=[TextContent])

客户端发起查询
  →
  ① IntentRoute 完成 → 推送 {"stage": "routing", "intent": "search", "elapsed_ms": 200}
  ② Dense 检索完成  → 推送 {"stage": "dense", "hits": 20, "elapsed_ms": 350}
  ③ Sparse 检索完成 → 推送 {"stage": "sparse", "hits": 20, "elapsed_ms": 120}
  ④ RRF Fusion 完成 → 推送 {"stage": "fusion", "hits": 10, "elapsed_ms": 50}
  ⑤ Rerank 完成     → 推送 {"stage": "rerank", "hits": 5, "elapsed_ms": 300}
  ⑥ 最终结果流式返回，每个 chunk 附 citations

async def query_knowledge_hub_streaming_handler(query, top_k, collection):
    for stage_event in pipeline.stream_execute(query, top_k, collection):
        yield types.CallToolResult(
            content=[types.TextContent(type="text", text=stage_event.to_json())]
        )

RRF_score(chunk) = Σ (1 / (k + rank_i))
  其中 k = 60, rank_i = chunk 在第 i 个来源列表中的排名

示例：
  chunk_A 在 Dense 排名第 2，Sparse 排名第 5
  RRF_score = 1/(60+2) + 1/(60+5) = 0.0161 + 0.0154 = 0.0315