xsl/meida_backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(llm): four-tier daily budget degradation with manual override

Browse Source 
把单一日预算熔断升级为架构 §13.3 的四档分级降级（T-P1-25b / NFR-COST-02）：

- budget.py：新增 BudgetTier 配置模型与档位判定（current_tier/tier_name/
  blocked_scenes）。阈值相对 LLM_DAILY_BUDGET_CENTS：warn ≥60% 只告警；
  degrade_secondary ≥80% 拒 hd_export/report_regen；degrade_generation ≥90%
  追加拒 hairstyle/tryon；block_all ≥100% 拒全部非 critical、allow 名单
  （默认 agent_chat）放行。高档位 blocked 取「截至当前档的并集」（架构原文
  「追加拒绝」）。critical 场景全档豁免但照常计成本。
- 手动 override：Redis 键 llm:budget:override（值 "1"~"4"），优先于成本水
  位，读不到/非法则忽略；set_override TTL 到当日 UTC+8 24 点，跨预算日必然
  失效。CostCounter 协议相应增加 set/delete。
- 指标：Gauge llm_budget_tier + Counter llm_budget_blocked_total{scene}
  挂到共享 REGISTRY 随 /metrics 暴露；档位上升沿只告警一次
  （budget_tier_raised，实例内记 last_tier）。
- errors.py 新增 SceneDegradedError（tier2/3 拒绝）；gateway.py 调用点换成
  ensure_scene_allowed（保留 ensure_within_budget 兼容入口）；agent.py 映射
  LLM_SCENE_DEGRADED(60006)；100% 档维持 DailyBudgetExceededError/60003 语义。
- 单测：新增 tests/unit/test_tiered_budget.py（27 例：四档边界、各档放行/
  拒绝、critical 豁免、override 加严/放宽/非法值/TTL、配置驱动）。
- 集成：tests/integration/test_vertical_slice.py 新增用例——预算 1 分 +
  block_all allow 清空 + Redis 注入当日成本，POST /agent/chat 返回 60003，
  断言 bl_ledger 中 consume(refunded,-1)+refund(committed,+1) 净和为 0，
  覆盖「reserve 之后才熔断 → refund」路径；fixture 重构为共享
  FakeProvider/fake_client，断言语义未弱化。

门禁：ruff check/format、mypy（packages/llm strict + 其余包）、lint-imports、
单测 196 passed、集成 4 passed。

偏差说明：
1. tests/unit/test_gateway.py 的 test_daily_budget_rejects_non_critical_
   scenes_before_spending 把被测场景 AGENT_CHAT 换成 REPORT_COPY（test_daily_
   budget.py 同用例换 tryon）：block_all 档默认 allow=["agent_chat"]（架构
   「仅保留纯文本对话」）后 agent_chat 在 100% 档不再被拒，原场景选择与新
   语义冲突；断言本身（进 provider 前被拒）未弱化。
2. 集成测试的 block_all 档配置为 blocked="*"+allow=[]（文档示例为
   blocked=["agent_chat"]）：后者按档位规则会走 SceneDegradedError/60006，
   前者才能按设计走既有 60003 语义（文档允许二者取其一且"按上面设计应为
   60003"）。
