Grounding¶

grounding - domain-specific grounding agents.

Cluster: Knowledge & Grounding | Type: component | MCP Tools: None

Overview¶

grounding provides a domain knowledge base with TF-IDF retrieval across 19 built-in domains (science, medicine, mathematics, engineering, technology, law, arts, multimedia, logic_smt, proof_theory, model_theory, logic_programming, category_theory, type_theory, topology, algebra, causal_inference, annotated_logic, optimisation), each seeded with curated facts. A staleness detector adjusts confidence scores based on fact freshness.

Source capability discovery is available through GroundingBlock.source_capabilities() and the MCP grounding_capabilities tool. These surfaces report active and unavailable backend tiers before a query, including seed KB, manifest corpus, sklearn TF-IDF, OpenRouter, multimedia FAISS, and DuckDuckGo's network-dependent tier.

Source-quality gate for verified labels. A verified completion state is only emitted when the production corpus is available and not degraded, coverage is sufficient, no knowledge gap is detected, AND every cited source passes a deterministic source-quality gate: fresh within its source-registry SLA (default 365 days; per-source max_age_days honoured), carries a self-declared provenance/authority marker, is non-hostile, and passes a citation-faithfulness check. Every one of these is a check on metadata the source record declares; none of them authenticate the source itself. Any failure emits a stable G6_E_GROUNDING_SOURCE_* / G6_E_GROUNDING_CITATION_UNFAITHFUL code, ratchets the result down to qualified-draft, and records the reason in evidence["source_quality"]. Per-source SLA/provenance entries are exposed via source_registry_entry().

Freshness / provenance: sources missing fetched_at/indexed_at freshness metadata, or exceeding their SLA, or lacking a provenance marker, are flagged and cannot reach verified.
Hostile-source isolation: retrieved source text containing prompt/tool directives (e.g. "ignore previous instructions", "mark this verified") is treated as DATA only — it never alters control flow — flagged G6_E_GROUNDING_SOURCE_HOSTILE, and isolated from the verified decision.
Citation faithfulness is a HEURISTIC, not a proof: it is a content-token-overlap tripwire between the emitted claim text and the cited support text (evidence["source_quality"]["faithfulness_scope"] == "heuristic_token_overlap_not_proof"). It catches gross unfaithfulness; it is not an entailment/NLI guarantee.

The TF-IDF vectorizer is rebuilt on each infer() call. This is acceptable for the seed KB but can add latency when a large manifest-corpus result set is loaded; vectorizer caching is deferred to a dedicated performance phase.

When to use:

Checking methodology against retrieved domain knowledge
Retrieving domain-specific facts with a retrieval-similarity score (confidence is the mean query-fact similarity, not a relevance or answer-trust judgement)
Grounding reasoning in cited sources with freshness-aware retrieval

Example:

from mvp.grounding import GroundingBlock, GroundingInput

block = GroundingBlock(name="g")
result = block.infer(GroundingInput(query="backpropagation", domain="science"))
# result.ok → True; result.value → GroundingOutput with retrieved_facts and confidence

Works well with: ctx_rag, ctx_colbert, align_evals, goal_engine

Public API¶

`GroundingBlock(AIBlock[GroundingInput, GroundingOutput, None])`¶

Domain-specific grounding agent.

Field	Type	Default
`name`	`str`	'grounding'
`resource_bounds`	`ResourceBounds \\| None`	None
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)
`staleness_detector`	`StalenessDetector`	field(default_factory=StalenessDetector)

Methods:

`source_health(domain: str = 'science') -> dict[str, object]`¶

Return backend health without executing retrieval or network calls.

`source_capabilities(domain: str = 'science') -> dict[str, object]`¶

Return source tier capabilities for routing and preflight checks.

`infer(data: GroundingInput) -> Result[GroundingOutput]`¶

`list_patterns() -> dict[str, object]`¶

Return the applied deterministic-reliability pattern catalog.

`bias() -> dict`¶

`GroundingInput(BaseModel)`¶

Input to GroundingBlock.

Field	Type	Default
`query`	`str`	required
`domain`	`DOMAIN_LITERAL`	'science'
`top_k`	`int`	3
`include_sources`	`bool`	True
`use_web`	`bool`	False
`use_perplexity`	`bool`	False
`extra_facts`	`list[tuple[str, str]]`	Field(default_factory=list)
`domain_override`	`str`	''
`agent_name`	`str`	''
`corpus_agents`	`list[str]`	Field(default_factory=list)
`stakes`	`str`	'medium'
`strict_grounding`	`bool`	True
`image_url`	`str`	''
`audio_url`	`str`	''
`video_url`	`str`	''
`pdf_url`	`str`	''

