Agent 工程体系 · 第 88/98 篇。内容以 2026 年 9 月可验证的公开规范和稳定接口为基线;框架版本敏感能力会明确标注,不把实验行为写成通用保证。

Agent 缓存与路由:Prompt Cache、语义缓存、工具缓存和失效

Agent 的一次请求通常不是“调用一次模型”这么简单,而是由多个阶段组成:

请求路由模型调用工具调用再次调用模型最终回答\text{请求} \rightarrow \text{路由} \rightarrow \text{模型调用} \rightarrow \text{工具调用} \rightarrow \text{再次调用模型} \rightarrow \text{最终回答}

每个阶段都可能重复消耗延迟、Token、外部 API 配额和模型预算。缓存的目标,是在结果仍然正确的前提下,跳过其中某些重复工作;路由的目标,则是选择一个满足能力、延迟、成本和稳定性约束的执行路径。

两者容易被混为一谈:

  • Prompt Cache 缓存的是模型输入中的可复用前缀或中间计算;
  • 语义缓存 缓存的是“相似问题对应的最终答案或中间答案”;
  • 工具缓存 缓存的是工具调用结果;
  • 路由 决定请求应该进入哪个模型、Agent、工具或降级路径;
  • 失效 决定缓存结果从什么时候开始不能再被信任。

缓存命中并不等于业务正确。对 Agent 而言,真正的问题是:

是否可以复用=输入等价环境等价权限等价新鲜度仍满足要求\text{是否可以复用} = \text{输入等价} \land \text{环境等价} \land \text{权限等价} \land \text{新鲜度仍满足要求}

只要其中一个条件不成立,就不能安全复用。


一、先建立 Agent 执行模型

1. 一次运行由多个可观测步骤组成

把一次 Agent 运行表示为:

R=(s0,a1,o1,a2,o2,,an,y)R = (s_0, a_1, o_1, a_2, o_2, \dots, a_n, y)

其中:

  • s0s_0:初始输入和上下文;
  • aia_i:第 ii 步动作,例如模型调用、工具调用、路由决策;
  • oio_i:该动作的输出;
  • yy:最终结果;
  • nn:运行中实际发生的步骤数。

一次运行的总延迟可以粗略写成:

Ttotal=Troute+i=1n(Tqueue,i+TTTFT,i+Toutput,i+Ttool,i)T_{\text{total}} = T_{\text{route}} + \sum_{i=1}^{n} \left( T_{\text{queue},i} + T_{\text{TTFT},i} + T_{\text{output},i} + T_{\text{tool},i} \right)

其中:

  • TrouteT_{\text{route}}:路由判断耗时;
  • TTTFTT_{\text{TTFT}}:Time To First Token,模型产生第一个输出 Token 前的时间;
  • ToutputT_{\text{output}}:生成剩余输出的时间;
  • TtoolT_{\text{tool}}:工具或外部系统耗时。

缓存只能减少它所覆盖的那一段成本。例如:

  • Prompt Cache 主要减少输入处理和部分首 Token 延迟;
  • 工具缓存可以直接消除一次外部调用;
  • 语义缓存可能跳过整个 Agent 运行;
  • 路由缓存可以减少分类器调用,但会引入“错误复用路由”的风险。

因此,不应只看“缓存命中率”,而要看缓存命中后是否真正减少了:

ΔT,Δtokens,Δtool calls,Δcost\Delta T,\quad \Delta \text{tokens},\quad \Delta \text{tool calls},\quad \Delta \text{cost}

2. Agent 的缓存点

一个典型流程可以拆成以下缓存点:

flowchart LR
    U[用户请求] --> R[路由器]
    R --> SC{语义缓存}
    SC -- 命中且可复用 --> A[返回缓存答案]
    SC -- 未命中 --> M[模型调用]
    M --> TC{工具缓存]
    TC -- 命中 --> TR[复用工具结果]
    TC -- 未命中 --> T[执行工具]
    T --> W[写入工具缓存]
    W --> M2[继续模型调用]
    TR --> M2
    M2 --> O[最终输出]
    M2 --> PC[利用 Prompt Cache]

关键区别是:

  1. Prompt Cache 仍然会执行模型推理,只是复用了输入前缀相关的计算;
  2. 工具缓存跳过工具执行,但仍可能需要模型读取工具结果并决定下一步;
  3. 语义缓存可能直接返回答案,甚至不进入模型;
  4. 路由缓存可能跳过一次路由模型调用,但不能改变最终业务安全边界。

二、Prompt Cache:缓存模型输入前缀,而不是缓存答案

1. Prompt Cache 的定义

Prompt Cache 是模型服务侧对重复输入前缀进行复用的一种机制。

假设两个请求的输入分别是:

系统规则
工具定义
企业知识
用户上下文 A

和:

系统规则
工具定义
企业知识
用户上下文 B

如果前面三部分完全一致,模型服务可能复用前缀相关计算,仅处理变化部分。

因此 Prompt Cache 的缓存对象不是:

