mirror of
https://github.com/tiennm99/litellm.git
synced 2026-08-05 08:23:17 +00:00
Feat/add posthog observability (#14610)
* feat: add posthog observability * docs: add posthog logging docs * docs: posthog integration in proxy mode
This commit is contained in:
Carlos Marchal
@@ -0,0 +1,216 @@
|
||||
# PostHog - Tracking LLM Usage Analytics
|
||||
|
||||
## What is PostHog?
|
||||
|
||||
PostHog is an open-source product analytics platform that helps you track and analyze how users interact with your product. For LLM applications, PostHog provides specialized AI features to track model usage, performance, and user interactions with your AI features.
|
||||
|
||||
## Usage with LiteLLM Proxy (LLM Gateway)
|
||||
|
||||
**Step 1**: Create a `config.yaml` file and set `litellm_settings`: `success_callback`
|
||||
|
||||
```yaml
|
||||
model_list:
|
||||
- model_name: gpt-3.5-turbo
|
||||
litellm_params:
|
||||
model: gpt-3.5-turbo
|
||||
|
||||
litellm_settings:
|
||||
success_callback: ["posthog"]
|
||||
failure_callback: ["posthog"]
|
||||
```
|
||||
|
||||
**Step 2**: Set required environment variables
|
||||
|
||||
```shell
|
||||
export POSTHOG_API_KEY="your-posthog-api-key"
|
||||
# Optional, defaults to https://app.posthog.com
|
||||
export POSTHOG_API_URL="https://app.posthog.com" # optional
|
||||
```
|
||||
|
||||
**Step 3**: Start the proxy, make a test request
|
||||
|
||||
Start proxy
|
||||
|
||||
```shell
|
||||
litellm --config config.yaml --debug
|
||||
```
|
||||
|
||||
Test Request
|
||||
|
||||
```shell
|
||||
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": "what llm are you"
|
||||
}
|
||||
],
|
||||
"metadata": {
|
||||
"user_id": "user-123",
|
||||
"custom_field": "custom_value"
|
||||
}
|
||||
}'
|
||||
```
|
||||
|
||||
## Usage with LiteLLM Python SDK
|
||||
|
||||
### Quick Start
|
||||
|
||||
Use just 2 lines of code, to instantly log your responses **across all providers** with PostHog:
|
||||
|
||||
```python
|
||||
litellm.success_callback = ["posthog"]
|
||||
litellm.failure_callback = ["posthog"] # logs errors to posthog
|
||||
```
|
||||
```python
|
||||
import litellm
|
||||
import os
|
||||
|
||||
# from PostHog
|
||||
os.environ["POSTHOG_API_KEY"] = ""
|
||||
# Optional, defaults to https://app.posthog.com
|
||||
os.environ["POSTHOG_API_URL"] = "" # optional
|
||||
|
||||
# LLM API Keys
|
||||
os.environ['OPENAI_API_KEY']=""
|
||||
|
||||
# set posthog as a callback, litellm will send the data to posthog
|
||||
litellm.success_callback = ["posthog"]
|
||||
|
||||
# openai call
|
||||
response = litellm.completion(
|
||||
model="gpt-3.5-turbo",
|
||||
messages=[
|
||||
{"role": "user", "content": "Hi - i'm openai"}
|
||||
],
|
||||
metadata = {
|
||||
"user_id": "user-123", # set posthog user ID
|
||||
}
|
||||
)
|
||||
```
|
||||
|
||||
### Advanced
|
||||
|
||||
#### Set User ID and Custom Metadata
|
||||
|
||||
Pass `user_id` in `metadata` to associate events with specific users in PostHog:
|
||||
|
||||
**With LiteLLM Python SDK:**
|
||||
|
||||
```python
|
||||
import litellm
|
||||
|
||||
litellm.success_callback = ["posthog"]
|
||||
|
||||
response = litellm.completion(
|
||||
model="gpt-3.5-turbo",
|
||||
messages=[
|
||||
{"role": "user", "content": "Hello world"}
|
||||
],
|
||||
metadata={
|
||||
"user_id": "user-123", # Add user ID for PostHog tracking
|
||||
"custom_field": "custom_value" # Add custom metadata
|
||||
}
|
||||
)
|
||||
```
|
||||
|
||||
**With LiteLLM Proxy using OpenAI Python SDK:**
|
||||
|
||||
```python
|
||||
import openai
|
||||
|
||||
client = openai.OpenAI(
|
||||
api_key="sk-1234", # Your LiteLLM Proxy API key
|
||||
base_url="http://0.0.0.0:4000" # Your LiteLLM Proxy URL
|
||||
)
|
||||
|
||||
response = client.chat.completions.create(
|
||||
model="gpt-3.5-turbo",
|
||||
messages=[
|
||||
{"role": "user", "content": "Hello world"}
|
||||
],
|
||||
extra_body={
|
||||
"metadata": {
|
||||
"user_id": "user-123", # Add user ID for PostHog tracking
|
||||
"project_name": "my-project", # Add custom metadata
|
||||
"environment": "production"
|
||||
}
|
||||
}
|
||||
)
|
||||
```
|
||||
|
||||
#### Disable Logging for Specific Calls
|
||||
|
||||
Use the `no-log` flag to prevent logging for specific calls:
|
||||
|
||||
```python
|
||||
import litellm
|
||||
|
||||
litellm.success_callback = ["posthog"]
|
||||
|
||||
response = litellm.completion(
|
||||
model="gpt-3.5-turbo",
|
||||
messages=[
|
||||
{"role": "user", "content": "This won't be logged"}
|
||||
],
|
||||
metadata={"no-log": True}
|
||||
)
|
||||
```
|
||||
|
||||
## What's Logged to PostHog?
|
||||
|
||||
When LiteLLM logs to PostHog, it captures detailed information about your LLM usage:
|
||||
|
||||
### For Completion Calls
|
||||
- **Model Information**: Provider, model name, model parameters
|
||||
- **Usage Metrics**: Input tokens, output tokens, total cost
|
||||
- **Performance**: Latency, completion time
|
||||
- **Content**: Input messages, model responses (respects privacy settings)
|
||||
- **Metadata**: Custom fields, user ID, trace information
|
||||
|
||||
### For Embedding Calls
|
||||
- **Model Information**: Provider, model name
|
||||
- **Usage Metrics**: Input tokens, total cost
|
||||
- **Performance**: Latency
|
||||
- **Content**: Input text (respects privacy settings)
|
||||
- **Metadata**: Custom fields, user ID, trace information
|
||||
|
||||
### For Errors
|
||||
- **Error Details**: Error type, error message, stack trace
|
||||
- **Context**: Model, provider, input that caused the error
|
||||
- **Timing**: When the error occurred, request duration
|
||||
|
||||
## Environment Variables
|
||||
|
||||
| Variable | Required | Description |
|
||||
|----------|----------|-------------|
|
||||
| `POSTHOG_API_KEY` | Yes | Your PostHog project API key |
|
||||
| `POSTHOG_API_URL` | No | PostHog API URL (defaults to https://app.posthog.com) |
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
### 1. Missing API Key
|
||||
```
|
||||
Error: POSTHOG_API_KEY is not set
|
||||
```
|
||||
|
||||
Set your PostHog API key:
|
||||
```python
|
||||
import os
|
||||
os.environ["POSTHOG_API_KEY"] = "your-api-key"
|
||||
```
|
||||
|
||||
### 2. Custom PostHog Instance
|
||||
If you're using a self-hosted PostHog instance:
|
||||
```python
|
||||
import os
|
||||
os.environ["POSTHOG_API_URL"] = "https://your-posthog-instance.com"
|
||||
```
|
||||
|
||||
### 3. Events Not Appearing
|
||||
- Check that your API key is correct
|
||||
- Verify network connectivity to PostHog
|
||||
- Events may take a few minutes to appear in PostHog dashboard
|
||||
@@ -147,6 +147,7 @@ _custom_logger_compatible_callbacks_literal = Literal[
|
||||
"vector_store_pre_call_hook",
|
||||
"dotprompt",
|
||||
"cloudzero",
|
||||
"posthog",
|
||||
]
|
||||
configured_cold_storage_logger: Optional[
|
||||
_custom_logger_compatible_callbacks_literal
|
||||
|
||||
@@ -0,0 +1,333 @@
|
||||
"""
|
||||
PostHog Integration - sends LLM analytics events to PostHog
|
||||
|
||||
Follows PostHog's LLM Analytics format: https://posthog.com/docs/llm-analytics/manual-capture
|
||||
|
||||
async_log_success_event: stores batch of events in memory and flushes to PostHog
|
||||
async_log_failure_event: logs failed LLM calls with error information
|
||||
|
||||
For batching specific details see CustomBatchLogger class
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
import uuid
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
|
||||
from litellm._logging import verbose_logger
|
||||
from litellm.integrations.custom_batch_logger import CustomBatchLogger
|
||||
from litellm.llms.custom_httpx.http_handler import (
|
||||
_get_httpx_client,
|
||||
get_async_httpx_client,
|
||||
httpxSpecialProvider,
|
||||
)
|
||||
from litellm.types.integrations.posthog import (
|
||||
POSTHOG_MAX_BATCH_SIZE,
|
||||
PostHogEventPayload,
|
||||
)
|
||||
from litellm.types.utils import StandardLoggingPayload
|
||||
|
||||
|
||||
class PostHogLogger(CustomBatchLogger):
|
||||
def __init__(self, **kwargs):
|
||||
"""
|
||||
Initializes the PostHog logger, checks if the correct env variables are set
|
||||
|
||||
Required environment variables:
|
||||
`POSTHOG_API_KEY` - your PostHog API key
|
||||
`POSTHOG_API_URL` - your PostHog API URL (defaults to https://app.posthog.com)
|
||||
"""
|
||||
try:
|
||||
verbose_logger.debug("PostHog: in init posthog logger")
|
||||
if os.getenv("POSTHOG_API_KEY", None) is None:
|
||||
raise Exception("POSTHOG_API_KEY is not set, set 'POSTHOG_API_KEY=<>'")
|
||||
|
||||
self.async_client = get_async_httpx_client(
|
||||
llm_provider=httpxSpecialProvider.LoggingCallback
|
||||
)
|
||||
self.sync_client = _get_httpx_client()
|
||||
|
||||
self.POSTHOG_API_KEY = os.getenv("POSTHOG_API_KEY")
|
||||
posthog_api_url = os.getenv("POSTHOG_API_URL", "https://us.i.posthog.com")
|
||||
self.posthog_host = posthog_api_url.rstrip('/')
|
||||
self.capture_url = f"{self.posthog_host}/batch/"
|
||||
|
||||
self._async_initialized = False
|
||||
self.flush_lock = None
|
||||
self.log_queue = []
|
||||
|
||||
super().__init__(
|
||||
**kwargs, flush_lock=None, batch_size=POSTHOG_MAX_BATCH_SIZE
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
verbose_logger.exception(
|
||||
f"PostHog: Got exception on init PostHog client {str(e)}"
|
||||
)
|
||||
raise e
|
||||
|
||||
def log_success_event(self, kwargs, response_obj, start_time, end_time):
|
||||
try:
|
||||
verbose_logger.debug(
|
||||
"PostHog: Sync logging - Enters logging function for model %s", kwargs
|
||||
)
|
||||
|
||||
event_payload = self.create_posthog_event_payload(kwargs)
|
||||
|
||||
headers = {
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
payload = self._create_posthog_payload([event_payload])
|
||||
|
||||
response = self.sync_client.post(
|
||||
url=self.capture_url,
|
||||
json=payload,
|
||||
headers=headers,
|
||||
)
|
||||
response.raise_for_status()
|
||||
|
||||
if response.status_code != 200:
|
||||
raise Exception(
|
||||
f"Response from PostHog API status_code: {response.status_code}, text: {response.text}"
|
||||
)
|
||||
|
||||
verbose_logger.debug("PostHog: Sync event successfully sent")
|
||||
|
||||
except Exception as e:
|
||||
verbose_logger.exception(f"PostHog Sync Layer Error - {str(e)}")
|
||||
|
||||
async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
|
||||
try:
|
||||
verbose_logger.debug(
|
||||
"PostHog: Async logging - Enters logging function for model %s", kwargs
|
||||
)
|
||||
self._ensure_async_setup() # Lazy initialization
|
||||
await self._log_async_event(kwargs, response_obj, start_time, end_time)
|
||||
except Exception as e:
|
||||
verbose_logger.exception(f"PostHog Layer Error - {str(e)}")
|
||||
pass
|
||||
|
||||
async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
|
||||
try:
|
||||
verbose_logger.debug(
|
||||
"PostHog: Async logging - Enters logging function for model %s", kwargs
|
||||
)
|
||||
self._ensure_async_setup() # Lazy initialization
|
||||
await self._log_async_event(kwargs, response_obj, start_time, end_time)
|
||||
except Exception as e:
|
||||
verbose_logger.exception(f"PostHog Layer Error - {str(e)}")
|
||||
pass
|
||||
|
||||
async def _log_async_event(self, kwargs, response_obj=None, start_time=0.0, end_time=0.0):
|
||||
# Note: response_obj, start_time, end_time not used - all data comes from kwargs
|
||||
event_payload = self.create_posthog_event_payload(kwargs)
|
||||
|
||||
self.log_queue.append(event_payload)
|
||||
verbose_logger.debug(
|
||||
f"PostHog, event added to queue. Will flush in {self.flush_interval} seconds..."
|
||||
)
|
||||
|
||||
if len(self.log_queue) >= self.batch_size:
|
||||
await self.flush_queue()
|
||||
|
||||
def create_posthog_event_payload(self, kwargs: Dict[str, Any]) -> PostHogEventPayload:
|
||||
"""
|
||||
Helper function to create a PostHog event payload for logging
|
||||
|
||||
Args:
|
||||
kwargs (Dict[str, Any]): request kwargs containing standard_logging_object
|
||||
|
||||
Returns:
|
||||
PostHogEventPayload: defined in types.py
|
||||
"""
|
||||
standard_logging_object: Optional[StandardLoggingPayload] = kwargs.get(
|
||||
"standard_logging_object", None
|
||||
)
|
||||
if standard_logging_object is None:
|
||||
raise ValueError("standard_logging_object not found in kwargs")
|
||||
|
||||
call_type = standard_logging_object.get("call_type", "")
|
||||
event_name = "$ai_embedding" if call_type == "embedding" else "$ai_generation"
|
||||
|
||||
properties = self._create_posthog_properties(
|
||||
standard_logging_object=standard_logging_object,
|
||||
kwargs=kwargs,
|
||||
event_name=event_name,
|
||||
)
|
||||
|
||||
distinct_id = self._get_distinct_id(standard_logging_object, kwargs)
|
||||
|
||||
return PostHogEventPayload(
|
||||
event=event_name,
|
||||
properties=properties,
|
||||
distinct_id=distinct_id,
|
||||
)
|
||||
|
||||
def _create_posthog_properties(
|
||||
self,
|
||||
standard_logging_object: StandardLoggingPayload,
|
||||
kwargs: Dict[str, Any],
|
||||
event_name: str,
|
||||
) -> Dict[str, Any]:
|
||||
"""Create PostHog properties following LLM Analytics spec"""
|
||||
properties = {}
|
||||
|
||||
# Core model information
|
||||
properties["$ai_model"] = self._safe_get(standard_logging_object, "model", "")
|
||||
properties["$ai_provider"] = self._safe_get(standard_logging_object, "custom_llm_provider", "")
|
||||
|
||||
# Input/Output data
|
||||
messages = self._safe_get(standard_logging_object, "messages")
|
||||
if messages is not None:
|
||||
properties["$ai_input"] = messages
|
||||
|
||||
if event_name == "$ai_generation":
|
||||
response = self._safe_get(standard_logging_object, "response")
|
||||
if response is not None:
|
||||
properties["$ai_output_choices"] = response
|
||||
|
||||
# Token information
|
||||
properties["$ai_input_tokens"] = self._safe_get(standard_logging_object, "prompt_tokens", 0)
|
||||
if event_name == "$ai_generation":
|
||||
properties["$ai_output_tokens"] = self._safe_get(standard_logging_object, "completion_tokens", 0)
|
||||
|
||||
# Cost and performance
|
||||
response_cost = self._safe_get(standard_logging_object, "response_cost")
|
||||
if response_cost is not None:
|
||||
properties["$ai_total_cost_usd"] = response_cost
|
||||
|
||||
properties["$ai_latency"] = self._safe_get(standard_logging_object, "response_time", 0.0)
|
||||
|
||||
# Error handling
|
||||
if self._safe_get(standard_logging_object, "status") == "failure":
|
||||
properties["$ai_is_error"] = True
|
||||
error_str = self._safe_get(standard_logging_object, "error_str")
|
||||
if error_str is not None:
|
||||
properties["$ai_error"] = error_str
|
||||
|
||||
# Add trace properties
|
||||
self._add_trace_properties(properties, kwargs)
|
||||
|
||||
# Add custom metadata fields
|
||||
self._add_custom_metadata_properties(properties, kwargs)
|
||||
|
||||
return properties
|
||||
|
||||
def _add_trace_properties(self, properties: Dict[str, Any], kwargs: Dict[str, Any]):
|
||||
standard_logging_object = self._safe_get(kwargs, "standard_logging_object", {})
|
||||
|
||||
trace_id = self._safe_get(standard_logging_object, "trace_id", self._safe_uuid())
|
||||
properties["$ai_trace_id"] = trace_id
|
||||
|
||||
span_id = self._safe_get(standard_logging_object, "id", self._safe_uuid())
|
||||
properties["$ai_span_id"] = span_id
|
||||
|
||||
metadata = self._extract_metadata(kwargs)
|
||||
parent_id = metadata.get("parent_run_id") or metadata.get("parent_id")
|
||||
if parent_id:
|
||||
properties["$ai_parent_id"] = parent_id
|
||||
|
||||
def _add_custom_metadata_properties(self, properties: Dict[str, Any], kwargs: Dict[str, Any]):
|
||||
"""Add custom metadata fields to PostHog properties"""
|
||||
metadata = self._extract_metadata(kwargs)
|
||||
if not isinstance(metadata, dict):
|
||||
return
|
||||
|
||||
litellm_internal_fields = {
|
||||
"endpoint", "caching_groups", "user_api_key_hash", "user_api_key_alias",
|
||||
"user_api_key_team_id", "user_api_key_user_id", "user_api_key_org_id",
|
||||
"user_api_key_team_alias", "user_api_key_end_user_id", "user_api_key_user_email",
|
||||
"user_api_key", "user_api_end_user_max_budget", "litellm_api_version",
|
||||
"global_max_parallel_requests", "user_api_key_team_max_budget", "user_api_key_team_spend",
|
||||
"user_api_key_spend", "user_api_key_max_budget", "user_api_key_model_max_budget",
|
||||
"user_api_key_metadata", "headers", "litellm_parent_otel_span", "requester_ip_address",
|
||||
"model_group", "model_group_size", "deployment", "model_info", "api_base",
|
||||
"caching_groups", "hidden_params", "parent_run_id", "parent_id", "user_id"
|
||||
}
|
||||
|
||||
for key, value in metadata.items():
|
||||
if key not in litellm_internal_fields:
|
||||
properties[key] = value
|
||||
|
||||
def _get_distinct_id(
|
||||
self, standard_logging_object: StandardLoggingPayload, kwargs: Dict[str, Any]
|
||||
) -> str:
|
||||
metadata = self._extract_metadata(kwargs)
|
||||
user_id = self._safe_get(metadata, "user_id")
|
||||
if user_id:
|
||||
return str(user_id)
|
||||
end_user = self._safe_get(standard_logging_object, "end_user")
|
||||
if end_user:
|
||||
return str(end_user)
|
||||
trace_id = self._safe_get(standard_logging_object, "trace_id")
|
||||
if trace_id:
|
||||
return str(trace_id)
|
||||
|
||||
return self._safe_uuid()
|
||||
|
||||
async def async_send_batch(self):
|
||||
"""
|
||||
Sends the in memory logs queue to PostHog API
|
||||
|
||||
Raises:
|
||||
Raises a NON Blocking verbose_logger.exception if an error occurs
|
||||
"""
|
||||
try:
|
||||
if not self.log_queue:
|
||||
return
|
||||
|
||||
verbose_logger.debug(
|
||||
f"PostHog: Sending batch of {len(self.log_queue)} events"
|
||||
)
|
||||
|
||||
headers = {
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
payload = self._create_posthog_payload(list(self.log_queue))
|
||||
|
||||
response = await self.async_client.post(
|
||||
url=self.capture_url,
|
||||
json=payload,
|
||||
headers=headers,
|
||||
)
|
||||
response.raise_for_status()
|
||||
|
||||
if response.status_code != 200:
|
||||
raise Exception(
|
||||
f"Response from PostHog API status_code: {response.status_code}, text: {response.text}"
|
||||
)
|
||||
|
||||
verbose_logger.debug(
|
||||
f"PostHog: Batch of {len(self.log_queue)} events successfully sent"
|
||||
)
|
||||
except Exception as e:
|
||||
verbose_logger.exception(f"PostHog Error sending batch API - {str(e)}")
|
||||
|
||||
def _ensure_async_setup(self):
|
||||
if not self._async_initialized:
|
||||
try:
|
||||
self.flush_lock = asyncio.Lock()
|
||||
asyncio.create_task(self.periodic_flush())
|
||||
self._async_initialized = True
|
||||
verbose_logger.debug("PostHog: Async components initialized")
|
||||
except Exception as e:
|
||||
verbose_logger.error(f"PostHog: Failed to initialize async components: {str(e)}")
|
||||
raise
|
||||
|
||||
def _extract_metadata(self, kwargs: Dict[str, Any]) -> Dict[str, Any]:
|
||||
litellm_params = kwargs.get("litellm_params", {}) or {}
|
||||
return litellm_params.get("metadata", {}) or {}
|
||||
|
||||
def _safe_uuid(self) -> str:
|
||||
return str(uuid.uuid4())
|
||||
|
||||
def _create_posthog_payload(self, events: list) -> Dict[str, Any]:
|
||||
return {"api_key": self.POSTHOG_API_KEY, "batch": events}
|
||||
|
||||
def _safe_get(self, obj: Any, key: str, default: Any = None) -> Any:
|
||||
if obj is None or not hasattr(obj, 'get'):
|
||||
return default
|
||||
return obj.get(key, default)
|
||||
@@ -33,6 +33,7 @@ from litellm.integrations.mlflow import MlflowLogger
|
||||
from litellm.integrations.openmeter import OpenMeterLogger
|
||||
from litellm.integrations.opentelemetry import OpenTelemetry
|
||||
from litellm.integrations.opik.opik import OpikLogger
|
||||
from litellm.integrations.posthog import PostHogLogger
|
||||
|
||||
try:
|
||||
from litellm_enterprise.integrations.prometheus import PrometheusLogger
|
||||
@@ -88,6 +89,7 @@ class CustomLoggerRegistry:
|
||||
"vector_store_pre_call_hook": VectorStorePreCallHook,
|
||||
"dotprompt": DotpromptManager,
|
||||
"cloudzero": CloudZeroLogger,
|
||||
"posthog": PostHogLogger,
|
||||
}
|
||||
|
||||
try:
|
||||
|
||||
@@ -138,6 +138,7 @@ from ..integrations.logfire_logger import LogfireLevel, LogfireLogger
|
||||
from ..integrations.lunary import LunaryLogger
|
||||
from ..integrations.openmeter import OpenMeterLogger
|
||||
from ..integrations.opik.opik import OpikLogger
|
||||
from ..integrations.posthog import PostHogLogger
|
||||
from ..integrations.prompt_layer import PromptLayerLogger
|
||||
from ..integrations.s3 import S3Logger
|
||||
from ..integrations.s3_v2 import S3Logger as S3V2Logger
|
||||
@@ -193,7 +194,6 @@ _in_memory_loggers: List[Any] = []
|
||||
sentry_sdk_instance = None
|
||||
capture_exception = None
|
||||
add_breadcrumb = None
|
||||
posthog = None
|
||||
slack_app = None
|
||||
alerts_channel = None
|
||||
heliconeLogger = None
|
||||
@@ -3068,7 +3068,7 @@ def set_callbacks(callback_list, function_id=None): # noqa: PLR0915
|
||||
"""
|
||||
Globally sets the callback client
|
||||
"""
|
||||
global sentry_sdk_instance, capture_exception, add_breadcrumb, posthog, slack_app, alerts_channel, traceloopLogger, athinaLogger, heliconeLogger, supabaseClient, lunaryLogger, promptLayerLogger, langFuseLogger, customLogger, weightsBiasesLogger, logfireLogger, dynamoLogger, s3Logger, dataDogLogger, prometheusLogger, greenscaleLogger, openMeterLogger, deepevalLogger
|
||||
global sentry_sdk_instance, capture_exception, add_breadcrumb, slack_app, alerts_channel, traceloopLogger, athinaLogger, heliconeLogger, supabaseClient, lunaryLogger, promptLayerLogger, langFuseLogger, customLogger, weightsBiasesLogger, logfireLogger, dynamoLogger, s3Logger, dataDogLogger, prometheusLogger, greenscaleLogger, openMeterLogger, deepevalLogger
|
||||
|
||||
try:
|
||||
for callback in callback_list:
|
||||
@@ -3107,19 +3107,6 @@ def set_callbacks(callback_list, function_id=None): # noqa: PLR0915
|
||||
)
|
||||
capture_exception = sentry_sdk_instance.capture_exception
|
||||
add_breadcrumb = sentry_sdk_instance.add_breadcrumb
|
||||
elif callback == "posthog":
|
||||
try:
|
||||
from posthog import Posthog
|
||||
except ImportError:
|
||||
print_verbose("Package 'posthog' is missing. Installing it...")
|
||||
subprocess.check_call(
|
||||
[sys.executable, "-m", "pip", "install", "posthog"]
|
||||
)
|
||||
from posthog import Posthog
|
||||
posthog = Posthog(
|
||||
project_api_key=os.environ.get("POSTHOG_API_KEY"),
|
||||
host=os.environ.get("POSTHOG_API_URL"),
|
||||
)
|
||||
elif callback == "slack":
|
||||
try:
|
||||
from slack_bolt import App
|
||||
@@ -3214,6 +3201,14 @@ def _init_custom_logger_compatible_class( # noqa: PLR0915
|
||||
_openmeter_logger = OpenMeterLogger()
|
||||
_in_memory_loggers.append(_openmeter_logger)
|
||||
return _openmeter_logger # type: ignore
|
||||
elif logging_integration == "posthog":
|
||||
for callback in _in_memory_loggers:
|
||||
if isinstance(callback, PostHogLogger):
|
||||
return callback # type: ignore
|
||||
|
||||
_posthog_logger = PostHogLogger()
|
||||
_in_memory_loggers.append(_posthog_logger)
|
||||
return _posthog_logger # type: ignore
|
||||
elif logging_integration == "braintrust":
|
||||
from litellm.integrations.braintrust_logging import BraintrustLogger
|
||||
|
||||
|
||||
@@ -0,0 +1,18 @@
|
||||
from typing import Any, Dict, TypedDict
|
||||
|
||||
POSTHOG_MAX_BATCH_SIZE = 100
|
||||
|
||||
|
||||
class PostHogEventPayload(TypedDict):
|
||||
"""PostHog event payload structure"""
|
||||
|
||||
event: str # "$ai_generation" or "$ai_embedding"
|
||||
properties: Dict[str, Any]
|
||||
distinct_id: str
|
||||
|
||||
|
||||
class PostHogCredentialsObject(TypedDict):
|
||||
"""PostHog credentials configuration"""
|
||||
|
||||
POSTHOG_API_KEY: str
|
||||
POSTHOG_HOST: str
|
||||
@@ -0,0 +1,258 @@
|
||||
import os
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, os.path.abspath("../.."))
|
||||
|
||||
import pytest
|
||||
|
||||
from litellm.integrations.posthog import PostHogLogger
|
||||
from litellm.types.utils import StandardLoggingPayload
|
||||
from typing import cast
|
||||
|
||||
# Set env vars for tests
|
||||
os.environ["POSTHOG_API_KEY"] = "test_key"
|
||||
os.environ["POSTHOG_API_URL"] = "https://app.posthog.com"
|
||||
|
||||
|
||||
def create_standard_logging_payload() -> StandardLoggingPayload:
|
||||
# Use cast to bypass strict TypedDict requirements for tests
|
||||
return cast(StandardLoggingPayload, {
|
||||
"id": "test_id",
|
||||
"trace_id": "test_trace_id",
|
||||
"call_type": "completion",
|
||||
"stream": False,
|
||||
"response_cost": 0.1,
|
||||
"status": "success",
|
||||
"custom_llm_provider": "openai",
|
||||
"total_tokens": 30,
|
||||
"prompt_tokens": 20,
|
||||
"completion_tokens": 10,
|
||||
"startTime": 1234567890.0,
|
||||
"endTime": 1234567891.0,
|
||||
"completionStartTime": 1234567890.5,
|
||||
"response_time": 1.0,
|
||||
"model": "gpt-3.5-turbo",
|
||||
"model_id": "model-123",
|
||||
"api_base": "https://api.openai.com",
|
||||
"cache_hit": False,
|
||||
"saved_cache_cost": 0.0,
|
||||
"request_tags": [],
|
||||
"end_user": None,
|
||||
"messages": [{"role": "user", "content": "Hello, world!"}],
|
||||
"response": {"choices": [{"message": {"content": "Hi there!"}}]},
|
||||
"error_str": None,
|
||||
"model_parameters": {"stream": True},
|
||||
})
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_posthog_event_payload():
|
||||
posthog_logger = PostHogLogger()
|
||||
standard_payload = create_standard_logging_payload()
|
||||
kwargs = {"standard_logging_object": standard_payload}
|
||||
|
||||
event_payload = posthog_logger.create_posthog_event_payload(kwargs)
|
||||
|
||||
assert event_payload["event"] == "$ai_generation"
|
||||
assert event_payload["properties"]["$ai_model"] == "gpt-3.5-turbo"
|
||||
assert event_payload["properties"]["$ai_input_tokens"] == 20
|
||||
assert event_payload["properties"]["$ai_output_tokens"] == 10
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_posthog_failure_logging():
|
||||
posthog_logger = PostHogLogger()
|
||||
standard_payload = create_standard_logging_payload()
|
||||
standard_payload["status"] = "failure"
|
||||
standard_payload["error_str"] = "Test error"
|
||||
|
||||
kwargs = {"standard_logging_object": standard_payload}
|
||||
|
||||
event_payload = posthog_logger.create_posthog_event_payload(kwargs)
|
||||
|
||||
assert event_payload["properties"]["$ai_is_error"] is True
|
||||
assert event_payload["properties"]["$ai_error"] == "Test error"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_posthog_embedding_event():
|
||||
posthog_logger = PostHogLogger()
|
||||
standard_payload = create_standard_logging_payload()
|
||||
standard_payload["call_type"] = "embedding"
|
||||
|
||||
kwargs = {"standard_logging_object": standard_payload}
|
||||
|
||||
event_payload = posthog_logger.create_posthog_event_payload(kwargs)
|
||||
|
||||
assert event_payload["event"] == "$ai_embedding"
|
||||
assert "$ai_output_tokens" not in event_payload["properties"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_trace_id_fallback_from_standard_logging_object():
|
||||
"""Test that trace_id is properly extracted from standard_logging_object"""
|
||||
posthog_logger = PostHogLogger()
|
||||
standard_payload = create_standard_logging_payload()
|
||||
standard_payload["trace_id"] = "test-trace-123"
|
||||
|
||||
kwargs = {"standard_logging_object": standard_payload}
|
||||
|
||||
event_payload = posthog_logger.create_posthog_event_payload(kwargs)
|
||||
|
||||
assert event_payload["properties"]["$ai_trace_id"] == "test-trace-123"
|
||||
assert event_payload["properties"]["$ai_span_id"] == "test_id" # from standard_payload["id"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_trace_id_uuid_fallback():
|
||||
"""Test that UUID is generated when no trace_id is available"""
|
||||
posthog_logger = PostHogLogger()
|
||||
standard_payload = create_standard_logging_payload()
|
||||
# Remove trace_id to test fallback
|
||||
del standard_payload["trace_id"]
|
||||
del standard_payload["id"]
|
||||
|
||||
kwargs = {"standard_logging_object": standard_payload}
|
||||
|
||||
event_payload = posthog_logger.create_posthog_event_payload(kwargs)
|
||||
|
||||
# Should have generated UUIDs
|
||||
assert len(event_payload["properties"]["$ai_trace_id"]) == 36 # UUID length
|
||||
assert len(event_payload["properties"]["$ai_span_id"]) == 36 # UUID length
|
||||
assert "-" in event_payload["properties"]["$ai_trace_id"] # UUID format
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_distinct_id_fallback_chain():
|
||||
"""Test the distinct_id fallback priority chain"""
|
||||
posthog_logger = PostHogLogger()
|
||||
|
||||
# Test 1: user_id from metadata (highest priority)
|
||||
standard_payload = create_standard_logging_payload()
|
||||
kwargs = {
|
||||
"standard_logging_object": standard_payload,
|
||||
"litellm_params": {"metadata": {"user_id": "metadata-user-123"}}
|
||||
}
|
||||
|
||||
distinct_id = posthog_logger._get_distinct_id(standard_payload, kwargs)
|
||||
assert distinct_id == "metadata-user-123"
|
||||
|
||||
# Test 2: trace_id from standard_logging_object (second priority)
|
||||
kwargs = {"standard_logging_object": standard_payload} # no metadata
|
||||
distinct_id = posthog_logger._get_distinct_id(standard_payload, kwargs)
|
||||
assert distinct_id == "test_trace_id"
|
||||
|
||||
# Test 3: end_user from standard_logging_object (third priority)
|
||||
standard_payload_no_trace = create_standard_logging_payload()
|
||||
del standard_payload_no_trace["trace_id"]
|
||||
standard_payload_no_trace["end_user"] = "end-user-456"
|
||||
|
||||
distinct_id = posthog_logger._get_distinct_id(standard_payload_no_trace, {})
|
||||
assert distinct_id == "end-user-456"
|
||||
|
||||
# Test 4: UUID fallback (lowest priority)
|
||||
standard_payload_empty = create_standard_logging_payload()
|
||||
del standard_payload_empty["trace_id"]
|
||||
del standard_payload_empty["end_user"]
|
||||
|
||||
distinct_id = posthog_logger._get_distinct_id(standard_payload_empty, {})
|
||||
assert len(distinct_id) == 36 # UUID length
|
||||
assert "-" in distinct_id # UUID format
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_missing_standard_logging_object():
|
||||
"""Test error handling when standard_logging_object is missing"""
|
||||
posthog_logger = PostHogLogger()
|
||||
|
||||
kwargs = {} # Missing standard_logging_object
|
||||
|
||||
with pytest.raises(ValueError, match="standard_logging_object not found in kwargs"):
|
||||
posthog_logger.create_posthog_event_payload(kwargs)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_custom_metadata_support():
|
||||
"""Test that custom metadata fields are added directly to properties"""
|
||||
posthog_logger = PostHogLogger()
|
||||
standard_payload = create_standard_logging_payload()
|
||||
|
||||
kwargs = {
|
||||
"standard_logging_object": standard_payload,
|
||||
"litellm_params": {
|
||||
"metadata": {
|
||||
"user_id": "user-123", # should be used for distinct_id, not custom property
|
||||
"project_name": "test_project", # should appear as project_name
|
||||
"environment": "staging", # should appear as environment
|
||||
"custom_field": "custom_value" # should appear as custom_field
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
event_payload = posthog_logger.create_posthog_event_payload(kwargs)
|
||||
|
||||
# Check that custom fields are added directly
|
||||
assert event_payload["properties"]["project_name"] == "test_project"
|
||||
assert event_payload["properties"]["environment"] == "staging"
|
||||
assert event_payload["properties"]["custom_field"] == "custom_value"
|
||||
|
||||
# Check that user_id is used for distinct_id, not as custom property
|
||||
assert event_payload["distinct_id"] == "user-123"
|
||||
assert "user_id" not in event_payload["properties"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_custom_metadata_filters_internal_fields():
|
||||
"""Test that LiteLLM internal fields are filtered out from custom metadata"""
|
||||
posthog_logger = PostHogLogger()
|
||||
standard_payload = create_standard_logging_payload()
|
||||
|
||||
kwargs = {
|
||||
"standard_logging_object": standard_payload,
|
||||
"litellm_params": {
|
||||
"metadata": {
|
||||
"custom_field": "should_appear",
|
||||
"endpoint": "/chat/completions", # internal field - should be filtered
|
||||
"user_api_key_hash": "hash123", # internal field - should be filtered
|
||||
"headers": {"content-type": "application/json"}, # internal field - should be filtered
|
||||
"model_info": {"id": "123"}, # internal field - should be filtered
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
event_payload = posthog_logger.create_posthog_event_payload(kwargs)
|
||||
|
||||
# Check that custom field appears
|
||||
assert event_payload["properties"]["custom_field"] == "should_appear"
|
||||
|
||||
# Check that internal fields are filtered out
|
||||
assert "endpoint" not in event_payload["properties"]
|
||||
assert "user_api_key_hash" not in event_payload["properties"]
|
||||
assert "headers" not in event_payload["properties"]
|
||||
assert "model_info" not in event_payload["properties"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_custom_metadata_with_no_metadata():
|
||||
"""Test that logger handles cases with no metadata gracefully"""
|
||||
posthog_logger = PostHogLogger()
|
||||
standard_payload = create_standard_logging_payload()
|
||||
|
||||
# Test with no litellm_params
|
||||
kwargs = {"standard_logging_object": standard_payload}
|
||||
event_payload = posthog_logger.create_posthog_event_payload(kwargs)
|
||||
|
||||
# Should not error and should have standard properties
|
||||
assert event_payload["event"] == "$ai_generation"
|
||||
assert event_payload["properties"]["$ai_model"] == "gpt-3.5-turbo"
|
||||
|
||||
# Test with empty metadata
|
||||
kwargs = {
|
||||
"standard_logging_object": standard_payload,
|
||||
"litellm_params": {"metadata": {}}
|
||||
}
|
||||
event_payload = posthog_logger.create_posthog_event_payload(kwargs)
|
||||
|
||||
# Should not error and should have standard properties
|
||||
assert event_payload["event"] == "$ai_generation"
|
||||
assert event_payload["properties"]["$ai_model"] == "gpt-3.5-turbo"
|
||||
Reference in New Issue
Block a user