Merge pull request MervinPraison#493 from MervinPraison/claude/issue-411-20250524_055043

Fix Mini Agents sequential task data passing issue

Author: MervinPraison

src/praisonai-agents/praisonaiagents/agents/agents.py

@@ -13,6 +13,16 @@
 from ..process.process import Process, LoopItems
 import asyncio
 import uuid
+from enum import Enum
+
+# Task status constants
+class TaskStatus(Enum):
+    """Enumeration for task status values to ensure consistency"""
+    COMPLETED = "completed"
+    IN_PROGRESS = "in progress"
+    NOT_STARTED = "not started"
+    FAILED = "failed"
+    UNKNOWN = "unknown"
 
 # Set up logger
 logger = logging.getLogger(__name__)
@@ -49,6 +59,55 @@ def process_video(video_path: str, seconds_per_frame=2):
     video.release()
     return base64_frames
 
+def process_task_context(context_item, verbose=0, user_id=None):
+    """
+    Process a single context item for task execution.
+    This helper function avoids code duplication between async and sync execution methods.
+    
+    Args:
+        context_item: The context item to process (can be string, list, task object, or dict)
+        verbose: Verbosity level for logging
+        user_id: User ID for database queries
+        
+    Returns:
+        str: Formatted context string for this item
+    """
+    if isinstance(context_item, str):
+        return f"Input Content:\n{context_item}"
+    elif isinstance(context_item, list):
+        return f"Input Content: {' '.join(str(x) for x in context_item)}"
+    elif hasattr(context_item, 'result'):  # Task object
+        # Ensure the previous task is completed before including its result
+        task_status = getattr(context_item, 'status', None)
+        task_name = context_item.name if context_item.name else context_item.description
+        
+        if context_item.result and task_status == TaskStatus.COMPLETED.value:
+            return f"Result of previous task {task_name}:\n{context_item.result.raw}"
+        elif task_status == TaskStatus.COMPLETED.value and not context_item.result:
+            return f"Previous task {task_name} completed but produced no result."
+        else:
+            return f"Previous task {task_name} is not yet completed (status: {task_status or TaskStatus.UNKNOWN.value})."
+    elif isinstance(context_item, dict) and "vector_store" in context_item:
+        from ..knowledge.knowledge import Knowledge
+        try:
+            # Handle both string and dict configs
+            cfg = context_item["vector_store"]
+            if isinstance(cfg, str):
+                cfg = json.loads(cfg)
+            
+            knowledge = Knowledge(config={"vector_store": cfg}, verbose=verbose)
+            
+            # Only use user_id as filter
+            db_results = knowledge.search(
+                context_item.get("query", ""),  # Use query from context if available
+                user_id=user_id if user_id else None
+            )
+            return f"[DB Context]: {str(db_results)}"
+        except Exception as e:
+            return f"[Vector DB Error]: {e}"
+    else:
+        return str(context_item)  # Fallback for unknown types
+
 class PraisonAIAgents:
     def __init__(self, agents, tasks=None, verbose=0, completion_checker=None, max_retries=5, process="sequential", manager_llm=None, memory=False, memory_config=None, embedder=None, user_id=None, max_iter=10, stream=True, name: Optional[str] = None):
         # Add check at the start if memory is requested
@@ -250,44 +309,20 @@ async def aexecute_task(self, task_id):
         if task.context:
             context_results = []  # Use list to avoid duplicates
             for context_item in task.context:
-                if isinstance(context_item, str):
-                    context_results.append(f"Input Content:\n{context_item}")
-                elif isinstance(context_item, list):
-                    context_results.append(f"Input Content: {' '.join(str(x) for x in context_item)}")
-                elif hasattr(context_item, 'result'):  # Task object
-                    if context_item.result:
-                        context_results.append(
-                            f"Result of previous task {context_item.name if context_item.name else context_item.description}:\n{context_item.result.raw}"
-                        )
-                    else:
-                        context_results.append(
-                            f"Previous task {context_item.name if context_item.name else context_item.description} has no result yet."
-                        )
-                elif isinstance(context_item, dict) and "vector_store" in context_item:
-                    from ..knowledge.knowledge import Knowledge
-                    try:
-                        # Handle both string and dict configs
-                        cfg = context_item["vector_store"]
-                        if isinstance(cfg, str):
-                            cfg = json.loads(cfg)
-                        
-                        knowledge = Knowledge(config={"vector_store": cfg}, verbose=self.verbose)
-                        
-                        # Only use user_id as filter
-                        db_results = knowledge.search(
-                            task.description,
-                            user_id=self.user_id if self.user_id else None
-                        )
-                        context_results.append(f"[DB Context]: {str(db_results)}")
-                    except Exception as e:
-                        context_results.append(f"[Vector DB Error]: {e}")
+                # Use the centralized helper function
+                context_str = process_task_context(context_item, self.verbose, self.user_id)
+                context_results.append(context_str)
 
