LiteLLM

pip install netra-sdk litellm

from netra import Netra
from litellm import completion
import os

# Initialize Netra
Netra.init(
    headers=f"x-api-key={os.environ.get('NETRA_API_KEY')}",
    trace_content=True
)

# Use LiteLLM as normal - automatically traced
response = completion(
    model="gpt-3.5-turbo",
    messages=[{"role": "user", "content": "Hello!"}]
)
print(response.choices[0].message.content)

from netra import task, SpanWrapper
from litellm import completion

@task()
def call_multiple_providers(prompt: str) -> dict:
    span = SpanWrapper("multi-provider", {
        "prompt": prompt
    }).start()
    
    results = {}
    
    # OpenAI
    openai_response = completion(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}]
    )
    results["openai"] = openai_response.choices[0].message.content
    
    # Anthropic
    anthropic_response = completion(
        model="claude-3-sonnet-20240229",
        messages=[{"role": "user", "content": prompt}]
    )
    results["anthropic"] = anthropic_response.choices[0].message.content
    
    span.set_attribute("providers.count", len(results))
    span.end()
    
    return results

from netra import workflow, SpanWrapper
from litellm import completion

@workflow()
def efficient_completion(messages: list[dict], model: str = "gpt-3.5-turbo"):
    span = SpanWrapper("litellm-completion", {
        "model": model,
        "messages.count": len(messages)
    }).start()
    
    response = completion(
        model=model,
        messages=messages,
        timeout=30
    )
    
    content = response.choices[0].message.content
    span.set_attribute("response", content)
    span.set_attribute("tokens.total", response.usage.total_tokens)
    span.end()
    
    return content

from netra import agent, SpanWrapper
from litellm import completion

@agent()
def completion_with_fallback(prompt: str, models: list[str]):
    span = SpanWrapper("fallback-completion", {
        "prompt": prompt,
        "models": ",".join(models)
    }).start()
    
    for model in models:
        try:
            response = completion(
                model=model,
                messages=[{"role": "user", "content": prompt}],
                timeout=10
            )
            
            span.set_attribute("successful_model", model)
            span.end()
            
            return response.choices[0].message.content
        except Exception as e:
            span.set_attribute(f"error.{model}", str(e))
            continue
    
    span.set_status(code=1, message="All models failed")
    span.end()
    raise Exception("All models failed")

from netra import task, SpanWrapper
from litellm import completion

@task()
def stream_any_provider(model: str, prompt: str):
    stream_span = SpanWrapper("litellm-stream", {
        "model": model
    }).start()
    
    response = completion(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        stream=True
    )
    
    full_response = ""
    for chunk in response:
        if chunk.choices[0].delta.content:
            content = chunk.choices[0].delta.content
            print(content, end='', flush=True)
            full_response += content
    
    stream_span.set_attribute("response", full_response)
    stream_span.end()
    
    return full_response

from netra import task, SpanWrapper
from litellm import acompletion
import asyncio

@task()
async def async_completion(prompt: str):
    span = SpanWrapper("async-completion").start()
    
    response = await acompletion(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}]
    )
    
    content = response.choices[0].message.content
    span.set_attribute("response", content)
    span.end()
    
    return content

from netra import Netra
from netra.instrumentation.instruments import InstrumentSet

Netra.init(
    headers=f"x-api-key={os.environ.get('NETRA_API_KEY')}",
    trace_content=True,
    instruments={InstrumentSet.LITELLM}
)