`GroundingOutput(BaseModel)`¶

Output from GroundingBlock.

Field	Type	Default
`query`	`str`	required
`domain`	`str`	required
`answer`	`str`	required
`sources`	`list[str]`	Field(default_factory=list)
`confidence`	`float`	required
`retrieved_facts`	`list[str]`	Field(default_factory=list)
`provenance`	`list[dict]`	Field(default_factory=list)
`retrieval_scores`	`list[float]`	Field(default_factory=list)
`retrieval_method`	`str`	'tfidf'
`degraded`	`bool`	False
`degradation_reason`	`str`	''
`corpus_available`	`bool`	True
`corpus_degradation_reason`	`str`	''
`completion_state`	`Literal['verified', 'qualified-draft', 'blocked-escalated']`	'qualified-draft'
`warning_card`	`dict`	Field(default_factory=dict)
`evidence`	`dict`	Field(default_factory=dict)
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''
`grounding_coverage`	`float`	0.0
`coverage_sufficient`	`bool`	False
`coverage_threshold`	`float`	0.0
`authority_weighted_confidence`	`float`	0.5
`citation_weighted_confidence`	`float`	0.5
`knowledge_gap_detected`	`bool`	False
`knowledge_gap_message`	`str`	''
`corpus_density_score`	`float`	0.0
`epistemic_status`	`str`	'UNVERIFIABLE'
`perplexity_requested`	`bool`	False
`perplexity_used`	`bool`	False
`perplexity_warning`	`str`	''
`relace_used`	`bool`	False
`relace_warning`	`str`	''
`grounding_quality`	`dict`	Field(default_factory=dict)

`DatedFact`¶

A fact with provenance and temporal metadata.

Field	Type	Default
`text`	`str`	required
`source`	`str`	required
`domain`	`str`	required
`published_date`	`str`	''
`added_date`	`str`	field(default_factory=lambda: datetime.now(timezone.utc).strftime('%Y-%m-%d'))
`expiry_days`	`int`	DEFAULT_EXPIRE_DAYS

`StalenessReport`¶

Report on the freshness status of a set of facts.

Field	Type	Default
`total_facts`	`int`	0
`fresh_count`	`int`	0
`stale_count`	`int`	0
`expired_count`	`int`	0
`unknown_age_count`	`int`	0
`details`	`list[dict[str, Any]]`	field(default_factory=list)

`StalenessDetector`¶

Validates KB fact freshness and adjusts confidence accordingly.

Constructor:

Parameter	Type	Default
`warn_days`	`int`	DEFAULT_WARN_DAYS
`expire_days`	`int`	DEFAULT_EXPIRE_DAYS

Methods:

`check_freshness(facts: list[tuple[str, str]], domain: str = 'general') -> StalenessReport`¶

Check freshness of a list of (fact, source) tuples.

`adjust_confidence(base_confidence: float, facts: list[tuple[str, str]], domain: str = 'general') -> tuple[float, StalenessReport]`¶

Adjust a confidence score by penalising stale facts.

`filter_expired(facts: list[tuple[str, str]], domain: str = 'general') -> list[tuple[str, str]]`¶

Remove expired facts from a list.

Grounding¶

Overview¶

Public API¶

GroundingBlock(AIBlock[GroundingInput, GroundingOutput, None])¶

source_health(domain: str = 'science') -> dict[str, object]¶

source_capabilities(domain: str = 'science') -> dict[str, object]¶

infer(data: GroundingInput) -> Result[GroundingOutput]¶

list_patterns() -> dict[str, object]¶

bias() -> dict¶

GroundingInput(BaseModel)¶

GroundingOutput(BaseModel)¶

DatedFact¶

StalenessReport¶

StalenessDetector¶

check_freshness(facts: list[tuple[str, str]], domain: str = 'general') -> StalenessReport¶

adjust_confidence(base_confidence: float, facts: list[tuple[str, str]], domain: str = 'general') -> tuple[float, StalenessReport]¶

filter_expired(facts: list[tuple[str, str]], domain: str = 'general') -> list[tuple[str, str]]¶