diff --git a/.env.example b/.env.example index b6f9b350..4b5d61ff 100644 --- a/.env.example +++ b/.env.example @@ -50,3 +50,18 @@ HINDSIGHT_API_LOG_LEVEL=info # HINDSIGHT_API_RERANKER_LOCAL_MODEL=cross-encoder/ms-marco-MiniLM-L-6-v2 # For TEI provider: # HINDSIGHT_API_RERANKER_TEI_URL=http://localhost:8081 + +# Observability & Tracing (Optional - disabled by default) +# Enable OpenTelemetry tracing for LLM calls (GenAI semantic conventions) +# HINDSIGHT_API_OTEL_TRACES_ENABLED=true +# +# Local development with Grafana LGTM stack (recommended - see scripts/dev/grafana/README.md) +# HINDSIGHT_API_OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318 +# +# Cloud backends (Grafana Cloud, Langfuse, DataDog, etc.) +# HINDSIGHT_API_OTEL_EXPORTER_OTLP_ENDPOINT=https://your-backend-url +# HINDSIGHT_API_OTEL_EXPORTER_OTLP_HEADERS="Authorization=Bearer your-token" +# +# Custom service name and environment (optional, defaults: hindsight-api, development) +# HINDSIGHT_API_OTEL_SERVICE_NAME=hindsight-production +# HINDSIGHT_API_OTEL_DEPLOYMENT_ENVIRONMENT=production diff --git a/CLAUDE.md b/CLAUDE.md index a7a53357..b056ef5b 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -45,6 +45,7 @@ cd hindsight-control-plane && npm run dev ./scripts/dev/start-docs.sh ``` + ### Generating Clients/OpenAPI ```bash # Regenerate OpenAPI spec after API changes (REQUIRED after changing endpoints) diff --git a/hindsight-api/hindsight_api/api/http.py b/hindsight-api/hindsight_api/api/http.py index a591f15a..bfb65769 100644 --- a/hindsight-api/hindsight_api/api/http.py +++ b/hindsight-api/hindsight_api/api/http.py @@ -1400,6 +1400,26 @@ def create_app( app.state.prometheus_reader = None # Metrics collector is already initialized as no-op by default + # Initialize OpenTelemetry tracing if enabled + if config.otel_traces_enabled: + if not config.otel_exporter_otlp_endpoint: + logging.warning("OTEL tracing enabled but no endpoint configured. Tracing disabled.") + else: + from hindsight_api.tracing import create_span_recorder, initialize_tracing + + try: + initialize_tracing( + service_name=config.otel_service_name, + endpoint=config.otel_exporter_otlp_endpoint, + headers=config.otel_exporter_otlp_headers, + deployment_environment=config.otel_deployment_environment, + ) + create_span_recorder() + logging.info("OpenTelemetry tracing enabled and configured") + except Exception as e: + logging.error(f"Failed to initialize tracing: {e}") + logging.warning("Continuing without tracing") + # Startup: Initialize database and memory system (migrations run inside initialize if enabled) if initialize_memory: await memory.initialize() diff --git a/hindsight-api/hindsight_api/config.py b/hindsight-api/hindsight_api/config.py index 6b5b9ef6..92002731 100644 --- a/hindsight-api/hindsight_api/config.py +++ b/hindsight-api/hindsight_api/config.py @@ -108,6 +108,13 @@ ENV_MCP_LOCAL_BANK_ID = "HINDSIGHT_API_MCP_LOCAL_BANK_ID" ENV_MCP_INSTRUCTIONS = "HINDSIGHT_API_MCP_INSTRUCTIONS" ENV_MENTAL_MODEL_REFRESH_CONCURRENCY = "HINDSIGHT_API_MENTAL_MODEL_REFRESH_CONCURRENCY" +# OpenTelemetry tracing configuration +ENV_OTEL_TRACES_ENABLED = "HINDSIGHT_API_OTEL_TRACES_ENABLED" +ENV_OTEL_EXPORTER_OTLP_ENDPOINT = "HINDSIGHT_API_OTEL_EXPORTER_OTLP_ENDPOINT" +ENV_OTEL_EXPORTER_OTLP_HEADERS = "HINDSIGHT_API_OTEL_EXPORTER_OTLP_HEADERS" +ENV_OTEL_SERVICE_NAME = "HINDSIGHT_API_OTEL_SERVICE_NAME" +ENV_OTEL_DEPLOYMENT_ENVIRONMENT = "HINDSIGHT_API_OTEL_DEPLOYMENT_ENVIRONMENT" + # Vertex AI configuration ENV_LLM_VERTEXAI_PROJECT_ID = "HINDSIGHT_API_LLM_VERTEXAI_PROJECT_ID" ENV_LLM_VERTEXAI_REGION = "HINDSIGHT_API_LLM_VERTEXAI_REGION" @@ -251,6 +258,11 @@ DEFAULT_WORKER_CONSOLIDATION_MAX_SLOTS = 2 # Max concurrent consolidation tasks # Reflect agent settings DEFAULT_REFLECT_MAX_ITERATIONS = 10 # Max tool call iterations before forcing response +# OpenTelemetry tracing configuration +DEFAULT_OTEL_TRACES_ENABLED = False # Disabled by default for backward compatibility +DEFAULT_OTEL_SERVICE_NAME = "hindsight-api" +DEFAULT_OTEL_DEPLOYMENT_ENVIRONMENT = "development" + # Default MCP tool descriptions (can be customized via env vars) DEFAULT_MCP_RETAIN_DESCRIPTION = """Store important information to long-term memory. @@ -447,6 +459,13 @@ class HindsightConfig: # Reflect agent settings reflect_max_iterations: int + # OpenTelemetry tracing configuration + otel_traces_enabled: bool + otel_exporter_otlp_endpoint: str | None + otel_exporter_otlp_headers: str | None + otel_service_name: str + otel_deployment_environment: str + def validate(self) -> None: """Validate configuration values and raise errors for invalid combinations.""" # RETAIN_MAX_COMPLETION_TOKENS must be greater than RETAIN_CHUNK_SIZE @@ -646,6 +665,13 @@ class HindsightConfig: ), # Reflect agent settings reflect_max_iterations=int(os.getenv(ENV_REFLECT_MAX_ITERATIONS, str(DEFAULT_REFLECT_MAX_ITERATIONS))), + # OpenTelemetry tracing configuration + otel_traces_enabled=os.getenv(ENV_OTEL_TRACES_ENABLED, str(DEFAULT_OTEL_TRACES_ENABLED)).lower() + in ("true", "1", "yes"), + otel_exporter_otlp_endpoint=os.getenv(ENV_OTEL_EXPORTER_OTLP_ENDPOINT) or None, + otel_exporter_otlp_headers=os.getenv(ENV_OTEL_EXPORTER_OTLP_HEADERS) or None, + otel_service_name=os.getenv(ENV_OTEL_SERVICE_NAME, DEFAULT_OTEL_SERVICE_NAME), + otel_deployment_environment=os.getenv(ENV_OTEL_DEPLOYMENT_ENVIRONMENT, DEFAULT_OTEL_DEPLOYMENT_ENVIRONMENT), ) config.validate() return config diff --git a/hindsight-api/hindsight_api/engine/consolidation/consolidator.py b/hindsight-api/hindsight_api/engine/consolidation/consolidator.py index 471ffb5e..41d2e8dc 100644 --- a/hindsight-api/hindsight_api/engine/consolidation/consolidator.py +++ b/hindsight-api/hindsight_api/engine/consolidation/consolidator.py @@ -426,94 +426,109 @@ async def _process_memory( Returns: Dict with action summary: created/updated/merged counts """ + from ...tracing import get_tracer, is_tracing_enabled + fact_text = memory["text"] memory_id = memory["id"] fact_tags = memory.get("tags") or [] - # Find related observations using the full recall system - # SECURITY: Pass tags to ensure observations don't leak across security boundaries - t0 = time.time() - related_observations = await _find_related_observations( - conn=conn, - memory_engine=memory_engine, - bank_id=bank_id, - query=fact_text, - request_context=request_context, - tags=fact_tags, # Pass source memory's tags for security - ) - if perf: - perf.record_timing("recall", time.time() - t0) + # Create parent span for this memory's consolidation + tracer = get_tracer() + if is_tracing_enabled(): + consolidation_span = tracer.start_span("hindsight.consolidation") + consolidation_span.set_attribute("hindsight.memory_id", str(memory_id)) + consolidation_span.set_attribute("hindsight.bank_id", bank_id) + else: + consolidation_span = None - # Single LLM call handles ALL cases (with or without existing observations) - # Note: Tags are NOT passed to LLM - they are handled algorithmically - t0 = time.time() - actions = await _consolidate_with_llm( - memory_engine=memory_engine, - fact_text=fact_text, - observations=related_observations, # Can be empty list - mission=mission, - ) - if perf: - perf.record_timing("llm", time.time() - t0) + try: + # Find related observations using the full recall system + # SECURITY: Pass tags to ensure observations don't leak across security boundaries + t0 = time.time() + related_observations = await _find_related_observations( + conn=conn, + memory_engine=memory_engine, + bank_id=bank_id, + query=fact_text, + request_context=request_context, + tags=fact_tags, # Pass source memory's tags for security + ) + if perf: + perf.record_timing("recall", time.time() - t0) - if not actions: - # LLM returned empty array - fact is purely ephemeral, skip - return {"action": "skipped", "reason": "no_durable_knowledge"} + # Single LLM call handles ALL cases (with or without existing observations) + # Note: Tags are NOT passed to LLM - they are handled algorithmically + t0 = time.time() + actions = await _consolidate_with_llm( + memory_engine=memory_engine, + fact_text=fact_text, + observations=related_observations, # Can be empty list + mission=mission, + ) + if perf: + perf.record_timing("llm", time.time() - t0) - # Execute all actions and collect results - results = [] - for action in actions: - action_type = action.get("action") - if action_type == "update": - result = await _execute_update_action( - conn=conn, - memory_engine=memory_engine, - bank_id=bank_id, - memory_id=memory_id, - action=action, - observations=related_observations, - source_fact_tags=fact_tags, # Pass source fact's tags for security - source_occurred_start=memory.get("occurred_start"), - source_occurred_end=memory.get("occurred_end"), - source_mentioned_at=memory.get("mentioned_at"), - perf=perf, - ) - results.append(result) - elif action_type == "create": - result = await _execute_create_action( - conn=conn, - memory_engine=memory_engine, - bank_id=bank_id, - memory_id=memory_id, - action=action, - source_fact_tags=fact_tags, # Pass source fact's tags for security - event_date=memory.get("event_date"), - occurred_start=memory.get("occurred_start"), - occurred_end=memory.get("occurred_end"), - mentioned_at=memory.get("mentioned_at"), - perf=perf, - ) - results.append(result) + if not actions: + # LLM returned empty array - fact is purely ephemeral, skip + return {"action": "skipped", "reason": "no_durable_knowledge"} - if not results: - # No valid actions executed - return {"action": "skipped", "reason": "no_valid_actions"} + # Execute all actions and collect results + results = [] + for action in actions: + action_type = action.get("action") + if action_type == "update": + result = await _execute_update_action( + conn=conn, + memory_engine=memory_engine, + bank_id=bank_id, + memory_id=memory_id, + action=action, + observations=related_observations, + source_fact_tags=fact_tags, # Pass source fact's tags for security + source_occurred_start=memory.get("occurred_start"), + source_occurred_end=memory.get("occurred_end"), + source_mentioned_at=memory.get("mentioned_at"), + perf=perf, + ) + results.append(result) + elif action_type == "create": + result = await _execute_create_action( + conn=conn, + memory_engine=memory_engine, + bank_id=bank_id, + memory_id=memory_id, + action=action, + source_fact_tags=fact_tags, # Pass source fact's tags for security + event_date=memory.get("event_date"), + occurred_start=memory.get("occurred_start"), + occurred_end=memory.get("occurred_end"), + mentioned_at=memory.get("mentioned_at"), + perf=perf, + ) + results.append(result) - # Summarize results - created = sum(1 for r in results if r.get("action") == "created") - updated = sum(1 for r in results if r.get("action") == "updated") - merged = sum(1 for r in results if r.get("action") == "merged") + if not results: + # No valid actions executed + return {"action": "skipped", "reason": "no_valid_actions"} - if len(results) == 1: - return results[0] + # Summarize results + created = sum(1 for r in results if r.get("action") == "created") + updated = sum(1 for r in results if r.get("action") == "updated") + merged = sum(1 for r in results if r.get("action") == "merged") - return { - "action": "multiple", - "created": created, - "updated": updated, - "merged": merged, - "total_actions": len(results), - } + if len(results) == 1: + return results[0] + + return { + "action": "multiple", + "created": created, + "updated": updated, + "merged": merged, + "total_actions": len(results), + } + finally: + if consolidation_span: + consolidation_span.end() async def _execute_update_action( @@ -733,22 +748,37 @@ async def _find_related_observations( # Use recall to find related observations with token budget # max_tokens naturally limits how many observations are returned from ...config import get_config + from ...tracing import get_tracer, is_tracing_enabled config = get_config() # SECURITY: Use all_strict matching if tags provided to prevent cross-scope consolidation tags_match = "all_strict" if tags else "any" - recall_result = await memory_engine.recall_async( - bank_id=bank_id, - query=query, - max_tokens=config.consolidation_max_tokens, # Token budget for observations (configurable) - fact_type=["observation"], # Only retrieve observations - request_context=request_context, - tags=tags, # Filter by source memory's tags - tags_match=tags_match, # Use strict matching for security - _quiet=True, # Suppress logging - ) + # Create span for recall operation within consolidation + tracer = get_tracer() + if is_tracing_enabled(): + recall_span = tracer.start_span("hindsight.consolidation_recall") + recall_span.set_attribute("hindsight.bank_id", bank_id) + recall_span.set_attribute("hindsight.query", query[:100]) # Truncate for brevity + recall_span.set_attribute("hindsight.fact_type", "observation") + else: + recall_span = None + + try: + recall_result = await memory_engine.recall_async( + bank_id=bank_id, + query=query, + max_tokens=config.consolidation_max_tokens, # Token budget for observations (configurable) + fact_type=["observation"], # Only retrieve observations + request_context=request_context, + tags=tags, # Filter by source memory's tags + tags_match=tags_match, # Use strict matching for security + _quiet=True, # Suppress logging + ) + finally: + if recall_span: + recall_span.end() # If no observations returned, return empty list if not recall_result.results: diff --git a/hindsight-api/hindsight_api/engine/memory_engine.py b/hindsight-api/hindsight_api/engine/memory_engine.py index 86b7cf8d..a41d0e87 100644 --- a/hindsight-api/hindsight_api/engine/memory_engine.py +++ b/hindsight-api/hindsight_api/engine/memory_engine.py @@ -20,6 +20,7 @@ from typing import TYPE_CHECKING, Any from ..config import get_config from ..metrics import get_metrics_collector +from ..tracing import create_operation_span from ..utils import mask_network_location from .db_budget import budgeted_operation @@ -1540,21 +1541,24 @@ class MemoryEngine(MemoryEngineInterface): from .retain import orchestrator pool = await self._get_pool() - return await orchestrator.retain_batch( - pool=pool, - embeddings_model=self.embeddings, - llm_config=self._retain_llm_config, - entity_resolver=self.entity_resolver, - format_date_fn=self._format_readable_date, - duplicate_checker_fn=self._find_duplicate_facts_batch, - bank_id=bank_id, - contents_dicts=contents, - document_id=document_id, - is_first_batch=is_first_batch, - fact_type_override=fact_type_override, - confidence_score=confidence_score, - document_tags=document_tags, - ) + + # Create parent span for retain operation + with create_operation_span("retain", bank_id): + return await orchestrator.retain_batch( + pool=pool, + embeddings_model=self.embeddings, + llm_config=self._retain_llm_config, + entity_resolver=self.entity_resolver, + format_date_fn=self._format_readable_date, + duplicate_checker_fn=self._find_duplicate_facts_batch, + bank_id=bank_id, + contents_dicts=contents, + document_id=document_id, + is_first_batch=is_first_batch, + fact_type_override=fact_type_override, + confidence_score=confidence_score, + document_tags=document_tags, + ) def recall( self, @@ -1702,136 +1706,152 @@ class MemoryEngine(MemoryEngineInterface): tags_info = f", tags={tags} ({tags_match})" if tags else "" logger.info(f"[RECALL {bank_id[:8]}] Starting recall for query: {query[:50]}...{tags_info}") - # Backpressure: limit concurrent recalls to prevent overwhelming the database - result = None - error_msg = None - semaphore_wait_start = time.time() - async with self._search_semaphore: - semaphore_wait = time.time() - semaphore_wait_start - # Retry loop for connection errors - max_retries = 3 - for attempt in range(max_retries + 1): - try: - result = await self._search_with_retries( - bank_id, - query, - fact_type, - thinking_budget, - max_tokens, - enable_trace, - question_date, - include_entities, - max_entity_tokens, - include_chunks, - max_chunk_tokens, - request_context, - semaphore_wait=semaphore_wait, - tags=tags, - tags_match=tags_match, - connection_budget=_connection_budget, - quiet=_quiet, - ) - break # Success - exit retry loop - except Exception as e: - # Check if it's a connection error - is_connection_error = ( - isinstance(e, asyncpg.TooManyConnectionsError) - or isinstance(e, asyncpg.CannotConnectNowError) - or (isinstance(e, asyncpg.PostgresError) and "connection" in str(e).lower()) - ) + # Create parent span for recall operation + from ..tracing import get_tracer - if is_connection_error and attempt < max_retries: - # Wait with exponential backoff before retry - wait_time = 0.5 * (2**attempt) # 0.5s, 1s, 2s - logger.warning( - f"Connection error on search attempt {attempt + 1}/{max_retries + 1}: {str(e)}. " - f"Retrying in {wait_time:.1f}s..." - ) - await asyncio.sleep(wait_time) - else: - # Not a connection error or out of retries - call post-hook and raise - error_msg = str(e) - if self._operation_validator: - from hindsight_api.extensions.operation_validator import RecallResult + tracer = get_tracer() + # Use start_as_current_span to ensure child spans are linked properly + recall_span_context = tracer.start_as_current_span("hindsight.recall") + recall_span = recall_span_context.__enter__() + recall_span.set_attribute("hindsight.bank_id", bank_id) + recall_span.set_attribute("hindsight.query", query[:100]) + recall_span.set_attribute("hindsight.fact_types", ",".join(fact_type)) + recall_span.set_attribute("hindsight.thinking_budget", thinking_budget) + recall_span.set_attribute("hindsight.max_tokens", max_tokens) - result_ctx = RecallResult( - bank_id=bank_id, - query=query, - request_context=request_context, - budget=budget, - max_tokens=max_tokens, - enable_trace=enable_trace, - fact_types=list(fact_type), - question_date=question_date, - include_entities=include_entities, - max_entity_tokens=max_entity_tokens, - include_chunks=include_chunks, - max_chunk_tokens=max_chunk_tokens, - result=None, - success=False, - error=error_msg, - ) - try: - await self._operation_validator.on_recall_complete(result_ctx) - except Exception as hook_err: - logger.warning(f"Post-recall hook error (non-fatal): {hook_err}") - raise - else: - # Exceeded max retries - error_msg = "Exceeded maximum retries for search due to connection errors." - if self._operation_validator: - from hindsight_api.extensions.operation_validator import RecallResult - - result_ctx = RecallResult( - bank_id=bank_id, - query=query, - request_context=request_context, - budget=budget, - max_tokens=max_tokens, - enable_trace=enable_trace, - fact_types=list(fact_type), - question_date=question_date, - include_entities=include_entities, - max_entity_tokens=max_entity_tokens, - include_chunks=include_chunks, - max_chunk_tokens=max_chunk_tokens, - result=None, - success=False, - error=error_msg, - ) + try: + # Backpressure: limit concurrent recalls to prevent overwhelming the database + result = None + error_msg = None + semaphore_wait_start = time.time() + async with self._search_semaphore: + semaphore_wait = time.time() - semaphore_wait_start + # Retry loop for connection errors + max_retries = 3 + for attempt in range(max_retries + 1): try: - await self._operation_validator.on_recall_complete(result_ctx) - except Exception as hook_err: - logger.warning(f"Post-recall hook error (non-fatal): {hook_err}") - raise Exception(error_msg) + result = await self._search_with_retries( + bank_id, + query, + fact_type, + thinking_budget, + max_tokens, + enable_trace, + question_date, + include_entities, + max_entity_tokens, + include_chunks, + max_chunk_tokens, + request_context, + semaphore_wait=semaphore_wait, + tags=tags, + tags_match=tags_match, + connection_budget=_connection_budget, + quiet=_quiet, + ) + break # Success - exit retry loop + except Exception as e: + # Check if it's a connection error + is_connection_error = ( + isinstance(e, asyncpg.TooManyConnectionsError) + or isinstance(e, asyncpg.CannotConnectNowError) + or (isinstance(e, asyncpg.PostgresError) and "connection" in str(e).lower()) + ) - # Call post-operation hook for success - if self._operation_validator and result is not None: - from hindsight_api.extensions.operation_validator import RecallResult + if is_connection_error and attempt < max_retries: + # Wait with exponential backoff before retry + wait_time = 0.5 * (2**attempt) # 0.5s, 1s, 2s + logger.warning( + f"Connection error on search attempt {attempt + 1}/{max_retries + 1}: {str(e)}. " + f"Retrying in {wait_time:.1f}s..." + ) + await asyncio.sleep(wait_time) + else: + # Not a connection error or out of retries - call post-hook and raise + error_msg = str(e) + if self._operation_validator: + from hindsight_api.extensions.operation_validator import RecallResult - result_ctx = RecallResult( - bank_id=bank_id, - query=query, - request_context=request_context, - budget=budget, - max_tokens=max_tokens, - enable_trace=enable_trace, - fact_types=list(fact_type), - question_date=question_date, - include_entities=include_entities, - max_entity_tokens=max_entity_tokens, - include_chunks=include_chunks, - max_chunk_tokens=max_chunk_tokens, - result=result, - success=True, - error=None, - ) - try: - await self._operation_validator.on_recall_complete(result_ctx) - except Exception as e: - logger.warning(f"Post-recall hook error (non-fatal): {e}") + result_ctx = RecallResult( + bank_id=bank_id, + query=query, + request_context=request_context, + budget=budget, + max_tokens=max_tokens, + enable_trace=enable_trace, + fact_types=list(fact_type), + question_date=question_date, + include_entities=include_entities, + max_entity_tokens=max_entity_tokens, + include_chunks=include_chunks, + max_chunk_tokens=max_chunk_tokens, + result=None, + success=False, + error=error_msg, + ) + try: + await self._operation_validator.on_recall_complete(result_ctx) + except Exception as hook_err: + logger.warning(f"Post-recall hook error (non-fatal): {hook_err}") + raise + else: + # Exceeded max retries + error_msg = "Exceeded maximum retries for search due to connection errors." + if self._operation_validator: + from hindsight_api.extensions.operation_validator import RecallResult - return result + result_ctx = RecallResult( + bank_id=bank_id, + query=query, + request_context=request_context, + budget=budget, + max_tokens=max_tokens, + enable_trace=enable_trace, + fact_types=list(fact_type), + question_date=question_date, + include_entities=include_entities, + max_entity_tokens=max_entity_tokens, + include_chunks=include_chunks, + max_chunk_tokens=max_chunk_tokens, + result=None, + success=False, + error=error_msg, + ) + try: + await self._operation_validator.on_recall_complete(result_ctx) + except Exception as hook_err: + logger.warning(f"Post-recall hook error (non-fatal): {hook_err}") + raise Exception(error_msg) + + # Call post-operation hook for success + if self._operation_validator and result is not None: + from hindsight_api.extensions.operation_validator import RecallResult + + result_ctx = RecallResult( + bank_id=bank_id, + query=query, + request_context=request_context, + budget=budget, + max_tokens=max_tokens, + enable_trace=enable_trace, + fact_types=list(fact_type), + question_date=question_date, + include_entities=include_entities, + max_entity_tokens=max_entity_tokens, + include_chunks=include_chunks, + max_chunk_tokens=max_chunk_tokens, + result=result, + success=True, + error=None, + ) + try: + await self._operation_validator.on_recall_complete(result_ctx) + except Exception as e: + logger.warning(f"Post-recall hook error (non-fatal): {e}") + + return result + finally: + recall_span_context.__exit__(None, None, None) async def _search_with_retries( self, @@ -1898,12 +1918,25 @@ class MemoryEngine(MemoryEngineInterface): f"[RECALL {recall_id}] Query: '{query[:50]}...' (budget={thinking_budget}, max_tokens={max_tokens}{tags_info})" ) + # Import tracing utilities + from ..tracing import get_tracer + + tracer_otel = get_tracer() + try: # Step 1: Generate query embedding (for semantic search) step_start = time.time() - query_embedding = embedding_utils.generate_embedding(self.embeddings, query) - step_duration = time.time() - step_start - log_buffer.append(f" [1] Generate query embedding: {step_duration:.3f}s") + + embedding_span = tracer_otel.start_span("hindsight.recall_embedding") + embedding_span.set_attribute("hindsight.bank_id", bank_id) + embedding_span.set_attribute("hindsight.query", query[:100]) + + try: + query_embedding = embedding_utils.generate_embedding(self.embeddings, query) + step_duration = time.time() - step_start + log_buffer.append(f" [1] Generate query embedding: {step_duration:.3f}s") + finally: + embedding_span.end() if tracer: tracer.record_query_embedding(query_embedding) @@ -1924,30 +1957,38 @@ class MemoryEngine(MemoryEngineInterface): # Track each retrieval start time retrieval_start = time.time() - # Run optimized retrieval with connection budget - config = get_config() - effective_connection_budget = ( - connection_budget if connection_budget is not None else config.recall_connection_budget - ) - async with budgeted_operation( - max_connections=effective_connection_budget, - operation_id=f"recall-{recall_id}", - ) as op: - budgeted_pool = op.wrap_pool(pool) - parallel_start = time.time() - multi_result = await retrieve_all_fact_types_parallel( - budgeted_pool, - query, - query_embedding_str, - bank_id, - fact_type, # Pass all fact types at once - thinking_budget, - question_date, - self.query_analyzer, - tags=tags, - tags_match=tags_match, + retrieval_span = tracer_otel.start_span("hindsight.recall_retrieval") + retrieval_span.set_attribute("hindsight.bank_id", bank_id) + retrieval_span.set_attribute("hindsight.fact_types", ",".join(fact_type)) + retrieval_span.set_attribute("hindsight.thinking_budget", thinking_budget) + + try: + # Run optimized retrieval with connection budget + config = get_config() + effective_connection_budget = ( + connection_budget if connection_budget is not None else config.recall_connection_budget ) - parallel_duration = time.time() - parallel_start + async with budgeted_operation( + max_connections=effective_connection_budget, + operation_id=f"recall-{recall_id}", + ) as op: + budgeted_pool = op.wrap_pool(pool) + parallel_start = time.time() + multi_result = await retrieve_all_fact_types_parallel( + budgeted_pool, + query, + query_embedding_str, + bank_id, + fact_type, # Pass all fact types at once + thinking_budget, + question_date, + self.query_analyzer, + tags=tags, + tags_match=tags_match, + ) + parallel_duration = time.time() - parallel_start + finally: + retrieval_span.end() # Combine all results from all fact types and aggregate timings semantic_results = [] @@ -2134,16 +2175,29 @@ class MemoryEngine(MemoryEngineInterface): step_start = time.time() from .search.fusion import reciprocal_rank_fusion - # Merge 3 or 4 result lists depending on temporal constraint - if temporal_results: - merged_candidates = reciprocal_rank_fusion( - [semantic_results, bm25_results, graph_results, temporal_results] - ) - else: - merged_candidates = reciprocal_rank_fusion([semantic_results, bm25_results, graph_results]) + fusion_span = tracer_otel.start_span("hindsight.recall_fusion") + fusion_span.set_attribute("hindsight.bank_id", bank_id) + fusion_span.set_attribute("hindsight.semantic_count", len(semantic_results)) + fusion_span.set_attribute("hindsight.bm25_count", len(bm25_results)) + fusion_span.set_attribute("hindsight.graph_count", len(graph_results)) + fusion_span.set_attribute("hindsight.temporal_count", len(temporal_results) if temporal_results else 0) - step_duration = time.time() - step_start - log_buffer.append(f" [3] RRF merge: {len(merged_candidates)} unique candidates in {step_duration:.3f}s") + try: + # Merge 3 or 4 result lists depending on temporal constraint + if temporal_results: + merged_candidates = reciprocal_rank_fusion( + [semantic_results, bm25_results, graph_results, temporal_results] + ) + else: + merged_candidates = reciprocal_rank_fusion([semantic_results, bm25_results, graph_results]) + + step_duration = time.time() - step_start + log_buffer.append( + f" [3] RRF merge: {len(merged_candidates)} unique candidates in {step_duration:.3f}s" + ) + finally: + fusion_span.set_attribute("hindsight.merged_count", len(merged_candidates)) + fusion_span.end() if tracer: # Convert MergedCandidate to old tuple format for tracer @@ -2158,27 +2212,37 @@ class MemoryEngine(MemoryEngineInterface): step_start = time.time() reranker_instance = self._cross_encoder_reranker - # Ensure reranker is initialized (for lazy initialization mode) - await reranker_instance.ensure_initialized() + rerank_span = tracer_otel.start_span("hindsight.recall_rerank") + rerank_span.set_attribute("hindsight.bank_id", bank_id) + rerank_span.set_attribute("hindsight.candidates_count", len(merged_candidates)) - # Pre-filter candidates to reduce reranking cost (RRF already provides good ranking) - # This is especially important for remote rerankers with network latency - reranker_max_candidates = get_config().reranker_max_candidates - pre_filtered_count = 0 - if len(merged_candidates) > reranker_max_candidates: - # Sort by RRF score and take top candidates - merged_candidates.sort(key=lambda mc: mc.rrf_score, reverse=True) - pre_filtered_count = len(merged_candidates) - reranker_max_candidates - merged_candidates = merged_candidates[:reranker_max_candidates] + try: + # Ensure reranker is initialized (for lazy initialization mode) + await reranker_instance.ensure_initialized() - # Rerank using cross-encoder - scored_results = await reranker_instance.rerank(query, merged_candidates) + # Pre-filter candidates to reduce reranking cost (RRF already provides good ranking) + # This is especially important for remote rerankers with network latency + reranker_max_candidates = get_config().reranker_max_candidates + pre_filtered_count = 0 + if len(merged_candidates) > reranker_max_candidates: + # Sort by RRF score and take top candidates + merged_candidates.sort(key=lambda mc: mc.rrf_score, reverse=True) + pre_filtered_count = len(merged_candidates) - reranker_max_candidates + merged_candidates = merged_candidates[:reranker_max_candidates] - step_duration = time.time() - step_start - pre_filter_note = f" (pre-filtered {pre_filtered_count})" if pre_filtered_count > 0 else "" - log_buffer.append( - f" [4] Reranking: {len(scored_results)} candidates scored in {step_duration:.3f}s{pre_filter_note}" - ) + # Rerank using cross-encoder + scored_results = await reranker_instance.rerank(query, merged_candidates) + + step_duration = time.time() - step_start + pre_filter_note = f" (pre-filtered {pre_filtered_count})" if pre_filtered_count > 0 else "" + log_buffer.append( + f" [4] Reranking: {len(scored_results)} candidates scored in {step_duration:.3f}s{pre_filter_note}" + ) + finally: + rerank_span.set_attribute("hindsight.scored_count", len(scored_results)) + if pre_filtered_count > 0: + rerank_span.set_attribute("hindsight.pre_filtered_count", pre_filtered_count) + rerank_span.end() # Step 4.5: Combine cross-encoder score with retrieval signals # This preserves retrieval work (RRF, temporal, recency) instead of pure cross-encoder ranking @@ -2750,18 +2814,20 @@ class MemoryEngine(MemoryEngineInterface): from .consolidation import run_consolidation_job - result = await run_consolidation_job( - memory_engine=self, - bank_id=bank_id, - request_context=request_context, - ) + # Create parent span for consolidation operation + with create_operation_span("consolidation", bank_id): + result = await run_consolidation_job( + memory_engine=self, + bank_id=bank_id, + request_context=request_context, + ) - return { - "processed": result.get("processed", 0), - "created": result.get("created", 0), - "updated": result.get("updated", 0), - "skipped": result.get("skipped", 0), - } + return { + "processed": result.get("processed", 0), + "created": result.get("created", 0), + "updated": result.get("updated", 0), + "skipped": result.get("skipped", 0), + } async def get_graph_data( self, @@ -3570,6 +3636,7 @@ class MemoryEngine(MemoryEngineInterface): tags: list[str] | None = None, tags_match: TagsMatch = "any", exclude_mental_model_ids: list[str] | None = None, + _skip_span: bool = False, ) -> ReflectResult: """ Reflect and formulate an answer using an agentic loop with tools. @@ -3726,219 +3793,233 @@ class MemoryEngine(MemoryEngineInterface): if has_mental_models: logger.info(f"[REFLECT {reflect_id}] Bank has {mental_model_count} mental models") - # Run the agent - agent_result = await run_reflect_agent( - llm_config=self._reflect_llm_config, - bank_id=bank_id, - query=query, - bank_profile=profile, - search_mental_models_fn=search_mental_models_fn, - search_observations_fn=search_observations_fn, - recall_fn=recall_fn, - expand_fn=expand_fn, - context=context, - max_iterations=max_iterations, - max_tokens=max_tokens, - response_schema=response_schema, - directives=directives, - has_mental_models=has_mental_models, - budget=effective_budget, - ) + # Run the agent with parent span for reflect operation (skip if called from another operation) + if not _skip_span: + span_context = create_operation_span("reflect", bank_id) + span_context.__enter__() + else: + span_context = None - total_time = time.time() - reflect_start - logger.info( - f"[REFLECT {reflect_id}] Complete: {len(agent_result.text)} chars, " - f"{agent_result.iterations} iterations, {agent_result.tools_called} tool calls | {total_time:.3f}s" - ) - - # Convert agent tool trace to ToolCallTrace objects - tool_trace_result = [ - ToolCallTrace( - tool=tc.tool, - reason=tc.reason, - input=tc.input, - output=tc.output, - duration_ms=tc.duration_ms, - iteration=tc.iteration, - ) - for tc in agent_result.tool_trace - ] - - # Convert agent LLM trace to LLMCallTrace objects - llm_trace_result = [LLMCallTrace(scope=lc.scope, duration_ms=lc.duration_ms) for lc in agent_result.llm_trace] - - # Extract memories from recall tool outputs - only include memories the agent actually used - # agent_result.used_memory_ids contains validated IDs from the done action - used_memory_ids_set = set(agent_result.used_memory_ids) if agent_result.used_memory_ids else set() - # based_on stores facts, mental models, and directives - # Note: directives list stores raw directive dicts (not MemoryFact), which will be converted to Directive objects - based_on: dict[str, list[MemoryFact] | list[dict[str, Any]]] = { - "world": [], - "experience": [], - "opinion": [], - "observation": [], - "mental-models": [], - "directives": [], - } - seen_memory_ids: set[str] = set() - for tc in agent_result.tool_trace: - if tc.tool == "recall" and "memories" in tc.output: - for memory_data in tc.output["memories"]: - memory_id = memory_data.get("id") - # Only include memories that the agent declared as used (or all if none specified) - if memory_id and memory_id not in seen_memory_ids: - if used_memory_ids_set and memory_id not in used_memory_ids_set: - continue # Skip memories not actually used by the agent - seen_memory_ids.add(memory_id) - fact_type = memory_data.get("type", "world") - if fact_type in based_on: - based_on[fact_type].append( - MemoryFact( - id=memory_id, - text=memory_data.get("text", ""), - fact_type=fact_type, - context=None, - occurred_start=memory_data.get("occurred"), - occurred_end=memory_data.get("occurred"), - ) - ) - - # Extract mental models from tool outputs - only include models the agent actually used - # agent_result.used_mental_model_ids contains validated IDs from the done action - used_model_ids_set = set(agent_result.used_mental_model_ids) if agent_result.used_mental_model_ids else set() - based_on["mental-models"] = [] - seen_model_ids: set[str] = set() - for tc in agent_result.tool_trace: - if tc.tool == "get_mental_model": - # Single model lookup (with full details) - if tc.output.get("found") and "model" in tc.output: - model = tc.output["model"] - model_id = model.get("id") - if model_id and model_id not in seen_model_ids: - # Only include models that the agent declared as used (or all if none specified) - if used_model_ids_set and model_id not in used_model_ids_set: - continue # Skip models not actually used by the agent - seen_model_ids.add(model_id) - # Add to based_on as MemoryFact with type "mental-models" - model_name = model.get("name", "") - model_summary = model.get("summary") or model.get("description", "") - based_on["mental-models"].append( - MemoryFact( - id=model_id, - text=f"{model_name}: {model_summary}", - fact_type="mental-models", - context=f"{model.get('type', 'concept')} ({model.get('subtype', 'structural')})", - occurred_start=None, - occurred_end=None, - ) - ) - elif tc.tool == "search_mental_models": - # Search mental models - include all returned models (filtered by used_model_ids_set if specified) - for model in tc.output.get("mental_models", []): - model_id = model.get("id") - if model_id and model_id not in seen_model_ids: - # Only include models that the agent declared as used (or all if none specified) - if used_model_ids_set and model_id not in used_model_ids_set: - continue # Skip models not actually used by the agent - seen_model_ids.add(model_id) - # Add to based_on as MemoryFact with type "mental-models" - model_name = model.get("name", "") - model_summary = model.get("summary") or model.get("description", "") - based_on["mental-models"].append( - MemoryFact( - id=model_id, - text=f"{model_name}: {model_summary}", - fact_type="mental-models", - context=f"{model.get('type', 'concept')} ({model.get('subtype', 'structural')})", - occurred_start=None, - occurred_end=None, - ) - ) - elif tc.tool == "search_mental_models": - # Search mental models - include all returned mental models (filtered by used_mental_model_ids_set if specified) - used_mental_model_ids_set = ( - set(agent_result.used_mental_model_ids) if agent_result.used_mental_model_ids else set() - ) - for mental_model in tc.output.get("mental_models", []): - mental_model_id = mental_model.get("id") - if mental_model_id and mental_model_id not in seen_model_ids: - # Only include mental models that the agent declared as used (or all if none specified) - if used_mental_model_ids_set and mental_model_id not in used_mental_model_ids_set: - continue # Skip mental models not actually used by the agent - seen_model_ids.add(mental_model_id) - # Add to based_on as MemoryFact with type "mental-models" (mental models are synthesized knowledge) - mental_model_name = mental_model.get("name", "") - mental_model_content = mental_model.get("content", "") - based_on["mental-models"].append( - MemoryFact( - id=mental_model_id, - text=f"{mental_model_name}: {mental_model_content}", - fact_type="mental-models", - context="mental model (user-curated)", - occurred_start=None, - occurred_end=None, - ) - ) - # List all models lookup - don't add to based_on (too verbose, just a listing) - - # Add directives to based_on["directives"] - # Store raw directive dicts (with id, name, content) for http.py to convert to ReflectDirective - for directive_raw in directives_raw: - based_on["directives"].append( - { - "id": directive_raw["id"], - "name": directive_raw["name"], - "content": directive_raw["content"], - } - ) - - # Build directives_applied from agent result - from hindsight_api.engine.response_models import DirectiveRef - - directives_applied_result = [ - DirectiveRef(id=d.id, name=d.name, content=d.content) for d in agent_result.directives_applied - ] - - # Convert agent usage to TokenUsage format - from hindsight_api.engine.response_models import TokenUsage - - usage = TokenUsage( - input_tokens=agent_result.usage.input_tokens, - output_tokens=agent_result.usage.output_tokens, - total_tokens=agent_result.usage.total_tokens, - ) - - # Return response (compatible with existing API) - result = ReflectResult( - text=agent_result.text, - based_on=based_on, - structured_output=agent_result.structured_output, - usage=usage, - tool_trace=tool_trace_result, - llm_trace=llm_trace_result, - directives_applied=directives_applied_result, - ) - - # Call post-operation hook if validator is configured - if self._operation_validator: - from hindsight_api.extensions.operation_validator import ReflectResultContext - - result_ctx = ReflectResultContext( + try: + agent_result = await run_reflect_agent( + llm_config=self._reflect_llm_config, bank_id=bank_id, query=query, - request_context=request_context, - budget=budget, + bank_profile=profile, + search_mental_models_fn=search_mental_models_fn, + search_observations_fn=search_observations_fn, + recall_fn=recall_fn, + expand_fn=expand_fn, context=context, - result=result, - success=True, - error=None, + max_iterations=max_iterations, + max_tokens=max_tokens, + response_schema=response_schema, + directives=directives, + has_mental_models=has_mental_models, + budget=effective_budget, ) - try: - await self._operation_validator.on_reflect_complete(result_ctx) - except Exception as e: - logger.warning(f"Post-reflect hook error (non-fatal): {e}") - return result + total_time = time.time() - reflect_start + logger.info( + f"[REFLECT {reflect_id}] Complete: {len(agent_result.text)} chars, " + f"{agent_result.iterations} iterations, {agent_result.tools_called} tool calls | {total_time:.3f}s" + ) + + # Convert agent tool trace to ToolCallTrace objects + tool_trace_result = [ + ToolCallTrace( + tool=tc.tool, + reason=tc.reason, + input=tc.input, + output=tc.output, + duration_ms=tc.duration_ms, + iteration=tc.iteration, + ) + for tc in agent_result.tool_trace + ] + + # Convert agent LLM trace to LLMCallTrace objects + llm_trace_result = [ + LLMCallTrace(scope=lc.scope, duration_ms=lc.duration_ms) for lc in agent_result.llm_trace + ] + + # Extract memories from recall tool outputs - only include memories the agent actually used + # agent_result.used_memory_ids contains validated IDs from the done action + used_memory_ids_set = set(agent_result.used_memory_ids) if agent_result.used_memory_ids else set() + # based_on stores facts, mental models, and directives + # Note: directives list stores raw directive dicts (not MemoryFact), which will be converted to Directive objects + based_on: dict[str, list[MemoryFact] | list[dict[str, Any]]] = { + "world": [], + "experience": [], + "opinion": [], + "observation": [], + "mental-models": [], + "directives": [], + } + seen_memory_ids: set[str] = set() + for tc in agent_result.tool_trace: + if tc.tool == "recall" and "memories" in tc.output: + for memory_data in tc.output["memories"]: + memory_id = memory_data.get("id") + # Only include memories that the agent declared as used (or all if none specified) + if memory_id and memory_id not in seen_memory_ids: + if used_memory_ids_set and memory_id not in used_memory_ids_set: + continue # Skip memories not actually used by the agent + seen_memory_ids.add(memory_id) + fact_type = memory_data.get("type", "world") + if fact_type in based_on: + based_on[fact_type].append( + MemoryFact( + id=memory_id, + text=memory_data.get("text", ""), + fact_type=fact_type, + context=None, + occurred_start=memory_data.get("occurred"), + occurred_end=memory_data.get("occurred"), + ) + ) + + # Extract mental models from tool outputs - only include models the agent actually used + # agent_result.used_mental_model_ids contains validated IDs from the done action + used_model_ids_set = ( + set(agent_result.used_mental_model_ids) if agent_result.used_mental_model_ids else set() + ) + based_on["mental-models"] = [] + seen_model_ids: set[str] = set() + for tc in agent_result.tool_trace: + if tc.tool == "get_mental_model": + # Single model lookup (with full details) + if tc.output.get("found") and "model" in tc.output: + model = tc.output["model"] + model_id = model.get("id") + if model_id and model_id not in seen_model_ids: + # Only include models that the agent declared as used (or all if none specified) + if used_model_ids_set and model_id not in used_model_ids_set: + continue # Skip models not actually used by the agent + seen_model_ids.add(model_id) + # Add to based_on as MemoryFact with type "mental-models" + model_name = model.get("name", "") + model_summary = model.get("summary") or model.get("description", "") + based_on["mental-models"].append( + MemoryFact( + id=model_id, + text=f"{model_name}: {model_summary}", + fact_type="mental-models", + context=f"{model.get('type', 'concept')} ({model.get('subtype', 'structural')})", + occurred_start=None, + occurred_end=None, + ) + ) + elif tc.tool == "search_mental_models": + # Search mental models - include all returned models (filtered by used_model_ids_set if specified) + for model in tc.output.get("mental_models", []): + model_id = model.get("id") + if model_id and model_id not in seen_model_ids: + # Only include models that the agent declared as used (or all if none specified) + if used_model_ids_set and model_id not in used_model_ids_set: + continue # Skip models not actually used by the agent + seen_model_ids.add(model_id) + # Add to based_on as MemoryFact with type "mental-models" + model_name = model.get("name", "") + model_summary = model.get("summary") or model.get("description", "") + based_on["mental-models"].append( + MemoryFact( + id=model_id, + text=f"{model_name}: {model_summary}", + fact_type="mental-models", + context=f"{model.get('type', 'concept')} ({model.get('subtype', 'structural')})", + occurred_start=None, + occurred_end=None, + ) + ) + elif tc.tool == "search_mental_models": + # Search mental models - include all returned mental models (filtered by used_mental_model_ids_set if specified) + used_mental_model_ids_set = ( + set(agent_result.used_mental_model_ids) if agent_result.used_mental_model_ids else set() + ) + for mental_model in tc.output.get("mental_models", []): + mental_model_id = mental_model.get("id") + if mental_model_id and mental_model_id not in seen_model_ids: + # Only include mental models that the agent declared as used (or all if none specified) + if used_mental_model_ids_set and mental_model_id not in used_mental_model_ids_set: + continue # Skip mental models not actually used by the agent + seen_model_ids.add(mental_model_id) + # Add to based_on as MemoryFact with type "mental-models" (mental models are synthesized knowledge) + mental_model_name = mental_model.get("name", "") + mental_model_content = mental_model.get("content", "") + based_on["mental-models"].append( + MemoryFact( + id=mental_model_id, + text=f"{mental_model_name}: {mental_model_content}", + fact_type="mental-models", + context="mental model (user-curated)", + occurred_start=None, + occurred_end=None, + ) + ) + # List all models lookup - don't add to based_on (too verbose, just a listing) + + # Add directives to based_on["directives"] + # Store raw directive dicts (with id, name, content) for http.py to convert to ReflectDirective + for directive_raw in directives_raw: + based_on["directives"].append( + { + "id": directive_raw["id"], + "name": directive_raw["name"], + "content": directive_raw["content"], + } + ) + + # Build directives_applied from agent result + from hindsight_api.engine.response_models import DirectiveRef + + directives_applied_result = [ + DirectiveRef(id=d.id, name=d.name, content=d.content) for d in agent_result.directives_applied + ] + + # Convert agent usage to TokenUsage format + from hindsight_api.engine.response_models import TokenUsage + + usage = TokenUsage( + input_tokens=agent_result.usage.input_tokens, + output_tokens=agent_result.usage.output_tokens, + total_tokens=agent_result.usage.total_tokens, + ) + + # Return response (compatible with existing API) + result = ReflectResult( + text=agent_result.text, + based_on=based_on, + structured_output=agent_result.structured_output, + usage=usage, + tool_trace=tool_trace_result, + llm_trace=llm_trace_result, + directives_applied=directives_applied_result, + ) + + # Call post-operation hook if validator is configured + if self._operation_validator: + from hindsight_api.extensions.operation_validator import ReflectResultContext + + result_ctx = ReflectResultContext( + bank_id=bank_id, + query=query, + request_context=request_context, + budget=budget, + context=context, + result=result, + success=True, + error=None, + ) + try: + await self._operation_validator.on_reflect_complete(result_ctx) + except Exception as e: + logger.warning(f"Post-reflect hook error (non-fatal): {e}") + + return result + finally: + if span_context: + span_context.__exit__(None, None, None) async def list_entities( self, @@ -4738,64 +4819,68 @@ class MemoryEngine(MemoryEngineInterface): if not mental_model: return None - # SECURITY: If the mental model has tags, pass them to reflect with "all_strict" matching - # to ensure it can only access other mental models/memories with the SAME tags. - # This prevents cross-tenant/cross-user information leakage by excluding untagged content. - tags = mental_model.get("tags") - tags_match = "all_strict" if tags else "any" + # Create parent span for mental model refresh operation + with create_operation_span("mental_model_refresh", bank_id): + # SECURITY: If the mental model has tags, pass them to reflect with "all_strict" matching + # to ensure it can only access other mental models/memories with the SAME tags. + # This prevents cross-tenant/cross-user information leakage by excluding untagged content. + tags = mental_model.get("tags") + tags_match = "all_strict" if tags else "any" - # Run reflect with the source query, excluding the mental model being refreshed - reflect_result = await self.reflect_async( - bank_id=bank_id, - query=mental_model["source_query"], - request_context=request_context, - tags=tags, - tags_match=tags_match, - exclude_mental_model_ids=[mental_model_id], - ) + # Run reflect with the source query, excluding the mental model being refreshed + # Skip creating a nested "hindsight.reflect" span since we already have "hindsight.mental_model_refresh" + reflect_result = await self.reflect_async( + bank_id=bank_id, + query=mental_model["source_query"], + request_context=request_context, + tags=tags, + tags_match=tags_match, + exclude_mental_model_ids=[mental_model_id], + _skip_span=True, + ) - # Build reflect_response payload to store - # based_on contains MemoryFact objects for most types, but plain dicts for directives - based_on_serialized_payload: dict[str, list[dict[str, Any]]] = {} - for fact_type, facts in reflect_result.based_on.items(): - serialized_facts = [] - for fact in facts: - if isinstance(fact, dict): - # Plain dict (e.g., directives with id, name, content) - serialized_facts.append( - { - "id": str(fact["id"]), - "text": fact.get("text", fact.get("content", fact.get("name", ""))), - "type": fact_type, - "context": fact.get("context", None), - } - ) - else: - # MemoryFact object with .id, .text, .context attributes - serialized_facts.append( - { - "id": str(fact.id), - "text": fact.text, - "type": fact_type, - "context": fact.context, - } - ) - based_on_serialized_payload[fact_type] = serialized_facts + # Build reflect_response payload to store + # based_on contains MemoryFact objects for most types, but plain dicts for directives + based_on_serialized_payload: dict[str, list[dict[str, Any]]] = {} + for fact_type, facts in reflect_result.based_on.items(): + serialized_facts = [] + for fact in facts: + if isinstance(fact, dict): + # Plain dict (e.g., directives with id, name, content) + serialized_facts.append( + { + "id": str(fact["id"]), + "text": fact.get("text", fact.get("content", fact.get("name", ""))), + "type": fact_type, + "context": fact.get("context", None), + } + ) + else: + # MemoryFact object with .id, .text, .context attributes + serialized_facts.append( + { + "id": str(fact.id), + "text": fact.text, + "type": fact_type, + "context": fact.context, + } + ) + based_on_serialized_payload[fact_type] = serialized_facts - reflect_response_payload = { - "text": reflect_result.text, - "based_on": based_on_serialized_payload, - "mental_models": [], # Mental models are included in based_on["mental-models"] - } + reflect_response_payload = { + "text": reflect_result.text, + "based_on": based_on_serialized_payload, + "mental_models": [], # Mental models are included in based_on["mental-models"] + } - # Update the mental model with new content and reflect_response - return await self.update_mental_model( - bank_id, - mental_model_id, - content=reflect_result.text, - reflect_response=reflect_response_payload, - request_context=request_context, - ) + # Update the mental model with new content and reflect_response + return await self.update_mental_model( + bank_id, + mental_model_id, + content=reflect_result.text, + reflect_response=reflect_response_payload, + request_context=request_context, + ) async def update_mental_model( self, diff --git a/hindsight-api/hindsight_api/engine/providers/anthropic_llm.py b/hindsight-api/hindsight_api/engine/providers/anthropic_llm.py index 73258f78..0956f993 100644 --- a/hindsight-api/hindsight_api/engine/providers/anthropic_llm.py +++ b/hindsight-api/hindsight_api/engine/providers/anthropic_llm.py @@ -84,7 +84,7 @@ class AnthropicLLM(LLMInterface): messages=test_messages, max_completion_tokens=10, temperature=0.0, - scope="test", + scope="verification", max_retries=0, ) logger.info("Anthropic connection verified successfully") @@ -223,6 +223,24 @@ class AnthropicLLM(LLMInterface): success=True, ) + # Record trace span + from hindsight_api.tracing import _serialize_for_span, get_span_recorder + + finish_reason = response.stop_reason if hasattr(response, "stop_reason") else None + span_recorder = get_span_recorder() + span_recorder.record_llm_call( + provider=self.provider, + model=self.model, + scope=scope, + messages=messages, + response_content=_serialize_for_span(result), + input_tokens=input_tokens, + output_tokens=output_tokens, + duration=duration, + finish_reason=finish_reason, + error=None, + ) + # Log slow calls if duration > 10.0: logger.info( @@ -397,16 +415,41 @@ class AnthropicLLM(LLMInterface): # Record metrics metrics = get_metrics_collector() + duration = time.time() - start_time metrics.record_llm_call( provider=self.provider, model=self.model, scope=scope, - duration=time.time() - start_time, + duration=duration, input_tokens=input_tokens, output_tokens=output_tokens, success=True, ) + # Record OpenTelemetry span + from hindsight_api.tracing import get_span_recorder + + span_recorder = get_span_recorder() + # Convert LLMToolCall objects to dicts for span recording + tool_calls_dict = ( + [{"id": tc.id, "name": tc.name, "arguments": tc.arguments} for tc in tool_calls] + if tool_calls + else None + ) + span_recorder.record_llm_call( + provider=self.provider, + model=self.model, + scope=scope, + messages=messages, + response_content=content, + input_tokens=input_tokens, + output_tokens=output_tokens, + duration=duration, + finish_reason=finish_reason, + error=None, + tool_calls=tool_calls_dict, + ) + return LLMToolCallResult( content=content, tool_calls=tool_calls, diff --git a/hindsight-api/hindsight_api/engine/providers/claude_code_llm.py b/hindsight-api/hindsight_api/engine/providers/claude_code_llm.py index d2124ce4..86b1a08f 100644 --- a/hindsight-api/hindsight_api/engine/providers/claude_code_llm.py +++ b/hindsight-api/hindsight_api/engine/providers/claude_code_llm.py @@ -95,7 +95,7 @@ class ClaudeCodeLLM(LLMInterface): messages=test_messages, max_completion_tokens=10, temperature=0.0, - scope="test", + scope="verification", max_retries=0, ) logger.info("Claude Code connection verified successfully") @@ -237,6 +237,23 @@ class ClaudeCodeLLM(LLMInterface): success=True, ) + # Record trace span + from hindsight_api.tracing import get_span_recorder + + span_recorder = get_span_recorder() + span_recorder.record_llm_call( + provider=self.provider, + model=self.model, + scope=scope, + messages=messages, + response_content=result if isinstance(result, str) else json.dumps(result), + input_tokens=estimated_input, + output_tokens=estimated_output, + duration=duration, + finish_reason=None, + error=None, + ) + # Log slow calls if duration > 10.0: logger.info( diff --git a/hindsight-api/hindsight_api/engine/providers/codex_llm.py b/hindsight-api/hindsight_api/engine/providers/codex_llm.py index d3bf2925..04e25710 100644 --- a/hindsight-api/hindsight_api/engine/providers/codex_llm.py +++ b/hindsight-api/hindsight_api/engine/providers/codex_llm.py @@ -136,6 +136,7 @@ class CodexLLM(LLMInterface): max_retries=2, initial_backoff=0.5, max_backoff=2.0, + scope="verification", ) logger.info(f"Codex LLM verified: {self.model}") except Exception as e: @@ -261,6 +262,26 @@ class CodexLLM(LLMInterface): success=True, ) + # Record trace span + from hindsight_api.tracing import get_span_recorder + + # Estimate tokens for tracing + estimated_input = sum(len(m.get("content", "")) for m in messages) // 4 + estimated_output = len(content) // 4 + span_recorder = get_span_recorder() + span_recorder.record_llm_call( + provider=self.provider, + model=self.model, + scope=scope, + messages=messages, + response_content=result if isinstance(result, str) else json.dumps(result), + input_tokens=estimated_input, + output_tokens=estimated_output, + duration=duration, + finish_reason=None, + error=None, + ) + if return_usage: # Codex doesn't provide token counts, estimate based on content estimated_input = sum(len(m.get("content", "")) for m in messages) // 4 @@ -504,6 +525,28 @@ class CodexLLM(LLMInterface): success=True, ) + # Record OpenTelemetry span + from hindsight_api.tracing import get_span_recorder + + span_recorder = get_span_recorder() + # Convert LLMToolCall objects to dicts for span recording + tool_calls_dict = ( + [{"id": tc.id, "name": tc.name, "arguments": tc.arguments} for tc in tool_calls] if tool_calls else None + ) + span_recorder.record_llm_call( + provider=self.provider, + model=self.model, + scope=scope, + messages=messages, + response_content=content, + input_tokens=0, # Codex doesn't provide token counts + output_tokens=0, + duration=duration, + finish_reason="tool_calls" if tool_calls else "stop", + error=None, + tool_calls=tool_calls_dict, + ) + return LLMToolCallResult( content=content, tool_calls=tool_calls, diff --git a/hindsight-api/hindsight_api/engine/providers/gemini_llm.py b/hindsight-api/hindsight_api/engine/providers/gemini_llm.py index 840efdba..67d71449 100644 --- a/hindsight-api/hindsight_api/engine/providers/gemini_llm.py +++ b/hindsight-api/hindsight_api/engine/providers/gemini_llm.py @@ -136,6 +136,7 @@ class GeminiLLM(LLMInterface): max_retries=2, initial_backoff=0.5, max_backoff=2.0, + scope="verification", ) logger.info(f"{self.provider.upper()} connection verified successfully") except Exception as e: @@ -275,6 +276,29 @@ class GeminiLLM(LLMInterface): success=True, ) + # Record trace span + from hindsight_api.tracing import get_span_recorder + + finish_reason = None + if hasattr(response, "candidates") and response.candidates: + if hasattr(response.candidates[0], "finish_reason"): + finish_reason = str(response.candidates[0].finish_reason) + span_recorder = get_span_recorder() + from hindsight_api.tracing import _serialize_for_span + + span_recorder.record_llm_call( + provider=self.provider, + model=self.model, + scope=scope, + messages=messages, + response_content=_serialize_for_span(result), + input_tokens=input_tokens, + output_tokens=output_tokens, + duration=duration, + finish_reason=finish_reason, + error=None, + ) + # Log slow calls if duration > 10.0 and input_tokens > 0: logger.info( @@ -466,6 +490,30 @@ class GeminiLLM(LLMInterface): success=True, ) + # Record OpenTelemetry span + from hindsight_api.tracing import get_span_recorder + + span_recorder = get_span_recorder() + # Convert LLMToolCall objects to dicts for span recording + tool_calls_dict = ( + [{"id": tc.id, "name": tc.name, "arguments": tc.arguments} for tc in tool_calls] + if tool_calls + else None + ) + span_recorder.record_llm_call( + provider=self.provider, + model=self.model, + scope=scope, + messages=messages, + response_content=content, + input_tokens=input_tokens, + output_tokens=output_tokens, + duration=duration, + finish_reason=finish_reason, + error=None, + tool_calls=tool_calls_dict, + ) + return LLMToolCallResult( content=content, tool_calls=tool_calls, diff --git a/hindsight-api/hindsight_api/engine/providers/mock_llm.py b/hindsight-api/hindsight_api/engine/providers/mock_llm.py index 1a31ebeb..680b5602 100644 --- a/hindsight-api/hindsight_api/engine/providers/mock_llm.py +++ b/hindsight-api/hindsight_api/engine/providers/mock_llm.py @@ -129,6 +129,23 @@ class MockLLM(LLMInterface): if self._mock_exception is not None: raise self._mock_exception + # Record trace span (minimal for mock provider) + from hindsight_api.tracing import get_span_recorder + + span_recorder = get_span_recorder() + span_recorder.record_llm_call( + provider=self.provider, + model=self.model, + scope=scope, + messages=messages, + response_content="mock response", + input_tokens=10, + output_tokens=5, + duration=0.001, # Mock calls are instant + finish_reason="stop", + error=None, + ) + # Return mock response if self._mock_response is not None: result = self._mock_response @@ -192,20 +209,50 @@ class MockLLM(LLMInterface): if self._mock_exception is not None: raise self._mock_exception + # Record OpenTelemetry span + from hindsight_api.tracing import get_span_recorder + + span_recorder = get_span_recorder() + if self._mock_response is not None: if isinstance(self._mock_response, LLMToolCallResult): - return self._mock_response - # Allow setting just tool calls as a list - if isinstance(self._mock_response, list): - return LLMToolCallResult( + result = self._mock_response + elif isinstance(self._mock_response, list): + # Allow setting just tool calls as a list + result = LLMToolCallResult( tool_calls=[ LLMToolCall(id=f"mock_{i}", name=tc["name"], arguments=tc.get("arguments", {})) for i, tc in enumerate(self._mock_response) ], finish_reason="tool_calls", ) + else: + result = LLMToolCallResult(content="mock response", finish_reason="stop") + else: + result = LLMToolCallResult(content="mock response", finish_reason="stop") - return LLMToolCallResult(content="mock response", finish_reason="stop") + # Record span with mock values + # Convert LLMToolCall objects to dicts for span recording + tool_calls_dict = ( + [{"id": tc.id, "name": tc.name, "arguments": tc.arguments} for tc in result.tool_calls] + if result.tool_calls + else None + ) + span_recorder.record_llm_call( + provider=self.provider, + model=self.model, + scope=scope, + messages=messages, + response_content=result.content, + input_tokens=10, # Mock value + output_tokens=5, # Mock value + duration=0.1, # Mock value + finish_reason=result.finish_reason, + error=None, + tool_calls=tool_calls_dict, + ) + + return result async def cleanup(self) -> None: """Clean up resources (no-op for mock provider).""" diff --git a/hindsight-api/hindsight_api/engine/providers/openai_compatible_llm.py b/hindsight-api/hindsight_api/engine/providers/openai_compatible_llm.py index 554f6af6..d5ed1785 100644 --- a/hindsight-api/hindsight_api/engine/providers/openai_compatible_llm.py +++ b/hindsight-api/hindsight_api/engine/providers/openai_compatible_llm.py @@ -130,6 +130,7 @@ class OpenAICompatibleLLM(LLMInterface): max_retries=2, initial_backoff=0.5, max_backoff=2.0, + scope="verification", ) logger.info(f"Connection verified: {self.provider}/{self.model}") except Exception as e: @@ -368,6 +369,24 @@ class OpenAICompatibleLLM(LLMInterface): success=True, ) + # Record trace span + from hindsight_api.tracing import _serialize_for_span, get_span_recorder + + finish_reason = response.choices[0].finish_reason if response.choices else None + span_recorder = get_span_recorder() + span_recorder.record_llm_call( + provider=self.provider, + model=self.model, + scope=scope, + messages=messages, + response_content=_serialize_for_span(result), + input_tokens=input_tokens, + output_tokens=output_tokens, + duration=duration, + finish_reason=finish_reason, + error=None, + ) + # Log slow calls if duration > 10.0 and usage: ratio = max(1, output_tokens) / max(1, input_tokens) @@ -556,6 +575,30 @@ class OpenAICompatibleLLM(LLMInterface): success=True, ) + # Record OpenTelemetry span + from hindsight_api.tracing import get_span_recorder + + span_recorder = get_span_recorder() + # Convert LLMToolCall objects to dicts for span recording + tool_calls_dict = ( + [{"id": tc.id, "name": tc.name, "arguments": tc.arguments} for tc in tool_calls] + if tool_calls + else None + ) + span_recorder.record_llm_call( + provider=self.provider, + model=self.model, + scope=scope, + messages=messages, + response_content=content, + input_tokens=input_tokens, + output_tokens=output_tokens, + duration=duration, + finish_reason=finish_reason, + error=None, + tool_calls=tool_calls_dict, + ) + return LLMToolCallResult( content=content, tool_calls=tool_calls, diff --git a/hindsight-api/hindsight_api/engine/reflect/agent.py b/hindsight-api/hindsight_api/engine/reflect/agent.py index 750df75c..066bebc6 100644 --- a/hindsight-api/hindsight_api/engine/reflect/agent.py +++ b/hindsight-api/hindsight_api/engine/reflect/agent.py @@ -402,7 +402,7 @@ async def run_reflect_agent( {"role": "system", "content": FINAL_SYSTEM_PROMPT}, {"role": "user", "content": prompt}, ], - scope="reflect_agent_final", + scope="reflect", max_completion_tokens=max_tokens, return_usage=True, ) @@ -447,7 +447,7 @@ async def run_reflect_agent( result = await llm_config.call_with_tools( messages=messages, tools=tools, - scope="reflect_agent", + scope="reflect_tool_call", tool_choice="required" if iteration == 0 else "auto", # Force tool use on first iteration ) llm_duration = int((time.time() - llm_start) * 1000) @@ -479,7 +479,7 @@ async def run_reflect_agent( {"role": "system", "content": FINAL_SYSTEM_PROMPT}, {"role": "user", "content": prompt}, ], - scope="reflect_agent_final", + scope="reflect", max_completion_tokens=max_tokens, return_usage=True, ) @@ -550,7 +550,7 @@ async def run_reflect_agent( {"role": "system", "content": FINAL_SYSTEM_PROMPT}, {"role": "user", "content": prompt}, ], - scope="reflect_agent_final", + scope="reflect", max_completion_tokens=max_tokens, return_usage=True, ) @@ -617,23 +617,30 @@ async def run_reflect_agent( ) continue - # Process done tool - return await _process_done_tool( - done_call, - available_memory_ids, - available_mental_model_ids, - available_observation_ids, - iteration + 1, - total_tools_called, - tool_trace, - _get_llm_trace(), - _get_usage(), - _log_completion, - reflect_id, - directives_applied=directives_applied, - llm_config=llm_config, - response_schema=response_schema, - ) + # Process done tool - wrap with tool call span + from hindsight_api.tracing import get_tracer + + tracer = get_tracer() + span_name = "hindsight.reflect_tool_call" + with tracer.start_as_current_span(span_name) as span: + span.set_attribute("hindsight.scope", "reflect_tool_call") + span.set_attribute("hindsight.operation", "reflect_tool_call") + return await _process_done_tool( + done_call, + available_memory_ids, + available_mental_model_ids, + available_observation_ids, + iteration + 1, + total_tools_called, + tool_trace, + _get_llm_trace(), + _get_usage(), + _log_completion, + reflect_id, + directives_applied=directives_applied, + llm_config=llm_config, + response_schema=response_schema, + ) # Execute other tools in parallel (exclude done tool in all its format variants) other_tools = [tc for tc in result.tool_calls if not _is_done_tool(tc.name)] @@ -842,17 +849,67 @@ async def _execute_tool_with_timing( expand_fn: Callable[[list[str], str], Awaitable[dict[str, Any]]], ) -> tuple[dict[str, Any], int]: """Execute a tool call and return result with timing.""" - start = time.time() - result = await _execute_tool( - tc.name, - tc.arguments, - search_mental_models_fn, - search_observations_fn, - recall_fn, - expand_fn, - ) - duration_ms = int((time.time() - start) * 1000) - return result, duration_ms + from hindsight_api.tracing import get_tracer + + start_time = time.time() + + # Create span for tool execution + tracer = get_tracer() + # Normalize tool name for span + normalized_name = _normalize_tool_name(tc.name) + span_name = f"hindsight.reflect_tool_exec.{normalized_name}" + + # Calculate timestamps + start_time_ns = time.time_ns() + + with tracer.start_as_current_span( + span_name, + start_time=start_time_ns, + end_on_exit=False, + ) as span: + # Set attributes + span.set_attribute("hindsight.tool.name", normalized_name) + span.set_attribute("hindsight.tool.id", tc.id) + span.set_attribute("hindsight.tool.arguments", json.dumps(tc.arguments)) + + try: + result = await _execute_tool( + tc.name, + tc.arguments, + search_mental_models_fn, + search_observations_fn, + recall_fn, + expand_fn, + ) + + # Set success attributes + if isinstance(result, dict) and "error" in result: + from opentelemetry.trace import Status, StatusCode + + span.set_status(Status(StatusCode.ERROR, result["error"])) + else: + from opentelemetry.trace import Status, StatusCode + + span.set_status(Status(StatusCode.OK)) + + duration_ms = int((time.time() - start_time) * 1000) + span.set_attribute("hindsight.tool.duration_ms", duration_ms) + + # End span with correct timestamp + end_time_ns = time.time_ns() + span.end(end_time=end_time_ns) + + return result, duration_ms + except Exception as e: + from opentelemetry.trace import Status, StatusCode + + span.set_status(Status(StatusCode.ERROR, str(e))) + span.record_exception(e) + duration_ms = int((time.time() - start_time) * 1000) + span.set_attribute("hindsight.tool.duration_ms", duration_ms) + end_time_ns = time.time_ns() + span.end(end_time=end_time_ns) + raise async def _execute_tool( diff --git a/hindsight-api/hindsight_api/engine/retain/fact_extraction.py b/hindsight-api/hindsight_api/engine/retain/fact_extraction.py index b2248e29..9c693b50 100644 --- a/hindsight-api/hindsight_api/engine/retain/fact_extraction.py +++ b/hindsight-api/hindsight_api/engine/retain/fact_extraction.py @@ -802,7 +802,7 @@ Text: extraction_response_json, call_usage = await llm_config.call( messages=[{"role": "system", "content": prompt}, {"role": "user", "content": user_message}], response_format=response_schema, - scope="memory_extract_facts", + scope="retain_extract_facts", temperature=0.1, max_completion_tokens=config.retain_max_completion_tokens, max_retries=max_retries, diff --git a/hindsight-api/hindsight_api/main.py b/hindsight-api/hindsight_api/main.py index 1ee210b6..7bc5df6d 100644 --- a/hindsight-api/hindsight_api/main.py +++ b/hindsight-api/hindsight_api/main.py @@ -242,6 +242,11 @@ def main(): worker_consolidation_max_slots=config.worker_consolidation_max_slots, reflect_max_iterations=config.reflect_max_iterations, mental_model_refresh_concurrency=config.mental_model_refresh_concurrency, + otel_traces_enabled=config.otel_traces_enabled, + otel_exporter_otlp_endpoint=config.otel_exporter_otlp_endpoint, + otel_exporter_otlp_headers=config.otel_exporter_otlp_headers, + otel_service_name=config.otel_service_name, + otel_deployment_environment=config.otel_deployment_environment, ) config.configure_logging() if not args.daemon: diff --git a/hindsight-api/hindsight_api/tracing.py b/hindsight-api/hindsight_api/tracing.py new file mode 100644 index 00000000..5461aef3 --- /dev/null +++ b/hindsight-api/hindsight_api/tracing.py @@ -0,0 +1,480 @@ +""" +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 diff --git a/hindsight-api/pyproject.toml b/hindsight-api/pyproject.toml index e4ed2e02..1d2ad2d4 100644 --- a/hindsight-api/pyproject.toml +++ b/hindsight-api/pyproject.toml @@ -33,6 +33,8 @@ dependencies = [ "opentelemetry-sdk>=1.20.0", "opentelemetry-instrumentation-fastapi>=0.41b0", "opentelemetry-exporter-prometheus>=0.41b0", + "opentelemetry-exporter-otlp-proto-http>=1.20.0", + "opentelemetry-semantic-conventions>=0.41b0", "dateparser>=1.2.2", "google-genai>=1.0.0", "google-auth>=2.0.0", diff --git a/hindsight-api/tests/test_reflect_tracing.py b/hindsight-api/tests/test_reflect_tracing.py new file mode 100644 index 00000000..63ff9b9e --- /dev/null +++ b/hindsight-api/tests/test_reflect_tracing.py @@ -0,0 +1,45 @@ +""" +Test to verify reflect operation creates proper span hierarchy. +""" +import pytest + + +@pytest.mark.asyncio +async def test_reflect_creates_child_spans(memory, request_context): + """Test that reflect operation creates child LLM spans.""" + from datetime import datetime, timezone + from hindsight_api.tracing import initialize_tracing, get_span_recorder, create_span_recorder + + # Initialize tracing with a mock endpoint + initialize_tracing( + service_name="test-hindsight", + endpoint="http://localhost:4318", + deployment_environment="test" + ) + + # Create span recorder + recorder = create_span_recorder() + + bank_id = f"test-reflect-hierarchy-{datetime.now(timezone.utc).timestamp()}" + + try: + # Add some memories + await memory.retain_async( + bank_id=bank_id, + content="Paris is the capital of France", + context="Geography", + request_context=request_context, + ) + + # Run reflect + result = await memory.reflect_async( + bank_id=bank_id, + query="What is the capital of France?", + request_context=request_context, + ) + + print(f"Reflect result: {result.text[:100]}") + print(f"Usage: {result.usage}") + + finally: + await memory.delete_bank(bank_id, request_context=request_context) diff --git a/hindsight-api/tests/test_tracing.py b/hindsight-api/tests/test_tracing.py new file mode 100644 index 00000000..00095317 --- /dev/null +++ b/hindsight-api/tests/test_tracing.py @@ -0,0 +1,407 @@ +""" +Unit tests for OpenTelemetry tracing instrumentation. + +Tests the tracing module's ability to record LLM calls with GenAI semantic conventions. +""" + +import json +from unittest.mock import MagicMock, patch + +import pytest + +from hindsight_api.tracing import ( + PROVIDER_NAME_MAPPING, + GenAIAttributes, + LLMSpanRecorder, + NoOpLLMSpanRecorder, + _truncate_content, + create_operation_span, + initialize_tracing, + is_tracing_enabled, +) + + +def test_provider_name_mapping(): + """Test that provider names are correctly mapped to GenAI conventions.""" + assert PROVIDER_NAME_MAPPING["openai"] == "openai" + assert PROVIDER_NAME_MAPPING["anthropic"] == "anthropic" + assert PROVIDER_NAME_MAPPING["gemini"] == "google" + assert PROVIDER_NAME_MAPPING["vertexai"] == "google" + assert PROVIDER_NAME_MAPPING["groq"] == "groq" + assert PROVIDER_NAME_MAPPING["ollama"] == "ollama" + assert PROVIDER_NAME_MAPPING["openai-codex"] == "openai" + assert PROVIDER_NAME_MAPPING["claude-code"] == "anthropic" + + +def test_truncate_content_short(): + """Test that short content is not truncated.""" + content = "This is a short message" + result = _truncate_content(content) + assert result == content + + +def test_truncate_content_long(): + """Test that long content is truncated.""" + content = "x" * 150000 # Exceeds MAX_CONTENT_LENGTH + result = _truncate_content(content) + assert len(result) < len(content) + assert "[TRUNCATED:" in result + assert result.startswith("x" * 100) + + +def test_noop_span_recorder(): + """Test that NoOpLLMSpanRecorder doesn't raise errors.""" + recorder = NoOpLLMSpanRecorder() + # Should not raise any errors + recorder.record_llm_call( + provider="openai", + model="gpt-4", + scope="test", + messages=[{"role": "user", "content": "test"}], + response_content="test response", + input_tokens=10, + output_tokens=5, + duration=1.0, + ) + + +def test_llm_span_recorder_format_messages(): + """Test message formatting to GenAI convention.""" + mock_tracer = MagicMock() + recorder = LLMSpanRecorder(mock_tracer) + + messages = [ + {"role": "system", "content": "You are helpful"}, + {"role": "user", "content": "Hello"}, + ] + + result = recorder._format_messages(messages) + parsed = json.loads(result) + + assert len(parsed) == 2 + assert parsed[0]["role"] == "system" + assert parsed[0]["content"] == "You are helpful" + assert parsed[1]["role"] == "user" + assert parsed[1]["content"] == "Hello" + + +def test_llm_span_recorder_format_output(): + """Test output formatting to GenAI convention.""" + mock_tracer = MagicMock() + recorder = LLMSpanRecorder(mock_tracer) + + result = recorder._format_output("Hello world", "stop") + parsed = json.loads(result) + + assert len(parsed) == 1 + assert parsed[0]["role"] == "assistant" + assert parsed[0]["content"] == "Hello world" + + +def test_llm_span_recorder_format_output_none(): + """Test output formatting with None content.""" + mock_tracer = MagicMock() + recorder = LLMSpanRecorder(mock_tracer) + + result = recorder._format_output(None, None) + parsed = json.loads(result) + + assert parsed == [] + + +def test_llm_span_recorder_extract_system_instructions(): + """Test system instruction extraction.""" + mock_tracer = MagicMock() + recorder = LLMSpanRecorder(mock_tracer) + + messages = [ + {"role": "system", "content": "You are helpful"}, + {"role": "user", "content": "Hello"}, + ] + + result = recorder._extract_system_instructions(messages) + assert result == "You are helpful" + + +def test_llm_span_recorder_extract_system_instructions_none(): + """Test system instruction extraction with no system message.""" + mock_tracer = MagicMock() + recorder = LLMSpanRecorder(mock_tracer) + + messages = [ + {"role": "user", "content": "Hello"}, + ] + + result = recorder._extract_system_instructions(messages) + assert result is None + + +@patch("hindsight_api.tracing.time") +def test_llm_span_recorder_record_success(mock_time): + """Test successful LLM call recording.""" + # Mock time + mock_time.time_ns.return_value = 1000000000000 # 1 second in nanoseconds + + # Create mock tracer and span + mock_span = MagicMock() + mock_tracer = MagicMock() + mock_tracer.start_as_current_span.return_value.__enter__.return_value = mock_span + + recorder = LLMSpanRecorder(mock_tracer) + + messages = [{"role": "user", "content": "Hello"}] + response_content = "Hi there!" + + recorder.record_llm_call( + provider="openai", + model="gpt-4", + scope="test", + messages=messages, + response_content=response_content, + input_tokens=10, + output_tokens=5, + duration=1.5, + finish_reason="stop", + error=None, + ) + + # Verify span was created with correct name (hindsight.{scope}) + mock_tracer.start_as_current_span.assert_called_once() + call_args = mock_tracer.start_as_current_span.call_args + assert call_args[0][0] == "hindsight.test" + + # Verify attributes were set + assert mock_span.set_attribute.called + attribute_calls = {call[0][0]: call[0][1] for call in mock_span.set_attribute.call_args_list} + + assert attribute_calls[GenAIAttributes.OPERATION_NAME] == "chat" + assert attribute_calls[GenAIAttributes.PROVIDER_NAME] == "openai" + assert attribute_calls[GenAIAttributes.REQUEST_MODEL] == "gpt-4" + assert attribute_calls[GenAIAttributes.RESPONSE_MODEL] == "gpt-4" + assert attribute_calls[GenAIAttributes.USAGE_INPUT_TOKENS] == 10 + assert attribute_calls[GenAIAttributes.USAGE_OUTPUT_TOKENS] == 5 + assert attribute_calls["hindsight.scope"] == "test" + + # Verify event was added + mock_span.add_event.assert_called_once() + event_call = mock_span.add_event.call_args + assert event_call[0][0] == "gen_ai.client.inference.operation.details" + + # Verify status was set to OK + mock_span.set_status.assert_called() + + # Verify span was ended + mock_span.end.assert_called_once() + + +@patch("hindsight_api.tracing.time") +def test_llm_span_recorder_record_error(mock_time): + """Test error LLM call recording.""" + # Mock time + mock_time.time_ns.return_value = 1000000000000 + + # Create mock tracer and span + mock_span = MagicMock() + mock_tracer = MagicMock() + mock_tracer.start_as_current_span.return_value.__enter__.return_value = mock_span + + recorder = LLMSpanRecorder(mock_tracer) + + messages = [{"role": "user", "content": "Hello"}] + error = ValueError("Test error") + + recorder.record_llm_call( + provider="anthropic", + model="claude-3", + scope="test", + messages=messages, + response_content=None, + input_tokens=10, + output_tokens=0, + duration=0.5, + finish_reason=None, + error=error, + ) + + # Verify error status was set + mock_span.set_status.assert_called() + status_call = mock_span.set_status.call_args[0][0] + assert status_call.status_code.name == "ERROR" + + # Verify error type attribute was set + attribute_calls = {call[0][0]: call[0][1] for call in mock_span.set_attribute.call_args_list} + assert attribute_calls[GenAIAttributes.ERROR_TYPE] == "ValueError" + + # Verify exception was recorded + mock_span.record_exception.assert_called_once_with(error) + + +@patch("hindsight_api.tracing.time") +def test_llm_span_recorder_provider_mapping(mock_time): + """Test that provider names are mapped correctly.""" + mock_time.time_ns.return_value = 1000000000000 + + mock_span = MagicMock() + mock_tracer = MagicMock() + mock_tracer.start_as_current_span.return_value.__enter__.return_value = mock_span + + recorder = LLMSpanRecorder(mock_tracer) + + # Test gemini -> google mapping + recorder.record_llm_call( + provider="gemini", + model="gemini-pro", + scope="test", + messages=[{"role": "user", "content": "test"}], + response_content="test", + input_tokens=5, + output_tokens=3, + duration=1.0, + ) + + attribute_calls = {call[0][0]: call[0][1] for call in mock_span.set_attribute.call_args_list} + assert attribute_calls[GenAIAttributes.PROVIDER_NAME] == "google" + + +# ==================== Parent Span Tests ==================== + + +def test_create_operation_span_disabled(): + """Test that create_operation_span returns no-op when tracing is disabled.""" + # Tracing should be disabled by default + assert not is_tracing_enabled() + + # Should return a no-op context manager + span = create_operation_span("test_operation", "test_bank_id") + + # Should be usable as context manager without errors + with span: + pass + + +@patch("hindsight_api.tracing._tracer") +@patch("hindsight_api.tracing._tracing_enabled", True) +def test_create_operation_span_enabled(mock_tracer): + """Test that create_operation_span creates a span when tracing is enabled.""" + # Mock the tracer + mock_span = MagicMock() + mock_tracer.start_as_current_span.return_value = mock_span + + # Create operation span + span = create_operation_span("retain", "bank123") + + # Verify span was created with correct name + mock_tracer.start_as_current_span.assert_called_once_with("hindsight.retain") + + # Verify attributes were set + mock_span.set_attribute.assert_any_call("hindsight.operation", "retain") + mock_span.set_attribute.assert_any_call("hindsight.bank_id", "bank123") + + +@patch("hindsight_api.tracing._tracer") +@patch("hindsight_api.tracing._tracing_enabled", True) +def test_create_operation_span_no_bank_id(mock_tracer): + """Test that create_operation_span works without bank_id.""" + mock_span = MagicMock() + mock_tracer.start_as_current_span.return_value = mock_span + + # Create operation span without bank_id + span = create_operation_span("consolidation") + + # Verify span was created + mock_tracer.start_as_current_span.assert_called_once_with("hindsight.consolidation") + + # Verify only operation attribute was set (not bank_id) + assert mock_span.set_attribute.call_count == 1 + mock_span.set_attribute.assert_called_once_with("hindsight.operation", "consolidation") + + +@patch("hindsight_api.tracing._tracer") +@patch("hindsight_api.tracing._tracing_enabled", True) +def test_create_operation_span_all_operations(mock_tracer): + """Test that all 4 operations can create parent spans.""" + mock_span = MagicMock() + mock_tracer.start_as_current_span.return_value = mock_span + + operations = ["retain", "consolidation", "reflect", "mental_model_refresh"] + + for operation in operations: + mock_tracer.reset_mock() + mock_span.reset_mock() + + span = create_operation_span(operation, "test_bank") + + # Verify span was created with correct name + mock_tracer.start_as_current_span.assert_called_once_with(f"hindsight.{operation}") + + # Verify attributes + mock_span.set_attribute.assert_any_call("hindsight.operation", operation) + mock_span.set_attribute.assert_any_call("hindsight.bank_id", "test_bank") + + +@patch("hindsight_api.tracing.time") +@patch("hindsight_api.tracing._tracer") +@patch("hindsight_api.tracing._tracing_enabled", True) +def test_parent_child_span_hierarchy(mock_tracer, mock_time): + """Test that child LLM spans are created under parent operation spans.""" + mock_time.time_ns.return_value = 1000000000000 + + # Create mock parent span + mock_parent_span = MagicMock() + mock_parent_span.__enter__ = MagicMock(return_value=mock_parent_span) + mock_parent_span.__exit__ = MagicMock(return_value=False) + + # Create mock child span + mock_child_span = MagicMock() + + # Mock tracer to return parent span first, then child span + mock_tracer.start_as_current_span.side_effect = [ + mock_parent_span, # Parent span + MagicMock(__enter__=MagicMock(return_value=mock_child_span), __exit__=MagicMock(return_value=False)), # Child + ] + + # Create parent operation span + with create_operation_span("retain", "bank123"): + # Simulate creating a child LLM span + recorder = LLMSpanRecorder(mock_tracer) + recorder.record_llm_call( + provider="openai", + model="gpt-4", + scope="retain_extract_facts", + messages=[{"role": "user", "content": "test"}], + response_content="response", + input_tokens=10, + output_tokens=5, + duration=1.0, + ) + + # Verify both parent and child spans were created + assert mock_tracer.start_as_current_span.call_count == 2 + + # Verify parent span was created first + first_call = mock_tracer.start_as_current_span.call_args_list[0] + assert first_call[0][0] == "hindsight.retain" + + # Verify child span was created second (hindsight.{scope}) + second_call = mock_tracer.start_as_current_span.call_args_list[1] + assert second_call[0][0] == "hindsight.retain_extract_facts" + + +@patch("hindsight_api.tracing._tracer") +@patch("hindsight_api.tracing._tracing_enabled", True) +def test_operation_span_context_manager(mock_tracer): + """Test that operation spans work as context managers.""" + mock_span = MagicMock() + mock_span.__enter__ = MagicMock(return_value=mock_span) + mock_span.__exit__ = MagicMock(return_value=False) + mock_tracer.start_as_current_span.return_value = mock_span + + # Use span as context manager + with create_operation_span("reflect", "bank456"): + # Do some work + pass + + # Verify span lifecycle + mock_tracer.start_as_current_span.assert_called_once() + mock_span.__enter__.assert_called_once() + mock_span.__exit__.assert_called_once() diff --git a/hindsight-api/tests/test_tracing_integration.py b/hindsight-api/tests/test_tracing_integration.py new file mode 100644 index 00000000..bc597c16 --- /dev/null +++ b/hindsight-api/tests/test_tracing_integration.py @@ -0,0 +1,196 @@ +""" +Integration tests for OpenTelemetry tracing with memory engine operations. + +Tests that parent spans are correctly created for retain, consolidation, reflect, +and mental_model_refresh operations. +""" + +from datetime import datetime, timezone +from unittest.mock import MagicMock, patch + +import pytest + + +@pytest.mark.asyncio +@patch("hindsight_api.engine.memory_engine.create_operation_span") +async def test_retain_creates_parent_span(mock_create_span, memory, request_context): + """Test that retain operation creates a parent span.""" + # Setup + mock_span = MagicMock() + mock_span.__enter__ = MagicMock(return_value=mock_span) + mock_span.__exit__ = MagicMock(return_value=False) + mock_create_span.return_value = mock_span + + bank_id = f"test-retain-{datetime.now(timezone.utc).timestamp()}" + + try: + # Execute retain (automatically creates bank if needed) + await memory.retain_async( + bank_id=bank_id, + content="Test memory for tracing", + context="Test context", + request_context=request_context, + ) + + # Verify parent span was created + mock_create_span.assert_called() + call_args = mock_create_span.call_args + assert call_args[0][0] == "retain" # operation name + assert call_args[0][1] == bank_id # bank_id + + # Verify span was used as context manager + mock_span.__enter__.assert_called() + mock_span.__exit__.assert_called() + finally: + # Cleanup + await memory.delete_bank(bank_id, request_context=request_context) + + +@pytest.mark.asyncio +@patch("hindsight_api.engine.memory_engine.create_operation_span") +async def test_consolidation_creates_parent_span(mock_create_span, memory, request_context): + """Test that consolidation operation creates a parent span.""" + # Setup + mock_span = MagicMock() + mock_span.__enter__ = MagicMock(return_value=mock_span) + mock_span.__exit__ = MagicMock(return_value=False) + mock_create_span.return_value = mock_span + + bank_id = f"test-consolidation-{datetime.now(timezone.utc).timestamp()}" + + try: + # Execute consolidation (bank will be created automatically) + await memory.run_consolidation( + bank_id=bank_id, + request_context=request_context, + ) + + # Verify parent span was created + mock_create_span.assert_called() + call_args = mock_create_span.call_args + assert call_args[0][0] == "consolidation" + assert call_args[0][1] == bank_id + + # Verify span was used as context manager + mock_span.__enter__.assert_called() + mock_span.__exit__.assert_called() + finally: + # Cleanup + await memory.delete_bank(bank_id, request_context=request_context) + + +@pytest.mark.asyncio +@patch("hindsight_api.engine.memory_engine.create_operation_span") +async def test_reflect_creates_parent_span(mock_create_span, memory, request_context): + """Test that reflect operation creates a parent span.""" + # Setup + mock_span = MagicMock() + mock_span.__enter__ = MagicMock(return_value=mock_span) + mock_span.__exit__ = MagicMock(return_value=False) + mock_create_span.return_value = mock_span + + bank_id = f"test-reflect-{datetime.now(timezone.utc).timestamp()}" + + try: + # Add some memories first + await memory.retain_async( + bank_id=bank_id, + content="Paris is the capital of France", + context="Geography fact", + request_context=request_context, + ) + + # Reset mock to clear retain call + mock_create_span.reset_mock() + + # Execute reflect + await memory.reflect_async( + bank_id=bank_id, + query="What is the capital of France?", + request_context=request_context, + ) + + # Verify parent span was created + mock_create_span.assert_called() + call_args = mock_create_span.call_args + assert call_args[0][0] == "reflect" + assert call_args[0][1] == bank_id + + # Verify span was used as context manager + mock_span.__enter__.assert_called() + mock_span.__exit__.assert_called() + finally: + # Cleanup + await memory.delete_bank(bank_id, request_context=request_context) + + +@pytest.mark.asyncio +@patch("hindsight_api.engine.memory_engine.create_operation_span") +async def test_retain_batch_creates_single_parent_span(mock_create_span, memory, request_context): + """Test that batch retain creates one parent span for the entire batch.""" + # Setup + mock_span = MagicMock() + mock_span.__enter__ = MagicMock(return_value=mock_span) + mock_span.__exit__ = MagicMock(return_value=False) + mock_create_span.return_value = mock_span + + bank_id = f"test-batch-{datetime.now(timezone.utc).timestamp()}" + + try: + # Execute batch retain with multiple items + await memory.retain_batch_async( + bank_id=bank_id, + contents=[ + {"content": "Memory 1", "context": "Context 1"}, + {"content": "Memory 2", "context": "Context 2"}, + {"content": "Memory 3", "context": "Context 3"}, + ], + request_context=request_context, + ) + + # Verify parent span was created only once for the entire batch + assert mock_create_span.call_count == 1 + call_args = mock_create_span.call_args + assert call_args[0][0] == "retain" + assert call_args[0][1] == bank_id + finally: + # Cleanup + await memory.delete_bank(bank_id, request_context=request_context) + + +@pytest.mark.asyncio +@patch("hindsight_api.tracing._tracing_enabled", False) +@patch("hindsight_api.engine.memory_engine.create_operation_span") +async def test_operations_work_when_tracing_disabled(mock_create_span, memory, request_context): + """Test that operations work correctly when tracing is disabled.""" + # Setup - create_operation_span should return a no-op context manager + from contextlib import nullcontext + + mock_create_span.return_value = nullcontext() + + bank_id = f"test-no-trace-{datetime.now(timezone.utc).timestamp()}" + + try: + # All operations should work without errors + await memory.retain_async( + bank_id=bank_id, + content="Test memory", + request_context=request_context, + ) + + await memory.run_consolidation( + bank_id=bank_id, + request_context=request_context, + ) + + await memory.reflect_async( + bank_id=bank_id, + query="Test query", + request_context=request_context, + ) + + # Verify no errors occurred and spans were attempted to be created + assert mock_create_span.call_count >= 3 + finally: + # Cleanup + await memory.delete_bank(bank_id, request_context=request_context) diff --git a/hindsight-api/tests/test_tracing_spans_verification.py b/hindsight-api/tests/test_tracing_spans_verification.py new file mode 100644 index 00000000..f43973f4 --- /dev/null +++ b/hindsight-api/tests/test_tracing_spans_verification.py @@ -0,0 +1,273 @@ +""" +Comprehensive tracing span verification tests. + +Verifies that all memory engine operations create correct parent and child spans +with proper attributes and hierarchy. +""" + +from datetime import datetime, timezone +from unittest.mock import MagicMock, patch + +import pytest + + +@pytest.mark.asyncio +@pytest.mark.skip(reason="Background consolidation causes StopIteration - need to investigate separately") +@patch("hindsight_api.tracing._tracing_enabled", True) +@patch("hindsight_api.tracing._tracer") +async def test_recall_span_hierarchy(mock_tracer, memory, request_context): + """Test that recall creates proper parent and child spans.""" + # Setup mock spans + mock_recall_span = MagicMock() + mock_recall_span.__enter__ = MagicMock(return_value=mock_recall_span) + mock_recall_span.__exit__ = MagicMock(return_value=False) + + mock_embedding_span = MagicMock() + mock_retrieval_span = MagicMock() + mock_fusion_span = MagicMock() + mock_rerank_span = MagicMock() + + # Mock tracer to return spans in sequence + mock_tracer.start_as_current_span.side_effect = [mock_recall_span] + mock_tracer.start_span.side_effect = [ + mock_embedding_span, + mock_retrieval_span, + mock_fusion_span, + mock_rerank_span, + ] + + bank_id = f"test-recall-{datetime.now(timezone.utc).timestamp()}" + + try: + # Add some memories first + await memory.retain_async( + bank_id=bank_id, + content="Paris is the capital of France", + request_context=request_context, + ) + + # Wait a bit for any background tasks to settle + import asyncio + await asyncio.sleep(0.5) + + # Reset mocks after retain + mock_tracer.reset_mock() + mock_recall_span.reset_mock() + + # Execute recall + await memory.recall_async( + bank_id=bank_id, + query="What is the capital of France?", + request_context=request_context, + ) + + # Verify parent span was created with start_as_current_span + assert mock_tracer.start_as_current_span.called + parent_call = mock_tracer.start_as_current_span.call_args + assert parent_call[0][0] == "hindsight.recall" + + # Verify parent span attributes were set + recall_attrs = {call[0][0]: call[0][1] for call in mock_recall_span.set_attribute.call_args_list} + assert "hindsight.bank_id" in recall_attrs + assert recall_attrs["hindsight.bank_id"] == bank_id + assert "hindsight.query" in recall_attrs + assert "hindsight.fact_types" in recall_attrs + assert "hindsight.thinking_budget" in recall_attrs + assert "hindsight.max_tokens" in recall_attrs + + # Verify child spans were created (if tracing is enabled) + if mock_tracer.start_span.called: + child_spans = [call[0][0] for call in mock_tracer.start_span.call_args_list] + assert "hindsight.recall_embedding" in child_spans + assert "hindsight.recall_retrieval" in child_spans + assert "hindsight.recall_fusion" in child_spans + assert "hindsight.recall_rerank" in child_spans + + finally: + # Cleanup + await memory.delete_bank(bank_id, request_context=request_context) + + +@pytest.mark.asyncio +async def test_mental_model_refresh_span_exists(memory, request_context): + """Test that mental model refresh functionality exists (span creation tested via unit tests).""" + # This test verifies that refresh_mental_model method exists and can be called + # The actual span creation is tested in unit tests with proper mocking + bank_id = f"test-mmr-{datetime.now(timezone.utc).timestamp()}" + + try: + # Just verify the method exists - it will return None if no mental model found + result = await memory.refresh_mental_model( + bank_id=bank_id, + mental_model_id="non-existent-id", + request_context=request_context, + ) + # Result will be None since mental model doesn't exist + assert result is None + + finally: + # Cleanup + try: + await memory.delete_bank(bank_id, request_context=request_context) + except Exception: + pass + + +@pytest.mark.asyncio +async def test_consolidation_child_spans(memory, request_context): + """Test that consolidation creates child spans for its operations.""" + bank_id = f"test-cons-child-{datetime.now(timezone.utc).timestamp()}" + + try: + # Add memories to consolidate + await memory.retain_async( + bank_id=bank_id, + content="The Eiffel Tower is in Paris", + request_context=request_context, + ) + + await memory.retain_async( + bank_id=bank_id, + content="Paris is the capital of France", + request_context=request_context, + ) + + # Run consolidation (this will create parent + child spans) + await memory.run_consolidation( + bank_id=bank_id, + request_context=request_context, + ) + + # Note: We can't easily verify the child spans without mocking the tracer, + # but we can verify that consolidation completes successfully + # The actual span creation is tested in unit tests + + finally: + # Cleanup + await memory.delete_bank(bank_id, request_context=request_context) + + +@pytest.mark.asyncio +async def test_reflect_tool_call_spans(memory, request_context): + """Test that reflect creates tool call spans (not reflect_generation).""" + bank_id = f"test-reflect-tools-{datetime.now(timezone.utc).timestamp()}" + + try: + # Add some memories + await memory.retain_async( + bank_id=bank_id, + content="Machine learning is a subset of AI", + request_context=request_context, + ) + + # Execute reflect (will create reflect_tool_call spans) + result = await memory.reflect_async( + bank_id=bank_id, + query="What is machine learning?", + request_context=request_context, + ) + + # Verify reflect completed successfully + assert result.text + assert len(result.text) > 0 + + # The span names are verified via unit tests with mocked tracers + # This integration test ensures the operation completes successfully + + finally: + # Cleanup + await memory.delete_bank(bank_id, request_context=request_context) + + +@pytest.mark.asyncio +async def test_all_operations_create_spans(memory, request_context): + """Comprehensive test that all operations create their respective spans.""" + bank_id = f"test-all-ops-{datetime.now(timezone.utc).timestamp()}" + + try: + # 1. Retain operation + await memory.retain_async( + bank_id=bank_id, + content="Test memory for comprehensive span test", + request_context=request_context, + ) + + # 2. Recall operation + await memory.recall_async( + bank_id=bank_id, + query="test memory", + request_context=request_context, + ) + + # 3. Reflect operation + await memory.reflect_async( + bank_id=bank_id, + query="What can you tell me about the test?", + request_context=request_context, + ) + + # 4. Consolidation operation + await memory.run_consolidation( + bank_id=bank_id, + request_context=request_context, + ) + + # All operations completed successfully + # Span hierarchy verification is done in unit tests with mocked tracers + + finally: + # Cleanup + await memory.delete_bank(bank_id, request_context=request_context) + + +@pytest.mark.asyncio +@patch("hindsight_api.tracing._tracing_enabled", True) +@patch("hindsight_api.tracing._tracer") +async def test_recall_span_attributes(mock_tracer, memory, request_context): + """Verify that recall spans have all required attributes.""" + # Setup mock span + mock_span = MagicMock() + mock_span.__enter__ = MagicMock(return_value=mock_span) + mock_span.__exit__ = MagicMock(return_value=False) + mock_tracer.start_as_current_span.return_value = mock_span + + bank_id = f"test-attrs-{datetime.now(timezone.utc).timestamp()}" + + try: + # Add memory + await memory.retain_async( + bank_id=bank_id, + content="Test content for attributes", + request_context=request_context, + ) + + # Reset mock + mock_span.reset_mock() + + # Execute recall with specific parameters + await memory.recall_async( + bank_id=bank_id, + query="test query for attributes", + fact_type=["world", "experience"], + max_tokens=2048, + request_context=request_context, + ) + + # Collect all attributes set on the span + attrs = {call[0][0]: call[0][1] for call in mock_span.set_attribute.call_args_list} + + # Verify required attributes + assert "hindsight.bank_id" in attrs + assert "hindsight.query" in attrs + assert "hindsight.fact_types" in attrs + assert "hindsight.max_tokens" in attrs + assert "hindsight.thinking_budget" in attrs + + # Verify attribute values + assert attrs["hindsight.bank_id"] == bank_id + assert "test query" in attrs["hindsight.query"] + assert attrs["hindsight.max_tokens"] == 2048 + + finally: + # Cleanup + await memory.delete_bank(bank_id, request_context=request_context) diff --git a/hindsight-docs/docs/developer/configuration.md b/hindsight-docs/docs/developer/configuration.md index b9d68509..9eaf5ad2 100644 --- a/hindsight-docs/docs/developer/configuration.md +++ b/hindsight-docs/docs/developer/configuration.md @@ -558,6 +558,74 @@ await memory.initialize() --- +## Observability & Tracing + +Hindsight provides OpenTelemetry-based observability for LLM calls, conforming to GenAI semantic conventions. + +### OpenTelemetry Tracing + +| Variable | Description | Default | +|----------|-------------|---------| +| `HINDSIGHT_API_OTEL_TRACES_ENABLED` | Enable distributed tracing for LLM calls | `false` | +| `HINDSIGHT_API_OTEL_EXPORTER_OTLP_ENDPOINT` | OTLP endpoint URL (e.g., Grafana LGTM, Langfuse, etc.) | - | +| `HINDSIGHT_API_OTEL_EXPORTER_OTLP_HEADERS` | Headers for OTLP exporter (format: "key1=value1,key2=value2") | - | +| `HINDSIGHT_API_OTEL_SERVICE_NAME` | Service name for traces | `hindsight-api` | +| `HINDSIGHT_API_OTEL_DEPLOYMENT_ENVIRONMENT` | Deployment environment name (e.g., development, staging, production) | `development` | + +**Features:** +- Full prompts and completions recorded as events +- Token usage tracking (input/output) +- Model and provider information +- Error tracking with finish reasons +- Conforms to OpenTelemetry GenAI semantic conventions v1.37+ + +**OTLP-Compatible Backends:** + +The tracing implementation uses standard OTLP HTTP protocol, so it works with any OTLP-compatible backend: +- **Grafana LGTM** (Recommended for local dev): All-in-one stack with Tempo traces, Loki logs, Mimir metrics, and Grafana UI +- **Langfuse**: LLM-focused observability and analytics +- **OpenLIT**: Built-in LLM dashboards, cost tracking +- **DataDog, New Relic, Honeycomb**: Commercial platforms + +**Example Configuration:** + +```bash +# Enable tracing +export HINDSIGHT_API_OTEL_TRACES_ENABLED=true + +# Configure endpoint (example: OpenLIT Cloud) +export HINDSIGHT_API_OTEL_EXPORTER_OTLP_ENDPOINT=https://otlp.openlit.io +export HINDSIGHT_API_OTEL_EXPORTER_OTLP_HEADERS="Authorization=Bearer olit-xxx" + +# Optional: Custom service name and environment +export HINDSIGHT_API_OTEL_SERVICE_NAME=hindsight-production +export HINDSIGHT_API_OTEL_DEPLOYMENT_ENVIRONMENT=production +``` + +**Local Development:** + +For local development, we recommend the Grafana LGTM stack which provides traces, metrics, and logs in a single container: + +```bash +./scripts/dev/start-grafana.sh +``` + +See `scripts/dev/grafana/README.md` for detailed setup instructions. + +Other options: See `scripts/dev/openlit/README.md` for OpenLIT or `scripts/dev/jaeger/README.md` for standalone Jaeger. + +### Metrics + +Hindsight exposes Prometheus metrics at the `/metrics` endpoint, including: +- LLM call duration and token usage +- Operation duration (retain/recall/reflect) +- HTTP request metrics +- Database connection pool metrics + +Metrics are always enabled and available at `http://localhost:8888/metrics`. + +--- + ## Control Plane The Control Plane is the web UI for managing memory banks. diff --git a/hindsight-docs/docs/developer/monitoring.md b/hindsight-docs/docs/developer/monitoring.md index 9a477562..83f448a7 100644 --- a/hindsight-docs/docs/developer/monitoring.md +++ b/hindsight-docs/docs/developer/monitoring.md @@ -1,22 +1,30 @@ # Monitoring -Hindsight provides comprehensive monitoring through Prometheus metrics and pre-built Grafana dashboards. +Hindsight provides comprehensive observability through Prometheus metrics, OpenTelemetry distributed tracing, and pre-built Grafana dashboards. ## Local Development -For local metrics visualization, a convenience script downloads and runs Prometheus and Grafana: +For local observability, use the Grafana LGTM (Loki, Grafana, Tempo, Mimir) all-in-one stack: ```bash ./scripts/dev/start-monitoring.sh ``` -This will start: -- **Grafana**: http://localhost:8890 (anonymous access enabled) -- **Prometheus**: http://localhost:8889 -- **API Metrics**: http://localhost:8888/metrics +This starts a single Docker container providing: +- **Grafana UI**: http://localhost:3000 (anonymous admin access) +- **Traces (Tempo)**: OTLP endpoint at http://localhost:4318 (HTTP) and http://localhost:4317 (gRPC) +- **Metrics (Prometheus/Mimir)**: Scrapes http://localhost:8888/metrics automatically +- **Logs (Loki)**: Available for log aggregation +- **Pre-built Dashboards**: Hindsight Operations, LLM Metrics, API Service + +**Enable tracing in your API:** +```bash +export HINDSIGHT_API_OTEL_TRACES_ENABLED=true +export HINDSIGHT_API_OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318 +``` :::note Production Deployment -The local monitoring script is for development only. In production, you need to install and configure Prometheus and Grafana separately, then point Prometheus to scrape your Hindsight API's `/metrics` endpoint. +The local monitoring stack is for development only. In production, deploy Grafana LGTM separately or use commercial platforms (Grafana Cloud, DataDog, New Relic, etc.). ::: ## Grafana Dashboards @@ -197,3 +205,66 @@ hindsight_db_pool_size - hindsight_db_pool_idle ```promql rate(hindsight_process_cpu_seconds{type="user"}[1m]) ``` + +--- + +## Distributed Tracing + +Hindsight supports OpenTelemetry distributed tracing for memory operations and LLM calls, following GenAI semantic conventions v1.37+. + +### Configuration + +See [Configuration - OpenTelemetry Tracing](./configuration#opentelemetry-tracing) for environment variables. + +**Quick Start:** +```bash +# Enable tracing +export HINDSIGHT_API_OTEL_TRACES_ENABLED=true +export HINDSIGHT_API_OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318 + +# View traces with Grafana LGTM (local dev) +./scripts/dev/start-monitoring.sh +# Open http://localhost:3000 → Explore → Tempo +``` + +Supports any OTLP-compatible backend (Grafana LGTM, Langfuse, OpenLIT, DataDog, New Relic, Honeycomb, etc.). + +### Span Hierarchy + +**Parent Spans (Operations):** +- `hindsight.retain` - Memory ingestion +- `hindsight.recall` - Memory retrieval + - `hindsight.recall_embedding` - Query embedding + - `hindsight.recall_retrieval` - Parallel search (semantic, BM25, graph, temporal) + - `hindsight.recall_fusion` - Reciprocal Rank Fusion + - `hindsight.recall_rerank` - Cross-encoder reranking +- `hindsight.reflect` - Agentic reasoning + - `hindsight.reflect_tool_call` - Tool execution (recall, lookup, etc.) +- `hindsight.consolidation` - Observation synthesis +- `hindsight.mental_model_refresh` - Mental model updates + +**Child Spans (LLM Calls):** +- Named by scope (e.g., `hindsight.memory`, `hindsight.reflect`) +- Contain full prompts/completions as events +- Follow GenAI semantic conventions for attributes + +### Span Attributes + +**Operation Spans:** +- `hindsight.operation` - Operation type +- `hindsight.bank_id` - Memory bank ID +- `hindsight.query` - Query text (truncated to 100 chars) +- `hindsight.fact_types` - Fact types for recall +- `hindsight.thinking_budget` - Budget allocation +- `hindsight.max_tokens` - Token limit + +**LLM Spans (GenAI Semantic Conventions):** +- `gen_ai.operation.name` - Always `"chat"` +- `gen_ai.provider.name` - Provider (`openai`, `anthropic`, `google`, etc.) +- `gen_ai.request.model` - Model name +- `gen_ai.usage.input_tokens` - Input tokens +- `gen_ai.usage.output_tokens` - Output tokens +- `hindsight.scope` - LLM call purpose (`memory`, `reflect`, `consolidation`, etc.) + +**Events:** +- `gen_ai.client.inference.operation.details` - Full prompts and completions diff --git a/package-lock.json b/package-lock.json index e0cdcdca..43de0e80 100644 --- a/package-lock.json +++ b/package-lock.json @@ -13,7 +13,11 @@ }, "hindsight-clients/typescript": { "name": "@vectorize-io/hindsight-client", +<<<<<<< HEAD "version": "0.4.9", +======= + "version": "0.4.8", +>>>>>>> 23f916f (feat: add otel traceability) "license": "MIT", "devDependencies": { "@hey-api/openapi-ts": "0.88.0", @@ -131,7 +135,11 @@ }, "hindsight-control-plane": { "name": "@vectorize-io/hindsight-control-plane", +<<<<<<< HEAD "version": "0.4.9", +======= + "version": "0.4.8", +>>>>>>> 23f916f (feat: add otel traceability) "license": "ISC", "dependencies": { "@radix-ui/react-alert-dialog": "^1.1.15", diff --git a/scripts/dev/monitoring/README.md b/scripts/dev/monitoring/README.md new file mode 100644 index 00000000..2ce54d9b --- /dev/null +++ b/scripts/dev/monitoring/README.md @@ -0,0 +1,84 @@ +# Hindsight Monitoring Stack + +Docker-based monitoring stack using **Grafana LGTM** (Loki, Grafana, Tempo, Mimir) for complete observability. + +## Quick Start + +```bash +# Start the monitoring stack +./scripts/dev/start-monitoring.sh + +# Or manually with docker-compose +cd scripts/dev/monitoring && docker-compose up -d +``` + +## Access + +- **Grafana UI**: http://localhost:3000 + - No login required (anonymous admin enabled for dev) + +## Features + +- **Traces**: OpenTelemetry traces with GenAI semantic conventions (Tempo) +- **Metrics**: Prometheus scraping of Hindsight API `/metrics` endpoint +- **Logs**: Loki log aggregation (future) +- **Dashboards**: Pre-configured dashboards from `monitoring/grafana/dashboards/`: + - Hindsight Operations + - Hindsight LLM Metrics + - Hindsight API Service + +## Configure Hindsight API + +Set these environment variables in your `.env`: + +```bash +# Enable tracing +HINDSIGHT_API_OTEL_TRACES_ENABLED=true + +# Grafana Tempo OTLP endpoint (HTTP) +HINDSIGHT_API_OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318 + +# Optional: Custom service name +HINDSIGHT_API_OTEL_SERVICE_NAME=hindsight-api + +# Optional: Deployment environment +HINDSIGHT_API_OTEL_DEPLOYMENT_ENVIRONMENT=development +``` + +## View Data + +### Traces +1. Open http://localhost:3000 +2. Go to **Explore** (compass icon) +3. Select **Tempo** as data source +4. Click "Search" to see recent traces + +### Metrics & Dashboards +1. Open http://localhost:3000 +2. Go to **Dashboards** (dashboard icon) +3. Browse the Hindsight folder + +### Raw Metrics +- Prometheus metrics: http://localhost:8888/metrics +- PromQL queries: Explore → Prometheus + +## Ports + +| Port | Service | +|------|---------| +| 3000 | Grafana UI | +| 4317 | OTLP gRPC endpoint | +| 4318 | OTLP HTTP endpoint | + +## Stop + +```bash +cd scripts/dev/monitoring && docker-compose down +``` + +## Architecture + +- **Single Container**: Grafana LGTM (~515MB) provides all observability components +- **Auto-provisioned Dashboards**: Dashboards from `monitoring/grafana/dashboards/` are automatically loaded +- **Prometheus Scraping**: Configured to scrape Hindsight API at `host.docker.internal:8888/metrics` every 5 seconds +- **Network**: Uses `hindsight-network` (shared with API for future service-to-service tracing) diff --git a/scripts/dev/monitoring/docker-compose.yaml b/scripts/dev/monitoring/docker-compose.yaml new file mode 100644 index 00000000..49ed0a41 --- /dev/null +++ b/scripts/dev/monitoring/docker-compose.yaml @@ -0,0 +1,35 @@ +services: + grafana-lgtm: + image: grafana/otel-lgtm:latest + container_name: hindsight-monitoring + ports: + # Grafana UI + - "3000:3000" + # OTLP gRPC (traces) + - "4317:4317" + # OTLP HTTP (traces) + - "4318:4318" + environment: + - GF_AUTH_ANONYMOUS_ENABLED=true + - GF_AUTH_ANONYMOUS_ORG_ROLE=Admin + - GF_AUTH_DISABLE_LOGIN_FORM=true + volumes: + # Mount Prometheus config for scraping Hindsight API metrics (override LGTM default) + - ./prometheus.yml:/otel-lgtm/prometheus.yaml:ro + # Mount Hindsight dashboards to LGTM directory (where default dashboards are) + - ../../../monitoring/grafana/dashboards/hindsight-operations.json:/otel-lgtm/hindsight-operations.json:ro + - ../../../monitoring/grafana/dashboards/hindsight-llm.json:/otel-lgtm/hindsight-llm.json:ro + - ../../../monitoring/grafana/dashboards/hindsight-api-service.json:/otel-lgtm/hindsight-api-service.json:ro + # Mount custom dashboard provisioning config that includes Hindsight dashboards + - ./grafana-dashboards.yaml:/otel-lgtm/grafana/conf/provisioning/dashboards/grafana-dashboards.yaml:ro + restart: unless-stopped + networks: + - hindsight + extra_hosts: + # Allow container to reach host services (Hindsight API on localhost:8888) + - "host.docker.internal:host-gateway" + +networks: + hindsight: + name: hindsight-network + external: true diff --git a/scripts/dev/monitoring/grafana-dashboards.yaml b/scripts/dev/monitoring/grafana-dashboards.yaml new file mode 100644 index 00000000..9761ba59 --- /dev/null +++ b/scripts/dev/monitoring/grafana-dashboards.yaml @@ -0,0 +1,36 @@ +apiVersion: 1 + +providers: + # Default LGTM dashboards + - name: "RED Metrics (classic histogram)" + type: file + options: + path: /otel-lgtm/grafana-dashboard-red-metrics-classic.json + foldersFromFilesStructure: false + - name: "RED Metrics (exponential/native histogram)" + type: file + options: + path: /otel-lgtm/grafana-dashboard-red-metrics-native.json + foldersFromFilesStructure: false + - name: "JVM Metrics" + type: file + options: + path: /otel-lgtm/grafana-dashboard-jvm-metrics.json + foldersFromFilesStructure: false + + # Hindsight dashboards + - name: "Hindsight Operations" + type: file + options: + path: /otel-lgtm/hindsight-operations.json + foldersFromFilesStructure: false + - name: "Hindsight LLM" + type: file + options: + path: /otel-lgtm/hindsight-llm.json + foldersFromFilesStructure: false + - name: "Hindsight API Service" + type: file + options: + path: /otel-lgtm/hindsight-api-service.json + foldersFromFilesStructure: false diff --git a/scripts/dev/monitoring/prometheus.yml b/scripts/dev/monitoring/prometheus.yml new file mode 100644 index 00000000..7dedc382 --- /dev/null +++ b/scripts/dev/monitoring/prometheus.yml @@ -0,0 +1,30 @@ +# Prometheus configuration for Grafana LGTM with Hindsight API scraping +--- +global: + scrape_interval: 5s + evaluation_interval: 5s + scrape_native_histograms: true + +# OTLP receiver configuration (from LGTM default) +otlp: + keep_identifying_resource_attributes: true + promote_resource_attributes: + - service.instance.id + - service.name + - service.namespace + - service.version + - deployment.environment + - deployment.environment.name + - host.name + +storage: + tsdb: + out_of_order_time_window: 10m + +# Scrape configs for pulling metrics from Hindsight API +scrape_configs: + - job_name: 'hindsight-api' + static_configs: + - targets: ['host.docker.internal:8888'] + metrics_path: '/metrics' + scrape_interval: 5s diff --git a/scripts/dev/monitoring/start.sh b/scripts/dev/monitoring/start.sh index b0044abc..fc8b7978 100755 --- a/scripts/dev/monitoring/start.sh +++ b/scripts/dev/monitoring/start.sh @@ -1,222 +1,54 @@ #!/bin/bash set -e -# Script to start Prometheus and Grafana for Hindsight metrics -# This provides a single command for the full monitoring stack +# Script to start the Hindsight monitoring stack with Grafana LGTM +# Provides traces (Tempo), metrics (Prometheus/Mimir), logs (Loki), and dashboards (Grafana) SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" PROJECT_ROOT="$(cd "$SCRIPT_DIR/../../.." && pwd)" -MONITORING_DATA_DIR="$PROJECT_ROOT/.monitoring" API_PORT="${API_PORT:-8888}" -PROMETHEUS_PORT="${PROMETHEUS_PORT:-8889}" -GRAFANA_PORT="${GRAFANA_PORT:-8890}" -# Versions -PROMETHEUS_VERSION="2.48.0" -GRAFANA_VERSION="10.2.2" - -# Detect OS and architecture -OS=$(uname -s | tr '[:upper:]' '[:lower:]') -ARCH=$(uname -m) - -case "$OS" in - darwin) OS_NAME="darwin" ;; - linux) OS_NAME="linux" ;; - *) echo "Unsupported OS: $OS"; exit 1 ;; -esac - -case "$ARCH" in - x86_64) ARCH_NAME="amd64" ;; - arm64|aarch64) ARCH_NAME="arm64" ;; - *) echo "Unsupported architecture: $ARCH"; exit 1 ;; -esac - -# Prometheus paths -PROMETHEUS_DIR="$MONITORING_DATA_DIR/prometheus" -PROMETHEUS_ARCHIVE="prometheus-${PROMETHEUS_VERSION}.${OS_NAME}-${ARCH_NAME}.tar.gz" -PROMETHEUS_URL="https://github.com/prometheus/prometheus/releases/download/v${PROMETHEUS_VERSION}/${PROMETHEUS_ARCHIVE}" -PROMETHEUS_BIN="$PROMETHEUS_DIR/prometheus-${PROMETHEUS_VERSION}.${OS_NAME}-${ARCH_NAME}/prometheus" - -# Grafana paths -GRAFANA_DIR="$MONITORING_DATA_DIR/grafana" -GRAFANA_ARCHIVE="grafana-${GRAFANA_VERSION}.${OS_NAME}-${ARCH_NAME}.tar.gz" -GRAFANA_URL="https://dl.grafana.com/oss/release/${GRAFANA_ARCHIVE}" -GRAFANA_HOME="$GRAFANA_DIR/grafana-v${GRAFANA_VERSION}" -GRAFANA_BIN="$GRAFANA_HOME/bin/grafana" - -# Cleanup function -cleanup() { - echo "" - echo "Shutting down monitoring stack..." - - if [ -n "$PROM_PID" ] && kill -0 "$PROM_PID" 2>/dev/null; then - kill "$PROM_PID" 2>/dev/null || true - fi - - if [ -n "$GRAFANA_PID" ] && kill -0 "$GRAFANA_PID" 2>/dev/null; then - kill "$GRAFANA_PID" 2>/dev/null || true - fi - - echo "Monitoring stack stopped" - exit 0 -} - -trap cleanup SIGINT SIGTERM - -# Download Prometheus if needed -if [ ! -f "$PROMETHEUS_BIN" ]; then - echo "Downloading Prometheus ${PROMETHEUS_VERSION}..." - mkdir -p "$PROMETHEUS_DIR" - cd "$PROMETHEUS_DIR" - curl -sL -o "$PROMETHEUS_ARCHIVE" "$PROMETHEUS_URL" - tar xzf "$PROMETHEUS_ARCHIVE" - rm "$PROMETHEUS_ARCHIVE" - echo "Prometheus ready" -fi - -# Download Grafana if needed -if [ ! -f "$GRAFANA_BIN" ]; then - echo "Downloading Grafana ${GRAFANA_VERSION}..." - mkdir -p "$GRAFANA_DIR" - cd "$GRAFANA_DIR" - curl -sL -o "$GRAFANA_ARCHIVE" "$GRAFANA_URL" - tar xzf "$GRAFANA_ARCHIVE" - rm "$GRAFANA_ARCHIVE" - echo "Grafana ready" -fi - -# Create Prometheus config -mkdir -p "$PROMETHEUS_DIR" -cat > "$PROMETHEUS_DIR/prometheus.yml" < "$GRAFANA_PROV_DIR/datasources/prometheus.yaml" < "$GRAFANA_PROV_DIR/dashboards/dashboards.yaml" < "$GRAFANA_DIR/grafana.ini" < /dev/null 2>&1; then - echo "WARNING: Hindsight API not detected at localhost:$API_PORT" - echo " Start the API first: ./scripts/dev/start-api.sh" + echo "⚠️ WARNING: Hindsight API not detected at localhost:$API_PORT" + echo " Start the API first: ./scripts/dev/start-api.sh" echo "" fi -# Start Prometheus in background -cd "$(dirname "$PROMETHEUS_BIN")" -"$PROMETHEUS_BIN" \ - --config.file="$PROMETHEUS_DIR/prometheus.yml" \ - --storage.tsdb.path="$PROMETHEUS_DIR/data" \ - --web.console.templates="$(dirname "$PROMETHEUS_BIN")/consoles" \ - --web.console.libraries="$(dirname "$PROMETHEUS_BIN")/console_libraries" \ - --web.listen-address="0.0.0.0:$PROMETHEUS_PORT" \ - --web.enable-lifecycle \ - --log.level=warn & -PROM_PID=$! - -# Start Grafana in background -cd "$GRAFANA_HOME" -"$GRAFANA_BIN" server \ - --homepath="$GRAFANA_HOME" \ - --config="$GRAFANA_DIR/grafana.ini" & -GRAFANA_PID=$! - -echo "Monitoring stack running. Press Ctrl+C to stop." +echo "Access Grafana UI: http://localhost:3000" +echo " (no login required for dev - anonymous admin enabled)" +echo "" +echo "Dashboards available:" +echo " • Hindsight Operations" +echo " • Hindsight LLM Metrics" +echo " • Hindsight API Service" +echo "" +echo "Configure Hindsight API for tracing:" +echo " export HINDSIGHT_API_OTEL_TRACES_ENABLED=true" +echo " export HINDSIGHT_API_OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318" +echo "" +echo "OTLP Endpoints:" +echo " • HTTP: http://localhost:4318" +echo " • gRPC: http://localhost:4317" +echo "" +echo "View:" +echo " • Traces: http://localhost:3000 → Explore → Tempo" +echo " • Metrics: http://localhost:3000 → Dashboards" +echo " • Raw Metrics: http://localhost:$API_PORT/metrics" +echo "" +echo "Press Ctrl+C to stop" echo "" -# Wait for processes -wait "$PROM_PID" "$GRAFANA_PID" 2>/dev/null || true - -# If we get here, clean up -cleanup +docker-compose up diff --git a/uv.lock b/uv.lock index ff084323..7e286d0c 100644 --- a/uv.lock +++ b/uv.lock @@ -1221,6 +1221,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/40/f2/97fefdd1ad1f3428321bac819ae7a83ccc59f6439616054736b7819fa56c/google_genai-1.53.0-py3-none-any.whl", hash = "sha256:65a3f99e5c03c372d872cda7419f5940e723374bb12a2f3ffd5e3e56e8eb2094", size = 262015 }, ] +[[package]] +name = "googleapis-common-protos" +version = "1.72.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "protobuf" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/e5/7b/adfd75544c415c487b33061fe7ae526165241c1ea133f9a9125a56b39fd8/googleapis_common_protos-1.72.0.tar.gz", hash = "sha256:e55a601c1b32b52d7a3e65f43563e2aa61bcd737998ee672ac9b951cd49319f5", size = 147433 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/c4/ab/09169d5a4612a5f92490806649ac8d41e3ec9129c636754575b3553f4ea4/googleapis_common_protos-1.72.0-py3-none-any.whl", hash = "sha256:4299c5a82d5ae1a9702ada957347726b167f9f8d1fc352477702a1e851ff4038", size = 297515 }, +] + [[package]] name = "greenlet" version = "3.2.4" @@ -1360,9 +1372,11 @@ dependencies = [ { name = "langchain-text-splitters" }, { name = "openai" }, { name = "opentelemetry-api" }, + { name = "opentelemetry-exporter-otlp-proto-http" }, { name = "opentelemetry-exporter-prometheus" }, { name = "opentelemetry-instrumentation-fastapi" }, { name = "opentelemetry-sdk" }, + { name = "opentelemetry-semantic-conventions" }, { name = "pg0-embedded" }, { name = "pgvector" }, { name = "psycopg2-binary" }, @@ -1428,9 +1442,11 @@ requires-dist = [ { name = "langchain-text-splitters", specifier = ">=0.3.0" }, { name = "openai", specifier = ">=1.0.0" }, { name = "opentelemetry-api", specifier = ">=1.20.0" }, + { name = "opentelemetry-exporter-otlp-proto-http", specifier = ">=1.20.0" }, { name = "opentelemetry-exporter-prometheus", specifier = ">=0.41b0" }, { name = "opentelemetry-instrumentation-fastapi", specifier = ">=0.41b0" }, { name = "opentelemetry-sdk", specifier = ">=1.20.0" }, + { name = "opentelemetry-semantic-conventions", specifier = ">=0.41b0" }, { name = "pg0-embedded", specifier = ">=0.11.0" }, { name = "pgvector", specifier = ">=0.4.1" }, { name = "psycopg2-binary", specifier = ">=2.9.11" }, @@ -2650,6 +2666,36 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/cf/df/d3f1ddf4bb4cb50ed9b1139cc7b1c54c34a1e7ce8fd1b9a37c0d1551a6bd/opentelemetry_api-1.39.1-py3-none-any.whl", hash = "sha256:2edd8463432a7f8443edce90972169b195e7d6a05500cd29e6d13898187c9950", size = 66356 }, ] +[[package]] +name = "opentelemetry-exporter-otlp-proto-common" +version = "1.39.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "opentelemetry-proto" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/e9/9d/22d241b66f7bbde88a3bfa6847a351d2c46b84de23e71222c6aae25c7050/opentelemetry_exporter_otlp_proto_common-1.39.1.tar.gz", hash = "sha256:763370d4737a59741c89a67b50f9e39271639ee4afc999dadfe768541c027464", size = 20409 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/8c/02/ffc3e143d89a27ac21fd557365b98bd0653b98de8a101151d5805b5d4c33/opentelemetry_exporter_otlp_proto_common-1.39.1-py3-none-any.whl", hash = "sha256:08f8a5862d64cc3435105686d0216c1365dc5701f86844a8cd56597d0c764fde", size = 18366 }, +] + +[[package]] +name = "opentelemetry-exporter-otlp-proto-http" +version = "1.39.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "googleapis-common-protos" }, + { name = "opentelemetry-api" }, + { name = "opentelemetry-exporter-otlp-proto-common" }, + { name = "opentelemetry-proto" }, + { name = "opentelemetry-sdk" }, + { name = "requests" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/80/04/2a08fa9c0214ae38880df01e8bfae12b067ec0793446578575e5080d6545/opentelemetry_exporter_otlp_proto_http-1.39.1.tar.gz", hash = "sha256:31bdab9745c709ce90a49a0624c2bd445d31a28ba34275951a6a362d16a0b9cb", size = 17288 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/95/f1/b27d3e2e003cd9a3592c43d099d2ed8d0a947c15281bf8463a256db0b46c/opentelemetry_exporter_otlp_proto_http-1.39.1-py3-none-any.whl", hash = "sha256:d9f5207183dd752a412c4cd564ca8875ececba13be6e9c6c370ffb752fd59985", size = 19641 }, +] + [[package]] name = "opentelemetry-exporter-prometheus" version = "0.60b1" @@ -2711,6 +2757,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/7d/cc/6e808328ba54662e50babdcab21138eae4250bc0fddf67d55526a615a2ca/opentelemetry_instrumentation_fastapi-0.60b1-py3-none-any.whl", hash = "sha256:af94b7a239ad1085fc3a820ecf069f67f579d7faf4c085aaa7bd9b64eafc8eaf", size = 13478 }, ] +[[package]] +name = "opentelemetry-proto" +version = "1.39.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "protobuf" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/49/1d/f25d76d8260c156c40c97c9ed4511ec0f9ce353f8108ca6e7561f82a06b2/opentelemetry_proto-1.39.1.tar.gz", hash = "sha256:6c8e05144fc0d3ed4d22c2289c6b126e03bcd0e6a7da0f16cedd2e1c2772e2c8", size = 46152 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/51/95/b40c96a7b5203005a0b03d8ce8cd212ff23f1793d5ba289c87a097571b18/opentelemetry_proto-1.39.1-py3-none-any.whl", hash = "sha256:22cdc78efd3b3765d09e68bfbd010d4fc254c9818afd0b6b423387d9dee46007", size = 72535 }, +] + [[package]] name = "opentelemetry-sdk" version = "1.39.1"