llm-fiber

A thin, fast, and production-ready Python client for LLMs.

Async-first with wire-speed streaming, consistent ergonomics across providers, and built-in observability.

---

llm-fiber is designed for developers who need a reliable, high-performance, and observable way to interact with multiple LLM providers. It eliminates boilerplate and provides a robust toolkit for building production-grade applications, with minimal dependencies and a clean, modern API.

✨ Features

llm-fiber comes packed with features designed for production systems.

Core & Multi-Provider

• 🔄 Async-first with Sync Wrappers: Native async/await for performance, with convenient sync wrappers for simplicity.
• 🎯 Multi-Provider Support: Use OpenAI, Anthropic, and Google Gemini models through a single, unified interface.
• 🤖 Smart Provider Routing: Automatically detects the right provider from the model name (e.g., "gpt-4o-mini" → OpenAI).

Performance & Cost Control

• 🌊 First-Class Streaming: Consume tokens at wire-speed with a clean, backpressure-friendly event stream.
• 💾 Intelligent Caching: Reduce latency and cost with built-in, deterministic caching (in-memory LRU+TTL).
• 💰 Budget & Cost Controls: Enforce token and cost ceilings per-call to prevent runaway spend.
• 🚀 Batch Operations: Process multiple requests concurrently with configurable rate limiting and error handling.

Reliability & Observability

• 🔁 Robust Retries: Automatic, configurable exponential backoff with jitter for transient errors.
• ⏱️ Granular Timeouts: Control connect, read, and total call time to prevent hangs.
• 🔍 Built-in Observability: Get immediate insights with structured logging and metrics for latency, TTFB, retries, and token usage.
• 🛠️ Normalized Tool Calling: A consistent interface for tool/function calls across providers.

🚀 Quick Start

Get up and running in minutes.

1. Installation

pip install llm-fiber

2. Set Environment Variables

export OPENAI_API_KEY="sk-..."
export ANTHROPIC_API_KEY="sk-ant-..."
export GEMINI_API_KEY="..."

3. Make your first calls

import asyncio
from llm_fiber import Fiber, ChatMessage, StreamEventType

# Create a client with auto-detected providers
# The provider is automatically inferred from the model name
fiber = Fiber.from_env()

async def main():
    # Simple text-in, text-out using the best model for the job
    response = await fiber.ask(
        "What is the capital of France?",
        model="gpt-4o-mini" # -> routes to OpenAI
    )
    print(f"Response from GPT: {response}")

    # Full chat with streaming from another provider
    print("\nStreaming response from Claude:")
    async for event in fiber.chat_stream(
        model="claude-3-sonnet-20240229", # -> routes to Anthropic
        messages=[ChatMessage.user("Count from 1 to 5")],
        temperature=0.7
    ):
        if event.type == StreamEventType.CHUNK:
            print(event.delta, end="", flush=True)

# You can also use the sync interface
print("\n\nSync response from Gemini:")
sync_response = fiber.sync.ask(
    "Hello, world!",
    model="gemini-1.5-flash" # -> routes to Google
)
print(sync_response)

asyncio.run(main())

📖 Documentation

For a deep dive into the architecture, features, and design decisions, check out our Full Developer Documentation.

Key pages include:

• Architecture Overview (C4-lite)
• Feature: Retries & Budgets
• Feature: Caching
• Feature: Observability
• Project Roadmap

💡 More Examples

Tool Calling (normalized)

A concise example of receiving a tool call during streaming, normalizing it, and replying with a tool response message.

import asyncio
from llm_fiber import (
    Fiber,
    ChatMessage,
    StreamEventType,
    extract_tool_calls_from_stream_events,
    create_tool_response_message,
)

async def tool_calling_example():
    fiber = Fiber.from_env()

    tools = [{
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get weather by city",
            "parameters": {
                "type": "object",
                "properties": {"city": {"type": "string"}},
                "required": ["city"],
            },
        },
    }]

    events = []
    async for ev in fiber.chat_stream(
        model="gpt-4o-mini",
        messages=[ChatMessage.user("What's the weather in Paris?")],
        tools=tools,
    ):
        events.append(ev)

    # Normalize tool calls across providers
    tool_calls = extract_tool_calls_from_stream_events(events)
    for tc in tool_calls:
        # Call your function here; then respond with a tool message
        tool_response = create_tool_response_message(
            tool_call_id=tc.id,
            content='{"temp_c": 21, "condition": "sunny"}',
            name=tc.name,
        )

        final = await fiber.chat(
            model="gpt-4o-mini",
            messages=[ChatMessage.user("What's the weather in Paris?"), tool_response],
        )
        print(final.text)

asyncio.run(tool_calling_example())

Caching

Enable the in-memory cache to instantly return results for repeated requests, saving time and money.

from llm_fiber import Fiber
# Enable in-memory cache
fiber = Fiber.from_env(
    enable_memory_cache=True
)

# First call: miss -> provider -> writes to cache
result1 = fiber.sync.ask("Explain idempotency in one sentence.")
print("First call finished.")

# Second call: hit -> returns from cache instantly
result2 = fiber.sync.ask("Explain idempotency in one sentence.")
print("Second call finished.")

Budget Controls

Protect your application from unexpected costs by setting a ceiling per call.

from llm_fiber import (
    Fiber,
    BudgetPeriod,
    create_cost_budget,
    create_token_budget,
    BudgetExceededError,
)

budgets = [
    create_cost_budget("daily_spending", 0.10, BudgetPeriod.DAILY, hard_limit=True),
    create_token_budget("hourly_tokens", 500, BudgetPeriod.HOURLY),
]
fiber = Fiber.from_env(budgets=budgets)

try:
    result = fiber.sync.ask(
        "Write a short story about a programmer who discovers a magical bug.",
        model="gpt-4o",
    )
    print(result)
except BudgetExceededError as e:
    print(f"Call stopped: {e}")

Batch Processing

Run multiple jobs concurrently with a configurable worker limit.

import asyncio
from llm_fiber import Fiber, BatchConfig, BatchStrategy, create_batch_from_prompts

async def batch_example():
    fiber = Fiber.from_env()
    jobs = create_batch_from_prompts(
        ["What is the speed of light?", "What is the capital of Japan?", "What is 2+2?"],
        model="gpt-4o-mini",
    )
    cfg = BatchConfig(
        max_concurrent=10,
        strategy=BatchStrategy.CONCURRENT,
        retry_failed=True,
        max_retries=3,
    )
    results = await fiber.batch_chat(jobs, cfg)

    for r in results:
        print(f"[{r.id}] {'OK' if r.is_success else 'FAIL'}")

asyncio.run(batch_example())

⚙️ Installation

Standard Installation

pip install llm-fiber

Optional Dependencies

llm-fiber has optional extras for extended functionality.

# OpenTelemetry integration
pip install "llm-fiber[otel]"

# StatsD metrics export
pip install "llm-fiber[statsd]"

# Structured logging with structlog
pip install "llm-fiber[structlog]"

# Install all extras
pip install "llm-fiber[all]"

🧑‍💻 Development

Set up your local environment for development.

# Clone the repository
git clone https://github.com/aimlesx/llm-fiber
cd llm-fiber

# Install in editable mode with dev dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Format and lint code
ruff format .
ruff check .

# Run type checking
mypy src/

🙌 Contributing

Contributions are welcome! Please read our Architecture Decision Records (ADRs) to understand the design principles behind the library.

When contributing, please ensure you:

1. Update relevant documentation in docs/features/.
2. Add tests for new functionality.
3. Follow the existing code style and conventions.

📄 License

This project is licensed under the MIT License. See the LICENSE file for details.