
Installation
Install both the Netra SDK and LiteLLM:pip install netra-sdk litellm
npm install netra-sdk litellm
Usage
Initialize the Netra SDK to automatically trace all LiteLLM operations: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)
import { Netra } from "netra-sdk";
import { completion } from "litellm";
// Initialize Netra
await Netra.init({
headers: `x-api-key=${process.env.NETRA_API_KEY}`,
traceContent: true
});
// Use LiteLLM as normal - automatically traced
const response = await completion({
model: "gpt-3.5-turbo",
messages: [{ role: "user", content: "Hello!" }]
});
console.log(response.choices[0].message.content);
LiteLLM Features
Trace completions across different providers:from netra.decorators import task
from netra import 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
import { task, SpanWrapper } from "netra-sdk";
import { completion } from "litellm";
@task()
async function callMultipleProviders(prompt: string): Promise<Record<string, string>> {
const span = new SpanWrapper("multi-provider", {
"prompt": prompt
}).start();
const results: Record<string, string> = {};
// OpenAI
const openaiResponse = await completion({
model: "gpt-3.5-turbo",
messages: [{ role: "user", content: prompt }]
});
results["openai"] = openaiResponse.choices[0].message.content;
// Anthropic
const anthropicResponse = await completion({
model: "claude-3-sonnet-20240229",
messages: [{ role: "user", content: prompt }]
});
results["anthropic"] = anthropicResponse.choices[0].message.content;
span.setAttribute("providers.count", Object.keys(results).length);
span.end();
return results;
}
Lightweight Design
Trace with minimal overhead:from netra.decorators import workflow
from netra import 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
import { workflow, SpanWrapper } from "netra-sdk";
import { completion } from "litellm";
@workflow()
async function efficientCompletion(messages: Array<any>, model: string = "gpt-3.5-turbo") {
const span = new SpanWrapper("litellm-completion", {
"model": model,
"messages.count": messages.length
}).start();
const response = await completion({
model: model,
messages: messages,
timeout: 30
});
const content = response.choices[0].message.content;
span.setAttribute("response", content);
span.setAttribute("tokens.total", response.usage?.total_tokens || 0);
span.end();
return content;
}
Modular Architecture
Trace with custom configurations:from netra.decorators import agent
from netra import 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")
import { agent, SpanWrapper } from "netra-sdk";
import { completion } from "litellm";
@agent()
async function completionWithFallback(prompt: string, models: string[]) {
const span = new SpanWrapper("fallback-completion", {
"prompt": prompt,
"models": models.join(",")
}).start();
for (const model of models) {
try {
const response = await completion({
model: model,
messages: [{ role: "user", content: prompt }],
timeout: 10
});
span.setAttribute("successful_model", model);
span.end();
return response.choices[0].message.content;
} catch (error) {
span.setAttribute(`error.${model}`, String(error));
continue;
}
}
span.setStatus({ code: 1, message: "All models failed" });
span.end();
throw new Error("All models failed");
}
Cross-Platform Support
Trace streaming across platforms:from netra.decorators import task
from netra import 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
import { task, SpanWrapper } from "netra-sdk";
import { completion } from "litellm";
@task()
async function streamAnyProvider(model: string, prompt: string) {
const streamSpan = new SpanWrapper("litellm-stream", {
"model": model
}).start();
const response = await completion({
model: model,
messages: [{ role: "user", content: prompt }],
stream: true
});
let fullResponse = "";
for await (const chunk of response) {
if (chunk.choices[0]?.delta?.content) {
const content = chunk.choices[0].delta.content;
process.stdout.write(content);
fullResponse += content;
}
}
streamSpan.setAttribute("response", fullResponse);
streamSpan.end();
return fullResponse;
}
Async Support
Trace async completions:from netra.decorators import task
from netra import 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
import { task, SpanWrapper } from "netra-sdk";
import { completion } from "litellm";
@task()
async function asyncCompletion(prompt: string) {
const span = new SpanWrapper("async-completion").start();
const response = await completion({
model: "gpt-3.5-turbo",
messages: [{ role: "user", content: prompt }]
});
const content = response.choices[0].message.content;
span.setAttribute("response", content);
span.end();
return content;
}
Configuration
Configure LiteLLM instrumentation: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}
)
import { Netra, NetraInstruments } from "netra-sdk";
await Netra.init({
headers: `x-api-key=${process.env.NETRA_API_KEY}`,
traceContent: true,
instruments: new Set([NetraInstruments.LITELLM])
});
Next Steps
- Quick Start Guide - Complete setup and configuration
- Decorators - Add custom tracing with
@workflow,@agent, and@taskdecorators - LiteLLM Documentation - Official LiteLLM documentation
