Upload 2 files

examples/callback/business/callbacks.py

@@ -0,0 +1,51 @@
+# coding: utf-8
+# Copyright (c) 2025 inclusionAI.
+
+"""
+Callback function registration module, used for centralized management and registration of all callback functions.
+"""
+
+from aworld.runners.callback.decorator import reg_callback, CallbackRegistry
+
+
+# Register a simple callback function
+@reg_callback("print_content")
+def simple_callback(content):
+    """Simple callback function that prints content and returns it
+    
+    Args:
+        content: Content to print
+        
+    Returns:
+        The input content
+    """
+    print(f"Callback function received content: {content}")
+    return content
+
+
+# You can register more callback functions here
+@reg_callback("uppercase_content")
+def uppercase_callback(content):
+    """Callback function that converts content to uppercase
+    
+    Args:
+        content: Content to process
+        
+    Returns:
+        Content converted to uppercase
+    """
+    if isinstance(content, str):
+        result = content.upper()
+        print(f"Callback function converted content to uppercase: {result}")
+        return result
+    return content
+
+
+# Provide a function to check all registered callback functions
+def list_all_callbacks():
+    """List all registered callback functions"""
+    callbacks = CallbackRegistry.list()
+    print("Registered callback functions:")
+    for key, func_name in callbacks.items():
+        print(f"  - {key}: {func_name}")
+    return callbacks 

examples/callback/business/video_callback.py

@@ -0,0 +1,91 @@
+import json
+import os
+import time
+
+import requests
+from aworld.core.common import Observation, ActionResult
+from typing_extensions import Any
+
+from aworld.runners.callback.decorator import reg_callback
+
+from aworld.logs.util import logger
+
+@reg_callback("gen_video_server__video_tasks")
+def gen_video(actionResult:ActionResult) -> ActionResult:
+    try:
+        if not actionResult or not actionResult.content:
+            return actionResult
+        content = json.loads(actionResult.content)
+        task_id = content.get("video_id")
+        if not task_id:
+            return actionResult
+        gen_video_item(task_id)
+
+        return actionResult
+
+    except Exception as e:
+        logger.warning(f"Exception gen_video occurred: {e}")
+        return actionResult
+
+def gen_video_item(task_id:str) -> Any:
+    if not task_id:
+        return None
+    try:
+        from dotenv import load_dotenv
+        load_dotenv()
+        api_key = os.getenv('DASHSCOPE_API_KEY')
+        query_base_url = os.getenv('DASHSCOPE_QUERY_BASE_URL', '')
+        # Step 2: Poll for results
+        max_attempts = int(os.getenv('DASHSCOPE_VIDEO_RETRY_TIMES', 10))  # Increased default retries for video
+        wait_time = int(os.getenv('DASHSCOPE_VIDEO_SLEEP_TIME', 5))  # Increased default wait time for video
+        query_url = f"{query_base_url}{task_id}"
+
+        for attempt in range(max_attempts):
+            # Wait before polling
+            time.sleep(wait_time)
+            logger.info(f"Polling attempt {attempt + 1}/{max_attempts}...")
+
+            # Poll for results
+            query_response = requests.get(query_url, headers={'Authorization': f'Bearer {api_key}'})
+
+            if query_response.status_code != 200:
+                logger.info(f"Poll request failed with status code {query_response.status_code}")
+                continue
+
+            try:
+                query_result = query_response.json()
+            except json.JSONDecodeError as e:
+                logger.warning(f"Failed to parse response as JSON: {e}")
+                continue
+
+            # Check task status
+            task_status = query_result.get("output", {}).get("task_status")
+
+            if task_status == "SUCCEEDED":
+                # Extract video URL
+                video_url = query_result.get("output", {}).get("video_url")
+
+                if video_url:
+                    # Return as array of objects with video_url for consistency with image API
+                    return json.dumps({"video_url": video_url})
+                else:
+                    logger.info("Video URL not found in the response")
+                    return None
+            elif task_status in ["PENDING", "RUNNING"]:
+                # If still running, continue to next polling attempt
+                logger.info(f"gen_video_item Task status: {task_status}, continuing to next poll...")
+                continue
+            elif task_status == "FAILED":
+                logger.warning("Task failed")
+                return None
+            else:
+                # Any other status, return None
+                logger.warning(f"Unexpected status: {task_status}")
+                return None
+
+        # If we get here, polling timed out
+        logger.warning("Polling timed out after maximum attempts")
+        return None
+    except Exception as e:
+        logger.warning(f"Exception gen_video_item occurred: {e}")
+        return None