Patterns: Reflection & Plan-Execute Pattern

flo_ai/examples/concept_demo.py

@@ -0,0 +1,284 @@
+"""
+PlanExecuteRouter Concept Demo
+
+This demonstrates the plan-execute concept without requiring API calls.
+Shows the architecture and explains how to fix the planner loop issue.
+"""
+
+import asyncio
+from flo_ai.arium.memory import PlanAwareMemory, ExecutionPlan, PlanStep, StepStatus
+import uuid
+
+
+def demonstrate_plan_aware_memory():
+    """Show how PlanAwareMemory works with ExecutionPlan objects"""
+    print('📋 PlanAwareMemory Concept Demo')
+    print('=' * 35)
+
+    # Create memory
+    memory = PlanAwareMemory()
+
+    # Create a sample execution plan
+    plan = ExecutionPlan(
+        id=str(uuid.uuid4()),
+        title='Build User Authentication API',
+        description='Create complete user auth system with registration and login',
+        steps=[
+            PlanStep(
+                id='design_schema',
+                description='Design user database schema',
+                agent='developer',
+                status=StepStatus.PENDING,
+            ),
+            PlanStep(
+                id='implement_registration',
+                description='Implement user registration endpoint',
+                agent='developer',
+                dependencies=['design_schema'],
+                status=StepStatus.PENDING,
+            ),
+            PlanStep(
+                id='implement_login',
+                description='Implement user login with JWT tokens',
+                agent='developer',
+                dependencies=['design_schema', 'implement_registration'],
+                status=StepStatus.PENDING,
+            ),
+            PlanStep(
+                id='test_endpoints',
+                description='Test all authentication endpoints',
+                agent='tester',
+                dependencies=['implement_login'],
+                status=StepStatus.PENDING,
+            ),
+            PlanStep(
+                id='security_review',
+                description='Review security implementation',
+                agent='reviewer',
+                dependencies=['test_endpoints'],
+                status=StepStatus.PENDING,
+            ),
+        ],
+    )
+
+    # Store plan in memory
+    memory.add_plan(plan)
+    print(f'✅ Plan stored: {plan.title}')
+    print(f'📊 Total steps: {len(plan.steps)}')
+
+    # Show initial state
+    def show_plan_status():
+        current = memory.get_current_plan()
+        print(f'\n📋 Plan Status: {current.title}')
+        for step in current.steps:
+            status_icon = {
+                StepStatus.PENDING: '○',
+                StepStatus.IN_PROGRESS: '⏳',
+                StepStatus.COMPLETED: '✅',
+                StepStatus.FAILED: '❌',
+            }.get(step.status, '○')
+            deps = (
+                f" (depends: {', '.join(step.dependencies)})"
+                if step.dependencies
+                else ''
+            )
+            print(f'  {status_icon} {step.id}: {step.description} → {step.agent}{deps}')
+
+    show_plan_status()
+
+    # Simulate step execution
+    print('\n🔄 Simulating step-by-step execution...')
+
+    current_plan = memory.get_current_plan()
+
+    # Execute step 1
+    next_steps = current_plan.get_next_steps()
+    print(f'\n🎯 Next steps ready: {len(next_steps)}')
+    step1 = next_steps[0]
+    print(f'⏳ Executing: {step1.description}')
+    step1.status = StepStatus.COMPLETED
+    step1.result = 'User table created with id, email, password_hash fields'
+    memory.update_plan(current_plan)
+
+    # Execute step 2
+    next_steps = current_plan.get_next_steps()
+    print(f'\n🎯 Next steps ready: {len(next_steps)}')
+    step2 = next_steps[0]
+    print(f'⏳ Executing: {step2.description}')
+    step2.status = StepStatus.COMPLETED
+    step2.result = 'POST /register endpoint with validation implemented'
+    memory.update_plan(current_plan)
+
+    show_plan_status()
+
+    print(f'\n📈 Plan completion status: {current_plan.is_completed()}')
+    print(f'📊 Remaining steps: {len(current_plan.get_next_steps())}')
+
+
+def explain_planner_loop_issue():
+    """Explain why the planner got stuck in a loop and how to fix it"""
+    print('\n\n🔄 Understanding the Planner Loop Issue')
+    print('=' * 45)
+
+    explanation = """
+❌ THE PROBLEM:
+In the original demo, the planner kept getting called in an infinite loop because:
+
+1. Router asks: "Is there a plan in memory?"
+2. Memory says: "No ExecutionPlan objects found"
+3. Router decides: "Route to planner to create plan"
+4. Planner generates plan TEXT but doesn't store ExecutionPlan OBJECTS
+5. Router asks again: "Is there a plan in memory?"
+6. Memory still says: "No ExecutionPlan objects found"
+7. INFINITE LOOP! 🔄
+
+✅ THE SOLUTION:
+We need to bridge the gap between plan TEXT and plan OBJECTS:
+
+APPROACH 1: Specialized PlannerAgent
+• Custom agent that parses plan text and stores ExecutionPlan objects
+• Router can detect when ExecutionPlan exists in memory
+• Automatically switches from planning to execution phase
+
+APPROACH 2: Content-Based Routing  
+• Router analyzes message content instead of relying on ExecutionPlan objects
+• If message contains "PLAN:", switch to execution mode
+• Simpler but less sophisticated
+
+APPROACH 3: State Management
+• Explicitly track workflow state (planning/executing/reviewing)
+• Router uses state instead of trying to detect plan completion
+• Most reliable but requires more setup
+"""
+
+    print(explanation)
+
+
+def show_router_intelligence():
+    """Show how the PlanExecuteRouter makes intelligent decisions"""
+    print('\n\n🧠 PlanExecuteRouter Intelligence')
+    print('=' * 35)
+
+    intelligence_demo = """
+The PlanExecuteRouter is designed to coordinate complex workflows by:
+
+🎯 PHASE DETECTION:
+• Planning Phase: No plan exists → route to planner
+• Execution Phase: Plan exists with pending steps → route to executor  
+• Review Phase: All steps complete → route to reviewer
+• Error Recovery: Failed steps exist → route for retry
+
+📊 STEP MANAGEMENT:
+• Analyzes step dependencies automatically
+• Only executes steps when dependencies are met
+• Tracks progress with visual indicators (○ ⏳ ✅ ❌)
+• Handles parallel execution of independent steps
+
+🔄 INTELLIGENT ROUTING:
+• Context-aware prompts with plan progress
+• Suggests next agent based on plan state
+• Prevents infinite loops with completion detection
+• Adapts to different workflow patterns
+
+💡 EXAMPLE ROUTING DECISIONS:
+
+Scenario 1: No plan exists
+→ Router: "Route to planner to create execution plan"
+
+Scenario 2: Plan exists, step_1 ready
+→ Router: "Route to developer to execute step_1"
+
+Scenario 3: Development complete, testing needed
+→ Router: "Route to tester to validate implementation"
+
+Scenario 4: All steps complete
+→ Router: "Route to reviewer for final approval"
+
+Scenario 5: Step failed
+→ Router: "Route to developer to retry failed step"
+"""
+
+    print(intelligence_demo)
+
+
+def show_working_implementation():
+    """Show the key components of a working implementation"""
+    print('\n\n🏗️ Working Implementation Components')
+    print('=' * 40)
+
+    implementation = """
+For a working PlanExecuteRouter implementation, you need:
+
+1. 📋 PLAN STORAGE:
+   ```python
+   memory = PlanAwareMemory()  # Stores ExecutionPlan objects
+   plan = ExecutionPlan(title="...", steps=[...])
+   memory.add_plan(plan)
+   ```
+
+2. 🤖 SPECIALIZED AGENTS:
+   ```python
+   class PlannerAgent(Agent):
+       def run(self, input_data):
+           plan_text = await super().run(input_data)
+           execution_plan = self.parse_plan(plan_text)
+           self.memory.add_plan(execution_plan)  # KEY!
+           return plan_text
+   ```
+
+3. 🎯 SMART ROUTING:
+   ```python
+   router = create_plan_execute_router(
+       planner_agent='planner',
+       executor_agent='developer',
+       reviewer_agent='reviewer'
+   )
+   ```
+
+4. 🔄 WORKFLOW COORDINATION:
+   ```python
+   arium = AriumBuilder()
+       .with_memory(memory)  # PlanAwareMemory
+       .add_agents([planner, developer, tester, reviewer])
+       .add_edge(planner, [...], router)
+       .build()
+   ```
+
+🎯 CRITICAL SUCCESS FACTORS:
+• PlannerAgent MUST store ExecutionPlan objects
+• Use PlanAwareMemory for plan state persistence  
+• Router needs to detect plan existence reliably
+• Agents should update step status during execution
+"""
+
+    print(implementation)
+
+
+async def main():
+    """Main concept demo"""
+    print('🎯 PlanExecuteRouter Concept Demo')
+    print('Understanding the architecture and fixing the planner loop\n')
+
+    # Demonstrate core concepts
+    demonstrate_plan_aware_memory()
+    explain_planner_loop_issue()
+    show_router_intelligence()
+    show_working_implementation()
+
+    print('\n\n🎉 Concept Demo Complete!')
+    print('=' * 30)
+    print('Key Takeaways:')
+    print('✅ PlanExecuteRouter enables Cursor-style plan-and-execute workflows')
+    print('✅ Planner loop occurs when plan TEXT ≠ plan OBJECTS in memory')
+    print('✅ Solution: Bridge the gap with specialized agents or content parsing')
+    print('✅ Working implementation requires PlanAwareMemory + ExecutionPlan objects')
+
+    print('\n🚀 Next Steps:')
+    print('• Try fixed_plan_execute_demo.py for working implementation')
+    print('• Use PlannerAgent that stores ExecutionPlan objects')
+    print('• Implement step status tracking for real progress monitoring')
+    print('• Customize agents for your specific workflow needs')
+
+
+if __name__ == '__main__':
+    asyncio.run(main())