luzia/examples/prompt_engineering_demo.py

#!/usr/bin/env python3
"""
Prompt Engineering Demo Script

Demonstrates the advanced prompt engineering techniques implemented in Luzia.

Run with: python3 examples/prompt_engineering_demo.py
"""

import sys
from pathlib import Path

# Add lib to path
sys.path.insert(0, str(Path(__file__).parent.parent / "lib"))

from prompt_techniques import (
    TaskType, PromptStrategy, PromptEngineer,
    ChainOfThoughtEngine, FewShotExampleBuilder, RoleBasedPrompting,
    TaskSpecificPatterns
)
from prompt_integration import (
    PromptIntegrationEngine, ComplexityAdaptivePrompting,
    DomainSpecificAugmentor
)


def demo_chain_of_thought():
    """Demonstrate Chain-of-Thought prompting"""
    print("\n" + "="*80)
    print("DEMO 1: Chain-of-Thought Prompting")
    print("="*80)

    task = "Implement a distributed caching layer for database queries with TTL-based invalidation"

    print(f"\nOriginal Task:\n{task}\n")

    print("CoT Prompt (Complexity 3):")
    print("-" * 40)
    cot_prompt = ChainOfThoughtEngine.generate_cot_prompt(task, complexity=3)
    print(cot_prompt)


def demo_few_shot():
    """Demonstrate Few-Shot Learning"""
    print("\n" + "="*80)
    print("DEMO 2: Few-Shot Learning")
    print("="*80)

    builder = FewShotExampleBuilder()
    examples = builder.build_examples_for_task(TaskType.IMPLEMENTATION, num_examples=2)

    print("\nFew-Shot Examples for IMPLEMENTATION tasks:")
    print("-" * 40)
    formatted = builder.format_examples_for_prompt(examples)
    print(formatted)


def demo_role_based():
    """Demonstrate Role-Based Prompting"""
    print("\n" + "="*80)
    print("DEMO 3: Role-Based Prompting")
    print("="*80)

    for task_type in [TaskType.DEBUGGING, TaskType.ANALYSIS, TaskType.IMPLEMENTATION]:
        print(f"\n{task_type.value.upper()}:")
        print("-" * 40)
        role_prompt = RoleBasedPrompting.get_role_prompt(task_type)
        print(role_prompt[:200] + "...")


def demo_task_patterns():
    """Demonstrate Task-Specific Patterns"""
    print("\n" + "="*80)
    print("DEMO 4: Task-Specific Patterns")
    print("="*80)

    patterns = TaskSpecificPatterns()

    print("\nAnalysis Pattern:")
    print("-" * 40)
    analysis = patterns.get_analysis_pattern(
        "Database Performance",
        ["Query optimization", "Index efficiency", "Cache effectiveness"]
    )
    print(analysis[:300] + "...")

    print("\n\nDebugging Pattern:")
    print("-" * 40)
    debug = patterns.get_debugging_pattern(
        "Intermittent 503 errors under high load",
        "API Gateway"
    )
    print(debug[:300] + "...")


def demo_complexity_detection():
    """Demonstrate Complexity Estimation"""
    print("\n" + "="*80)
    print("DEMO 5: Complexity Adaptation")
    print("="*80)

    test_tasks = [
        ("Fix typo in README", "Simple"),
        ("Add logging to error handler", "Basic"),
        ("Implement rate limiting for API", "Moderate"),
        ("Refactor authentication system with concurrent access and security considerations", "Complex"),
        ("Design and implement distributed transaction system with encryption, failover, and performance optimization", "Very Complex")
    ]

    print("\nTask Complexity Detection:")
    print("-" * 40)

    for task, expected_level in test_tasks:
        complexity = ComplexityAdaptivePrompting.estimate_complexity(task, TaskType.IMPLEMENTATION)
        strategies = ComplexityAdaptivePrompting.get_prompting_strategies(complexity)

        print(f"\nTask: {task}")
        print(f"Expected: {expected_level} | Detected Complexity: {complexity}/5")
        print(f"Strategies: {len(strategies)} - {', '.join(s.value for s in strategies[:3])}")


def demo_integration_engine():
    """Demonstrate Full Integration Engine"""
    print("\n" + "="*80)
    print("DEMO 6: Full Prompt Integration Engine")
    print("="*80)

    # Initialize engine
    project_config = {
        "name": "luzia",
        "path": "/opt/server-agents/orchestrator",
        "focus": "Self-improving orchestrator for multi-project coordination"
    }

    engine = PromptIntegrationEngine(project_config)

    # Example 1: Simple implementation task
    task1 = "Add request rate limiting to the API endpoint"
    augmented1, metadata1 = engine.augment_for_task(
        task1,
        task_type=TaskType.IMPLEMENTATION,
        domain="backend"
    )

    print(f"\nTask 1: {task1}")
    print(f"Complexity: {metadata1['complexity']}/5")
    print(f"Strategies: {metadata1['strategies']}")
    print(f"Augmentation Ratio: {metadata1['final_token_estimate'] / len(task1.split()):.1f}x")
    print("\nAugmented Prompt (first 400 chars):")
    print("-" * 40)
    print(augmented1[:400] + "...\n")

    # Example 2: Complex debugging task
    task2 = """Debug intermittent race condition in async event handler that only manifests under high concurrent load.
    The issue causes occasional data corruption in shared state and we need to identify the synchronization issue and fix it."""

    augmented2, metadata2 = engine.augment_for_task(
        task2,
        task_type=TaskType.DEBUGGING,
        domain="backend"
    )

    print(f"Task 2: {task2[:80]}...")
    print(f"Complexity: {metadata2['complexity']}/5")
    print(f"Strategies: {metadata2['strategies']}")
    print(f"Augmentation Ratio: {metadata2['final_token_estimate'] / len(task2.split()):.1f}x")

    # Example 3: Security analysis
    task3 = "Analyze security implications of the current token storage approach"
    augmented3, metadata3 = engine.augment_for_task(
        task3,
        task_type=TaskType.ANALYSIS,
        domain="crypto",
        context={
            "previous_results": {
                "current_approach": "JWT stored in localStorage",
                "threat_model": "Browser-based XSS attacks"
            },
            "blockers": ["Need to decide on alternative storage mechanism"]
        }
    )

    print(f"\n\nTask 3: {task3}")
    print(f"Complexity: {metadata3['complexity']}/5")
    print(f"Strategies: {metadata3['strategies']}")
    print(f"Augmentation Ratio: {metadata3['final_token_estimate'] / len(task3.split()):.1f}x")


