新增 chat_prompts_rag检模块

applications/prompts/chat_prompts.py

@@ -1,160 +1,202 @@
 from typing import List, Dict, Optional, Any
 
 
-def map_prompt(question, format_context):
-    prompt = f"""
-【任务说明】：针对下方单个文档片段，提炼与“问题”直接相关的‘可被引用的事实要点’
-【问题】： {question}
-【输入】：{format_context}
-【输出】：输出只输出 JSON，不需要多余问题，输出结果如下：
-    {{
-        "id": "id", # 返回输入 text 中的 DOC id
-        "claims": [ 
-            {{"point": "事实要点1（尽量原文转述/精准改写）"}},
-            {{"point": "事实要点2}},
-        ],
-        "conflicts_or_limits": ["该片段的限制/含糊点（如时间、定义口径、版本号等）"]
-    }}
-    """
+def map_prompt(question: str, format_context: str) -> str:
+    prompt = (
+        f"【任务说明】：针对下方单个文档片段，提炼与“问题”直接相关的‘可被引用的事实要点’\n"
+        f"【问题】： {question}\n"
+        f"【输入】：{format_context}\n"
+        "【输出】：只输出 JSON，不要多余文本，格式如下：\n"
+        "{\n"
+        '  "id": "id",\n'
+        '  "claims": [\n'
+        '    {"point": "事实要点1（尽量原文转述/精准改写）"},\n'
+        '    {"point": "事实要点2"}\n'
+        '  ],\n'
+        '  "conflicts_or_limits": ["该片段的限制/含糊点（如时间、定义口径、版本号等）"]\n'
+        "}"
+    )
     return prompt
 
 
-def reduce_prompt(question, mapped_results_json_list):
-    prompt = f"""
-【任务】：合并多份 Map 结果，完成一下三点：
-    "1) 去重并合并同义要点；
-    "2) 标注并归纳冲突点；
-    "3) 输出最终回答（含引用）。
-【问题】：{question}
-【Map 结果】:{mapped_results_json_list}
-【输出（Markdown）】:
-    - 简要结论（2-4句）
-    - 关键要点（每点附主要引用，如 [C12] [C4], 用 map 结果中的 id来表示）
-    - 证据信息不一致（如有）：列出冲突内容、涉及的 doc_id、可能原因
-    - 信息缺口（如有）
-"""
+def reduce_prompt(question: str, mapped_results_json_list: Optional[str]) -> str:
+    prompt = (
+        "【任务】：合并多份 Map 结果，完成以下三点：\n"
+        "1) 去重并合并同义要点；\n"
+        "2) 标注并归纳冲突点；\n"
+        "3) 输出最终回答（含引用）。\n"
+        f"【问题】：{question}\n"
+        f"【Map 结果】：{mapped_results_json_list or '[]'}\n"
+        "【输出（Markdown）】:\n"
+        "- 简要结论（2-4句）\n"
+        "- 关键要点（每点附主要引用，如 [Chunk-12] [Chunk-4]，用 map 结果中的 id 来表示）\n"
+        "- 证据信息不一致（如有）：列出冲突内容、涉及的 doc_id、可能原因\n"
+        "- 信息缺口（如有）\n"
+    )
     return prompt
 
 
