feat: update xapi event to send the object in right way with location and send usage

Author: Henrrypg

backend/openedx_ai_extensions/processors/llm/educator_assistant_processor.py

@@ -58,7 +58,7 @@ def _call_completion_api(self, system_role):
 
             return {
                 "response": content,
-                "tokens_used": response.usage.total_tokens if response.usage else 0,
+                "usage": response.usage if response.usage else None,
                 "model_used": self.extra_params.get("model", "unknown"),
                 "status": "success",
             }
@@ -93,12 +93,11 @@ def generate_quiz_questions(self, input_data):
             logger.exception(f"Error calling LiteLLM: {e}")
             return {"error": f"AI processing failed: {str(e)}"}
 
-        tokens_used = result.get("tokens_used", 0)
         response = json.loads(result['response'])
 
         return {
             "response": response,
-            "tokens_used": tokens_used,
+            "usage": result.get("usage", None),
             "model_used": self.extra_params.get("model", "unknown"),
             "status": "success",
         }
@@ -121,19 +120,17 @@ def refine_quiz_question(self, input_data):
         for key, value in input_data.items():
             placeholder = f"{{{{{key.upper()}}}}}"
             prompt = prompt.replace(placeholder, str(value))
-        logger.info(f"Refinement prompt after placeholder replacement: {prompt}")
 
         try:
             result = self._call_completion_api(prompt)
         except Exception as e:  # pylint: disable=broad-exception-caught
             logger.exception(f"Error calling LiteLLM during refinement: {e}")
             return {"error": f"AI processing failed: {str(e)}"}
 
-        tokens_used = result.get("tokens_used", 0)
         response = json.loads(result['response'])
         return {
             "response": response,
-            "tokens_used": tokens_used,
+            "usage": result.get("usage", None),
             "model_used": self.extra_params.get("model", "unknown"),
             "status": "success",
         }

backend/openedx_ai_extensions/processors/llm/llm_processor.py

@@ -83,12 +83,10 @@ def _handle_streaming_completion(self, response):
     def _handle_non_streaming_completion(self, response):
         """Handles the non-streaming logic, returning a response dict."""
         content = response.choices[0].message.content
-        total_tokens = response.usage.total_tokens if response.usage else 0
-        logger.info(f"[LLM NON-STREAM] Tokens used: {total_tokens}")
 
         return {
             "response": content,
-            "tokens_used": total_tokens,
+            "usage": response.usage if response.usage else None,
             "model_used": self.provider,
             "status": "success",
         }
@@ -181,12 +179,10 @@ def _call_responses_wrapper(self, params, initialize=False):
             if response_id:
                 self.user_session.remote_response_id = response_id
                 self.user_session.save()
-            total_tokens = response.usage.total_tokens if response.usage else 0
-            logger.info(f"[LLM NON-STREAM] Tokens used: {total_tokens}")
 
             result = {
                 "response": content,
-                "tokens_used": total_tokens,
+                "usage": response.usage if response.usage else None,
                 "model_used": self.extra_params.get("model", "unknown"),
                 "status": "success",
             }
@@ -682,12 +678,11 @@ def generate_flashcards(self):
         if "error" in result:
             return result
 
-        tokens_used = result.get("tokens_used", 0)
         response = json.loads(result['response'])
 
         return {
             "response": response,
-            "tokens_used": tokens_used,
+            "usage": result.get("usage", None),
             "model_used": self.extra_params.get("model", "unknown"),
             "status": "success",
         }

backend/openedx_ai_extensions/workflows/orchestrators/base_orchestrator.py

@@ -19,21 +19,68 @@ def __init__(self, workflow, user, context):
         self.location_id = context.get("location_id", None)
         self.course_id = context.get("course_id", None)
 
-    def _emit_workflow_event(self, event_name: str) -> None:
+    def _convert_usage_to_json_serializable(self, usage) -> dict:
+        """
+        Convert usage data to a JSON-serializable format.
+
+        This is necessary because usage data may contain complex objects
+        (e.g. litellm Usage Pydantic models) that cannot be directly
+        serialized to JSON when included in xAPI event data. This method
+        first normalizes the input to a plain dict, then ensures every
+        value is JSON-serializable.
+
+        Args:
+            usage: A dictionary or Pydantic model containing usage data.
+
+        Returns:
+            A new dictionary with all values converted to JSON-serializable formats.
+        """
+        if isinstance(usage, dict):
+            usage_dict = usage
+        elif hasattr(usage, "model_dump"):
+            # Pydantic v2 models (e.g. litellm Usage)
+            usage_dict = usage.model_dump()
+        elif hasattr(usage, "dict"):
+            # Pydantic v1 models
+            usage_dict = usage.dict()
+        else:
+            usage_dict = vars(usage)
+
+        serializable_usage = {}
+        for key, value in usage_dict.items():
+            if isinstance(value, (str, int, float, bool)) or value is None:
+                serializable_usage[key] = value
+            else:
+                # For non-serializable types, convert to string representation
+                serializable_usage[key] = str(value)
+        return serializable_usage
+
+    def _emit_workflow_event(self, event_name: str, usage: dict = None) -> None:
         """
         Emit an xAPI event for this workflow.
 
         Args:
             event_name: The event name constant (e.g., EVENT_NAME_WORKFLOW_COMPLETED)
         """
