PlantAI-rag-new/sdk/app/services/privateqa.py

"""services/privateqa.py —— 私有资料问答服务（FR-PRIV-004）。

BFF 直连 DeepSeek 生成 300 字内带 [资料n] 引用的 markdown 短答：
- ask_private(question, *, top_k=8, min_score=0.30) -> dict
  检索 chunk 索引 → 空则 {hit:false, mode:"empty", answer:"", items:[]}（不调 LLM）
  → 先按 doc_id 聚合（doc 级 max score 排序取 top 文档）再取各 doc 内
    最相关块（每 doc ≤3，缺陷 #52c）
  → 上下文拼 `[资料n]《name》（相关度 x.xx）\nsnippet`（12k 预算截断）
  → complete_text(system 约束, temperature=0.2, max_tokens=800, timeout=40)
  → {hit:true, mode:"vector", answer, items:[...带 ref], elapsed_ms}

技术纪律：本模块只用 dict/stdlib；无生成器/无 SSE
（Cython 安全，断流崩溃回归）。
异常不吞：DocgenUnavailable / Exception / LLMError 原样上抛，由路由转 503/502 信封。
"""
from __future__ import annotations

import logging
import time
from typing import Any

from .docgen.client import search_chunks
from .docgen.llm import complete_text

logger = logging.getLogger(__name__)

# ---- 常量（任务书契约）----
PRIVATE_QA_LLM_TIMEOUT = 40.0
PRIVATE_QA_MAX_CTX_CHARS = 12000
PRIVATE_QA_PER_DOC_MAX = 3
PRIVATE_QA_MAX_TOKENS = 800

SYSTEM_PROMPT = (
    "你是一个只依赖客户上传资料回答的助手。只能依据资料回答，不得编造；"
    "每个关键论断标注 [资料n]；资料不足则答『资料中未找到相关信息』；"
    "300 字内简体中文 markdown。"
)

# BUG-PRIV-R1-03（TL 裁决）：system prompt 固定的自谦答语。LLM 返回匹配
# 任一即改判未命中（底噪越线时不展示低质块；不调默认阈值）。
SELF_DEPRECATING_ANSWERS = (
    "资料中未找到相关信息",
    "资料中没有相关信息",
    "资料中未找到相关的内容",
    "资料中没有相关的内容",
    "资料中找不到相关信息",
    "未找到相关信息",
    "没有找到相关信息",
)


def _is_self_deprecating(text: str) -> bool:
    """确定性自谦答语判断（子串匹配）。"""
    src = (text or "").strip()
    if not src:
        return True  # 空答案视同未命中
    return any(phrase in src for phrase in SELF_DEPRECATING_ANSWERS)


def _safe_float(value: Any) -> bool:
    try:
        float(value)
        return True
    except (TypeError, ValueError):
        return False


def _pick_items(items: list[dict[str, Any]], top_k: int) -> list[dict[str, Any]]:
    """先按 doc_id 聚合再取块（缺陷 #52c）：单一文档不再占满全部名额。

    原实现按输入 score 序逐条取（每 doc ≤ PRIVATE_QA_PER_DOC_MAX 块）直到
    top_k——重复上传的作者页块（得分几乎相同）会占满 top-8、正文块进不了
    上下文。新语义：
    1. 按 doc_id 分组，doc 级得分 = 组内 max score；
    2. doc 级得分降序取前 top_k 个文档；
    3. 每选中文档取组内得分最高的 ≤ PRIVATE_QA_PER_DOC_MAX 个块，
       且组内近重复块（内容指纹相同，重复上传残留产物）只取 1 条——
       否则「同内容 200 条作者页块」仍会把该 doc 的 3 个名额占满、
       正文块永远进不了上下文；
    4. 全部选中块按 score 降序稳定排序、截断到 top_k 返回（_build_context
       的 [资料n] 编号与 items 顺序不变，回归语义保留）。
    """
    picked: list[dict[str, Any]] = []
    groups: dict[str, list[dict[str, Any]]] = {}
    doc_score: dict[str, float] = {}
    doc_order: list[str] = []
    for item in items:
        if not isinstance(item, dict):
            continue
        doc_id = str(item.get("doc_id") or "")
        if doc_id not in doc_order:
            doc_order.append(doc_id)
        groups.setdefault(doc_id, []).append(item)
        try:
            score = float(item.get("score"))
        except (TypeError, ValueError):
            score = 0.0
        if score > doc_score.get(doc_id, float("-inf")):
            doc_score[doc_id] = score
    # 先聚合后排序：得分相同时按输入首现顺序（稳定）
    top_docs = sorted(doc_order, key=lambda d: doc_score.get(d, 0.0), reverse=True)[:top_k]
    for doc_id in top_docs:
        group = sorted(groups.get(doc_id, []), key=_score_of, reverse=True)
        taken_fingerprints: set[str] = set()
        for item in group:
            fingerprint = _chunk_fingerprint(item)
            if fingerprint in taken_fingerprints:
                continue  # 组内近重复块（重复上传残留）不重复占名额
            taken_fingerprints.add(fingerprint)
            picked.append(dict(item))
            if len(taken_fingerprints) >= PRIVATE_QA_PER_DOC_MAX:
                break
    picked.sort(key=_score_of, reverse=True)
    return picked[:top_k]


