system-prompts-and-models-o.../dealix/auto_client_acquisition/ai/model_router.py
Dealix Builder e34cc729aa feat(dealix): py3.10/3.11 compat shim + 54 unit tests for business/innovation/ai
PROBLEM
The codebase used Python 3.11+ stdlib features (`from datetime import UTC`,
`from enum import StrEnum`) in 22 files, breaking local dev on Python 3.10
(Windows users) and any pytest run that imports the affected modules.

SOLUTION
1. New `core/_py_compat.py` providing UTC + StrEnum shims that:
   - On 3.11+ re-export the stdlib names (zero overhead)
   - On 3.10 fall back to `timezone.utc` and a (str, Enum) backport

2. All 22 affected files patched to import from the shim:
   - core/utils.py, core/config/models.py
   - api/routers/admin.py
   - auto_client_acquisition/{ai/model_router, agents/{intake,icp_matcher},
     v3/{memory,agents,compliance_os,market_radar},
     personal_operator/{operator,memory,launch_report},
     innovation/{proof_ledger_repo,command_feed_live}}.py
   - autonomous_growth/agents/sector_intel.py
   - dealix/{trust/{approval,tool_verification,policy},
     observability/cost_tracker,
     contracts/{evidence_pack,event_envelope,audit_log,decision},
     classifications/__init__,
     governance/approvals}.py

3. Three new test suites for previously-untested layers (54 tests):
   - tests/unit/test_business_suite.py — gtm_plan, launch_metrics,
     market_positioning, pricing_strategy, proof_pack, unit_economics,
     verticals (28 tests covering plan recommendation, performance fee,
     ROI math, account health grading, vertical playbook structure)
   - tests/unit/test_innovation_suite.py — aeo_radar, command_feed,
     deal_rooms, experiments, growth_missions, proof_ledger, ten_in_ten
     (18 tests covering deterministic reproducibility, card type taxonomy,
     pending-approval invariant, kill-mission visibility)
   - tests/unit/test_ai_model_router.py — ModelTask + get_model_route +
     estimate_model_cost_class + requires_guardrail (8 tests covering
     enum integrity, route round-trip, guardrail bool contract)

VERIFICATION
- ast.parse green on all 22 patched files
- pytest tests/unit/ → 477 passed, 2 skipped (provider smoke needs API keys)
  on Python 3.10.12 venv with project requirements installed
- No behavior change on 3.11+: the shim re-exports stdlib symbols

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-01 14:50:04 +03:00

79 lines
2.8 KiB
Python

"""Task-based model routing — provider-agnostic, deterministic."""
from __future__ import annotations
from dataclasses import dataclass
from core._py_compat import StrEnum
from typing import Literal
CostClass = Literal["low", "medium", "high"]
class ModelTask(StrEnum):
STRATEGIC_REASONING = "strategic_reasoning"
ARABIC_WRITING = "arabic_writing"
CLASSIFICATION = "classification"
COMPLIANCE_GUARDRAIL = "compliance_guardrail"
PROJECT_CODE_UNDERSTANDING = "project_code_understanding"
SUMMARIZATION = "summarization"
EXTRACTION = "extraction"
FORECASTING = "forecasting"
CUSTOMER_SUPPORT = "customer_support"
BULK_ENRICHMENT = "bulk_enrichment"
@dataclass(frozen=True)
class ModelRoute:
task: ModelTask
quality_tier: Literal["standard", "high"]
latency: Literal["low", "medium", "high"]
cost_class: CostClass
fallback_task: ModelTask | None
guardrail_required: bool
eval_metric: str
def get_model_route(task: ModelTask) -> ModelRoute:
"""Return routing guidance without binding to a vendor model name."""
table: dict[ModelTask, ModelRoute] = {
ModelTask.STRATEGIC_REASONING: ModelRoute(
task, "high", "medium", "high", ModelTask.SUMMARIZATION, True, "decision_accuracy",
),
ModelTask.ARABIC_WRITING: ModelRoute(
task, "high", "medium", "medium", ModelTask.SUMMARIZATION, True, "arabic_tone_and_grounding",
),
ModelTask.CLASSIFICATION: ModelRoute(
task, "standard", "low", "low", None, True, "precision_recall",
),
ModelTask.COMPLIANCE_GUARDRAIL: ModelRoute(
task, "high", "low", "medium", ModelTask.CLASSIFICATION, True, "block_rate_vs_false_positives",
),
ModelTask.PROJECT_CODE_UNDERSTANDING: ModelRoute(
task, "high", "medium", "high", ModelTask.SUMMARIZATION, True, "grounded_citations",
),
ModelTask.SUMMARIZATION: ModelRoute(
task, "standard", "low", "low", None, False, "faithfulness",
),
ModelTask.EXTRACTION: ModelRoute(
task, "standard", "medium", "medium", ModelTask.CLASSIFICATION, True, "field_f1",
),
ModelTask.FORECASTING: ModelRoute(
task, "high", "high", "high", ModelTask.SUMMARIZATION, True, "forecast_error",
),
ModelTask.CUSTOMER_SUPPORT: ModelRoute(
task, "standard", "low", "medium", ModelTask.SUMMARIZATION, True, "resolution_rate",
),
ModelTask.BULK_ENRICHMENT: ModelRoute(
task, "standard", "high", "low", None, False, "cost_per_row",
),
}
return table.get(task, table[ModelTask.SUMMARIZATION])
def estimate_model_cost_class(task: ModelTask) -> CostClass:
return get_model_route(task).cost_class
def requires_guardrail(task: ModelTask) -> bool:
return get_model_route(task).guardrail_required