WorkBuddy
c89bfe2af7
P2-1 no_data 门控依赖文案子串(脆弱): - context_builder 新增 SliceResult(text/has_data/n_records), 5 个切片函数改为显式声明 has_data - base._slice_has_data() 优先取结构化结果,str 返回仍走文案回退 (兼容既有测试 mock 与自定义切片) - build_context 的 has_stats/has_injuries 直接取切片声明 P2-2 agent_weights 无校验即落库: - validation 新增 AgentWeightsSchema / validate_agent_weights: 未知专家名丢弃、越界值钳制、总和非 1 时归一化 - orchestrator 落库前对 agent_weights 做校验 P2-3 1x2 与比分不一致被静默修正: - 仍以比分修正,但补 logger.warning 暴露 LLM 自相矛盾 P2-5/P2-6 prompt 缓存不可刷新 + 缓存键不含模板内容: - 新增 clear_prompt_cache() 供改模板后显式失效 - 缓存键纳入模板内容 hash,模板一改缓存自动失效 P2-7 ingest/backtest/settle 接口无鉴权: - 新增 require_admin_key 依赖(X-API-Key), ADMIN_API_KEY 未设置时放行并告警(不破坏本地开发) - 挂到 3 个 ingest 接口 + backtest + eval/settle P2-8 前端请求竞态 + 未使用游标分页: - Matches.tsx 用递增 seq 丢弃过期响应,避免旧筛选结果覆盖新筛选 - 接入后端已有的 cursor 分页 + 「加载更多」按钮 附带: .env.example 补齐 LLM_TIMEOUT / 分档模型 / ADMIN_API_KEY; tests 新增 10 个用例覆盖 P2-1/2/3。
319 lines
12 KiB
Python
319 lines
12 KiB
Python
"""多 agent 架构测试。"""
|
|
import pytest
|
|
from src.llm.agents.base import AgentReport, load_agent_prompt, _is_no_data
|
|
from src.llm.context_builder import MatchHeader
|
|
|
|
|
|
class TestNoDataGate:
|
|
"""no_data 门控: 切片无数据 → 跳过 LLM。"""
|
|
|
|
def test_is_no_data_all_lines(self):
|
|
assert _is_no_data("── 伤停 ──\n 无数据") is True
|
|
|
|
def test_is_no_data_with_content(self):
|
|
assert _is_no_data("── 交锋 ──\n 2026-03: A 2-1 B") is False
|
|
|
|
def test_is_no_data_empty(self):
|
|
assert _is_no_data("") is True
|
|
|
|
def test_is_no_data_mixed(self):
|
|
# 部分有数据部分无 → 不是 no_data
|
|
assert _is_no_data("── xG ──\n A: xG 1.5 vs B 0.8\n B: 无 xG 数据") is False
|
|
|
|
def test_stub_no_data_report(self):
|
|
from src.llm.agents.base import _stub_no_data
|
|
r = _stub_no_data("injuries")
|
|
assert r.status == "no_data"
|
|
assert r.data_sufficiency == "none"
|
|
assert r.home_edge is None
|
|
|
|
|
|
class TestPromptLoading:
|
|
"""agent prompt 模板加载。"""
|
|
|
|
@pytest.mark.parametrize("name", ["form", "stats", "home_away", "injuries", "h2h", "aggregator"])
|
|
def test_all_prompts_exist(self, name):
|
|
tpl = load_agent_prompt(name, "v1")
|
|
assert "{{context}}" in tpl or "{{agent_reports}}" in tpl
|
|
|
|
def test_prompt_not_found(self):
|
|
with pytest.raises(FileNotFoundError):
|
|
load_agent_prompt("nonexistent", "v1")
|
|
|
|
|
|
class TestReportParsing:
|
|
"""LLM JSON 输出 → AgentReport 解析(宽容处理)。"""
|
|
|
|
def _make_header(self) -> MatchHeader:
|
|
return MatchHeader(
|
|
match_id=1, home_name="A", away_name="B", league_name="PL",
|
|
season="2026-2027", match_date="2026-09-15", match_dt=None,
|
|
stage=None, home_team_id=10, away_team_id=20, league_id=1,
|
|
)
|
|
|
|
def test_parse_full_report(self):
|
|
from src.llm.agents.base import _parse_report
|
|
from src.llm.provider import LLMResponse
|
|
|
|
parsed = {
|
|
"data_sufficiency": "high",
|
|
"analysis": "主队交锋占优",
|
|
"home_edge": 0.6,
|
|
"subjective_confidence": 0.8,
|
|
"key_evidence": ["近5次交锋主队4胜", "主场交锋3连胜"],
|
|
}
|
|
resp = LLMResponse(content="{}", parsed=parsed, prompt_tokens=100, completion_tokens=50, latency_ms=500)
|
|
r = _parse_report("h2h", parsed, resp, "gpt-4o-mini")
|
|
assert r.status == "ok"
|
|
assert r.home_edge == 0.6
|
|
assert r.subjective_confidence == 0.8
|
|
assert len(r.key_evidence) == 2
|
|
assert r.data_sufficiency == "high"
|
|
|
|
def test_parse_xg_report_with_score(self):
|
|
from src.llm.agents.base import _parse_report
|
|
from src.llm.provider import LLMResponse
|
|
|
|
parsed = {
|
|
"data_sufficiency": "medium",
|
|
"analysis": "主队火力更强",
|
|
"home_edge": 0.4,
|
|
"subjective_confidence": 0.7,
|
|
"key_evidence": ["场均xG 2.1"],
|
|
"exp_home_goals": 2.1,
|
|
"exp_away_goals": 1.2,
|
|
"probable_score": {"home": 2, "away": 1, "prob": 0.14},
|
|
}
|
|
resp = LLMResponse(content="{}", parsed=parsed)
|
|
r = _parse_report("xg", parsed, resp, "gpt-4o-mini")
|
|
assert r.exp_home_goals == 2.1
|
|
assert r.probable_score == "2-1"
|
|
|
|
def test_parse_bad_values_forgiving(self):
|
|
"""非法数值/字段宽容降级,不抛异常。"""
|
|
from src.llm.agents.base import _parse_report
|
|
from src.llm.provider import LLMResponse
|
|
|
|
parsed = {
|
|
"data_sufficiency": "bogus", # 非法 → medium
|
|
"home_edge": "very strong", # 非法 → None (宽容降级)
|
|
"subjective_confidence": None,
|
|
"key_evidence": "单字符串", # → [str]
|
|
}
|
|
resp = LLMResponse(content="{}", parsed=parsed)
|
|
r = _parse_report("form", parsed, resp, "m")
|
|
# 宽容降级: 不抛异常,非法字段变 None 或默认值
|
|
assert r.status == "ok"
|
|
assert r.data_sufficiency == "medium"
|
|
assert r.home_edge is None
|
|
assert r.key_evidence == ["单字符串"]
|
|
|
|
# 验证范围约束: confidence > 1 会被截断或拒绝
|
|
parsed2 = {"subjective_confidence": 1.5, "home_edge": 2.0}
|
|
r2 = _parse_report("form", parsed2, resp, "m")
|
|
# Pydantic 会拒绝越界值 → parse_error
|
|
assert r2.status == "parse_error"
|
|
|
|
|
|
class TestRunAgent:
|
|
"""run_agent 执行器: 门控 + fail-open。"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_data_skips_llm(self):
|
|
"""切片无数据 → 不调 LLM,直接 stub。"""
|
|
from src.llm.agents.base import AgentSpec, run_agent
|
|
|
|
async def empty_slice(header, before=None):
|
|
return "── 伤停 ──\n 无数据"
|
|
|
|
spec = AgentSpec(name="injuries", system_prompt="s", slice_fn=empty_slice)
|
|
header = self._make_header()
|
|
|
|
class ExplodingProvider:
|
|
async def chat(self, *a, **kw):
|
|
raise AssertionError("LLM 不应被调用")
|
|
|
|
r = await run_agent(spec, header, ExplodingProvider(), version="v1")
|
|
assert r.status == "no_data"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_llm_error_fail_open(self):
|
|
"""LLM 调用失败 → status=error,不抛异常。"""
|
|
from src.llm.agents.base import AgentSpec, run_agent
|
|
|
|
async def good_slice(header, before=None):
|
|
return "── 交锋 ──\n 2026-03: A 2-1 B"
|
|
|
|
spec = AgentSpec(name="h2h", system_prompt="s", slice_fn=good_slice)
|
|
header = self._make_header()
|
|
|
|
class FailProvider:
|
|
async def chat(self, *a, **kw):
|
|
from src.llm.provider import LLMResponse
|
|
return LLMResponse(content="", error="timeout")
|
|
|
|
r = await run_agent(spec, header, FailProvider(), version="v1")
|
|
assert r.status == "error"
|
|
assert "LLM 调用失败" in r.analysis
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_success_path(self):
|
|
"""正常路径: 切片 → LLM → 解析。"""
|
|
from src.llm.agents.base import AgentSpec, run_agent
|
|
from src.llm.provider import LLMResponse
|
|
|
|
async def good_slice(header, before=None):
|
|
return "── 交锋 ──\n 2026-03: A 2-1 B"
|
|
|
|
spec = AgentSpec(name="h2h", system_prompt="s", slice_fn=good_slice)
|
|
header = self._make_header()
|
|
|
|
class OkProvider:
|
|
model = "test-model"
|
|
async def chat(self, system, user, **kw):
|
|
assert "{{context}}" not in user # 模板已渲染
|
|
assert "A 2-1 B" in user
|
|
return LLMResponse(
|
|
content="{}",
|
|
parsed={"data_sufficiency": "high", "analysis": "ok", "home_edge": 0.5, "subjective_confidence": 0.9},
|
|
prompt_tokens=10, completion_tokens=5, latency_ms=100,
|
|
)
|
|
|
|
r = await run_agent(spec, header, OkProvider(), version="v1")
|
|
assert r.status == "ok"
|
|
assert r.home_edge == 0.5
|
|
assert r.model == "test-model"
|
|
|
|
def _make_header(self) -> MatchHeader:
|
|
return MatchHeader(
|
|
match_id=1, home_name="A", away_name="B", league_name="PL",
|
|
season="2026-2027", match_date="2026-09-15", match_dt=None,
|
|
stage=None, home_team_id=10, away_team_id=20, league_id=1,
|
|
)
|
|
|
|
|
|
class TestOrchestratorAggregation:
|
|
"""终裁输入拼装逻辑。"""
|
|
|
|
def test_reports_to_json(self):
|
|
from src.llm.agents.orchestrator import _reports_to_json
|
|
import json
|
|
|
|
reports = [
|
|
AgentReport(agent="h2h", status="ok", home_edge=0.5, subjective_confidence=0.8, analysis="a"),
|
|
AgentReport(agent="injuries", status="no_data", data_sufficiency="none"),
|
|
]
|
|
text = _reports_to_json(reports)
|
|
data = json.loads(text)
|
|
assert len(data) == 2
|
|
assert data[0]["agent"] == "h2h"
|
|
assert data[1]["status"] == "no_data"
|
|
|
|
def test_aggregator_prompt_renders(self):
|
|
"""终裁 prompt 模板两占位符都能渲染。"""
|
|
tpl = load_agent_prompt("aggregator", "v1")
|
|
rendered = (
|
|
tpl
|
|
.replace("{{match_header}}", "对阵: A vs B")
|
|
.replace("{{agent_reports}}", '[{"agent": "h2h"}]')
|
|
)
|
|
assert "{{match_header}}" not in rendered
|
|
assert "{{agent_reports}}" not in rendered
|
|
|
|
|
|
class TestSliceResultGate:
|
|
"""P2-1: 结构化 has_data 门控(替代脆弱的文案子串匹配)。"""
|
|
|
|
def test_sliceresult_empty_is_no_data(self):
|
|
from src.llm.agents.base import _slice_has_data
|
|
from src.llm.context_builder import SliceResult
|
|
|
|
text, has = _slice_has_data(SliceResult(text="── x ──\n 无数据", has_data=False))
|
|
assert has is False
|
|
|
|
def test_sliceresult_with_data_beats_text(self):
|
|
"""即使文案里出现「无数据」字样,结构化 has_data=True 也应胜出。
|
|
|
|
这正是旧实现的漏洞:文案匹配会把「主队: 无伤停数据 / 客队: 2人伤停」
|
|
这类混合输出……这里显式验证结构化声明优先。
|
|
"""
|
|
from src.llm.agents.base import _slice_has_data
|
|
from src.llm.context_builder import SliceResult
|
|
|
|
tricky = SliceResult(
|
|
text="── 阵容完整性 ──\n 主队: 无伤停数据\n 客队伤停(1人):\n - X: 拉伤",
|
|
has_data=True,
|
|
)
|
|
_, has = _slice_has_data(tricky)
|
|
assert has is True
|
|
|
|
def test_str_fallback_still_works(self):
|
|
"""旧式 str 切片(测试 mock / 自定义切片)仍走文案回退,保持兼容。"""
|
|
from src.llm.agents.base import _slice_has_data
|
|
|
|
assert _slice_has_data("── 伤停 ──\n 无数据")[1] is False
|
|
assert _slice_has_data("── 交锋 ──\n A 2-1 B")[1] is True
|
|
|
|
def test_sliceresult_str_compat(self):
|
|
"""SliceResult 可当 str 用(老调用点无需改)。"""
|
|
from src.llm.context_builder import SliceResult
|
|
|
|
s = SliceResult(text="hello", has_data=True)
|
|
assert str(s) == "hello"
|
|
|
|
|
|
class TestAgentWeightsValidation:
|
|
"""P2-2: agent_weights 必须过校验才能落库。"""
|
|
|
|
def test_unknown_agent_dropped(self):
|
|
from src.llm.validation import validate_agent_weights
|
|
|
|
w = validate_agent_weights({"form": 0.4, "h2h": 0.4, "bogus": 0.2, "stats": 0.4})
|
|
assert "bogus" not in w
|
|
assert set(w) <= {"form", "stats", "home_away", "injuries", "h2h"}
|
|
|
|
def test_out_of_range_clamped(self):
|
|
from src.llm.validation import validate_agent_weights
|
|
|
|
w = validate_agent_weights({"form": 5.0, "h2h": -1.0})
|
|
assert w["form"] == 1.0
|
|
assert w["h2h"] == 0.0
|
|
|
|
def test_sum_normalized(self):
|
|
from src.llm.validation import validate_agent_weights
|
|
|
|
w = validate_agent_weights({"form": 2.0, "stats": 2.0})
|
|
assert abs(sum(w.values()) - 1.0) < 1e-9
|
|
|
|
def test_no_weights_returns_empty(self):
|
|
from src.llm.validation import validate_agent_weights
|
|
|
|
assert validate_agent_weights(None) == {}
|
|
assert validate_agent_weights("not a dict") == {}
|
|
|
|
|
|
class TestPredictionConsistencyWarn:
|
|
"""P2-3: 比分与 1x2 不一致 → 以比分修正(且告警)。"""
|
|
|
|
def test_mismatch_is_corrected_to_score(self):
|
|
from src.llm.validation import validate_prediction_output
|
|
|
|
v = validate_prediction_output({
|
|
"pred_home_goals": 2.0,
|
|
"pred_away_goals": 1.0,
|
|
"pred_1x2": "X", # 与 2-1 矛盾
|
|
"subjective_confidence": 0.7,
|
|
})
|
|
assert v.pred_1x2 == "1" # 按比分修正
|
|
|
|
def test_consistent_passes_through(self):
|
|
from src.llm.validation import validate_prediction_output
|
|
|
|
v = validate_prediction_output({
|
|
"pred_home_goals": 0.0,
|
|
"pred_away_goals": 0.0,
|
|
"pred_1x2": "X",
|
|
"subjective_confidence": 0.5,
|
|
})
|
|
assert v.pred_1x2 == "X"
|