Skip to main content

Brokle Platform Python SDK for intelligent LLM routing, caching, and observability

Project description

Brokle Python SDK

Three integration patterns. One powerful platform.

The Brokle Python SDK provides intelligent routing across 250+ LLM providers, semantic caching (30-50% cost reduction), and comprehensive observability. Choose your integration level:

🎯 Three Integration Patterns

Pattern 1: Wrapper Functions Wrap existing SDK clients (OpenAI, Anthropic) for automatic observability and platform features.

Pattern 2: Universal Decorator Framework-agnostic @observe() decorator with automatic hierarchical tracing. Works with any AI library.

Pattern 3: Native SDK (Sync & Async) Full platform capabilities: intelligent routing, semantic caching, cost optimization. OpenAI-compatible interface with Brokle extensions.

Installation

pip install brokle

Setup

export BROKLE_API_KEY="bk_your_api_key_here"
export BROKLE_HOST="http://localhost:8080"

Quick Start

Pattern 1: Wrapper Functions

# Wrap existing SDK clients for automatic observability
from openai import OpenAI
from anthropic import Anthropic
from brokle import wrap_openai, wrap_anthropic

# OpenAI wrapper
openai_client = wrap_openai(
    OpenAI(api_key="sk-..."),
    tags=["production"],
    session_id="user_session_123"
)

# Anthropic wrapper
anthropic_client = wrap_anthropic(
    Anthropic(api_key="sk-ant-..."),
    tags=["claude", "analysis"]
)

response = openai_client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Hello!"}]
)
# ✅ Automatic Brokle observability, routing, caching, optimization

Pattern 2: Universal Decorator

# Automatic hierarchical tracing with just @observe()
from brokle import observe
import openai

client = openai.OpenAI()

@observe(name="parent-workflow")
def main_workflow(data: str):
    # Parent span automatically created
    result1 = analyze_data(data)
    result2 = summarize_findings(result1)
    return f"Final result: {result1} -> {result2}"

@observe(name="data-analysis")
def analyze_data(data: str):
    # Child span automatically linked to parent
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": f"Analyze: {data}"}]
    )
    return response.choices[0].message.content

@observe(name="summarization")
def summarize_findings(analysis: str):
    # Another child span automatically linked to parent
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": f"Summarize: {analysis}"}]
    )
    return response.choices[0].message.content

# Automatic hierarchical tracing - no manual workflow management needed
result = main_workflow("User behavior data from Q4 2024")
# ✅ Complete span hierarchy: parent -> analyze_data + summarize_findings

Pattern 3: Native SDK

Sync Client:

from brokle import Brokle

# Context manager (recommended)
with Brokle(
    api_key="bk_...",
    host="http://localhost:8080"
) as client:
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": "Hello!"}],
        routing_strategy="cost_optimized",  # Brokle extension
        cache_strategy="semantic"           # Brokle extension
    )
    print(f"Response: {response.choices[0].message.content}")

Async Client:

from brokle import AsyncBrokle
import asyncio

async def main():
    async with AsyncBrokle(
        api_key="bk_...",
    ) as client:
        response = await client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": "Hello!"}],
            routing_strategy="cost_optimized",  # Smart routing
            cache_strategy="semantic",          # Semantic caching
            tags=["async", "production"]       # Analytics tags
        )
        print(f"Response: {response.choices[0].message.content}")

asyncio.run(main())

Why Choose Brokle?

  • 🚀 30-50% Cost Reduction: Intelligent routing and semantic caching
  • ⚡ <3ms Overhead: High-performance observability
  • 🔄 250+ Providers: Route across all major LLM providers
  • 📊 Complete Visibility: Real-time analytics and quality scoring
  • 🛠️ Three Patterns: Start simple, scale as needed

Next Steps

Examples

Check the examples/ directory:

Support


Simple. Powerful. Three patterns to fit your needs.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

brokle-0.2.8.tar.gz (122.4 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

brokle-0.2.8-py3-none-any.whl (109.9 kB view details)

Uploaded Python 3

File details

Details for the file brokle-0.2.8.tar.gz.

File metadata

  • Download URL: brokle-0.2.8.tar.gz
  • Upload date:
  • Size: 122.4 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.11.13

File hashes

Hashes for brokle-0.2.8.tar.gz
Algorithm Hash digest
SHA256 97338dd41bb5e9315a49f029cd7ec6a6220320942288b1bfad5230f0a5c7ce12
MD5 f5f5a11a49dc3f970ba259ae4beee5f4
BLAKE2b-256 3b3590d2caa1044ffbabf76d78a37a509b353d4ca155f81623df55c567e7a6a4

See more details on using hashes here.

File details

Details for the file brokle-0.2.8-py3-none-any.whl.

File metadata

  • Download URL: brokle-0.2.8-py3-none-any.whl
  • Upload date:
  • Size: 109.9 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.11.13

File hashes

Hashes for brokle-0.2.8-py3-none-any.whl
Algorithm Hash digest
SHA256 ee4dd228526131dea537653d2ee14e94d969ddc2d44411e79326c22f6e390a49
MD5 8d1757e2dcdbe3a41f49cab54bad895e
BLAKE2b-256 3cb187dcd88a5a9b7ebe106c8fef9f4da1a0c6ddda1c8419eae41ac89889558b

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page