《AI Agent 高级编程》1.2 统一抽象:一个 Protocol 适配多家并写适配层
本节定义一个统一后端抽象 AgentBackend Protocol(单个 async run),并为它写适配层:本地假后端、PydanticAI 的 TestModel、以及自定义 FunctionModel 三种实现挂到同一个 Protocol 上,用同一份调用代码真跑并断言输出,证明它们可互换,最后讨论错误与超时契约及真实框架适配骨架。
category
本节定义一个统一后端抽象 AgentBackend Protocol(单个 async run),并为它写适配层:本地假后端、PydanticAI 的 TestModel、以及自定义 FunctionModel 三种实现挂到同一个 Protocol 上,用同一份调用代码真跑并断言输出,证明它们可互换,最后讨论错误与超时契约及真实框架适配骨架。
本节给出 Agent 框架的四层定位图:引擎层(loop 内核)、编排层(LangGraph 图执行器)、SDK 层(PydanticAI、Claude Agent SDK、OpenAI Agents SDK、smolagents)、协议层(MCP、A2A),用本机真实 import 打印模块归属与版本号作证,并把卷一 minagent 与卷二 ResearchCopilot 钉到同一张图上。
并发开得越大,越怕一个慢工具拖住整轮。本节用 asyncio.timeout 给每个工具调用设墙钟上限,演示用户断开时取消如何从 run 一路传到最内层的工具,剖析忘了 await 的 Task 与 TaskGroup 失败取消两种泄漏。真跑注入一个睡 10 秒的慢工具,验证 1 秒超时后整个 run 及时止损并返回明确错误,最后给出超时值的决策表。
上一节把并发开到最大,这一节给它装刹车。无限制的并发会打爆下游:模型 API 限流、数据库连接池耗尽。本节用 asyncio.Semaphore 做并发上限,用手写最小连接池做资源复用,讲清背压「生产快于消费时必须让上游慢下来」,真跑 200 个任务并发上限 10 打印峰值并发与总耗时,并给出按下游 QPS 反推并发上限的估算方法。
Agent 天生并发:一轮可以发出多个工具调用,一个任务可以拆成多个子 Agent。本节把卷一 4.3 的线程池写法升级成 asyncio.TaskGroup 的真并发版本,演示结果按调用顺序回填而非完成顺序,再并行跑多个子 Agent 汇总,用带 sleep 的假工具实测串行与并行的墙钟差,并给出并发安全边界表。
检索解决「怎么取」,本节解决「怎么存」。如果什么都往记忆里塞,召回会被噪声淹没;塞得太少又会漏掉关键信息。本节给 minagent 加上写入策略:用 should_write 判断什么值得记(偏好、事实、结论 vs 寒暄),用规范化字符串比较做 dedup 去重,用 slot 加时间戳做冲突消解让新信息覆盖旧信息,并真跑一组写入用例。
缓冲与摘要只能记住最近的对话,用户上周说过的偏好照样找不回来。本节让记忆支持按相关度召回:把记忆存成带文本的条目列表,用关键词与中文二元组打分,实现 retrieve(query, k) 挑出最相关的几条,再拼进本轮上下文。我们用假模型演示「用户提到上周的偏好,Agent 从记忆里召回并回答」,并列出关键词检索的三个陷阱。
第 3 章的 history 只增不减,对话一长必然撑爆上下文。本节给 minagent 立下统一的 Memory 协议(add / get_context),并实现两种短期记忆:BufferMemory 保留最近 N 条或 N token,SummaryMemory 超阈值时用模型把旧对话压成摘要。我们用假模型跑一段长对话,对比两版注入 messages 的条数差异,并给出三者对照表。
队列里除了用户触发的 run,还需要两类「自己动」的任务:每 N 分钟重建语料索引的周期任务,和 run 完成后延迟几分钟生成摘要的延迟任务。本节用 Redis 的 SET NX PX 加 token 校验做分布式锁,讲清多 worker 抢同一个定时任务时怎么保证只有一个执行、释放与续期为什么要比 token,并真跑两个并发 worker 争抢同一把锁。
上游抖动让 worker 一次失败就判死太脆弱。本节复用卷一 14.1 的错误分类,给队列加上带指数退避与抖动的重试、超过上限进死信队列、以及「重试必须配幂等」的三件套:把 run id 当幂等键,已 succeeded 的 run 直接复用旧结果不重复执行。真跑注入一个「前两次必失败、第三次成功」的假模型打印重试轨迹,再跑一个永远失败的进死信并读回死信记录。
把一次 run 从同步请求改造成异步任务:API 只写库并入队、立刻返回 202,worker 在另一个进程消费。本节讲清裸 LPUSH/BRPOP 为什么会丢任务,改用 BLMOVE 把任务原子搬到 processing 做可靠队列,并真连本机 Redis 7.4.11 与 Postgres 17.11,跑出完整的 run 生命周期日志与同步/异步取舍表。
Agent 跑到一半崩溃,内存里的 history 与步骤全部蒸发,重启只能从头再来。本节把 Session(含消息、步骤、工具中间结果)序列化成 JSON 存到临时目录,崩溃后用 restore 从断点续跑。我们用假模型脚本化「跑到一半 → 存盘 → 新建对象 → 读盘 → 续跑」,真跑并打印续跑后的消息条数与「未重复已完成的工具调用」的证据,最后讨论检查点频率与体积的取舍。
history 焊在 Agent 实例里,一个 Agent 只服务一段对话,也无法把对话存下来或跨进程搬走。本节引入 Session 状态对象,持有 messages、session_id、metadata、created_at,提供 append/snapshot/restore,把 history 剥出 Agent。用假模型跑两轮对话,验证 history 累积与 session_id 稳定。
一次性返回把 Agent 变成黑箱:用户只看到最后那句话,中间的工具调用与进度全都看不见。本节给 minagent 定义统一的 Event 类型,让 Model 通过 stream() 逐 token 产出事件,并把 3.3 的两个回调升级成单一事件生成器——调用方用 for ev in agent.stream(...) 即可订阅全过程。用假模型脚本化 token 序列,真跑并打印完整事件流。
本节用 Redis 给 ResearchCopilot 加两道闸:按租户的令牌桶限流(Lua 脚本保证读-算-写原子),以及按消耗 token 计费的配额——不是限制请求数,而是限制烧掉的 token。真连本机 Redis 跑出连续请求直到 429 的每一步剩余令牌与剩余额度,并给出限流粒度表与「被限流时 Agent 该怎么办」的策略。
6.1 的字面缓存只认逐字节相同的 key,可用户换个说法问同一个问题它就失灵。本节实现语义缓存:把问题向量化后按相似度阈值命中历史答案,用卷一 9.1 的手写哈希嵌入真跑三种情况——同义改写命中、无关问题不命中、相似但答案相反时误命中,贴出真实相似度数值,并重点讲错配检测与哪些问题绝不能走语义缓存。
本节给 ResearchCopilot 的检索结果与嵌入向量加一层 Redis 缓存:缓存 key 带上 workspace_id 以对齐 4.2 的租户隔离,真连本机 Redis 跑出「首次未命中回源、二次命中」的耗时对比与 70% 命中率,并给出检索、嵌入、会话三类 TTL 策略表,以及缓存穿透、击穿、雪崩在 Agent 场景下的具体形态与对策。
本节给 minagent 加上结构化输出:先讲 JSON mode 与 schema 约束的接口设计(只讲设计不真调),再把重点放在解析兜底上——实现 parse_json 稳健处理代码围栏、前置寒暄、嵌套括号等脏输出,用一组样例做单元测试并真跑,最后演示解析失败时把错误回灌给模型重出。
本节给 minagent 加上主动压缩:区别于 2.3 的被动截断,这里把更早的轮次交给模型压成一段摘要再拼接,实现 compact(messages, budget) 与可插拔的摘要器,真跑对比压缩前后的消息条数与近似 token,并给出「何时摘要、何时直接丢」的取舍表。
本节给 minagent 加上指令分层:把全局角色与安全边界、本次任务目标、本轮临时约束拆成三层,用 build_system 与 build_messages 拼装;再给出一张「哪些内容该进 system、哪些进 user、哪些进工具结果」的判断表,并用假模型真跑对比同一问题在三种分层下的消息序列与 token 数。