KAILib — the foundation every system stands on.

from kai_lib import KAIApp, BreakerState

app = KAIApp()                                   # local model + local store, sovereign

def answer(question: str) -> str:
    h = app.llm.health()                          # typed Health — no dict keys to guess
    if not h.healthy or h.breaker == BreakerState.OPEN:
        return "Service degraded — please retry shortly."   # fail honest, not fake

    reply = app.llm.ask(question, system="Answer only from what you know; say so if unsure.")
    app.store.append_event("assistant", "answered",
                          {"q": question, "a": reply, "model": h.model})
    return reply

# later: replay the whole conversation for review / export
for e in app.store.read_events("assistant", limit=500):
    print(e.ts, e.payload["q"], "→", e.payload["a"][:60])

import asyncio
from kai_lib import KAIApp
from kai_lib.config import ResilienceConfig

# admit 8 calls at a time: gather() may launch thousands; the limiter holds the line
app = KAIApp(resilience=ResilienceConfig(max_concurrency=8))

async def summarize_all(docs: list[str]) -> list[str]:
    async def one(text):
        return await asyncio.to_thread(
            app.llm.ask, f"Summarize in 2 lines:\n{text}")
    return await asyncio.gather(*(one(d) for d in docs))

summaries = asyncio.run(summarize_all(load_docs()))
app.store.put("runs", "latest", {"n": len(summaries)})
print(app.llm.health().cache_hit_rate)          # typed Health.cache_hit_rate

from kai_lib import KAIApp
from kai_lib.config import WebConfig
from kai_lib.errors import WebError   # parent — also catches WebBlockedError

# web is off unless you construct it; here we opt in deliberately
app = KAIApp(web_config=WebConfig(timeout=15))

def brief(topic: str, source_url: str | None = None) -> str:
    context = ""
    if source_url:
        try:
            page = app.web.fetch(source_url)         # SSRF-guarded, size-capped
            context = page.text[:4000]
        except WebError as e:                     # blocked, too large, timeout — all WebError
            context = f"(source unavailable: {e})"    # degrade, don't crash
    return app.llm.ask(f"Brief me on {topic}.\nContext:\n{context}")

from kai_lib import KAIApp
from kai_lib.config import StoreConfig

app = KAIApp(store_config=StoreConfig(dsn="/var/lib/edge/events.db"))

def on_reading(sensor: str, value: float):         # called from the device loop
    app.store.append_event(f"sensor/{sensor}", "reading", {"v": value})

def hourly_summary(sensor: str, last_id: int) -> str:
    events = app.store.read_events(f"sensor/{sensor}", after_id=last_id, limit=1000)
    series = [e.payload["v"] for e in events][-200:]    # bound the window on constrained hardware
    lo, hi = min(series), max(series)
    return app.llm.ask(f"Range {lo}–{hi}. Flag anything unusual in these {len(series)} readings: {series}")

import time
from kai_lib import KAIApp, BreakerState

with KAIApp() as app:                        # context manager → guaranteed close()
    while serving:
        h = app.llm.health()
        if h.breaker == BreakerState.OPEN or not h.healthy:
            time.sleep(10); continue                # fail fast, breathe, retry
        job = next_job()
        deliver(app.llm.ask(job.prompt))
        app.store.append_event("service", "served", {"job": job.id})

from kai_lib import LLMClient
from kai_lib.llm import AIBBProvider        # the kai_lib[aibb] extra

client = LLMClient(AIBBProvider(
    "https://127.0.0.1:8443",          # a KQAI in front of a boxed Ollama
    token="…",                          # bearer token (client-cert mTLS is on the KQAI roadmap)
    pin="sha256/…"))                     # pin the box's cert — no MITM

print(client.ask("What's the torque spec for the caliper bolt?"))
# same call as a local model — now authenticated, encrypted, and audited on the box