《AI Agent 高级编程》1.2 统一抽象：一个 Protocol 适配多家并写适配层

1.2 统一抽象：一个 Protocol 适配多家并写适配层

1.1 的四层图告诉你 SDK 层各家用起来不一样：PydanticAI 是 Agent(model, output_type=...).run()，Claude Agent SDK 是 query() 会话，OpenAI Agents SDK 是 Runner.run(agent, ...)。

如果业务代码直接写死某家的调用形态，你就把整条业务逻辑绑死在了这家框架上——这正是 1.1 结尾说的「锁定度」。本节用一个只有单个方法的 Protocol 把这件事解耦。

本节把「统一抽象」推进到可运行：定义 AgentBackend Protocol，写本地假后端与 PydanticAI（TestModel / FunctionModel）三个适配器，用同一份调用代码跑出三种输出并逐条断言。跑完之后你会看到——换框架只需换一个适配器，调用方一行不改。

1.2.1 抽象应该有多薄

先定原则：抽象层只暴露「业务真正需要的最小面」。业务需要的是什么？一句话——「给我一个 prompt，还我一个字符串」。至于这背后是 PydanticAI 的循环、Claude 的会话还是本地函数，调用方不该知道。

一个常见的错误是把抽象做厚：把「工具」「多轮」「流式」「重试」「检查点」全塞进接口。结果是任何一家框架只要有一项不满足，你的抽象就漏了，而且厚接口本身就变成了新的锁定。正确的做法是薄接口 + 按需扩展：

工具、多轮、持久化这些能力，用组合而不是继承：需要工具时，让后端内部去管；需要编排时，交给 1.1 图里的编排层。抽象层保持薄，才能真的「适配多家」。

还有一条经验：接口里出现的每一个方法，你都得为每一家后端写一遍实现。方法越多，适配成本越高，能适配的家数越少。这就是「薄」的经济学。

1.2.2 为什么是 Protocol 而不是 ABC

Python 里有两种「接口」写法：抽象基类 abc.ABC 与结构化协议 typing.Protocol。适配第三方框架时，Protocol 明显更合适：

关键在第二行：你不可能让 PydanticAI 的 Agent 去继承你的 ABC——那是别人的类。但你可以写一个适配器类，让它「长得像」Protocol，然后靠结构化类型通过检查。这正是适配器模式在 Python 里最自然的落地方式。

代价是运行期检查弱：isinstance 只确认方法名在，不确认签名。所以真正的契约要靠测试兜底——本节的 assert 就是在做这件事。

1.2.3 定义 AgentBackend

from typing import Protocol, runtime_checkable


@runtime_checkable
class AgentBackend(Protocol):
    """统一后端：任何能 await run(prompt) -> str 的实现都算合格。"""

    name: str

    async def run(self, prompt: str) -> str: ...

三行说明：

• async run：本卷主线是生产化，网络调用天然是异步的。同步接口在这里会逼所有适配器开线程，得不偿失。
• name 属性：用于日志、路由与打分表（1.3）识别「这次是谁在跑」。
• @runtime_checkable：让 isinstance(b, AgentBackend) 在运行时可用。注意它只检查方法名存在，不检查签名——这是运行期检查的已知局限，真正的契约靠测试兜底。

整个协议就这么多。没有基类、没有 __init__、没有 register。越薄，越好适配。

1.2.4 三个适配器

现在写三个实现，覆盖「完全本地」与「挂第三方框架」两类：

from pydantic_ai import Agent
from pydantic_ai.models.test import TestModel
from pydantic_ai.models.function import FunctionModel, AgentInfo
from pydantic_ai.messages import ModelResponse, TextPart


class LocalFakeBackend:
    """零依赖假后端：输出完全由本地函数决定。"""

    def __init__(self, name: str = "local-fake") -> None:
        self.name = name

    async def run(self, prompt: str) -> str:
        return f"[local] {prompt.strip().upper()}"


class PydanticAIBackend:
    """把任意 pydantic-ai Model 挂到同一 Protocol 上。"""

    def __init__(self, name: str, model) -> None:
        self.name = name
        self._agent = Agent(model, output_type=str)

    async def run(self, prompt: str) -> str:
        result = await self._agent.run(prompt)
        return result.output


def echo_function(messages, info: AgentInfo) -> ModelResponse:
    """FunctionModel 的回调：把最后一条输入原样大写后返回。"""
    last = messages[-1]
    text = last.parts[-1].content if last.parts else ""
    return ModelResponse(parts=[TextPart(f"[function] {text.strip().upper()}")])

两个细节是本机核对过的、凭记忆会写错的：

• Agent(model, output_type=str)：PydanticAI 2.55.0 里输出参数叫 output_type，不是旧版的 result_type。Agent.run 的签名里也是 output_type（本机 inspect.signature(Agent.run) 实测）。
• FunctionModel 的回调签名是 (messages, info)，返回 ModelResponse；info: AgentInfo 是第二个参数。TestModel 则按 output_type 自动合成输出——output_type=str 时它返回固定串 "success (no tool calls)"。

TestModel 与 FunctionModel 是 PydanticAI 官方提供的两种「假模型」，不需要任何 API Key、不发起网络请求，是本卷「零凭据确定性对照」的主力（第 2 章会大量复用）。

三个适配器的差异一览：

1.2.5 真跑：同一份调用代码跑三种后端

关键证明在这一步：调用方 run_suite 只认 AgentBackend，完全不知道背后是谁。

import asyncio


async def run_suite(backends: list[AgentBackend], prompt: str) -> dict[str, str]:
    return {b.name: await b.run(prompt) for b in backends}


