LLMワークフローをステートマシンとして構築するためのPythonフレームワーク。形式検証が組み込まれています。
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="Full name")
budget: int = Field(description="Budget in USD")
timeline: str = Field(description="Buying timeline")
# 抽出を行うノードを定義
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()
# 決定論的な計算（LLM無し）を行うノードを定義
class QualifyBudget(Node):
system_prompt = "Score the lead."
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="Hi, I'm Sarah. Budget is $450k.")
engine.process()を呼び出すと、以下の手順が順に実行されます：
1. Adaptive DAGヘルスチェック → このノードをスキップまたは再試行すべきか？
2. GraphRAGキャッシュ検索 → この入力が以前にあったか？LLMをスキップ。
3. Few-shotインジェクション → 類似の成功例を探し、例として注入。
4. LLM抽出＋検証 → データを抽出し、Z3で検証、間違いなら再試行。
5. ノードのhandle()メソッド → ビジネスロジックはここで実行。
6. MCTSルーティング（UCB1） → UCB1＋AdaptiveDAGメトリクスでブランチをスコアリング。
7. 状態のシリアライズ → タイムトラベルデバッグ用に状態を保存。
8. 投機的実行 → 並列で次のノードを事前計算。
これこそがAura-Stateを他のフレームワークと実際に差別化するものです。
ノードグラフはKripke構造にコンパイルされ、時相論理プロパティに対してチェックされます：
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モデル検査、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）— 分布の仮定は不要です。
4ノードパイプラインで不動産販売のトランスクリプト10件を実行（GPT-4o-mini、合計30回のAPI呼び出し）：
フィールド 精度
────────────── ──────────
name 100%
budget 100%
bedrooms 100%
pre_approved 90%
timeline 90%
city 80%
時間的プロパティ: 3/3 証明済み
Z3証明義務: 20/20 パス
ルーティング精度: 90%
平均レイテンシ: 1.4s
# 自分で試してみる — APIキー不要
python examples/benchmark/run_benchmark.py
# 実際のLLM呼び出し（.envにOPENAI_API_KEYが必要）
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スキーマ → ノードクラス
│ └── dspy_tuner.py # KNN few-shot選択
├── 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