《AI Agent 工程实战》2.3 模型路由与降级：真/假模型开关，离线默认假模型

from dataclasses import dataclass, field
from typing import Any, Literal, Protocol

Role = Literal["system", "user", "assistant", "tool"]


@dataclass
class ToolCall:
    id: str
    name: str
    arguments: dict[str, Any]


@dataclass
class Message:
    role: Role
    content: str = ""
    tool_calls: list[ToolCall] = field(default_factory=list)
    tool_call_id: str | None = None


@dataclass
class ModelResponse:
    message: Message
    stop_reason: str = "end_turn"
    usage: dict[str, int] = field(default_factory=dict)


class Model(Protocol):
    def complete(self, messages: list[Message], tools: list[dict]) -> ModelResponse: ...

class FakeModel:
    def __init__(self, script: list[Any], name: str = "fake") -> None:
        self._script = list(script)
        self._i = 0
        self.name = name
        self.calls: list[tuple[list[Message], list[dict]]] = []

    def complete(self, messages: list[Message], tools: list[dict]) -> ModelResponse:
        self.calls.append((list(messages), list(tools)))
        step = self._script[self._i]
        self._i += 1
        if isinstance(step, str):
            return ModelResponse(message=Message(role="assistant", content=step))
        return ModelResponse(
            message=Message(role="assistant", tool_calls=list(step)),
            stop_reason="tool_use",
        )

class ModelRouter:
    """按顺序尝试若干后端，第一个成功的胜出。"""

    def __init__(self, backends: list[Model]) -> None:
        assert backends, "至少要有一个后端"
        self._backends = backends
        self.trace: list[tuple[str, str]] = []  # (后端名, 结果)

    def complete(self, messages: list[Message], tools: list[dict]) -> ModelResponse:
        last: Exception | None = None
        for b in self._backends:
            try:
                resp = b.complete(messages, tools)
            except Exception as e:
                self.trace.append((b.name, f"FAIL {type(e).__name__}: {e}"))
                last = e
                continue
            self.trace.append((b.name, "OK"))
            return resp
        assert last is not None
        raise last

from dataclasses import dataclass


@dataclass
class Settings:
    use_fake_model: bool = True
    model_name: str = "claude-sonnet-4-5"
    fallback_model: str = "claude-haiku-4-5"
    degrade_to_fake: bool = True


class UnavailableModel:
    """模拟真实适配器：网络 / 额度问题直接抛错。"""

    def __init__(self, name: str, err: str = "429 rate_limit") -> None:
        self.name = name
        self.err = err

    def complete(self, messages: list[Message], tools: list[dict]) -> ModelResponse:
        raise RuntimeError(self.err)


def build_anthropic(name: str) -> Model:
    return UnavailableModel(name)  # 真实项目里是 AnthropicModel(api_key=..., model=name)


def build_model(s: Settings) -> Model:
    if s.use_fake_model:
        return FakeModel(["（离线假模型）已收到你的问题。"], name="fake-default")
    chain: list[Model] = [build_anthropic(s.model_name), build_anthropic(s.fallback_model)]
    if s.degrade_to_fake:
        chain.append(FakeModel(["（降级）上游暂不可用，已用离线模型作答。"], name="fake-last-resort"))
    return ModelRouter(chain)

print("== ① 默认配置：离线假模型，零凭据 ==")
m = build_model(Settings())
print("后端类型:", type(m).__name__)
print("回复:", m.complete([Message(role="user", content="ResearchCopilot 是什么？")], []).message.content)

print()
print("== ② use_fake_model=False：主 -> 备 -> 假模型兜底 ==")
router = build_model(Settings(use_fake_model=False))
resp = router.complete([Message(role="user", content="ResearchCopilot 是什么？")], [])
print("最终回复:", resp.message.content)
print("降级轨迹:")
for name, outcome in router.trace:
    print(f"  {name:<20} {outcome}")
print("请求成功:", resp.message.content != "")

== ① 默认配置：离线假模型，零凭据 ==
后端类型: FakeModel
回复: （离线假模型）已收到你的问题。

== ② use_fake_model=False：主 -> 备 -> 假模型兜底 ==
最终回复: （降级）上游暂不可用，已用离线模型作答。
降级轨迹:
  claude-sonnet-4-5    FAIL RuntimeError: 429 rate_limit
  claude-haiku-4-5     FAIL RuntimeError: 429 rate_limit
  fake-last-resort     OK
请求成功: True