Coverage for astrocyte/pipeline/wiki_lint.py: 100%
52 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-10-29 22:10 +0000
1"""Wiki lint pass — Karpathy's periodic "lint" op (M12.5).
3Karpathy's LLM Wiki spec includes a lint operation that checks compiled
4pages for contradictions, stale claims, orphans, missing cross-links,
5and data gaps. This module implements **contradiction detection** for
6v1 — the highest-bench-leverage check.
8Mechanism: per WikiPage, send (page content + top-K facts in same scope)
9to an LLM judge with a yes/no contradiction prompt. If the judge flags a
10contradiction, the page is marked unclean and callers can filter it out
11of the synth context.
13The lint is **stateless per call** — callers either (a) run once per
14bank and cache the report, or (b) run inline per-question on the
15small set of wiki hits actually surfaced by recall. The latter pays
16~1-2 LLM calls per question instead of N upfront but only lints
17pages that would have been seen.
19Generic across benches — operates on WikiPage shape (title + content +
20scope) without bench-specific assumptions.
22See:
23- ``docs/_design/llm-wiki-compile.md`` §7 (lint design)
24- Karpathy gist: https://gist.github.com/karpathy/442a6bf555914893e9891c11519de94f
25"""
27from __future__ import annotations
29import json
30import logging
31from dataclasses import dataclass, field
32from typing import TYPE_CHECKING
34from astrocyte.pipeline._json_tolerant import tolerant_json_loads_or_raise
36if TYPE_CHECKING:
37 from astrocyte.provider import LLMProvider
39logger = logging.getLogger("astrocyte.pipeline.wiki_lint")
42@dataclass
43class WikiLintIssue:
44 """An issue detected by the lint pass for one wiki page."""
46 page_id: str
47 kind: str # 'contradicted' | 'stale' | 'orphan'
48 detail: str = ""
51@dataclass
52class WikiLintReport:
53 """Aggregate of lint issues for a set of pages."""
55 bank_id: str
56 issues: list[WikiLintIssue] = field(default_factory=list)
58 def is_clean(self, page_id: str) -> bool:
59 """True if no issues were flagged for this page_id."""
60 return not any(i.page_id == page_id for i in self.issues)
62 def kinds_for(self, page_id: str) -> set[str]:
63 """Set of issue kinds flagged for this page."""
64 return {i.kind for i in self.issues if i.page_id == page_id}
67_CONTRADICTION_PROMPT = """You are auditing a knowledge-base entry against the source facts it summarises.
69Knowledge entry (titled "{title}"):
70{content}
72Source facts (each independently true, most relevant first):
73{facts_block}
75Does the entry contain ANY claim that DISAGREES with one of the source facts in a way that matters for answering questions? A contradiction is when the entry directly asserts something the facts refute. Vagueness or omission is NOT contradiction.
77Respond with JSON only:
78{{"verdict": "OK" or "CONTRADICTED", "explanation": "one sentence; only when CONTRADICTED"}}"""
81async def lint_one_wiki(
82 *,
83 page_id: str,
84 title: str,
85 content: str,
86 facts: list[str],
87 llm_provider: LLMProvider,
88 model: str = "gpt-4o-mini",
89) -> WikiLintIssue | None:
90 """Lint a single wiki page against pre-fetched facts.
92 Returns:
93 ``WikiLintIssue`` with ``kind='contradicted'`` if the LLM judge
94 flags a contradiction, ``None`` otherwise (clean / no facts /
95 LLM failure — resilient).
96 """
97 if not facts or not content.strip():
98 return None
100 facts_block = "\n".join(f"{i + 1}. {fact}" for i, fact in enumerate(facts))
101 prompt = _CONTRADICTION_PROMPT.format(
102 title=title,
103 content=content,
104 facts_block=facts_block,
105 )
107 try:
108 completion = await llm_provider.complete(
109 prompt,
110 model=model,
111 response_format={"type": "json_object"},
112 )
113 except Exception as exc: # noqa: BLE001
114 logger.warning(
115 "wiki_lint llm call failed for page=%s: %s: %s",
116 page_id,
117 type(exc).__name__,
118 exc,
119 )
120 return None
122 try:
123 parsed = tolerant_json_loads_or_raise(completion.text)
124 except (json.JSONDecodeError, AttributeError):
125 logger.warning("wiki_lint json parse failed for page=%s", page_id)
126 return None
128 verdict = str(parsed.get("verdict", "")).strip().upper()
129 if verdict == "CONTRADICTED":
130 return WikiLintIssue(
131 page_id=page_id,
132 kind="contradicted",
133 detail=str(parsed.get("explanation", "")).strip(),
134 )
135 return None
138async def lint_wiki_pages(
139 *,
140 pages: list[tuple[str, str, str, list[str]]],
141 llm_provider: LLMProvider,
142 bank_id: str,
143 model: str = "gpt-4o-mini",
144) -> WikiLintReport:
145 """Batch lint a set of pages.
147 Args:
148 pages: List of ``(page_id, title, content, facts)`` tuples. The
149 tuple shape lets callers adapt from WikiPage / WikiPageHit
150 without forcing a particular dataclass.
151 llm_provider: Provider for the contradiction-detection LLM call.
152 bank_id: Bank scope recorded on the report.
153 model: LLM model for the judge.
155 Returns:
156 ``WikiLintReport`` with one issue per contradicted page. Pages
157 with no facts or that fail the LLM call are silently skipped
158 (treated as clean — fail-open).
159 """
160 report = WikiLintReport(bank_id=bank_id)
161 for page_id, title, content, facts in pages:
162 if not page_id:
163 continue
164 # Orphan: page has no source facts to anchor against. The
165 # caller can still treat orphans as clean for filtering; we
166 # flag for visibility.
167 if not facts:
168 report.issues.append(
169 WikiLintIssue(
170 page_id=page_id,
171 kind="orphan",
172 detail="no facts in scope",
173 )
174 )
175 continue
176 issue = await lint_one_wiki(
177 page_id=page_id,
178 title=title,
179 content=content,
180 facts=facts,
181 llm_provider=llm_provider,
182 model=model,
183 )
184 if issue is not None:
185 report.issues.append(issue)
186 return report