上下文投毒：AI Agent失败的真凶在数据而非模型

当一个AI Agent被要求查询某人自2月1日以来的所有销售通话时，它可能返回这个人根本没参与的通话——尽管正确的通话记录就在它的上下文窗口里。问题出在到达模型的数据上，而非模型本身。

Airbyte CEO兼联合创始人Michel Tricot在一场直播演示中，直接对Gong API执行了这一查询。由于该API无法按用户过滤通话，Agent只能逐页遍历约一个半月的全部通话记录。Tricot表示，这条路径通常需要约三分钟，由于臃肿的上下文让Agent无法筛选出正确的人，返回结果往往超过45条通话，并且比通过上下文存储的路径多消耗了近30000个token。这些是他单次演示的粗略数字，并非基准测试，但问题形态很常见：工具把本该由应用完成的过滤工作交给了模型。

这就是「上下文投毒」（Context Poisoning）——错误、过时或结构糟糕的数据进入上下文窗口，Agent基于这些数据推理，无需任何恶意行为就会出错。

正如大家所说，今天的问题不在模型，而在于你提供给模型的数据。

——Michel Tricot，Airbyte CEO兼联合创始人，Chain of Thought 第52期

在插入点做日志记录

要在工具结果变为模型可见文本的那个位置进行埋点。仅记录HTTP响应大小是不够的，因为中间的任何过滤或转换都会改变模型实际看到的内容。

对每次工具调用，记录序列化结果新增的token数量以及它传递的记录ID。运行结束后，记录答案实际使用了其中多少条记录。保留精确的载荷（在正常数据访问控制下），这样你能读到模型读到的内容。

import json

def append_tool_result(call_id, tool, records, messages, audit):
    content = json.dumps({"records": records}, separators=(",", ":"))
    messages.append({"role": "tool", "tool_call_id": call_id,
                     "content": content})
    audit[call_id] = {
        "tool": tool,
        "tokens_added": count_tokens(content),
        "returned_ids": {r["id"] for r in records},
        "ids_used": None,  # unknown until someone checks the answer
    }

def record_usage(call_id, verified_ids, audit):
    entry = audit[call_id]
    entry["ids_used"] = len(entry["returned_ids"] & set(verified_ids))

在所有数据塑形之后调用这个包装器，确保记录的内容与模型收到的完全一致。verified_ids应从核对过的引用或审查中填入——引用展示的是模型声称使用了什么，而不一定是它实际依赖了什么，所以在有人核查之前，该字段应保持未知状态。

上下文窗口（Context Window） 指模型单次推理能容纳的输入token上限，所有系统提示、用户消息、工具结果和对话历史都共享这一预算。超出后要么截断要么报错，而截断往往正是Agent丢失关键记录的原因。Token 是模型处理文本的最小单位，英文中约等于四分之三个单词，中文里一个汉字通常占1到2个token。工具调用（Tool Call） 指Agent通过结构化消息请求外部函数执行并把结果回传的机制，回传内容以 role: "tool" 的消息形式进入对话历史，从此永久占据上下文。载荷（Payload） 则指工具实际返回的原始数据体。理解这四者关系，才能明白为何每次工具结果都会持续消耗后续每一轮的token预算——这也是插入点日志必须记录token增量而非仅HTTP大小的根本原因。

精度（Precision）与召回（Recall） 在此语境中描述检索质量：召回是找到所有相关记录的能力，精度则是返回结果中相关记录所占比例。上下文投毒通常是精度问题——相关记录可能都在，但被大量无关记录淹没，模型因此难以正确加权。

关注精度，而不只是大小

大结果不一定是坏结果——比较40条通话确实就需要40条通话。

真正的警示信号是：大结果中充满了问题范围之外的记录。近似重复的记录挤占上下文窗口，模型可能即使持有正确记录也无法给予其足够权重。这种trace读起来像是推理失败，导致团队去评估模型，而问题其实出在输入端。

对于失败的运行，将传递的ID与任务所需的记录做对比，并检查任何错误声明与载荷中相邻记录的关系。然后用经过塑形的结果、相同的模型和提示词重新运行任务，检查正确性和覆盖率，同时观察上下文窗口是否变小。

在模型看到之前完成过滤

Tricot描述了在检索端做这项工作的做法：

第一件事是，很多工作在后端完成，确保我们只返回我们认为最相关的信息。

——Michel Tricot，Airbyte CEO兼联合创始人，Chain of Thought 第52期

将任务约束推送到源头能支持的地方：按ID获取、按用户过滤、应用日期范围。只返回问题需要的字段，加上用于归因的标识符。当源头无法表达查询时，在应用代码中过滤或查询为检索同步的存储。不要因为上游API恰好分页就把每一页都发给模型。

投影是按任务定制的。一个通话列表工具可以返回紧凑的元数据。一个回答某人在通话中说了什么的问题工具，则需要相关转录文本——裁掉它虽然省token，却会让答案失去支撑。

投影（Projection） 借自数据库术语，指从一条记录中只挑选部分字段返回，而非整行。它同时减少token消耗和干扰信息。下推（Pushdown） 指把过滤、排序、聚合等操作尽量推向数据源头（API参数、SQL WHERE子句），避免把原始数据拉到应用层再处理。当上游API支持 owner_id 或日期范围参数时，下推能显著缩小返回集；当API不支持时，退而求其次是在应用代码或检索存储中完成，但绝不能把未过滤的全量数据交给模型。检索同步的存储 指把源系统数据周期性同步到本地向量库或文档库，以便支持源头无法表达的查询形态，代价是引入新鲜度问题，这也是下一节讨论实时读取的原因。

决定哪些读取必须实时

同步存储引发新鲜度问题。Tricot自己的方案是混合模式：在存储中搜索和发现，然后在任务需要最新版本时前往API获取特定对象。

API非常擅长检索单个对象——只要你知道这个对象是什么。

——Michel Tricot，Airbyte CEO兼联合创始人，Chain of Thought 第52期

把这变成工具策略，而不是让Agent临场发挥。

FIELDS = ("id", "owner_id", "started_at", "title")

def get_call(call_id, require_live):
    source = "live_api" if require_live else "synced_store"
    record = fetch_record(source, call_id, fields=FIELDS)
    return {f: record[f] for f in FIELDS}

最终投影控制进入上下文窗口的内容，即使上游响应携带更多信息。暴露同步时间戳让过时判断可见。一个需要完整、最新列表的任务，需要一条同样最新的发现路径。

新鲜度（Freshness） 指存储中的数据与源头最新状态之间的时间差，通常用同步时间戳衡量。不同任务对新鲜度的容忍度差异极大：发现类任务（列出某人的所有通话）可以接受几小时延迟，而读取单条记录内容并需要最新版本时，则可能要求秒级实时。混合模式（Hybrid Pattern） 正是为平衡这两类需求而设计——用同步存储承担宽范围搜索和过滤，再针对少数关键对象回源到API获取实时版本。关键是把这套策略固化到工具接口里（例如通过 require_live 参数显式表达），而不是让模型在每次调用时自行猜测该用哪条路径，因为模型无法可靠判断数据的陈旧程度。

核心要点

面对下一次Agent运行失败：

• 读取进入上下文的精确工具载荷
• 按工具调用记录新增token和答案使用的记录
• 在插入之前完成过滤和投影，而非之后
• 为发现和单记录读取分别设定新鲜度规则

更详细的解释（含节目片段）可在Chain of Thought查看。