Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .github/e2e-stack/select_tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
r"|^tests/e2e/batches/test_managed_files_enforcement_e2e\.py$"
r"|^tests/e2e/guardrails/test_presidio_masking_e2e\.py$"
r"|^tests/e2e/logging/test_otel_v2_langfuse_generation_output_e2e\.py$"
r"|^tests/e2e/logging/test_langsmith_batch_serialization_e2e\.py$"
)
HARNESS: Final = re.compile(
r"^tests/e2e/[A-Za-z0-9_.-]+\.(py|ini)$"
Expand Down
5 changes: 3 additions & 2 deletions litellm/integrations/langsmith.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
#### What this does ####
# On success, logs events to Langsmith
import asyncio
import json
import os
import random
import traceback
Expand Down Expand Up @@ -415,7 +416,7 @@ async def _log_batch_on_langsmith(
langsmith_api_key: Final = credentials["LANGSMITH_API_KEY"]
langsmith_tenant_id: Final = credentials.get("LANGSMITH_TENANT_ID")
url: Final = self._add_endpoint_to_url(langsmith_api_base, "runs/batch")
headers: Final = {"x-api-key": langsmith_api_key}
headers: Final = {"x-api-key": langsmith_api_key, "Content-Type": "application/json"}
if langsmith_tenant_id:
headers["x-tenant-id"] = langsmith_tenant_id
elements_to_log: Final = [queue_object["data"] for queue_object in queue_objects]
Expand All @@ -426,7 +427,7 @@ async def _log_batch_on_langsmith(
verbose_logger.debug("[LANGSMITH MOCK] Mock mode enabled - API calls will be intercepted")
response: Final = await self.async_httpx_client.post(
url=url,
json={"post": elements_to_log},
content=json.dumps({"post": elements_to_log}, default=str, allow_nan=False),
headers=headers,
)
response.raise_for_status()
Expand Down
3 changes: 3 additions & 0 deletions litellm/llms/custom_httpx/http_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -831,6 +831,7 @@ async def post(
params=params,
headers=headers,
stream=stream,
content=content,
)
finally:
await new_client.aclose()
Expand Down Expand Up @@ -971,6 +972,7 @@ async def patch(
params=params,
headers=headers,
stream=stream,
content=content,
)
finally:
await new_client.aclose()
Expand Down Expand Up @@ -1037,6 +1039,7 @@ async def delete(
params=params,
headers=headers,
stream=stream,
content=content,
)
finally:
await new_client.aclose()
Expand Down
1 change: 1 addition & 0 deletions tests/e2e/coverage_registry/logging.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
- {id: logging.otel.failure.exports_metric, module: logging, tier: P0, event: failure, assertions: [exports_metric], exercised_on: [chat_completions, messages], source: "integrations/otel/logger.py", rationale: "Error spans for observability continuity"}
- {id: logging.braintrust.success.logs_spend, module: logging, tier: P1, event: success, assertions: [logs_spend], exercised_on: [chat_completions, messages], source: "integrations/braintrust_logging.py", rationale: "Evals platform spend"}
- {id: logging.langsmith.success.logs_spend, module: logging, tier: P1, event: success, assertions: [logs_spend], exercised_on: [chat_completions, messages], source: "integrations/langsmith.py", rationale: "LangChain ecosystem"}
- {id: logging.langsmith.success.serializes_non_native_metadata, module: logging, tier: P1, event: success, assertions: [serializes_non_native_metadata], exercised_on: [sdk], source: "integrations/langsmith.py", rationale: "datetime/Decimal/UUID metadata used to TypeError in json.dumps and drop the whole batch (LIT-8310)"}
- {id: logging.arize.success.logs_spend, module: logging, tier: P1, event: success, assertions: [logs_spend], exercised_on: [chat_completions, embeddings], source: "integrations/arize/arize.py", rationale: "ML-ops observability"}
- {id: logging.mlflow.success.logs_spend, module: logging, tier: P1, event: success, assertions: [logs_spend], exercised_on: [chat_completions], source: "integrations/mlflow.py", rationale: "Experiment tracking cost/run"}
- {id: logging.opik.success.logs_spend, module: logging, tier: P1, event: success, assertions: [logs_spend], exercised_on: [chat_completions], source: "integrations/opik/opik.py", rationale: "Eval platform spend/case"}
Expand Down
125 changes: 125 additions & 0 deletions tests/e2e/logging/test_langsmith_batch_serialization_e2e.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,125 @@
"""Live e2e: a LangSmith batch whose metadata holds non JSON-native Python values
(datetime, Decimal) must reach the real LangSmith API instead of dying in
json.dumps and dropping the whole batch. Only the SDK path can put such values
into the batch (the proxy JSON-decodes request metadata), so this test drives
litellm.acompletion in-process against the real OpenAI API with a LangsmithLogger
injected per request, flushes the batch, and reads the run back by id through
LangSmith's own API. Nothing is mocked.
"""

from __future__ import annotations

import asyncio
import datetime
import decimal
import os
import time
import uuid
from dataclasses import dataclass
from typing import Final

import pytest
from e2e_config import CHEAP_OPENAI_MODEL, POLL_INTERVAL, POLL_TIMEOUT, unique_marker
from e2e_http import Headers, Success, get_external
from pydantic import BaseModel, ConfigDict, Field, JsonValue

import litellm
from litellm.integrations.langsmith import LangsmithLogger

pytestmark = pytest.mark.e2e


class LangsmithHeaders(Headers):
x_api_key: str = Field(serialization_alias="x-api-key")


class LangsmithRunExtra(BaseModel):
model_config = ConfigDict(extra="allow")
requester_metadata: dict[str, JsonValue] | None = None


class LangsmithRun(BaseModel):
id: str
session_name: str | None = None
extra: LangsmithRunExtra


@dataclass(frozen=True, slots=True)
class LangsmithCreds:
api_key: str
base_url: str
project: str


def load_langsmith_creds() -> LangsmithCreds:
api_key = os.getenv("LANGSMITH_API_KEY")
if not api_key:
pytest.fail("LangSmith e2e requires LANGSMITH_API_KEY; missing credentials is a hard failure, not a skip")
if os.getenv("LANGSMITH_MOCK"):
pytest.fail("LANGSMITH_MOCK is set; this e2e must hit the real LangSmith API")
return LangsmithCreds(
api_key=api_key,
base_url=(os.getenv("LANGSMITH_BASE_URL") or "https://api.smith.langchain.com").rstrip("/"),
project=os.getenv("LANGSMITH_PROJECT") or "litellm-e2e",
)


def _fetch_run(creds: LangsmithCreds, run_id: uuid.UUID) -> LangsmithRun | None:
result = get_external(
f"{creds.base_url}/runs/{run_id}",
response_type=LangsmithRun,
headers=LangsmithHeaders(x_api_key=creds.api_key),
)
match result:
case Success(data=run):
return run
case _:
return None


def _poll_run(creds: LangsmithCreds, run_id: uuid.UUID) -> LangsmithRun:
deadline: Final = time.monotonic() + POLL_TIMEOUT
while time.monotonic() < deadline:
run = _fetch_run(creds, run_id)
if run is not None:
return run
time.sleep(POLL_INTERVAL)
pytest.fail(f"LangSmith run {run_id} never appeared within {POLL_TIMEOUT}s; the batch flush dropped it")


class TestLangsmithBatchSerialization:
@pytest.mark.asyncio
@pytest.mark.covers("logging.langsmith.success.serializes_non_native_metadata")
async def test_non_json_native_metadata_reaches_langsmith(self) -> None:
creds: Final = load_langsmith_creds()
logger: Final = LangsmithLogger(
langsmith_api_key=creds.api_key, langsmith_project=creds.project, langsmith_base_url=creds.base_url
)
assert not logger.is_mock_mode, "LangsmithLogger initialised in mock mode; this e2e needs the real API"
marker: Final = unique_marker()
run_id: Final = uuid.uuid4()
created_at: Final = datetime.datetime(2026, 1, 2, 3, 4, 5, tzinfo=datetime.timezone.utc)
spend: Final = decimal.Decimal("0.0042")
response: Final = await litellm.acompletion(
model=f"openai/{CHEAP_OPENAI_MODEL}",
messages=[{"role": "user", "content": f"Reply with the single word ok ({marker})"}],
max_completion_tokens=5,
callbacks=[logger],
metadata={"run_id": str(run_id), "metadata": {"marker": marker, "created_at": created_at, "spend": spend}},
)
assert isinstance(response, litellm.ModelResponse) and response.id, (
"a non-streaming completion must return a ModelResponse before the batch flush is meaningful"
)
enqueue_deadline: Final = time.monotonic() + POLL_TIMEOUT
while time.monotonic() < enqueue_deadline and len(logger.log_queue) == 0:
await asyncio.sleep(0.5)
assert len(logger.log_queue) == 1, (
f"the completion must be queued for the LangSmith batch, got {len(logger.log_queue)} queued entries"
)
await logger.async_send_batch()
run: Final = _poll_run(creds, run_id)
requester_metadata: Final = run.extra.requester_metadata
assert requester_metadata is not None, "the run must carry the caller metadata under extra.requester_metadata"
assert requester_metadata["marker"] == marker
assert requester_metadata["created_at"] == str(created_at)
assert requester_metadata["spend"] == str(spend)
8 changes: 5 additions & 3 deletions tests/logging_callback_tests/test_langsmith_unit_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -356,10 +356,12 @@ async def test_langsmith_key_based_logging():
# tenant_id should not be in headers if not provided
assert "x-tenant-id" not in call_args[1]["headers"]

assert call_args[1]["headers"]["Content-Type"] == "application/json"

# Verify the request body contains the expected data
request_body = call_args[1]["json"]
request_body = json.loads(call_args[1]["content"])
assert "post" in request_body
assert len(request_body["post"]) == 1 # Should contain one run
assert len(request_body["post"]) == 1

# EXPECTED BODY
expected_body = {
Expand Down Expand Up @@ -404,7 +406,7 @@ async def test_langsmith_key_based_logging():
}

# Print both bodies for debugging
actual_body = call_args[1]["json"]
actual_body = json.loads(call_args[1]["content"])
print("\nExpected body:")
print(json.dumps(expected_body, indent=2))
print("\nActual body:")
Expand Down
127 changes: 122 additions & 5 deletions tests/test_litellm/integrations/test_langsmith_init.py
Original file line number Diff line number Diff line change
@@ -1,13 +1,17 @@
import asyncio
import json
import os
from datetime import datetime, timezone
from decimal import Decimal
from typing import Final
from unittest.mock import AsyncMock, MagicMock, patch

import httpx
import pytest


import litellm
from litellm.integrations.langsmith import LangsmithLogger
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
from litellm.types.integrations.langsmith import LangsmithQueueObject


Expand Down Expand Up @@ -219,6 +223,121 @@ async def test_async_log_failure_event_lazily_starts_periodic_flush(self):
assert len(logger.log_queue) == 1


class TestLangsmithBatchSerialization:
async def _logger(self, transport_handler, tenant_id=None):
logger = LangsmithLogger(
langsmith_api_key="test-key",
langsmith_project="test-project",
langsmith_base_url="https://api.smith.langchain.com",
langsmith_tenant_id=tenant_id,
)
if logger._flush_task is not None:
logger._flush_task.cancel()
handler = AsyncHTTPHandler()
await handler.client.aclose()
handler.client = httpx.AsyncClient(
transport=httpx.MockTransport(transport_handler)
)
logger.async_httpx_client = handler
return logger

@staticmethod
def _capturing_transport(captured):
async def handle(request: httpx.Request) -> httpx.Response:
captured.append(request)
return httpx.Response(200, request=request, json={"ok": True})

return handle

def _queue(self, logger, extra):
return [
LangsmithQueueObject(
data={"id": "run-1", "name": "LLMRun", "extra": extra},
credentials=logger.default_credentials,
)
]

@pytest.mark.asyncio
async def test_datetime_and_decimal_metadata_reach_langsmith_as_strings(self):
captured: list[httpx.Request] = [] # mutable-ok: transport capture buffer
logger = await self._logger(self._capturing_transport(captured))
logger.log_queue = self._queue(
logger,
{
"created_at": datetime(2026, 1, 2, 3, 4, 5, tzinfo=timezone.utc),
"spend": Decimal("0.0042"),
},
)

await logger.async_send_batch()

assert len(captured) == 1, "batch was dropped instead of being sent"
body = json.loads(captured[0].content)
assert body["post"][0]["extra"] == {
"created_at": "2026-01-02 03:04:05+00:00",
"spend": "0.0042",
}
await logger.async_httpx_client.client.aclose()

@pytest.mark.asyncio
async def test_nan_metadata_is_dropped_instead_of_shipping_invalid_json(self):
captured: list[httpx.Request] = [] # mutable-ok: transport capture buffer
logger = await self._logger(self._capturing_transport(captured))
logger.log_queue = self._queue(logger, {"score": float("nan")})

await logger.async_send_batch()

assert captured == [], (
"nan metadata must abort the batch: a bare NaN token is invalid JSON and LangSmith rejects it"
)
await logger.async_httpx_client.client.aclose()

@pytest.mark.asyncio
async def test_batch_declares_json_content_type(self):
captured: list[httpx.Request] = [] # mutable-ok: transport capture buffer
logger = await self._logger(self._capturing_transport(captured))
logger.log_queue = self._queue(logger, {"model": "gpt-4.1-mini"})

await logger.async_send_batch()

assert captured[0].headers["content-type"] == "application/json", (
"a content= body carries no implicit content type; LangSmith refuses it without this header"
)
assert captured[0].url.path.endswith("/api/v1/runs/batch")
assert captured[0].headers["x-api-key"] == "test-key"
assert "x-tenant-id" not in captured[0].headers
await logger.async_httpx_client.client.aclose()

@pytest.mark.asyncio
async def test_tenant_id_is_forwarded_on_the_batch_request(self):
captured: list[httpx.Request] = [] # mutable-ok: transport capture buffer
logger = await self._logger(
self._capturing_transport(captured), tenant_id="tenant-1"
)
logger.log_queue = self._queue(logger, {"model": "gpt-4.1-mini"})

await logger.async_send_batch()

assert captured[0].headers["x-tenant-id"] == "tenant-1"
await logger.async_httpx_client.client.aclose()

@pytest.mark.asyncio
async def test_langsmith_error_response_does_not_propagate(self):
captured: list[httpx.Request] = [] # mutable-ok: transport capture buffer

async def reject(request: httpx.Request) -> httpx.Response:
captured.append(request)
return httpx.Response(422, request=request, text="bad run")

logger = await self._logger(reject)
logger.log_queue = self._queue(logger, {"model": "gpt-4.1-mini"})

await logger.async_send_batch()

assert len(captured) == 1, "the batch never left the process"
await logger.async_httpx_client.client.aclose()


class TestLangsmithPrepareLogData:
"""Regression test for #24001: _prepare_log_data must inject
usage_metadata into outputs so LangSmith's Cost column is populated."""
Expand Down Expand Up @@ -544,12 +663,10 @@ async def test_events_appended_during_flush_are_not_dropped():
credentials=logger.default_credentials, data={"id": "late"}
)

async def fake_post(
url: str, json: dict[str, list[dict[str, str]]], headers: dict[str, str]
) -> MagicMock:
async def fake_post(url: str, content: str, headers: dict[str, str]) -> MagicMock:
if not sent_batches:
logger.log_queue.append(late_event)
sent_batches.append(json["post"])
sent_batches.append(json.loads(content)["post"])
response = MagicMock()
response.status_code = 200
response.raise_for_status = MagicMock()
Expand Down
Loading
Loading