-            # Join unique context results
+            # Join unique context results with proper formatting
             unique_contexts = list(dict.fromkeys(context_results))  # Remove duplicates
+            if self.verbose >= 3:
+                logger.info(f"Task {task_id} context items: {len(unique_contexts)}")
+                for i, ctx in enumerate(unique_contexts):
+                    logger.info(f"Context {i+1}: {ctx[:100]}...")
             task_prompt += f"""
 Context:
 
-{'  '.join(unique_contexts)}
+{'\n\n'.join(unique_contexts)}
 """
         task_prompt += "Please provide only the final result of your work. Do not add any conversation or extra explanation."
 
@@ -573,44 +608,20 @@ def execute_task(self, task_id):
         if task.context:
             context_results = []  # Use list to avoid duplicates
             for context_item in task.context:
-                if isinstance(context_item, str):
-                    context_results.append(f"Input Content:\n{context_item}")
-                elif isinstance(context_item, list):
-                    context_results.append(f"Input Content: {' '.join(str(x) for x in context_item)}")
-                elif hasattr(context_item, 'result'):  # Task object
-                    if context_item.result:
-                        context_results.append(
-                            f"Result of previous task {context_item.name if context_item.name else context_item.description}:\n{context_item.result.raw}"
-                        )
-                    else:
-                        context_results.append(
-                            f"Previous task {context_item.name if context_item.name else context_item.description} has no result yet."
-                        )
-                elif isinstance(context_item, dict) and "vector_store" in context_item:
-                    from ..knowledge.knowledge import Knowledge
-                    try:
-                        # Handle both string and dict configs
-                        cfg = context_item["vector_store"]
-                        if isinstance(cfg, str):
-                            cfg = json.loads(cfg)
-                        
-                        knowledge = Knowledge(config={"vector_store": cfg}, verbose=self.verbose)
-                        
-                        # Only use user_id as filter
-                        db_results = knowledge.search(
-                            task.description,
-                            user_id=self.user_id if self.user_id else None
-                        )
-                        context_results.append(f"[DB Context]: {str(db_results)}")
-                    except Exception as e:
-                        context_results.append(f"[Vector DB Error]: {e}")
+                # Use the centralized helper function
+                context_str = process_task_context(context_item, self.verbose, self.user_id)
+                context_results.append(context_str)
 
-            # Join unique context results
+            # Join unique context results with proper formatting
             unique_contexts = list(dict.fromkeys(context_results))  # Remove duplicates
+            if self.verbose >= 3:
+                logger.info(f"Task {task_id} context items: {len(unique_contexts)}")
+                for i, ctx in enumerate(unique_contexts):
+                    logger.info(f"Context {i+1}: {ctx[:100]}...")
             task_prompt += f"""
 Context:
 
-{'  '.join(unique_contexts)}
+{'\n\n'.join(unique_contexts)}
 """
 
         # Add memory context if available

tests/unit/agent/test_mini_agents_fix.py

