""" OpenTelemetry distributed tracing instrumentation for Hindsight API. This module provides tracing for: - LLM API calls with full prompts/completions following GenAI semantic conventions - Token usage and model information - Error tracking and finish reasons Tracing is conditional and disabled by default. When enabled, traces are exported to Langfuse (or any OTLP-compatible backend) via OTLP HTTP protocol. """ import json import logging import time from typing import Any, Optional from opentelemetry import trace from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.trace import Status, StatusCode logger = logging.getLogger(__name__) def _serialize_for_span(obj: Any) -> str: """Serialize an object for span recording, handling Pydantic models.""" if isinstance(obj, str): return obj if hasattr(obj, "model_dump_json"): # Pydantic v2 model return obj.model_dump_json() if hasattr(obj, "json"): # Pydantic v1 model return obj.json() if hasattr(obj, "model_dump"): # Pydantic v2 model - convert to dict then json return json.dumps(obj.model_dump()) if hasattr(obj, "dict"): # Pydantic v1 model - convert to dict then json return json.dumps(obj.dict()) # Fallback to json.dumps for dicts and other types return json.dumps(obj) # No-op tracer for when tracing is disabled class NoOpTracer: """No-op tracer that provides the same interface as OpenTelemetry Tracer but does nothing.""" def start_as_current_span(self, name: str, **kwargs): """Return a no-op context manager that yields a NoOpSpan.""" from contextlib import contextmanager @contextmanager def noop_span_context(): yield NoOpSpan() return noop_span_context() def start_span(self, name: str, **kwargs): """Return a no-op span.""" return NoOpSpan() class NoOpSpan: """No-op span that provides the same interface as OpenTelemetry Span but does nothing.""" def set_attribute(self, key: str, value: Any) -> None: """No-op.""" pass def set_status(self, status: Any) -> None: """No-op.""" pass def record_exception(self, exception: Exception) -> None: """No-op.""" pass def add_event(self, name: str, attributes: dict | None = None) -> None: """No-op.""" pass def end(self, end_time: int | None = None) -> None: """No-op.""" pass # Global tracer instance _tracer: trace.Tracer | NoOpTracer = NoOpTracer() _tracing_enabled: bool = False # GenAI semantic convention attribute names (based on v1.37 spec) class GenAIAttributes: """GenAI semantic convention attribute names.""" # Operation and provider OPERATION_NAME = "gen_ai.operation.name" PROVIDER_NAME = "gen_ai.provider.name" # Model information REQUEST_MODEL = "gen_ai.request.model" RESPONSE_MODEL = "gen_ai.response.model" # Token usage USAGE_INPUT_TOKENS = "gen_ai.usage.input_tokens" USAGE_OUTPUT_TOKENS = "gen_ai.usage.output_tokens" # Messages and prompts SYSTEM_INSTRUCTIONS = "gen_ai.system_instructions" INPUT_MESSAGES = "gen_ai.input.messages" OUTPUT_MESSAGES = "gen_ai.output.messages" # Response metadata FINISH_REASONS = "gen_ai.response.finish_reasons" # Error tracking ERROR_TYPE = "error.type" # Provider name mapping (Hindsight internal -> GenAI semantic convention) PROVIDER_NAME_MAPPING = { "openai": "openai", "anthropic": "anthropic", "gemini": "google", "vertexai": "google", "groq": "groq", "ollama": "ollama", "lmstudio": "lmstudio", "openai-codex": "openai", "claude-code": "anthropic", "mock": "mock", } def initialize_tracing( service_name: str, endpoint: str, headers: Optional[str] = None, deployment_environment: str = "development", ) -> None: """ Initialize OpenTelemetry tracing with OTLP exporter. Args: service_name: Name of the service for resource attributes endpoint: OTLP endpoint URL (e.g., https://cloud.langfuse.com/api/public/otel) headers: Optional headers in format "key1=value1,key2=value2" deployment_environment: Deployment environment (e.g., development, staging, production) """ global _tracer, _tracing_enabled # Create resource with service information resource = Resource.create( { "service.name": service_name, "service.version": "0.4.8", # Could import from __version__ "deployment.environment.name": deployment_environment, } ) # Parse headers headers_dict = {} if headers: for pair in headers.split(","): if "=" in pair: key, value = pair.split("=", 1) headers_dict[key.strip()] = value.strip() # Create OTLP HTTP exporter # Note: Langfuse expects /v1/traces path appended to base endpoint otlp_endpoint = endpoint if endpoint.endswith("/v1/traces") else f"{endpoint}/v1/traces" otlp_exporter = OTLPSpanExporter( endpoint=otlp_endpoint, headers=headers_dict, ) # Create tracer provider with batch processor provider = TracerProvider(resource=resource) provider.add_span_processor(BatchSpanProcessor(otlp_exporter)) # Set global tracer provider trace.set_tracer_provider(provider) # Get tracer for this application _tracer = trace.get_tracer(__name__) _tracing_enabled = True logger.info(f"Tracing initialized: endpoint={otlp_endpoint}, service={service_name}") def get_tracer() -> trace.Tracer | NoOpTracer: """ Get the global tracer instance. Returns a no-op tracer if tracing is disabled, so callers don't need to check for None. This improves code readability by allowing direct use without null checks. """ return _tracer def create_operation_span(operation: str, bank_id: str | None = None): """ Create a parent span for a Hindsight operation (retain, reflect, consolidation, etc.). This creates the span hierarchy: - hindsight.{operation} (parent) - chat {model} (child LLM calls) Args: operation: Operation name (retain, reflect, consolidation, mental_model_refresh) bank_id: Optional bank ID for context Returns: Span context manager """ if not _tracing_enabled or _tracer is None: # Return a no-op context manager from contextlib import nullcontext return nullcontext() span_name = f"hindsight.{operation}" span = _tracer.start_as_current_span(span_name) # Add operation-specific attributes if span and hasattr(span, "set_attribute"): span.set_attribute("hindsight.operation", operation) if bank_id: span.set_attribute("hindsight.bank_id", bank_id) return span def is_tracing_enabled() -> bool: """Check if tracing is enabled.""" return _tracing_enabled # Maximum content length before truncation (to stay within span size limits) MAX_CONTENT_LENGTH = 100_000 # characters def _truncate_content(content: str) -> str: """Truncate content if too large for span.""" if len(content) > MAX_CONTENT_LENGTH: return content[:MAX_CONTENT_LENGTH] + f"\n\n[TRUNCATED: {len(content) - MAX_CONTENT_LENGTH} chars omitted]" return content class LLMSpanRecorder: """ Records OpenTelemetry spans for LLM calls following GenAI semantic conventions. """ def __init__(self, tracer: trace.Tracer): self.tracer = tracer def record_llm_call( self, provider: str, model: str, scope: str, messages: list[dict[str, str]], response_content: Optional[str], input_tokens: int, output_tokens: int, duration: float, finish_reason: Optional[str] = None, error: Optional[Exception] = None, tool_calls: Optional[list[dict[str, Any]]] = None, ) -> None: """ Record a completed LLM call as a span with GenAI semantic conventions. This creates a span AFTER the call completes, using timestamps to set the correct start/end times. This approach works better with the existing sync metrics recording pattern. Args: provider: Hindsight provider name model: Model name scope: Scope identifier (memory, reflect, consolidation, etc.) messages: Input messages (chat history) response_content: Response text from LLM input_tokens: Input token count output_tokens: Output token count duration: Call duration in seconds finish_reason: Reason the model stopped (stop, length, tool_calls, etc.) error: Exception if call failed tool_calls: List of tool calls made (for function calling) """ try: # Map provider name to GenAI semantic convention genai_provider = PROVIDER_NAME_MAPPING.get(provider.lower(), provider.lower()) # Determine operation name based on scope/context operation_name = "chat" # Default for GenAI semantic conventions # Create span name: "hindsight.{scope}" for consistency with parent spans # Model info is available in span attributes (gen_ai.request.model) if scope: span_name = f"hindsight.{scope}" else: # Fallback to chat {model} if no scope provided span_name = f"{operation_name} {model}" # Calculate timestamps end_time_ns = time.time_ns() start_time_ns = end_time_ns - int(duration * 1_000_000_000) # Create span with explicit timestamps with self.tracer.start_as_current_span( span_name, start_time=start_time_ns, end_on_exit=False, # We'll set end time manually ) as span: # Set required attributes span.set_attribute(GenAIAttributes.OPERATION_NAME, operation_name) span.set_attribute(GenAIAttributes.PROVIDER_NAME, genai_provider) span.set_attribute(GenAIAttributes.REQUEST_MODEL, model) span.set_attribute(GenAIAttributes.RESPONSE_MODEL, model) span.set_attribute(GenAIAttributes.USAGE_INPUT_TOKENS, input_tokens) span.set_attribute(GenAIAttributes.USAGE_OUTPUT_TOKENS, output_tokens) # Add custom attributes for Hindsight context span.set_attribute("hindsight.scope", scope) span.set_attribute("hindsight.provider.internal", provider) # Add tool call information if present if tool_calls: span.set_attribute("gen_ai.tool_calls.count", len(tool_calls)) # Add tool names as comma-separated list tool_names = [tc.get("name", "") for tc in tool_calls] span.set_attribute("gen_ai.tool_calls.names", ",".join(tool_names)) # Format messages for GenAI conventions (as JSON) input_messages_json = self._format_messages(messages) output_messages_json = self._format_output(response_content, finish_reason) # Extract system instructions if present system_instructions = self._extract_system_instructions(messages) # Add event with prompts/completions following v1.37 conventions event_attrs = {} if input_messages_json: event_attrs[GenAIAttributes.INPUT_MESSAGES] = input_messages_json if output_messages_json: event_attrs[GenAIAttributes.OUTPUT_MESSAGES] = output_messages_json if system_instructions: event_attrs[GenAIAttributes.SYSTEM_INSTRUCTIONS] = system_instructions if finish_reason: event_attrs[GenAIAttributes.FINISH_REASONS] = json.dumps([finish_reason]) span.add_event( "gen_ai.client.inference.operation.details", attributes=event_attrs, ) # Add individual tool call events with details if tool_calls: for i, tc in enumerate(tool_calls): tool_event_attrs = { "tool.name": tc.get("name", ""), "tool.id": tc.get("id", ""), "tool.arguments": json.dumps(tc.get("arguments", {})), } span.add_event(f"gen_ai.tool_call.{i}", attributes=tool_event_attrs) # Handle errors if error: span.set_status(Status(StatusCode.ERROR, str(error))) span.set_attribute(GenAIAttributes.ERROR_TYPE, type(error).__name__) span.record_exception(error) else: span.set_status(Status(StatusCode.OK)) # Set end time span.end(end_time=end_time_ns) except Exception as e: # Don't let tracing errors break LLM calls logger.error(f"Failed to record LLM span: {e}", exc_info=True) def _format_messages(self, messages: list[dict[str, str]]) -> str: """ Format messages into GenAI semantic convention format (JSON array). Returns JSON string representation of message array. """ try: formatted = [] for msg in messages: content = msg.get("content", "") # Truncate if needed if isinstance(content, str): content = _truncate_content(content) formatted.append( { "role": msg.get("role", "user"), "content": content, } ) return json.dumps(formatted) except Exception as e: logger.warning(f"Failed to format input messages: {e}") return "[]" def _format_output( self, content: Optional[str], finish_reason: Optional[str], ) -> str: """Format output message into GenAI semantic convention format.""" try: if content is None: return "[]" # Truncate if needed if isinstance(content, str): content = _truncate_content(content) return json.dumps( [ { "role": "assistant", "content": content, } ] ) except Exception as e: logger.warning(f"Failed to format output message: {e}") return "[]" def _extract_system_instructions(self, messages: list[dict[str, str]]) -> Optional[str]: """Extract system instructions from messages if present.""" try: for msg in messages: if msg.get("role") == "system": content = msg.get("content", "") if isinstance(content, str): return _truncate_content(content) return str(content) except Exception as e: logger.warning(f"Failed to extract system instructions: {e}") return None class NoOpLLMSpanRecorder: """No-op span recorder for when tracing is disabled.""" def record_llm_call(self, **kwargs) -> None: """No-op.""" pass # Global span recorder instance _span_recorder: Optional[LLMSpanRecorder] = None def get_span_recorder() -> LLMSpanRecorder | NoOpLLMSpanRecorder: """Get the global span recorder (NoOp if tracing disabled).""" if _span_recorder is None: return NoOpLLMSpanRecorder() return _span_recorder def create_span_recorder() -> LLMSpanRecorder: """Create and set the global span recorder.""" global _span_recorder tracer = get_tracer() if tracer is None: raise RuntimeError("Tracing not initialized. Call initialize_tracing() first.") _span_recorder = LLMSpanRecorder(tracer) return _span_recorder