《AI Agent 工程实战》5.3 查询优化:EXPLAIN 定位慢查询并加索引
会话内拉最近 N 条消息是 ResearchCopilot 最热的查询。本节造 5 万条真实规模数据,用 EXPLAIN (ANALYZE, BUFFERS) 真跑,看它从 Seq Scan(8.8 ms、扫描 718 个页)变成 Index Scan(0.07 ms、只读 4 个页),并给出该加与不该加的索引清单。
category
会话内拉最近 N 条消息是 ResearchCopilot 最热的查询。本节造 5 万条真实规模数据,用 EXPLAIN (ANALYZE, BUFFERS) 真跑,看它从 Seq Scan(8.8 ms、扫描 718 个页)变成 Index Scan(0.07 ms、只读 4 个页),并给出该加与不该加的索引清单。
run 是一台状态机:queued → running → succeeded/failed/cancelled。本节用 UPDATE ... WHERE status='queued' RETURNING 的返回行数判断谁抢到执行权,真跑两个并发 worker 抢同一个 run,并演示「先 SELECT 再 UPDATE」为什么会把同一个任务跑两遍。
ResearchCopilot 的数据层从四张表开始:sessions、messages、runs、memories。本节给出可直接执行的 DDL,说明为什么消息要存原始 JSON 而不是拆成列,用 psycopg 真连本机 PostgreSQL 17.11 跑通建表、写入与查询,再配一套带版本表与咨询锁的迁移脚本。
ReAct 与 Plan-and-Solve 都是「一次跑到底」,失败了不会回头。本节给循环外面再套一层 Reflexion:执行后由 evaluator 判定成败,失败就生成一段反思文本写回上下文,带着反思重跑,最多 N 轮。我们用 FakeModel 脚本化「第一次算错 → 反思 → 第二次算对」并真跑打印两轮轨迹,最后给何时值得 Reflexion 的决策清单。
ReAct 边想边做,Plan-and-Solve 则先谋后动。本节把循环拆成 Planner 与 Executor 两段:Planner 用结构化输出(JSON 数组)一次性产出子任务列表,Executor 顺序执行每一步且每步内部仍是一个完整小循环。我们用 FakeModel 演示一个三步计划,打印每步状态,并给出与 ReAct 的取舍表。
本节给 minagent 装上第一种推理循环 ReAct:把 Thought、Action、Observation 三元组显式写进每轮。我们说明现代 tool-use API 下 Thought 就是 assistant 消息的文本部分,实现一个 ReActAgent,用 FakeModel 脚本化「想—做—看」两轮后收敛的完整轨迹并真跑打印,最后给出与 3.1 朴素循环的差异表。
本节给 minagent 的 ToolRegistry 加一层权限包装:工具按只读、写入、危险分三档,危险工具调用前挂起等人工审批。我们用假模型驱动一条完整轨迹——模型请求危险工具、被挂起、审批通过、继续执行,把每一步贴出来,并给出权限档位表与审批超时策略。
本节把 workspace_id 贯穿所有资源,做行级租户隔离:所有查询强制带租户条件,同一个 run id 在别的 workspace 下取不到,且统一返回 404 而非 403,避免泄露资源是否存在。我们用真连的本机 Postgres 跑一遍「A 租户访问 B 租户资源被拒」,并列出隔离失效的三个高频坑。
本节给 ResearchCopilot 装上第一道门:用 FastAPI 依赖注入实现认证,同时支持 Bearer JWT 与 API Key 两种凭证。我们用 PyJWT 2.15.1 真跑签发与校验,演示合法 token 通过、过期 token 401、篡改签名 401 三种情况,并把认证失败统一到 3.3 的错误契约。
模型一轮可以同时发出多个 tool_call。本节先看串行执行的浪费,再用 concurrent.futures.ThreadPoolExecutor 把多个工具并发跑起来,对比串行与并行的实测耗时,并给出 asyncio.gather 的等价写法,最后讲清结果如何按 tool_call_id 对齐合并回消息。
模型填的参数可能类型错、缺字段、甚至多编一个。本节用 pydantic 2.14.0 给每个工具配一个参数模型,把校验失败变成结构化错误回灌给模型而不中断循环,再用线程池加超时、捕获工具异常、统一序列化返回值,把工具执行彻底隔离起来,全部代码在 Python 3.14.6 实测跑通。
本节把工具层拆成 Tool 与 ToolRegistry 两个小对象:先讲 JSON Schema 的两种生成方式(手写与从类型注解推导),再用极简版 get_type_hints 自动产出 schema,最后实现按名分发、未知工具返回 error 而不抛异常。全部代码在 Python 3.14 实测跑通。
本节补齐 ResearchCopilot 接口的三个边角:用 Idempotency-Key 让同一 key 重复 POST 只创建一个 run、设计 code / message / details / request_id 四字段统一错误体并给出错误码表、用游标分页取代 offset 分页,三处都用 TestClient 真跑并贴出真实输出。
本节给 ResearchCopilot 的 run 接上 SSE 流式接口:定义 id / event / data 三行事件 envelope,把卷一第 7 章的 token / tool_call / tool_result / step / final 映射成 text/event-stream 帧,用 httpx 真消费一条完整事件流,最后给出 SSE 与 WebSocket 的取舍表。
本节为 ResearchCopilot 设计会话与任务 API:把 session、run、message 三类资源钉成 REST 语义,讲清为什么 run 必须独立成资源而不是 session 的一个动作,给出资源关系图与状态字段,并用 FastAPI 0.143.0 真跑一组路由,贴出 TestClient 打出的真实 JSON 响应与 OpenAPI 契约。
上两节的 run() 是纯函数:配置靠参数传、状态散在局部、过程只能事后拿到。本节把它收进 Agent 类——构造时持有 model、registry、system 与 max_steps,run() 返回 RunResult,并新增 on_step 与 on_tool_call 两个回调钩子与 history 状态。我们用假模型演示回调如何逐步打印轨迹,并说明回调为何是流式输出的地基。
上节的 run() 只会跑到自然收敛,模型一卡住就束手无策。本节给循环装上五条止损线:无工具调用、达到 max_steps、stop_reason 为 max_tokens、工具连续失败、重复动作。我们用 AgentLoopError 把异常终止变成可诊断的数据,再用两个假模型分别制造死循环与原地打转,逐一验证每条止损线是否真的生效。
本节给 minagent 装上第一个 Agent 循环:run() 反复调用 model.complete,把 assistant 消息与工具结果追加进同一个消息列表,直到某一轮不再产生工具调用为止。我们用 FakeModel 跑通 (2+3)*4 的两步收敛,打印每一步的 Step 轨迹、最终回答与消息条数,并讨论这个朴素实现留下了哪些坑。
配置与密钥就位后,本节决定 ResearchCopilot 用哪个模型。我们实现一个 ModelRouter:按配置在真实 Anthropic 适配器与假模型之间切换,离线默认假模型(零凭据可跑);并让主模型失败时按链式顺序自动降级。我们用假模型真跑一次「主模型抛错 → 自动降级 → 请求仍然成功」的完整轨迹,并给出路由决策表。
上一节的 Settings 里还嵌着明文密码。本节给 ResearchCopilot 补上密钥治理:用 SecretStr 让密钥在打印、日志、序列化时自动脱敏,用 .env 与 .gitignore 配合完成本地注入,并在启动时对必需密钥做 fail-fast 检查。我们真跑两段代码展示脱敏效果与缺失报错,最后给出一份密钥清单表。