diff --git a/src/llm/agents/orchestrator.py b/src/llm/agents/orchestrator.py index 35fc16d..f54621b 100644 --- a/src/llm/agents/orchestrator.py +++ b/src/llm/agents/orchestrator.py @@ -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"] diff --git a/tests/test_agent_weights_persist.py b/tests/test_agent_weights_persist.py index 2fe3163..e2a11cc 100644 --- a/tests/test_agent_weights_persist.py +++ b/tests/test_agent_weights_persist.py @@ -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): diff --git a/tests/test_multi_agent_cutoff.py b/tests/test_multi_agent_cutoff.py index 920b511..e03100a 100644 --- a/tests/test_multi_agent_cutoff.py +++ b/tests/test_multi_agent_cutoff.py @@ -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 [] diff --git a/tests/test_multi_agent_degraded.py b/tests/test_multi_agent_degraded.py index 9528be8..94a0a25 100644 --- a/tests/test_multi_agent_degraded.py +++ b/tests/test_multi_agent_degraded.py @@ -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): diff --git a/tests/test_review_required_fixes.py b/tests/test_review_required_fixes.py index 4eaa07f..e070e45 100644 --- a/tests/test_review_required_fixes.py +++ b/tests/test_review_required_fixes.py @@ -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 可能又走回隐式中转,并发下会串味" + ) # ============================================================