mobai
8dea68a25b
Corrects a common misconception: the 60s heartbeat must not call any LLM.
It runs the mood decay equation -- microseconds of float math. Giving it a
model is an architecture error, not a resource problem.
Real token spend lands in two places:
- per-turn input context (ontology + state + dyad + memory)
- background task frequency
So route by scene, not by time:
none rule/math tick, reach-out decision, fillers, avatar drive
low local small life events, memory gist, fact extract
standard mid + short speculative-response second segment
high flagship main dialogue, proactive message, crisis
Hard constraints:
- dialogue floor is LOW, never NONE (can get dumber, cannot lose lines)
- crisis never degrades (product constraint, not perf tuning)
- degradation is predictable: one rung per request, floor only at
the hard budget threshold
Cost levers, by actual payoff:
1. prompt caching (fixed prefix) -- by far the largest
2. upstream compression (small model condenses 20 memories to 3)
3. batching (one batch per 30min, not 20 single calls)
4. swapping to a smaller model -- smallest payoff, hurts persona
Also: cost degradation is narratable. 'Net error -> I zoned out' already
exists in the projection table, so being throttled is in-character and true.
- config.py: tiers / routing / budget, replacing llm+llm_fast+llm_background
- think/: scenes, budget, router, wiring
- models.yaml restructured around tiers
- SPEC gains 9.5
- 20 new tests; 78 total pass; ruff clean
211 lines
6.7 KiB
Python
211 lines
6.7 KiB
Python
"""模型路由与预算测试。
|
||
|
||
守的是这一层的核心判断：
|
||
**大模型按关键时刻分配，不按时间分配。**
|
||
|
||
如果哪天有人把 tick 路由到 high，test_heartbeat_never_uses_model 会红。
|
||
如果哪天预算紧张把主对话降成"无模型"，test_dialogue_floor_is_low 会红。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
|
||
from cybergril.providers import EchoLLM
|
||
from cybergril.think import (
|
||
BudgetConfig,
|
||
BudgetGuard,
|
||
ModelRouter,
|
||
Scene,
|
||
Tier,
|
||
build_router,
|
||
floor_for,
|
||
is_model_scene,
|
||
)
|
||
from cybergril.think.scenes import CONTEXT_BUDGET, NEVER_NONE
|
||
|
||
|
||
@pytest.fixture
|
||
def providers() -> dict[Tier, EchoLLM | None]:
|
||
return {
|
||
Tier.HIGH: EchoLLM(prefix="[high]"),
|
||
Tier.STANDARD: EchoLLM(prefix="[std]"),
|
||
Tier.LOW: EchoLLM(prefix="[low]"),
|
||
Tier.NONE: None,
|
||
}
|
||
|
||
|
||
@pytest.fixture
|
||
def router(providers) -> ModelRouter:
|
||
return build_router(providers=providers)
|
||
|
||
|
||
# ---------------------------------------------------------------- 路由基表
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"scene",
|
||
[Scene.TICK, Scene.FILLER, Scene.REACH_OUT, Scene.AVATAR_DRIVE],
|
||
)
|
||
def test_heartbeat_and_rules_never_use_model(router: ModelRouter, scene: Scene) -> None:
|
||
"""心跳、填充词、主动性判定、表情：不调用任何模型。"""
|
||
d = router.route(scene)
|
||
assert d.effective is Tier.NONE
|
||
assert d.provider is None
|
||
assert d.is_rule_based
|
||
assert d.context_budget == 0
|
||
|
||
|
||
def test_dialogue_main_uses_high(router: ModelRouter) -> None:
|
||
d = router.route(Scene.DIALOGUE_MAIN)
|
||
assert d.effective is Tier.HIGH
|
||
assert d.provider is not None
|
||
assert d.context_budget == CONTEXT_BUDGET[Tier.HIGH]
|
||
|
||
|
||
def test_background_tasks_use_low(router: ModelRouter) -> None:
|
||
for scene in (Scene.MEMORY_GIST, Scene.FACT_EXTRACT, Scene.LIFE_EVENT):
|
||
d = router.route(scene)
|
||
assert d.effective is Tier.LOW, scene
|
||
|
||
|
||
def test_is_model_scene_helper() -> None:
|
||
assert not is_model_scene(Scene.TICK)
|
||
assert is_model_scene(Scene.DIALOGUE_MAIN)
|
||
|
||
|
||
# ---------------------------------------------------------------- 预算降级
|
||
|
||
|
||
def test_budget_tight_drops_to_floor(providers) -> None:
|
||
"""到达硬阈值时直接降到该场景下限，而不是温和过渡。"""
|
||
guard = BudgetGuard(BudgetConfig(daily_tokens=1000, hard_stop_ratio=0.95), today="2026-01-01")
|
||
guard.record(960, "test") # 96%
|
||
router = build_router(providers=providers)
|
||
router.guard = guard
|
||
|
||
d = router.route(Scene.DIALOGUE_MAIN)
|
||
assert d.requested is Tier.HIGH
|
||
assert d.effective is Tier.LOW # dialogue 的下限
|
||
assert d.degraded
|
||
assert "预算" in (d.degraded_reason or "")
|
||
|
||
|
||
def test_dialogue_floor_is_low_not_none(providers) -> None:
|
||
"""预算耗尽时，主对话可以降成小模型，但绝不能没有台词。"""
|
||
guard = BudgetGuard(BudgetConfig(daily_tokens=1000, hard_stop_ratio=0.5), today="2026-01-01")
|
||
guard.record(999, "test")
|
||
router = build_router(providers=providers)
|
||
router.guard = guard
|
||
|
||
d = router.route(Scene.DIALOGUE_MAIN)
|
||
assert d.effective is Tier.LOW
|
||
assert d.provider is not None # 还有模型可用
|
||
assert floor_for(Scene.DIALOGUE_MAIN) is Tier.LOW
|
||
|
||
|
||
def test_crisis_never_downgrades(providers) -> None:
|
||
"""危机场景即使预算耗尽也保持高档位。
|
||
|
||
这是产品意义上的强制约束，不是性能优化。
|
||
"""
|
||
guard = BudgetGuard(BudgetConfig(daily_tokens=100, hard_stop_ratio=0.1), today="2026-01-01")
|
||
guard.record(1000, "test")
|
||
router = build_router(providers=providers)
|
||
router.guard = guard
|
||
|
||
d = router.route(Scene.CRISIS)
|
||
assert d.effective is Tier.HIGH
|
||
assert floor_for(Scene.CRISIS) is Tier.HIGH
|
||
|
||
|
||
def test_never_none_scenes_are_declared() -> None:
|
||
assert Scene.DIALOGUE_MAIN in NEVER_NONE
|
||
assert Scene.CRISIS in NEVER_NONE
|
||
|
||
|
||
def test_estimated_tokens_downgrades_one_step(providers) -> None:
|
||
"""单次请求放不下时只降一档——降级要可预测。"""
|
||
guard = BudgetGuard(BudgetConfig(daily_tokens=1000, hard_stop_ratio=0.95), today="2026-01-01")
|
||
guard.record(900, "test") # 90%，还没到硬阈值
|
||
router = build_router(providers=providers)
|
||
router.guard = guard
|
||
|
||
d = router.route(Scene.DIALOGUE_MAIN, estimated_tokens=500)
|
||
assert d.effective is Tier.STANDARD
|
||
assert "降一档" in (d.degraded_reason or "")
|
||
|
||
|
||
# ---------------------------------------------------------------- 缺 provider
|
||
|
||
|
||
def test_missing_tier_falls_through(providers) -> None:
|
||
"""某一档没配 provider 时应该往下找，而不是崩。"""
|
||
providers[Tier.HIGH] = None
|
||
router = build_router(providers=providers)
|
||
|
||
d = router.route(Scene.PROACTIVE_MESSAGE)
|
||
assert d.requested is Tier.HIGH
|
||
assert d.effective is Tier.STANDARD
|
||
assert d.provider is not None
|
||
assert d.degraded
|
||
|
||
|
||
# ---------------------------------------------------------------- 预算记账
|
||
|
||
|
||
def test_budget_records_by_scene() -> None:
|
||
guard = BudgetGuard(BudgetConfig(daily_tokens=10_000), today="2026-01-01")
|
||
guard.record(100, Scene.DIALOGUE_MAIN.value)
|
||
guard.record(300, Scene.MEMORY_GIST.value)
|
||
guard.record(50, Scene.DIALOGUE_MAIN.value)
|
||
|
||
assert guard.used == 450
|
||
top = guard.top_scenes()
|
||
assert top[0][0] == Scene.MEMORY_GIST.value
|
||
assert top[0][1] == 300
|
||
|
||
|
||
def test_budget_usage_dict() -> None:
|
||
guard = BudgetGuard(BudgetConfig(daily_tokens=10_000), today="2026-01-01")
|
||
guard.record_usage({"prompt_tokens": 100, "completion_tokens": 50}, "x")
|
||
assert guard.used == 150
|
||
|
||
|
||
def test_budget_ratio_and_remaining() -> None:
|
||
guard = BudgetGuard(BudgetConfig(daily_tokens=1000), today="2026-01-01")
|
||
guard.record(250, "x")
|
||
assert guard.ratio() == 0.25
|
||
assert guard.remaining() == 750
|
||
assert not guard.is_tight()
|
||
assert not guard.is_warning()
|
||
|
||
|
||
def test_budget_warning_threshold() -> None:
|
||
guard = BudgetGuard(BudgetConfig(daily_tokens=1000, warn_ratio=0.7), today="2026-01-01")
|
||
guard.record(750, "x")
|
||
assert guard.is_warning()
|
||
assert not guard.is_tight()
|
||
|
||
|
||
def test_budget_summary_readable() -> None:
|
||
guard = BudgetGuard(BudgetConfig(daily_tokens=1000), today="2026-01-01")
|
||
guard.record(300, "dialogue_main")
|
||
text = guard.summary()
|
||
assert "300" in text
|
||
assert "dialogue_main" in text
|
||
|
||
|
||
# ---------------------------------------------------------------- 上下文预算
|
||
|
||
|
||
def test_context_budget_decreases_with_tier() -> None:
|
||
assert CONTEXT_BUDGET[Tier.HIGH] > CONTEXT_BUDGET[Tier.STANDARD]
|
||
assert CONTEXT_BUDGET[Tier.STANDARD] > CONTEXT_BUDGET[Tier.LOW]
|
||
assert CONTEXT_BUDGET[Tier.NONE] == 0
|
||
|
||
|
||
def test_routing_decision_str(router: ModelRouter) -> None:
|
||
assert "dialogue_main" in str(router.route(Scene.DIALOGUE_MAIN))
|
||
assert "tick" in str(router.route(Scene.TICK))
|