-def verify_prompt(question, draft, formatted_contexts):
-    prompt = f"""
-【任务】: 对“初稿答案”的关键断言逐条核验，严格限定仅使用上下文，请标注每条断言是否被证据‘支持/相矛盾/证据不足’，必要时修正结论
-【问题】
-    {question}
-【初稿答案】
-    {draft}
-【上下文】
-    {formatted_contexts}
-【输出（只输出 JSON）】
-{
-        "verdicts": [
-        {
-            "claim": "断言内容", "status": "supported|contradicted|insufficient", "citations": ["[Cid]"]}
-    ],
-    "final_answer": "（如需修正，请给出修正后的简明答案，并附引用）"
-}
-"""
+def verify_prompt(question: str, draft: str, formatted_contexts: str) -> str:
+    prompt = (
+        "【任务】: 对“初稿答案”的关键断言逐条核验，严格限定仅使用上下文；"
+        "请标注每条断言是否被证据‘支持/相矛盾/证据不足’，必要时修正结论。\n"
+        f"【问题】\n{question}\n"
+        f"【初稿答案】\n{draft}\n"
+        f"【上下文】\n{formatted_contexts}\n"
+        "【输出（只输出 JSON）】\n"
+        "{\n"
+        '  "verdicts": [\n'
+        '    {"claim": "断言内容", "status": "supported|contradicted|insufficient", "citations": ["[Chunk-{id}]"]}\n'
+        "  ],\n"
+        '  "final_answer": "（如需修正，请给出修正后的简明答案，并附引用）"\n'
+        "}"
+    )
+    return prompt
+
+
+def merge_prompt(question: str, rag_answer: str, formatted_contexts: str) -> str:
+    """
+    合并策略：
+    1) 先判定 RAG 初稿是否“足够/部分/不足”；
+    2) 若不足或部分，用通用大模型（允许常识/公开资料）补充‘外部要点’，并标注为 [EXT]；
+    3) 产出融合后的最终答案，优先采用带 [C{id}] 的已证据支撑结论；
+    4) 输出审计区：区分 RAG 引用与外部补充，列出信息缺口。
+    """
+    prompt = (
+        "【任务】：先判断 RAG 初稿是否足以回答问题；若不足或仅部分覆盖，则在“允许使用常识/公开资料”的前提下补充，并与上下文证据进行“无缝融合”，产出一份单一成稿（禁止把上下文与外部补充分栏或分章节呈现）。\n"
+        "【必须遵守】\n"
+        "- 严禁以“基于上下文/外部补充/对标/工具列表”等标题分隔来源内容；最终仅保留一份融合后的成稿。\n"
+        "- 句内引用：凡是有依据的关键信息，紧随其后附上来源标记；上下文证据用 [Chunk-{id}]，外部补充用 [EXT]，可选在括号中加来源类型（如[EXT-百科]、[EXT-官网]）。\n"
+        "- 证据优先级：优先采用【上下文】已被证实的结论；若与外部资料冲突，优先保留【上下文】并在句末以 [EXT-Conflict] 标注冲突存在。\n"
+        "- 去重与融合：对上下文与外部的相似要点进行合并表述，避免重复罗列；统一术语与口径。\n"
+        "- 若信息仍不足，明确列出“信息缺口”。\n"
+        "- 最终答案必须是可直接给用户的完整回应（可含小标题/列表），但不得按照来源划分结构。\n\n"
+        f"【问题】\n{question}\n\n"
+        f"【RAG 初稿】\n{rag_answer}\n\n"
+        f"【上下文】\n{formatted_contexts}\n\n"
+        # "【写作要求】\n"
+        # "- 开头 1–2 句给出结论或核心抓手；随后组织为自然段或要点清单（3–8 条）。\n"
+        # "- 每条关键结论或数据点后紧跟其来源标记；能不标就不写空标注。\n"
+        # "- 语气务实、可执行，避免流水账与堆砌名词。\n"
+        # "- 禁止出现“基于上下文…/外部补充…”等来源分栏表述。\n\n"
+        "【输出（只输出 JSON）】\n"
+        "{\n"
+        '  "status": "判断最终结果是否能回答用户问题，若能回答输出 1，若不能则输出 0",\n'
+        '  "reason": "判断的理由（例如：核心问题是否被覆盖、关键数据/定义是否缺失）",\n'
+        '  "final_answer_markdown": "融合后的最终答案，不要输出信息缺口（每一条信息需标注 [Chunk-{id}] 与 [EXT]）",\n'
+        '  "audit": {\n'
+        '    "rag_citations": ["[Chunk-12]", "[Chunk-4]"],\n'
+        '    "external_claims": [\n'
+        '      {"claim":"外部补充要点1","note":"来源类型/可信度（如：官网/百科/标准） [EXT]"},\n'
+        '      {"claim":"外部补充要点2","note":"… [EXT]"}\n'
+        '    ],\n'
+        '    "gaps": ["仍然缺失或无法确认的信息"]\n'
+        "  }\n"
+        "}\n"
+        "【合规自检（执行后自检，不输出给用户）】：若发现答案结构仍以来源分栏或出现“外部补充/基于上下文”等分栏标题，则立即重写为单一成稿。"
+    )
+
     return prompt
 
 
 def build_rag_prompt(
     question: str,
     contexts: List[Dict],
-    mode: str = "single",  # 可选：single | map | reduce | rerank | verify | map_reduce
+    mode: str = "single",  # single | map | reduce | rerank | verify | merge
     max_chars_per_chunk: int = 800,
-    draft_answer: Optional[str] = None,
-    mapped_results_json_list: Optional[str] = None,
+    draft_answer: Optional[str] = None,              # verify/merge 用：RAG 初稿或草稿
+    mapped_results: Optional[str] = None,  # reduce 用
 ) -> Dict[str, Any]:
     """
     生成 RAG 聚合阶段所需的提示词（Prompt）。
     返回值依据 mode 不同而不同：
-      - single:      {"system": str, "user": str}
-      - map:         {"system": str, "user_list": List[str]}   # 每个片段一条 Map 提示
-      - reduce:      {"system": str, "user": str}
-      - rerank:      {"system": str, "user": str}
-      - verify:      {"system": str, "user": str}
-      - map_reduce:  {"map": {...}, "reduce": {...}}           # 组合骨架
-    使用方式：
-      1) 单步聚合：把返回的 system/user 丢给 LLM 即可；
-      2) Map-Reduce：先用 map.user_list 逐条调用 LLM 得到 JSON，再把合并后的 JSON 列表给 reduce。
+      - single: {"system": str, "user": str}
+      - map:    {"system": str, "user_list": List[str]}   # 每个片段一条 Map 提示
+      - reduce: {"system": str, "user": str}
+      - rerank: {"system": str, "user": str}
+      - verify: {"system": str, "user": str}
+      - merge:  {"system": str, "user": str}
     """
 
