diff --git a/continuation.md b/continuation.md index d87cb9e..a0b4ecf 100644 --- a/continuation.md +++ b/continuation.md @@ -19,17 +19,19 @@ - B4 `vectorstore/client.py`: 检索读取 payload.sources - A1 `scheduler/reporter.py`: 日报 HTML 多源时显示「财联社 / 新浪 📰」 - A2 `a_share_cli/main.py` + `mcp_server/tools.py`: 检索展示「cecn / cscn [多源]」,MCP 返回 sources 数组 -- 存量回填 `scripts/backfill_qdrant_sources.py`(新):以指纹库 source_ids 为权威源,scroll+批量 upsert 回写 3.3 万条 payload - - ⚠️ 坑:本地文件模式 Qdrant 的 `set_payload` 逐点极慢(0.65s/点,33k 条约 6h),改为与 ingest 相同的 scroll+upsert 快速路径;且本地模式单进程锁,必须避开调度窗口 +- 存量回填 `scripts/backfill_qdrant_sources.py`(新):以指纹库 source_ids 为权威源回填 Qdrant payload + - ⚠️ 坑:本地文件模式 Qdrant 写入逐点 ≈0.4~0.65s,**全量 3.3 万条回填需 4~6 小时**且占用单进程锁,不可行 + - 改为**定向回填**:只有多源 point 需要 sources(单源点展示层兜底 source_id),仅回填指纹库 227 条多源 → **1m57s 完成 226 条**(1 条不在集合) + - 执行时机:避开定时调度窗口(22:00 pipeline 结束后 22:34 执行) **验证**: - 新增 `tests/test_multisource.py` 10 个(validator/`_result_to_point` 三优先级/文本构造透传/回填加载) - 全量 **276 passed**(3 个 crawler 基线失败与本次无关);ruff 干净(reporter 5 个 N806/SIM115 为既有问题未动) - 今日 M5/M6 重跑:313 条全含 sources(11 条多源);CLI 检索显示「来源: cecn / cscn [多源]」✓,MCP 返回 `sources: ['cecn','cscn']` ✓ -- 存量回填:待执行(等 22:00 pipeline 结束后运行,避开 Qdrant 单进程锁) +- 存量回填:226 条多源 point payload 已补 sources;CLI 检索「碧根果反倾销」显示「来源: cnstock / cscn [多源]」✓;日报 HTML 冒烟测试多源渲染「财联社 / 新浪财经 / 东方财富 📰」✓ **待办/遗留**: -- 回填脚本执行 + 回填后检索验证 +- 单源存量 point 无 sources 字段(展示层兜底 source_id,行为不变,如需统一可后续补) - P1-2(补跑 steps 含 cninfo)、P1-3(时区一致性),待用户决策 --- diff --git a/scripts/backfill_qdrant_sources.py b/scripts/backfill_qdrant_sources.py index 0dcd3ae..5ce7315 100644 --- a/scripts/backfill_qdrant_sources.py +++ b/scripts/backfill_qdrant_sources.py @@ -1,22 +1,18 @@ -"""一次性回填脚本:为 Qdrant 存量 point 的 payload 补充 sources 多源字段。 +"""一次性回填脚本:为 Qdrant 存量**多源** point 的 payload 补充 sources 字段。 背景: M5/M6 旧链路未把 M3 多源记录写入 EmbeddingResult / Qdrant payload, 导致知识库检索只能看到单源。新链路修复后,此脚本为存量数据补齐。 -数据来源(按优先级): - 1. 指纹库 data/dedup/fingerprints.sqlite3 的 source_ids 列(权威多源记录); - 2. source_ids 为 NULL 的旧行回退 [source_id]; - 3. 指纹库查不到的 point 回退 payload.source_id。 +范围(关键设计): + 本地文件模式 Qdrant 写入逐点成本 ≈0.4~0.65s,全量 3.3 万条回填需数小时, + 且本地模式为单进程锁,长时间占用会阻塞后续定时调度。 + 但**只有多源 point 需要 sources 字段**——单源 point 的展示层 + (CLI/MCP/日报)会在 sources 为空时兜底显示 source_id。 + 因此本脚本仅回填指纹库中 source_ids 长度 > 1 的多源记录,分钟级完成。 -实现: - 本地文件模式 Qdrant 的 set_payload 逐点极慢(≈0.65s/点,33k 条约 6 小时), - 故采用与正常 ingest 相同的快速路径: - scroll 分页取全部 point(含向量)→ 合并 sources → 批量 upsert 回写。 - -注意: - Qdrant 为本地文件模式,同一时刻仅允许一个进程打开; - 执行前请确认无 pipeline 子进程正在运行(避开调度窗口)。 +数据来源: + 指纹库 data/dedup/fingerprints.sqlite3 的 source_ids 列(权威多源记录)。 用法: uv run python -m scripts.backfill_qdrant_sources # 实际执行 @@ -29,15 +25,13 @@ import argparse import json import sqlite3 import sys +from collections import defaultdict from pathlib import Path from loguru import logger -from qdrant_client.http.models import PointStruct from vectorstore import DEFAULT_COLLECTION, make_qdrant_client - -_SCROLL_LIMIT = 1000 -_UPSERT_BATCH = 100 +from vectorstore.client import url_hash_to_uuid def _setup_logger(level: str) -> None: @@ -49,8 +43,8 @@ def _setup_logger(level: str) -> None: ) -def _load_fingerprint_sources(db_path: Path) -> dict[str, list[str]]: - """从指纹库读 url_hash -> sources(source_ids 为 NULL 时回退 [source_id])。""" +def _load_multi_source(db_path: Path) -> dict[str, list[str]]: + """从指纹库读多源记录: url_hash -> sources(仅 source_ids 长度 > 1)。""" conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row rows = conn.execute( @@ -60,47 +54,26 @@ def _load_fingerprint_sources(db_path: Path) -> dict[str, list[str]]: result: dict[str, list[str]] = {} for r in rows: - sources: list[str] = [] - if r["source_ids"]: - try: - val = json.loads(r["source_ids"]) - if isinstance(val, list): - sources = [s for s in val if s] - except (TypeError, ValueError): - sources = [] - if not sources: - sources = [r["source_id"]] if r["source_id"] else [] + if not r["source_ids"]: + continue + try: + val = json.loads(r["source_ids"]) + except (TypeError, ValueError): + continue + if not isinstance(val, list): + continue + sources = [s for s in val if s] + if len(sources) <= 1: + continue # 保证主源居首 if r["source_id"] and r["source_id"] in sources: sources = [r["source_id"], *[s for s in sources if s != r["source_id"]]] - if sources: - result[r["url_hash"]] = sources + result[r["url_hash"]] = sources return result -def _build_points( - records: list, - fp_sources: dict[str, list[str]], -) -> list[PointStruct]: - """把 scroll 记录转回 PointStruct,payload 补入 sources 字段。""" - points: list[PointStruct] = [] - for rec in records: - payload = dict(rec.payload or {}) - url_hash = payload.get("url_hash") or "" - source_id = payload.get("source_id") or "" - sources = fp_sources.get(url_hash) or ([source_id] if source_id else []) - if sources: - payload["sources"] = sources - points.append(PointStruct( - id=rec.id, - vector=rec.vector, - payload=payload, - )) - return points - - def main() -> int: - parser = argparse.ArgumentParser(description="Qdrant 存量 payload 回填 sources 多源字段") + parser = argparse.ArgumentParser(description="Qdrant 存量多源 payload 回填 sources 字段") parser.add_argument("--db", default="data/dedup/fingerprints.sqlite3", help="指纹库路径") parser.add_argument("--collection", default=DEFAULT_COLLECTION) @@ -110,56 +83,53 @@ def main() -> int: _setup_logger(args.log_level) - fp_sources = _load_fingerprint_sources(Path(args.db)) - logger.info("指纹库加载完成: {} 条,其中多源 {} 条", - len(fp_sources), sum(1 for v in fp_sources.values() if len(v) > 1)) - - client = make_qdrant_client() - - # scroll 分页 + (非 dry-run) 批量 upsert 回写 - total = 0 - multi_cnt = 0 - updated = 0 - offset = None - while True: - records, offset = client.scroll( - collection_name=args.collection, - limit=_SCROLL_LIMIT, - offset=offset, - with_payload=True, - with_vectors=True, - ) - if not records: - if offset is None: - break - continue - total += len(records) - multi_cnt += sum( - 1 for r in records - if len((r.payload or {}).get("sources") or []) > 1 - ) - if not args.dry_run: - points = _build_points(records, fp_sources) - for i in range(0, len(points), _UPSERT_BATCH): - client.upsert( - collection_name=args.collection, - points=points[i : i + _UPSERT_BATCH], - ) - updated += len(points) - if offset is None: - break - - logger.info("Qdrant 现存 point: {} 条(其中 payload 已含多源 {} 条)", total, multi_cnt) - if args.dry_run: - logger.info("--dry-run: 不执行写入") - client.close() + multi = _load_multi_source(Path(args.db)) + logger.info("指纹库多源记录: {} 条", len(multi)) + if not multi: + logger.info("无多源记录,退出") return 0 - logger.info("回填完成: 重新 upsert {} 条 point(payload 已补 sources)", updated) + # 按 sources 值分组(同值一次 set_payload) + groups: dict[str, list[str]] = defaultdict(list) + for url_hash, sources in multi.items(): + groups[json.dumps(sources, ensure_ascii=False)].append(url_hash) + + if args.dry_run: + logger.info("待回填: {} 组 / {} 条多源 point (--dry-run 不写入)", + len(groups), len(multi)) + return 0 + + client = make_qdrant_client() + updated = 0 + missing = 0 + for sources_json, url_hashes in groups.items(): + payload = {"sources": json.loads(sources_json)} + ids = [url_hash_to_uuid(h) for h in url_hashes] + try: + client.set_payload( + collection_name=args.collection, + payload=payload, + points=ids, + ) + updated += len(ids) + except KeyError: + # 部分 point 已不在集合中(如指纹库保留但向量库已清理),逐点重试 + for uid in ids: + try: + client.set_payload( + collection_name=args.collection, + payload=payload, + points=[uid], + ) + updated += 1 + except KeyError: + missing += 1 + logger.info("回填完成: 更新 {} 条多源 point,{} 条不存在于集合", + updated, missing) + client.close() return 0 if __name__ == "__main__": raise SystemExit(main()) - diff --git a/tests/test_multisource.py b/tests/test_multisource.py index a3ffed8..0da2ebc 100644 --- a/tests/test_multisource.py +++ b/tests/test_multisource.py @@ -166,7 +166,7 @@ def test_build_text_from_article_reads_sources(tmp_path: Path) -> None: # --------------------------------------------------------------------------- # -# 回填脚本 _load_fingerprint_sources +# 回填脚本 _load_multi_source # --------------------------------------------------------------------------- # def _make_fp_db(tmp_path: Path) -> Path: @@ -183,23 +183,30 @@ def _make_fp_db(tmp_path: Path) -> Path: conn.execute( "INSERT INTO fingerprints VALUES (?, ?, ?)", ("h2", "zqrb", None) ) + conn.execute( + "INSERT INTO fingerprints VALUES (?, ?, ?)", ("h3", "cls", "not-json") + ) + conn.execute( + "INSERT INTO fingerprints VALUES (?, ?, ?)", + ("h4", "cls", json.dumps(["cls"])), + ) conn.commit() conn.close() return db -def test_load_fingerprint_sources_parses_and_falls_back(tmp_path: Path) -> None: - from scripts.backfill_qdrant_sources import _load_fingerprint_sources +def test_load_multi_source_only_multi(tmp_path: Path) -> None: + """仅返回多源记录;NULL/损坏/单源均排除。""" + from scripts.backfill_qdrant_sources import _load_multi_source db = _make_fp_db(tmp_path) - res = _load_fingerprint_sources(db) - assert res["h1"] == ["cls", "sina"] - assert res["h2"] == ["zqrb"] # source_ids NULL → 回退 [主源] + res = _load_multi_source(db) + assert res == {"h1": ["cls", "sina"]} -def test_load_fingerprint_sources_main_first(tmp_path: Path) -> None: +def test_load_multi_source_main_first(tmp_path: Path) -> None: """source_ids 顺序异常时保证主源居首。""" - from scripts.backfill_qdrant_sources import _load_fingerprint_sources + from scripts.backfill_qdrant_sources import _load_multi_source db = tmp_path / "fp.sqlite3" conn = sqlite3.connect(db) @@ -213,5 +220,5 @@ def test_load_fingerprint_sources_main_first(tmp_path: Path) -> None: ) conn.commit() conn.close() - res = _load_fingerprint_sources(db) + res = _load_multi_source(db) assert res["h1"] == ["cls", "sina"]