xsl
f46ecc5eb0
Two defects that only surface against the real DeepSeek API, both of which break the cost model rather than the code: deepseek-v4-flash enables thinking mode by default. The same single-turn question took 13.5s with it on and 1.8s with it off, and the thinking tokens are billed at the output rate. That misses NFR-PERF-02b (15s per turn) and inflates every conversational call. SceneConfig now carries an explicit `thinking` flag, all four V1 scenes set it to false, and the provider always sends the parameter rather than relying on the upstream default. Reasoning tokens are recorded separately in Usage.tokens_reasoning so the cost of ever turning it back on is visible. DeepSeek moves to peak/off-peak pricing on 2026-08-17, with peak (Beijing 09:00-12:00 and 14:00-18:00) at twice the off-peak rate. Reports are generated mostly during peak hours, so a single blended rate would leave ob_ai_call -- the only input to V2 pricing -- off by nearly a factor of two. PeakAwareRateTable resolves the rate at call time; cost is computed and stored then, never recomputed later at whatever the current rate happens to be. Also here: - Replace the live API key in the redaction test with a synthetic one. The test that proves we redact secrets was itself a secret leak waiting for its first commit. - Correct LLM_COST_RATES_JSON, which held placeholder rates. - Make ids.SnowflakeGenerator take an injectable clock so rollover and clock-rollback are testable without patching a slots class. - Rename exceptions to the -Error suffix (N818) and clear the remaining ruff, mypy and import-linter findings, including two dead lines: a duplicate pytestmark assignment and a no-op assert on a None return. Gates: ruff, mypy (96 files incl. tests), 4/4 import contracts, detect-secrets, 166 unit + 3 integration tests, migration up/down round-trip.
80 lines
2.7 KiB
Python
80 lines
2.7 KiB
Python
"""思考模式必须显式声明（deepseek-v4 默认开启）。
|
|
|
|
保护的是两件事：漏传 ``thinking`` 会让对话延迟从约 2s 涨到 13s 以上（实测同一个单句
|
|
问答），并且思考 token 按输出价计费 —— 一个用户看不到的字段同时打破响应时间目标
|
|
与成本模型。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from packages.llm.providers.deepseek import DeepSeekProvider
|
|
from packages.llm.routing import build_scene_configs
|
|
from packages.llm.types import Message, ProviderRequest, Scene
|
|
|
|
|
|
def _request(*, thinking: bool) -> ProviderRequest:
|
|
return ProviderRequest(
|
|
model="deepseek-v4-flash",
|
|
messages=(Message(role="user", content="面试穿什么"),),
|
|
temperature=0.7,
|
|
max_tokens=800,
|
|
timeout_seconds=30.0,
|
|
thinking=thinking,
|
|
)
|
|
|
|
|
|
def _provider(captured: dict[str, Any]) -> DeepSeekProvider:
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
captured.update(json.loads(request.content))
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"choices": [{"message": {"content": "ok"}, "finish_reason": "stop"}],
|
|
"usage": {
|
|
"prompt_tokens": 12,
|
|
"completion_tokens": 34,
|
|
"completion_tokens_details": {"reasoning_tokens": 20},
|
|
},
|
|
},
|
|
)
|
|
|
|
return DeepSeekProvider(
|
|
api_key="test-key",
|
|
client=httpx.AsyncClient(transport=httpx.MockTransport(handler)),
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(("thinking", "expected"), [(False, "disabled"), (True, "enabled")])
|
|
async def test_thinking_is_always_sent_explicitly(thinking: bool, expected: str) -> None:
|
|
captured: dict[str, Any] = {}
|
|
provider = _provider(captured)
|
|
|
|
await provider.complete(_request(thinking=thinking))
|
|
|
|
# 关键是这个字段一定存在：省略它就等于接受供应商的"开启"默认值
|
|
assert captured["thinking"] == {"type": expected}
|
|
await provider.aclose()
|
|
|
|
|
|
async def test_reasoning_tokens_are_recorded_for_cost_visibility() -> None:
|
|
provider = _provider({})
|
|
response = await provider.complete(_request(thinking=True))
|
|
|
|
# 思考 token 已计入 tokens_out，单独留一份用于回答"输出成本有多少花在推理上"
|
|
assert response.usage.tokens_out == 34
|
|
assert response.usage.tokens_reasoning == 20
|
|
await provider.aclose()
|
|
|
|
|
|
def test_no_v1_scene_enables_thinking_by_default() -> None:
|
|
configs = build_scene_configs(primary_provider="deepseek")
|
|
assert [s for s, c in configs.items() if c.thinking] == []
|
|
# 且每个场景都显式给了值，不存在"忘了配"的场景
|
|
assert set(configs) == set(Scene)
|