ai2070-l0

Rename note: This project was previously called L0-Python. The package/repository is being renamed to reliable-ai-streams-py. Existing imports remain supported. Install: pip install reliable-ai-streams-py

L0 - Deterministic Streaming Execution Substrate (DSES) for AI

The missing reliability and observability layer for all AI streams.

LLMs produce high-value reasoning over a low-integrity transport layer. Streams stall, drop tokens, reorder events, violate timing guarantees, and expose no deterministic contract.

This breaks retries. It breaks supervision. It breaks reproducibility. It makes reliable AI systems impossible to build on top of raw provider streams.

L0 is the deterministic execution substrate that fixes the transport - with guardrails designed specifically for the streaming layer: stream-neutral, pattern-based, loop-safe, and timing-aware.

The result: production-grade, integrity-preserving, deterministic AI streams you can finally build real systems on.

It works with OpenAI and LiteLLM (100+ providers including Anthropic, Cohere, Bedrock, Vertex, Gemini). Supports tool calls and provides full observability.

pip install ai2070-l0

Also available in TypeScript: @ai-2070/l0 npm install @ai2070/l0 - native implementation with full lifecycle and event signature parity.

Production-grade reliability. Just pass your stream. L0'll take it from here.

L0 includes 1,800+ tests covering all major reliability features.

   Any AI Stream                    L0 Layer                         Your App
 ─────────────────    ┌──────────────────────────────────────┐    ─────────────
                      │                                      │
   OpenAI / LiteLLM   │   Retry · Fallback · Resume          │      Reliable
   Custom Streams  ──▶│   Guardrails · Timeouts · Consensus  │─────▶ Output
                      │   Full Observability                 │
                      │                                      │
                      └──────────────────────────────────────┘
 ─────────────────                                                ─────────────
                           L0 = Token-Level Reliability

Features

Know what you're doing? Skip the tutorial

Quick Start

Wrap Your Client (Recommended)

import asyncio
from openai import AsyncOpenAI
import l0

async def main():
    # Wrap the client once - L0 reliability is automatic
    client = l0.wrap(AsyncOpenAI())

    # Use normally - no lambdas needed!
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Hello!"}],
        stream=True,
    )

    # Stream with L0 events
    async for event in response:
        if event.is_token:
            print(event.text, end="", flush=True)

    # Or read all at once
    text = await response.read()

asyncio.run(main())

With Configuration

import l0
from openai import AsyncOpenAI

# Configure once, use everywhere
client = l0.wrap(
    AsyncOpenAI(),
    guardrails=l0.Guardrails.recommended(),
    retry=l0.Retry(attempts=5),
    timeout=l0.Timeout(initial_token=10000, inter_token=30000),
    continue_from_last_good_token=True,  # Resume from checkpoint on failure
)

response = await client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}],
    stream=True,
)

See Also: API.md for all options, ADVANCED.md for full examples

With LiteLLM (100+ Providers)

import asyncio
import litellm
import l0

async def main():
    # For LiteLLM, use l0.run() with a factory function
    result = await l0.run(
        stream=lambda: litellm.acompletion(
            model="anthropic/claude-3-haiku-20240307",
            messages=[{"role": "user", "content": "Hello!"}],
            stream=True,
        ),
        guardrails=l0.Guardrails.recommended(),
    )

    # Read full text
    text = await result.read()
    print(text)

asyncio.run(main())

Structured Output with Pydantic

from pydantic import BaseModel
import l0

class UserProfile(BaseModel):
    name: str
    age: int
    occupation: str

result = await l0.structured(
    schema=UserProfile,
    stream=lambda: client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Generate a fictional person as JSON"}],
        stream=True,
    ),
    auto_correct=True,  # Fix trailing commas, missing braces, markdown fences
)

print(result.name)  # "Alice"
print(result.age)   # 32

Lifecycle Events

import l0

result = await l0.run(
    stream=lambda: client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        stream=True,
    ),
    on_stream_event=lambda event: (
        print(event.text, end="") if event.is_token else
        print(f"\nError: {event.error}") if event.is_error else
        print("\nDone!") if event.is_complete else None
    ),
)

Fallback Models & Providers

import l0

result = await l0.run(
    # Primary model
    stream=lambda: client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        stream=True,
    ),
    # Fallbacks: tried in order if primary fails
    fallbacks=[
        lambda: client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            stream=True,
        ),
        lambda: litellm.acompletion(
            model="anthropic/claude-3-haiku-20240307",
            messages=[{"role": "user", "content": prompt}],
            stream=True,
        ),
    ],
    on_fallback=lambda index, reason: print(f"Switched to fallback {index}"),
)

Parallel Execution

import l0

prompts = ["Name a fruit", "Name a color", "Name an animal"]

result = await l0.parallel(
    tasks=[
        lambda p=p: l0.run(
            stream=lambda: client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": p}],
                stream=True,
            ),
        )
        for p in prompts
    ],
    concurrency=3,
)

for prompt, stream in zip(prompts, result.results):
    text = await stream.read()
    print(f"{prompt}: {text.strip()}")

Pydantic Validation Models

L0 provides Pydantic models for all its types, enabling runtime validation, JSON serialization, and schema generation:

from l0.pydantic import StateModel, RetryModel, DriftResultModel

# Validate external data
state = StateModel(content="Hello", token_count=5, completed=True)

# Serialize to JSON
json_data = state.model_dump_json()

# Generate JSON schema for documentation or APIs
schema = StateModel.model_json_schema()

All L0 types have corresponding Pydantic models: StateModel, RetryModel, TimeoutModel, ConsensusResultModel, DriftResultModel, MetricsSnapshotModel, and more.

Philosophy

• No magic - Everything is explicit and predictable
• Streaming-first - Built for real-time token delivery
• Signals, not rewrites - Guardrails detect issues, don't modify output
• Model-agnostic - Works with any provider via adapters
• Pure asyncio - No compatibility layers, native Python async
• Own retry logic - No tenacity, full control over behavior

Performance

Benchmarks on Apple M1 Max, Python 3.13, zero-delay mock streams (2000 tokens):

Full stack = JSON + Markdown + zero-output guardrails + drift detection + checkpointing. See BENCHMARKS.md for details.

Documentation

Installation

# Basic installation
pip install ai2070-l0

# With OpenAI support
pip install ai2070-l0[openai]

# With LiteLLM (100+ providers)
pip install ai2070-l0[litellm]

# With OpenTelemetry
pip install ai2070-l0[otel]

# With Sentry
pip install ai2070-l0[sentry]

# Development (dev is a dependency-group, not a pip extra)
uv sync --group dev

Or with uv:

uv add ai2070-l0
uv add ai2070-l0 --extra openai
uv add ai2070-l0 --extra litellm

Dependencies

Optional Dependencies

License

Apache-2.0