Adapt Experta¶

Adapt Experta — mvp.adapt_experta

Cluster: Agents & LLM | Type: component | MCP Tools: 44

Overview¶

Forward-chaining expert system block built on the Experta library (a Python CLIPS port) with a pure-Python fallback that runs without any optional dependency. Maintains a persistent working memory of typed facts across calls and fires production rules in priority order until no more rules match. Supports crisp production rules and Prolog-style clause rules via a Prolog delegate, making it suitable for both symbolic reasoning and logic-program-style inference.

When to use:

• Encoding domain knowledge as IF-THEN production rules for automated reasoning
• Running forward-chaining inference over a fact base assembled by upstream agent steps
• Querying the current working memory for all facts of a given type
• Combining rule-based expert reasoning with LLM-extracted facts in a hybrid pipeline

Launch-readiness caveat: adapt_experta is most useful when the user already has a structured fact schema and explicit rules, or when an upstream template/agent converts domain language into those structures. It provides real value for deterministic diagnostics, compliance checks, configuration decisions, and auditable fallback logic, but it is not yet a standalone non-developer rule-authoring experience. For first-user onboarding, pair it with guided templates, examples, or natural-language fact extraction so a domain specialist does not have to design __type__ facts and rule objects from scratch.

Backend honesty: when the preferred Experta backend is unavailable or fails during execution, results use the pure-Python fallback but are marked degraded with backend metadata. MCP also exposes backend_capabilities and audited, allowlisted backend_native_read; native backend mutation remains disabled by default.

Example:

from mvp.adapt_experta import AdaptExpertaBlock, ExpertaInput, Rule

block = AdaptExpertaBlock(name="experta")
# Add rules then run inference
block.infer(ExpertaInput(operation="add_rules", rules=[
    Rule(name="fever_rule", conditions=["Temperature"], actions=["Fever"], priority=10),
]))
block.infer(ExpertaInput(operation="add_facts", facts=[{"__type__": "Temperature", "value": 39.5}]))
result = block.infer(ExpertaInput(operation="run"))
# result.value.rules_fired → ["fever_rule"]; result.value.inferred_facts contains Fever fact

Works well with: adapt_eurisko, formal_methods, goal_engine

Public API¶

ExpertaFactRecommendation¶

Validated natural-language -> facts recommendation decision record.

Methods:

fact_count() -> int¶

type_count() -> int¶

to_metadata() -> dict[str, Any]¶

ExpertaDecisionError(ValueError)¶

The LLM did not produce a usable, validated fact recommendation.

LLMExpertaRuntime¶

Provider-neutral natural-language -> facts recommendation runtime over G6's

Constructor:

Methods:

synthesize(text: str, domain: str = '') -> Result[list[dict]]¶

ExpertaFactPlanner¶

Runtime-first facade with the deterministic floor as honest fallback.

Constructor:

Methods:

recommend_facts(text: str, domain: str = '') -> ExpertaFactRecommendation¶

AuditRecord¶

AuditLogger¶

Synchronous SQLite audit logger.

Constructor:

Methods:

log(step_type: str, run_id: str, llm_model: str = '', prompt_text: str = '', response_text: str = '', latency_ms: float = 0.0, token_in: int = 0, token_out: int = 0, cf_value: float = 0.0, metadata: dict[str, Any] | None = None) -> str¶

Write one audit record. Returns the record id. Raises on write failure.

get_run_audit(run_id: str) -> list[AuditRecord]¶

Return all audit records for a run_id, ordered by timestamp ascending.

list_runs(limit: int = 50) -> list[dict[str, Any]]¶

Return summary rows: run_id, started, ended, step_count.

AdaptExpertaBlock(AIBlock[ExpertaInput, ExpertaOutput, dict])¶

Forward-chaining expert system block.

Methods:

infer(data: ExpertaInput) -> Result[ExpertaOutput]¶

Rule(BaseModel)¶

A production rule in the expert system.

ExpertaInput(BaseModel)¶

Input to AdaptExpertaBlock.

ExpertaOutput(BaseModel)¶

Output from AdaptExpertaBlock.

AdaptExpertaMCPBlock(AIBlock[MCPExpertaInput, MCPExpertaOutput, dict])¶

44-op expert system block with SQLite persistence.

Methods:

infer(data: MCPExpertaInput) -> Result[MCPExpertaOutput]¶

MCPExpertaInput(BaseModel)¶

MCPExpertaOutput(BaseModel)¶

ExpertaStore¶

Sync SQLite store with 7 tables.

Constructor:

Methods:

close() -> None¶

save_session(name: str, facts_json: str, rules_json: str, notes: str) -> str¶

load_session(name: str) -> dict[str, Any] | None¶

list_sessions() -> list[dict[str, Any]]¶

delete_session(name: str) -> bool¶

save_ruleset(name: str, rules_json: str, description: str, domain: str, tags: list[str]) -> str¶

load_ruleset(name: str) -> dict[str, Any] | None¶

increment_ruleset_use(name: str) -> None¶

save_ruleset_spec(ruleset_id: str, name: str, spec_json: str) -> None¶

Persist a RulesetSpec attestation alongside a ruleset.

load_ruleset_spec(name: str) -> str¶

Return the RulesetSpec JSON for a ruleset name, or '' if none.

list_rulesets(domain: str = '') -> list[dict[str, Any]]¶

add_run(session_name: str, rules_fired: list[str], n_cycles: int, n_facts: int, notes: str, tags: list[str]) -> str¶

query_runs(session_name: str = '', limit: int = 50) -> list[dict[str, Any]]¶

summarize_runs(session_name: str = '') -> dict[str, Any]¶

get_cf(fact_type: str) -> float | None¶

add_cf(fact_type: str, rule_name: str, cf_value: float, combined_cf: float) -> str¶

add_job(op: str, params_json: str, priority: int) -> str¶

get_job(job_id: str) -> dict[str, Any] | None¶

update_job(job_id: str, status: str, result_json: str) -> None¶

text_search(query: str, top_k: int = 5) -> list[dict[str, Any]]¶

count_all() -> dict[str, int]¶

add_approval_candidate(run_id: str, rule_json: str, cf_score: float, proposer: str = '') -> str¶

list_approval_queue(status: str = 'pending') -> list[dict]¶

get_approval_candidate(candidate_id: str) -> dict | None¶

Return a single approval-queue row by id (any status), or None.

update_approval_status(candidate_id: str, status: str, reviewer_note: str = '', reviewer_id: str = '') -> None¶

get_pending_count() -> int¶

add_durable_fact(fact_json: str, cf_value: float, run_id: str) -> str¶

list_durable_facts() -> list[dict]¶

get_rule_performance(rule_name: str) -> dict | None¶

upsert_rule_performance(rule_name: str, fires_delta: int = 0, positives_delta: int = 0, negatives_delta: int = 0, cf_adjustment: float | None = None, status: str | None = None) -> None¶

list_weak_rules(min_fires: int = 10, max_accuracy: float = 0.2) -> list[dict]¶

Functions¶

agentic_planner_enabled(default_enabled: bool = True) -> bool¶

Decide whether the agentic experta fact-planner should be used.

validate_fact_recommendation(facts: list[dict]) -> list[dict]¶

Reject any fact recommendation outside the canonical shape / bounds.

recommend_facts_floor(text: str, domain: str = '') -> list[dict]¶

Deterministic, deliberately-brittle NL->facts selector (the honest floor).

MCP Tools¶