"""多 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, "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.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, "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 "confidence": None, "key_evidence": "单字符串", # → [str] } resp = LLMResponse(content="{}", parsed=parsed) r = _parse_report("form", parsed, resp, "m") assert r.data_sufficiency == "medium" assert r.home_edge is None assert r.key_evidence == ["单字符串"] 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, "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, 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