[feature] ConfidentAI logging enabled for proxy and sdk (#10649)

* async success implemented * fail async event * sync events added * docs added * docs added * test added * style * test * . * lock file genrated due to tenacity change * mypy errors * resolved comments * resolved comments * resolved comments * resolved comments * style * style * resolved comments
2026-07-08 13:12:16 +00:00 · 2025-05-24 00:10:48 +05:30
parent e9b7059af4
commit 8da898c55e
13 changed files with 622 additions and 4933 deletions
@@ -0,0 +1,55 @@
+import Image from '@theme/IdealImage';
+
+# 🔭 DeepEval - Open-Source Evals with Tracing
+
+### What is DeepEval?
+[DeepEval](https://deepeval.com) is an open-source evaluation framework for LLMs ([Github](https://github.com/confident-ai/deepeval)). 
+
+### What is Confident AI?
+
+[Confident AI](https://documentation.confident-ai.com) (the ***deepeval*** platfrom) offers an Observatory for teams to trace and monitor LLM applications. Think Datadog for LLM apps. The observatory allows you to:
+
+- Detect and debug issues in your LLM applications in real-time
+- Search and analyze historical generation data with powerful filters
+- Collect human feedback on model responses
+- Run evaluations to measure and improve performance
+- Track costs and latency to optimize resource usage
+
+<Image img={require('../../img/deepeval_dashboard.png')} />
+
+### Quickstart
+
+```python
+import os
+import time
+import litellm
+
+
+os.environ['OPENAI_API_KEY']='<your-openai-api-key>'
+os.environ['CONFIDENT_API_KEY']='<your-confident-api-key>'
+
+litellm.success_callback = ["deepeval"]
+litellm.failure_callback = ["deepeval"]
+
+try:
+    response = litellm.completion(
+        model="gpt-3.5-turbo",
+        messages=[
+            {"role": "user", "content": "What's the weather like in San Francisco?"}
+        ],
+    )
+except Exception as e:
+    print(e)
+
+print(response)
+```
+
+:::info
+You can obtain your `CONFIDENT_API_KEY` by logging into [Confident AI](https://app.confident-ai.com/project) platform. 
+:::
+
+## Support & Talk with Deepeval team
+- [Confident AI Docs 📝](https://documentation.confident-ai.com)
+- [Platform 🚀](https://confident-ai.com)
+- [Community Discord 💭](https://discord.gg/wuPM9dRgDw)
+- Support ✉️ support@confident-ai.com
@@ -11,6 +11,7 @@ Log Proxy input, output, and exceptions using:
 - GCS, s3, Azure (Blob) Buckets
 - Lunary
 - MLflow
+- Deepeval
 - Custom Callbacks - Custom code and API endpoints
 - Langsmith
 - DataDog
@@ -1182,7 +1183,58 @@ curl --location 'http://0.0.0.0:4000/chat/completions' \
 '
 ```

+## Deepeval
+LiteLLM supports logging on [Confidential AI](https://documentation.confident-ai.com/) (The Deepeval Platform):

+### Usage:
+1. Add `deepeval` in the LiteLLM `config.yaml`
+
+```yaml
+model_list:
+  - model_name: gpt-4o
+    litellm_params:
+      model: gpt-4o
+litellm_settings:
+  success_callback: ["deepeval"]
+  failure_callback: ["deepeval"]
+```
+
+2. Set your environment variables in `.env` file. 
+```shell
+CONFIDENT_API_KEY=<your-api-key>
+```
+:::info
+You can obtain your `CONFIDENT_API_KEY` by logging into [Confident AI](https://app.confident-ai.com/project) platform. 
+:::
+
+3. Start your proxy server:
+```shell
+litellm --config config.yaml --debug
+```
+
+4. Make a request:
+```shell
+curl -X POST 'http://0.0.0.0:4000/chat/completions' \
+-H 'Content-Type: application/json' \
+-H 'Authorization: Bearer sk-1234' \
+-d '{
+    "model": "gpt-3.5-turbo",
+    "messages": [
+      {
+        "role": "system",
+        "content": "You are a helpful math tutor. Guide the user through the solution step by step."
+      },
+      {
+        "role": "user",
+        "content": "how can I solve 8x + 7 = -23"
+      }
+    ]
+}'
+```
+
+5. Check trace on platform: 
+
+<Image img={require('../../img/deepeval_visible_trace.png')} />

 ## s3 Buckets

@@ -460,6 +460,7 @@ const sidebars = {
        "observability/agentops_integration",
        "observability/langfuse_integration",
        "observability/lunary_integration",
+        "observability/deepeval_integration",
        "observability/mlflow",
        "observability/gcs_bucket_integration",
        "observability/langsmith_integration",
@@ -118,6 +118,7 @@ _custom_logger_compatible_callbacks_literal = Literal[
    "generic_api",
    "resend_email",
    "smtp_email",
+    "deepeval"
 ]
 logged_real_time_event_types: Optional[Union[List[str], Literal["*"]]] = None
 _known_custom_logger_compatible_callbacks: List = list(
@@ -0,0 +1,120 @@
+# duplicate -> https://github.com/confident-ai/deepeval/blob/main/deepeval/confident/api.py
+import logging
+import httpx
+from enum import Enum
+from litellm._logging import verbose_logger
+
+DEEPEVAL_BASE_URL = "https://deepeval.confident-ai.com"
+DEEPEVAL_BASE_URL_EU = "https://eu.deepeval.confident-ai.com"
+API_BASE_URL = "https://api.confident-ai.com"
+API_BASE_URL_EU = "https://eu.api.confident-ai.com"
+retryable_exceptions = httpx.HTTPError
+
+from litellm.llms.custom_httpx.http_handler import (
+    HTTPHandler,
+    get_async_httpx_client,
+    httpxSpecialProvider,
+)
+
+
+def log_retry_error(details):
+    exception = details.get("exception")
+    tries = details.get("tries")
+    if exception:
+        logging.error(f"Confident AI Error: {exception}. Retrying: {tries} time(s)...")
+    else:
+        logging.error(f"Retrying: {tries} time(s)...")
+
+
+class HttpMethods(Enum):
+    GET = "GET"
+    POST = "POST"
+    DELETE = "DELETE"
+    PUT = "PUT"
+
+
+class Endpoints(Enum):
+    DATASET_ENDPOINT = "/v1/dataset"
+    TEST_RUN_ENDPOINT = "/v1/test-run"
+    TRACING_ENDPOINT = "/v1/tracing"
+    EVENT_ENDPOINT = "/v1/event"
+    FEEDBACK_ENDPOINT = "/v1/feedback"
+    PROMPT_ENDPOINT = "/v1/prompt"
+    RECOMMEND_ENDPOINT = "/v1/recommend-metrics"
+    EVALUATE_ENDPOINT = "/evaluate"
+    GUARD_ENDPOINT = "/guard"
+    GUARDRAILS_ENDPOINT = "/guardrails"
+    BASELINE_ATTACKS_ENDPOINT = "/generate-baseline-attacks"
+
+
+class Api:
+    def __init__(self, api_key: str, base_url=None):
+        self.api_key = api_key
+        self._headers = {
+            "Content-Type": "application/json",
+            # "User-Agent": "Python/Requests",
+            "CONFIDENT_API_KEY": api_key,
+        }
+        # using the global non-eu variable for base url
+        self.base_api_url = base_url or API_BASE_URL
+        self.sync_http_handler = HTTPHandler()
+        self.async_http_handler = get_async_httpx_client(
+            llm_provider=httpxSpecialProvider.LoggingCallback
+        )
+
+    def _http_request(
+        self, method: str, url: str, headers=None, json=None, params=None
+    ):
+        if method != "POST":
+            raise Exception("Only POST requests are supported")
+        try:
+            self.sync_http_handler.post(
+                url=url,
+                headers=headers,
+                json=json,
+                params=params,
+            )
+        except httpx.HTTPStatusError as e:
+            raise Exception(f"DeepEval logging error: {e.response.text}")
+        except Exception as e:
+            raise e
+
+    def send_request(
+        self, method: HttpMethods, endpoint: Endpoints, body=None, params=None
+    ):
+        url = f"{self.base_api_url}{endpoint.value}"
+        res = self._http_request(
+            method=method.value,
+            url=url,
+            headers=self._headers,
+            json=body,
+            params=params,
+        )
+
+        if res.status_code == 200:
+            try:
+                return res.json()
+            except ValueError:
+                return res.text
+        else:
+            verbose_logger.debug(res.json())
+            raise Exception(res.json().get("error", res.text))
+
+    async def a_send_request(
+        self, method: HttpMethods, endpoint: Endpoints, body=None, params=None
+    ):
+        if method != HttpMethods.POST:
+            raise Exception("Only POST requests are supported")
+
+        url = f"{self.base_api_url}{endpoint.value}"
+        try:
+            await self.async_http_handler.post(
+                url=url,
+                headers=self._headers,
+                json=body,
+                params=params,
+            )
+        except httpx.HTTPStatusError as e:
+            raise Exception(f"DeepEval logging error: {e.response.text}")
+        except Exception as e:
+            raise e
@@ -0,0 +1,175 @@
+import os
+import uuid
+from litellm.integrations.custom_logger import CustomLogger
+from litellm.integrations.deepeval.api import Api, Endpoints, HttpMethods
+from litellm.integrations.deepeval.types import (
+    BaseApiSpan,
+    SpanApiType,
+    TraceApi,
+    TraceSpanApiStatus,
+)
+from litellm.integrations.deepeval.utils import (
+    to_zod_compatible_iso,
+    validate_environment,
+)
+from litellm._logging import verbose_logger
+
+
+# This file includes the custom callbacks for LiteLLM Proxy
+# Once defined, these can be passed in proxy_config.yaml
+class DeepEvalLogger(CustomLogger):
+    """Logs litellm traces to DeepEval's platform."""
+
+    def __init__(self, *args, **kwargs):
+        api_key = os.getenv("CONFIDENT_API_KEY")
+        self.litellm_environment = os.getenv("LITELM_ENVIRONMENT", "development")
+        validate_environment(self.litellm_environment)
+        if not api_key:
+            raise ValueError(
+                "Please set 'CONFIDENT_API_KEY=<>' in your environment variables."
+            )
+        self.api = Api(api_key=api_key)
+        super().__init__(*args, **kwargs)
+
+    def log_success_event(self, kwargs, response_obj, start_time, end_time):
+        """Logs a success event to DeepEval's platform."""
+        self._sync_event_handler(
+            kwargs, response_obj, start_time, end_time, is_success=True
+        )
+
+    def log_failure_event(self, kwargs, response_obj, start_time, end_time):
+        """Logs a failure event to DeepEval's platform."""
+        self._sync_event_handler(
+            kwargs, response_obj, start_time, end_time, is_success=False
+        )
+
+    async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
+        """Logs a failure event to DeepEval's platform."""
+        await self._async_event_handler(
+            kwargs, response_obj, start_time, end_time, is_success=False
+        )
+
+    async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
+        """Logs a success event to DeepEval's platform."""
+        await self._async_event_handler(
+            kwargs, response_obj, start_time, end_time, is_success=True
+        )
+
+    def _prepare_trace_api(
+        self, kwargs, response_obj, start_time, end_time, is_success
+    ):
+        _start_time = to_zod_compatible_iso(start_time)
+        _end_time = to_zod_compatible_iso(end_time)
+        _standard_logging_object = kwargs.get("standard_logging_object", {})
+        base_api_span = self._create_base_api_span(
+            kwargs,
+            standard_logging_object=_standard_logging_object,
+            start_time=_start_time,
+            end_time=_end_time,
+            is_success=is_success,
+        )
+        trace_api = self._create_trace_api(
+            base_api_span,
+            standard_logging_object=_standard_logging_object,
+            start_time=_start_time,
+            end_time=_end_time,
+            litellm_environment=self.litellm_environment,
+        )
+
+        body = {}
+
+        try:
+            body = trace_api.model_dump(by_alias=True, exclude_none=True)
+        except AttributeError:
+            # Pydantic version below 2.0
+            body = trace_api.dict(by_alias=True, exclude_none=True)
+        return body
+
+    def _sync_event_handler(
+        self, kwargs, response_obj, start_time, end_time, is_success
+    ):
+        body = self._prepare_trace_api(
+            kwargs, response_obj, start_time, end_time, is_success
+        )
+        try:
+            response = self.api.send_request(
+                method=HttpMethods.POST,
+                endpoint=Endpoints.TRACING_ENDPOINT,
+                body=body,
+            )
+        except Exception as e:
+            raise e
+        verbose_logger.debug(
+            "DeepEvalLogger: sync_log_failure_event: Api response", response
+        )
+
+    async def _async_event_handler(
+        self, kwargs, response_obj, start_time, end_time, is_success
+    ):
+        body = self._prepare_trace_api(
+            kwargs, response_obj, start_time, end_time, is_success
+        )
+        response = await self.api.a_send_request(
+            method=HttpMethods.POST,
+            endpoint=Endpoints.TRACING_ENDPOINT,
+            body=body,
+        )
+
+        verbose_logger.debug(
+            "DeepEvalLogger: async_event_handler: Api response", response
+        )
+
+    def _create_base_api_span(
+        self, kwargs, standard_logging_object, start_time, end_time, is_success
+    ):
+        # extract usage
+        usage = standard_logging_object.get("response", {}).get("usage", {})
+        if is_success:
+            output = (
+                standard_logging_object.get("response", {})
+                .get("choices", [{}])[0]
+                .get("message", {})
+                .get("content", "NO_OUTPUT")
+            )
+        else:
+            output = str(standard_logging_object.get("error_string", ""))
+        return BaseApiSpan(
+            uuid=standard_logging_object.get("id", uuid.uuid4()),
+            name=(
+                "litellm_success_callback" if is_success else "litellm_failure_callback"
+            ),
+            status=(
+                TraceSpanApiStatus.SUCCESS if is_success else TraceSpanApiStatus.ERRORED
+            ),
+            type=SpanApiType.LLM,
+            traceUuid=standard_logging_object.get("trace_id", uuid.uuid4()),
+            startTime=str(start_time),
+            endTime=str(end_time),
+            input=kwargs.get("input", "NO_INPUT"),
+            output=output,
+            model=standard_logging_object.get("model", None),
+            inputTokenCount=usage.get("prompt_tokens", None) if is_success else None,
+            outputTokenCount=(
+                usage.get("completion_tokens", None) if is_success else None
+            ),
+        )
+
+    def _create_trace_api(
+        self,
+        base_api_span,
+        standard_logging_object,
+        start_time,
+        end_time,
+        litellm_environment,
+    ):
+        return TraceApi(
+            uuid=standard_logging_object.get("trace_id", uuid.uuid4()),
+            baseSpans=[],
+            agentSpans=[],
+            llmSpans=[base_api_span],
+            retrieverSpans=[],
+            toolSpans=[],
+            startTime=str(start_time),
+            endTime=str(end_time),
+            environment=litellm_environment,
+        )
@@ -0,0 +1,64 @@
+# Duplicate -> https://github.com/confident-ai/deepeval/blob/main/deepeval/tracing/api.py
+from enum import Enum
+from typing import Any, Dict, List, Optional, Union, Literal
+from pydantic import BaseModel, Field
+
+
+class SpanApiType(Enum):
+    BASE = "base"
+    AGENT = "agent"
+    LLM = "llm"
+    RETRIEVER = "retriever"
+    TOOL = "tool"
+
+
+span_api_type_literals = Literal["base", "agent", "llm", "retriever", "tool"]
+
+
+class TraceSpanApiStatus(Enum):
+    SUCCESS = "SUCCESS"
+    ERRORED = "ERRORED"
+
+
+class BaseApiSpan(BaseModel):
+    uuid: str
+    name: Optional[str] = None
+    status: TraceSpanApiStatus
+    type: SpanApiType
+    trace_uuid: str = Field(alias="traceUuid")
+    parent_uuid: Optional[str] = Field(None, alias="parentUuid")
+    start_time: str = Field(alias="startTime")
+    end_time: str = Field(alias="endTime")
+    input: Optional[Union[Dict, list, str]] = None
+    output: Optional[Union[Dict, list, str]] = None
+    error: Optional[str] = None
+
+    # llm
+    model: Optional[str] = None
+    input_token_count: Optional[int] = Field(None, alias="inputTokenCount")
+    output_token_count: Optional[int] = Field(None, alias="outputTokenCount")
+    cost_per_input_token: Optional[float] = Field(None, alias="costPerInputToken")
+    cost_per_output_token: Optional[float] = Field(None, alias="costPerOutputToken")
+
+    class Config:
+        use_enum_values = True
+
+
+class TraceApi(BaseModel):
+    uuid: str
+    base_spans: List[BaseApiSpan] = Field(alias="baseSpans")
+    agent_spans: List[BaseApiSpan] = Field(alias="agentSpans")
+    llm_spans: List[BaseApiSpan] = Field(alias="llmSpans")
+    retriever_spans: List[BaseApiSpan] = Field(alias="retrieverSpans")
+    tool_spans: List[BaseApiSpan] = Field(alias="toolSpans")
+    start_time: str = Field(alias="startTime")
+    end_time: str = Field(alias="endTime")
+    metadata: Optional[Dict[str, Any]] = Field(None)
+    tags: Optional[List[str]] = Field(None)
+    environment: Optional[str] = Field(None)
+
+
+class Environment(Enum):
+    PRODUCTION = "production"
+    DEVELOPMENT = "development"
+    STAGING = "staging"
@@ -0,0 +1,18 @@
+from datetime import datetime, timezone
+from litellm.integrations.deepeval.types import Environment
+
+
+def to_zod_compatible_iso(dt: datetime) -> str:
+    return (
+        dt.astimezone(timezone.utc)
+        .isoformat(timespec="milliseconds")
+        .replace("+00:00", "Z")
+    )
+
+
+def validate_environment(environment: str):
+    if environment not in [env.value for env in Environment]:
+        valid_values = ", ".join(f'"{env.value}"' for env in Environment)
+        raise ValueError(
+            f"Invalid environment: {environment}. Please use one of the following instead: {valid_values}"
+        )
@@ -52,6 +52,7 @@ from litellm.integrations.anthropic_cache_control_hook import AnthropicCacheCont
 from litellm.integrations.arize.arize import ArizeLogger
 from litellm.integrations.custom_guardrail import CustomGuardrail
 from litellm.integrations.custom_logger import CustomLogger
+from litellm.integrations.deepeval.deepeval import DeepEvalLogger
 from litellm.integrations.mlflow import MlflowLogger
 from litellm.integrations.vector_stores.bedrock_vector_store import BedrockVectorStore
 from litellm.litellm_core_utils.get_litellm_params import get_litellm_params
@@ -198,6 +199,7 @@ s3Logger = None
 greenscaleLogger = None
 lunaryLogger = None
 supabaseClient = None
+deepevalLogger = None
 callback_list: Optional[List[str]] = []
 user_logger_fn = None
 additional_details: Optional[Dict[str, str]] = {}
@@ -1723,7 +1725,6 @@ class Logging(LiteLLMLoggingBaseClass):
                                start_time=start_time,
                                end_time=end_time,
                            )
-
                    if (
                        isinstance(callback, CustomLogger)
                        and self.model_call_details.get("litellm_params", {}).get(
@@ -2671,7 +2672,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
+    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

    try:
        for callback in callback_list:
@@ -2955,6 +2956,15 @@ def _init_custom_logger_compatible_class(  # noqa: PLR0915
            galileo_logger = GalileoObserve()
            _in_memory_loggers.append(galileo_logger)
            return galileo_logger  # type: ignore
+        
+        elif logging_integration == "deepeval":
+            for callback in _in_memory_loggers:
+                if isinstance(callback, DeepEvalLogger):
+                    return callback  # type: ignore
+            deepeval_logger = DeepEvalLogger()
+            _in_memory_loggers.append(deepeval_logger)
+            return deepeval_logger  # type: ignore
+        
        elif logging_integration == "logfire":
            if "LOGFIRE_TOKEN" not in os.environ:
                raise ValueError("LOGFIRE_TOKEN not found in environment variables")
@@ -3124,6 +3134,10 @@ def get_custom_logger_compatible_class(  # noqa: PLR0915
            for callback in _in_memory_loggers:
                if isinstance(callback, GalileoObserve):
                    return callback
+        elif logging_integration == "deepeval":
+            for callback in _in_memory_loggers:
+                if isinstance(callback, DeepEvalLogger):
+                    return callback
        elif logging_integration == "langsmith":
            for callback in _in_memory_loggers:
                if isinstance(callback, LangsmithLogger):
@@ -0,0 +1,120 @@
+import unittest
+from unittest.mock import patch, MagicMock
+from datetime import datetime, timezone
+import uuid
+import os
+
+from litellm.integrations.deepeval.deepeval import DeepEvalLogger
+from litellm.integrations.deepeval.api import HttpMethods, Endpoints
+from litellm.integrations.deepeval.types import TraceSpanApiStatus, SpanApiType
+
+
+class TestDeepEvalLogger(unittest.TestCase):
+    @patch.dict(os.environ, {"CONFIDENT_API_KEY": "test-api-key"})
+    def setUp(self):
+        # Mock the Api class before initializing DeepEvalLogger
+        self.api_patcher = patch("litellm.integrations.deepeval.deepeval.Api")
+        self.mock_api_class = self.api_patcher.start()
+        self.mock_api_instance = MagicMock()
+        self.mock_api_class.return_value = self.mock_api_instance
+
+        self.logger = DeepEvalLogger()
+        self.start_time = datetime(2023, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
+        self.end_time = datetime(2023, 1, 1, 12, 0, 1, tzinfo=timezone.utc)
+        self.mock_response_obj = {"id": "resp_123"}
+        self.trace_id = str(uuid.uuid4())
+        self.span_id = str(uuid.uuid4())
+        self.model = "gpt-3.5-turbo"
+        self.input_str = "Hello, world!"
+
+    def tearDown(self):
+        self.api_patcher.stop()
+
+    def _common_assertions(
+        self, expected_status: TraceSpanApiStatus, expected_output: str
+    ):
+        self.mock_api_instance.send_request.assert_called_once()
+        call_args = self.mock_api_instance.send_request.call_args
+
+        self.assertEqual(call_args.kwargs["method"], HttpMethods.POST)
+        self.assertEqual(call_args.kwargs["endpoint"], Endpoints.TRACING_ENDPOINT)
+
+        body = call_args.kwargs["body"]
+
+        self.assertIsInstance(body, dict)
+        self.assertEqual(body["uuid"], self.trace_id)
+        self.assertIn("startTime", body)
+        self.assertIn("endTime", body)
+
+        self.assertIsInstance(body["llmSpans"], list)
+        self.assertEqual(len(body["llmSpans"]), 1)
+
+        llm_span = body["llmSpans"][0]
+        self.assertEqual(llm_span["uuid"], self.span_id)
+
+        expected_name = (
+            "litellm_success_callback"
+            if expected_status == TraceSpanApiStatus.SUCCESS
+            else "litellm_failure_callback"
+        )
+        self.assertEqual(llm_span["name"], expected_name)
+
+        self.assertEqual(llm_span["status"], expected_status.value)
+        self.assertEqual(llm_span["type"], SpanApiType.LLM.value)
+        self.assertEqual(llm_span["traceUuid"], self.trace_id)
+        self.assertIn("startTime", llm_span)
+        self.assertIn("endTime", llm_span)
+        self.assertEqual(llm_span["input"], self.input_str)
+        self.assertEqual(llm_span["output"], expected_output)
+        self.assertEqual(llm_span["model"], self.model)
+
+        return llm_span
+
+    def test_log_success_event(self):
+        kwargs = {
+            "input": self.input_str,
+            "standard_logging_object": {
+                "id": self.span_id,
+                "trace_id": self.trace_id,
+                "model": self.model,
+                "response": {
+                    "usage": {"prompt_tokens": 10, "completion_tokens": 20},
+                    "choices": [{"message": {"content": "This is a success."}}],
+                },
+            },
+        }
+
+        self.logger.log_success_event(
+            kwargs, self.mock_response_obj, self.start_time, self.end_time
+        )
+
+        llm_span = self._common_assertions(
+            TraceSpanApiStatus.SUCCESS, "This is a success."
+        )
+        self.assertEqual(llm_span["inputTokenCount"], 10)
+        self.assertEqual(llm_span["outputTokenCount"], 20)
+
+    def test_log_failure_event(self):
+        error_message = "This is an error."
+        kwargs = {
+            "input": self.input_str,
+            "standard_logging_object": {
+                "id": self.span_id,
+                "trace_id": self.trace_id,
+                "model": self.model,
+                "error_string": error_message,
+                "response": {},
+            },
+        }
+
+        self.logger.log_failure_event(
+            kwargs, self.mock_response_obj, self.start_time, self.end_time
+        )
+
+        llm_span = self._common_assertions(TraceSpanApiStatus.ERRORED, error_message)
+        self.assertIsNone(llm_span.get("inputTokenCount"))
+        self.assertIsNone(llm_span.get("outputTokenCount"))
+
+
+if __name__ == "__main__":
+    unittest.main()