@@ -0,0 +1,111 @@
+#!/usr/bin/env python3
+"""
+Simple test to verify the Mini Agents sequential task data passing fix.
+This tests the core functionality without external dependencies.
+"""
+
+import sys
+import os
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))
+
+try:
+    from praisonaiagents import Agent, Agents
+    print("✅ Successfully imported PraisonAI Agents")
+except ImportError as e:
+    print(f"❌ Failed to import: {e}")
+    sys.exit(1)
+
+def test_context_processing():
+    """Test the context processing logic without running actual agents"""
+    
+    print("\n=== Testing Context Processing Logic ===")
+    
+    # Simulate a task result object
+    class MockTaskResult:
+        def __init__(self, raw_output):
+            self.raw = raw_output
+    
+    # Simulate a completed task object
+    class MockTask:
+        def __init__(self, name, result_text, status="completed"):
+            self.name = name
+            self.result = MockTaskResult(result_text) if result_text else None
+            self.status = status
+            self.description = f"Mock task: {name}"
+    
+    # Test the context processing logic
+    task1 = MockTask("research_task", "AI 2024 analysis: Major breakthroughs in machine learning")
+    task2 = MockTask("summary_task", None, "in progress")  # Task not completed yet
+    task3 = MockTask("analysis_task", "Key trends: Neural networks, transformers, LLMs", "completed")
+    
+    # Simulate context items like the real code does
+    context_items = [task1, task2, task3]
+    context_results = []
+    
+    for context_item in context_items:
+        if hasattr(context_item, 'result'):  # Task object
+            # Apply our fix: Ensure the previous task is completed before including its result
+            if context_item.result and getattr(context_item, 'status', None) == "completed":
+                context_results.append(
+                    f"Result of previous task {context_item.name if context_item.name else context_item.description}:\n{context_item.result.raw}"
+                )
+            elif getattr(context_item, 'status', None) == "completed" and not context_item.result:
+                context_results.append(
+                    f"Previous task {context_item.name if context_item.name else context_item.description} completed but produced no result."
+                )
+            else:
+                context_results.append(
+                    f"Previous task {context_item.name if context_item.name else context_item.description} is not yet completed (status: {getattr(context_item, 'status', 'unknown')})."
+                )
+    
+    # Apply our fix: Join with proper formatting
+    unique_contexts = list(dict.fromkeys(context_results))  # Remove duplicates
+    formatted_context = '\n\n'.join(unique_contexts)
+    
+    print("Context Results:")
+    print("================")
+    print(formatted_context)
+    print("================")
+    
+    # Verify the fix works
+    expected_patterns = [
+        "Result of previous task research_task:",
+        "AI 2024 analysis: Major breakthroughs in machine learning",
+        "summary_task is not yet completed (status: in progress)",
+        "Result of previous task analysis_task:",
+        "Key trends: Neural networks, transformers, LLMs"
+    ]
+    
+    success = True
+    for pattern in expected_patterns:
+        if pattern not in formatted_context:
+            print(f"❌ Missing expected pattern: {pattern}")
+            success = False
+        else:
+            print(f"✅ Found expected pattern: {pattern}")
+    
+    # Check formatting improvement
+    if '\n\n' in formatted_context and '  ' not in formatted_context.replace('  ', ' '):
+        print("✅ Context is properly formatted with newlines instead of spaces")
+    else:
+        print("❌ Context formatting issue")
+        success = False
+    
+    return success
+
+def main():
+    print("Testing Mini Agents Sequential Task Data Passing Fix")
+    print("=" * 60)
+    
+    success = test_context_processing()
+    
+    if success:
+        print("\n🎉 All tests passed! The fix should resolve the data passing issue.")
+    else:
+        print("\n❌ Tests failed. The fix needs more work.")
+    
+    return success
+
+if __name__ == "__main__":
+    success = main()
+    sys.exit(0 if success else 1)

tests/unit/agent/test_mini_agents_sequential.py

@@ -0,0 +1,64 @@
+#!/usr/bin/env python3
+"""
+Test script to reproduce the Mini Agents sequential task data passing issue.
+This will help verify the fix works correctly.
+"""
+
+import sys
+import os
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))
+
+from praisonaiagents import Agent, Agents
+
+def test_mini_agents_sequential_data_passing():
+    """Test that output from previous task is passed to next task in Mini Agents"""
+    
+    print("Testing Mini Agents Sequential Data Passing...")
+    
+    # Create two agents for sequential processing
+    agent1 = Agent(instructions="Generate the number 42 as your output. Only return the number 42, nothing else.")
+    agent2 = Agent(instructions="Take the input number and multiply it by 2. Only return the result number, nothing else.")
+    
+    # Create agents with sequential processing (Mini Agents pattern)
+    agents = Agents(agents=[agent1, agent2], verbose=True)
+    
+    # Execute the agents
+    result = agents.start(return_dict=True)
+    
+    print("\n=== Results ===")
+    print("Task Status:", result['task_status'])
+    print("\nTask Results:")
+    for task_id, task_result in result['task_results'].items():
+        if task_result:
+            print(f"Task {task_id}: {task_result.raw}")
+        else:
+            print(f"Task {task_id}: No result")
+    
+    # Check if the second task received the first task's output
+    task_results = result['task_results']
+    if len(task_results) >= 2:
+        task1_result = task_results[0]
+        task2_result = task_results[1]
+        
+        if task1_result and task2_result:
+            print(f"\nFirst task output: {task1_result.raw}")
+            print(f"Second task output: {task2_result.raw}")
+            
+            # The second task should have received "42" and returned "84"
+            if "42" in str(task1_result.raw) and "84" in str(task2_result.raw):
+                print("✅ SUCCESS: Data was passed correctly between tasks!")
+                return True
+            else:
+                print("❌ FAILED: Data was not passed correctly between tasks")
+                print(f"Expected first task to output '42', got: {task1_result.raw}")
+                print(f"Expected second task to output '84', got: {task2_result.raw}")
+                return False
+        else:
+            print("❌ FAILED: One or both tasks produced no result")
+            return False
+    else:
+        print("❌ FAILED: Not enough tasks were executed")
+        return False
+
+if __name__ == "__main__":
+    test_mini_agents_sequential_data_passing()