“这个问题的答案”

而是:

“这段输入前缀对应的模型处理结果”

这也是它与语义缓存的根本区别。

Prompt Cache 通常要求:

Pprefix(1)=Pprefix(2)P_{\text{prefix}}^{(1)} = P_{\text{prefix}}^{(2)}

这里的“相等”通常是序列级相等,而不是语义相似。下面这些变化可能导致前缀不再匹配:

  • 系统提示词多一个空格;
  • 工具定义顺序变化;
  • JSON 字段顺序变化;
  • 动态时间被插入到系统提示词前部;
  • 会话 ID、用户姓名等动态字段放在了稳定内容之前;
  • 使用了不同模型或不同模型配置;
  • 请求分片方式发生改变。

因此,“内容意思一样”不代表 Prompt Cache 会命中。

2. Prompt Cache 的输入布局

适合缓存的布局通常是:

[稳定系统规则]
[稳定工具定义]
[稳定领域知识]
[租户级稳定配置]
[用户身份与权限]
[本轮对话]
[本轮动态数据]

不适合缓存的布局是:

[当前时间]
[随机请求 ID]
[用户本轮问题]
[稳定系统规则]
[工具定义]

原因很直接:前缀缓存要求变化尽量靠后。如果高频变化的数据出现在前面,后续稳定内容也无法复用。

可以把请求拆成:

P=SDP = S \Vert D

其中:

  • SS:稳定前缀;
  • DD:动态后缀;
  • \Vert:拼接操作。

如果请求频繁变化的是 DD,就应尽量最大化 S|S|,而不是把所有内容都塞进固定模板。

3. OpenAI API 中的 Prompt Cache 观测

截至 2026 年 9 月,OpenAI 文档区分了隐式和显式 Prompt Cache 配置。隐式模式下,平台自动选择缓存断点;显式模式下,应用可以标记可复用的 Prompt 前缀。相关接口还支持 prompt_cache_key,请求使用情况可以通过 usage.prompt_tokens_details.cached_tokens 观察。具体字段和支持范围应以所使用模型的当前 API 文档为准。(developers.openai.com)

这意味着生产系统不能只记录总输入 Token,而应至少记录:

{
  "input_tokens": 12000,
  "cached_input_tokens": 9000,
  "uncached_input_tokens": 3000,
  "cache_write_tokens": 0
}

若只看总 Token,可能误以为 Prompt Cache 无效;若只看命中率,又可能忽略缓存写入成本、缓存前缀过短或命中后节省很小等问题。

4. Prompt Cache 的完整算例

假设一次 Agent 调用包含:

  • 稳定系统规则:3000 Token;
  • 工具定义:4000 Token;
  • 企业知识:5000 Token;
  • 用户上下文:1000 Token;
  • 当前问题:500 Token。

总输入:

I=3000+4000+5000+1000+500=13500I = 3000 + 4000 + 5000 + 1000 + 500 = 13500

其中前 12000 Token 在 100 次请求中保持不变。

若第 1 次请求建立缓存,之后 99 次请求命中稳定前缀,则:

  • 无缓存重复处理量:

100×12000=1,200,000100 \times 12000 = 1{,}200{,}000

  • 有缓存的前缀处理量,近似为:

1×120001 \times 12000

  • 动态部分仍需处理:

100×1500=150,000100 \times 1500 = 150{,}000

所以缓存只能减少稳定前缀相关的输入处理,并不会减少:

  • 当前问题的理解;
  • 当前用户上下文的处理;
  • 输出 Token;
  • 工具执行;
  • Agent 的额外步骤。

如果每次请求都产生新的动态工具描述:

当前用户可用工具:根据请求临时生成

并且它位于稳定企业知识之前,那么实际稳定前缀可能只剩系统规则,缓存收益会显著下降。

5. Prompt Cache 的反例

下面的设计看起来“内容没有变”,实际上很容易破坏命中:

prompt = f"""
当前时间:{now}
请求 ID:{request_id}
用户:{user_id}

你是企业客服 Agent。
以下是固定工具定义:
{TOOLS}

以下是固定规则:
{POLICY}

用户问题:
{question}
"""

动态字段位于最前面,导致后续固定内容都无法作为同一个前缀复用。

更合理的结构是:

prompt = f"""
你是企业客服 Agent。

固定规则:
{POLICY}

固定工具定义:
{TOOLS}

用户:
{user_id}

当前时间:
{now}

用户问题:
{question}
"""

这并不保证一定命中,因为实际服务还可能依赖模型、请求结构、缓存键和平台策略;但它满足“稳定内容在前、动态内容在后”的必要设计方向。


三、语义缓存:缓存“相似请求的答案”,因此必须证明可等价

1. 语义缓存的定义

语义缓存把请求转换为向量或规范化表示,然后查找语义上相似的历史请求。

设请求 qq 的向量为:

e(q)Rde(q) \in \mathbb{R}^d

缓存中已有请求 qq',相似度为:

