《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 加时间戳做冲突消解让新信息覆盖旧信息,并真跑一组写入用例。
本节用中文二元组与英文词做词面匹配,并按相关度和时间排序。它是透明的教学召回器,不是跨用户的长期记忆服务。记忆命中不代表事实可信,最新时间也不代表正确。生产中先按租户与权限过滤,保留来源和更正机制;不可仅靠提示词阻止读取他人资料。
BufferMemory 和 SummaryMemory 用于文本上下文,不用于逐条裁剪带工具调用的运行历史;工具历史应使用第 6.2 节的完整轮次策略。摘要是有损模型输出,不保证最早事实永不丢失。get_context 把摘要标记为资料 user 消息,不能提升为 system 指令。长度截断也可能丢信息,业务重要事实应单独结构化保存并抽查。
队列里除了用户触发的 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 生命周期日志与同步/异步取舍表。
本节只证明固定脚本在完整工具轮次后保存、恢复并继续运行。恢复时校验版本、脚本指纹、游标和工具结果配对。模型游标属于 FakeModel,不能推广为真实供应商游标。真实模型即使看到完整历史,也可能再请求相同工具;已发生副作用不能靠历史完整性保证不重做。
history 焊在 Agent 实例里,一个 Agent 只服务一段对话,也无法把对话存下来或跨进程搬走。本节引入 Session 状态对象,持有 messages、session_id、metadata、created_at,提供 append/snapshot/restore,把 history 剥出 Agent。用假模型跑两轮对话,验证 history 累积与 session_id 稳定。
本节建立工具、步骤、文本与终态的事件契约,供日志和 HTTP 层复用。修订版复用第三章循环,不另写一套丢失保护的 Agent.stream。这个生成器在第一次拉取时先执行完整同步循环,再回放队列。因此消费侧 break 只停止回放,不能取消已经完成的工具,也不对上游模型形成背压;模型首字延迟不会降低。
本节用 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,并给出「何时摘要、何时直接丢」的取舍表。
把角色、任务和本轮约束分开,可以让调用方看清每段文字由谁提供、何时变化。应用中的分层约定不等于供应商角色协议或安全策略。优先级提示只帮助组织文字,模型仍可能忽略;不能依靠一句安全红线授权工具。工具白名单、参数校验和人工确认应在执行代码里完成;缓存也由供应商具体条件决定。