Spaces:
Sleeping
Sleeping
File size: 6,255 Bytes
157b149 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 | """
Skill Dependency DAG: models how low scores in one skill block progress in another.
Hand-designed for MVP, encodes domain knowledge about coaching interdependencies.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Dict, List, Tuple
import numpy as np
@dataclass
class DependencyEdge:
"""An edge in the skill dependency graph."""
source: str # Upstream skill (the blocker)
target: str # Downstream skill (the blocked)
weight: float # Blocking strength [0, 1]
# Hand-designed dependency edges for MVP
DEFAULT_EDGES = [
DependencyEdge("emotional_reg", "consistency", 0.4),
DependencyEdge("emotional_reg", "focus", 0.2),
DependencyEdge("task_clarity", "follow_through", 0.5),
DependencyEdge("task_clarity", "consistency", 0.3),
DependencyEdge("self_trust", "social_courage", 0.3),
DependencyEdge("focus", "systems_thinking", 0.4),
]
class DependencyGraph:
"""
DAG of skill dependencies with blocking/bottleneck analysis.
Used to identify which upstream "dents" in the sphere block downstream
improvement, and to prioritize interventions at the root cause.
"""
def __init__(self, edges: List[DependencyEdge] | None = None):
self.edges = edges or DEFAULT_EDGES
self._adjacency: Dict[str, List[Tuple[str, float]]] = {}
self._reverse: Dict[str, List[Tuple[str, float]]] = {}
self._build()
def _build(self) -> None:
self._adjacency.clear()
self._reverse.clear()
for edge in self.edges:
self._adjacency.setdefault(edge.source, []).append(
(edge.target, edge.weight)
)
self._reverse.setdefault(edge.target, []).append(
(edge.source, edge.weight)
)
def find_bottlenecks(
self,
beliefs: Dict[str, np.ndarray],
low_threshold: float = 0.4,
) -> List[Dict]:
"""
Identify skills that are low AND block other skills.
A bottleneck occurs when:
- source skill's expected score is below low_threshold
- at least one downstream skill exists
Args:
beliefs: Dict mapping skill name -> belief vector (5 levels)
low_threshold: Normalized score threshold (0-1) below which
a skill is considered "low"
Returns:
List of bottleneck dicts sorted by impact (highest first):
[{"blocker": str, "blocked": [str], "score": float, "impact": float}]
"""
level_values = np.array([0.1, 0.3, 0.5, 0.7, 0.9])
bottlenecks = []
for source, targets in self._adjacency.items():
if source not in beliefs:
continue
score = float(np.dot(beliefs[source], level_values))
if score < low_threshold:
blocked_skills = []
total_impact = 0.0
for target, weight in targets:
blocked_skills.append(target)
total_impact += weight * (low_threshold - score)
if blocked_skills:
bottlenecks.append({
"blocker": source,
"blocked": blocked_skills,
"score": score,
"impact": total_impact,
})
bottlenecks.sort(key=lambda b: b["impact"], reverse=True)
return bottlenecks
def compute_impact_ranking(
self, beliefs: Dict[str, np.ndarray]
) -> List[Tuple[str, float]]:
"""
Rank all skills by improvement impact, considering downstream effects.
Impact of improving skill s = direct deficit + sum of downstream unblocking.
Returns:
List of (skill_name, impact_score) sorted by impact descending.
"""
level_values = np.array([0.1, 0.3, 0.5, 0.7, 0.9])
impacts = []
for skill_name, belief in beliefs.items():
score = float(np.dot(belief, level_values))
direct_deficit = max(0.0, 0.5 - score)
downstream_impact = 0.0
for target, weight in self._adjacency.get(skill_name, []):
downstream_impact += weight * direct_deficit
total = direct_deficit + downstream_impact
impacts.append((skill_name, total))
impacts.sort(key=lambda x: x[1], reverse=True)
return impacts
def get_blockers_for(self, skill: str) -> List[Tuple[str, float]]:
"""Get upstream skills that block a given skill."""
return self._reverse.get(skill, [])
def get_blocked_by(self, skill: str) -> List[Tuple[str, float]]:
"""Get downstream skills blocked by a given skill."""
return self._adjacency.get(skill, [])
def get_explanation(self, blocker: str, blocked: str) -> str:
"""Generate a human-readable explanation of a blocking relationship."""
explanations = {
("emotional_reg", "consistency"):
"When emotions are hard to manage, maintaining routines becomes much harder.",
("emotional_reg", "focus"):
"Emotional turbulence steals attention and makes focus difficult.",
("task_clarity", "follow_through"):
"Without a clear picture of what 'done' looks like, follow-through stalls.",
("task_clarity", "consistency"):
"Ambiguity about tasks makes it hard to build consistent habits.",
("self_trust", "social_courage"):
"When you doubt your own judgment, speaking up feels riskier.",
("focus", "systems_thinking"):
"Systems thinking needs sustained attention to hold multiple pieces together.",
}
return explanations.get(
(blocker, blocked),
f"Low {blocker.replace('_', ' ')} tends to limit {blocked.replace('_', ' ')}.",
)
def get_all_edges(self) -> List[Dict]:
"""Get all edges as dicts for visualization."""
return [
{"source": e.source, "target": e.target, "weight": e.weight}
for e in self.edges
]
|