fix(langfuse): Send structured input for traces

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

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"