"""会商「模型自动容错（换下一个模型）」测试。 覆盖： - 中间件兜底/熔断 AIMessage 带可识别标记 + ``is_llm_error_message`` 检测（标记 + 正文回退）； - ``fallback_model_chain`` 按 config.yaml models[] 顺序排候选（当前优先、去重、开关关闭=单个）； - ``reason_text`` 文案； - 引擎 Step2 单席位异常隔离（一个席位抛错不拖垮整链/整批，仅置空交付）； - 网关带回的模型容错记录（switches）→ 引擎在时间线插「系统提示」气泡。 纯单元/纯逻辑：引擎用 JobProgressWriter(store=None) 干跑 + 假网关，不需要 DB / 真模型。 """ from __future__ import annotations import asyncio from types import SimpleNamespace import app.gateway.roundtable_model_fallback as mf from app.gateway.roundtable_model_fallback import ( fallback_model_chain, is_llm_error_message, reason_text, ) from deerflow.agents.middlewares.llm_error_handling_middleware import ( LLM_ERROR_MARKER, LLMErrorHandlingMiddleware, ) from deerflow.agents.middlewares.llm_error_handling_middleware import ( is_llm_error_message as harness_is_llm_error_message, ) from deerflow.agents.roundtable_orchestrator import ( JobProgressWriter, SeatRef, run_orchestration, ) from deerflow.agents.roundtable_orchestrator.types import LeaderTurn, ModelSwitch, SeatTurn # ── is_llm_error_message ───────────────────────────────────────────────────── def test_is_llm_error_message_marker_dict() -> None: msg = {"type": "ai", "content": "随便什么", "additional_kwargs": {LLM_ERROR_MARKER: "busy"}} assert is_llm_error_message(msg) == "busy" assert harness_is_llm_error_message(msg) == "busy" def test_is_llm_error_message_content_fallback() -> None: # additional_kwargs 被剥离时，按已知兜底文案正文回退识别。 msg = {"type": "ai", "content": "大模型本次调用未能成功完成，请稍后再试或更换模型。"} assert is_llm_error_message(msg) == "generic" def test_is_llm_error_message_normal_is_none() -> None: assert is_llm_error_message({"type": "ai", "content": "这是正常的席位交付内容"}) is None assert is_llm_error_message(None) is None assert is_llm_error_message({"type": "ai", "content": ""}) is None # ── 中间件给兜底 AIMessage 打标记 ──────────────────────────────────────────── def _middleware() -> LLMErrorHandlingMiddleware: app_config = SimpleNamespace(circuit_breaker=SimpleNamespace(failure_threshold=5, recovery_timeout_sec=30)) return LLMErrorHandlingMiddleware(app_config=app_config) def test_middleware_marks_error_message() -> None: mw = _middleware() def _boom(_req): raise RuntimeError("insufficient_quota: 余额不足") # quota → 非重试 → 立即兜底 out = mw.wrap_model_call(None, _boom) assert harness_is_llm_error_message(out) == "quota" assert out.additional_kwargs.get(LLM_ERROR_MARKER) == "quota" # ── fallback_model_chain ───────────────────────────────────────────────────── def _patch_models(monkeypatch, names: list[str]) -> None: cfg = SimpleNamespace(models=[SimpleNamespace(name=n) for n in names]) monkeypatch.setattr(mf, "get_app_config", lambda: cfg) def test_chain_current_first_then_config_order(monkeypatch) -> None: _patch_models(monkeypatch, ["m1", "m2", "m3"]) monkeypatch.delenv("ROUNDTABLE_MODEL_FALLBACK", raising=False) monkeypatch.delenv("ROUNDTABLE_MODEL_FALLBACK_MAX", raising=False) assert fallback_model_chain("m2") == ["m2", "m1", "m3"] def test_chain_empty_current_uses_config_order(monkeypatch) -> None: _patch_models(monkeypatch, ["m1", "m2", "m3"]) monkeypatch.delenv("ROUNDTABLE_MODEL_FALLBACK", raising=False) assert fallback_model_chain("") == ["m1", "m2", "m3"] assert fallback_model_chain(None) == ["m1", "m2", "m3"] def test_chain_unknown_current_kept_first(monkeypatch) -> None: _patch_models(monkeypatch, ["m1", "m2"]) monkeypatch.delenv("ROUNDTABLE_MODEL_FALLBACK", raising=False) assert fallback_model_chain("custom") == ["custom", "m1", "m2"] def test_chain_disabled_single(monkeypatch) -> None: _patch_models(monkeypatch, ["m1", "m2", "m3"]) monkeypatch.setenv("ROUNDTABLE_MODEL_FALLBACK", "0") assert fallback_model_chain("m2") == ["m2"] def test_chain_cap(monkeypatch) -> None: _patch_models(monkeypatch, ["m1", "m2", "m3", "m4"]) monkeypatch.delenv("ROUNDTABLE_MODEL_FALLBACK", raising=False) monkeypatch.setenv("ROUNDTABLE_MODEL_FALLBACK_MAX", "2") assert fallback_model_chain("m1") == ["m1", "m2"] def test_chain_no_config_returns_current(monkeypatch) -> None: _patch_models(monkeypatch, []) monkeypatch.delenv("ROUNDTABLE_MODEL_FALLBACK", raising=False) assert fallback_model_chain("m9") == ["m9"] assert fallback_model_chain("") == [] def test_reason_text() -> None: assert reason_text("busy") == "服务繁忙" assert reason_text("empty") == "未产出有效内容" assert reason_text("timeout") == "长时间无响应（卡死）" assert reason_text(None) == "调用失败" assert reason_text("不存在的码") == "调用失败" # ── 引擎：单席位异常隔离 + 模型容错记录 → 系统提示气泡 ────────────────────────── _AGENTS = [SeatRef("seat-a", "情报"), SeatRef("seat-b", "环境"), SeatRef("seat-c", "方案")] def _writer() -> JobProgressWriter: return JobProgressWriter(None, job_id="job-x", agents=_AGENTS) def _thread_ids() -> dict[str, str]: return {a.agent_id: f"tid-{a.agent_id}" for a in _AGENTS} | {"roundtable-coordinator": "tid-co"} class _ChainGateway: """chain 模式假网关：可指定某席位抛异常 / 带回模型容错记录。""" def __init__(self, *, fail_agent: str | None = None, switch_agent: str | None = None) -> None: self.fail_agent = fail_agent self.switch_agent = switch_agent self.seen: list[str] = [] async def run_leader(self, *, thread_ids, coordinator_name, message, model) -> LeaderTurn: return LeaderTurn(final_text="总控说明", dispatched=[]) async def run_seat(self, *, thread_ids, agent_id, task, model) -> SeatTurn: self.seen.append(agent_id) if agent_id == self.fail_agent: raise RuntimeError("boom: 模型连接被打满") switches = [] if agent_id == self.switch_agent: switches = [ModelSwitch(failed_model="m1", next_model="m2", reason_label="服务繁忙")] return SeatTurn(final_text=f"{agent_id} 的交付", switches=switches) async def create_report_thread(self) -> str: # pragma: no cover raise AssertionError("enable_report=False 不应绘制报告") async def _run_chain(gw: _ChainGateway) -> tuple[object, JobProgressWriter]: writer = _writer() result = await run_orchestration( gateway=gw, progress=writer, agents=_AGENTS, thread_ids=_thread_ids(), coordinator_name="roundtable-coordinator", model="m1", seed_message="种子任务", mode="chain", enable_report=False, ) return result, writer def test_chain_seat_failure_isolated() -> None: gw = _ChainGateway(fail_agent="seat-b") result, writer = asyncio.run(_run_chain(gw)) # 单席位失败不拖垮整链：仍走完三个席位并收口为 done。 assert result.status == "done" assert gw.seen == ["seat-a", "seat-b", "seat-c"] # 失败席位的交付气泡为空，其余正常。 seat_dlgs = {d["agentId"]: d["content"] for d in writer.dialogues() if d["role"] == "seat"} assert seat_dlgs["seat-b"] == "" assert seat_dlgs["seat-a"] == "seat-a 的交付" assert seat_dlgs["seat-c"] == "seat-c 的交付" def test_model_switch_emits_system_notice() -> None: gw = _ChainGateway(switch_agent="seat-b") _result, writer = asyncio.run(_run_chain(gw)) systems = [d for d in writer.dialogues() if d["role"] == "system"] assert len(systems) == 1 assert "模型容错" in systems[0]["content"] assert "服务繁忙" in systems[0]["content"] assert "m2" in systems[0]["content"]