feat(crewai): Add GenAI memory operation tracing for unified memory system

packages/opentelemetry-instrumentation-crewai/opentelemetry/instrumentation/crewai/instrumentation.py

@@ -10,13 +10,39 @@
 from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
 from opentelemetry.instrumentation.crewai.version import __version__
 from opentelemetry.semconv._incubating.attributes import (
+    error_attributes as ErrorAttributes,
     gen_ai_attributes as GenAIAttributes,
 )
 from opentelemetry.semconv_ai import SpanAttributes, TraceloopSpanKindValues, Meters
 from .crewai_span_attributes import CrewAISpanAttributes, set_span_attribute
 
 _instruments = ("crewai >= 0.70.0",)
 
+# GenAI memory semantic convention attribute keys (fallback to string
+# literals when the installed semconv package doesn't define them yet).
+_GEN_AI_OPERATION_NAME = getattr(GenAIAttributes, "GEN_AI_OPERATION_NAME", "gen_ai.operation.name")
+_GEN_AI_PROVIDER_NAME = getattr(GenAIAttributes, "GEN_AI_PROVIDER_NAME", "gen_ai.provider.name")
+_GEN_AI_MEMORY_SCOPE = getattr(GenAIAttributes, "GEN_AI_MEMORY_SCOPE", "gen_ai.memory.scope")
+_GEN_AI_MEMORY_TYPE = getattr(GenAIAttributes, "GEN_AI_MEMORY_TYPE", "gen_ai.memory.type")
+_GEN_AI_MEMORY_QUERY = getattr(GenAIAttributes, "GEN_AI_MEMORY_QUERY", "gen_ai.memory.query")
+_GEN_AI_MEMORY_CONTENT = getattr(GenAIAttributes, "GEN_AI_MEMORY_CONTENT", "gen_ai.memory.content")
+_GEN_AI_MEMORY_NAMESPACE = getattr(GenAIAttributes, "GEN_AI_MEMORY_NAMESPACE", "gen_ai.memory.namespace")
+_GEN_AI_MEMORY_SEARCH_RESULT_COUNT = getattr(
+    GenAIAttributes, "GEN_AI_MEMORY_SEARCH_RESULT_COUNT", "gen_ai.memory.search.result.count"
+)
+_GEN_AI_MEMORY_UPDATE_STRATEGY = getattr(
+    GenAIAttributes, "GEN_AI_MEMORY_UPDATE_STRATEGY", "gen_ai.memory.update.strategy"
+)
+_GEN_AI_MEMORY_IMPORTANCE = getattr(GenAIAttributes, "GEN_AI_MEMORY_IMPORTANCE", "gen_ai.memory.importance")
+_ERROR_TYPE = getattr(ErrorAttributes, "ERROR_TYPE", "error.type")
+
+_PROVIDER = "crewai"
+
+
+def _capture_content() -> bool:
+    """Check if memory content capture is enabled."""
+    return os.environ.get("OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT", "").lower() in ("true", "1")
+
 
 class CrewAIInstrumentor(BaseInstrumentor):
 
@@ -44,12 +70,40 @@ def _instrument(self, **kwargs):
         wrap_function_wrapper("crewai.llm", "LLM.call",
                               wrap_llm_call(tracer, duration_histogram, token_histogram))
 
+        # Memory operations (crewai.memory.unified_memory.Memory)
+        try:
+            wrap_function_wrapper(
+                "crewai.memory.unified_memory", "Memory.remember",
+                wrap_memory_remember(tracer, duration_histogram))
+            wrap_function_wrapper(
+                "crewai.memory.unified_memory", "Memory.recall",
+                wrap_memory_recall(tracer, duration_histogram))
+            wrap_function_wrapper(
+                "crewai.memory.unified_memory", "Memory.forget",
+                wrap_memory_forget(tracer, duration_histogram))
+            wrap_function_wrapper(
+                "crewai.memory.unified_memory", "Memory.reset",
+                wrap_memory_reset(tracer, duration_histogram))
+        except Exception:
+            # CrewAI versions before unified_memory may not have these classes
+            pass
+
     def _uninstrument(self, **kwargs):
         unwrap("crewai.crew.Crew", "kickoff")
         unwrap("crewai.agent.Agent", "execute_task")
         unwrap("crewai.task.Task", "execute_sync")
         unwrap("crewai.llm.LLM", "call")
 
+        # Memory unwrap (ignore if not wrapped)
+        try:
+            from crewai.memory.unified_memory import Memory as UnifiedMemory
+            unwrap(UnifiedMemory, "remember")
+            unwrap(UnifiedMemory, "recall")
+            unwrap(UnifiedMemory, "forget")
+            unwrap(UnifiedMemory, "reset")
+        except Exception:
+            pass
+
 
 def with_tracer_wrapper(func):
     """Helper for providing tracer for wrapper functions."""
