munshi007/aura-statePython-фреймворк для создания LLM-рабочих процессов в виде конечных автоматов с формальной верификацией через теорию доказательств Z3, CTL-модел-чекинг и конформное предсказание для гарантированно корректного извлечения данных.
Фреймворк на Python для построения LLM-воркфлоу в виде конечных автоматов со встроенной формальной верификацией.
pip install git+https://github.com/munshi007/Aura-State.git
Большинство LLM-фреймворков позволяют вам выстраивать цепочки вызовов API и надеяться на лучшее. Aura-State предлагает иной подход: вы определяете свой воркфлоу как граф узлов, каждый из которых выполняет определённую задачу, а фреймворк отвечает за извлечение, верификацию и маршрутизацию.
Ключевое отличие — то, что происходит между узлами:
from aura_state import AuraEngine, Node, CompiledTransition
from pydantic import BaseModel, Field
from openai import OpenAI
# Определите, что вы хотите извлечь
class LeadData(BaseModel):
name: str = Field(description="Полное имя")
budget: int = Field(description="Бюджет в долларах США")
timeline: str = Field(description="Срок покупки")
# Определите узел, который извлекает эти данные
class ExtractLead(Node):
system_prompt = "Извлеките информацию о лиде из стенограммы звонка продаж."
extracts = LeadData
def handle(self, user_text, extracted_data=None, memory=None):
return "QualifyBudget", extracted_data.model_dump()
# Определите узел, выполняющий детерминированную арифметику (без LLM)
class QualifyBudget(Node):
system_prompt = "Оцените лида."
sandbox_rule = "result = budget > 100000" # выполняется в изолированном AST, не через LLM
def handle(self, user_text, extracted_data=None, memory=None):
return "END", memory
# Соберите всё вместе
engine = AuraEngine(llm_client=OpenAI())
engine.register(ExtractLead, QualifyBudget)
engine.connect([
CompiledTransition(from_node=ExtractLead, to_node=QualifyBudget),
])
# Запустите
next_state, data = engine.process("ExtractLead", user_text="Здравствуйте, я Сара. Бюджет — 450 000 долларов.")
При вызове engine.process() выполняются следующие шаги:
1. Адаптивная проверка DAG → Следует ли пропустить или повторить этот узел?
2. Поиск в кеше GraphRAG → Видели ли мы этот вход раньше? Пропускаем LLM.
3. Инжекция few-shot примеров → Поиск похожих успешных случаев, вставка как примеров.
4. Извлечение LLM + верификация → Извлечение данных, проверка с помощью Z3, повтор при ошибке.
5. Метод handle() вашего узла → Здесь выполняется ваша бизнес-логика.
6. Маршрутизация MCTS (UCB1) → Оценка ветвей с помощью UCB1 + метрик AdaptiveDAG.
7. Сериализация состояния → Сохранение состояния для отладки с перемещением во времени.
8. Упреждающее выполнение → Предварительное вычисление вероятных следующих узлов параллельно.
Именно это по-настоящему отличает Aura-State от других фреймворков.
Ваш граф узлов компилируется в структуру Крипке и проверяется на соответствие свойствам темпоральной логики:
from aura_state import verify_engine, reachability, mutual_exclusion, eventual_completion
results = verify_engine(engine, [
{"description": "QualifyBudget достижим", "formula": reachability("QualifyBudget")},
{"description": "Все пути завершаются", "formula": eventual_completion("QualifyBudget")},
])
# Результат: PROVEN или VIOLATED, с точными состояниями, которые удовлетворяют/нарушают свойство
Это тот же метод, который используется для верификации аппаратных схем и систем управления полётом (CTL model checking, Clarke et al., 1986).
После того как LLM извлекает значения, Z3 (теорема-проверщик от Microsoft Research) может формально доказать, что они удовлетворяют вашим ограничениям:
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
# Если False, Z3 выдаёт контрпример, показывающий, что именно нарушено
Запустите извлечение несколько раз и получите непараметрические доверительные интервалы:
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
Здесь используется конформное предсказание (Vovk et al., 2005) — никаких предположений о распределении не требуется.
Мы пропустили 10 стенограмм продаж недвижимости через пайплайн из 4 узлов с использованием GPT-4o-mini (всего 30 вызовов API):
Field Accuracy
────────────── ──────────
name 100%
budget 100%
bedrooms 100%
pre_approved 90%
timeline 90%
city 80%
Временны́е свойства: 3/3 доказаны
Обязательства Z3: 20/20 выполнены
Точность маршрутизации: 90%
Средняя задержка: 1,4 с
# Попробуйте сами — ключ API не нужен
python examples/benchmark/run_benchmark.py
# С реальными LLM вызовами (требуется OPENAI_API_KEY в .env)
python examples/benchmark/run_live.py --model gpt-4o-mini --runs 3
aura_state/
├── core/
│ ├── engine.py # Основной движок — process() + маршрутизация MCTS/UCB1
│ ├── adaptive_graph.py # Мониторинг состояния узлов
│ ├── verification_loop.py # Цикл извлечение → проверка → повтор
│ └── providers.py # Маршрутизация по нескольким моделям + учёт затрат
├── compiler/
│ ├── schema_compiler.py # JSON Schema → классы узлов
│ └── dspy_tuner.py # Выбор few-shot примеров по KNN
├── verification/
│ ├── temporal_verifier.py # Проверка моделей Kripke + CTL
│ ├── conformal.py # Конформные интервалы предсказания
│ └── proof_engine.py # Доказательства Z3
├── execution/
│ ├── tracer.py # Сериализация состояния (отладка с перемещением во времени)
│ └── sandbox.py # Безопасное выполнение арифметики (проверка AST)
├── memory/
│ ├── trajectory_cache.py # Кеш изоморфизма подграфов
│ └── pruner.py # Оптимизация окна контекста
└── consensus/
└── auto_vote.py # Многозапусковое извлечение с голосованием
pip install git+https://github.com/munshi007/Aura-State.git
Требуется Python 3.10+. Зависимости: pydantic, instructor, openai, networkx, pyModelChecking, z3-solver, pyyaml.
python -m pytest tests/ -v
# Проходит 65 тестов
MIT