Process Manager Tools

process_manager_tools

# SCENARIO: Comprehensive customer analysis pipeline with multiple specialized agents

print("=== Starting Customer Analysis Pipeline ===\n")

# Step 1: Data Extraction (delegate to specialized extractor)
print("Step 1: Extracting customer data...")
extraction_result = _delegate_work(
    prompt="""
Extract customer data for Q1 2024 including:
1. Transaction history from database
2. Support ticket data
3. Product usage metrics
4. Customer satisfaction scores

Return data in JSON format with structure:
{
  "transactions": [...],
  "support_tickets": [...],
  "usage_metrics": {...},
  "satisfaction_scores": {...}
}
""",
    target_bot="data_extractor_bot",
    timeout_seconds=300
)

if not extraction_result['success']:
    print(f"❌ Extraction failed: {extraction_result['error']}")
    exit(1)

print("✅ Data extraction complete\n")
customer_data = extraction_result['result']

# Step 2: Parallel Analysis (using async sub-threads)
print("Step 2: Running parallel analysis tasks...")

# Start transaction analysis (async)
transaction_analysis = sub_thread(
    task=f"""
Analyze transaction data: {customer_data['transactions']}

Calculate:
1. Total revenue by customer segment
2. Average transaction value
3. Purchase frequency trends
4. Top product categories
5. Customer lifetime value estimates

Return structured analysis with key metrics.
""",
    allowed_tools=['code_executor_tools', 'data_connector_tools'],
    async_mode=True,
    timeout_seconds=600
)

# Start support analysis (async)
support_analysis = sub_thread(
    task=f"""
Analyze support ticket data: {customer_data['support_tickets']}

Identify:
1. Most common issues
2. Resolution time trends
3. Customer satisfaction correlation
4. Issue severity distribution
5. Recurring problem patterns

Return structured analysis with insights.
""",
    allowed_tools=['code_executor_tools', 'data_connector_tools'],
    async_mode=True,
    timeout_seconds=600
)

# Start usage analysis (async)
usage_analysis = sub_thread(
    task=f"""
Analyze product usage metrics: {customer_data['usage_metrics']}

Determine:
1. Feature adoption rates
2. User engagement levels
3. Churn risk indicators
4. Power user characteristics
5. Underutilized features

Return structured analysis with recommendations.
""",
    allowed_tools=['code_executor_tools', 'data_connector_tools'],
    async_mode=True,
    timeout_seconds=600
)

# Wait for all analyses to complete
import time
analyses = {
    'transaction': transaction_analysis['thread_id'],
    'support': support_analysis['thread_id'],
    'usage': usage_analysis['thread_id']
}

results = {}
max_wait = 600  # 10 minutes total
start_time = time.time()

while len(results) < 3 and (time.time() - start_time) < max_wait:
    for name, thread_id in analyses.items():
        if name not in results:
            status = check_sub_thread_status(thread_id=thread_id)
            if status['status'] == 'completed':
                results[name] = status['result']
                print(f"✅ {name.title()} analysis complete")
            elif status['status'] == 'error':
                results[name] = {'error': status['error']}
                print(f"❌ {name.title()} analysis failed: {status['error']}")
    
    if len(results) < 3:
        time.sleep(10)  # Check every 10 seconds

print(f"\n✅ All analyses complete ({len(results)}/3)\n")

# Step 3: Generate insights report (delegate to analyst)
print("Step 3: Generating comprehensive insights...")
insights_result = _delegate_work(
    prompt=f"""
Create comprehensive customer insights report from these analyses:

Transaction Analysis:
{results.get('transaction', 'Not available')}

Support Analysis:
{results.get('support', 'Not available')}

Usage Analysis:
{results.get('usage', 'Not available')}

Generate report including:
1. Executive summary
2. Key findings and trends
3. Customer segments with characteristics
4. Risk and opportunity analysis
5. Strategic recommendations
6. Data-driven action items

Format as structured markdown report.
""",
    target_bot="senior_analyst_bot",
    timeout_seconds=300
)

if not insights_result['success']:
    print(f"❌ Insights generation failed: {insights_result['error']}")
    exit(1)

print("✅ Insights report generated\n")

# Step 4: Generate visualizations (delegate to viz specialist)
print("Step 4: Creating visualizations...")
viz_result = _delegate_work(
    prompt=f"""
Create visualizations for customer insights report:

Data:
{results}

Generate:
1. Revenue trend chart (line chart)
2. Customer segment distribution (pie chart)
3. Support ticket categories (bar chart)
4. Feature adoption heatmap
5. Churn risk matrix

Return Plotly JSON specifications for each chart.
""",
    target_bot="visualization_bot",
    timeout_seconds=180
)

print("✅ Visualizations created\n")

# Step 5: Generate customer-specific reports (batch processing)
print("Step 5: Generating individual customer reports...")

top_customers = [
    {"customer_id": "C001", "name": "Acme Corp", "tier": "Enterprise"},
    {"customer_id": "C002", "name": "TechStart Inc", "tier": "Pro"},
    {"customer_id": "C003", "name": "Global Solutions", "tier": "Enterprise"}
]

customer_reports = repeat(
    instruction="""
Generate personalized report for this customer:
- Customer ID: {customer_id}
- Customer Name: {name}
- Tier: {tier}

Include:
1. Personalized insights from overall analysis
2. Customer-specific metrics
3. Tailored recommendations
4. Account health score

Save report to customer_reports/{customer_id}_report.pdf
""",
    items=top_customers,
    timeout_seconds=120
)

