Automate Strategic Planning 🚀

# Production OKR Dashboard with LangGraph (Real-time monitoring)
from langgraph.graph import Graph, END
from typing import TypedDict, List, Optional
import anthropic
import asyncio
from datetime import datetime, timedelta
import json

class OKRState(TypedDict):
    okrs: List[dict]
    current_metrics: dict
    historical_data: List[dict]
    analysis: Optional[dict]
    risks: List[dict]
    insights: List[dict]
    actions: List[dict]
    notifications_sent: bool
    dashboard_updated: bool

class ProductionOKRSystem:
    def __init__(self):
        self.client = anthropic.Anthropic()
        self.graph = self.build_graph()
    
    async def fetch_data_node(self, state: OKRState) -> OKRState:
        """Fetch OKRs and current metrics from multiple sources"""
        # Parallel data fetching
        okrs, metrics, historical = await asyncio.gather(
            self.fetch_okrs_async(),
            self.fetch_current_metrics_async(),
            self.fetch_historical_data_async()
        )
        
        state['okrs'] = okrs
        state['current_metrics'] = metrics
        state['historical_data'] = historical
        return state
    
    async def analyze_progress_node(self, state: OKRState) -> OKRState:
        """Analyze current progress vs targets"""
        prompt = f"""Analyze OKR progress with historical context.

Current OKRs:
{json.dumps(state['okrs'], indent=2)}

Current metrics:
{json.dumps(state['current_metrics'], indent=2)}

Historical data (last 4 weeks):
{json.dumps(state['historical_data'], indent=2)}

Provide JSON analysis with:
- overall_health (on_track/caution/critical)
- progress_by_objective (array with objective_id, completion_percent, trend)
- velocity_analysis (are we accelerating or decelerating?)
- forecast (will we hit targets at current pace?)
"""
        
        response = self.client.messages.create(
            model='claude-3-5-sonnet-20241022',
            max_tokens=3072,
            messages=[{'role': 'user', 'content': prompt}]
        )
        
        state['analysis'] = json.loads(response.content[0].text)
        return state
    
    async def identify_risks_node(self, state: OKRState) -> OKRState:
        """Identify and prioritize risks"""
        prompt = f"""Identify risks based on this analysis.

Analysis:
{json.dumps(state['analysis'], indent=2)}

For each risk provide:
- kr_id
- risk_type (velocity/trajectory/external)
- severity (critical/high/medium/low)
- time_to_impact (days until this becomes a problem)
- recommended_action
- estimated_effort (hours to address)

Return as JSON array."""
        
        response = self.client.messages.create(
            model='claude-3-5-sonnet-20241022',
            max_tokens=2048,
            messages=[{'role': 'user', 'content': prompt}]
        )
        
        state['risks'] = json.loads(response.content[0].text)
        return state
    
    async def generate_insights_node(self, state: OKRState) -> OKRState:
        """Generate strategic insights from analysis"""
        prompt = f"""Generate strategic insights from this OKR analysis.

Analysis:
{json.dumps(state['analysis'], indent=2)}

Risks:
{json.dumps(state['risks'], indent=2)}

Provide insights as JSON array with:
- insight (what we learned)
- implication (what this means for strategy)
- confidence (high/medium/low)
- data_points (supporting evidence)
- suggested_pivot (should we adjust targets or tactics?)
"""
        
        response = self.client.messages.create(
            model='claude-3-5-sonnet-20241022',
            max_tokens=2048,
            messages=[{'role': 'user', 'content': prompt}]
        )
        
        state['insights'] = json.loads(response.content[0].text)
        return state
    
    async def generate_actions_node(self, state: OKRState) -> OKRState:
        """Generate prioritized action items"""
        prompt = f"""Generate prioritized actions based on risks and insights.

Risks:
{json.dumps(state['risks'], indent=2)}

Insights:
{json.dumps(state['insights'], indent=2)}

For each action provide:
- priority (1-5, 1 is highest)
- action (specific, actionable task)
- owner (recommended role/team)
- deadline (relative: 'this week', 'next 2 weeks', etc.)
- impact (what KR this affects)
- effort_estimate (hours)

Return as JSON array sorted by priority."""
        
        response = self.client.messages.create(
            model='claude-3-5-sonnet-20241022',
            max_tokens=2048,
            messages=[{'role': 'user', 'content': prompt}]
        )
        
        state['actions'] = json.loads(response.content[0].text)
        return state
    
    async def notify_stakeholders_node(self, state: OKRState) -> OKRState:
        """Send notifications based on severity"""
        critical_risks = [r for r in state['risks'] if r['severity'] == 'critical']
        high_risks = [r for r in state['risks'] if r['severity'] == 'high']
        
        # Send Slack alerts for critical risks
        if critical_risks:
            await self.send_slack_alert(
                channel='#executive-alerts',
                message=f"🚨 {len(critical_risks)} CRITICAL OKR risks detected",
                risks=critical_risks,
                actions=[a for a in state['actions'] if a['priority'] == 1]
            )
        
        # Send daily digest to strategy team
        await self.send_email_digest(
            to='strategy-team@company.com',
            subject=f"OKR Update - {state['analysis']['overall_health'].upper()}",
            analysis=state['analysis'],
            top_actions=state['actions'][:5]
        )
        
        state['notifications_sent'] = True
        return state
    
    async def update_dashboards_node(self, state: OKRState) -> OKRState:
        """Update Notion, Asana, and internal dashboards"""
        # Update Notion dashboard
        await self.update_notion_dashboard(
            analysis=state['analysis'],
            risks=state['risks'],
            insights=state['insights']
        )
        
        # Create Asana tasks for high-priority actions
        for action in [a for a in state['actions'] if a['priority'] <= 2]:
            await self.create_asana_task(
                name=action['action'],
                assignee=action['owner'],
                due_date=self.parse_relative_date(action['deadline']),
                description=f"Impact: {action['impact']}\nEstimated effort: {action['effort_estimate']} hours"
            )
        
        state['dashboard_updated'] = True
        return state
    
    def should_send_alerts(self, state: OKRState) -> str:
        """Determine if we need to send alerts"""
        critical_count = len([r for r in state['risks'] if r['severity'] == 'critical'])
        high_count = len([r for r in state['risks'] if r['severity'] == 'high'])
        
        if critical_count > 0 or high_count >= 3:
            return "send_alerts"
        return "skip_alerts"
    
    def build_graph(self) -> Graph:
        """Build the LangGraph workflow"""
        graph = Graph()
        
        # Add nodes
        graph.add_node("fetch_data", self.fetch_data_node)
        graph.add_node("analyze_progress", self.analyze_progress_node)
        graph.add_node("identify_risks", self.identify_risks_node)
        graph.add_node("generate_insights", self.generate_insights_node)
        graph.add_node("generate_actions", self.generate_actions_node)
        graph.add_node("notify_stakeholders", self.notify_stakeholders_node)
        graph.add_node("update_dashboards", self.update_dashboards_node)
        
        # Define flow
        graph.set_entry_point("fetch_data")
        graph.add_edge("fetch_data", "analyze_progress")
        graph.add_edge("analyze_progress", "identify_risks")
        graph.add_edge("identify_risks", "generate_insights")
        graph.add_edge("generate_insights", "generate_actions")
        
        # Conditional notification
        graph.add_conditional_edges(
            "generate_actions",
            self.should_send_alerts,
            {
                "send_alerts": "notify_stakeholders",
                "skip_alerts": "update_dashboards"
            }
        )
        
        graph.add_edge("notify_stakeholders", "update_dashboards")
        graph.add_edge("update_dashboards", END)
        
        return graph.compile()
    
    async def run_daily_analysis(self) -> OKRState:
        """Run the full analysis workflow"""
        initial_state: OKRState = {
            'okrs': [],
            'current_metrics': {},
            'historical_data': [],
            'analysis': None,
            'risks': [],
            'insights': [],
            'actions': [],
            'notifications_sent': False,
            'dashboard_updated': False
        }
        
        result = await self.graph.ainvoke(initial_state)
        return result

# Usage (deploy as scheduled job)
async def main():
    system = ProductionOKRSystem()
    result = await system.run_daily_analysis()
    
    print(f"Analysis complete: {result['analysis']['overall_health']}")
    print(f"Risks identified: {len(result['risks'])}")
    print(f"Actions generated: {len(result['actions'])}")
    print(f"Notifications sent: {result['notifications_sent']}")

# Run daily at 8am via cron or Kubernetes CronJob
if __name__ == '__main__':
    asyncio.run(main())