sim(q,q)=e(q)e(q)e(q)e(q)\operatorname{sim}(q,q') = \frac{e(q)\cdot e(q')} {\|e(q)\|\|e(q')\|}

当:

sim(q,q)τ\operatorname{sim}(q,q') \ge \tau

时,系统可能返回 qq' 的缓存结果。

但这个条件只说明“语义相似”,不说明“答案可以复用”。安全复用至少还需要:

Reusable(q,q)=Sim(q,q)SamePolicySameScopeFreshEnoughSameOutputContract\operatorname{Reusable}(q,q') = \operatorname{Sim}(q,q') \land \operatorname{SamePolicy} \land \operatorname{SameScope} \land \operatorname{FreshEnough} \land \operatorname{SameOutputContract}

其中:

  • SamePolicy:系统规则、业务规则、模型策略一致;
  • SameScope:租户、用户、权限和数据范围一致;
  • FreshEnough:缓存结果仍满足新鲜度要求;
  • SameOutputContract:输出格式和用途一致。

2. 相似不等价的反例

以下两个问题语义非常相似:

杭州今天适合户外跑步吗?
杭州明天适合户外跑步吗?

如果把前一个问题的答案复用给后一个问题,可能得到错误结果,因为日期是答案的关键变量。

再例如:

我的订单为什么还没发货?
我的订单为什么还没发货?订单号 A1001

第一个问题缺少订单标识。若缓存键只使用自然语言相似度,可能把另一个用户的订单状态泄露出来。

因此,语义缓存不能把所有文本都当作可交换的自然语言。应先提取影响答案的业务变量

{
  "intent": "order_shipping_status",
  "tenant_id": "t1",
  "user_id": "u1",
  "order_id": "A1001",
  "as_of": "2026-09-01T10:00:00+08:00"
}

然后再决定哪些字段进入匹配键,哪些字段进入 TTL 或失效条件。

3. 语义缓存的分层策略

一个较安全的语义缓存可以分为三层。

第一层:精确缓存

适合:

  • 固定 FAQ;
  • 版本明确的产品文档;
  • 不含用户数据的公开知识;
  • 输出具有严格确定性的查询。

键可以是:

Kexact=H(normalized_queryprompt_version)K_{\text{exact}} = H(\text{normalized\_query} \Vert \text{prompt\_version})

第二层:受约束的语义缓存

适合:

  • 允许答案在短时间内近似复用的解释型问题;
  • 同一租户内的知识问答;
  • 不涉及实时状态和个性化权限的问答。

除了向量相似度,还应检查:

  • 租户;
  • 知识库版本;
  • 语言;
  • 输出类型;
  • 过滤条件;
  • 安全策略版本。

第三层:不缓存最终答案,只缓存中间结果

适合:

  • 复杂 Agent;
  • 需要实时工具确认的流程;
  • 答案包含个性化数据;
  • 结果受权限影响;
  • 结果具有高风险。

例如可以缓存:

“该问题涉及退款政策第 3.2 节”

但不缓存:

“用户张三可以退款 128 元,预计 2 天到账”

前者是相对稳定的检索或分类结果,后者依赖用户、订单、时间和外部状态。

4. 语义缓存的阈值不是越高越好

设一次请求命中语义缓存的收益为 BB,错误复用造成的期望损失为 LL,命中概率为 p(τ)p(\tau),错误概率为 e(τ)e(\tau),则缓存策略的期望收益可表示为:

E(τ)=p(τ)Be(τ)LE(\tau) = p(\tau)B - e(\tau)L

提高相似度阈值 τ\tau 往往会:

  • 降低命中率;
  • 降低误命中率;
  • 让缓存更接近精确匹配。

对于天气、库存、支付、权限等高损失场景,LL 很大,应提高阈值,甚至禁用最终答案语义缓存。对于低风险的格式转换、固定概念解释,LL 较小,可以接受更宽松的相似度范围。

这不是一个全局参数,而是一个按意图和风险分层的策略


四、工具缓存:缓存外部世界的观察结果

1. 工具缓存的定义

工具缓存保存工具调用的输入与输出:

Ctool:(tool_name,normalized_args,scope)(result,timestamp,version)C_{\text{tool}}: (\text{tool\_name}, \text{normalized\_args}, \text{scope}) \rightarrow (\text{result}, \text{timestamp}, \text{version})

例如:

{
  "tool": "get_exchange_rate",
  "args": {
    "base": "CNY",
    "quote": "USD"
  },
  "result": {
    "rate": 0.1372
  },
  "observed_at": "2026-09-01T10:00:00+08:00",
  "ttl_seconds": 60
}

工具缓存通常比语义缓存更容易推理,因为它可以使用结构化参数进行精确匹配。但它面对的是一个更困难的问题:外部世界会变化

工具结果是否可复用,不仅取决于输入参数,还取决于:

Reusable Tool Result=Same ArgsSame IdentitySame PermissionFresh EnoughTool Version Compatible\text{Reusable Tool Result} = \text{Same Args} \land \text{Same Identity} \land \text{Same Permission} \land \text{Fresh Enough} \land \text{Tool Version Compatible}

2. 读工具和写工具必须分开

可以缓存的通常是幂等读操作:

get_user_profile
get_product_detail
search_documents
get_exchange_rate
get_weather

不能直接缓存执行结果的通常是有副作用的写操作:

create_order
charge_payment
send_email
delete_file
transfer_money

对于写操作,缓存的重点不是“复用结果”,而是幂等键和请求去重

例如:

idempotency_key = f"{tenant_id}:{user_id}:{operation_id}"

如果网络超时后重试 charge_payment,系统应该通过幂等键确认:

同一个 operation_id 是否已经成功执行?

而不是盲目返回上一次模型生成的文字。

3. 工具缓存键必须包含隐藏上下文

下面这个键不安全:

get_invoice:invoice_id=INV-001

如果发票内容受租户权限控制,就必须至少包含:

tenant_id
principal_id 或权限版本
invoice_id
tool_version
请求参数

形式化表示:

K=H(tenant_idprincipal_scopetool_nametool_versionnormalized_args)K = H( \text{tenant\_id} \Vert \text{principal\_scope} \Vert \text{tool\_name} \Vert \text{tool\_version} \Vert \text{normalized\_args} )

否则可能发生跨租户数据泄露:

  1. 用户 A 查询发票 INV-001
  2. 结果被写入缓存;
  3. 用户 B 查询同一发票编号;
  4. 因键缺少租户或权限信息,命中 A 的结果。

缓存命中率越高,漏洞传播越快。

4. 工具结果的 TTL 应由业务变化速度决定

TTL 是 Time To Live,表示缓存条目的最长可接受存活时间。

如果数据变化频率为 λ\lambda,TTL 为 tt,在简单的泊松变化近似下,缓存仍未过期但数据已经变化的概率可以近似为:

P(stale)=1eλtP(\text{stale}) = 1-e^{-\lambda t}

例如某库存平均每 10 分钟变化一次:

λ=1600\lambda = \frac{1}{600}

若设置 TTL 为 60 秒,则:

P(stale)=1e60/6009.5%P(\text{stale}) = 1-e^{-60/600} \approx 9.5\%

这并不意味着库存一定有 9.5% 的错误率,因为实际变化过程未必服从泊松分布;但它说明 TTL 越长,陈旧结果风险通常越高。

对于不同工具,可以采用不同策略:

工具类型 典型策略
静态产品文档 版本化,发布时失效
汇率 秒级或分钟级 TTL
库存 短 TTL,结算前强制回源
用户余额 不缓存,或只缓存展示值
搜索结果 短 TTL,允许降级
支付状态 回源查询,不能仅依赖模型或旧缓存

五、路由:缓存前后都需要路由,但路由输入不能被缓存污染

1. 路由的定义

路由是根据请求特征选择执行路径:

r(x){model,agent,tool,fallback}r(x) \rightarrow \{\text{model}, \text{agent}, \text{tool}, \text{fallback}\}

请求特征 xx 可以包括:

  • 任务意图;
  • 风险等级;
  • 是否需要实时信息;
  • 是否需要工具;
  • 输入长度;
  • 输出格式;
  • 用户等级;
  • 当前延迟和容量;
  • 成本预算;
  • 模型健康状态。

路由的目标不是“永远选最强模型”,而是在约束下优化目标函数:

minrαT(r)+βC(r)+γE(r)\min_r \quad \alpha T(r) + \beta C(r) + \gamma E(r)

同时满足:

Q(r)QminQ(r) \ge Q_{\min}

其中:

  • T(r)T(r):延迟;
  • C(r)C(r):成本;
  • E(r)E(r):错误或失败风险;
  • Q(r)Q(r):质量;
  • α,β,γ\alpha,\beta,\gamma:业务对各项指标的权重。

2. 缓存命中应当位于路由流程的什么位置?

没有唯一答案,取决于缓存类型。

语义缓存通常先于模型路由

请求
→ 安全检查
→ 权限与租户作用域
→ 语义缓存查询
→ 命中则返回
→ 未命中才进行模型路由

但不能在认证之前查缓存。否则缓存系统可能成为越权读取接口。

工具缓存通常位于工具执行器内部

模型决定调用工具
→ 工具执行器校验权限
→ 工具缓存查询
→ 命中则返回工具结果
→ 未命中则调用外部系统

模型不能自行决定某个工具结果是否“足够新”。这应由工具执行器根据工具策略决定。

Prompt Cache 发生在模型请求内部

模型路由先选择模型,然后该模型服务再判断 Prompt Cache 是否命中。因此不同模型通常不能共享同一个 Prompt Cache 语义空间。

3. 路由缓存的风险

可以缓存“路由分类结果”,例如:

{
  "intent": "product_faq",
  "risk": "low",
  "requires_fresh_data": false,
  "route": "small_model"
}

但路由结果也可能过期。下面这些变化都可能使旧路由失效:

  • 工具新增或下线;
  • 模型能力发生变化;
  • 业务政策改变;
  • 用户权限变化;
  • 服务容量变化;
  • 当前请求带有新的约束;
  • 旧模型出现质量回归。

所以路由缓存键至少需要包含:

Kroute=H(normalized_requestpolicy_versiontool_catalog_versionmodel_routing_versionuser_scope)K_{\text{route}} = H( \text{normalized\_request} \Vert \text{policy\_version} \Vert \text{tool\_catalog\_version} \Vert \text{model\_routing\_version} \Vert \text{user\_scope} )

路由缓存不应把“当前服务健康状态”长时间写入静态缓存。容量、错误率和熔断状态属于动态信号,应实时或短周期读取。


六、缓存与模型降级的组合关系

缓存命中、模型路由和降级之间有一个重要顺序:

高风险约束检查
    ↓
精确缓存或安全的工具缓存
    ↓
语义缓存
    ↓
模型路由
    ↓
主模型
    ↓
备用模型
    ↓
非模型降级

但这不是固定流水线。对于实时数据场景,可能必须反过来:

请求
→ 判断需要实时数据
→ 禁止使用最终答案语义缓存
→ 调用工具
→ 工具缓存只允许极短 TTL 或完全回源
→ 使用模型生成回答

1. 缓存命中不是无条件优先级最高

假设用户问:

“我现在账户余额是多少?”

即使语义缓存中有高度相似的问题,也不能直接返回旧答案。因为:

  • 账户状态是用户专属的;
  • 余额是实时或准实时数据;
  • 错误损失可能是金融风险;
  • 权限和身份可能已经变化。

更安全的策略是:

缓存历史解释模板
+
实时查询余额
+
模型或模板生成最终结果

2. 降级时不得放宽安全语义

当主模型不可用时,可以:

  • 切换到备用模型;
  • 返回工具原始结果;
  • 返回结构化状态;
  • 请求用户稍后重试。

不能因为主模型不可用,就把一个陈旧语义缓存结果当成实时结果返回。

降级策略应区分:

失败类型 可接受降级
主模型超时 备用模型或简化提示
低风险 FAQ 模型失败 精确缓存答案
实时天气工具失败 明确告知无法获取最新数据
支付状态查询失败 保守返回“状态待确认”
权限服务失败 拒绝访问,不使用未确认缓存
写操作超时 通过幂等查询确认,不重复执行

七、失效:缓存系统真正困难的部分

1. TTL 只是失效的一种形式

常见失效机制包括:

  1. 时间失效:超过 TTL;
  2. 版本失效:提示词、工具、知识库或政策版本变化;
  3. 事件失效:订单更新、库存变化、用户权限变更;
  4. 作用域失效:用户登出、租户配置切换;
  5. 手动失效:运营或管理员主动清理;
  6. 负缓存失效:错误结果、空结果和“未找到”结果单独设置更短 TTL。

缓存条目的有效性可以表示为:

Valid(c,t)=(ttcTTLc)(Vcurrent=Vc)(scope current=scopec)¬InvalidationEvent\operatorname{Valid}(c,t) = (t-t_c \le TTL_c) \land (V_{\text{current}} = V_c) \land (\text{scope current} = \text{scope}_c) \land \neg \text{InvalidationEvent}

这里的 VV 可以是:

  • prompt 版本;
  • 工具实现版本;
  • 知识库版本;
  • 路由配置版本;
  • 安全策略版本。

2. 版本化通常比全量删除更可靠

假设产品文档更新后需要让语义缓存失效。可以选择:

删除所有缓存键

也可以把知识库版本加入键:

semantic:v42:tenant:t1:embedding:... 

新版本上线后,直接切换到 v43。旧数据可以异步回收。

版本化的优点是:

  • 不需要同步删除海量键;
  • 新旧版本不会混用;
  • 回滚时可以切回旧版本;
  • 便于比较两个版本的质量。

3. 失效和并发写入

缓存通常会遇到缓存击穿:

  1. 某个热门键过期;
  2. 大量请求同时发现未命中;
  3. 所有请求并发访问模型或工具;
  4. 外部服务被突发流量压垮。

常见解决方式是请求合并,也称 single-flight:

import asyncio
from collections import defaultdict

_locks = defaultdict(asyncio.Lock)
_cache = {}

async def get_or_compute(key, compute, ttl_seconds=60):
    item = _cache.get(key)
    if item is not None:
        value, expires_at = item
        if expires_at > asyncio.get_running_loop().time():
            return value, "hit"

    async with _locks[key]:
        # 双重检查:等待锁时,其他请求可能已经填充缓存
        item = _cache.get(key)
        if item is not None:
            value, expires_at = item
            if expires_at > asyncio.get_running_loop().time():
                return value, "coalesced-hit"

        value = await compute()
        expires_at = (
            asyncio.get_running_loop().time() + ttl_seconds
        )
        _cache[key] = (value, expires_at)
        return value, "miss"

这个示例的关键不是 dict,而是“双重检查”:

  • 第一次检查避免正常命中时加锁;
  • 获得锁后再次检查,避免重复计算;
  • 只有真正的第一个请求执行 compute()

生产系统还需要考虑:

  • 分布式锁;
  • 锁超时;
  • 计算失败是否写入负缓存;
  • 进程崩溃后的锁释放;
  • 单个热门键的并发上限;
  • 是否允许返回 stale-while-revalidate 结果。

4. Stale-While-Revalidate 的边界

Stale-While-Revalidate,即“先返回短暂过期结果,同时后台刷新”,适合:

  • 搜索推荐;
  • 低风险 FAQ;
  • 产品展示信息;
  • 非关键统计数据。

不适合:

  • 权限判断;
  • 支付状态;
  • 账户余额;
  • 库存扣减;
  • 删除或转账结果;
  • 任何需要证明“当前状态”的操作。

应明确区分:

fresh:在 TTL 内
stale-but-servable:过期但允许展示
expired:禁止使用

“过期”不等于“删除”。过期条目可以暂时保留用于诊断,但不能继续作为业务事实返回。


八、一个可运行的缓存与路由骨架

下面给出一个不依赖具体模型厂商的 Python 示例,展示:

  • 路由;
  • 精确工具缓存;
  • 语义缓存的作用域检查;
  • 实时数据禁止最终答案缓存;
  • 主模型失败后的降级。
from __future__ import annotations

import hashlib
import json
import time
from dataclasses import dataclass
from typing import Any, Callable


@dataclass
class CacheEntry:
    value: Any
    expires_at: float
    version: str
    scope: str


class Cache:
    def __init__(self) -> None:
        self.data: dict[str, CacheEntry] = {}

    def get(
        self,
        key: str,
        *,
        version: str,
        scope: str,
        allow_stale: bool = False,
    ) -> Any | None:
        entry = self.data.get(key)
        if entry is None:
            return None

        if entry.version != version or entry.scope != scope:
            return None

        if entry.expires_at < time.time() and not allow_stale:
            return None

        return entry.value

    def put(
        self,
        key: str,
        value: Any,
        *,
        ttl: int,
        version: str,
        scope: str,
    ) -> None:
        self.data[key] = CacheEntry(
            value=value,
            expires_at=time.time() + ttl,
            version=version,
            scope=scope,
        )


def stable_json(value: Any) -> str:
    return json.dumps(
        value,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    )


def make_key(prefix: str, value: Any) -> str:
    digest = hashlib.sha256(stable_json(value).encode()).hexdigest()
    return f"{prefix}:{digest}"


def route(request: dict[str, Any]) -> str:
    if request["requires_fresh_data"]:
        return "realtime_agent"

    if request["risk"] == "high":
        return "strong_model"

    if request["estimated_input_tokens"] > 20_000:
        return "long_context_model"

    return "fast_model"


def get_tool_result(
    cache: Cache,
    *,
    tenant_id: str,
    principal_scope: str,
    tool_name: str,
    tool_version: str,
    args: dict[str, Any],
    ttl: int,
    execute_tool: Callable[[], Any],
) -> tuple[Any, str]:
    key = make_key(
        "tool",
        {
            "tenant_id": tenant_id,
            "principal_scope": principal_scope,
            "tool_name": tool_name,
            "tool_version": tool_version,
            "args": args,
        },
    )

    cached = cache.get(
        key,
        version=tool_version,
        scope=principal_scope,
    )
    if cached is not None:
        return cached, "tool-cache-hit"

    result = execute_tool()

    cache.put(
        key,
        result,
        ttl=ttl,
        version=tool_version,
        scope=principal_scope,
    )
    return result, "tool-cache-miss"


def answer_request(
    request: dict[str, Any],
    *,
    semantic_cache: Cache,
    tool_cache: Cache,
    call_model: Callable[[str, dict[str, Any]], str],
    call_realtime_tool: Callable[[dict[str, Any]], Any],
) -> dict[str, Any]:
    policy_version = "policy-v7"
    knowledge_version = "kb-v42"
    scope = f'{request["tenant_id"]}:{request["principal_scope"]}'

    selected_route = route(request)

    # 实时问题禁止直接复用最终答案
    can_use_final_answer_cache = not request["requires_fresh_data"]

    semantic_key = make_key(
        "semantic",
        {
            "normalized_question": request["normalized_question"],
            "language": request["language"],
            "output_contract": request["output_contract"],
        },
    )

    if can_use_final_answer_cache:
        cached_answer = semantic_cache.get(
            semantic_key,
            version=f"{policy_version}:{knowledge_version}",
            scope=scope,
        )
        if cached_answer is not None:
            return {
                "answer": cached_answer,
                "route": selected_route,
                "cache": "semantic-hit",
            }

    tool_result = None
    tool_cache_status = "not-used"

    if request["requires_fresh_data"]:
        tool_result, tool_cache_status = get_tool_result(
            tool_cache,
            tenant_id=request["tenant_id"],
            principal_scope=request["principal_scope"],
            tool_name=request["tool_name"],
            tool_version=request["tool_version"],
            args=request["tool_args"],
            ttl=request["tool_ttl_seconds"],
            execute_tool=lambda: call_realtime_tool(
                request["tool_args"]
            ),
        )

    try:
        answer = call_model(
            selected_route,
            {
                "question": request["question"],
                "tool_result": tool_result,
                "policy_version": policy_version,
            },
        )
    except TimeoutError:
        if selected_route != "fast_model":
            answer = call_model(
                "fast_model",
                {
                    "question": request["question"],
                    "tool_result": tool_result,
                    "policy_version": policy_version,
                    "degraded": True,
                },
            )
        else:
            return {
                "answer": "当前服务繁忙,请稍后重试。",
                "route": selected_route,
                "cache": "miss",
                "tool_cache": tool_cache_status,
                "degraded": True,
            }

    if can_use_final_answer_cache:
        semantic_cache.put(
            semantic_key,
            answer,
            ttl=300,
            version=f"{policy_version}:{knowledge_version}",
            scope=scope,
        )

    return {
        "answer": answer,
        "route": selected_route,
        "cache": "semantic-miss",
        "tool_cache": tool_cache_status,
        "degraded": False,
    }

这个示例中每一步为何成立

stable_json

缓存键必须避免同一组参数因为字典顺序不同而产生不同键。排序后的 JSON 只是键规范化的一种实现。

scope

scope 将租户和权限作用域带入缓存读取。它不是完整的权限校验替代品;工具执行前仍必须重新鉴权。

requires_fresh_data

这是一个业务约束,不应由模型自由决定。只要请求被判断为需要实时数据,就禁止直接返回最终答案语义缓存。

tool_versionknowledge_version

版本字段让发布和回滚可以通过切换版本实现,而不依赖同步删除所有旧键。

模型超时后的降级

降级只改变执行模型,不改变权限、新鲜度和副作用边界。实时工具结果仍然保留,不能因为模型超时就返回旧的最终答案。


九、如何评估缓存是否真正改善了 Agent

缓存需要同时评估性能、成本和质量。

1. 不要只统计命中率

基本指标包括:

Hit Rate=cache hitscache lookups\text{Hit Rate} = \frac{\text{cache hits}} {\text{cache lookups}}

但更重要的是:

Useful Hit Rate=safe hits that avoided workall cache lookups\text{Useful Hit Rate} = \frac{\text{safe hits that avoided work}} {\text{all cache lookups}}

一次 Prompt Cache 命中可能只减少部分输入处理;一次工具缓存命中可能消除数百毫秒外部调用;一次语义缓存命中可能消除整个 Agent 运行。三者不能放在同一个指标中平均。

建议分别统计:

  • Prompt Cache 命中率;
  • 缓存 Token 数;
  • 语义缓存命中率;
  • 语义缓存误命中率;
  • 工具缓存命中率;
  • 工具调用减少次数;
  • 缓存命中后的 TTFT;
  • 总步骤数;
  • 总模型调用次数;
  • 总工具耗时;
  • 降级比例;
  • 缓存命中后的任务成功率。

2. 缓存必须进入 Agent Trace

OpenAI Agent Evals 文档将 trace 描述为一次工作流的端到端记录,其中可以包含模型调用、工具调用、guardrail 和 handoff;trace grading 可以用来检查工具选择、handoff、提示词变化和路由变化带来的工作流影响。(developers.openai.com)

因此,每个缓存决策都应记录为可观测事件:

{
  "cache_layer": "semantic",
  "cache_status": "hit",
  "cache_key_hash": "sha256:...",
  "cache_version": "policy-v7:kb-v42",
  "scope_hash": "sha256:...",
  "similarity": 0.96,
  "ttl_remaining_seconds": 182,
  "bypass_reason": null
}

不要记录原始用户隐私数据作为缓存键日志。应使用哈希、脱敏字段或结构化标签。

Agents SDK 的 tracing 以 trace 和 span 表示一次工作流及其内部操作;文档中提供了 generation、function、handoff、guardrail 和 custom span 等类型,也支持将多个 trace 通过 group_id 关联到同一会话或过程。(openai.github.io)

一个自定义缓存 span 可以这样写:

from agents.tracing import custom_span

def record_cache_lookup(layer: str, status: str, key_hash: str):
    with custom_span(
        name="cache_lookup",
        data={
            "layer": layer,
            "status": status,
            "key_hash": key_hash,
        },
    ):
        pass

实际项目中应把 span 包围在真实缓存查询代码外:

from agents.tracing import custom_span

with custom_span(
    "semantic_cache_lookup",
    {
        "status": "miss",
        "cache_version": "policy-v7:kb-v42",
    },
):
    cached_answer = semantic_cache.get(...)

Tracing 的同步处理器不应阻塞主请求,也不应因为观测系统失败而影响业务请求;Agents SDK 文档明确建议 span 生命周期使用上下文管理,并在错误时正确记录错误状态。(openai.github.io)

3. 评测集必须覆盖“应该不命中”的样本

如果评测集只包含重复 FAQ,缓存系统很容易得到虚假的高收益。至少应加入以下样本:

同义但日期不同
同义但租户不同
同义但用户不同
同义但权限不同
同义但知识库版本不同
同义但需要实时工具
同义但输出格式不同
相似但业务意图不同
缓存刚失效
工具结果刚更新
主模型失败并触发降级

评测目标不仅是回答质量,还包括:

Cache Safety=应命中且正确复用的请求所有发生复用的请求\text{Cache Safety} = \frac{\text{应命中且正确复用的请求}} {\text{所有发生复用的请求}}

对于高风险场景,应把“错误复用”作为硬失败,而不是普通质量扣分。

OpenAI 的 Agent Evals 建议先通过 traces 调试工作流行为,再将已知的“什么是好结果”固化为 datasets 和 eval runs,以便重复比较提示词、路由和工作流变化。(developers.openai.com)


十、常见误解与对应诊断

误解一:Prompt Cache 命中就不会调用模型

错误。Prompt Cache 通常只复用模型输入前缀相关工作,模型仍需完成本轮理解和生成。

诊断方法:

  • 查看模型调用次数;
  • 查看缓存输入 Token;
  • 对比 TTFT,而不是只看总请求延迟;
  • 对比输出 Token 和工具耗时。

误解二:语义相似度高就可以返回旧答案

错误。语义相似只是一种候选检索条件,不是业务等价证明。

诊断方法:

  • 检查日期、用户、租户、权限和版本字段;
  • 抽样查看高相似低正确率样本;
  • 对比“命中回答”和“重新执行回答”的差异;
  • 对实时意图强制绕过最终答案缓存。

误解三:工具缓存只要把参数序列化就够了

错误。权限作用域、工具版本和数据新鲜度同样属于输入。

诊断方法:

  • 复查缓存键是否包含租户和主体作用域;
  • 查看权限变更后旧缓存是否仍可读取;
  • 检查工具发布后旧结果是否被隔离;
  • 检查写操作是否误用了普通结果缓存。

误解四:TTL 越长,成本越低

不一定。TTL 延长会提高命中率,但也会提高陈旧结果概率。最终成本还包括:

  • 错误答案造成的人工处理;
  • 用户重试;
  • 业务补偿;
  • 风险事件;
  • 缓存失效后的突发回源。

应优化:

总成本=缓存存储成本+未命中计算成本+错误复用期望损失\text{总成本} = \text{缓存存储成本} + \text{未命中计算成本} + \text{错误复用期望损失}

误解五:缓存命中率高说明系统变快

不一定。若语义缓存查询本身耗时很高,或者 Prompt Cache 只覆盖很短的前缀,命中率高也可能没有明显收益。

应测量:

Twith cache=Tcache lookup+Tcache validation+Tremaining workT_{\text{with cache}} = T_{\text{cache lookup}} + T_{\text{cache validation}} + T_{\text{remaining work}}

只有当:

Tcache lookup<Tavoided workT_{\text{cache lookup}} < T_{\text{avoided work}}

缓存才在延迟上有正收益。


十一、生产中的最小安全边界

一个可上线的 Agent 缓存系统至少应做到以下几点:

  1. 认证和权限检查先于缓存读取
  2. 最终答案缓存与工具结果缓存分开管理
  3. 读操作和有副作用的写操作分开处理
  4. 缓存键包含必要的租户、主体、版本和参数作用域
  5. 实时数据意图默认绕过最终答案语义缓存
  6. Prompt Cache、语义缓存和工具缓存使用独立指标
  7. 每次命中、未命中、绕过和失效都可追踪
  8. 模型降级不能放宽权限和新鲜度要求
  9. 缓存失效支持 TTL、版本和事件三种路径
  10. 评测集必须包含相似但不可复用的反例

缓存的正确抽象不是“把结果存起来”,而是:

缓存=可证明的复用+作用域约束+新鲜度约束+失效机制+可观测验证\text{缓存} = \text{可证明的复用} + \text{作用域约束} + \text{新鲜度约束} + \text{失效机制} + \text{可观测验证}

Prompt Cache 解决的是模型输入计算的重复;语义缓存解决的是相似请求的重复;工具缓存解决的是外部观察结果的重复;路由决定哪些请求进入哪条执行路径;失效则决定系统何时必须重新面对真实的模型、工具和外部世界。

当缓存设计无法回答“为什么这个结果仍然等价”时,它就不是性能优化,而是在隐藏不确定性。


系列导航与关联阅读

官方资料

本文依据 Agent、模型、协议与框架官方资料重新梳理;正文、示例与生产清单由 WR BLOG 编写。