Rifqi Hafizuddin
[NOTICKET] feat(knowledge_extraction): paid extraction stage + validate, diff, queue
ab5ea78 | """The four extraction branches. This is the only stage that costs money. | |
| **One call per TERM CLUSTER** β not per mention, not per chunk. That is the | |
| whole economic argument for clustering: 200 mentions of "PA" cost one call, not | |
| 200. It is also what makes conflict detection possible, since contradictory | |
| definitions can only be compared when they arrive together. | |
| Each branch returns `(entry, usage)`, with `None` for the entry when the | |
| response fails schema validation. A failed parse is not an exception: one bad | |
| response must not abort a corpus-scale run that has already paid for parsing. | |
| """ | |
| from __future__ import annotations | |
| from ...middlewares.logging import get_logger | |
| from ..models import ( | |
| BriefContext, | |
| CallUsage, | |
| Chunk, | |
| FormulaEntry, | |
| FormulaVariable, | |
| GlossaryEntry, | |
| Provenance, | |
| RuleCandidate, | |
| RuleEntry, | |
| TermCluster, | |
| ) | |
| from ..rank import top_k | |
| from ..settings import EVIDENCE_K | |
| from .base import evidence_block, load_prompt | |
| from .schemas import FormulaDraft, GlossaryDraft, RuleDraft, SummaryDraft, schema_for | |
| logger = get_logger("knowledge_extract") | |
| def _prov(draft_prov, doc_id: str, chunk_id: str | None = None) -> Provenance: | |
| return Provenance( | |
| doc_id=doc_id, | |
| span=draft_prov.span, | |
| page=draft_prov.page, | |
| section_no=draft_prov.section_no, | |
| chunk_id=chunk_id, | |
| ) | |
| def _evidence_for( | |
| cluster: TermCluster, chunks: list[Chunk], k: int, round_index: int | |
| ) -> tuple[list[Chunk], list[float]]: | |
| by_id = {c.chunk_id: c for c in chunks} | |
| ids = top_k(cluster, k=k, round_index=round_index) | |
| scores = cluster.evidence_scores[round_index * k : round_index * k + k] | |
| return [by_id[i] for i in ids if i in by_id], scores | |
| # ββ glossary ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def build_glossary_prompt( | |
| cluster: TermCluster, chunks: list[Chunk], k: int = EVIDENCE_K, round_index: int = 0 | |
| ) -> tuple[str, str]: | |
| evidence, scores = _evidence_for(cluster, chunks, k, round_index) | |
| user = ( | |
| f"CANDIDATE TERM: {cluster.canonical}\n" | |
| f"KNOWN VARIANTS: {', '.join(cluster.variants)}\n" | |
| f"MENTION COUNT: {cluster.mention_count}\n\n" | |
| + evidence_block(evidence, scores) | |
| ) | |
| return load_prompt("glossary"), user | |
| def extract_glossary( | |
| cluster: TermCluster, | |
| chunks: list[Chunk], | |
| extractor, | |
| doc_id: str, | |
| k: int = EVIDENCE_K, | |
| round_index: int = 0, | |
| ) -> tuple[GlossaryEntry | None, CallUsage]: | |
| system, user = build_glossary_prompt(cluster, chunks, k, round_index) | |
| result = extractor.complete( | |
| "glossary", system, user, schema_for("glossary"), "GlossaryEntry" | |
| ) | |
| try: | |
| draft = GlossaryDraft.model_validate(result.data) | |
| except Exception as exc: | |
| logger.warning( | |
| "glossary draft invalid", cluster=cluster.canonical, error=repr(exc) | |
| ) | |
| return None, result.usage | |
| evidence, _ = _evidence_for(cluster, chunks, k, round_index) | |
| entry = GlossaryEntry( | |
| term=draft.term, | |
| full_name=draft.full_name, | |
| source_wording=_heading_wording(cluster, evidence) or draft.source_wording, | |
| definition=draft.definition, | |
| formula_latex=draft.formula_latex, | |
| interpretation=draft.interpretation, | |
| subdomain_tags=draft.subdomain_tags, | |
| domain=draft.domain, | |
| company=draft.company, | |
| language=draft.language, | |
| mention_count=cluster.mention_count, | |
| provenance=_prov( | |
| draft.provenance, doc_id, evidence[0].chunk_id if evidence else None | |
| ), | |
| ) | |
| return entry, result.usage | |
| def _heading_wording(cluster: TermCluster, evidence: list[Chunk]) -> str | None: | |
| """The verbatim heading of the evidence chunk whose title names this term. | |
| Preferred over whatever the model chose to quote, because the section | |
| heading is where the document formally names the term. Measured on the | |
| reference standard: the model quoted "Physical Availability (PA)" from the | |
| page-1 intro β a real verbatim quote β while the section itself is headed | |
| "Physical **of** Availability (PA)". Both occur in the document; only the | |
| heading form reveals that the two disagree. | |
| Recording the literal form is a locked decision: the discrepancy belongs to | |
| the expert, not to us. Taking it deterministically rather than asking the | |
| model to volunteer it means it cannot be normalised away. | |
| """ | |
| from ..cluster.normalize import normalize | |
| from ..rank.evidence import _word_match | |
| variants = [normalize(v) for v in cluster.variants] | |
| for chunk in evidence: | |
| heading = chunk.heading | |
| if heading and any(_word_match(v, normalize(heading)) for v in variants): | |
| return heading | |
| return None | |
| # ββ rule of thumb βββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def extract_rule( | |
| candidate: RuleCandidate, chunk: Chunk, extractor, doc_id: str | |
| ) -> tuple[RuleEntry | None, CallUsage]: | |
| user = ( | |
| f"CUE: {candidate.cue}\n\n" | |
| + evidence_block([chunk]) | |
| + f"\n\nFOCUS PASSAGE:\n{candidate.snippet}" | |
| ) | |
| result = extractor.complete( | |
| "rule", load_prompt("rule"), user, schema_for("rule"), "RuleEntry" | |
| ) | |
| try: | |
| draft = RuleDraft.model_validate(result.data) | |
| except Exception as exc: | |
| logger.warning("rule draft invalid", chunk_id=candidate.chunk_id, error=repr(exc)) | |
| return None, result.usage | |
| return ( | |
| RuleEntry( | |
| rule_id=draft.rule_id, | |
| statement=draft.statement, | |
| condition=draft.condition, | |
| consequence=draft.consequence, | |
| applies_to=draft.applies_to, | |
| subdomain_tags=draft.subdomain_tags, | |
| language=draft.language, | |
| provenance=_prov(draft.provenance, doc_id, chunk.chunk_id), | |
| ), | |
| result.usage, | |
| ) | |
| # ββ formula βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def extract_formula( | |
| chunk: Chunk, extractor, doc_id: str | |
| ) -> tuple[FormulaEntry | None, CallUsage]: | |
| user = evidence_block([chunk]) | |
| result = extractor.complete( | |
| "formula", load_prompt("formula"), user, schema_for("formula"), "FormulaEntry" | |
| ) | |
| try: | |
| draft = FormulaDraft.model_validate(result.data) | |
| except Exception as exc: | |
| logger.warning("formula draft invalid", chunk_id=chunk.chunk_id, error=repr(exc)) | |
| return None, result.usage | |
| return ( | |
| FormulaEntry( | |
| name=draft.name, | |
| formula_latex=draft.formula_latex, | |
| variables=[ | |
| FormulaVariable(symbol=v.symbol, meaning=v.meaning) for v in draft.variables | |
| ], | |
| unit=draft.unit, | |
| provenance=_prov(draft.provenance, doc_id, chunk.chunk_id), | |
| ), | |
| result.usage, | |
| ) | |
| # ββ summary βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def extract_summary( | |
| chunks: list[Chunk], extractor, doc_id: str | |
| ) -> tuple[BriefContext | None, CallUsage]: | |
| """Whole-document summary β the quiet cost risk. Few calls, but a large | |
| share of all input tokens, because summarisation cannot be filtered: it | |
| needs the whole document. | |
| It is also the only branch that cannot be span-checked at all. A plausible | |
| summary is indistinguishable from a correct one, which is exactly why it | |
| belongs on a larger tier as soon as one exists. | |
| """ | |
| user = evidence_block(chunks) | |
| result = extractor.complete( | |
| "summary", load_prompt("summary"), user, schema_for("summary"), "BriefContext" | |
| ) | |
| try: | |
| draft = SummaryDraft.model_validate(result.data) | |
| except Exception as exc: | |
| logger.warning("summary draft invalid", error=repr(exc)) | |
| return None, result.usage | |
| return ( | |
| BriefContext( | |
| title=draft.title, | |
| purpose=draft.purpose, | |
| scope=draft.scope, | |
| key_parameters=draft.key_parameters, | |
| summary_md=draft.summary_md, | |
| provenance=_prov(draft.provenance, doc_id), | |
| ), | |
| result.usage, | |
| ) | |