RAG升级合并到develop分支

.gitignore

@@ -202,10 +202,20 @@ Agent/knowledge_base/unfinish_source/
 skills/
 
 # test
-tests/
+tests/*
+!tests/test_multimodal_*.py
 
 # study
 study/
 
-# docs
-docs/
+# local-only documents
+/docs/
+!/docs/
+!docs/难题.md
+Agent/knowledge_base/rag/docs/
+Agent/knowledge_base/rag/output/reports/
+Agent/knowledge_base/multimodal_assets/
+Agent/knowledge_base/multimodal_indexes/
+Agent/knowledge_base/multimodal_runtime/
+Agent/knowledge_base/multimodal_benchmarks/
+app/DESIGN.md

AGENTS.md

@@ -82,14 +82,18 @@
 - `create_app()` 会先执行 `app/db.py` 中的 `check_database_readiness()`，确认数据库和关键表已就绪，然后再注册蓝图。
 - 当前实际注册的蓝图有 6 个：`auth`、`chat`、`files`、`agent`、`main`、`admin`。
 - Web 进程只负责登录态校验、短请求、analysis job 入队和 SSE 推送；Agent/RAG/MCP 长任务不在 Web 进程内执行，而是由独立 worker 进程处理。
-- 后台 worker 入口是 `python -m app.agent.worker`；worker 启动流程是：数据库就绪检查 -> 初始化 LLM -> 检查 RAG 可用性 -> 按 `JOB_WORKERS` 启动多个 slot。
-- 每个 worker slot 会独占一组 MCP server process、一个通过 `MultiServerMCPClient.session("causal")` 打开的持久 `ClientSession`、一组由 `load_mcp_tools(session)` 生成的 LangChain tools，以及一个编译好的 Agent graph；真实执行单元是 slot，不是 Flask 请求线程。旧 `open_mcp_session()` / 手写 `list_tools()` 包装仅保留作历史兼容入口。
-- 父图当前只暴露 `mcp`、`rag` 两个工具阶段节点：`mcp` 子图内部执行 `mcp_planner -> mcp_tool_node -> mcp_result_parser`，`rag` 子图内部执行 `rag_question_planner -> rag_tool_node -> rag_result_parser`；worker 事件流仍沿用旧 `astream(stream_mode="updates")` 适配，不要求本轮输出 tool-level SSE 事件。
+- 后台 worker 入口是 `python -m app.agent.worker`；worker 启动流程是：数据库就绪检查 -> 初始化主 LLM -> 从多模态 active pointer 严格创建一次进程级 `RagRuntime/RagService`（失败则绑定不可用 Service）-> 按 `JOB_WORKERS` 启动多个 slot。
+- 每个 worker slot 会独占一组 MCP server process、一个通过 `MultiServerMCPClient.session("causal")` 打开的持久 `ClientSession`、一组由 `load_mcp_tools(session)` 生成的 LangChain tools，以及一个直接绑定进程级 RagService 的 Agent graph；所有 slot 共享同一个多模态 RagService/Runtime。真实执行单元是 slot，不是 Flask 请求线程。
+- 父图保留 `mcp`、`rag` 两个业务阶段：`mcp` 仍是 `mcp_planner -> mcp_tool_node -> mcp_result_parser` 子图；`rag` 已收缩为普通节点，内部直接执行“生成问题 -> 调用 RagService -> 写入 knowledge_base_result”。worker 只输出 `rag_result`，不再输出 RAG tool-call 事件。
 - 配置统一由 `config/settings.py` 从系统环境变量读取；若项目根目录存在 `.env`，会先通过 `python-dotenv` 加载到环境变量。
 - 前端当前仍是 Flask 静态资源方案，不是 Node/Vite/React 工程；关键文件是：
   - `app/static/chat.html`
   - `app/static/css/style.css`
   - `app/static/js/script.js`
+  - `app/static/rag_eval.html`
+  - `app/static/css/rag_eval.css`
+  - `app/static/js/rag_eval.js`
+- RAG 评测控制台页面可通过 `/rag_eval` 或 `/rag-eval` 访问；静态文件仍位于 `app/static/rag_eval.html`。
 - `Database/database_init.py` 只负责加载环境变量、确保数据库存在并检查连接；业务表结构维护入口是 Alembic，而不是这个脚本。
 - Alembic 迁移目录由 `alembic.ini` 指向 `Database/migrations`；业务 schema 变更应以迁移脚本为准。
 - 数据库生产化升级前应先执行 `Database/audit_before_db_upgrade.py`；它是只读审计，不会修改数据，重点检查孤立消息、孤立附件、非法附件类型和分区状态。
@@ -109,7 +113,27 @@
 - Docker 是当前首选开发方式；`docker-compose.replica.yml` 中 `app` 和 `worker` 都会挂载以下知识库目录：
   - `Agent/knowledge_base/models`
   - `Agent/knowledge_base/db`
-- RAG 启动期只检查知识库目录是否可用，不会在启动时完整加载向量库；若 `Agent/knowledge_base/db` 不存在，worker 会记录 warning，并以“无知识库模式”继续运行。
+- 默认生产 RAG 使用 `RagRuntime -> RagService -> rag 普通节点` 链路；Runtime 默认从多模态 active pointer 解析不可变 Chroma collection，并校验 pointer、manifest 和运行时 embedding 指纹；失败时绑定不可用 Service，绝不回退到 PubMedQA。`rag_enrichment_search` 仅保留为兼容工具入口，不参与父图默认执行。
+- `query_rag.py` 的 dense + BM25s + rerank + answer 流程继续作为默认检索实现，并已兼容 `document_id`、`page_number`、`asset_uri`、`modality` 和 `content_kind` 等多模态 metadata。PubMedQA 构建与专用评测入口暂作为医疗兼容代码保留，后续分阶段清理。
+- 多模态公共知识库维护模块位于 `Agent/knowledge_base/multimodal/`，其 assets、暂存索引、active pointer 和 OmniDocBench 下载资料分别使用独立目录，严禁写入或清理 `Agent/knowledge_base/db/` 与 PubMedQA collection。PDF 当前默认 Docling；manifest 必须保存 source、parser 原始产物、标准化单元与资源的 URI/内容哈希关联，发布门禁必须回读校验。WCode 仅可接收 `remote_samples.json` 中固定 Pearl 页或 OmniDocBench 精确文件路径，模型固定 `qwen/qwen3-vl-flash`、域名必须为 `wcode.net`，默认预算不超过 100 且 smoke 应显式限制；审计日志不得记录图片、提示词、响应正文或密钥。
+- 正式 PDF 摄取按物理页在独立子进程中运行 Docling + RapidOCR，并保存页级 checkpoint；失败续跑不得重复解析已完成页面。Chroma 必须分批写入独立 attempt 目录，成功后才能提交为版本的 `chroma/`。生产评测命中必须同时匹配文档、页码和 `expected_modality`；`run` 默认不发布，只有显式 `--publish` 才允许切换 active pointer。
+- `Agent/knowledge_base/build_knowledge.py` 当前支持 `--profile default` 和 `--profile medical`：
+  - `default` 从 `Agent/knowledge_base/source/` 读取 Pearl/因果资料，并使用本地 `bge-small-zh-v1.5`。
+  - `medical` 从 `rag_config.py` 的 `MEDICAL_KNOWLEDGE_BUILD_CONFIG["corpus_path"]` 读取当前 active 医疗语料；当前指向 PubMedQA processed corpus，embedding provider 由 `RAG_EMBEDDING_PROVIDER` 控制：`auto` 保持旧兼容行为（存在 `MEDICAL_EMBEDDING_API_KEY` 或 `KNOWLEDGE_BUILD_PROFILE=medical` 时使用 OpenAI-compatible API，否则使用本地模型），`openai_compatible` 强制使用 `MEDICAL_EMBEDDING_API_KEY`、`MEDICAL_EMBEDDING_BASE_URL`、`MEDICAL_EMBEDDING_MODEL`，`local` 强制使用 `RAG_LOCAL_EMBEDDING_MODEL_PATH` 或默认 `Agent/knowledge_base/models/bge-small-zh-v1.5`。
+  - 两个 profile 都写入原 `Agent/knowledge_base/db` 持久化目录；切换 profile 前如果要清空旧索引，必须先获得用户明确确认。
+- 旧医疗兼容 benchmark 是 PubMedQA labeled；其 processed corpus/eval 均为 1000 条，不再属于默认 RAG 测试链路。
+- 当前本地 `Agent/knowledge_base/db` 已替换为 PubMedQA 医疗知识库，医疗查询与 medical 构建默认 collection 为 `pubmedqa_clean`；`causal_agent_default` 也指向 PubMedQA 但存在重复 chunk，旧 RAGCare 向量库已备份到 `tmp/RAGCare`。
+- 当前医疗知识库应以 PubMedQA corpus 为准；若本地 `Agent/knowledge_base/db` 与 active benchmark 不一致，retrieval_eval 会通过向量库/benchmark mismatch 防护阻止误跑。
+- `build_knowledge.py` 的旧构建入口仍支持 `RAG_VECTOR_DB_DIR`、`RAG_COLLECTION_NAME` 等显式覆盖；默认查询 Runtime 不再读取这两个旧医疗路径变量，而是通过 `MULTIMODAL_INDEX_ROOT` 与 `MULTIMODAL_ACTIVE_INDEX_CONFIG` 定位已发布多模态索引。embedding provider 仍由 `RAG_EMBEDDING_PROVIDER` 与 `RAG_LOCAL_EMBEDDING_MODEL_PATH` 等现有配置控制。
+- 默认多模态 RAG 查询继续读取 `Agent/knowledge_base/rag/runtime/production_rag_config.json` 中的 dense、BM25s、rerank 和证据长度参数。
+- `build_knowledge.py` 默认拒绝向非空 Chroma collection 追加写入，并记录到 `Agent/knowledge_base/build_knowledge.log`；只有明确传 `--allow-append` 才允许追加，避免重复 chunk 污染默认库。
+- retrieval_eval 已增加向量库/benchmark mismatch 防护：若 active gold doc 前缀与当前向量库 doc 前缀不一致，会在检索前失败。
+- `run_rag_eval.py` 默认步骤是 `validate_datasets -> retrieval_eval -> ragas_eval -> trace_export -> summary`；`claim_eval` 已从默认链路和前端工作台调参入口屏蔽，坏例链路只统计 retrieval/Ragas 相关问题。
+- RAG 评测已移除 CLI 调参备份层；前端不再展示 CLI 等价字段，后端不再提供 `GET /api/rag_eval/cli-params`，也不再接受 `cli_overrides`。
+- 当前 `pubmedqa_pipeline` 默认跑前 30 条，样本数通过前端动态调整；Ragas generation 默认使用 6 个 context、单 context 1600 字符、回答 1100 字符、PubMedQA prompt v6；100 条 `pubmedqa_eval100` profile 默认使用 `ragas_max_workers=4`、`ragas_max_retries=3`、`ragas_max_wait=20` 控制 Ragas judge 并发和限流等待；Ragas 运行会先校验 `rag_eval_result.json` 的样本数、问题顺序、检索配置和向量库摘要，不一致则自动刷新 retrieval latest，避免不同样本规模的 cross-metric 误报；默认 `trace_export` 产物在前端按 Bad Case Traces / 坏例链路展示；前端取消会在 `retrieval_eval` 和 Ragas dataset 构建/refresh 的当前样本结束后停止，并用 `step_progress` 展示样本进度。
+- `ragas_eval.py` 和 `claim_eval.py` 会在导入 LangChain/Ragas 前用 `os.environ.setdefault("KMP_DUPLICATE_LIB_OK", "TRUE")` 做 Windows OpenMP 进程级兜底；不要写入 `.env`，命令行显式设置仅用于覆盖默认值。
+- Active benchmark 测试集使用通用 `benchmark_v2` schema，`gold_doc_ids` 是通用 doc-level 检索 gold；旧因果数据集仍可继续使用 `gold_chunk_ids`。
+- 当前 LLM 若不支持 `response_format` 结构化输出，`query_rag.py` 会退回普通 JSON answer 生成路径。
 
 
 ### 3.1 常用命令

Agent/causal_agent/edges.py

@@ -71,6 +71,9 @@ def mcp_router(state: CausalChatState) -> str:
     if isinstance(mcp_result, dict) and mcp_result.get("success") is True:
         logging.info("路由决策 -> MCP分析成功, 前往[RAG工具阶段]")
         return "rag"
+    if isinstance(mcp_result, dict) and mcp_result.get("success") is False:
+        logging.info("路由决策 -> MCP分析失败, 前往[普通问答]终止本轮")
+        return "normal_chat"
     logging.info("路由决策 -> MCP分析缺失不充足, 前往[Agent决策路由]")
     return "agent"
 

Agent/causal_agent/graph.py

@@ -2,8 +2,9 @@
 from .state import CausalChatState
 from . import nodes, edges
 from .graph_utils import bind_node
-from .tool_subgraphs import build_mcp_subgraph, build_rag_subgraph
+from .tool_subgraphs import build_mcp_subgraph
 from .fault_tolerance import (
+    degrade_rag_tool_result,
     recover_postprocess_to_report,
     recover_report,
     recover_terminal_message,
@@ -21,6 +22,14 @@
 from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
 
 
+def _mcp_parent_update(result: dict) -> dict:
+    """将 MCP 子图终态投影为父图所需的两个业务字段。"""
+    return {
+        "causal_analysis_result": result.get("causal_analysis_result"),
+        "tool_call_request": result.get("tool_call_request"),
+    }
+
+
 def _build_default_checkpointer():
     """Create the default MySQL checkpointer when configuration allows it."""
     try:
@@ -52,19 +61,24 @@ def _build_default_checkpointer():
 
 
 
-def build_graph(llm: "ChatOpenAI", mcp_tools: list, rag_tools: list, checkpointer=None):
+def build_graph(llm: "ChatOpenAI", mcp_tools: list, rag_service, checkpointer=None):
     """
     构建父图。
 
-    父图只表达业务阶段顺序，MCP/RAG 的 tool-calling 细节封装在各自子图内。
+    父图保留 MCP 工具子图；默认多模态 RAG 作为单一普通节点执行。
     """
     workflow = StateGraph(CausalChatState)
 
     agent_node_with_llm = bind_node(nodes.agent_node, llm=llm)
     fold_node_with_llm = bind_node(nodes.fold_node, llm=llm)
     preprocess_node_with_llm = bind_node(nodes.preprocess_node, llm=llm)
     mcp_subgraph = build_mcp_subgraph(llm=llm, mcp_tools=mcp_tools)
-    rag_subgraph = build_rag_subgraph(llm=llm, rag_tools=rag_tools)
+
+    async def run_mcp_subgraph(state: CausalChatState) -> dict:
+        """执行 MCP 子图，并仅回写父图需要的稳定业务输出。"""
+        result = await mcp_subgraph.ainvoke(state)
+        return _mcp_parent_update(result)
+    rag_node_with_resources = bind_node(nodes.rag_node, llm=llm, rag_service=rag_service)
     postprocess_node_with_llm = bind_node(nodes.postprocess_node, llm=llm)
     inquiry_answer_node_with_llm = bind_node(nodes.inquiry_answer_node, llm=llm)
     report_node_with_llm = bind_node(nodes.report_node, llm=llm)
@@ -91,8 +105,14 @@ def build_graph(llm: "ChatOpenAI", mcp_tools: list, rag_tools: list, checkpointe
         timeout=timeout(run_timeout=180, idle_timeout=60),
         error_handler=recover_to_agent,
     )
-    workflow.add_node("mcp", mcp_subgraph)
-    workflow.add_node("rag", rag_subgraph)
+    workflow.add_node("mcp", run_mcp_subgraph)
+    workflow.add_node(
+        "rag",
+        rag_node_with_resources,
+        retry_policy=tool_retry(max_attempts=2),
+        timeout=timeout(run_timeout=180, idle_timeout=60),
+        error_handler=degrade_rag_tool_result,
+    )
     workflow.add_node(
         "postprocess",
         postprocess_node_with_llm,
@@ -146,12 +166,13 @@ def build_graph(llm: "ChatOpenAI", mcp_tools: list, rag_tools: list, checkpointe
         edges.mcp_router, 
         {
             "rag": "rag", 
-            "agent": "agent"
+            "agent": "agent",
+            "normal_chat": "normal_chat"
         }
     )
     workflow.add_edge(
         "rag", 
-        "agent"
+        "postprocess"
     )
     workflow.add_conditional_edges(
         "postprocess", 
@@ -176,12 +197,9 @@ def build_graph(llm: "ChatOpenAI", mcp_tools: list, rag_tools: list, checkpointe
 
 
 
-def create_graph_from_tools(llm: "ChatOpenAI", mcp_tools: list):
-    """使用已加载的 MCP tools 构建父图，返回可直接执行的 compiled graph。"""
-    from Agent.tool_node.rag_tool_registry import build_rag_tools
-    # graph 层不再关心 MCP session。
-    rag_tools = build_rag_tools()
-    return build_graph(llm=llm, mcp_tools=mcp_tools, rag_tools=rag_tools)
+def create_graph_from_tools(llm: "ChatOpenAI", mcp_tools: list, rag_service):
+    """使用已加载的 MCP tools 和多模态 RagService 构建父图。"""
+    return build_graph(llm=llm, mcp_tools=mcp_tools, rag_service=rag_service)
 
 
 agent_graph = None

Agent/causal_agent/nodes.py

@@ -481,7 +481,6 @@ async def preprocess_node(state: CausalChatState, llm: ChatOpenAI) -> dict:
 from Agent.tool_node.mcp_tool_call_adapter import normalize_mcp_tool_call_message
 from Agent.tool_node.tool_message_adapter import (
     attach_tool_call_metadata,
-    latest_ai_tool_call_ids,
     latest_matching_tool_result,
     parse_tool_message_json,
 )
@@ -542,73 +541,12 @@ async def mcp_result_parser_node(state: CausalChatState) -> dict:
     }
 
 
-async def rag_question_planner_node(state: CausalChatState, llm: ChatOpenAI, rag_tools: list) -> dict:
-    """子节点：获取rag问题"""
-    logging.info("正在启动 RAG 问题生成任务...")
-
-    max_questions = 3
-    rag_questions = await get_rag_questions(state, llm, max_questions=max_questions)
-    tool_name = "rag_enrichment_search"
-    if rag_tools:
-        tool_name = getattr(rag_tools[0], "name", tool_name)
-
-    ai_message = AIMessage(
-        content="",
-        tool_calls=[
-            {
-                "name": tool_name,
-                "args": {
-                    "questions": rag_questions,
-                    "max_results": 5,
-                },
-                "id": "rag_enrichment_search_1",
-            }
-        ],
-    )
-    return {"messages": [ai_message]}
-
-
-async def rag_result_parser_node(state: CausalChatState) -> dict:
-    """子节点：获取rag返回内容，注入state当中"""
-    messages = state.get("messages", [])
-    latest_tool_message, latest_tool_call = latest_matching_tool_result(messages)
-    if latest_tool_message is None:
-        if latest_ai_tool_call_ids(messages):
-            return {
-                "knowledge_base_result": {
-                    "success": False,
-                    "summary": "知识库增强暂不可用，报告将仅基于因果分析结果生成。",
-                    "questions": [],
-                    "evidence_count": 0,
-                    "error": "No RAG tool result was produced.",
-                }
-            }
-        existing_result = state.get("knowledge_base_result")
-        if isinstance(existing_result, dict):
-            return {"knowledge_base_result": existing_result}
-        return {
-            "knowledge_base_result": {
-                "success": False,
-                "summary": "知识库增强暂不可用，报告将仅基于因果分析结果生成。",
-                "questions": [],
-                "evidence_count": 0,
-                "error": "No RAG tool result was produced.",
-            }
-        }
-
-    parsed = parse_tool_message_json(latest_tool_message)
-    if not parsed.get("success"):
-        parsed.setdefault("summary", "知识库增强暂不可用，报告将仅基于因果分析结果生成。")
-        parsed.setdefault("questions", [])
-        parsed.setdefault("evidence_count", 0)
-    result = attach_tool_call_metadata(
-        parsed,
-        latest_tool_message,
-        latest_tool_call,
-    )
-    return {
-        "knowledge_base_result": result,
-    }
+async def rag_node(state: CausalChatState, llm: ChatOpenAI, rag_service: Any) -> dict:
+    """生成多模态检索问题，调用共享 RagService 并写回结构化结果。"""
+    logging.info("正在执行默认多模态 RAG 增强...")
+    questions = await get_rag_questions(state, llm, max_questions=3)
+    result = await rag_query_task(questions, rag_service)
+    return {"knowledge_base_result": result}
 
 
 # 环路检测模块

Agent/causal_agent/tool_subgraphs.py

@@ -6,9 +6,7 @@
 
 from Agent.causal_agent import nodes
 from Agent.causal_agent.fault_tolerance import (
-    degrade_rag_tool_result,
     recover_mcp_tool_failure,
-    short_retry,
     timeout,
     tool_retry,
 )
@@ -34,29 +32,3 @@ def build_mcp_subgraph(llm, mcp_tools):
     graph.add_edge("mcp_tool_node", "mcp_result_parser")
     graph.add_edge("mcp_result_parser", END)
     return graph.compile(name="mcp")
-
-
-def build_rag_subgraph(llm, rag_tools):
-    """Build the RAG enrichment subgraph used as one parent-graph stage."""
-    graph = StateGraph(CausalChatState)
-    graph.add_node(
-        "rag_question_planner",
-        bind_node(nodes.rag_question_planner_node, llm=llm, rag_tools=rag_tools),
-        retry_policy=short_retry(max_attempts=2),
-        timeout=timeout(run_timeout=60, idle_timeout=30),
-        error_handler=degrade_rag_tool_result,
-    )
-    graph.add_node(
-        "rag_tool_node",
-        ToolNode(rag_tools),
-        retry_policy=tool_retry(max_attempts=2),
-        timeout=timeout(run_timeout=120, idle_timeout=45),
-        error_handler=degrade_rag_tool_result,
-    )
-    graph.add_node("rag_result_parser", nodes.rag_result_parser_node)
-    graph.set_entry_point("rag_question_planner")
-
-    graph.add_edge("rag_question_planner", "rag_tool_node")
-    graph.add_edge("rag_tool_node", "rag_result_parser")
-    graph.add_edge("rag_result_parser", END)
-    return graph.compile(name="rag")