munshi007/aura-stateFramework Python per costruire flussi di lavoro LLM come macchine a stati con verifica formale tramite dimostrazione di teoremi Z3, model checking CTL, e predizione conforme per estrazione dati dimostrabilmente corretta.
Un framework Python per costruire workflow LLM come macchine a stati, con verifica formale integrata.
pip install git+https://github.com/munshi007/Aura-State.git
La maggior parte dei framework LLM ti permettono di concatenare chiamate API e sperare per il meglio. Aura-State adotta un approccio diverso: definisci il tuo workflow come un grafo di nodi, ciascuno con un compito specifico, e il framework gestisce estrazione, verifica e routing.
La differenza fondamentale è ciò che accade tra i nodi:
from aura_state import AuraEngine, Node, CompiledTransition
from pydantic import BaseModel, Field
from openai import OpenAI
# Define what you want to extract
class LeadData(BaseModel):
name: str = Field(description="Full name")
budget: int = Field(description="Budget in USD")
timeline: str = Field(description="Buying timeline")
# Define a node that extracts it
class ExtractLead(Node):
system_prompt = "Extract lead info from a sales call transcript."
extracts = LeadData
def handle(self, user_text, extracted_data=None, memory=None):
return "QualifyBudget", extracted_data.model_dump()
# Define a node that does deterministic math (no LLM)
class QualifyBudget(Node):
system_prompt = "Score the lead."
sandbox_rule = "result = budget > 100000" # runs in sandboxed AST, not LLM
def handle(self, user_text, extracted_data=None, memory=None):
return "END", memory
# Wire it up
engine = AuraEngine(llm_client=OpenAI())
engine.register(ExtractLead, QualifyBudget)
engine.connect([
CompiledTransition(from_node=ExtractLead, to_node=QualifyBudget),
])
# Run
next_state, data = engine.process("ExtractLead", user_text="Hi, I'm Sarah. Budget is $450k.")
Quando chiami engine.process(), vengono eseguiti questi passaggi in ordine:
1. Adaptive DAG health check → Questo nodo dovrebbe essere saltato o riprovato?
2. GraphRAG cache lookup → Abbiamo già visto questo input? Salta l'LLM.
3. Few-shot injection → Trova successi simili passati, inserisci come esempi.
4. LLM extraction + verification → Estrai dati, verifica con Z3, riprova se errato.
5. Your node's handle() method → La tua logica di business viene eseguita qui.
6. MCTS Routing (UCB1) → Valuta i rami usando UCB1 + metriche AdaptiveDAG.
7. State serialization → Salva lo stato per debugging temporale.
8. Speculative execution → Pre-calcola i nodi successivi probabili in parallelo.
Questo è ciò che rende Aura-State realmente diverso dagli altri framework.
Il tuo grafo di nodi viene compilato in una Kripke structure e controllato rispetto a proprietà di logica temporale:
from aura_state import verify_engine, reachability, mutual_exclusion, eventual_completion
results = verify_engine(engine, [
{"description": "QualifyBudget is reachable", "formula": reachability("QualifyBudget")},
{"description": "All paths terminate", "formula": eventual_completion("QualifyBudget")},
])
# Result: PROVEN or VIOLATED, with the exact states that satisfy/violate
Questa è la stessa tecnica utilizzata per verificare circuiti hardware e sistemi di controllo di volo (CTL model checking, Clarke et al. 1986).
Dopo che l'LLM estrae i valori, Z3 (un dimostratore di teoremi di Microsoft Research) può dimostrare formalmente che soddisfano i tuoi vincoli:
from aura_state import prove_extraction
result = prove_extraction(
{"budget": 450000, "cost_per_sqft": 3, "total": 1350000},
obligations=["budget > 0", "total == budget * cost_per_sqft"],
)
# result.verified = True
# If False, Z3 gives you a counterexample showing exactly what broke
Esegui l'estrazione più volte e ottieni intervalli di confidenza senza distribuzione:
from aura_state import conformal_interval
budgets = [450000, 452000, 448000, 450000, 451000]
ci = conformal_interval(budgets, confidence=0.95)
# ci.lower = 447800, ci.upper = 452200
Questo utilizza la predizione conformale (Vovk et al., 2005) — non sono richieste assunzioni distribuzionali.
Abbiamo eseguito 10 trascrizioni di vendite immobiliari attraverso un pipeline a 4 nodi usando GPT-4o-mini (30 chiamate API totali):
Campo Precisione
────────────── ──────────
name 100%
budget 100%
bedrooms 100%
pre_approved 90%
timeline 90%
city 80%
Proprietà temporali: 3/3 dimostrate
Obblighi di prova Z3: 20/20 passati
Accuratezza del routing: 90%
Latenza media: 1.4s
# Prova tu stesso — non serve una chiave API
python examples/benchmark/run_benchmark.py
# Con chiamate LLM reali (necessita OPENAI_API_KEY in .env)
python examples/benchmark/run_live.py --model gpt-4o-mini --runs 3
aura_state/
├── core/
│ ├── engine.py # Motore principale — process() + routing MCTS/UCB1
│ ├── adaptive_graph.py # Monitoraggio salute nodi
│ ├── verification_loop.py # Ciclo estrai → verifica → riprova
│ └── providers.py # Routing multi-modello + tracciamento costi
├── compiler/
│ ├── schema_compiler.py # JSON Schema → classi Node
│ └── dspy_tuner.py # Selezione few-shot KNN
├── verification/
│ ├── temporal_verifier.py # Kripke + model checking CTL
│ ├── conformal.py # Intervalli di predizione conformale
│ └── proof_engine.py # Dimostrazioni Z3
├── execution/
│ ├── tracer.py # Serializzazione stato (debug temporale)
│ └── sandbox.py # Esecuzione matematica sicura (AST validato)
├── memory/
│ ├── trajectory_cache.py # Cache per isomorfismo di sottografi
│ └── pruner.py # Ottimizzazione finestra contesto
└── consensus/
└── auto_vote.py # Estrazione multi-esecuzione con voto
pip install git+https://github.com/munshi007/Aura-State.git
Richiede Python 3.10+. Dipendenze: pydantic, instructor, openai, networkx, pyModelChecking, z3-solver, pyyaml.
python -m pytest tests/ -v
# 65 test passati
MIT