一个用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
# 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.")
当你调用 engine.process() 时,它会按顺序执行以下步骤:
1. 自适应DAG健康检查 → 此节点应跳过还是重试?
2. GraphRAG缓存查找 → 之前是否见过完全相同的输入?跳过LLM。
3. Few-shot注入 → 查找相似的成功案例,注入作为示例。
4. LLM提取 + 验证 → 提取数据,用Z3验证,错误则重试。
5. 节点的handle()方法 → 你的业务逻辑在此运行。
6. MCTS路由(UCB1) → 使用UCB1 + 自适应DAG指标对分支评分。
7. 状态序列化 → 保存状态以便时间旅行调试。
8. 投机执行 → 并行预计算可能的后续节点。
这正是Aura-State与其他框架不同的地方。
你的节点图会被编译成一个Kripke结构,并针对时态逻辑属性进行检查:
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
这与验证硬件电路和飞行控制系统所用的技术相同(CTL模型检验,Clarke et al. 1986)。
在LLM提取值之后,Z3(微软研究院的定理证明器)可以形式化证明这些值满足你的约束:
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
多次运行提取,获取无分布假设的置信区间:
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份房地产销售通话记录(共调用30次GPT-4o-mini API):
Field Accuracy
────────────── ──────────
name 100%
budget 100%
bedrooms 100%
pre_approved 90%
timeline 90%
city 80%
Temporal properties: 3/3 proven
Z3 proof obligations: 20/20 passed
Routing accuracy: 90%
Avg latency: 1.4s
# Try it yourself — no API key needed
python examples/benchmark/run_benchmark.py
# With real LLM calls (needs OPENAI_API_KEY in .env)
python examples/benchmark/run_live.py --model gpt-4o-mini --runs 3
aura_state/
├── core/
│ ├── engine.py # Main engine — process() + MCTS/UCB1 routing
│ ├── adaptive_graph.py # Node health monitoring
│ ├── verification_loop.py # Extract → verify → retry loop
│ └── providers.py # Multi-model routing + cost tracking
├── compiler/
│ ├── schema_compiler.py # JSON Schema → Node classes
│ └── dspy_tuner.py # KNN few-shot selection
├── verification/
│ ├── temporal_verifier.py # Kripke + CTL model checking
│ ├── conformal.py # Conformal prediction intervals
│ └── proof_engine.py # Z3 proofs
├── execution/
│ ├── tracer.py # State serialization (time-travel debug)
│ └── sandbox.py # Safe math execution (AST validated)
├── memory/
│ ├── trajectory_cache.py # Subgraph isomorphism cache
│ └── pruner.py # Context window optimization
└── consensus/
└── auto_vote.py # Multi-run extraction with voting
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 tests passing
MIT