-    # ——Tools
+    # —— 辅助函数 ——
     def _trim(text: str, limit: int) -> str:
-        text = text.strip().replace("\n", " ")
+        text = (text or "").strip().replace("\n", " ")
         return text if len(text) <= limit else text[: max(0, limit - 1)] + "…"
 
-    def _format_contexts(chunks: List[Dict]) -> str:
-        lines = [_format_each_chunk(i) for i in chunks]
-        return "\n".join(lines).strip()
-
     def _format_each_chunk(chunk: Dict) -> str:
-        bits = [f"DOC id={chunk['id']}"]
-        if chunk.get("score"):
-            bits.append(f"score={round(chunk['score'], 4)}")
-
+        # 兼容不同返回字段：content / text
+        content = chunk.get("content") or chunk.get("text") or ""
+        bits = [f"Chunk id={chunk.get('id', '')}"]
+        if "score" in chunk and chunk["score"] is not None:
+            try:
+                bits.append(f"score={round(float(chunk['score']), 4)}")
+            except Exception:
+                pass
+        if chunk.get("url"):
+            bits.append(f"url={chunk['url']}")
         prefix = "[" + " ".join(bits) + "]"
-        snippets = _trim(chunk["content"], max_chars_per_chunk)
-        item = f"{prefix}\n{snippets}\n"
-        return item
+        snippet = _trim(content, max_chars_per_chunk)
+        return f"{prefix}\n{snippet}\n"
+
+    def _format_contexts(chunks: List[Dict]) -> str:
+        return "\n".join(_format_each_chunk(c) for c in chunks).strip()
 
     # —— 统一 System 约束（全中文） ——
     system_text = (
         "你是一位“基于证据的助手”。你必须只使用我提供的【上下文】来回答：\n"
         "- 不得使用外部常识或臆测；\n"
         "- 若上下文不足，请明确输出“信息不足”，并指出缺失的信息类型；\n"
-        "- 对关键结论附 [C{id}] 形式的出处；\n"
+        "- 对关键结论附 [Chunk-{id}] 形式的出处；\n"
         "- 如存在冲突证据，请列出冲突并给出谨慎结论与采信依据；\n"
         "- 用中文回答，保持简洁、结构化。"
     )
 
     formatted_contexts = _format_contexts(contexts)
