group_fqcd_jr/tools/seed_knowledge_r1r5_faq.py

"""补一条"r1到r5分别代表什么"的 FAQ 知识（幂等；解决该问法答不出来的问题）。

## 为什么需要它

用户实测：问「r1到r5分别代表什么」时客服不给答案（走兜底 + 转人工）。排查结论（2026-09-14）：

- 向量库**有**内容，检索也命中（top1 = 那一套"高频问答对"里的 `R1到R5风险等级是什么意思?`，
  相似度 **0.6621**）；
- 但客服的判定是"中置信（0.55–0.75）必须**领先次优 ≥0.07**"（
  `app/service/agent/implementations/customer_service.py:148-150`），而这一问的次优是
  0.6412 → **只领先 0.021** → 判"候选并列"，转人工。

根因是**同一主题有多个来源**：`高频问答对.txt` 的 R1–R5 条目、《个人投资者适当性管理指南》
第十一条（还切成好几块）、产品手册的 1.4 节——它们分数天然挤在一起，gap 永远拉不开。

## 这条为什么有效

实测（本机，qwen3-embedding）：

| 条目形态 | 「r1到r5分别代表什么」的 top1 | 结论 |
|---|---|---|
| 长条目（表格 + 详细说明，488+433 字，切两块） | 0.6622 | 向量被长正文稀释，仍不过门槛 |
| 短条目（272 字，标题与问法近似） | 0.7214 | gap 0.059，仍差一点 |
| **短条目 + 标题与问法逐字对齐**（本条的做法） | **0.7384**，次优 0.6623 → **gap 0.0761 ≥ 0.07** | ✅ **可答** |

也就是：**命中条目的标题/正文与客户问法越贴近、块越短，相似度越高**；高置信档（≥0.75）
不要求 gap，中置信档靠"领先次优"过关。内容侧这一条就能修好，不必改门槛策略。

## 用法

```bash
# 先看要做什么（默认只读探测）
python tools/seed_knowledge_r1r5_faq.py

# 真正上传（需要 API 在跑，且 Worker 在跑以便投向量）
python tools/seed_knowledge_r1r5_faq.py --apply
```

幂等：若已存在同 `source_file` 的**未过期**知识，直接跳过（要覆盖先加 `--force`，
它会先删旧的再传新的）。
"""

from __future__ import annotations

import argparse
import asyncio
import base64
import sys
import time
import uuid
from pathlib import Path

import httpx

if hasattr(sys.stdout, "reconfigure"):
    sys.stdout.reconfigure(errors="replace")

BASE_DEFAULT = "http://127.0.0.1:8000"
#: 知识正文放 `docs/`（与 `tools/seed_knowledge_demo.py` 读 `docs/43-…` 同一约定）：
#: `data/` 被 `.gitignore` 忽略，放那里换环境就取不到了。
PROJECT_ROOT = Path(__file__).resolve().parents[1]
SOURCE = PROJECT_ROOT / "docs" / "45-风险等级R1到R5问答（知识库入库版）.md"
FILENAME = "r1到r5分别代表什么.md"
PROBE_QUERY = "r1到r5分别代表什么"


def _login(client: httpx.Client, base: str, username: str, password: str) -> str:
    response = client.post(
        f"{base}/api/v1/auth/tokens", json={"username": username, "password": password}
    )
    if response.status_code != 200:
        raise SystemExit(f"登录失败：HTTP {response.status_code} {response.text[:160]}")
    return str(response.json()["data"]["access_token"])


def _headers(token: str) -> dict[str, str]:
    return {"Authorization": f"Bearer {token}", "Idempotency-Key": uuid.uuid4().hex}


def _existing_active(client: httpx.Client, base: str, token: str) -> list[dict[str, object]]:
    """列未过期知识，挑出同 source_file 的条目（用于幂等）。"""
    response = client.get(f"{base}/api/v1/knowledge/list", headers=_headers(token))
    body = response.json()
    items = body.get("items")
    if items is None:
        data = body.get("data")
        items = data.get("items") if isinstance(data, dict) else data
    if not isinstance(items, list):
        return []
    return [row for row in items if isinstance(row, dict) and row.get("source_file") == FILENAME]


def _probe_scores() -> None:
    """上传后核对检索评分（与客服判定同一口径）。"""
    from app.service.agent.bootstrap import get_knowledge_search_service

    from app.core.knowledge_contracts import ALL_TIERS

    service = get_knowledge_search_service()
    # 校验脚本按**全集档位**查：它要证明的是"这条知识进库了、能被检索到"，
    # 不是"访客能看到"，故不受档位隔离限制。
    outcome = asyncio.run(service.search(PROBE_QUERY, top_k=5, tiers=ALL_TIERS))
    if not outcome.hits:
        print(f"  ⚠️ 「{PROBE_QUERY}」仍然没有命中")
        return
    top = outcome.hits[0].score
    second = outcome.hits[1].score if len(outcome.hits) > 1 else 0.0
    gap = top - second
    verdict = "高置信直接答" if top >= 0.75 else (
        f"中置信 gap={gap:.4f} → " + ("可答" if gap >= 0.07 else "仍会转人工")
    )
    print(f"  检索：top1={top:.4f}（{outcome.hits[0].title[:30]}） 次优={second:.4f} → {verdict}")


def main() -> int:
    parser = argparse.ArgumentParser(description="补 R1–R5 的 FAQ 知识（幂等）")
    parser.add_argument("--base", default=BASE_DEFAULT, help="平台地址")
    parser.add_argument("--username", default="admin_t")
    parser.add_argument("--password", default="88888888")
    parser.add_argument("--apply", action="store_true", help="真正上传（默认只看要做什么）")
    parser.add_argument("--force", action="store_true", help="已存在时删旧的再传")
    args = parser.parse_args()

    if not SOURCE.exists():
        raise SystemExit(f"知识正文不存在：{SOURCE}")
    content = SOURCE.read_text(encoding="utf-8")
    print(f"知识正文：{SOURCE}（{len(content)} 字）")

    with httpx.Client(base_url=args.base, timeout=120) as client:
        token = _login(client, args.base, args.username, args.password)
        existing = _existing_active(client, args.base, token)
        print(f"库内同源未过期知识：{len(existing)} 条 {[r.get('knowledge_id') for r in existing]}")
        if existing and not args.force:
            print("已存在 → 跳过（要覆盖加 --force）。")
            _probe_scores()
            return 0
        if not args.apply:
            print("\n[dry-run] 未上传。加 --apply 真写。")
            return 0

        for row in existing:
            kid = row.get("knowledge_id")
            resp = client.delete(f"{args.base}/api/v1/knowledge/{kid}", headers=_headers(token))
            print(f"  删除旧条目 {kid}: HTTP {resp.status_code}")
        resp = client.post(
            f"{args.base}/api/v1/knowledge/upload",
            headers=_headers(token),
            json={
                "filename": FILENAME,
                "content_base64": base64.b64encode(content.encode("utf-8")).decode("ascii"),
                "knowledge_type": "faq",
            },
        )
        print(f"上传：HTTP {resp.status_code} {resp.text[:200]}")
        if resp.status_code != 201:
            return 1
        print("等 Worker 投向量 …")
        time.sleep(12)
        _probe_scores()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())