Skip to content

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).

Functions
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/openjarvis/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.

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

Get child skills in the hierarchy.

Source code in src/openjarvis/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/openjarvis/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).

Functions
get_competence
get_competence(skill_id: str) -> float

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

Source code in src/openjarvis/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/openjarvis/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/openjarvis/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/openjarvis/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/openjarvis/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/openjarvis/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)