Kagamicho/cs_chatbot
0
1"""Distill raw qa_pairs JSONL into the curated KB layer (data/curated/).2 3Usage:4 uv run python scripts/distill_kb.py --source arekore # no LLM5 uv run python scripts/distill_kb.py --source agentforce # Gemini Flash6 uv run python scripts/distill_kb.py --source cs_low_voltage7 uv run python scripts/distill_kb.py --source cs_low_voltage --limit 20 # trial run8 9Re-runnable: a checkpoint file (data/curated/.checkpoint_<source>.json) maps10raw entry id -> result, so crashes / re-runs don't re-spend LLM calls.11Output: data/curated/<source>.jsonl + data/curated/_report_<source>_<date>.md12 13Notes:14- Error entries are NOT checkpointed and will be retried on re-run. Only15 terminal classifications (canonical / episode / noise) are cached. This16 ensures transient failures (rate limits, safety blocks, parse errors) are17 automatically retried on the next invocation.18- --limit N interacts with checkpoint: a repeated --limit N run will be served19 largely from the checkpoint for entries already processed, so it is safe to20 use for incremental trials.21- The dedup pass (dedupe_entries) is a no-op for clean sources with distinct22 canonical questions.23"""24import argparse25import difflib26import json27import logging28import sys29import time30from datetime import date31from pathlib import Path32 33ROOT = Path(__file__).resolve().parent.parent34sys.path.insert(0, str(ROOT))35 36from ingestion.curated_schema import CuratedEntry # noqa: E40237 38logger = logging.getLogger(__name__)39 40SOURCES = {41 "arekore": "data/sources/qa_pairs/arekore_2026-06-09.jsonl",42 "agentforce": "data/sources/qa_pairs/agentforce_2026-05-25.jsonl",43 "cs_low_voltage": "data/sources/qa_pairs/cs_low_voltage_2026-06-08.jsonl",44}45LOW_CONFIDENCE = 0.746SIMILARITY_THRESHOLD = 0.8547MAX_ANSWER_CHARS = 3000 # truncate long answers before sending to LLM48 49 50def parse_llm_json(text: str) -> dict:51 """Parse the model's JSON, tolerating markdown code fences."""52 t = text.strip()53 if t.startswith("```"):54 t = t.split("\n", 1)[1] if "\n" in t else t55 t = t.rsplit("```", 1)[0]56 t = t.strip()57 start, end = t.find("{"), t.rfind("}")58 if start == -1 or end == -1:59 raise ValueError(f"no JSON object in LLM output: {text[:120]!r}")60 return json.loads(t[start : end + 1])61 62 63def load_raw(path: Path) -> list[dict]:64 out = []65 for line in path.read_text(encoding="utf-8").splitlines():66 line = line.strip()67 if line:68 out.append(json.loads(line))69 return out70 71 72def migrate_arekore(raw_entries: list[dict], *, today: str) -> list[CuratedEntry]:73 """Arekore is already clean: schema migration only, no LLM."""74 out = []75 for i, rec in enumerate(raw_entries, 1):76 out.append(CuratedEntry(77 id=f"qa-cur-arekore-{i:04d}",78 canonical_question=rec["question"],79 question_variants=[],80 answer=rec["answer"],81 topic=(rec.get("tags") or ["その他"])[0],82 audience="customer",83 status="canonical",84 priority=float(rec.get("priority", 2.0)),85 source_refs=[rec["id"]],86 distilled_by="migration",87 distilled_at=today,88 ))89 return out90 91 92class Distiller:93 def __init__(self, *, llm, source: str, out_dir: Path, today: str):94 self._llm = llm95 self._source = source96 self._out_dir = out_dir97 self._today = today98 self._ckpt_path = out_dir / f".checkpoint_{source}.json"99 self._ckpt: dict[str, dict] = {}100 if self._ckpt_path.exists():101 self._ckpt = json.loads(self._ckpt_path.read_text(encoding="utf-8"))102 self._prompt_template = (103 ROOT / "config" / "prompts" / "distill_classify_ja.md"104 ).read_text(encoding="utf-8")105 106 def _save_ckpt(self) -> None:107 self._out_dir.mkdir(parents=True, exist_ok=True)108 self._ckpt_path.write_text(109 json.dumps(self._ckpt, ensure_ascii=False), encoding="utf-8"110 )111 112 def _classify(self, rec: dict) -> dict:113 # FIX 3: truncate long answers so the prompt stays within token budget114 answer = rec.get("answer", "")115 if len(answer) > MAX_ANSWER_CHARS:116 answer = answer[:MAX_ANSWER_CHARS] + "\n…（以下省略）"117 # FIX 2: use .replace() so raw { } in question/answer/tags never118 # collide with Python's str.format placeholder syntax119 prompt = (120 self._prompt_template121 .replace("{question}", rec.get("question", ""))122 .replace("{answer}", answer)123 .replace("{tags}", ", ".join(rec.get("tags") or []))124 )125 # Long rewritten answers + Gemini 2.5 thinking tokens can exceed the126 # gateway default (4096) and truncate the JSON mid-output — give the127 # distiller a larger budget than the chat path needs.128 resp = self._llm.generate(129 prompt=prompt, profile="default", max_output_tokens=16384130 )131 return parse_llm_json(resp.text)132 133 def run(self, raw_entries: list[dict]) -> tuple[list[CuratedEntry], list[dict]]:134 entries: list[CuratedEntry] = []135 report: list[dict] = []136 seq = 0137 for rec in raw_entries:138 rid = rec["id"]139 if rid in self._ckpt:140 result = self._ckpt[rid]141 else:142 try:143 result = self._classify(rec)144 # FIX 1: only cache terminal (successful) classifications;145 # errors are NOT written to the checkpoint so they are146 # retried on the next run (handles rate limits, safety147 # blocks, parse errors transparently).148 self._ckpt[rid] = result149 self._save_ckpt()150 except Exception as e: # LLM/parse errors → review by exception151 logger.warning("distill failed for %s: %s", rid, e)152 result = {"status": "error", "reason": str(e)[:200]}153 time.sleep(0.1) # gentle rate limiting154 row = {"raw_id": rid, "status": result.get("status", "error"),155 "reason": result.get("reason", ""),156 "confidence": result.get("confidence")}157 report.append(row)158 if result.get("status") != "canonical":159 continue160 # FIX 4: build with seq+1 first; only advance seq on success so161 # schema errors don't burn a sequence number162 try:163 entries.append(CuratedEntry(164 id=f"qa-cur-{self._source}-{seq + 1:04d}",165 canonical_question=result["canonical_question"],166 question_variants=result.get("question_variants") or [],167 answer=result["answer"],168 context_note=result.get("context_note"),169 topic=result.get("topic", "その他"),170 audience=result.get("audience", "customer"),171 status="canonical",172 priority=2.0,173 source_refs=[rid],174 # honor a per-entry distilled_by (set when entries were175 # produced by Claude subagents and merged into the176 # checkpoint); default to gemini-flash for the API path177 distilled_by=result.get("distilled_by", "gemini-flash"),178 distilled_at=self._today,179 ))180 seq += 1181 except Exception as e:182 row["status"] = "error"183 row["reason"] = f"schema: {e}"[:200]184 return entries, report185 186 187def dedupe_entries(entries: list[CuratedEntry]) -> list[CuratedEntry]:188 """Merge near-duplicate canonical questions within the same topic.189 190 Newest distilled_at wins; source_refs are unioned so the audit trail191 survives the merge.192 """193 kept: list[CuratedEntry] = []194 for e in sorted(entries, key=lambda x: x.distilled_at, reverse=True):195 dup = next(196 (k for k in kept if k.topic == e.topic and difflib.SequenceMatcher(197 None, k.canonical_question, e.canonical_question198 ).ratio() >= SIMILARITY_THRESHOLD),199 None,200 )201 if dup is None:202 kept.append(e)203 else:204 merged_refs = list(dict.fromkeys([*dup.source_refs, *e.source_refs]))205 kept[kept.index(dup)] = dup.model_copy(update={"source_refs": merged_refs})206 return kept207 208 209def write_outputs(210 *, out_dir: Path, source: str, today: str,211 entries: list[CuratedEntry], report: list[dict],212) -> None:213 out_dir.mkdir(parents=True, exist_ok=True)214 out_path = out_dir / f"{source}.jsonl"215 out_path.write_text(216 "\n".join(json.dumps(e.model_dump(), ensure_ascii=False) for e in entries),217 encoding="utf-8",218 )219 counts: dict[str, int] = {}220 for r in report:221 counts[r["status"]] = counts.get(r["status"], 0) + 1222 low_conf = [r for r in report223 if r["status"] == "canonical"224 and r.get("confidence") is not None and r["confidence"] < LOW_CONFIDENCE]225 dropped = [r for r in report if r["status"] in ("episode", "noise", "error")]226 lines = [227 f"# Distillation report — {source} — {today}", "",228 f"Input entries: {len(report)}",229 *(f"- {k}: {v}" for k, v in sorted(counts.items())),230 f"\nKept after dedup: {len(entries)}", "",231 f"## Low-confidence rewrites (< {LOW_CONFIDENCE}) — spot-check these",232 *(f"- {r['raw_id']} (conf={r['confidence']}): {r['reason']}" for r in low_conf),233 "", "## Dropped entries",234 *(f"- {r['raw_id']} [{r['status']}]: {r['reason']}" for r in dropped),235 "",236 ]237 (out_dir / f"_report_{source}_{today}.md").write_text(238 "\n".join(lines), encoding="utf-8"239 )240 logger.info("wrote %d entries to %s", len(entries), out_path)241 242 243def main() -> None:244 logging.basicConfig(level=logging.INFO)245 parser = argparse.ArgumentParser()246 parser.add_argument("--source", required=True, choices=sorted(SOURCES))247 parser.add_argument("--limit", type=int, default=None,248 help="process only the first N raw entries (trial run)")249 args = parser.parse_args()250 251 today = date.today().isoformat()252 out_dir = ROOT / "data" / "curated"253 raw = load_raw(ROOT / SOURCES[args.source])254 if args.limit:255 raw = raw[: args.limit]256 257 if args.source == "arekore":258 entries = migrate_arekore(raw, today=today)259 report = [{"raw_id": r["id"], "status": "canonical",260 "reason": "migration", "confidence": 1.0} for r in raw]261 else:262 from app.core.llm_gateway import GeminiGateway263 from config.settings import settings264 llm = GeminiGateway(api_key=settings.gemini_api_key)265 distiller = Distiller(llm=llm, source=args.source, out_dir=out_dir, today=today)266 entries, report = distiller.run(raw)267 268 entries = dedupe_entries(entries)269 write_outputs(out_dir=out_dir, source=args.source, today=today,270 entries=entries, report=report)271 272 273if __name__ == "__main__":274 main()275