@@ -199,3 +253,229 @@ def _create_metrics(meter: Meter):
     )
 
     return token_histogram, duration_histogram
+
+
+# ---------------------------------------------------------------------------
+# Memory operation wrappers — aligned with GenAI memory semantic conventions
+# ---------------------------------------------------------------------------
+
+
+def _infer_memory_scope(instance) -> str:
+    """Infer memory scope from the Memory instance or its MemoryScope wrapper."""
+    # MemoryScope has a _root attribute like "/agent/1" or "/user/123"
+    root = getattr(instance, "_root", None)
+    if root:
+        parts = root.strip("/").split("/")
+        if parts:
+            first = parts[0].lower()
+            if first in ("user", "agent", "session", "team", "global"):
+                return first
+    return "agent"
+
+
+def _infer_memory_type(kwargs) -> str:
+    """Infer memory type from kwargs categories hint, defaulting to long_term."""
+    categories = kwargs.get("categories")
+    if categories and isinstance(categories, list):
+        for cat in categories:
+            cl = str(cat).lower()
+            if "short" in cl:
+                return "short_term"
+            if "entity" in cl:
+                return "entity"
+    return "long_term"
+
+
+def _set_memory_error(span, exc):
+    """Record error details on the span."""
+    error_type = type(exc).__qualname__
+    span.set_status(Status(StatusCode.ERROR, str(exc)))
+    set_span_attribute(span, _ERROR_TYPE, error_type)
+    return error_type
+
+
+def _record_memory_duration(duration_histogram, duration_s, operation, error_type=None):
+    """Record memory operation duration metric."""
+    if not duration_histogram:
+        return
+    attrs = {
+        _GEN_AI_OPERATION_NAME: operation,
+        GenAIAttributes.GEN_AI_SYSTEM: _PROVIDER,
+    }
+    if error_type:
+        attrs[_ERROR_TYPE] = error_type
+    duration_histogram.record(max(duration_s, 0.0), attributes=attrs)
+
+
+def wrap_memory_remember(tracer: Tracer, duration_histogram: Histogram):
+    """Wrap Memory.remember() → update_memory span."""
+    def _wrapper(wrapped, instance, args, kwargs):
+        operation = "update_memory"
+        span_name = f"{operation} {_PROVIDER}"
+        error_type = None
+        start_time = time.time()
+        with tracer.start_as_current_span(
+            span_name, kind=SpanKind.CLIENT,
+            attributes={GenAIAttributes.GEN_AI_SYSTEM: _PROVIDER}
+        ) as span:
+            set_span_attribute(span, _GEN_AI_OPERATION_NAME, operation)
+            set_span_attribute(span, _GEN_AI_PROVIDER_NAME, _PROVIDER)
+            set_span_attribute(span, _GEN_AI_MEMORY_SCOPE, _infer_memory_scope(instance))
+            set_span_attribute(span, _GEN_AI_MEMORY_TYPE, _infer_memory_type(kwargs))
+            set_span_attribute(span, _GEN_AI_MEMORY_UPDATE_STRATEGY, "merge")
+
+            # Namespace from source kwarg
+            source = kwargs.get("source")
+            if source:
+                set_span_attribute(span, _GEN_AI_MEMORY_NAMESPACE, str(source))
+
+            # Scope path
+            scope = kwargs.get("scope")
+            if scope:
+                set_span_attribute(span, "crewai.memory.scope_path", str(scope))
+
+            importance = kwargs.get("importance")
+            if importance is not None:
+                set_span_attribute(span, _GEN_AI_MEMORY_IMPORTANCE, float(importance))
+
+            # Content (opt-in)
+            if _capture_content() and args:
+                content = args[0] if args else kwargs.get("content")
+                if content and isinstance(content, str):
+                    set_span_attribute(span, _GEN_AI_MEMORY_CONTENT, content)
+
+            try:
+                result = wrapped(*args, **kwargs)
+                span.set_status(Status(StatusCode.OK))
+                # MemoryRecord has an id attribute
+                if result and hasattr(result, "id"):
+                    set_span_attribute(span, "gen_ai.memory.id", str(result.id))
+                return result
+            except Exception as ex:
+                error_type = _set_memory_error(span, ex)
+                raise
+            finally:
+                _record_memory_duration(
+                    duration_histogram, time.time() - start_time, operation, error_type
+                )
+    return _wrapper
+
+
+def wrap_memory_recall(tracer: Tracer, duration_histogram: Histogram):
+    """Wrap Memory.recall() → search_memory span."""
+    def _wrapper(wrapped, instance, args, kwargs):
+        operation = "search_memory"
+        span_name = f"{operation} {_PROVIDER}"
+        error_type = None
+        start_time = time.time()
+        with tracer.start_as_current_span(
+            span_name, kind=SpanKind.CLIENT,
+            attributes={GenAIAttributes.GEN_AI_SYSTEM: _PROVIDER}
+        ) as span:
+            set_span_attribute(span, _GEN_AI_OPERATION_NAME, operation)
+            set_span_attribute(span, _GEN_AI_PROVIDER_NAME, _PROVIDER)
+            set_span_attribute(span, _GEN_AI_MEMORY_SCOPE, _infer_memory_scope(instance))
+            set_span_attribute(span, _GEN_AI_MEMORY_TYPE, _infer_memory_type(kwargs))
+
+            # Query (opt-in)
+            query = args[0] if args else kwargs.get("query")
+            if _capture_content() and query and isinstance(query, str):
+                set_span_attribute(span, _GEN_AI_MEMORY_QUERY, query)
+
+            # Scope path
+            scope = kwargs.get("scope")
+            if scope:
+                set_span_attribute(span, "crewai.memory.scope_path", str(scope))
+
+            source = kwargs.get("source")
+            if source:
+                set_span_attribute(span, _GEN_AI_MEMORY_NAMESPACE, str(source))
+
+            try:
+                result = wrapped(*args, **kwargs)
+                span.set_status(Status(StatusCode.OK))
+                if isinstance(result, list):
+                    set_span_attribute(span, _GEN_AI_MEMORY_SEARCH_RESULT_COUNT, len(result))
+                return result
+            except Exception as ex:
+                error_type = _set_memory_error(span, ex)
+                raise
+            finally:
+                _record_memory_duration(
+                    duration_histogram, time.time() - start_time, operation, error_type
+                )
+    return _wrapper
+
+
+def wrap_memory_forget(tracer: Tracer, duration_histogram: Histogram):
+    """Wrap Memory.forget() → delete_memory span."""
+    def _wrapper(wrapped, instance, args, kwargs):
+        operation = "delete_memory"
+        span_name = f"{operation} {_PROVIDER}"
+        error_type = None
+        start_time = time.time()
+        with tracer.start_as_current_span(
+            span_name, kind=SpanKind.CLIENT,
+            attributes={GenAIAttributes.GEN_AI_SYSTEM: _PROVIDER}
+        ) as span:
+            set_span_attribute(span, _GEN_AI_OPERATION_NAME, operation)
+            set_span_attribute(span, _GEN_AI_PROVIDER_NAME, _PROVIDER)
+            set_span_attribute(span, _GEN_AI_MEMORY_SCOPE, _infer_memory_scope(instance))
+
+            scope = kwargs.get("scope")
+            if scope:
+                set_span_attribute(span, "crewai.memory.scope_path", str(scope))
+
+            record_ids = kwargs.get("record_ids")
+            if record_ids and isinstance(record_ids, list) and len(record_ids) == 1:
+                set_span_attribute(span, "gen_ai.memory.id", str(record_ids[0]))
+
+            try:
+                result = wrapped(*args, **kwargs)
+                span.set_status(Status(StatusCode.OK))
+                # forget() returns number of deleted records
+                if isinstance(result, int):
+                    set_span_attribute(span, "crewai.memory.deleted_count", result)
+                return result
+            except Exception as ex:
+                error_type = _set_memory_error(span, ex)
+                raise
+            finally:
+                _record_memory_duration(
+                    duration_histogram, time.time() - start_time, operation, error_type
+                )
+    return _wrapper
+
+
+def wrap_memory_reset(tracer: Tracer, duration_histogram: Histogram):
+    """Wrap Memory.reset() → delete_memory span (scope-level wipe)."""
+    def _wrapper(wrapped, instance, args, kwargs):
+        operation = "delete_memory"
+        span_name = f"{operation} {_PROVIDER}"
+        error_type = None
+        start_time = time.time()
+        with tracer.start_as_current_span(
+            span_name, kind=SpanKind.CLIENT,
+            attributes={GenAIAttributes.GEN_AI_SYSTEM: _PROVIDER}
+        ) as span:
+            set_span_attribute(span, _GEN_AI_OPERATION_NAME, operation)
+            set_span_attribute(span, _GEN_AI_PROVIDER_NAME, _PROVIDER)
+            set_span_attribute(span, _GEN_AI_MEMORY_SCOPE, _infer_memory_scope(instance))
+            set_span_attribute(span, "crewai.memory.reset", True)
+
+            scope = kwargs.get("scope")
+            if scope:
+                set_span_attribute(span, "crewai.memory.scope_path", str(scope))
+
+            try:
+                result = wrapped(*args, **kwargs)
+                span.set_status(Status(StatusCode.OK))
+                return result
+            except Exception as ex:
+                error_type = _set_memory_error(span, ex)
+                raise
+            finally:
+                _record_memory_duration(
+                    duration_histogram, time.time() - start_time, operation, error_type
+                )
+    return _wrapper