Spaces:
Paused
Paused
| """Consistency guard (Section 8.2). | |
| Deterministic topic/polarity check against the ledger first (cheap, keeps the | |
| hot path cheap); LLM semantic-entailment check only if the cheap pass is clean. | |
| Flags contradictions with already-committed claims. | |
| """ | |
| from __future__ import annotations | |
| from dataclasses import dataclass | |
| from ...llm.client import LLMClient | |
| from ...llm.prompts import PromptRegistry | |
| from ...models import Claim | |
| from ..ledger import LedgerStore | |
| class ConsistencyVerdict: | |
| contradicts: bool | |
| reason: str | |
| deterministic: bool = False | |
| class ConsistencyGuard: | |
| def __init__( | |
| self, client: LLMClient, prompts: PromptRegistry, ledger: LedgerStore | |
| ) -> None: | |
| self.client = client | |
| self.prompts = prompts | |
| self.ledger = ledger | |
| def check( | |
| self, *, character: str, draft: str, new_claims: list[Claim] | |
| ) -> ConsistencyVerdict: | |
| # 1) cheap deterministic pass | |
| hits = self.ledger.find_contradictions(character, new_claims) | |
| if hits: | |
| prior, new = hits[0] | |
| return ConsistencyVerdict( | |
| True, | |
| f"contradicts prior statement on '{prior.topic}': " | |
| f"\"{prior.proposition}\" vs \"{new.proposition}\"", | |
| deterministic=True, | |
| ) | |
| prior_claims = self.ledger.get(character).claims | |
| if not prior_claims: | |
| return ConsistencyVerdict(False, "no prior claims") | |
| # 2) LLM semantic entailment fallback | |
| prompt = self.prompts.render( | |
| "guard/consistency.md.j2", | |
| character=character, | |
| draft=draft, | |
| prior_claims=[ | |
| {"topic": c.topic, "proposition": c.proposition, | |
| "polarity": c.polarity} | |
| for c in prior_claims | |
| ], | |
| ) | |
| try: | |
| data, _ = self.client.complete_json( | |
| tier="guard", task="consistency_guard", user=prompt, | |
| ) | |
| except Exception: | |
| # On error, don't block on semantic check (deterministic pass is | |
| # the hard guarantee); allow but note. | |
| return ConsistencyVerdict(False, "semantic check skipped (error)") | |
| contradicts = ( | |
| bool(data.get("contradicts", False)) if isinstance(data, dict) else False | |
| ) | |
| reason = str(data.get("reason", "")) if isinstance(data, dict) else "" | |
| return ConsistencyVerdict(contradicts, reason) | |