types

types

Core data types for SkillOrchestra.

• Skill
• AgentProfile
• BetaCompetence
• ModeMetadata
• RoutingInsight
• CostStats

Classes

BetaCompetence dataclass

BetaCompetence(alpha: float = 1.0, beta: float = 1.0)

Bayesian competence estimate for an agent on a specific skill.

skill_scores / get_competence use empirical_rate (successes/attempts)

Attributes

empirical_rate property

empirical_rate: float

Empirical success rate: successes/attempts.

variance property

variance: float

Posterior variance.

total_observations property

total_observations: int

Total observations (excluding prior)

CostStats dataclass

CostStats(
    avg_prompt_tokens: float = 0.0,
    avg_completion_tokens: float = 0.0,
    avg_latency_s: float = 0.0,
    avg_cost_usd: float = 0.0,
    avg_completion_cost_usd: float = 0.0,
    avg_prompt_cost_usd: float = 0.0,
    total_executions: int = 0,
)

Execution cost statistics for an agent under a specific mode.

Tracks both total cost (prompt + completion) and completion-only cost separately, since completion cost is the variable component that differs most between models (prompt cost is roughly constant for the same query).

Methods:

update

update(
    prompt_tokens: float,
    completion_tokens: float,
    latency_s: float,
    cost_usd: float,
    completion_cost_usd: float = 0.0,
    prompt_cost_usd: float = 0.0,
) -> None

Incremental running-average update.

Source code in src/diapason/agents/hybrid/skillorchestra/types.py

def update(
    self,
    prompt_tokens: float,
    completion_tokens: float,
    latency_s: float,
    cost_usd: float,
    completion_cost_usd: float = 0.0,
    prompt_cost_usd: float = 0.0,
) -> None:
    """Incremental running-average update."""
    n = self.total_executions
    self.avg_prompt_tokens = (self.avg_prompt_tokens * n + prompt_tokens) / (n + 1)
    self.avg_completion_tokens = (
        self.avg_completion_tokens * n + completion_tokens
    ) / (n + 1)
    self.avg_latency_s = (self.avg_latency_s * n + latency_s) / (n + 1)
    self.avg_cost_usd = (self.avg_cost_usd * n + cost_usd) / (n + 1)
    self.avg_completion_cost_usd = (
        self.avg_completion_cost_usd * n + completion_cost_usd
    ) / (n + 1)
    self.avg_prompt_cost_usd = (self.avg_prompt_cost_usd * n + prompt_cost_usd) / (
        n + 1
    )
    self.total_executions = n + 1

RoutingInsight dataclass

RoutingInsight(
    insight_id: str = (lambda: hex[:8])(),
    content: str = "",
    insight_type: str = "",
    evidence_query_ids: List[str] = list(),
    confidence: float = 0.0,
)

A single routing insight learned from execution traces

ModeMetadata dataclass

ModeMetadata(
    mode: str = "",
    description: str = "",
    insights: List[RoutingInsight] = list(),
)

Mode-level routing metadata.

SkillProvenance dataclass

SkillProvenance(
    discovered_from_queries: List[str] = list(),
    positive_trajectories: List[str] = list(),
    negative_trajectories: List[str] = list(),
    discovery_round: int = 0,
    refinement_history: List[Dict[str, Any]] = list(),
)

Tracks how and why a skill was discovered.

Skill dataclass

Skill(
    skill_id: str = "",
    name: str = "",
    description: str = "",
    indicators: List[str] = list(),
    examples: List[str] = list(),
    mode: str = "",
    parent_skill_id: Optional[str] = None,
    provenance: SkillProvenance = SkillProvenance(),
)

A reusable capability abstraction.

Methods:

get_children

get_children(all_skills: Dict[str, Skill]) -> List[Skill]

Get child skills in the hierarchy.

Source code in src/diapason/agents/hybrid/skillorchestra/types.py

def get_children(self, all_skills: Dict[str, Skill]) -> List[Skill]:
    """Get child skills in the hierarchy."""
    return [s for s in all_skills.values() if s.parent_skill_id == self.skill_id]

is_leaf

is_leaf(all_skills: Dict[str, Skill]) -> bool

True if this skill has no children.

Source code in src/diapason/agents/hybrid/skillorchestra/types.py

def is_leaf(self, all_skills: Dict[str, Skill]) -> bool:
    """True if this skill has no children."""
    return len(self.get_children(all_skills)) == 0

