PraisonAI/examples/python/usecases/quantum-optimiser.py

from praisonaiagents import Agent, Task, AgentTeam
import time
from typing import Dict, List
import asyncio

def analyze_quantum_circuit():
    """Simulates quantum circuit analysis"""
    circuit_metrics = {
        "depth": 15 + (int(time.time()) % 10),
        "gates": {
            "single_qubit": 25 + (int(time.time()) % 15),
            "two_qubit": 10 + (int(time.time()) % 8),
            "measurement": 5 + (int(time.time()) % 3)
        },
        "qubits": 8 + (int(time.time()) % 4)
    }
    return circuit_metrics

def identify_optimization_opportunities(metrics: Dict):
    """Identifies potential optimization areas"""
    opportunities = []
    if metrics["depth"] > 20:
        opportunities.append({
            "type": "depth_reduction",
            "potential": "high",
            "technique": "parallel_gates"
        })
    if metrics["gates"]["two_qubit"] > 15:
        opportunities.append({
            "type": "gate_reduction",
            "potential": "medium",
            "technique": "gate_decomposition"
        })
    return opportunities

def simulate_optimization(opportunities: List[Dict]):
    """Simulates optimization results"""
    results = []
    for opt in opportunities:
        improvement = {
            "high": 0.3 + (time.time() % 2) / 10,
            "medium": 0.2 + (time.time() % 2) / 10,
            "low": 0.1 + (time.time() % 2) / 10
        }
        results.append({
            "type": opt["type"],
            "improvement": improvement[opt["potential"]],
            "success_probability": 0.8 + (time.time() % 2) / 10
        })
    return results

def benchmark_performance(original: Dict, optimized: List[Dict]):
    """Benchmarks optimization performance"""
    benchmarks = {
        "execution_time": {
            "original": 100 + (int(time.time()) % 50),
            "optimized": 70 + (int(time.time()) % 30)
        },
        "fidelity": {
            "original": 0.9 + (time.time() % 1) / 10,
            "optimized": 0.85 + (time.time() % 1) / 10
        },
        "resource_usage": {
            "original": original["qubits"],
            "optimized": original["qubits"] - len(optimized)
        }
    }
    return benchmarks

def validate_results(benchmarks: Dict):
    """Validates optimization results"""
    validations = {
        "performance_gain": (benchmarks["execution_time"]["original"] - 
                           benchmarks["execution_time"]["optimized"]) / 
                           benchmarks["execution_time"]["original"],
        "fidelity_maintained": benchmarks["fidelity"]["optimized"] > 0.8,
        "resource_efficient": benchmarks["resource_usage"]["optimized"] < 
                            benchmarks["resource_usage"]["original"]
    }
    return validations

# Create specialized agents
circuit_analyzer = Agent(
    name="Circuit Analyzer",
    role="Circuit Analysis",
    goal="Analyze quantum circuit structure",
    instructions="Evaluate quantum circuit metrics",
    tools=[analyze_quantum_circuit]
)

optimization_finder = Agent(
    name="Optimization Finder",
    role="Optimization Discovery",
    goal="Identify optimization opportunities",
    instructions="Find potential optimization techniques",
    tools=[identify_optimization_opportunities]
)

optimizer = Agent(
    name="Circuit Optimizer",
    role="Circuit Optimization",
    goal="Optimize quantum circuit",
    instructions="Apply optimization techniques",
    tools=[simulate_optimization]
)

benchmarker = Agent(
    name="Performance Benchmarker",
    role="Performance Analysis",
    goal="Benchmark optimization results",
    instructions="Measure performance improvements",
    tools=[benchmark_performance]
)

validator = Agent(
    name="Results Validator",
    role="Validation",
    goal="Validate optimization results",
    instructions="Ensure optimization quality",
    tools=[validate_results]
)

# Create workflow tasks
analysis_task = Task(
    name="analyze_circuit",
    description="Analyze quantum circuit",
    expected_output="Circuit metrics",
    agent=circuit_analyzer,
    is_start=True,
    next_tasks=["find_optimizations"]
)

optimization_task = Task(
    name="find_optimizations",
    description="Find optimization opportunities",
    expected_output="Optimization opportunities",
    agent=optimization_finder,
    next_tasks=["optimize_circuit"]
)

optimization_execution_task = Task(
    name="optimize_circuit",
    description="Execute circuit optimization",
    expected_output="Optimization results",
    agent=optimizer,
    next_tasks=["benchmark_performance"]
)

benchmark_task = Task(
    name="benchmark_performance",
    description="Benchmark optimization",
    expected_output="Performance metrics",
    agent=benchmarker,
    next_tasks=["validate_results"],
    context=[analysis_task, optimization_execution_task]
)

validation_task = Task(
    name="validate_results",
    description="Validate optimization results",
    expected_output="Validation results",
    agent=validator,
    task_type="decision",
    condition={
        "True": "",  # End workflow if validation passes
        "False": ["find_optimizations"]  # Retry optimization if validation fails
    }
)

# Create workflow
workflow = AgentTeam(
    agents=[circuit_analyzer, optimization_finder, optimizer,
            benchmarker, validator],
    tasks=[analysis_task, optimization_task, optimization_execution_task,
           benchmark_task, validation_task],
    process="workflow", output="verbose"
)

async def main():
    print("\nStarting Quantum Algorithm Optimization Workflow...")
    print("=" * 50)
    
    # Run workflow
    results = await workflow.astart()
    
    # Print results
    print("\nOptimization Results:")
    print("=" * 50)
    task_results = (
        results.get("task_results", {})
        if isinstance(results, dict)
        else {}
    )
    if not task_results and isinstance(results, str):
        print(f"\nWorkflow output:\n{results}")
    for task_id, result in task_results.items():
        if result:
            print(f"\nTask: {task_id}")
            print(f"Result: {result.raw}")
            print("-" * 50)

if __name__ == "__main__":
    asyncio.run(main())