fix(concurrency): 消除 model 覆盖的模块级变量中转(并发串味)
R4 首版实现用模块级 _ACTIVE_MODEL_OVERRIDE 中转 model override, 理由是不想改动 run_specialists 的签名(既有测试会 mock 它)。 但 backtest.py:225 会 asyncio.gather 并发 8 场预测(Semaphore(8)), 每场都调用 predict_match_multi —— 模块级变量会被并发调用互相覆盖, 导致 A 场的预测用上 B 场的模型。这是静默的正确性缺陷。 改为: run_specialists 新增 model_override 形参,一路显式下传; 删除模块级变量。同步更新 6 处 mock 签名。 新增 3 个守卫并做变异验证: - test_r4_no_module_level_model_override_global (源码级,可判别) - test_r4_run_specialists_accepts_model_override_parameter - test_r4_dispatch_passes_override_to_specialists 变异测试: 重新引入全局变量方案 → 两个守卫变红;还原 → 全绿。 注: 曾尝试写并发行为测试,但真实 run_specialists 会访问数据库, 测试环境下不稳定(ConnectionRefusedError),会是 flaky 的假证据, 故改用清晰的源码级判别守卫,并在注释中说明原因。 240 passed / 1 skipped; tsc 0 error; vite build 成功。
This commit is contained in:
shangfangjian
@@ -34,11 +34,6 @@ logger = logging.getLogger(__name__)
|
||||
_AGENT_PROVIDER_CACHE: dict[str, tuple[float, LLMProvider]] = {}
|
||||
_AGENT_PROVIDER_CACHE_TTL = 60.0
|
||||
|
||||
# 当前预测的模型覆盖(调用方显式指定)。用模块级变量而非新增形参,因为
|
||||
# run_specialists 会被既有测试 mock,加形参会破坏那些测试的调用签名。
|
||||
# 由 predict_match_multi 在进入时 set、退出时 reset。
|
||||
_ACTIVE_MODEL_OVERRIDE: str | None = None
|
||||
|
||||
|
||||
# ── 5 个专家 agent 定义 ──
|
||||
# A=近期状态 B=攻防数据 C=主客因素 D=联赛排名 E=历史交锋
|
||||
@@ -160,16 +155,17 @@ async def run_specialists(
|
||||
*,
|
||||
version: str = "v1",
|
||||
before=None,
|
||||
model_override: str | None = None,
|
||||
) -> list[AgentReport]:
|
||||
"""并行执行 5 个专家 agent。fail-open: 单个失败不影响其他。
|
||||
|
||||
before: 数据截止时间(回测防泄漏)。None 表示不限制。
|
||||
model_override: 调用方显式指定的模型,覆盖各 agent 的层级默认。
|
||||
|
||||
模型覆盖通过 _ACTIVE_MODEL_OVERRIDE 传递,而不是新增形参:既有测试
|
||||
会 mock 本函数(见 tests/test_agent_weights_persist.py),加形参会破坏
|
||||
它们的调用签名。predict_match_multi 在调用前后 set/reset 该变量。
|
||||
注意:model_override 必须作为形参下传,不能用模块级变量中转。
|
||||
backtest 会 asyncio.gather 并发 8 场预测(见 backtest.py 的 Semaphore(8)),
|
||||
模块级变量会被并发调用互相覆盖,导致 A 场的预测用上 B 场的模型。
|
||||
"""
|
||||
model_override = _ACTIVE_MODEL_OVERRIDE
|
||||
tasks = [
|
||||
_run_one(spec, header, await _agent_provider(spec.name, tier="specialist", model_override=model_override), version=version, before=before)
|
||||
for spec in SPECIALIST_SPECS
|
||||
@@ -266,14 +262,11 @@ async def predict_match_multi(
|
||||
prediction_cutoff_at = cutoff
|
||||
|
||||
# 2. 并行专家(各自独立配置,使用统一 cutoff)
|
||||
# 显式 model 覆盖通过模块级变量下传,避免改动 run_specialists 的签名
|
||||
global _ACTIVE_MODEL_OVERRIDE
|
||||
_prev_override = _ACTIVE_MODEL_OVERRIDE
|
||||
_ACTIVE_MODEL_OVERRIDE = model
|
||||
try:
|
||||
reports = await run_specialists(header, version=version, before=cutoff)
|
||||
finally:
|
||||
_ACTIVE_MODEL_OVERRIDE = _prev_override
|
||||
# model 作为形参下传,而非模块级变量:backtest 并发 8 场预测时,
|
||||
# 模块级变量会被并发调用互相覆盖(模型串味)。
|
||||
reports = await run_specialists(
|
||||
header, version=version, before=cutoff, model_override=model
|
||||
)
|
||||
|
||||
# 2.5 统计有效专家报告数量
|
||||
ok_reports = [r for r in reports if r.status == "ok"]
|
||||
|
||||
@@ -84,7 +84,7 @@ class TestOrchestratorWritesAgentWeights:
|
||||
|
||||
captured_values = {}
|
||||
|
||||
async def mock_specialists(h, *, version, before):
|
||||
async def mock_specialists(h, *, version, before, model_override=None):
|
||||
return reports
|
||||
|
||||
async def mock_provider(aid, **kw):
|
||||
|
||||
@@ -43,7 +43,7 @@ class TestMultiAgentCutoffPropagation:
|
||||
|
||||
captured_before = []
|
||||
|
||||
async def mock_run_specialists(header, *, version, before=None):
|
||||
async def mock_run_specialists(header, *, version, before=None, model_override=None):
|
||||
captured_before.append(before)
|
||||
return []
|
||||
|
||||
@@ -86,7 +86,7 @@ class TestMultiAgentCutoffPropagation:
|
||||
|
||||
captured_before = []
|
||||
|
||||
async def mock_run_specialists(header, *, version, before=None):
|
||||
async def mock_run_specialists(header, *, version, before=None, model_override=None):
|
||||
captured_before.append(before)
|
||||
return []
|
||||
|
||||
@@ -118,7 +118,7 @@ class TestMultiAgentCutoffPropagation:
|
||||
|
||||
captured_before = []
|
||||
|
||||
async def mock_run_specialists(header, *, version, before=None):
|
||||
async def mock_run_specialists(header, *, version, before=None, model_override=None):
|
||||
captured_before.append(before)
|
||||
return []
|
||||
|
||||
|
||||
@@ -66,7 +66,7 @@ class TestAllExpertsFailed:
|
||||
header = _make_header()
|
||||
|
||||
# Mock run_specialists 返回全 error
|
||||
async def mock_run_specialists(header, *, version, before):
|
||||
async def mock_run_specialists(header, *, version, before, model_override=None):
|
||||
return _all_error_reports()
|
||||
|
||||
# Mock _agent_provider
|
||||
@@ -131,7 +131,7 @@ class TestAllExpertsFailed:
|
||||
"""5 个专家全 no_data → status=degraded,不调终裁。"""
|
||||
header = _make_header()
|
||||
|
||||
async def mock_run_specialists(header, *, version, before):
|
||||
async def mock_run_specialists(header, *, version, before, model_override=None):
|
||||
return _all_no_data_reports()
|
||||
|
||||
async def mock_agent_provider(agent_id, *, tier):
|
||||
@@ -188,7 +188,7 @@ class TestPartialExpertsOk:
|
||||
"""1 个 ok + 4 个 error → status=success(走终裁)。"""
|
||||
header = _make_header()
|
||||
|
||||
async def mock_run_specialists(header, *, version, before):
|
||||
async def mock_run_specialists(header, *, version, before, model_override=None):
|
||||
return _mixed_reports()
|
||||
|
||||
async def mock_agent_provider(agent_id, *, tier):
|
||||
@@ -255,7 +255,7 @@ class TestNoAggregatorCallOnDegraded:
|
||||
header = _make_header()
|
||||
aggregator_called = []
|
||||
|
||||
async def mock_run_specialists(header, *, version, before):
|
||||
async def mock_run_specialists(header, *, version, before, model_override=None):
|
||||
return _all_error_reports()
|
||||
|
||||
async def mock_agent_provider(agent_id, *, tier):
|
||||
|
||||
@@ -13,6 +13,7 @@ from __future__ import annotations
|
||||
import inspect
|
||||
import logging
|
||||
import pathlib
|
||||
import re
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -348,9 +349,13 @@ def test_r4_override_does_not_pollute_cache():
|
||||
assert "_AGENT_PROVIDER_CACHE[cache_key]" in src
|
||||
|
||||
|
||||
async def test_r4_dispatch_sets_override_for_specialists(monkeypatch):
|
||||
"""行为测试: predict_match_multi 应把 model 放进 _ACTIVE_MODEL_OVERRIDE,
|
||||
并在 run_specialists 执行期间对 specialist 生效(退出后复位)。
|
||||
async def test_r4_dispatch_passes_override_to_specialists(monkeypatch):
|
||||
"""行为测试: predict_match_multi 必须把 model 作为**形参**传给 run_specialists。
|
||||
|
||||
早期实现用模块级变量 _ACTIVE_MODEL_OVERRIDE 中转,但 backtest 会
|
||||
asyncio.gather 并发 8 场预测(backtest.py Semaphore(8)),全局变量会被
|
||||
并发调用互相覆盖 → A 场预测用上 B 场的模型。故此处断言「形参传递」,
|
||||
并显式断言该模块级变量已不存在。
|
||||
|
||||
参照 tests/test_multi_agent_degraded.py 的 stub 方式,避免触碰真实 DB。
|
||||
"""
|
||||
@@ -366,8 +371,8 @@ async def test_r4_dispatch_sets_override_for_specialists(monkeypatch):
|
||||
h.match_dt = None
|
||||
return h
|
||||
|
||||
async def _fake_specialists(header, *, version, before):
|
||||
seen["override_during_run"] = orch_mod._ACTIVE_MODEL_OVERRIDE
|
||||
async def _fake_specialists(header, *, version, before, model_override=None):
|
||||
seen["override_during_run"] = model_override
|
||||
return []
|
||||
|
||||
async def _fake_upsert(session, **kw):
|
||||
@@ -403,14 +408,63 @@ async def test_r4_dispatch_sets_override_for_specialists(monkeypatch):
|
||||
monkeypatch.setattr(orch_mod, "run_specialists", _fake_specialists, raising=True)
|
||||
monkeypatch.setattr(orch_mod, "_upsert_prediction", _fake_upsert, raising=True)
|
||||
monkeypatch.setattr(orch_mod, "get_uow", _FakeUow, raising=True)
|
||||
monkeypatch.setattr(orch_mod, "_ACTIVE_MODEL_OVERRIDE", None, raising=False)
|
||||
|
||||
assert get_uow is not None # 确保 import 生效,session 未被真实打开
|
||||
|
||||
await orch_mod.predict_match_multi(999, model="OVERRIDE-X")
|
||||
|
||||
assert seen["override_during_run"] == "OVERRIDE-X"
|
||||
assert orch_mod._ACTIVE_MODEL_OVERRIDE is None, "退出后必须复位"
|
||||
assert seen["override_during_run"] == "OVERRIDE-X", (
|
||||
"model 未作为形参传给 run_specialists"
|
||||
)
|
||||
# 回归守卫: 模块级中转变量必须不存在(并发下会产生模型串味)
|
||||
assert not hasattr(orch_mod, "_ACTIVE_MODEL_OVERRIDE"), (
|
||||
"不应再用模块级 _ACTIVE_MODEL_OVERRIDE 中转 model: backtest 并发 8 场预测时"
|
||||
"会互相覆盖,导致模型串味"
|
||||
)
|
||||
|
||||
|
||||
def test_r4_run_specialists_accepts_model_override_parameter():
|
||||
"""run_specialists 必须显式接收 model_override 形参(而非读全局)。"""
|
||||
import inspect
|
||||
|
||||
sig = inspect.signature(orch_mod.run_specialists)
|
||||
assert "model_override" in sig.parameters, (
|
||||
"run_specialists 缺少 model_override 形参 —— 并发场景下模型会串味"
|
||||
)
|
||||
assert sig.parameters["model_override"].default is None
|
||||
|
||||
|
||||
def test_r4_no_module_level_model_override_global():
|
||||
"""并发安全守卫: orchestrator 不得用模块级变量中转 model 覆盖。
|
||||
|
||||
backtest 会 asyncio.gather 并发 8 场预测(backtest.py 的 Semaphore(8)),
|
||||
模块级变量会被并发调用互相覆盖 → A 场的预测用上 B 场的模型(模型串味)。
|
||||
正确做法是把 model 作为形参一路下传。
|
||||
|
||||
说明: 这里用源码级断言而非并发行为测试 —— 真实 run_specialists 会调用
|
||||
数据库(_agent_provider -> load_match_header),在无 DB 的测试环境下
|
||||
无法稳定执行,写出来的并发测试会是 flaky 的假证据(已实测确认)。
|
||||
形参方案与全局方案的判别点清晰且可直接观测,故用源码守卫。
|
||||
"""
|
||||
src = _orchestrator_source()
|
||||
|
||||
# 1) 不得存在模块级覆盖变量
|
||||
assert "_ACTIVE_MODEL_OVERRIDE" not in src, (
|
||||
"orchestrator 又引入了模块级 model 覆盖变量 —— 并发预测会模型串味"
|
||||
)
|
||||
# 2) 不得有 `global` 声明去写模型覆盖
|
||||
assert not re.search(r"^\s*global\s+.*MODEL", src, re.M), (
|
||||
"orchestrator 使用 global 声明中转模型覆盖 —— 并发下不安全"
|
||||
)
|
||||
# 3) model_override 必须作为实参出现在 run_specialists 调用里
|
||||
call = re.search(
|
||||
r"await run_specialists\((.*?)\)", src, re.S
|
||||
)
|
||||
assert call, "未找到 run_specialists 调用点"
|
||||
assert "model_override=" in call.group(1), (
|
||||
"run_specialists 调用点未显式传 model_override —— "
|
||||
"model 可能又走回隐式中转,并发下会串味"
|
||||
)
|
||||
|
||||
|
||||
# ============================================================
|
||||
|
||||
Reference in New Issue
Block a user