-
-        tracker.emit(event_name, {
+        event_data = {
             "workflow_id": str(self.workflow.id),
             "action": self.workflow.action,
-            "course_id": str(self.course_id),
+            "course_id": str(self.course_id) if self.course_id else "",
             "profile_name": self.profile.slug,
-            "location_id": str(self.location_id),
-        })
+            "location_id": str(self.location_id) if self.location_id else "",
+        }
+        if usage:
+            event_data["usage"] = self._convert_usage_to_json_serializable(usage)
+
+        tracking_context = {}
+        if self.course_id:
+            tracking_context["course_id"] = str(self.course_id)
+
+        if tracking_context:
+            with tracker.get_tracker().context("ai_workflow", tracking_context):
+                tracker.emit(event_name, event_data)
+        else:
+            tracker.emit(event_name, event_data)
 
     def run(self, input_data):
         raise NotImplementedError("Subclasses must implement run method")

backend/openedx_ai_extensions/workflows/orchestrators/direct_orchestrator.py

@@ -70,17 +70,13 @@ def run(self, input_data):
             }
 
         # 6. Emit completed event for one-shot workflow
-        self._emit_workflow_event(EVENT_NAME_WORKFLOW_COMPLETED)
+        self._emit_workflow_event(EVENT_NAME_WORKFLOW_COMPLETED, usage=llm_result.get('usage', None))
 
         # --- 7. Return Structured Non-Streaming Result ---
         # If execution reaches this point, we have a successful, non-streaming result (Dict).
         response_data = {
             'response': llm_result.get('response', 'No response available'),
             'status': 'completed',
-            'metadata': {
-                'tokens_used': llm_result.get('tokens_used'),
-                'model_used': llm_result.get('model_used')
-            }
         }
         return response_data
 
@@ -199,14 +195,10 @@ def run(self, input_data):
                 "collection_id": collection_key,
             }
             self.session.save(update_fields=["metadata"])
-            self._emit_workflow_event(EVENT_NAME_WORKFLOW_COMPLETED)
+            self._emit_workflow_event(EVENT_NAME_WORKFLOW_COMPLETED, usage=llm_result.get('usage', None))
             return {
                 'response': f"authoring/library/{lib_key_str}/collection/{collection_key}",
-                'status': 'completed',
-                'metadata': {
-                    'tokens_used': llm_result.get('tokens_used'),
-                    'model_used': llm_result.get('model_used')
-                }
+                'status': 'completed'
             }
 
         # Iterative path: store questions for review

backend/openedx_ai_extensions/workflows/orchestrators/flashcards_orchestrator.py

@@ -107,7 +107,7 @@ def run(self, input_data):
                 'status': 'LLMProcessor error'
             }
 
-        self._emit_workflow_event(EVENT_NAME_WORKFLOW_COMPLETED)
+        self._emit_workflow_event(EVENT_NAME_WORKFLOW_COMPLETED, usage=llm_result.get('usage', None))
 
         response_obj = llm_result.get('response')
         cards = self._get_structured_cards(response_obj)
@@ -122,10 +122,6 @@ def run(self, input_data):
         response_data = {
             'response': cards,
             'status': 'completed',
-            'metadata': {
-                'tokens_used': llm_result.get('tokens_used'),
-                'model_used': llm_result.get('model_used')
-            }
         }
         return response_data
 

backend/openedx_ai_extensions/workflows/orchestrators/threaded_orchestrator.py

@@ -184,16 +184,12 @@ def run(self, input_data):
 
         # Emit appropriate event based on interaction state
         if is_first_interaction:
-            self._emit_workflow_event(EVENT_NAME_WORKFLOW_INITIALIZED)
+            self._emit_workflow_event(EVENT_NAME_WORKFLOW_INITIALIZED, usage=llm_result.get("usage", None))
         else:
-            self._emit_workflow_event(EVENT_NAME_WORKFLOW_INTERACTED)
+            self._emit_workflow_event(EVENT_NAME_WORKFLOW_INTERACTED, usage=llm_result.get("usage", None))
 
         # 4. Return result
         return {
             "response": llm_result.get("response", "No response available"),
             "status": "completed",
-            "metadata": {
-                "tokens_used": llm_result.get("tokens_used"),
-                "model_used": llm_result.get("model_used"),
-            },
         }

backend/openedx_ai_extensions/xapi/constants.py

@@ -22,6 +22,7 @@
 XAPI_EXTENSION_WORKFLOW_ACTION = "https://w3id.org/xapi/openedx/extension/ai-workflow-action"
 XAPI_EXTENSION_PROMPT_TEMPLATE_SLUG = "https://w3id.org/xapi/openedx/extension/prompt-template-slug"
 XAPI_EXTENSION_LOCATION_ID = "https://w3id.org/xapi/openedx/extension/location-id"
+XAPI_EXTENSION_USAGE = "https://w3id.org/xapi/openedx/extension/ai-usage"
 
 # Event names
 EVENT_NAME_WORKFLOW_INITIALIZED = "openedx.ai.workflow.initialized"

backend/openedx_ai_extensions/xapi/transformers.py

@@ -53,6 +53,11 @@ def get_object(self) -> Activity:
         if location_id:
             extensions[constants.XAPI_EXTENSION_LOCATION_ID] = location_id
 
+        # Add LLM usage data (token counts, etc.) if available
+        usage: Optional[dict] = self.get_data("usage")
+        if usage:
+            extensions[constants.XAPI_EXTENSION_USAGE] = usage
+
         return Activity(
             id=self.get_object_iri("ai_workflow", "__".join([profile_name, action])),
             definition=ActivityDefinition(