1. 长对话 Agent 为什么越聊越贵、越聊越蠢如果你正在做多轮 Agent、智能客服或者长会话助手大概率遇到过这个现象前 10 轮模型表现很稳聊到第 30 轮开始忘事、重复提问、把系统规则抛到脑后同时单次请求的 token 费用肉眼可见地往上涨。这不是模型退化而是上下文膨胀导致的注意力稀释——你把整段对话历史一股脑塞进窗口模型对中间部分的注意力本来就弱关键信息被淹没在噪声里。上下文工程Context Engineering要解决的就是这件事窗口就那么大历史对话、检索资料、业务规则、用户画像全要挤进去谁留下、谁压缩、谁放在显眼位置直接决定输出质量和成本。这篇从工程化视角拆一个可复用的 ContextManager把窗口管理、动态压缩、关键信息注入三件事做成一个统一入口并给出可复制的配置骨架和压缩触发阈值的验证动作。适合已经跑通单轮调用、准备把 Agent 推向长会话场景的开发者。2. 前置准备用 TaoToken 统一 Key 与 API 通道在写 ContextManager 之前先把模型调用通道固定下来。长会话场景会频繁触发摘要、事实抽取这类小模型调用如果每个模块各自维护 Key 和 base_url配置会散得到处都是。我的做法是统一走 TaoToken 的 API 通道一个 Key 覆盖对话、摘要、抽取三类调用后面 ContextManager 里所有 llm 调用都指向同一个入口。TaoToken 的 API 地址是https://taotoken.net/api兼容 OpenAI 风格的/v1/chat/completions协议所以现有基于 openai SDK 的代码基本不用改只换 base_url 和 api_key 即可。控制台里可以创建和管理 Key建议按项目分 Key方便后面做用量归因。2.1 在 settings.json 中固化接入项我习惯把通道配置写进项目的settings.json业务代码只读配置不硬编码。下面这份骨架可以直接抄把api_key换成你在控制台生成的即可{ llm: { provider: taotoken, base_url: https://taotoken.net/api, api_key: sk-你的TaoTokenKey, chat_model: gpt-4o, summary_model: gpt-4o-mini, timeout: 60, max_retries: 2 }, context: { token_budget: 4000, max_turns: 10, summarize_every: 6, fact_keep_last: 20, retrieval_top_k: 5 } }这里有两个设计点值得说明。第一chat_model和summary_model分开配置摘要和事实抽取是压缩型任务用便宜的小模型就够没必要拿主力模型烧钱。第二context段把窗口和压缩参数集中管理后面调阈值只改这一处不用翻业务代码。2.2 读取配置并初始化客户端import json from openai import OpenAI with open(settings.json, r, encodingutf-8) as f: cfg json.load(f) llm_cfg cfg[llm] client OpenAI( base_urlllm_cfg[base_url], api_keyllm_cfg[api_key], timeoutllm_cfg[timeout], max_retriesllm_cfg[max_retries], ) def llm(messages, modelNone): resp client.chat.completions.create( modelmodel or llm_cfg[chat_model], messagesmessages, ) return resp.choices[0].message.content到这一步通道就固定了。后面 ContextManager 里所有对模型的调用都走这个llm函数摘要时传summary_model即可。如果你还没创建 Key可以去控制台的 API Keys 页面生成一个接入文档里有完整的参数说明。3. 可复制配置ContextManager 的三层结构ContextManager 我拆成三层窗口层负责放得下压缩层负责放得精注入层负责放得对。三层各自独立最后在一个ask()方法里串起来。3.1 窗口层系统指令钉头对话滚动尾部朴素截断从头部删是致命错误因为开头是注意力最强的位置系统规则一旦被删模型立刻放飞。正确做法是滑动窗口系统提示词永远固定在第一位对话只保留最近 N 轮。class SlidingWindow: def __init__(self, system_prompt: str, max_turns: int 10): self.system_prompt system_prompt self.max_turns max_turns self.history: list[dict] [] def push(self, role: str, content: str): self.history.append({role: role, content: content}) if len(self.history) self.max_turns: self.history self.history[-self.max_turns:] def build_messages(self) - list[dict]: return [{role: system, content: self.system_prompt}] self.history3.2 压缩层按 token 预算触发滚动摘要轮数不是可靠度量有的一句话 500 字有的 5 个字。用 token 预算触发才准。token 计数用 tiktoken但要注意不同模型的 tokenizer 不一样留 20% 余量最稳。import tiktoken enc tiktoken.encoding_for_model(gpt-4o) def count_tokens(messages: list[dict]) - int: return sum(len(enc.encode(m[content])) for m in messages) class RollingSummarizer: def __init__(self, summarize_every: int 6): self.summarize_every summarize_every self.summary: str | None None def maybe_summarize(self, history: list[dict]) - list[dict]: if len(history) self.summarize_every: return history oldest history[:self.summarize_every] prompt ( 把下面的对话压缩成一段中文摘要保留用户意图、已确认的关键信息 地址/时间/金额等、未解决的问题。\n\n f对话\n{oldest}\n\n若已有旧摘要请融合{self.summary} ) self.summary llm([{role: user, content: prompt}], modelllm_cfg[summary_model]) return [{role: system, content: f对话摘要{self.summary}}] history[self.summarize_every:]3.3 事实库摘要会丢细节关键事实必须外置滚动摘要最大的坑是把具体数字压没了。摘要变成用户咨询了退货政策但到底是 7 天还是 15 天模型只能瞎猜。解法是给关键事实建独立的事实库压缩前先把事实抽出来单独存。class FactStore: def __init__(self, keep_last: int 20): self.facts: list[str] [] self.keep_last keep_last def extract(self, messages: list[dict]): prompt ( 从对话中抽取所有不可丢失的关键事实 地址、金额、日期、政策条款、用户明确要求每行一条\n f{messages} ) raw llm([{role: user, content: prompt}], modelllm_cfg[summary_model]) for f in raw.strip().splitlines(): f f.strip() if f and f not in self.facts: self.facts.append(f) def render(self) - str: return \n.join(f- {f} for f in self.facts[-self.keep_last:])3.4 注入层分区模板 检索结果排序系统提示词不要写成一大段散文要分区、用明确分隔符。顺序就是注意力优先级角色和任务放最前事实紧随其后摘要垫底。SYSTEM_TEMPLATE 【角色】你是{company}的智能客服助手。 【任务】解答用户问题涉及政策时必须以给定资料为准。 【关键事实】 {facts} 【约束】 1. 不知道就说不知道禁止编造。 2. 金额、日期必须引用原文不得换算或改写。 【对话历史摘要】 {summary} def build_system_prompt(company, fact_store, summarizer) - str: return SYSTEM_TEMPLATE.format( companycompany, factsfact_store.render() or 无, summarysummarizer.summary or 无, ) def inject_search_results(messages, hits, top_k5): top sorted(hits, keylambda h: -h[score])[:top_k] block \n\n.join(f[资料{i1}] {h[text]} for i, h in enumerate(top)) last messages[-1] enriched last[content] f\n\n可参考以下资料\n{block}\n\n请优先依据资料回答资料不足请明说。 return messages[:-1] [{role: last[role], content: enriched}]3.5 统一入口ContextManagerclass ContextManager: def __init__(self, company: str): c cfg[context] self.company company self.budget c[token_budget] self.window SlidingWindow(, c[max_turns]) self.summarizer RollingSummarizer(c[summarize_every]) self.facts FactStore(c[fact_keep_last]) self.top_k c[retrieval_top_k] def add(self, role: str, content: str): self.window.push(role, content) def ask(self, question: str, hits: list[dict] | None None) - str: self.add(user, question) messages self.window.build_messages() if count_tokens(messages) self.budget: self.facts.extract(messages[:self.summarizer.summarize_every]) while count_tokens(messages) self.budget: before len(messages) messages self.summarizer.maybe_summarize(messages) if len(messages) before: messages messages[-self.budget // 4:] messages[-1:] break if hits: messages inject_search_results(messages, hits, self.top_k) messages[0] {role: system, content: build_system_prompt(self.company, self.facts, self.summarizer)} reply llm(messages) self.add(assistant, reply) return reply4. 验证请求压缩触发阈值怎么测配置写完不算完得验证压缩到底有没有在正确的时机触发。我构造了一个 120 轮的长对话测试集含 30 个用户 N 轮前提过的信息追问统计关键信息回忆率和单轮平均 token 消耗。4.1 打印每次请求的 token 分布在ask()里加一行日志观察压缩前后的 token 变化def ask(self, question, hitsNone): self.add(user, question) messages self.window.build_messages() before_tokens count_tokens(messages) if before_tokens self.budget: self.facts.extract(messages[:self.summarizer.summarize_every]) while count_tokens(messages) self.budget: before len(messages) messages self.summarizer.maybe_summarize(messages) if len(messages) before: messages messages[-self.budget // 4:] messages[-1:] break after_tokens count_tokens(messages) print(f[ctx] before{before_tokens} after{after_tokens} fbudget{self.budget} facts{len(self.facts.facts)}) # ... 后续注入与调用跑一轮长对话你会看到类似这样的输出[ctx] before3820 after3820 budget4000 facts0 [ctx] before4150 after2680 budget4000 facts3 [ctx] before4310 after2740 budget4000 facts5第一次before没超预算不触发压缩第二次超了事实库抽了 3 条摘要把 token 压回 2680。这说明阈值工作正常。4.2 四种策略的效果对比我在同一批测试上跑了四种策略结果如下策略关键信息回忆率平均 token/轮备注全量历史直接塞41%8600又贵又蠢常超窗口滑动窗口10 轮58%2200便宜了但早期信息全丢滑动窗口 滚动摘要74%2600记得住大概细节靠猜 关键事实库 检索排序注入91%2800数字、地址不再丢最扎心的对比是第一行和最后一行同一个模型上下文管理做不做回忆率从 41% 涨到 91%成本只有原来的三分之一。模型没换提示词没大改纯粹是信息摆放的差别。4.3 用模型对话快速验证摘要质量摘要质量直接决定压缩效果。我习惯在接入前先去模型对话页面手动测几轮摘要 prompt看看小模型能不能稳定抽出地址、金额、日期这类事实。如果摘要总是丢细节就调整 prompt 里的保留项清单或者把summary_model换成更强的模型。这一步花 10 分钟能省掉后面大量调试。5. 本篇常见错排查5.1 摘要触发太频繁费用反而涨了有人把摘要放在用户每说一句话就触发这是纯浪费。压缩只应该发生在要调用模型生成、且超预算的时刻。检查你的ask()里maybe_summarize是不是被放在了循环外、且只在count_tokens budget时调用。5.2 token 计数口径不统一拿 tiktoken 的计数去卡其他模型的预算会算不准。不同模型的 tokenizer 不一样建议留 20% 余量或者直接用目标模型对应的 encoding。如果预算卡得太死会出现明明没超却触发压缩或者超了却没触发的情况。5.3 事实库无限增长FactStore如果不限制条数跑几百轮后会变成新的膨胀源。我在render()里只取最近 20 条更早的事实要么合并、要么归档。如果你的场景事实很多可以考虑给事实加时间戳按重要性淘汰。5.4 检索结果平铺在中间把召回的 10 段资料不分先后全拼在中间等于把最相关的内容放在模型最看不见的位置。正确做法是按相关度降序只注入 top 5紧贴最后一条用户消息。inject_search_results里的top_k和排序逻辑就是干这个的。5.5 系统提示词被覆盖messages[0]是系统提示词的位置如果你在注入检索结果时不小心改了索引系统规则就丢了。我的做法是最后一步才重建messages[0]确保它永远是分区模板生成的最新版本。6. 把上下文策略固化下来跑通之后建议把 ContextManager 做成项目里的独立模块业务代码只调ask()不碰窗口和压缩逻辑。参数全部走settings.json调阈值不用改代码。如果你的 Agent 要长期跑、还要接 coding 类任务可以考虑用 Coding Plan 把模型调用和额度管理也统一起来避免长会话把额度跑爆。接入通道统一走 TaoToken 的 API一个 Key 覆盖对话、摘要、抽取三类调用配置集中在settings.json里后面换模型或调预算只改一处。接入文档里有完整的参数说明和错误码对照遇到 401/429 这类问题可以直接查。先把这套骨架跑起来再根据你自己的场景调token_budget和summarize_every两个阈值上下文策略就稳了。