fix(langfuse): trace successful decisions calls with their token counts

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Rodrigo Azevedo 2026-10-27 10:39:57 -03:00
parent cfe0100e1f
commit 2c23948072
2 changed files with 57 additions and 3 deletions

View file

@ -15,6 +15,7 @@ from packaging.version import Version
import litellm
from litellm._logging import verbose_logger
from litellm.constants import MAX_LANGFUSE_INITIALIZED_CLIENTS
from litellm.cost_calculator import get_usage_object
from litellm.integrations.langfuse.langfuse_mock_client import (
create_mock_langfuse_client,
should_use_langfuse_mock,
@ -32,6 +33,7 @@ from litellm.litellm_core_utils.redact_messages import redact_user_api_key_info
from litellm.llms.custom_httpx.http_handler import get_httpx_client
from litellm.responses.utils import ResponseAPILoggingUtils
from litellm.secret_managers.main import str_to_bool
from litellm.types.decisions import DecisionsResponse, OpenAIDecisionResponse
from litellm.types.integrations.langfuse import *
from litellm.types.llms.openai import HttpxBinaryResponseContent, ResponseAPIUsage, ResponsesAPIResponse
from litellm.types.utils import (
@ -584,7 +586,9 @@ class LangFuseLogger:
| TranscriptionResponse
| RerankResponse
| HttpxBinaryResponseContent
| ResponsesAPIResponse,
| ResponsesAPIResponse
| DecisionsResponse
| OpenAIDecisionResponse,
prompt: dict,
level: str,
status_message: str | None,
@ -634,6 +638,9 @@ class LangFuseLogger:
elif response_obj is not None and isinstance(response_obj, litellm.ResponsesAPIResponse):
input = prompt
output = self._get_responses_api_content_for_langfuse(response_obj)
elif response_obj is not None and isinstance(response_obj, (DecisionsResponse, OpenAIDecisionResponse)):
input = prompt
output = response_obj.model_dump(mode="json", exclude={"usage"})
elif (
kwargs.get("call_type") is not None
and kwargs.get("call_type") == "_arealtime"
@ -859,10 +866,14 @@ class LangFuseLogger:
usage = None
usage_details = None
if response_obj is not None:
if hasattr(response_obj, "id") and response_obj.get("id", None) is not None:
if isinstance(response_obj, _ResponseWithId) and getattr(response_obj, "id", None) is not None:
generation_id = _logging_id(start_time, response_obj)
_raw_usage_obj: Final[_UsageObject | None] = getattr(response_obj, "usage", None)
_usage_obj: Final = _chat_usage(_raw_usage_obj)
_usage_obj: Final = (
get_usage_object(response_obj)
if isinstance(response_obj, (DecisionsResponse, OpenAIDecisionResponse))
else _chat_usage(_raw_usage_obj)
)
if _usage_obj:
# Safely get usage values, defaulting None to 0 for Langfuse compatibility.

View file

@ -433,6 +433,49 @@ async def test_langfuse_logging_vllm_embedding(
assert route.call_count == 1
@pytest.mark.asyncio
@pytest.mark.parametrize("response_id", [None, "decision-1"])
async def test_langfuse_logging_systemone(
langfuse_export: tuple[respx.Route, str], respx_mock: respx.MockRouter, response_id: str | None
) -> None:
route, trace_id = langfuse_export
answers: Final = {"urgent": {"type": "noul", "noul": 0.87}}
upstream: Final = respx_mock.post("https://openrouter.ai/api/alpha/decisions").respond(
json={
**({"id": response_id} if response_id is not None else {}),
"model": "typesafe/jev-1.13",
"answers": answers,
"usage": {"input_tokens": 283, "output_tokens": 20},
}
)
await litellm.asystemone(
model="openrouter/typesafe/jev-1.13",
state="Customer writes: my payout has failed three days in a row.",
questions={"urgent": {"type": "noul", "instructions": "Is this urgent?"}},
api_key="caller-key",
metadata={"trace_id": trace_id},
)
await _flush_langfuse()
assert upstream.call_count == 1
generations: Final = tuple(
TypeAdapter(dict[str, object]).validate_python(span["attributes"])
for span in _exported_spans(route)
if span["trace_id"] == resolve_trace_id(trace_id)
and TypeAdapter(dict[str, object]).validate_python(span["attributes"]).get("langfuse.observation.type")
== "generation"
)
assert len(generations) == 1
assert generations[0]["langfuse.observation.usage_details"] == {
"input": 283,
"output": 20,
"total": 303,
"cache_creation_input_tokens": 0,
"cache_read_input_tokens": 0,
}
output: Final = TypeAdapter(dict[str, object]).validate_python(generations[0]["langfuse.observation.output"])
assert output["answers"] == answers
@pytest.mark.asyncio
async def test_langfuse_logging_with_router(langfuse_export: tuple[respx.Route, str]) -> None:
route, trace_id = langfuse_export