mirror of
https://github.com/tiennm99/litellm.git
synced 2026-06-18 00:48:01 +00:00
feat: Add Levo AI integration (#18529)
This commit is contained in:
amangupta-20
@@ -0,0 +1,162 @@
|
||||
---
|
||||
sidebar_label: Levo AI
|
||||
---
|
||||
|
||||
import Image from '@theme/IdealImage';
|
||||
import Tabs from '@theme/Tabs';
|
||||
import TabItem from '@theme/TabItem';
|
||||
|
||||
# Levo AI
|
||||
|
||||
<div className="levo-logo-container" style={{ marginTop: '0.5rem', marginBottom: '1rem' }}>
|
||||
<div className="levo-logo-light">
|
||||
<Image img={require('../../img/levo_logo.png')} />
|
||||
</div>
|
||||
<div className="levo-logo-dark">
|
||||
<Image img={require('../../img/levo_logo_dark.png')} />
|
||||
</div>
|
||||
</div>
|
||||
|
||||
[Levo](https://levo.ai/) is an AI observability and compliance platform that provides comprehensive monitoring, analysis, and compliance tracking for LLM applications.
|
||||
|
||||
## Quick Start
|
||||
|
||||
Send all your LLM requests and responses to Levo for monitoring and analysis using LiteLLM's built-in Levo integration.
|
||||
|
||||
### What You'll Get
|
||||
|
||||
- **Complete visibility** into all LLM API calls across all providers
|
||||
- **Request and response data** including prompts, completions, and metadata
|
||||
- **Usage and cost tracking** with token counts and cost breakdowns
|
||||
- **Error monitoring** and performance metrics
|
||||
- **Compliance tracking** for audit and governance
|
||||
|
||||
### Setup Steps
|
||||
|
||||
**1. Install OpenTelemetry dependencies:**
|
||||
|
||||
```bash
|
||||
pip install opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp-proto-http opentelemetry-exporter-otlp-proto-grpc
|
||||
```
|
||||
|
||||
**2. Enable Levo callback in your LiteLLM config:**
|
||||
|
||||
Add to your `litellm_config.yaml`:
|
||||
|
||||
```yaml
|
||||
litellm_settings:
|
||||
callbacks: ["levo"]
|
||||
```
|
||||
|
||||
**3. Configure environment variables:**
|
||||
|
||||
[Contact Levo support](mailto:support@levo.ai) to get your collector endpoint URL, API key, organization ID, and workspace ID.
|
||||
|
||||
Set these required environment variables:
|
||||
|
||||
```bash
|
||||
export LEVOAI_API_KEY="<your-levo-api-key>"
|
||||
export LEVOAI_ORG_ID="<your-levo-org-id>"
|
||||
export LEVOAI_WORKSPACE_ID="<your-workspace-id>"
|
||||
export LEVOAI_COLLECTOR_URL="<your-levo-collector-url>"
|
||||
```
|
||||
|
||||
**Note:** The collector URL should be the full endpoint URL provided by Levo support. It will be used exactly as provided.
|
||||
|
||||
**4. Start LiteLLM:**
|
||||
|
||||
```bash
|
||||
litellm --config config.yaml
|
||||
```
|
||||
|
||||
**5. Make requests - they'll automatically be sent to Levo!**
|
||||
|
||||
```bash
|
||||
curl --location 'http://0.0.0.0:4000/chat/completions' \
|
||||
--header 'Content-Type: application/json' \
|
||||
--data '{
|
||||
"model": "gpt-3.5-turbo",
|
||||
"messages": [
|
||||
{
|
||||
"role": "user",
|
||||
"content": "Hello, this is a test message"
|
||||
}
|
||||
]
|
||||
}'
|
||||
```
|
||||
|
||||
## What Data is Captured
|
||||
|
||||
| Feature | Details |
|
||||
|---------|---------|
|
||||
| **What is logged** | OpenTelemetry Trace Data (OTLP format) |
|
||||
| **Events** | Success + Failure |
|
||||
| **Format** | OTLP (OpenTelemetry Protocol) |
|
||||
| **Headers** | Automatically includes `Authorization: Bearer {LEVOAI_API_KEY}`, `x-levo-organization-id`, and `x-levo-workspace-id` |
|
||||
|
||||
## Configuration Reference
|
||||
|
||||
### Required Environment Variables
|
||||
|
||||
| Variable | Description | Example |
|
||||
|----------|-------------|---------|
|
||||
| `LEVOAI_API_KEY` | Your Levo API key | `levo_abc123...` |
|
||||
| `LEVOAI_ORG_ID` | Your Levo organization ID | `org-123456` |
|
||||
| `LEVOAI_WORKSPACE_ID` | Your Levo workspace ID | `workspace-789` |
|
||||
| `LEVOAI_COLLECTOR_URL` | Full collector endpoint URL from Levo support | `https://collector.levo.ai/v1/traces` |
|
||||
|
||||
### Optional Environment Variables
|
||||
|
||||
| Variable | Description | Default |
|
||||
|----------|-------------|---------|
|
||||
| `LEVOAI_ENV_NAME` | Environment name for tagging traces | `None` |
|
||||
|
||||
**Note:** The collector URL is used exactly as provided by Levo support. No path manipulation is performed.
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
### Not seeing traces in Levo?
|
||||
|
||||
1. **Verify Levo callback is enabled**: Check LiteLLM startup logs for `initializing callbacks=['levo']`
|
||||
|
||||
2. **Check required environment variables**: Ensure all required variables are set:
|
||||
```bash
|
||||
echo $LEVOAI_API_KEY
|
||||
echo $LEVOAI_ORG_ID
|
||||
echo $LEVOAI_WORKSPACE_ID
|
||||
echo $LEVOAI_COLLECTOR_URL
|
||||
```
|
||||
|
||||
3. **Verify collector connectivity**: Test if your collector is reachable:
|
||||
```bash
|
||||
curl <your-collector-url>/health
|
||||
```
|
||||
|
||||
4. **Check for initialization errors**: Look for errors in LiteLLM startup logs. Common issues:
|
||||
- Missing OpenTelemetry packages: Install with `pip install opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp-proto-http opentelemetry-exporter-otlp-proto-grpc`
|
||||
- Missing required environment variables: All four required variables must be set
|
||||
- Invalid collector URL: Ensure the URL is correct and reachable
|
||||
|
||||
5. **Enable debug logging**:
|
||||
```bash
|
||||
export LITELLM_LOG="DEBUG"
|
||||
```
|
||||
|
||||
6. **Wait for async export**: OTLP sends traces asynchronously. Wait 10-15 seconds after making requests before checking Levo.
|
||||
|
||||
### Common Errors
|
||||
|
||||
**Error: "LEVOAI_COLLECTOR_URL environment variable is required"**
|
||||
- Solution: Set the `LEVOAI_COLLECTOR_URL` environment variable with your collector endpoint URL from Levo support.
|
||||
|
||||
**Error: "No module named 'opentelemetry'"**
|
||||
- Solution: Install OpenTelemetry packages: `pip install opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp-proto-http opentelemetry-exporter-otlp-proto-grpc`
|
||||
|
||||
## Additional Resources
|
||||
|
||||
- [Levo Documentation](https://docs.levo.ai)
|
||||
- [OpenTelemetry Specification](https://opentelemetry.io/docs/specs/otel/)
|
||||
|
||||
## Need Help?
|
||||
|
||||
For issues or questions about the Levo integration with LiteLLM, please [contact Levo support](mailto:support@levo.ai) or open an issue on the [LiteLLM GitHub repository](https://github.com/BerriAI/litellm/issues).
|
||||
@@ -4,7 +4,7 @@ import TabItem from '@theme/TabItem';
|
||||
|
||||
# OpenTelemetry - Tracing LLMs with any observability tool
|
||||
|
||||
OpenTelemetry is a CNCF standard for observability. It connects to any observability tool, such as Jaeger, Zipkin, Datadog, New Relic, Traceloop and others.
|
||||
OpenTelemetry is a CNCF standard for observability. It connects to any observability tool, such as Jaeger, Zipkin, Datadog, New Relic, Traceloop, Levo AI and others.
|
||||
|
||||
<Image img={require('../../img/traceloop_dash.png')} />
|
||||
|
||||
|
||||
Binary file not shown.
|
After Width: | Height: | Size: 12 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 11 KiB |
@@ -28,3 +28,34 @@
|
||||
--ifm-color-primary-lightest: #4fddbf;
|
||||
--docusaurus-highlighted-code-line-bg: rgba(0, 0, 0, 0.3);
|
||||
}
|
||||
|
||||
/* Levo logo sizing and theme switching */
|
||||
.levo-logo-container {
|
||||
position: relative;
|
||||
}
|
||||
|
||||
.levo-logo-container img,
|
||||
.levo-logo-container picture,
|
||||
.levo-logo-container .ideal-image {
|
||||
max-width: 200px !important;
|
||||
width: 200px !important;
|
||||
height: auto !important;
|
||||
}
|
||||
|
||||
/* Show light logo by default, hide dark logo */
|
||||
.levo-logo-dark {
|
||||
display: none !important;
|
||||
}
|
||||
|
||||
.levo-logo-light {
|
||||
display: block !important;
|
||||
}
|
||||
|
||||
/* In dark mode, hide light logo and show dark logo */
|
||||
[data-theme='dark'] .levo-logo-light {
|
||||
display: none !important;
|
||||
}
|
||||
|
||||
[data-theme='dark'] .levo-logo-dark {
|
||||
display: block !important;
|
||||
}
|
||||
|
||||
@@ -136,6 +136,7 @@ _custom_logger_compatible_callbacks_literal = Literal[
|
||||
"gitlab",
|
||||
"cloudzero",
|
||||
"posthog",
|
||||
"levo",
|
||||
]
|
||||
cold_storage_custom_logger: Optional[_custom_logger_compatible_callbacks_literal] = None
|
||||
logged_real_time_event_types: Optional[Union[List[str], Literal["*"]]] = None
|
||||
|
||||
@@ -0,0 +1,125 @@
|
||||
# Levo AI Integration
|
||||
|
||||
This integration enables sending LLM observability data to Levo AI using OpenTelemetry (OTLP) protocol.
|
||||
|
||||
## Overview
|
||||
|
||||
The Levo integration extends LiteLLM's OpenTelemetry support to automatically send traces to Levo's collector endpoint with proper authentication and routing headers.
|
||||
|
||||
## Features
|
||||
|
||||
- **Automatic OTLP Export**: Sends OpenTelemetry traces to Levo collector
|
||||
- **Levo-Specific Headers**: Automatically includes `x-levo-organization-id` and `x-levo-workspace-id` for routing
|
||||
- **Simple Configuration**: Just use `callbacks: ["levo"]` in your LiteLLM config
|
||||
- **Environment-Based Setup**: Configure via environment variables
|
||||
|
||||
## Quick Start
|
||||
|
||||
### 1. Install Dependencies
|
||||
|
||||
```bash
|
||||
pip install opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp-proto-http opentelemetry-exporter-otlp-proto-grpc
|
||||
```
|
||||
|
||||
### 2. Configure LiteLLM
|
||||
|
||||
Add to your `litellm_config.yaml`:
|
||||
|
||||
```yaml
|
||||
litellm_settings:
|
||||
callbacks: ["levo"]
|
||||
```
|
||||
|
||||
### 3. Set Environment Variables
|
||||
|
||||
```bash
|
||||
export LEVOAI_API_KEY="<your-levo-api-key>"
|
||||
export LEVOAI_ORG_ID="<your-levo-org-id>"
|
||||
export LEVOAI_WORKSPACE_ID="<your-workspace-id>"
|
||||
export LEVOAI_COLLECTOR_URL="<your-levo-collector-url>"
|
||||
```
|
||||
|
||||
### 4. Start LiteLLM
|
||||
|
||||
```bash
|
||||
litellm --config config.yaml
|
||||
```
|
||||
|
||||
All LLM requests will now automatically be sent to Levo!
|
||||
|
||||
## Configuration
|
||||
|
||||
### Required Environment Variables
|
||||
|
||||
| Variable | Description |
|
||||
|----------|-------------|
|
||||
| `LEVOAI_API_KEY` | Your Levo API key for authentication |
|
||||
| `LEVOAI_ORG_ID` | Your Levo organization ID for routing |
|
||||
| `LEVOAI_WORKSPACE_ID` | Your Levo workspace ID for routing |
|
||||
| `LEVOAI_COLLECTOR_URL` | Full collector endpoint URL from Levo support |
|
||||
|
||||
### Optional Environment Variables
|
||||
|
||||
| Variable | Description | Default |
|
||||
|----------|-------------|---------|
|
||||
| `LEVOAI_ENV_NAME` | Environment name for tagging traces | `None` |
|
||||
|
||||
**Important**: The `LEVOAI_COLLECTOR_URL` is used exactly as provided. No path manipulation is performed.
|
||||
|
||||
## How It Works
|
||||
|
||||
1. **LevoLogger** extends LiteLLM's `OpenTelemetry` class
|
||||
2. **Configuration** is read from environment variables via `get_levo_config()`
|
||||
3. **OTLP Headers** are automatically set:
|
||||
- `Authorization: Bearer {LEVOAI_API_KEY}`
|
||||
- `x-levo-organization-id: {LEVOAI_ORG_ID}`
|
||||
- `x-levo-workspace-id: {LEVOAI_WORKSPACE_ID}`
|
||||
4. **Traces** are sent to the collector endpoint in OTLP format
|
||||
|
||||
## Code Structure
|
||||
|
||||
```
|
||||
litellm/integrations/levo/
|
||||
├── __init__.py # Exports LevoLogger
|
||||
├── levo.py # LevoLogger implementation
|
||||
└── README.md # This file
|
||||
```
|
||||
|
||||
### Key Classes
|
||||
|
||||
- **LevoLogger**: Extends `OpenTelemetry`, handles Levo-specific configuration
|
||||
- **LevoConfig**: Pydantic model for Levo configuration (defined in `levo.py`)
|
||||
|
||||
## Testing
|
||||
|
||||
See the test files in `tests/test_litellm/integrations/levo/`:
|
||||
- `test_levo.py`: Unit tests for configuration
|
||||
- `test_levo_integration.py`: Integration tests for callback registration
|
||||
|
||||
## Error Handling
|
||||
|
||||
The integration validates all required environment variables at initialization:
|
||||
- Missing `LEVOAI_API_KEY`: Raises `ValueError` with clear message
|
||||
- Missing `LEVOAI_ORG_ID`: Raises `ValueError` with clear message
|
||||
- Missing `LEVOAI_WORKSPACE_ID`: Raises `ValueError` with clear message
|
||||
- Missing `LEVOAI_COLLECTOR_URL`: Raises `ValueError` with clear message
|
||||
|
||||
## Integration with LiteLLM
|
||||
|
||||
The Levo callback is registered in:
|
||||
- `litellm/litellm_core_utils/custom_logger_registry.py`: Maps `"levo"` to `LevoLogger`
|
||||
- `litellm/litellm_core_utils/litellm_logging.py`: Instantiates `LevoLogger` when `callbacks: ["levo"]` is used
|
||||
- `litellm/__init__.py`: Added to `_custom_logger_compatible_callbacks_literal`
|
||||
|
||||
## Documentation
|
||||
|
||||
For detailed documentation, see:
|
||||
- [LiteLLM Levo Integration Docs](../../../../docs/my-website/docs/observability/levo_integration.md)
|
||||
- [Levo Documentation](https://docs.levo.ai)
|
||||
|
||||
## Support
|
||||
|
||||
For issues or questions:
|
||||
- LiteLLM Issues: https://github.com/BerriAI/litellm/issues
|
||||
- Levo Support: support@levo.ai
|
||||
|
||||
@@ -0,0 +1,3 @@
|
||||
from litellm.integrations.levo.levo import LevoLogger
|
||||
|
||||
__all__ = ["LevoLogger"]
|
||||
@@ -0,0 +1,117 @@
|
||||
import os
|
||||
from typing import TYPE_CHECKING, Any, Optional, Union
|
||||
|
||||
from litellm.integrations.opentelemetry import OpenTelemetry
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from opentelemetry.trace import Span as _Span
|
||||
|
||||
from litellm.integrations.opentelemetry import OpenTelemetryConfig as _OpenTelemetryConfig
|
||||
from litellm.types.integrations.arize import Protocol as _Protocol
|
||||
|
||||
Protocol = _Protocol
|
||||
OpenTelemetryConfig = _OpenTelemetryConfig
|
||||
Span = Union[_Span, Any]
|
||||
else:
|
||||
Protocol = Any
|
||||
OpenTelemetryConfig = Any
|
||||
Span = Any
|
||||
|
||||
|
||||
class LevoConfig:
|
||||
"""Configuration for Levo OTLP integration."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
otlp_auth_headers: Optional[str],
|
||||
protocol: Protocol,
|
||||
endpoint: str,
|
||||
):
|
||||
self.otlp_auth_headers = otlp_auth_headers
|
||||
self.protocol = protocol
|
||||
self.endpoint = endpoint
|
||||
|
||||
|
||||
class LevoLogger(OpenTelemetry):
|
||||
"""Levo Logger that extends OpenTelemetry for OTLP integration."""
|
||||
|
||||
@staticmethod
|
||||
def get_levo_config() -> LevoConfig:
|
||||
"""
|
||||
Retrieves the Levo configuration based on environment variables.
|
||||
|
||||
Returns:
|
||||
LevoConfig: Configuration object containing Levo OTLP settings.
|
||||
|
||||
Raises:
|
||||
ValueError: If required environment variables are missing.
|
||||
"""
|
||||
# Required environment variables
|
||||
api_key = os.environ.get("LEVOAI_API_KEY", None)
|
||||
org_id = os.environ.get("LEVOAI_ORG_ID", None)
|
||||
workspace_id = os.environ.get("LEVOAI_WORKSPACE_ID", None)
|
||||
collector_url = os.environ.get("LEVOAI_COLLECTOR_URL", None)
|
||||
|
||||
# Validate required env vars
|
||||
if not api_key:
|
||||
raise ValueError(
|
||||
"LEVOAI_API_KEY environment variable is required for Levo integration."
|
||||
)
|
||||
if not org_id:
|
||||
raise ValueError(
|
||||
"LEVOAI_ORG_ID environment variable is required for Levo integration."
|
||||
)
|
||||
if not workspace_id:
|
||||
raise ValueError(
|
||||
"LEVOAI_WORKSPACE_ID environment variable is required for Levo integration."
|
||||
)
|
||||
if not collector_url:
|
||||
raise ValueError(
|
||||
"LEVOAI_COLLECTOR_URL environment variable is required for Levo integration. "
|
||||
"Please contact Levo support to get your collector URL."
|
||||
)
|
||||
|
||||
# Use collector URL exactly as provided by the user
|
||||
endpoint = collector_url
|
||||
protocol: Protocol = "otlp_http"
|
||||
|
||||
# Build OTLP headers string
|
||||
# Format: Authorization=Bearer {api_key},x-levo-organization-id={org_id},x-levo-workspace-id={workspace_id}
|
||||
headers_parts = [f"Authorization=Bearer {api_key}"]
|
||||
headers_parts.append(f"x-levo-organization-id={org_id}")
|
||||
headers_parts.append(f"x-levo-workspace-id={workspace_id}")
|
||||
|
||||
otlp_auth_headers = ",".join(headers_parts)
|
||||
|
||||
return LevoConfig(
|
||||
otlp_auth_headers=otlp_auth_headers,
|
||||
protocol=protocol,
|
||||
endpoint=endpoint,
|
||||
)
|
||||
|
||||
async def async_health_check(self):
|
||||
"""
|
||||
Health check for Levo integration.
|
||||
|
||||
Returns:
|
||||
dict: Health status with status and message/error_message keys.
|
||||
"""
|
||||
try:
|
||||
config = self.get_levo_config()
|
||||
|
||||
if not config.otlp_auth_headers:
|
||||
return {
|
||||
"status": "unhealthy",
|
||||
"error_message": "LEVOAI_API_KEY environment variable not set",
|
||||
}
|
||||
|
||||
return {
|
||||
"status": "healthy",
|
||||
"message": "Levo credentials are configured properly",
|
||||
}
|
||||
except ValueError as e:
|
||||
return {
|
||||
"status": "unhealthy",
|
||||
"error_message": str(e),
|
||||
}
|
||||
|
||||
@@ -76,6 +76,7 @@ class CustomLoggerRegistry:
|
||||
"arize_phoenix": OpenTelemetry,
|
||||
"langtrace": OpenTelemetry,
|
||||
"weave_otel": OpenTelemetry,
|
||||
"levo": OpenTelemetry,
|
||||
"mlflow": MlflowLogger,
|
||||
"langfuse": LangfusePromptManagement,
|
||||
"otel": OpenTelemetry,
|
||||
|
||||
@@ -3699,6 +3699,31 @@ def _init_custom_logger_compatible_class( # noqa: PLR0915
|
||||
)
|
||||
_in_memory_loggers.append(_arize_phoenix_otel_logger)
|
||||
return _arize_phoenix_otel_logger # type: ignore
|
||||
elif logging_integration == "levo":
|
||||
from litellm.integrations.levo.levo import LevoLogger
|
||||
from litellm.integrations.opentelemetry import (
|
||||
OpenTelemetry,
|
||||
OpenTelemetryConfig,
|
||||
)
|
||||
|
||||
levo_config = LevoLogger.get_levo_config()
|
||||
otel_config = OpenTelemetryConfig(
|
||||
exporter=levo_config.protocol,
|
||||
endpoint=levo_config.endpoint,
|
||||
headers=levo_config.otlp_auth_headers,
|
||||
)
|
||||
|
||||
# Check if LevoLogger instance already exists
|
||||
for callback in _in_memory_loggers:
|
||||
if (
|
||||
isinstance(callback, LevoLogger)
|
||||
and callback.callback_name == "levo"
|
||||
):
|
||||
return callback # type: ignore
|
||||
|
||||
_levo_otel_logger = LevoLogger(config=otel_config, callback_name="levo")
|
||||
_in_memory_loggers.append(_levo_otel_logger)
|
||||
return _levo_otel_logger # type: ignore
|
||||
elif logging_integration == "otel":
|
||||
from litellm.integrations.opentelemetry import OpenTelemetry
|
||||
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
# Levo integration tests
|
||||
@@ -0,0 +1,407 @@
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from litellm.integrations.levo.levo import LevoConfig, LevoLogger
|
||||
from litellm.integrations.opentelemetry import OpenTelemetryConfig
|
||||
|
||||
# Try to import OpenTelemetry packages, skip tests if not available
|
||||
try:
|
||||
from opentelemetry.sdk.trace import TracerProvider
|
||||
from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
|
||||
InMemorySpanExporter,
|
||||
)
|
||||
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
|
||||
|
||||
OPENTELEMETRY_AVAILABLE = True
|
||||
except ImportError:
|
||||
OPENTELEMETRY_AVAILABLE = False
|
||||
|
||||
|
||||
class TestLevoConfig(unittest.TestCase):
|
||||
"""Unit tests for LevoLogger configuration."""
|
||||
|
||||
@patch.dict(
|
||||
"os.environ",
|
||||
{
|
||||
"LEVOAI_API_KEY": "test-api-key",
|
||||
"LEVOAI_ORG_ID": "test-org-id",
|
||||
"LEVOAI_WORKSPACE_ID": "test-workspace-id",
|
||||
"LEVOAI_COLLECTOR_URL": "https://collector.levo.ai",
|
||||
},
|
||||
)
|
||||
def test_get_levo_config_with_all_required_vars(self):
|
||||
"""Test get_levo_config() with all required environment variables."""
|
||||
config = LevoLogger.get_levo_config()
|
||||
|
||||
# Verify headers include all three values
|
||||
self.assertIn("Authorization=Bearer test-api-key", config.otlp_auth_headers)
|
||||
self.assertIn("x-levo-organization-id=test-org-id", config.otlp_auth_headers)
|
||||
self.assertIn("x-levo-workspace-id=test-workspace-id", config.otlp_auth_headers)
|
||||
|
||||
# Verify endpoint uses provided collector URL exactly as-is
|
||||
self.assertEqual(config.endpoint, "https://collector.levo.ai")
|
||||
|
||||
# Verify protocol is otlp_http
|
||||
self.assertEqual(config.protocol, "otlp_http")
|
||||
|
||||
@patch.dict(
|
||||
"os.environ",
|
||||
{
|
||||
"LEVOAI_API_KEY": "test-api-key",
|
||||
"LEVOAI_ORG_ID": "test-org-id",
|
||||
"LEVOAI_WORKSPACE_ID": "test-workspace-id",
|
||||
"LEVOAI_COLLECTOR_URL": "https://custom.collector.com",
|
||||
},
|
||||
)
|
||||
def test_get_levo_config_with_custom_collector_url(self):
|
||||
"""Test get_levo_config() with custom collector URL."""
|
||||
config = LevoLogger.get_levo_config()
|
||||
|
||||
# Verify endpoint uses custom URL exactly as provided
|
||||
self.assertEqual(config.endpoint, "https://custom.collector.com")
|
||||
self.assertEqual(config.protocol, "otlp_http")
|
||||
|
||||
@patch.dict("os.environ", {}, clear=True)
|
||||
def test_get_levo_config_missing_api_key(self):
|
||||
"""Test get_levo_config() raises ValueError when LEVOAI_API_KEY is missing."""
|
||||
with pytest.raises(ValueError, match="LEVOAI_API_KEY"):
|
||||
LevoLogger.get_levo_config()
|
||||
|
||||
@patch.dict(
|
||||
"os.environ",
|
||||
{
|
||||
"LEVOAI_API_KEY": "test-api-key",
|
||||
},
|
||||
clear=True,
|
||||
)
|
||||
def test_get_levo_config_missing_org_id(self):
|
||||
"""Test get_levo_config() raises ValueError when LEVOAI_ORG_ID is missing."""
|
||||
with pytest.raises(ValueError, match="LEVOAI_ORG_ID"):
|
||||
LevoLogger.get_levo_config()
|
||||
|
||||
@patch.dict(
|
||||
"os.environ",
|
||||
{
|
||||
"LEVOAI_API_KEY": "test-api-key",
|
||||
"LEVOAI_ORG_ID": "test-org-id",
|
||||
},
|
||||
clear=True,
|
||||
)
|
||||
def test_get_levo_config_missing_workspace_id(self):
|
||||
"""Test get_levo_config() raises ValueError when LEVOAI_WORKSPACE_ID is missing."""
|
||||
with pytest.raises(ValueError, match="LEVOAI_WORKSPACE_ID"):
|
||||
LevoLogger.get_levo_config()
|
||||
|
||||
@patch.dict(
|
||||
"os.environ",
|
||||
{
|
||||
"LEVOAI_API_KEY": "test-api-key",
|
||||
"LEVOAI_ORG_ID": "test-org-id",
|
||||
"LEVOAI_WORKSPACE_ID": "test-workspace-id",
|
||||
},
|
||||
clear=True,
|
||||
)
|
||||
def test_get_levo_config_missing_collector_url(self):
|
||||
"""Test get_levo_config() raises ValueError when LEVOAI_COLLECTOR_URL is missing."""
|
||||
with pytest.raises(ValueError, match="LEVOAI_COLLECTOR_URL"):
|
||||
LevoLogger.get_levo_config()
|
||||
|
||||
@patch.dict(
|
||||
"os.environ",
|
||||
{
|
||||
"LEVOAI_API_KEY": "test-api-key",
|
||||
"LEVOAI_ORG_ID": "test-org-id",
|
||||
"LEVOAI_WORKSPACE_ID": "test-workspace-id",
|
||||
"LEVOAI_COLLECTOR_URL": "http://localhost:4318",
|
||||
},
|
||||
)
|
||||
def test_get_levo_config_with_http_endpoint(self):
|
||||
"""Test get_levo_config() with HTTP endpoint."""
|
||||
config = LevoLogger.get_levo_config()
|
||||
|
||||
# Should use HTTP endpoint exactly as provided
|
||||
self.assertEqual(config.endpoint, "http://localhost:4318")
|
||||
self.assertEqual(config.protocol, "otlp_http")
|
||||
|
||||
@patch.dict(
|
||||
"os.environ",
|
||||
{
|
||||
"LEVOAI_API_KEY": "test-api-key",
|
||||
"LEVOAI_ORG_ID": "test-org-id",
|
||||
"LEVOAI_WORKSPACE_ID": "test-workspace-id",
|
||||
"LEVOAI_COLLECTOR_URL": "https://collector.levo.ai",
|
||||
},
|
||||
)
|
||||
def test_levo_config_headers_format(self):
|
||||
"""Test that OTLP headers are formatted correctly."""
|
||||
config = LevoLogger.get_levo_config()
|
||||
|
||||
# Verify headers contain all required parts
|
||||
self.assertIn("Authorization=Bearer test-api-key", config.otlp_auth_headers)
|
||||
self.assertIn("x-levo-organization-id=test-org-id", config.otlp_auth_headers)
|
||||
self.assertIn("x-levo-workspace-id=test-workspace-id", config.otlp_auth_headers)
|
||||
|
||||
# Verify headers are comma-separated
|
||||
header_parts = config.otlp_auth_headers.split(",")
|
||||
self.assertEqual(len(header_parts), 3)
|
||||
|
||||
|
||||
class TestLevoIntegration(unittest.TestCase):
|
||||
"""Integration tests for LevoLogger."""
|
||||
|
||||
@patch.dict(
|
||||
"os.environ",
|
||||
{
|
||||
"LEVOAI_API_KEY": "test-api-key",
|
||||
"LEVOAI_ORG_ID": "test-org-id",
|
||||
"LEVOAI_WORKSPACE_ID": "test-workspace-id",
|
||||
"LEVOAI_COLLECTOR_URL": "https://collector.levo.ai",
|
||||
},
|
||||
)
|
||||
@pytest.mark.skipif(
|
||||
not OPENTELEMETRY_AVAILABLE, reason="OpenTelemetry packages not installed"
|
||||
)
|
||||
@patch(
|
||||
"litellm.integrations.opentelemetry.OpenTelemetry._init_otel_logger_on_litellm_proxy"
|
||||
)
|
||||
def test_levo_logger_instantiation(self, mock_init_proxy):
|
||||
"""Test that LevoLogger can be instantiated with proper config."""
|
||||
# Mock the proxy initialization to avoid importing proxy code
|
||||
mock_init_proxy.return_value = None
|
||||
|
||||
config = LevoLogger.get_levo_config()
|
||||
otel_config = OpenTelemetryConfig(
|
||||
exporter=config.protocol,
|
||||
endpoint=config.endpoint,
|
||||
headers=config.otlp_auth_headers,
|
||||
)
|
||||
|
||||
# Create a tracer provider with in-memory exporter to avoid requiring OTLP packages
|
||||
tracer_provider = TracerProvider()
|
||||
tracer_provider.add_span_processor(SimpleSpanProcessor(InMemorySpanExporter()))
|
||||
|
||||
# Create LevoLogger instance with mocked tracer provider
|
||||
levo_logger = LevoLogger(
|
||||
config=otel_config, callback_name="levo", tracer_provider=tracer_provider
|
||||
)
|
||||
|
||||
# Verify it's an instance of OpenTelemetry
|
||||
self.assertIsInstance(levo_logger, LevoLogger)
|
||||
# Check it extends OpenTelemetry by checking base classes
|
||||
from litellm.integrations.opentelemetry import OpenTelemetry
|
||||
|
||||
self.assertIsInstance(levo_logger, OpenTelemetry)
|
||||
|
||||
# Verify callback_name is set
|
||||
self.assertEqual(levo_logger.callback_name, "levo")
|
||||
|
||||
@patch.dict(
|
||||
"os.environ",
|
||||
{
|
||||
"LEVOAI_API_KEY": "test-api-key",
|
||||
"LEVOAI_ORG_ID": "test-org-id",
|
||||
"LEVOAI_WORKSPACE_ID": "test-workspace-id",
|
||||
"LEVOAI_COLLECTOR_URL": "https://collector.levo.ai",
|
||||
},
|
||||
)
|
||||
@pytest.mark.skipif(
|
||||
not OPENTELEMETRY_AVAILABLE, reason="OpenTelemetry packages not installed"
|
||||
)
|
||||
@patch(
|
||||
"litellm.integrations.opentelemetry.OpenTelemetry._init_otel_logger_on_litellm_proxy"
|
||||
)
|
||||
@pytest.mark.asyncio
|
||||
async def test_levo_logger_health_check_healthy(self, mock_init_proxy):
|
||||
"""Test health check returns healthy status when config is valid."""
|
||||
# Mock the proxy initialization to avoid importing proxy code
|
||||
mock_init_proxy.return_value = None
|
||||
|
||||
config = LevoLogger.get_levo_config()
|
||||
otel_config = OpenTelemetryConfig(
|
||||
exporter=config.protocol,
|
||||
endpoint=config.endpoint,
|
||||
headers=config.otlp_auth_headers,
|
||||
)
|
||||
|
||||
# Create tracer provider with in-memory exporter
|
||||
tracer_provider = TracerProvider()
|
||||
tracer_provider.add_span_processor(SimpleSpanProcessor(InMemorySpanExporter()))
|
||||
|
||||
levo_logger = LevoLogger(
|
||||
config=otel_config, callback_name="levo", tracer_provider=tracer_provider
|
||||
)
|
||||
|
||||
# Run health check
|
||||
result = await levo_logger.async_health_check()
|
||||
|
||||
self.assertEqual(result["status"], "healthy")
|
||||
self.assertIn("message", result)
|
||||
|
||||
@patch.dict("os.environ", {}, clear=True)
|
||||
def test_levo_logger_health_check_unhealthy(self):
|
||||
"""Test health check returns unhealthy status when required vars are missing."""
|
||||
# Try to create logger without required env vars
|
||||
# This should fail during config, but we can test health check logic
|
||||
with pytest.raises(ValueError):
|
||||
LevoLogger.get_levo_config()
|
||||
|
||||
@patch.dict(
|
||||
"os.environ",
|
||||
{
|
||||
"LEVOAI_API_KEY": "test-api-key",
|
||||
"LEVOAI_ORG_ID": "test-org-id",
|
||||
"LEVOAI_WORKSPACE_ID": "test-workspace-id",
|
||||
"LEVOAI_COLLECTOR_URL": "https://collector.levo.ai",
|
||||
},
|
||||
)
|
||||
@pytest.mark.skipif(
|
||||
not OPENTELEMETRY_AVAILABLE, reason="OpenTelemetry packages not installed"
|
||||
)
|
||||
@patch(
|
||||
"litellm.integrations.opentelemetry.OpenTelemetry._init_otel_logger_on_litellm_proxy"
|
||||
)
|
||||
def test_levo_logger_callback_name(self, mock_init_proxy):
|
||||
"""Test that callback_name is properly set and used."""
|
||||
# Mock the proxy initialization to avoid importing proxy code
|
||||
mock_init_proxy.return_value = None
|
||||
|
||||
config = LevoLogger.get_levo_config()
|
||||
otel_config = OpenTelemetryConfig(
|
||||
exporter=config.protocol,
|
||||
endpoint=config.endpoint,
|
||||
headers=config.otlp_auth_headers,
|
||||
)
|
||||
|
||||
# Create tracer provider with in-memory exporter
|
||||
tracer_provider = TracerProvider()
|
||||
tracer_provider.add_span_processor(SimpleSpanProcessor(InMemorySpanExporter()))
|
||||
|
||||
levo_logger = LevoLogger(
|
||||
config=otel_config, callback_name="levo", tracer_provider=tracer_provider
|
||||
)
|
||||
|
||||
# Verify callback_name attribute
|
||||
self.assertEqual(levo_logger.callback_name, "levo")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"env_vars, expected_headers_contains, expected_endpoint, expected_protocol",
|
||||
[
|
||||
pytest.param(
|
||||
{
|
||||
"LEVOAI_API_KEY": "test-key",
|
||||
"LEVOAI_ORG_ID": "test-org",
|
||||
"LEVOAI_WORKSPACE_ID": "test-workspace",
|
||||
"LEVOAI_COLLECTOR_URL": "https://collector.levo.ai",
|
||||
},
|
||||
[
|
||||
"Authorization=Bearer test-key",
|
||||
"x-levo-organization-id=test-org",
|
||||
"x-levo-workspace-id=test-workspace",
|
||||
],
|
||||
"https://collector.levo.ai",
|
||||
"otlp_http",
|
||||
id="collector URL with all required vars",
|
||||
),
|
||||
pytest.param(
|
||||
{
|
||||
"LEVOAI_API_KEY": "key-123",
|
||||
"LEVOAI_ORG_ID": "org-456",
|
||||
"LEVOAI_WORKSPACE_ID": "workspace-789",
|
||||
"LEVOAI_COLLECTOR_URL": "https://custom.example.com",
|
||||
},
|
||||
[
|
||||
"Authorization=Bearer key-123",
|
||||
"x-levo-organization-id=org-456",
|
||||
"x-levo-workspace-id=workspace-789",
|
||||
],
|
||||
"https://custom.example.com",
|
||||
"otlp_http",
|
||||
id="custom collector URL",
|
||||
),
|
||||
pytest.param(
|
||||
{
|
||||
"LEVOAI_API_KEY": "key-123",
|
||||
"LEVOAI_ORG_ID": "org-456",
|
||||
"LEVOAI_WORKSPACE_ID": "workspace-789",
|
||||
"LEVOAI_COLLECTOR_URL": "http://localhost:9999",
|
||||
},
|
||||
["Authorization=Bearer key-123"],
|
||||
"http://localhost:9999",
|
||||
"otlp_http",
|
||||
id="custom HTTP endpoint",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_get_levo_config_parametrized(
|
||||
monkeypatch,
|
||||
env_vars,
|
||||
expected_headers_contains,
|
||||
expected_endpoint,
|
||||
expected_protocol,
|
||||
):
|
||||
"""Parametrized tests for get_levo_config() with various configurations."""
|
||||
# Clear all Levo-related env vars first to ensure clean state
|
||||
for key in [
|
||||
"LEVOAI_API_KEY",
|
||||
"LEVOAI_ORG_ID",
|
||||
"LEVOAI_WORKSPACE_ID",
|
||||
"LEVOAI_COLLECTOR_URL",
|
||||
"LEVOAI_ENV_NAME",
|
||||
]:
|
||||
monkeypatch.delenv(key, raising=False)
|
||||
|
||||
for key, value in env_vars.items():
|
||||
monkeypatch.setenv(key, value)
|
||||
|
||||
config = LevoLogger.get_levo_config()
|
||||
|
||||
assert isinstance(config, LevoConfig)
|
||||
assert config.endpoint == expected_endpoint
|
||||
assert config.protocol == expected_protocol
|
||||
|
||||
# Verify all expected header parts are present
|
||||
for header_part in expected_headers_contains:
|
||||
assert header_part in config.otlp_auth_headers
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"missing_var",
|
||||
[
|
||||
pytest.param("LEVOAI_API_KEY", id="missing API key"),
|
||||
pytest.param("LEVOAI_ORG_ID", id="missing org ID"),
|
||||
pytest.param("LEVOAI_WORKSPACE_ID", id="missing workspace ID"),
|
||||
pytest.param("LEVOAI_COLLECTOR_URL", id="missing collector URL"),
|
||||
],
|
||||
)
|
||||
def test_get_levo_config_missing_required_vars(monkeypatch, missing_var):
|
||||
"""Test that missing required environment variables raise ValueError."""
|
||||
# Clear all Levo-related env vars
|
||||
for key in [
|
||||
"LEVOAI_API_KEY",
|
||||
"LEVOAI_ORG_ID",
|
||||
"LEVOAI_WORKSPACE_ID",
|
||||
"LEVOAI_COLLECTOR_URL",
|
||||
]:
|
||||
monkeypatch.delenv(key, raising=False)
|
||||
|
||||
# Set all required vars except the missing one
|
||||
required_vars = {
|
||||
"LEVOAI_API_KEY": "test-key",
|
||||
"LEVOAI_ORG_ID": "test-org",
|
||||
"LEVOAI_WORKSPACE_ID": "test-workspace",
|
||||
"LEVOAI_COLLECTOR_URL": "https://collector.levo.ai",
|
||||
}
|
||||
required_vars.pop(missing_var)
|
||||
|
||||
for key, value in required_vars.items():
|
||||
monkeypatch.setenv(key, value)
|
||||
|
||||
with pytest.raises(ValueError, match=missing_var):
|
||||
LevoLogger.get_levo_config()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user