def _chunk_fingerprint(item: dict[str, Any]) -> str:
    """组内近重复指纹：归一化（去空白）后的 snippet 前 48 字符。

    重复上传残留的块内容逐字相同（同内容多次入库切块结果一致）；
    正常相邻块（重叠 1 段）不会整段相同，不受影响。
    """
    import re

    snippet = str(item.get("snippet") or "")
    normalized = re.sub(r"\s+", "", snippet)
    return normalized[:48]


def _score_of(item: dict[str, Any]) -> float:
    """item 得分（无/非法 → 0.0）。"""
    try:
        return float(item.get("score"))
    except (TypeError, ValueError):
        return 0.0


def _build_context(picked: list[dict[str, Any]]) -> tuple[str, list[dict[str, Any]]]:
    """拼 LLM 上下文（12k 字符预算截断）并给条目补 ref 编号。"""
    parts: list[str] = []
    out_items: list[dict[str, Any]] = []
    budget = PRIVATE_QA_MAX_CTX_CHARS
    for n, item in enumerate(picked, 1):
        name = str(item.get("doc_name") or item.get("name") or "")
        score_s = f"{float(item.get('score')):.2f}" if _safe_float(item.get("score")) else "0.00"
        snippet = str(item.get("snippet") or "")
        part = f"[资料{n}]《{name}》（相关度 {score_s}）\n{snippet}"
        if len(part) > budget:
            part = part[:budget] if budget > 0 else ""
        if not part:
            continue
        budget -= len(part)
        parts.append(part)
        entry = dict(item)
        entry["ref"] = n
        out_items.append(entry)
    return "\n".join(parts), out_items


async def ask_private(
    question: str,
    *,
    top_k: int = 8,
    min_score: float = 0.30,
) -> dict[str, Any]:
    """私有资料问答：检索 → 去重截断 → 拼上下文 → DeepSeek 短答。

    返回 {hit, mode, answer, items, elapsed_ms}；未命中不调 LLM；
    异常（DocgenUnavailable / Exception / LLMError）原样上抛。
    """
    started = time.perf_counter()
    # 缺陷 #52c：先聚合再取块需要更宽的候选池——只取 top_k 条时重复文档
    # 的块会把池占满、聚合阶段无块可选。取 top_k×4（vstore 上限 20 clamp），
    # 最终 _pick_items 仍收敛到 top_k 条，响应量与 [资料n] 编号语义不变。
    pool_k = min(max(top_k * 4, top_k), 20)
    payload = await search_chunks(question, top_k=pool_k, min_score=min_score, timeout=12.0)
    items = [i for i in payload.get("items") or [] if isinstance(i, dict)]
    if not items:
        return {"hit": False, "mode": "empty", "answer": "", "items": []}

    picked = _pick_items(items, top_k)
    context, out_items = _build_context(picked)
    if not context:
        return {"hit": False, "mode": "empty", "answer": "", "items": []}

    # BUG-PRIV-R2-01（TL 裁决）：user 消息必须含问题——只传检索上下文时
    # LLM 不知用户问什么只能自谦，被 R1-03 改判误杀真命中。
    # 形态：用户问题 + 参考资料（[资料n] 编号顺序与 items 保持一致）。
    user_message = f"用户问题：{question}\n\n参考资料：\n{context}"
    answer = await complete_text(
        SYSTEM_PROMPT,
        user_message,
        temperature=0.2,
        max_tokens=PRIVATE_QA_MAX_TOKENS,
        timeout=PRIVATE_QA_LLM_TIMEOUT,
    )
    # BUG-PRIV-R1-03（TL 裁决）：LLM 自谦答语 → 改判未命中。
    # vstore 检索先行行为不变（search-chunks 仍返回 items 供内部判断），
    # 只是最终响应改判，前端零痕迹。
    if _is_self_deprecating(answer):
        return {"hit": False, "mode": "empty", "answer": "", "items": []}
    elapsed_ms = int((time.perf_counter() - started) * 1000)
    return {
        "hit": True,
        "mode": "vector",
        "answer": answer,
        "items": out_items,
        "elapsed_ms": elapsed_ms,
    }