lore-agents 0.1.0

Agent reliability layer: circuit breakers, dead letter queues, cost guards, and verification loops for any AI agent framework

lore-agents

Agent reliability layer. Adds circuit breakers, dead letter queues, cost guards, model routing, and execution tracing to any AI agent framework.

Install

pip install lore-agents

Quick Start

from lore import CircuitBreaker, CostGuard, DeadLetterQueue, LoreTrace, lore_wrap

# Wrap any LLM call with full reliability stack
safe_call = lore_wrap(
    my_llm_function,
    name="openai",
    failure_threshold=3,
    session_limit_usd=5.0,
)
result = safe_call("Hello, world!")

Features

Circuit Breaker

Prevents cascading failures when LLM providers go down.

from lore import CircuitBreaker

cb = CircuitBreaker(name="openai", failure_threshold=3, recovery_timeout=30)
result = cb.call(openai_chat, prompt="hello")

# Or use as a decorator
from lore import circuit_breaker

@circuit_breaker(name="anthropic", threshold=5)
def call_claude(prompt: str) -> str:
    return client.messages.create(...)

States: CLOSED (normal) -> OPEN (rejecting) -> HALF_OPEN (probing) -> CLOSED

Dead Letter Queue

SQLite-backed queue that captures failed tasks with automatic error classification.

from lore import DeadLetterQueue

dlq = DeadLetterQueue()

try:
    result = agent.run(task)
except Exception as e:
    dlq.push("task-123", e, context='{"model": "gpt-4o"}')

# Replay transient failures
entries = dlq.replay_transient(rate=10)

Classification: Transient (retry) | Permanent (park) | Ambiguous (one retry)

Cost Guard

Per-task and per-session spending limits with alerting.

from lore import CostGuard

guard = CostGuard(session_limit_usd=5.00, task_limit_usd=0.50)
guard.check_budget("task-1")  # Pre-flight check
guard.record("gpt-4o", input_tokens=1000, output_tokens=500, task_id="task-1")
print(guard.get_summary())

Model Router

3-tier routing (budget/mid/frontier) with latency tracking and fallback cascades.

from lore import ModelRouter, ModelTier

router = ModelRouter(
    task_type_map={"crud": ModelTier.BUDGET, "architecture": ModelTier.FRONTIER}
)
model = router.route("crud")  # -> cheapest healthy model

Execution Tracing

Step-level cost and duration tracking, exportable to JSON.

from lore import LoreTrace

tracer = LoreTrace("customer_support")
with tracer.step("classify", model="gpt-4o-mini") as s:
    result = classify(query)
    s.input_tokens = 150
    s.output_tokens = 20

tracer.finish()
print(tracer.to_json())

Audit Scanner

Scans agent code for missing reliability patterns.

from lore import LoreAudit

audit = LoreAudit("/path/to/project")
report = audit.scan()
print(report.to_json())

CLI

lore install .          # Install reliability rules + pre-commit hooks
lore audit .            # Scan for missing patterns
lore audit . -j         # JSON output
lore trace trace.json   # View a trace file
lore wrap               # Show wrap usage examples

Optional Dependencies

pip install lore-agents[redis]  # Redis-backed circuit breaker persistence
pip install lore-agents[dev]    # Development tools (pytest, mypy, ruff)

License

MIT