-    match mode:
-        case "single":
-            user_text = (
-                f"【问题】\n{question}\n\n"
-                f"【上下文（已按相关性排序）】\n{formatted_contexts}\n\n"
-                "【请按以下结构作答】\n"
-                "1) 简要结论（2-4句）\n"
-                "2) 关键要点（每点附1-2个引用，如 [C3]）\n"
-                "3) 证据信息不一致（如有）\n"
-                "4) 信息缺口（如有）"
-            )
-            return {"system": system_text, "user": user_text}
-
-        # —— Map 步骤：对每个片段单独提炼“可引用事实点” ——
-        case "map":
-            map_user_list = []
-            for context in contexts:
-                format_context = _format_each_chunk(context)
-                map_user_list.append(map_prompt(question, format_context))
-            return {"system": system_text, "user_list": map_user_list}
-
-        # 对 map 的结果进行聚合
-        case "reduce":
-            res = reduce_prompt(question, mapped_results_json_list)
-            return {"system": system_text, "user": res}
-
-        # —— 自重排（Rank-Then-Read 的“Rank”）：仅评分排序，不做总结 ——
-        case "rerank":
-            rerank_system = "你是一位严谨的重排器。请只输出 JSON。"
-            rerank_user = (
-                f"请比较下列候选段与问题“{question}”的相关性，仅打分并排序（不做总结）。\n"
-                "评分标准（由高到低）：直接回答性 > 主题一致性 > 细节重合度 > 时间匹配。\n\n"
-                f"【候选段】\n{formatted_contexts}\n\n"
-                "【只输出 JSON，格式如下（按 score 从高到低）】\n"
-                '[{"id":"DOC_ID","score":X.X}]'
-            )
-            return {"system": rerank_system, "user": rerank_user}
-
-        # —— 核验（Chain-of-Verification）：对初稿答案逐条校验并修正 ——
-        case "verify":
-            draft = draft_answer or "（此处为初稿答案）"
-            verify_user = verify_prompt(question, draft, formatted_contexts)
-            return {"system": system_text, "user": verify_user}
-
-        case _:
-            raise ValueError(f"不支持的模式：{mode}")
+
+    # —— 各模式分发 ——
+    if mode == "single":
+        user_text = (
+            f"【问题】\n{question}\n\n"
+            f"【上下文（已按相关性排序）】\n{formatted_contexts}\n\n"
+            "【请按以下结构作答】\n"
+            "1) 简要结论（2-4句）\n"
+            "2) 关键要点（每点附1-2个引用，如 [C3]）\n"
+            "3) 证据信息不一致（如有）\n"
+            "4) 信息缺口（如有）"
+        )
+        return {"system": system_text, "user": user_text}
+
+    if mode == "map":
+        map_user_list = []
+        for ctx in contexts:
+            format_context = _format_each_chunk(ctx)
+            map_user_list.append(map_prompt(question, format_context))
+        return {"system": system_text, "user_list": map_user_list}
+
+    if mode == "reduce":
+        return {"system": system_text, "user": reduce_prompt(question, mapped_results)}
+
+    if mode == "rerank":
+        rerank_system = "你是一位严谨的重排器。请只输出 JSON。"
+        rerank_user = (
+            f"请比较下列候选段与问题“{question}”的相关性，仅打分并排序（不做总结）。\n"
+            "评分标准（由高到低）：直接回答性 > 主题一致性 > 细节重合度 > 时间匹配。\n\n"
+            f"【候选段】\n{formatted_contexts}\n\n"
+            "【只输出 JSON，格式如下（按 score 从高到低）】\n"
+            '[{"id":"DOC_ID","score":X.X}]'
+        )
+        return {"system": rerank_system, "user": rerank_user}
+
+    if mode == "verify":
+        draft = draft_answer or "（此处为初稿答案）"
+        return {"system": system_text, "user": verify_prompt(question, draft, formatted_contexts)}
+
+    if mode == "merge":
+        if not draft_answer:
+            raise ValueError("merge 模式需要提供 draft_answer（即 RAG 初稿答案）。")
+        return {"system": system_text, "user": merge_prompt(question, draft_answer, formatted_contexts)}
+
+    raise ValueError(f"不支持的模式：{mode}")
 
 
 __all__ = ["build_rag_prompt"]