diff --git a/backend/app/core/langfuse/langfuse.py b/backend/app/core/langfuse/langfuse.py index 27d5e0224..dce85ec91 100644 --- a/backend/app/core/langfuse/langfuse.py +++ b/backend/app/core/langfuse/langfuse.py @@ -10,15 +10,32 @@ from langfuse import Langfuse, LangfuseOtelSpanAttributes from langfuse._client.span import LangfuseGeneration, LangfuseSpan from langfuse.api.core.api_error import ApiError +from opentelemetry.sdk.resources import SERVICE_NAME, Resource from opentelemetry.sdk.trace import TracerProvider +from app.core.config import settings +from app.core.telemetry.constants import GEN_AI_RESPONSE_ID, LLM_COMPLETION_TYPE +from app.core.telemetry.llm import billable_output_tokens from app.models.llm import ( + AudioContent, AudioOutput, + ImageContent, LLMCallResponse, NativeCompletionConfig, + PDFContent, QueryParams, + TextInput, TextOutput, ) +from app.models.llm.request import QueryInput + +TraceInput = dict[str, object] | list[dict[str, object]] +ModelParameterValue = str | int | float | bool | None + +# Prompt text belongs in the generation input, not in model parameters. +_PROMPT_PARAM_KEYS = frozenset( + {"model", "instructions", "input", "messages", "system", "director_notes"} +) logger = logging.getLogger(__name__) @@ -38,7 +55,9 @@ def get_langfuse_tracer_provider() -> TracerProvider: if _LANGFUSE_TRACER_PROVIDER is None: with _LANGFUSE_TRACER_PROVIDER_LOCK: if _LANGFUSE_TRACER_PROVIDER is None: - _LANGFUSE_TRACER_PROVIDER = TracerProvider() + _LANGFUSE_TRACER_PROVIDER = TracerProvider( + resource=Resource.create({SERVICE_NAME: settings.OTEL_SERVICE_NAME}) + ) return _LANGFUSE_TRACER_PROVIDER @@ -117,6 +136,117 @@ def set_trace_attributes( span._otel_span.set_attributes(attrs) +def _media_content_metadata( + content: AudioContent | ImageContent | PDFContent, +) -> dict[str, object]: + """Media fields without the base64 or URL payload, which is large and may be private.""" + return { + "format": content.format, + "mime_type": content.mime_type, + "length": len(content.value), + } + + +def _query_input_item(query_input: str | QueryInput) -> dict[str, object]: + if isinstance(query_input, str): + return {"type": "text", "content": query_input, "format": "text"} + if isinstance(query_input, TextInput): + return { + "type": "text", + "content": query_input.content.value, + "format": query_input.content.format, + "language_code": query_input.content.language_code, + } + contents = ( + query_input.content + if isinstance(query_input.content, list) + else [query_input.content] + ) + media = [_media_content_metadata(content) for content in contents] + if len(media) == 1: + return {"type": query_input.type, **media[0]} + return {"type": query_input.type, "items": media} + + +def extract_query_input( + query_input: str | QueryInput | list[QueryInput], +) -> TraceInput: + """Structured trace input (type, format, language code) instead of the Pydantic repr. + + Text keeps its content; audio, image and PDF send metadata only, like extract_response_output. + """ + if isinstance(query_input, list): + return [_query_input_item(item) for item in query_input] + return _query_input_item(query_input) + + +def _input_metadata(trace_input: TraceInput) -> dict[str, str]: + """Input type, format and language code as filterable metadata.""" + items = trace_input if isinstance(trace_input, list) else [trace_input] + types: dict[str, None] = {} + formats: dict[str, None] = {} + languages: dict[str, None] = {} + for item in items: + types[str(item["type"])] = None + parts = item.get("items") + for part in parts if isinstance(parts, list) else [item]: + if part.get("format"): + formats[str(part["format"])] = None + if item.get("language_code"): + languages[str(item["language_code"])] = None + metadata = { + "llm.input.type": ",".join(types), + "llm.input.format": ",".join(formats), + } + if languages: + metadata["llm.input.language_code"] = ",".join(languages) + return metadata + + +def _output_metadata(llm_output: TextOutput | AudioOutput | None) -> dict[str, str]: + if isinstance(llm_output, TextOutput): + metadata = { + "llm.output.type": "text", + "llm.output.format": llm_output.content.format, + } + if llm_output.content.language_code: + metadata["llm.output.language_code"] = llm_output.content.language_code + return metadata + if isinstance(llm_output, AudioOutput): + return { + "llm.output.type": "audio", + "llm.output.format": llm_output.content.format, + } + return {} + + +def _model_parameters(params: dict[str, object]) -> dict[str, ModelParameterValue]: + """Scalar config params, one nested level flattened (``reasoning.effort``).""" + scalars = (str, int, float, bool) + result: dict[str, ModelParameterValue] = {} + for key, value in params.items(): + if key in _PROMPT_PARAM_KEYS: + continue + if isinstance(value, scalars): + result[key] = value + elif isinstance(value, dict): + for sub_key, sub_value in value.items(): + if sub_key not in _PROMPT_PARAM_KEYS and isinstance(sub_value, scalars): + result[f"{key}.{sub_key}"] = sub_value + return result + + +def _generation_input(params: dict[str, object], trace_input: TraceInput) -> object: + """Chat-format input so Langfuse shows the system prompt.""" + instructions = params.get("instructions") + if not isinstance(instructions, str) or not instructions: + return trace_input + return [ + {"role": "system", "content": instructions}, + {"role": "user", "content": trace_input}, + ] + + def extract_response_output( llm_output: TextOutput | AudioOutput | None, ) -> str | dict[str, Any]: @@ -174,6 +304,7 @@ def __init__( secret_key=credentials["secret_key"], host=credentials["host"], tracing_enabled=True, # This ensures the client is active + environment=settings.ENVIRONMENT, tracer_provider=get_langfuse_tracer_provider(), ) except Exception as e: @@ -301,12 +432,14 @@ def flush(self) -> None: def observe_llm_execution( session_id: str | None = None, credentials: dict | None = None, + trace_metadata: dict[str, str] | None = None, ) -> Callable: """Decorator to add Langfuse observability to LLM provider execute methods. Args: credentials: Langfuse credentials with public_key, secret_key, and host session_id: Session ID for grouping traces (conversation_id) + trace_metadata: Call IDs, same keys as the Sentry spans (_CallContext.identity_attributes) Usage: decorated_execute = observe_llm_execution( @@ -330,6 +463,7 @@ def wrapper( public_key=credentials.get("public_key"), secret_key=credentials.get("secret_key"), host=credentials.get("host"), + environment=settings.ENVIRONMENT, tracer_provider=get_langfuse_tracer_provider(), ) logger.info( @@ -356,19 +490,25 @@ def langfuse_call(fn, *args, **kwargs): failed = True return None + params = completion_config.params + trace_input = extract_query_input(query.input) trace = langfuse_call( langfuse.start_observation, as_type="span", name="unified-llm-call", - input=query.input, + input=trace_input, + metadata=trace_metadata, ) if trace: langfuse_call( set_trace_attributes, trace, name="unified-llm-call", - input=query.input, - tags=[completion_config.provider], + input=trace_input, + session_id=session_id, + # OTel drops the whole list if any item is an Enum, not a plain str. + tags=[completion_config.provider, str(completion_config.type)], + metadata=trace_metadata, ) generation = None @@ -377,57 +517,77 @@ def langfuse_call(fn, *args, **kwargs): trace.start_observation, as_type="generation", name=f"{completion_config.provider}-completion", - input=query.input, - model=( - completion_config.params.get("model") - if isinstance(completion_config.params, dict) - else getattr(completion_config.params, "model", None) - ), + input=_generation_input(params, trace_input), + model=params.get("model"), + model_parameters=_model_parameters(params), + metadata={ + LLM_COMPLETION_TYPE: str(completion_config.type), + **_input_metadata(trace_input), + }, ) - response: LLMCallResponse | None - error: str | None - response, error = func(completion_config, query, **kwargs) - - if response: - if generation: - langfuse_call( - generation.update, - output={ - "status": "success", - "output": extract_response_output(response.response.output), - }, - usage_details={ - "input": response.usage.input_tokens, - "output": response.usage.output_tokens, - }, - model=response.response.model, - ) - langfuse_call(generation.end) + def finish_failure(error_msg: str) -> None: + output = {"status": "failure", "error": error_msg} + for observation in (generation, trace): + if observation: + langfuse_call( + observation.update, + output=output, + level="ERROR", + status_message=error_msg, + ) if trace: langfuse_call( set_trace_attributes, trace, - output={ - "status": "success", - "output": extract_response_output(response.response.output), - }, - session_id=session_id or response.response.conversation_id, - ) - langfuse_call(trace.end) - else: - error_msg = error or "Unknown error" - if generation: - langfuse_call(generation.update, output={"error": error_msg}) - langfuse_call(generation.end) - if trace: - langfuse_call( - set_trace_attributes, - trace, - output={"status": "failure", "error": error_msg}, + output=output, session_id=session_id, ) - langfuse_call(trace.end) + for observation in (generation, trace): + if observation: + langfuse_call(observation.end) + langfuse_call(langfuse.flush) + + response: LLMCallResponse | None + error: str | None + try: + response, error = func(completion_config, query, **kwargs) + except Exception as e: + finish_failure(str(e)) + raise + + if not response: + finish_failure(error or "Unknown error") + return response, error + + output = { + "status": "success", + "output": extract_response_output(response.response.output), + } + if generation: + langfuse_call( + generation.update, + output=output, + usage_details={ + "input": response.usage.input_tokens, + "output": billable_output_tokens(response.usage), + }, + model=response.response.model, + metadata={ + GEN_AI_RESPONSE_ID: response.response.provider_response_id, + **_output_metadata(response.response.output), + }, + ) + langfuse_call(generation.end) + if trace: + langfuse_call(trace.update, output=output) + langfuse_call( + set_trace_attributes, + trace, + output=output, + session_id=session_id or response.response.conversation_id, + ) + langfuse_call(trace.end) langfuse_call(langfuse.flush) return response, error diff --git a/backend/app/core/telemetry/constants.py b/backend/app/core/telemetry/constants.py index 492847109..4ded3473d 100644 --- a/backend/app/core/telemetry/constants.py +++ b/backend/app/core/telemetry/constants.py @@ -13,6 +13,7 @@ GEN_AI_REQUEST_PRESENCE_PENALTY, GEN_AI_REQUEST_TEMPERATURE, GEN_AI_REQUEST_TOP_P, + GEN_AI_RESPONSE_ID, GEN_AI_RESPONSE_MODEL, GEN_AI_SYSTEM, GEN_AI_USAGE_INPUT_TOKENS, @@ -38,6 +39,13 @@ GEN_AI_REQUEST_AVAILABLE_TOOLS = "gen_ai.request.available_tools" GEN_AI_USAGE_TOTAL_TOKENS = "gen_ai.usage.total_tokens" +# --- LLM call: same keys on Sentry spans and Langfuse trace metadata --- +LLM_CALL_ID = "llm.call_id" +LLM_CHAIN_ID = "llm.chain_id" +LLM_CONFIG_ID = "llm.config.id" +LLM_CONFIG_VERSION = "llm.config.version" +LLM_COMPLETION_TYPE = "llm.completion_type" + # --- HTTP --- HTTP_METHOD = "http.method" HTTP_ROUTE = "http.route" diff --git a/backend/app/core/telemetry/llm.py b/backend/app/core/telemetry/llm.py index 1013627b0..c928152bd 100644 --- a/backend/app/core/telemetry/llm.py +++ b/backend/app/core/telemetry/llm.py @@ -32,7 +32,18 @@ from app.core.telemetry.metrics import emit_sentry_metric if TYPE_CHECKING: - from app.models.llm.response import LLMCallResponse + from app.models.llm.response import LLMCallResponse, Usage + + +def billable_output_tokens(usage: "Usage") -> int: + """Output plus reasoning when the provider reports them separately (Gemini), so cost is right.""" + reasoning_tokens = usage.reasoning_tokens or 0 + if ( + reasoning_tokens + and usage.input_tokens + usage.output_tokens < usage.total_tokens + ): + return usage.output_tokens + reasoning_tokens + return usage.output_tokens def _llm_call_attrs( @@ -144,15 +155,13 @@ def set_gen_ai_response_attributes( """Set OTel GenAI response attributes (usage, model) on `span`.""" usage = response.usage if usage: - reasoning_tokens = usage.reasoning_tokens or 0 - output_tokens = usage.output_tokens - if reasoning_tokens and usage.input_tokens + output_tokens < usage.total_tokens: - output_tokens += reasoning_tokens span.set_attribute(GEN_AI_USAGE_INPUT_TOKENS, usage.input_tokens) - span.set_attribute(GEN_AI_USAGE_OUTPUT_TOKENS, output_tokens) + span.set_attribute(GEN_AI_USAGE_OUTPUT_TOKENS, billable_output_tokens(usage)) span.set_attribute(GEN_AI_USAGE_TOTAL_TOKENS, usage.total_tokens) - if reasoning_tokens: - span.set_attribute(GEN_AI_USAGE_REASONING_OUTPUT_TOKENS, reasoning_tokens) + if usage.reasoning_tokens: + span.set_attribute( + GEN_AI_USAGE_REASONING_OUTPUT_TOKENS, usage.reasoning_tokens + ) if response.response and response.response.model: span.set_attribute(GEN_AI_RESPONSE_MODEL, response.response.model) diff --git a/backend/app/services/llm/jobs.py b/backend/app/services/llm/jobs.py index 34f91b86f..ee0995f6f 100644 --- a/backend/app/services/llm/jobs.py +++ b/backend/app/services/llm/jobs.py @@ -25,6 +25,11 @@ GEN_AI_PROVIDER_NAME, GEN_AI_REQUEST_MODEL, JOB_ID, + LLM_CALL_ID, + LLM_CHAIN_ID, + LLM_COMPLETION_TYPE, + LLM_CONFIG_ID, + LLM_CONFIG_VERSION, TENANT_ORG_ID, TENANT_PROJECT_ID, ) @@ -116,9 +121,9 @@ def _set_traceability_attributes( if job_id is not None: span.set_attribute(JOB_ID, str(job_id)) if llm_call_id is not None: - span.set_attribute("llm.call_id", str(llm_call_id)) + span.set_attribute(LLM_CALL_ID, str(llm_call_id)) if chain_id is not None: - span.set_attribute("llm.chain_id", str(chain_id)) + span.set_attribute(LLM_CHAIN_ID, str(chain_id)) if trace_id is not None: span.set_attribute(CORRELATION_ID, trace_id) if project_id is not None: @@ -136,6 +141,7 @@ def _execute_provider_call( query: QueryParams, credentials: dict | None, session_id: str | None, + trace_metadata: dict[str, str] | None = None, **kwargs: Any, ) -> tuple[Any, Any]: kwargs.pop("organization_id", None) @@ -145,6 +151,7 @@ def _execute_provider_call( decorated = observe_llm_execution( session_id=session_id, credentials=credentials, + trace_metadata=trace_metadata, )(func) with suppress_http_instrumentation(): @@ -604,21 +611,29 @@ class _CallContext: trace_id: str | None record_call: bool llm_call_id: UUID | None = None + config_id: UUID | None = None + config_version: int | None = None started_at: float = 0.0 + def identity_attributes(self) -> dict[str, str]: + """IDs of this call, with the same keys on Sentry spans and Langfuse trace metadata.""" + ids = { + JOB_ID: self.job_id, + LLM_CALL_ID: self.llm_call_id, + LLM_CHAIN_ID: self.chain_id, + LLM_CONFIG_ID: self.config_id, + LLM_CONFIG_VERSION: self.config_version, + CORRELATION_ID: self.trace_id, + TENANT_ORG_ID: self.organization_id, + TENANT_PROJECT_ID: self.project_id, + } + return {key: str(value) for key, value in ids.items() if value is not None} + @contextmanager def span(self, name: str) -> Iterator[trace.Span]: span_tracer = tracer if self.record_call else trace.NoOpTracer() with span_tracer.start_as_current_span(name) as span: - _set_traceability_attributes( - span, - job_id=self.job_id, - llm_call_id=self.llm_call_id, - chain_id=self.chain_id, - trace_id=self.trace_id, - project_id=self.project_id, - organization_id=self.organization_id, - ) + span.set_attributes(self.identity_attributes()) yield span def metrics_started(self, *, provider: str, model: str) -> None: @@ -663,8 +678,9 @@ def _load_config_blob( with ctx.span("llm.resolve_config") as cfg_span: cfg_span.set_attribute("llm.config.is_stored", config.is_stored_config) if config.is_stored_config: - cfg_span.set_attribute("llm.config.id", str(config.id)) - cfg_span.set_attribute("llm.config.version", str(config.version)) + ctx.config_id, ctx.config_version = config.id, config.version + cfg_span.set_attribute(LLM_CONFIG_ID, str(config.id)) + cfg_span.set_attribute(LLM_CONFIG_VERSION, str(config.version)) config_crud = ConfigVersionCrud( session=session, project_id=ctx.project_id, config_id=config.id ) @@ -1354,7 +1370,7 @@ def execute_llm_call( with ctx.span(ai_span_name) as ai_span: ai_span.set_attribute("sentry.op", "gen_ai.chat") if completion_type: - ai_span.set_attribute("llm.completion_type", completion_type) + ai_span.set_attribute(LLM_COMPLETION_TYPE, completion_type) set_gen_ai_request_attributes( ai_span, provider=provider_name, @@ -1376,7 +1392,7 @@ def execute_llm_call( ) if completion_type: provider_span.set_attribute( - "llm.completion_type", completion_type + LLM_COMPLETION_TYPE, completion_type ) if model_name: provider_span.set_attribute(GEN_AI_REQUEST_MODEL, model_name) @@ -1387,6 +1403,11 @@ def execute_llm_call( query=query, credentials=langfuse_credentials, session_id=conversation_id, + trace_metadata={ + **ctx.identity_attributes(), + GEN_AI_PROVIDER_NAME: provider_name, + GEN_AI_REQUEST_MODEL: model_name, + }, organization_id=organization_id, project_id=project_id, telemetry_span=provider_span, diff --git a/backend/app/tests/core/test_langfuse/test_langfuse_tracer.py b/backend/app/tests/core/test_langfuse/test_langfuse_tracer.py index 9b86927f3..c802cbd70 100644 --- a/backend/app/tests/core/test_langfuse/test_langfuse_tracer.py +++ b/backend/app/tests/core/test_langfuse/test_langfuse_tracer.py @@ -6,8 +6,16 @@ from openai import OpenAIError -from app.core.langfuse.langfuse import LangfuseTracer +from app.core.langfuse.langfuse import LangfuseTracer, extract_query_input from app.models import Assistant, ResponsesAPIRequest +from app.models.llm import ( + AudioContent, + AudioInput, + ImageContent, + ImageInput, + TextContent, + TextInput, +) from app.services.response.response import generate_response, process_response @@ -653,3 +661,57 @@ def test_handles_openai_error_gracefully( ) assert result.success is False + + +class TestExtractQueryInput: + """Trace input is structured data, never the Pydantic repr or a media payload.""" + + def test_text_input_keeps_content_and_metadata(self) -> None: + result = extract_query_input( + TextInput(content=TextContent(value="why are you here", language_code="hi")) + ) + + assert result == { + "type": "text", + "content": "why are you here", + "format": "text", + "language_code": "hi", + } + + def test_plain_string_is_structured(self) -> None: + assert extract_query_input("hello") == { + "type": "text", + "content": "hello", + "format": "text", + } + + def test_audio_sends_metadata_not_payload(self) -> None: + payload = "QUJD" * 1000 + result = extract_query_input( + AudioInput(content=AudioContent(value=payload, mime_type="audio/wav")) + ) + + assert result == { + "type": "audio", + "format": "base64", + "mime_type": "audio/wav", + "length": len(payload), + } + assert payload not in str(result) + + def test_list_input_maps_each_part(self) -> None: + result = extract_query_input( + [ + TextInput(content=TextContent(value="caption this")), + ImageInput( + content=[ + ImageContent(value="abc", mime_type="image/png"), + ImageContent(format="url", value="https://e.x/a.png"), + ] + ), + ] + ) + + assert result[0]["content"] == "caption this" + assert result[1]["type"] == "image" + assert [item["format"] for item in result[1]["items"]] == ["base64", "url"]