AgentProfile dataclass

AgentProfile(
    agent_id: str = "",
    mode: str = "",
    model_name: str = "",
    tools: List[str] = list(),
    skill_competence: Dict[str, BetaCompetence] = dict(),
    total_attempts: int = 0,
    total_successes: int = 0,
    cost_stats: CostStats = CostStats(),
    routing_signals: List[str] = list(),
    strengths: List[str] = list(),
    weaknesses: List[str] = list(),
)

Agent profile for skill-aware orchestration.

Attributes

overall_success_rate property

overall_success_rate: float

Overall success rate (trajectory-level when available, else skill-level).

Methods:

get_competence

get_competence(skill_id: str) -> float

Get empirical success rate for a skill. Returns 0 if unseen.

Source code in src/diapason/agents/hybrid/skillorchestra/types.py

def get_competence(self, skill_id: str) -> float:
    """Get empirical success rate for a skill. Returns 0 if unseen."""
    if skill_id in self.skill_competence:
        return self.skill_competence[skill_id].empirical_rate
    return 0.0

get_competence_dist

get_competence_dist(skill_id: str) -> BetaCompetence

Get full Beta distribution for a skill, creating with prior if unseen.

Source code in src/diapason/agents/hybrid/skillorchestra/types.py

def get_competence_dist(self, skill_id: str) -> BetaCompetence:
    """Get full Beta distribution for a skill, creating with prior if unseen."""
    if skill_id not in self.skill_competence:
        self.skill_competence[skill_id] = BetaCompetence()
    return self.skill_competence[skill_id]

update_competence

update_competence(skill_id: str, success: bool) -> None

Update competence estimate for a skill.

Source code in src/diapason/agents/hybrid/skillorchestra/types.py

def update_competence(self, skill_id: str, success: bool) -> None:
    """Update competence estimate for a skill."""
    self.get_competence_dist(skill_id).update(success)

weighted_competence

weighted_competence(
    skill_weights: Dict[str, float],
) -> float

Compute weighted competence: sum w_{t,sigma} * alpha/(alpha+beta).

Source code in src/diapason/agents/hybrid/skillorchestra/types.py

def weighted_competence(self, skill_weights: Dict[str, float]) -> float:
    """Compute weighted competence: sum w_{t,sigma} * alpha/(alpha+beta)."""
    if not skill_weights:
        return 0.5
    total = 0.0
    for skill_id, weight in skill_weights.items():
        total += weight * self.get_competence(skill_id)
    return total

category_competence

category_competence(category_prefix: str) -> float

Aggregate competence on all skills under a category (skill_id prefix).

E.g. category_competence('entertainment_knowledge') = avg of get_competence(s) for all s where s.startswith('entertainment_knowledge.').

Source code in src/diapason/agents/hybrid/skillorchestra/types.py

def category_competence(self, category_prefix: str) -> float:
    """Aggregate competence on all skills under a category (skill_id prefix).

    E.g. category_competence('entertainment_knowledge') = avg of
    get_competence(s) for all s where s.startswith('entertainment_knowledge.').
    """
    prefix = category_prefix.rstrip(".") + "."
    scores = [
        self.get_competence(sid)
        for sid in self.skill_competence
        if sid.startswith(prefix)
    ]
    return sum(scores) / len(scores) if scores else 0.0

category_competence_for_skills

category_competence_for_skills(
    active_skill_ids: List[str],
) -> float

Category-level competence for hierarchical tie-breaking.

Extracts parent categories from active_skill_ids (e.g. 'entertainment_knowledge' from 'entertainment_knowledge.episodic_competition_outcome'), computes category_competence for each, returns average.

Source code in src/diapason/agents/hybrid/skillorchestra/types.py

def category_competence_for_skills(self, active_skill_ids: List[str]) -> float:
    """Category-level competence for hierarchical tie-breaking.

    Extracts parent categories from active_skill_ids (e.g. 'entertainment_knowledge'
    from 'entertainment_knowledge.episodic_competition_outcome'), computes
    category_competence for each, returns average.
    """
    categories: set = set()
    for sid in active_skill_ids:
        cat = sid.rsplit(".", 1)[0] if "." in sid else sid
        categories.add(cat)
    if not categories:
        return 0.0
    return sum(self.category_competence(cat) for cat in categories) / len(
        categories
    )