print(f"✅ Generated {customer_reports['completed_iterations']} customer reports\n")

# Step 6: Quality review (delegate to QA bot)
print("Step 6: Quality assurance review...")
qa_result = _delegate_work(
    prompt=f"""
Review the following deliverables for quality and accuracy:

1. Insights Report: {insights_result['result']}
2. Visualizations: {viz_result['result']}
3. Customer Reports: {customer_reports['results']}

Check for:
- Data accuracy and consistency
- Completeness of analysis
- Clarity of recommendations
- Visual quality and correctness
- Actionability of insights

Provide quality score (1-10) and list any issues found.
""",
    target_bot="qa_specialist_bot",
    timeout_seconds=180
)

print(f"✅ QA Review complete\n")
print(f"Quality Score: {qa_result['result'].get('score', 'N/A')}/10")

# Step 7: Final assembly and distribution
print("Step 7: Assembling final deliverable...")
final_result = _delegate_work(
    prompt=f"""
Create final customer analysis package:

Components:
1. Executive insights report
2. Visualization suite
3. Individual customer reports (3)
4. QA review summary

Package as:
- Single PDF with all components
- Separate files in organized folder structure
- Email summary for stakeholders

Save to: /customer_analysis_q1_2024/
Send email notification to: analytics-team@company.com
""",
    target_bot="document_specialist_bot",
    timeout_seconds=240
)

print("✅ Final deliverable assembled and distributed\n")

print("=== Customer Analysis Pipeline Complete ===")
print(f"Summary:")
print(f"- Data extracted for Q1 2024")
print(f"- 3 parallel analyses completed")
print(f"- Comprehensive insights generated")
print(f"- {len(top_customers)} customer-specific reports created")
print(f"- QA Score: {qa_result['result'].get('score', 'N/A')}/10")
print(f"- Final package delivered to analytics team")

# Start multiple async sub-threads
thread_ids = []

tasks = [
    "Analyze sales data for North region",
    "Analyze sales data for South region",
    "Analyze sales data for East region",
    "Analyze sales data for West region"
]

for task in tasks:
    result = sub_thread(
        task=task,
        async_mode=True,
        timeout_seconds=300
    )
    thread_ids.append(result['thread_id'])

# Wait for all to complete
import time
results = []
start_time = time.time()

while len(results) < len(thread_ids) and (time.time() - start_time) < 600:
    for thread_id in thread_ids:
        if thread_id not in [r['thread_id'] for r in results]:
            status = check_sub_thread_status(thread_id=thread_id)
            if status['status'] in ['completed', 'error']:
                results.append({
                    'thread_id': thread_id,
                    'status': status['status'],
                    'result': status.get('result')
                })
    
    if len(results) < len(thread_ids):
        time.sleep(5)

print(f"Completed {len(results)}/{len(thread_ids)} tasks")

def delegate_with_retry(prompt, target_bot, max_attempts=3):
    """Delegate work with automatic retry on failure"""
    for attempt in range(max_attempts):
        try:
            result = _delegate_work(
                prompt=prompt,
                target_bot=target_bot,
                timeout_seconds=300
            )
            
            if result['success']:
                return result
            
            print(f"Attempt {attempt + 1} failed: {result.get('error')}")
            
            # Exponential backoff
            if attempt < max_attempts - 1:
                wait_time = 2 ** attempt
                print(f"Retrying in {wait_time} seconds...")
                time.sleep(wait_time)
        
        except Exception as e:
            print(f"Attempt {attempt + 1} exception: {str(e)}")
            if attempt == max_attempts - 1:
                raise
    
    return {'success': False, 'error': 'Max retries exceeded'}

# Usage
result = delegate_with_retry(
    prompt="Extract customer data",
    target_bot="data_bot",
    max_attempts=3
)

# Step 1: Assess data quality
quality_check = _delegate_work(
    prompt="Assess data quality of customer_data.csv",
    target_bot="qa_bot"
)

# Step 2: Conditional path based on quality
if quality_check['result']['quality_score'] >= 8:
    # High quality: proceed with full analysis
    analysis = _delegate_work(
        prompt="Perform comprehensive analysis on customer_data.csv",
        target_bot="advanced_analyst_bot",
        timeout_seconds=600
    )
else:
    # Low quality: data cleaning required first
    cleaning = _delegate_work(
        prompt="Clean and validate customer_data.csv",
        target_bot="data_cleaner_bot"
    )
    
    # Then proceed with analysis
    analysis = _delegate_work(
        prompt=f"Analyze cleaned data: {cleaning['result']}",
        target_bot="analyst_bot"
    )

print(f"Analysis complete: {analysis['result']}")

# Start with basic analysis
current_result = _delegate_work(
    prompt="Perform basic statistical analysis on sales_data.csv",
    target_bot="stats_bot"
)

# Enhance with trend analysis
current_result = _delegate_work(
    prompt=f"Add trend analysis to these statistics: {current_result['result']}",
    target_bot="trends_bot"
)

# Enhance with forecasting
current_result = _delegate_work(
    prompt=f"Add 3-month forecast based on this analysis: {current_result['result']}",
    target_bot="forecast_bot"
)

# Final enhancement with recommendations
final_result = _delegate_work(
    prompt=f"Add strategic recommendations to this complete analysis: {current_result['result']}",
    target_bot="strategy_bot"
)

print(f"Final enhanced analysis: {final_result['result']}")