AetherCrew: AI 직원 11명의 가상 회사 설계 및 구현

모델 폴백

이 코드가 하는 일: 직원의 모델 우선순위 목록과 사용할 수 없는 모델 목록을 받아, 앞에서부터 차례로 확인해 사용 가능한 첫 모델을 고릅니다. 건너뛴 모델은 fallbacks에 기록해 몇 번째 모델로 넘어갔는지 남기고, 목록이 비어 있거나 모든 모델을 쓸 수 없으면 조용히 넘어가지 않고 오류를 냅니다. 목록 길이만큼만 확인하므로 폴백이 끝없이 이어지지 않습니다.

class FallbackManager:
    """Selects next available agent with bounded fallback traversal."""

    def select(
        self,
        priority_list: list[str],
        unavailable_agents: set[str] | None = None,
    ) -> FallbackResult:
        if not priority_list:
            raise ValueError("priority_list must not be empty")
        blocked = unavailable_agents or set()
        fallbacks: list[str] = []
        for idx, model in enumerate(priority_list):
            if model in blocked:
                fallbacks.append(model)
                continue
            return FallbackResult(selected_agent=model, selected_index=idx, fallbacks=fallbacks)
        raise RuntimeError("All agents are unavailable; fallback exhausted")

예산 상태

이 코드가 하는 일: 기록된 비용을 모두 더해 예산 대비 사용률을 계산하고, 그 값에 따라 OK, WARN_80, WARN_95, HARD_STOP 네 상태 중 하나를 돌려줍니다. 예산이 0 이하로 설정되면 사용률을 100%로 보아 바로 HARD_STOP이 되므로, 예산 설정 실수가 무제한 실행으로 이어지지 않습니다. 오케스트레이터는 이 상태가 HARD_STOP이면 새 작업을 막습니다.

    @property
    def total_cost(self) -> float:
        return round(sum(e.cost_usd for e in self.entries), 6)

    def percent_used(self) -> float:
        if self.budget_policy.total_budget <= 0:
            return 100.0
        return (self.total_cost / self.budget_policy.total_budget) * 100

    def budget_state(self) -> str:
        pct = self.percent_used()
        if pct >= self.budget_policy.hard_stop_at_pct:
            return "HARD_STOP"
        if pct >= self.budget_policy.warn_at_pct[1]:
            return "WARN_95"
        if pct >= self.budget_policy.warn_at_pct[0]:
            return "WARN_80"
        return "OK"