
Framework de Python para construir flujos de trabajo de LLM como máquinas de estado con verificación formal mediante demostración de teoremas Z3, verificación de modelos CTL y predicción conforme para extracción de datos demostrablemente correcta.
Un framework en Python para construir flujos de trabajo de LLM como máquinas de estado, con verificación formal integrada.
pip install git+https://github.com/munshi007/Aura-State.git
La mayoría de los frameworks de LLM te permiten encadenar llamadas a la API y esperar lo mejor. Aura-State adopta un enfoque diferente: defines tu flujo de trabajo como un grafo de nodos, cada uno con una tarea específica, y el framework se encarga de la extracción, verificación y enrutamiento.
La diferencia clave está en lo que ocurre entre los nodos:
from aura_state import AuraEngine, Node, CompiledTransition
from pydantic import BaseModel, Field
from openai import OpenAI
# Define lo que quieres extraer
class LeadData(BaseModel):
name: str = Field(description="Nombre completo")
budget: int = Field(description="Presupuesto en USD")
timeline: str = Field(description="Cronograma de compra")
# Define un nodo que lo extrae
class ExtractLead(Node):
system_prompt = "Extrae información del cliente potencial de una transcripción de llamada de ventas."
extracts = LeadData
def handle(self, user_text, extracted_data=None, memory=None):
return "QualifyBudget", extracted_data.model_dump()
# Define un nodo que realiza matemáticas deterministas (sin LLM)
class QualifyBudget(Node):
system_prompt = "Puntúa al cliente potencial."
sandbox_rule = "result = budget > 100000" # se ejecuta en un AST aislado, no en el LLM
def handle(self, user_text, extracted_data=None, memory=None):
return "END", memory
# Conéctalos
engine = AuraEngine(llm_client=OpenAI())
engine.register(ExtractLead, QualifyBudget)
engine.connect([
CompiledTransition(from_node=ExtractLead, to_node=QualifyBudget),
])
# Ejecuta
next_state, data = engine.process("ExtractLead", user_text="Hola, soy Sarah. El presupuesto es de $450k.")
Cuando llamas a engine.process(), se ejecutan estos pasos en orden:
1. Comprobación de salud del DAG adaptativo → ¿Este nodo debe omitirse o reintentarse?
2. Búsqueda en caché de GraphRAG → ¿Ya hemos visto esta entrada exacta? Omitir el LLM.
3. Inyección de pocos ejemplos (few-shot) → Encuentra éxitos similares pasados, inyecta como ejemplos.
4. Extracción + verificación del LLM → Extrae datos, verifica con Z3, reintenta si es incorrecto.
5. El método handle() de tu nodo → Aquí se ejecuta tu lógica de negocio.
6. Enrutamiento MCTS (UCB1) → Puntúa las ramas usando UCB1 + métricas del DAG adaptativo.
7. Serialización del estado → Guarda el estado para depuración con viaje en el tiempo.
8. Ejecución especulativa → Precalcula los siguientes nodos probables en paralelo.
Esto es lo que realmente diferencia a Aura-State de otros frameworks.
Tu grafo de nodos se compila en una estructura de Kripke y se verifica contra propiedades de lógica temporal:
from aura_state import verify_engine, reachability, mutual_exclusion, eventual_completion
results = verify_engine(engine, [
{"description": "QualifyBudget es alcanzable", "formula": reachability("QualifyBudget")},
{"description": "Todos los caminos terminan", "formula": eventual_completion("QualifyBudget")},
])
# Resultado: PROVEN o VIOLATED, con los estados exactos que satisfacen/violan
Esta es la misma técnica utilizada para verificar circuitos de hardware y sistemas de control de vuelo (verificación de modelos CTL, Clarke et al. 1986).
Después de que el LLM extrae los valores, Z3 (un demostrador de teoremas de Microsoft Research) puede probar formalmente que cumplen tus restricciones:
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
# Si es False, Z3 te da un contraejemplo que muestra exactamente qué falló
Ejecuta la extracción varias veces y obtén intervalos de confianza libres de distribución:
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
Esto utiliza predicción conforme (Vovk et al., 2005) — no se requieren supuestos distribucionales.
Ejecutamos 10 transcripciones de ventas inmobiliarias a través de un pipeline de 4 nodos usando GPT-4o-mini (30 llamadas a la API en total):
Campo Precisión
────────────── ──────────
name 100%
budget 100%
bedrooms 100%
pre_approved 90%
timeline 90%
city 80%
Propiedades temporales: 3/3 probadas
Obligaciones de prueba Z3: 20/20 pasadas
Precisión de enrutamiento: 90%
Latencia media: 1.4s
# Pruébalo tú mismo — no se necesita clave API
python examples/benchmark/run_benchmark.py
# Con llamadas reales al LLM (necesita OPENAI_API_KEY en .env)
python examples/benchmark/run_live.py --model gpt-4o-mini --runs 3
aura_state/
├── core/
│ ├── engine.py # Motor principal — process() + enrutamiento MCTS/UCB1
│ ├── adaptive_graph.py # Monitoreo de salud de nodos
│ ├── verification_loop.py # Bucle extraer → verificar → reintentar
│ └── providers.py # Enrutamiento multimodelo + seguimiento de costos
├── compiler/
│ ├── schema_compiler.py # JSON Schema → Clases de nodo
│ └── dspy_tuner.py # Selección de pocos ejemplos por KNN
├── verification/
│ ├── temporal_verifier.py # Kripke + verificación de modelo CTL
│ ├── conformal.py # Intervalos de predicción conforme
│ └── proof_engine.py # Pruebas Z3
├── execution/
│ ├── tracer.py # Serialización de estado (depuración con viaje en el tiempo)
│ └── sandbox.py # Ejecución segura de matemáticas (AST validado)
├── memory/
│ ├── trajectory_cache.py # Caché de isomorfismo de subgrafos
│ └── pruner.py # Optimización de ventana de contexto
└── consensus/
└── auto_vote.py # Extracción de múltiples ejecuciones con votación
pip install git+https://github.com/munshi007/Aura-State.git
Se requiere Python 3.10+. Dependencias: pydantic, instructor, openai, networkx, pyModelChecking, z3-solver, pyyaml.
python -m pytest tests/ -v
# 65 pruebas pasando
MIT