PraisonAI/examples/python/usecases/research-assistant.py

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

def analyze_research_papers():
    """Simulates research paper analysis"""
    papers = [
        {
            "topic": "quantum_computing",
            "gaps": ["error_correction", "scalability"],
            "methodology": "experimental",
            "impact_factor": 8.5
        },
        {
            "topic": "machine_learning",
            "gaps": ["interpretability", "robustness"],
            "methodology": "theoretical",
            "impact_factor": 7.8
        }
    ]
    return papers[int(time.time()) % 2]

def identify_knowledge_gaps(papers: Dict):
    """Identifies research gaps"""
    gaps = []
    for gap in papers["gaps"]:
        gaps.append({
            "area": gap,
            "significance": "high" if papers["impact_factor"] > 8 else "medium",
            "research_potential": 0.8 + (time.time() % 2) / 10
        })
    return gaps

def design_experiment(gaps: List[Dict]):
    """Designs experimental methodology"""
    experiments = []
    for gap in gaps:
        experiments.append({
            "area": gap["area"],
            "methodology": {
                "type": "quantitative",
                "duration": "3_months",
                "resources_needed": ["equipment_A", "dataset_B"],
                "expected_outcome": "validation_metrics"
            },
            "priority": gap["significance"]
        })
    return experiments

def validate_methodology(experiments: List[Dict]):
    """Validates experimental design"""
    validations = []
    for exp in experiments:
        validations.append({
            "area": exp["area"],
            "statistical_power": 0.9 + (time.time() % 1) / 10,
            "resource_feasibility": 0.8 + (time.time() % 1) / 10,
            "ethical_compliance": True,
            "recommendations": []
        })
    return validations

def predict_impact(experiments: List[Dict], validations: List[Dict]):
    """Predicts research impact"""
    predictions = []
    for exp, val in zip(experiments, validations):
        predictions.append({
            "area": exp["area"],
            "potential_impact": val["statistical_power"] * 10,
            "novelty_score": 0.7 + (time.time() % 3) / 10,
            "breakthrough_probability": 0.5 + (time.time() % 4) / 10
        })
    return predictions

# Create specialized agents
paper_analyzer = Agent(
    name="Paper Analyzer",
    role="Research Analysis",
    goal="Analyze research papers",
    instructions="Review and analyze scientific papers",
    tools=[analyze_research_papers]
)

gap_identifier = Agent(
    name="Gap Identifier",
    role="Gap Analysis",
    goal="Identify knowledge gaps",
    instructions="Identify research opportunities",
    tools=[identify_knowledge_gaps]
)

experiment_designer = Agent(
    name="Experiment Designer",
    role="Experimental Design",
    goal="Design research experiments",
    instructions="Create experimental methodologies",
    tools=[design_experiment]
)

methodology_validator = Agent(
    name="Methodology Validator",
    role="Validation",
    goal="Validate experimental design",
    instructions="Ensure methodology validity",
    tools=[validate_methodology]
)

impact_predictor = Agent(
    name="Impact Predictor",
    role="Impact Analysis",
    goal="Predict research impact",
    instructions="Assess potential impact",
    tools=[predict_impact]
)

# Create workflow tasks
analysis_task = Task(
    name="analyze_papers",
    description="Analyze research papers",
    expected_output="Paper analysis results",
    agent=paper_analyzer,
    is_start=True,
    next_tasks=["identify_gaps"]
)

gap_task = Task(
    name="identify_gaps",
    description="Identify knowledge gaps",
    expected_output="Research gaps",
    agent=gap_identifier,
    next_tasks=["design_experiments"]
)

design_task = Task(
    name="design_experiments",
    description="Design experiments",
    expected_output="Experimental designs",
    agent=experiment_designer,
    next_tasks=["validate_methodology"]
)

validation_task = Task(
    name="validate_methodology",
    description="Validate methodology",
    expected_output="Validation results",
    agent=methodology_validator,
    next_tasks=["predict_impact"]
)

prediction_task = Task(
    name="predict_impact",
    description="Predict research impact",
    expected_output="Impact predictions",
    agent=impact_predictor,
    task_type="decision",
    condition={
        "high": "",  # End workflow if high impact
        "medium": ["design_experiments"],  # Refine if medium impact
        "low": ["identify_gaps"]  # Restart if low impact
    },
    context=[design_task, validation_task]
)

# Create workflow
workflow = AgentTeam(
    agents=[paper_analyzer, gap_identifier, experiment_designer,
            methodology_validator, impact_predictor],
    tasks=[analysis_task, gap_task, design_task,
           validation_task, prediction_task],
    process="workflow", output="verbose"
)

async def main():
    print("\nStarting Research Assistant Workflow...")
    print("=" * 50)
    
    # Run workflow
    results = await workflow.astart()
    
    # Print results
    print("\nResearch Analysis 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())