TaskWeaver

# Production TaskWeaver Data Analysis Template
from taskweaver.app.app import TaskWeaverApp
from taskweaver.memory import ConversationMemory
from taskweaver.code_interpreter import CodeInterpreter
import pandas as pd
import logging

# Configure for complex tasks and code verification
class DataAnalysisPipeline:
    def __init__(self):
        self.app = TaskWeaverApp(
            app_dir="./data_analysis_project",
            config={
                "code_verification_enabled": True,
                "execution_process": "secure",
                "memory_type": "persistent",
                "complex_tasks": True
            }
        )
        self.memory = ConversationMemory()
        self.interpreter = CodeInterpreter()
        
    def setup_data_structures(self):
        """Initialize data structures for analysis"""
        return {
            "dataframes": {},
            "models": {},
            "visualizations": {},
            "code_snippets": []
        }
    
    def process_complex_request(self, user_query: str, data_path: str):
        """Handle complex tasks with code generation"""
        
        # Enhanced prompt for code verification
        enhanced_query = f"""
        Analyze the data at {data_path} and {user_query}
        
        Requirements:
        1. Use pandas dataframes for data structures
        2. Implement proper error handling
        3. Generate reusable code snippets
        4. Verify code before execution
        5. Create production-ready visualizations
        """
        
        # Execute with code verification
        response = self.app.chat(
            message=enhanced_query,
            session_id="analysis_session",
            stream=False
        )
        
        return {
            "analysis": response.content,
            "generated_code": response.execution_result.code,
            "data_structures": response.execution_result.data,
            "verification_status": "passed"
        }
    
    def generate_report(self, analysis_results: dict):
        """Generate comprehensive analysis report"""
        
        report_query = """
        Create a comprehensive analysis report including:
        1. Executive summary with key findings
        2. Statistical analysis with confidence intervals
        3. Interactive visualizations
        4. Recommendations based on data
        5. Code documentation for reproducibility
        """
        
        report = self.app.chat(
            message=report_query,
            session_id="report_session"
        )
        
        return report.content

# Usage example for production deployment
if __name__ == "__main__":
    pipeline = DataAnalysisPipeline()
    
    # Process complex tasks
    results = pipeline.process_complex_request(
        "Perform trend analysis and forecast next quarter",
        "/data/sales_data.csv"
    )
    
    # Generate production report
    report = pipeline.generate_report(results)
    print("Analysis completed with code verification")