async def main() -> None:
    backends = [
        LocalFakeBackend(),
        PydanticAIBackend("pydantic-test", TestModel()),
        PydanticAIBackend("pydantic-func", FunctionModel(echo_function)),
    ]
    print("isinstance 检查:", [isinstance(b, AgentBackend) for b in backends])

    out = await run_suite(backends, "hello agent")
    for k, v in out.items():
        print(f"{k:16s} -> {v!r}")

    assert out["local-fake"] == "[local] HELLO AGENT"
    assert out["pydantic-func"] == "[function] HELLO AGENT"
    assert out["pydantic-test"] == "success (no tool calls)"
    print("assertions OK")


if __name__ == "__main__":
    asyncio.run(main())

真跑输出（/tmp/agentbook/venv3/bin/python，Python 3.14.6）：

isinstance 检查: [True, True, True]
local-fake       -> '[local] HELLO AGENT'
pydantic-test    -> 'success (no tool calls)'
pydantic-func    -> '[function] HELLO AGENT'
assertions OK

三个后端跑完，调用代码 run_suite 一个字没变。isinstance 全为 True，说明 runtime_checkable 的结构化匹配成立。这就是统一抽象的全部价值：替换成本从「改遍业务代码」降到「换一个适配器对象」。

一个容易被忽略的点：pydantic-test 的输出 "success (no tool calls)" 看起来「没意义」，但它是可断言的确定值。测试的价值不在于输出好看，而在于每次跑都一样。本卷第 4.3 节做三框架同题对照时，正是靠这种确定性让「框架 A vs 框架 B」的差异可复现。

1.2.6 错误与超时：抽象层的另一条契约

薄接口容易漏掉一件重要的事：出错时怎么办。如果 run 抛出的异常五花八门（PydanticAI 抛 UnexpectedModelBehavior，某 SDK 抛自己的 AgentError），调用方就没法统一处理。

约定一条最小契约：适配器负责把底层异常翻译成本项目自己的异常，调用方只认这一种。

class BackendError(RuntimeError):
    """统一后端异常：所有适配器把底层错误翻译成它。"""


class LocalFakeBackend:
    # ... name / __init__ 同上
    async def run(self, prompt: str) -> str:
        if not prompt.strip():
            raise BackendError("empty prompt")
        return f"[local] {prompt.strip().upper()}"

配套约定：

• 超时归调用方管：run 内部不设死超时，由 asyncio.wait_for(backend.run(p), timeout=...) 在调用侧控制，这样超时策略可配置。
• 取消（cancel）透传：await 天然支持 CancelledError，适配器不要吞掉它，否则上层无法中断长任务。
• 错误里带 backend.name：排障时一眼看出是哪家在报错，呼应 1.1.4 的「分层排障」。

这条契约不写在 Protocol 里（Protocol 只描述成功路径），但它和接口一样重要——接口定义 happy path，约定定义 failure path。

1.2.7 为真实框架写适配器的骨架

本节的三个适配器都无需凭据。真实 SDK 的适配器形状完全一样，只是 run 内部换成它们的入口：

# 形状示意：本节不执行（运行会真实联网、消耗额度）
class ClaudeAgentBackend:
    def __init__(self, name: str, options) -> None:
        self.name = name
        self._options = options

    async def run(self, prompt: str) -> str:
        from claude_agent_sdk import query   # 延迟 import，避免无谓加载
        chunks = []
        async for msg in query(prompt=prompt, options=self._options):
            chunks.append(str(msg))
        return "".join(chunks)

三点必须写清楚：

1. 这段代码本节未执行。claude_agent_sdk.query 会 spawn 本机 claude CLI、发起真实请求并消耗额度。本卷第 3、4 章只核对 inspect.signature / dataclasses.fields 等接口表面，不运行。
2. 延迟 import：把框架 import 放进 run 内，避免服务启动时加载所有后端——多后端场景下这能显著缩短冷启动。
3. 适配器的职责是「翻译」：把 SDK 的消息流翻译成字符串、把 SDK 的异常翻译成 BackendError。翻译层越薄，未来换 SDK 越省事。

OpenAI Agents SDK 同理（Runner.run 入口），api.openai.com 在本机不可达，本卷也不发起调用。

1.2.8 这套抽象的边界与延伸

薄接口有边界，说清楚才不会踩坑：

• 它不隐藏语义差异。local-fake 是纯函数，pydantic-func 走了一个完整循环（含消息组装），二者耗时、可观测性、错误形态都不同。Protocol 统一的是「调用方式」，不是「运行语义」。别指望换了后端行为还一模一样。
• 它不解决工具/多轮。本节的后端都只处理单轮文本。要接工具时，扩展方式是在适配器内部加（比如给 PydanticAI 后端传 toolsets），而不是往 Protocol 上加方法。
• 它不替你选框架。统一抽象降低了「换」的成本，但不降低「选」的难度——选哪个仍是 1.3 那张打分表要回答的问题。

这套抽象替你做了什么：把「换框架」的爆炸半径从整个业务收敛到一个文件；让假后端和真后端在测试里可互换。你让渡了什么：多了一层间接，调试时要在适配器里多跳一次；以及薄接口表达不了框架的独有高级能力——那些能力你得从协议旁边开一个「后门」直接拿。本卷后续会反复出现这个张力。

阅读导航：上一节：1.1 框架分层：引擎/编排/SDK/协议四层定位图 · 下一节：1.3 选型方法：按可观测/持久化/生态/锁定度打分 。