def demo_domain_contexts():
    """Demonstrate Domain-Specific Contexts"""
    print("\n" + "="*80)
    print("DEMO 7: Domain-Specific Contexts")
    print("="*80)

    domains = ["backend", "frontend", "crypto", "devops", "research", "orchestration"]

    for domain in domains:
        context = DomainSpecificAugmentor.get_domain_context(domain)
        print(f"\n{domain.upper()}:")
        print(f"  Focus: {context['focus']}")
        print(f"  Priorities: {', '.join(context['priorities'][:2])}")
        print(f"  Best Practices: {context['best_practices'][0]}")


def demo_context_continuation():
    """Demonstrate Task Continuation with Context"""
    print("\n" + "="*80)
    print("DEMO 8: Task Continuation with Previous Context")
    print("="*80)

    project_config = {
        "name": "luzia",
        "path": "/opt/server-agents/orchestrator",
        "focus": "Self-improving orchestrator"
    }

    engine = PromptIntegrationEngine(project_config)

    # Initial task
    initial_task = "Design a caching strategy for frequently accessed user profiles"
    print(f"\nInitial Task: {initial_task}")

    augmented_initial, metadata_initial = engine.augment_for_task(
        initial_task,
        task_type=TaskType.PLANNING,
        domain="backend"
    )
    print(f"Initial complexity: {metadata_initial['complexity']}")

    # Continuation with previous context
    context = {
        "previous_results": {
            "chosen_strategy": "Redis with TTL-based invalidation",
            "estimated_hit_rate": "85%",
            "cache_size": "~500MB per instance"
        },
        "state": {
            "implementation_status": "Completed caching layer",
            "current_focus": "Optimizing invalidation strategy"
        },
        "blockers": [
            "Need to handle cache stampede on popular profiles",
            "Invalidation latency causing stale data"
        ]
    }

    continuation_task = "Continue: optimize cache invalidation to prevent stampede and reduce staleness"
    print(f"\nContinuation Task: {continuation_task}")

    augmented_cont, metadata_cont = engine.augment_for_task(
        continuation_task,
        task_type=TaskType.IMPLEMENTATION,
        domain="backend",
        context=context
    )

    print(f"Continuation complexity: {metadata_cont['complexity']}")
    print(f"Context included: {bool(context)}")
    print("\nAugmented Prompt includes:")
    print("  ✓ System Instructions")
    print("  ✓ Role-Based Prompting (Senior Engineer)")
    print("  ✓ Domain Context (Backend best practices)")
    print("  ✓ Task Continuation (Previous results, current state, blockers)")
    print("  ✓ Task-Specific Pattern (Implementation)")


def main():
    """Run all demonstrations"""
    print("\n" + "█"*80)
    print("█ LUZIA ADVANCED PROMPT ENGINEERING DEMONSTRATIONS")
    print("█"*80)

    demos = [
        ("Chain-of-Thought Prompting", demo_chain_of_thought),
        ("Few-Shot Learning", demo_few_shot),
        ("Role-Based Prompting", demo_role_based),
        ("Task-Specific Patterns", demo_task_patterns),
        ("Complexity Adaptation", demo_complexity_detection),
        ("Full Integration Engine", demo_integration_engine),
        ("Domain-Specific Contexts", demo_domain_contexts),
        ("Task Continuation", demo_context_continuation),
    ]

    for i, (name, demo_func) in enumerate(demos, 1):
        try:
            demo_func()
        except Exception as e:
            print(f"\n[ERROR in {name}]: {e}")
            import traceback
            traceback.print_exc()

    print("\n" + "█"*80)
    print("█ DEMONSTRATIONS COMPLETE")
    print("█"*80)
    print("\nKey Takeaways:")
    print("1. Chain-of-Thought breaks complex reasoning into steps")
    print("2. Few-Shot examples improve understanding of task patterns")
    print("3. Role-based prompting sets appropriate expertise level")
    print("4. Complexity adaptation optimizes strategy selection")
    print("5. Domain-specific contexts apply relevant best practices")
    print("6. Task continuation preserves state across multi-step work")
    print("\nIntegrate into Luzia with:")
    print("  from prompt_integration import PromptIntegrationEngine")
    print("  engine = PromptIntegrationEngine(project_config)")
    print("  augmented, metadata = engine.augment_for_task(task, task_type, domain)")
    print()


if __name__ == "__main__":
    main()