This commit is contained in:
xsl
2026-08-14 16:27:21 +08:00
parent 9e7a886863
commit a4d59aca30
8 changed files with 617 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/integration/test_vertical_slice.py
+97 -23
View File
@@ -12,6 +12,8 @@ from __future__ import annotations
import os
import uuid
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from decimal import Decimal
from typing import Any
@@ -31,40 +33,48 @@ pytestmark = [
]
@pytest.fixture
async def client() -> Any:
class FakeProvider:
"""替换真实 provider：集成用例验证配额与计量链路，不该花钱也不该受网络影响。"""
name = "deepseek"
default_model = "deepseek-v4-flash"
async def complete(self, request: Any) -> Any:
from packages.llm.types import ProviderResponse, Usage
return ProviderResponse(
text="建议白衬衫配西裤。",
model="deepseek-v4-flash",
usage=Usage(tokens_in=800, tokens_out=120, tokens_cached=600),
)
async def aclose(self) -> None:
return None
@asynccontextmanager
async def fake_client(app: Any) -> AsyncIterator[Any]:
"""起 app lifespan + ASGI 客户端，并把 deepseek 换成 FakeProvider。"""
import httpx
from apps.api.main import create_app
from packages.llm.types import ProviderResponse, Usage
from packages.shared.config import get_settings
app = create_app(get_settings())
class FakeProvider:
name = "deepseek"
default_model = "deepseek-v4-flash"
async def complete(self, request: Any) -> ProviderResponse:
return ProviderResponse(
text="建议白衬衫配西裤。",
model="deepseek-v4-flash",
usage=Usage(tokens_in=800, tokens_out=120, tokens_cached=600),
)
async def aclose(self) -> None:
return None
transport = httpx.ASGITransport(app=app)
async with (
httpx.AsyncClient(transport=transport, base_url="http://test") as http,
app.router.lifespan_context(app),
):
# 替换掉真实 provider：集成用例验证配额与计量链路，不该花钱也不该受网络影响
app.state.llm_gateway._providers["deepseek"] = FakeProvider()
yield http
@pytest.fixture
async def client() -> Any:
from apps.api.main import create_app
from packages.shared.config import get_settings
async with fake_client(create_app(get_settings())) as http:
yield http
async def test_chat_slice_reserves_commits_and_records_cost(client: Any) -> None:
token_resp = await client.post(
"/api/v1/auth/dev-token", json={"device_key": f"it-{uuid.uuid4().hex[:8]}"}
@@ -99,3 +109,67 @@ async def test_unauthenticated_requests_get_the_unified_envelope(client: Any) ->
resp = await client.post("/api/v1/agent/chat", json={"message": "hi"})
assert resp.status_code == 401
assert resp.json()["code"] == 40100
async def test_block_all_budget_rejects_and_refunds_the_reservation(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""T-P1-25b：reserve 之后才触发熔断 → gateway 拒绝 → refund → 净扣减 = 0。
这是四档降级（架构 §13.3）的验收路径：把 block_all 档的 allow 名单清空
（连纯文本对话也拦），让 agent_chat 走 100% 档的既有 60003 语义，从而
断言配额三段式在「先预扣、后熔断」时的补偿行为。
"""
import json
from apps.api.main import create_app
from packages.llm.budget import DailyBudgetGuard, InMemoryCostCounter
from packages.shared.config import get_settings
from packages.shared.redis_client import get_redis
monkeypatch.setenv("LLM_DAILY_BUDGET_CENTS", "1") # 预算 1 分钱
monkeypatch.setenv(
"LLM_BUDGET_TIERS_JSON",
json.dumps(
[
{"threshold": 0.6, "name": "warn", "blocked": []},
{"threshold": 0.8, "name": "degrade_secondary", "blocked": ["hd_export"]},
{"threshold": 0.9, "name": "degrade_generation", "blocked": ["hairstyle"]},
# allow 清空：agent_chat 也被 block_all 拦下（演练配置）
{"threshold": 1.0, "name": "block_all", "blocked": "*", "allow": []},
]
),
)
get_settings.cache_clear()
try:
app = create_app(get_settings())
async with fake_client(app) as http:
redis = get_redis()
# 注入当日成本到预算以上（键格式照 DailyBudgetGuard.current_key()）
key_helper = DailyBudgetGuard(InMemoryCostCounter(), daily_budget_cents=Decimal("1"))
cost_key = key_helper.current_key()
await redis.incrbyfloat(cost_key, 5)
token_resp = await http.post(
"/api/v1/auth/dev-token", json={"device_key": f"it-{uuid.uuid4().hex[:8]}"}
)
headers = {"Authorization": f"Bearer {token_resp.json()['data']['access_token']}"}
resp = await http.post(
"/api/v1/agent/chat", json={"message": "面试穿什么"}, headers=headers
)
assert resp.json()["code"] == 60003 # LLM_DAILY_BUDGET_EXCEEDED
# 被拒请求的配额净扣减 = 0：consume(-1, refunded) + refund(+1, committed)
rows = (await http.get("/api/v1/quota/ledger", headers=headers)).json()["data"]
consume = [r for r in rows if r["tx_type"] == "consume"]
refunds = [r for r in rows if r["tx_type"] == "refund"]
assert len(consume) == 1 and consume[0]["status"] == "refunded"
assert len(refunds) == 1 and refunds[0]["delta"] == 1
assert sum(r["delta"] for r in (*consume, *refunds)) == 0
await redis.delete(cost_key)
finally:
# 恢复全局 settings 缓存，避免本用例的环境覆盖渗进其它用例
monkeypatch.undo()
get_settings.cache_clear()
tests/unit/test_daily_budget.py
+4 -2
View File
@@ -37,9 +37,11 @@ async def test_allows_calls_below_the_threshold() -> None:
async def test_rejects_non_critical_scenes_once_the_budget_is_reached() -> None:
guard, _ = make_guard("100")
await guard.add_cost(Decimal("100"))
# block_all 档默认放行 agent_chat（「仅保留纯文本对话」），用非 allow 场景
# 断言 100% 档的既有拒绝语义不变
with pytest.raises(DailyBudgetExceededError) as excinfo:
await guard.ensure_within_budget(scene="agent_chat")
assert "agent_chat" in str(excinfo.value)
await guard.ensure_within_budget(scene="tryon")
assert "tryon" in str(excinfo.value)
async def test_critical_scenes_bypass_the_breaker() -> None:
tests/unit/test_gateway.py
+4 -1
View File
@@ -437,6 +437,7 @@ async def test_spend_accumulates_into_the_daily_budget(prompts_dir: Path) -> Non
async def test_daily_budget_rejects_non_critical_scenes_before_spending(
prompts_dir: Path,
) -> None:
# block_all 档默认放行 agent_chat（纯文本对话），用非 allow 场景断言拒绝
primary = FakeProvider("p1", [response("ok")])
gateway, _, guard = build_gateway(
prompts_dir, {"p1": primary}, backup=None, budget_cents="0.00001"
@@ -444,7 +445,9 @@ async def test_daily_budget_rejects_non_critical_scenes_before_spending(
await guard.add_cost(Decimal("1"))
with pytest.raises(DailyBudgetExceededError):
await gateway.complete(scene=Scene.AGENT_CHAT, prompt_vars={"user_input": "hi"})
await gateway.complete(
scene=Scene.REPORT_COPY, prompt_vars={"payload": "诊断结论与规则命中"}
)
assert primary.requests == [] # rejected, not merely alerted
tests/unit/test_tiered_budget.py
Executable
+245
View File
@@ -0,0 +1,245 @@
"""四档分级降级（架构 §13.3 / T-P1-25b）的判定逻辑单测。
不连网不连库：计数用 InMemoryCostCounter，时间用注入 clock。
档位语义：阈值相对 LLM_DAILY_BUDGET_CENTS；高档位**追加**拒绝低档位名单；
critical 全档豁免；手动 override 优先于成本水位。
"""
from __future__ import annotations
from datetime import UTC, datetime
from decimal import Decimal
import pytest
from packages.llm.budget import (
BLOCK_ALL,
OVERRIDE_KEY,
BudgetTier,
DailyBudgetGuard,
InMemoryCostCounter,
)
from packages.llm.errors import DailyBudgetExceededError, SceneDegradedError
def make_guard(
budget: str = "100",
*,
enabled: bool = True,
tiers: list[dict[str, object]] | None = None,
) -> tuple[DailyBudgetGuard, InMemoryCostCounter]:
counter = InMemoryCostCounter()
guard = DailyBudgetGuard(
counter,
daily_budget_cents=Decimal(budget),
enabled=enabled,
clock=lambda: datetime(2026, 8, 14, 3, 30, tzinfo=UTC), # UTC+8 已是 8-14
tiers=tiers,
)
return guard, counter
async def guard_spent(spent: str) -> DailyBudgetGuard:
guard, _ = make_guard()
await guard.add_cost(Decimal(spent))
return guard
# --------------------------------------------------------------------------
# 档位边界
# --------------------------------------------------------------------------
@pytest.mark.parametrize(
("spent", "expected_tier"),
[
("59.9", 0),
("60", 1),
("79.9", 1),
("80", 2),
("89.9", 2),
("90", 3),
("99.9", 3),
("100", 4),
],
)
async def test_tier_boundaries_follow_configured_thresholds(spent: str, expected_tier: int) -> None:
assert await (await guard_spent(spent)).current_tier() == expected_tier
async def test_tier_names_and_cumulative_blocked_scenes() -> None:
guard = await guard_spent("60")
assert await guard.tier_name() == "warn"
assert await guard.blocked_scenes() == frozenset() # warn 档无用户可感知变化
await guard.add_cost(Decimal("20")) # 80%
assert await guard.tier_name() == "degrade_secondary"
assert await guard.blocked_scenes() == frozenset({"hd_export", "report_regen"})
await guard.add_cost(Decimal("10")) # 90%：追加拒绝，而不是替换名单
assert await guard.tier_name() == "degrade_generation"
assert await guard.blocked_scenes() == frozenset(
{"hd_export", "report_regen", "hairstyle", "tryon"}
)
await guard.add_cost(Decimal("10")) # 100%
assert await guard.tier_name() == "block_all"
assert await guard.blocked_scenes() == BLOCK_ALL
# --------------------------------------------------------------------------
# 各档放行 / 拒绝
# --------------------------------------------------------------------------
async def test_warn_tier_blocks_nothing() -> None:
guard = await guard_spent("60")
for scene in ("agent_chat", "hd_export", "tryon"):
await guard.ensure_scene_allowed(scene=scene)
async def test_tier2_blocks_secondary_scenes_but_not_plain_chat() -> None:
guard = await guard_spent("80")
with pytest.raises(SceneDegradedError) as excinfo:
await guard.ensure_scene_allowed(scene="hd_export")
assert excinfo.value.scene == "hd_export"
assert excinfo.value.tier_name == "degrade_secondary"
await guard.ensure_scene_allowed(scene="agent_chat")
async def test_tier3_keeps_tier2_blocks_and_adds_generation_scenes() -> None:
guard = await guard_spent("90")
for scene in ("hd_export", "hairstyle", "tryon"):
with pytest.raises(SceneDegradedError):
await guard.ensure_scene_allowed(scene=scene)
await guard.ensure_scene_allowed(scene="agent_chat")
async def test_tier4_blocks_all_except_the_allow_list() -> None:
guard = await guard_spent("100")
with pytest.raises(DailyBudgetExceededError) as excinfo:
await guard.ensure_scene_allowed(scene="tryon")
assert excinfo.value.scene == "tryon"
# 架构原文「仅保留在途任务与纯文本对话」：agent_chat 默认在 allow 名单
await guard.ensure_scene_allowed(scene="agent_chat")
async def test_tier4_allow_list_is_configurable() -> None:
tiers: list[dict[str, object]] = [
# 裸字符串 "*" 是合法配置写法，用于演练「连纯文本对话也拦」
{"threshold": 1.0, "name": "block_all", "blocked": BLOCK_ALL, "allow": []},
]
guard, _ = make_guard(tiers=tiers)
await guard.add_cost(Decimal("100"))
assert await guard.current_tier() == 1 # 只有一档时 override/档位序号随之缩短
with pytest.raises(DailyBudgetExceededError):
await guard.ensure_scene_allowed(scene="agent_chat")
async def test_critical_scenes_are_exempt_from_every_tier() -> None:
guard = await guard_spent("500")
# 意图识别把守越界拒答：降级它等于让超范围提问漏过去，因此全档豁免。
await guard.ensure_scene_allowed(scene="intent_classify", critical=True)
async def test_disabled_guard_reports_no_tier() -> None:
guard, _ = make_guard(enabled=False)
await guard.add_cost(Decimal("999"))
assert await guard.current_tier() == 0
await guard.ensure_scene_allowed(scene="tryon")
async def test_legacy_entry_point_still_enforces() -> None:
"""ensure_within_budget 兼容入口：100% 档非 allow 场景仍抛既有异常。"""
guard = await guard_spent("100")
with pytest.raises(DailyBudgetExceededError):
await guard.ensure_within_budget(scene="tryon")
# --------------------------------------------------------------------------
# 手动 override
# --------------------------------------------------------------------------
async def test_override_forces_a_tier_above_the_water_level() -> None:
guard = await guard_spent("10") # 水位仅 10%，本未触发任何档
await guard.set_override(4)
assert await guard.current_tier() == 4
with pytest.raises(DailyBudgetExceededError):
await guard.ensure_scene_allowed(scene="tryon")
await guard.clear_override()
assert await guard.current_tier() == 0 # 清除后回落水位判定
await guard.ensure_scene_allowed(scene="tryon")
async def test_override_can_loosen_a_tier_the_water_level_reached() -> None:
"""运营解除加严：水位已到 tier4，手动放宽回 tier2。"""
guard = await guard_spent("100")
await guard.set_override(2)
assert await guard.current_tier() == 2
with pytest.raises(SceneDegradedError):
await guard.ensure_scene_allowed(scene="hd_export")
await guard.ensure_scene_allowed(scene="tryon")
@pytest.mark.parametrize("bad_value", ["0", "5", "abc", ""])
async def test_invalid_override_values_are_ignored(bad_value: str) -> None:
guard, counter = make_guard()
await guard.add_cost(Decimal("100")) # 水位 tier4
await counter.set(OVERRIDE_KEY, bad_value, 60)
assert await guard.current_tier() == 4 # 回落水位判定，override 不生效
async def test_set_override_rejects_out_of_range_tiers() -> None:
guard, _ = make_guard()
with pytest.raises(ValueError, match=r"1\.\.4"):
await guard.set_override(5)
with pytest.raises(ValueError, match=r"1\.\.4"):
await guard.set_override(0)
async def test_override_ttl_expires_at_the_utc8_day_boundary() -> None:
"""override TTL 到当日 UTC+8 24 点：跨预算日必然失效，不需要人工清理。"""
set_expiries: dict[str, int] = {}
counter = InMemoryCostCounter()
real_set = counter.set
async def recording_set(key: str, value: str, ttl_seconds: int) -> None:
set_expiries[key] = ttl_seconds
await real_set(key, value, ttl_seconds)
counter.set = recording_set # type: ignore[method-assign]
# 8-14 11:30 UTC+8 → 24:00 还有 12.5 小时
guard = DailyBudgetGuard(
counter,
daily_budget_cents=Decimal("100"),
clock=lambda: datetime(2026, 8, 14, 3, 30, tzinfo=UTC),
)
await guard.set_override(4)
assert set_expiries[OVERRIDE_KEY] == 12 * 3600 + 1800
# --------------------------------------------------------------------------
# 配置解析
# --------------------------------------------------------------------------
def test_budget_tier_normalizes_bare_star_and_defaults() -> None:
tier = BudgetTier.model_validate({"threshold": 1.0, "name": "block_all", "blocked": "*"})
assert tier.blocked == [BLOCK_ALL]
assert tier.blocks_all() is True
assert tier.allow == [] # allow 未配置时默认空：block_all 档将拒绝全部非 critical
assert BudgetTier.model_validate({"threshold": 0.6, "name": "warn", "blocked": []}).allow == []
async def test_custom_thresholds_drive_the_tiers() -> None:
tiers: list[dict[str, object]] = [
{"threshold": 0.5, "name": "warn", "blocked": []},
{"threshold": 0.75, "name": "degrade_secondary", "blocked": ["hd_export"]},
{"threshold": 1.0, "name": "block_all", "blocked": BLOCK_ALL, "allow": ["agent_chat"]},
]
guard, _ = make_guard(tiers=tiers)
await guard.add_cost(Decimal("75"))
assert await guard.current_tier() == 2
with pytest.raises(SceneDegradedError):
await guard.ensure_scene_allowed(scene="hd_export")