5-Minute Quickstart

# pip install electripy-ai && python quickstart.py
# Runs offline. No API key. No mocks. All five layers active.

import os
from electripy.ai.llm_gateway import LlmGatewaySyncClient, LlmGatewaySettings
from electripy.ai.llm_gateway import LlmMessage, LlmRequest, OpenAiSyncAdapter
from electripy.ai.policy_gateway import PolicyGateway, PolicyRule
from electripy.ai.policy_gateway import PolicyStage, PolicyAction, PolicySeverity
from electripy.ai.policy_gateway import build_llm_policy_hooks
from electripy.concurrency import CircuitBreaker
from electripy.observability.observe import ObservabilityService, InMemoryTracer
from electripy.ai.cost_ledger import CostLedger

# L01 Governance — redact PII before the model sees it
gateway = PolicyGateway(rules=[
    PolicyRule(rule_id="pii-email", code="PII_EMAIL",
               description="Redact emails in prompts",
               stage=PolicyStage.PREFLIGHT,
               pattern=r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}",
               action=PolicyAction.SANITIZE,
               severity=PolicySeverity.MEDIUM),
])

# L03 Reliability — stop cascading failures before they propagate
breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=30.0)

# L02 Observability — structured AI spans with ContextVar propagation
tracer = InMemoryTracer()
obs = ObservabilityService(tracer=tracer)

# L04 Model Runtime — gateway wired to policy hooks
request_hook, response_hook = build_llm_policy_hooks(gateway)
client = LlmGatewaySyncClient(
    port=OpenAiSyncAdapter(api_key=os.environ["OPENAI_API_KEY"]),
    settings=LlmGatewaySettings(
        request_hook=request_hook, response_hook=response_hook
    ),
)
ledger = CostLedger(cost_per_1k_tokens=0.00015)

# The call — emails are redacted before the model sees them
with obs.start_llm_span(provider="openai", model="gpt-4o-mini") as span:
    response = breaker.call(lambda: client.complete(LlmRequest(
        model="gpt-4o-mini",
        messages=[LlmMessage.user("Escalate to admin@acme.com ASAP")],
    )))
    span.set_attribute("gen_ai.usage.total_tokens", response.usage_total_tokens)

ledger.record(tokens=response.usage_total_tokens,
              labels={"model": "gpt-4o-mini", "feature": "support"})

# pip install electripy-ai
# python recipes/async_quickstart/async_quickstart.py
# Runs offline. No API key. asyncio all the way down.

import asyncio
import re
from dataclasses import dataclass
from electripy.ai.llm_gateway import (
    AsyncLlmPort, LlmGatewayAsyncClient, LlmGatewaySettings,
    LlmMessage, LlmRequest, LlmResponse,
)
from electripy.ai.policy_gateway import (
    PolicyGateway, PolicyRule, PolicyStage, PolicyAction,
    PolicySeverity, build_llm_policy_hooks,
)
from electripy.concurrency import CircuitBreaker, CircuitOpenError
from electripy.observability.observe import InMemoryTracer, ObservabilityService
from electripy.ai.cost_ledger import CostLedger

@dataclass
class FakeAsyncProvider(AsyncLlmPort):
    async def complete(self, request, *, timeout=None):
        await asyncio.sleep(0.01)          # simulated I/O
        last = request.messages[-1].content
        return LlmResponse(text=f"Acknowledged: {last}",
                           model=request.model, usage_total_tokens=22)

async def breaker_acall(breaker, coro_fn):
    """Async-safe circuit breaker wrapper."""
    breaker._before_call()
    try:
        result = await coro_fn()
        breaker._on_success()
        return result
    except Exception:
        breaker._on_failure()
        raise

async def main():
    gateway = PolicyGateway(rules=[PolicyRule(
        rule_id="pii-email", code="PII_EMAIL",
        description="Redact emails before model sees them",
        stage=PolicyStage.PREFLIGHT,
        pattern=r"[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}",
        action=PolicyAction.SANITIZE, severity=PolicySeverity.MEDIUM,
    )])
    breaker  = CircuitBreaker(failure_threshold=3, recovery_timeout=30.0)
    obs      = ObservabilityService(tracer=InMemoryTracer())
    ledger   = CostLedger(cost_per_1k_tokens=0.00015)

    req_hook, res_hook = build_llm_policy_hooks(gateway)
    client = LlmGatewayAsyncClient(
        port=FakeAsyncProvider(),
        settings=LlmGatewaySettings(request_hook=req_hook, response_hook=res_hook),
    )

    async with obs.start_llm_span(provider="openai", model="gpt-4o-mini") as span:
        response = await breaker_acall(breaker, lambda: client.complete(LlmRequest(
            model="gpt-4o-mini",
            messages=[LlmMessage.user(
                "Deploy the hotfix and notify ops@acme.com and cto@acme.com."
            )],
        )))
        span.set_attribute("gen_ai.usage.total_tokens", response.usage_total_tokens)

    ledger.record(tokens=response.usage_total_tokens,
                  labels={"model": "gpt-4o-mini", "feature": "ops"})
    print(response.text)   # → Acknowledged: Deploy the hotfix and notify [REDACTED] and [REDACTED].

asyncio.run(main())

# Circuit breaker opens after 3 failures — downstream never called again.
# Same pattern works for any provider adapter.

from electripy.concurrency import CircuitBreaker, CircuitOpenError
from electripy.ai.llm_gateway import (
    AsyncLlmPort, LlmGatewayAsyncClient, LlmGatewaySettings,
    LlmMessage, LlmRequest, LlmResponse,
)
import asyncio
from dataclasses import dataclass

@dataclass
class FlakyProvider(AsyncLlmPort):
    """Simulates an upstream that always times out."""
    async def complete(self, request, *, timeout=None):
        await asyncio.sleep(0.005)
        raise RuntimeError("upstream timeout")

async def breaker_acall(breaker, coro_fn):
    breaker._before_call()
    try:
        result = await coro_fn()
        breaker._on_success()
        return result
    except Exception:
        breaker._on_failure()
        raise

async def main():
    client  = LlmGatewayAsyncClient(port=FlakyProvider(),
                                     settings=LlmGatewaySettings())
    breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=60.0)

    for attempt in range(1, 5):
        try:
            await breaker_acall(breaker, lambda: client.complete(LlmRequest(
                model="gpt-4o-mini", messages=[LlmMessage.user("ping")],
            )))
        except CircuitOpenError:
            print(f"Attempt {attempt}: ⚡ CIRCUIT OPEN — fast-fail, upstream never called")
        except RuntimeError as exc:
            print(f"Attempt {attempt}: ✗ upstream error ({exc}) [circuit={breaker.state.name}]")

asyncio.run(main())