简介《从零实现多轮对话DeepSeek上下文推理模式开发手册》是一份面向开发者的41页实操指南帮助读者基于DeepSeek模型从零搭建具备上下文推理能力的多轮对话系统。内容涵盖基础概念、模型原理与开发准备包括多轮对话定义与特点、上下文推理作用、DeepSeek模型架构优势、开发环境搭建、数据收集清洗与标注划分等。架构与算法部分是重点详细讲解意图识别、上下文推理、回复生成等模块设计并给出注意力机制与记忆网络的PyTorch实现思路从上下文编码、分词处理到模型训练、推理与调优逐步展开。后半部分还覆盖系统测试评估、部署上线、监控优化及常见问题排查包括数据预处理、模型压缩与加速、服务配置优化等细节形成从需求到落地的完整链路。资源包共1个PDF文件大小约2.27MB目录结构清晰已有145人学习下载适合具备Python和深度学习基础的中高级开发者参考。1. 多轮对话的“无状态”真相为什么先做上下文推理模式很多人第一次调 DeepSeek API 时以为多轮对话就是把上一轮的问答原样拼到下一轮。结果没跑几轮界面弹出一句“达到对话长度上限请开启新对话”。这其实不是模型坏了而是上下文窗口被塞满了。所谓上下文推理模式就是开发者主动管理每次请求中放入哪些历史消息、留多少生成空间、超限后怎么降级。这篇文章按一份开发手册的骨架来拆解从最小 API 调用开始到 token 预算、截断、摘要、召回和排错最后落到可测试的 Session 封装。适合正在把 DeepSeek 接入产品、想让对话真正继承上一个用户问题的工程师。2. DeepSeek API 调用从一条消息到多轮对话的最小实现2.1 环境准备用 OpenAI SDK 兼容的方式调 DeepSeek APIDeepSeek API 兼容 OpenAI 的接口协议所以不需要另起一套客户端。常见做法是安装openai库再把base_url指到 DeepSeek 的 API 端点。这样社区里现有的工具链、测试框架和监控脚本都能直接复用。pip install openai export DEEPSEEK_API_KEY你的API Key这里的环境变量名建议统一叫DEEPSEEK_API_KEY与代码解耦。生产环境不要用 export 写进 shell 历史应该放到密钥管理服务里进程启动时从环境变量注入。把base_url显式传入OpenAI构造函数是为了避免某些全局配置或历史环境变量把请求带到别处。2.2 第一行请求messages 决定模型看到什么单轮请求是最小单元所有多轮逻辑都建立在它之上。构建请求时messages是唯一你完全可控的输入结构模型只会依据这里的内容生成回复。from openai import OpenAI client OpenAI( api_key你的API Key, base_urlhttps://api.deepseek.com ) response client.chat.completions.create( modeldeepseek-chat, messages[ {role: system, content: 你是一个中文技术助手。}, {role: user, content: 如何计算上下文 token} ], temperature0.7, max_tokens512, streamFalse ) print(response.choices[0].message.content)这段代码里的messages数组有两种角色system定义模型的人设和行为边界user是当前输入。temperature控制随机性排错时往往调到 0.2 便于复现max_tokens是生成侧上限它不会出现在 prompt 里但会占用上下文总预算。streamFalse表示一次性拿完整结果开发阶段这样最简单。2.3 从单轮到多轮自己维护 session 数组DeepSeek API 本身是无状态的同一个 client 连续发两次请求模型不会自动记住第一次聊了什么。所谓“继承上一个对话”本质是把历史问答按顺序放进新的messages里。我会维护一个会话数组每次调用后自动把助手回复追加进去。class ChatSession: def __init__(self, system_prompt, client, modeldeepseek-chat): self.model model self.client client self.messages [{role: system, content: system_prompt}] def append(self, role, content): self.messages.append({role: role, content: content}) def call(self, temperature0.7, max_tokens512): response self.client.chat.completions.create( modelself.model, messagesself.messages, temperaturetemperature, max_tokensmax_tokens, ) reply response.choices[0].message.content self.append(assistant, reply) return reply, response.usage使用方式很简单session ChatSession(你是后端工程师助手回答要简洁不要重复用户的话。, client) session.append(user, 帮我写一个 Python 重试函数) reply1, _ session.call() session.append(user, 给这个函数加上指数退避参数) reply2, usage session.call()第二次请求时session.messages已经包含system - user - assistant - user四段历史。模型看到前一轮的 assistant 回答才能顺着话头继续。下面这张表是call方法的主要参数实际开发时建议显式传参不要依赖默认值。参数类型作用说明modelstr模型标识以账号开通的模型名为准messageslist完整上下文顺序不能乱否则推理会断temperaturefloat随机性排错用 0.2创意用 0.8max_tokensint生成长度上限要计入上下文总预算streambool是否流式返回开发阶段建议 False做到这里多轮对话的最小闭环已经成立。但messages只增不减对话一长就会撞上长度上限。接下来进入上下文推理模式的核心设计。3. 上下文推理模式消息结构、token 预算与截断策略3.1 system 指令上下文推理模式的“定音鼓”在多轮对话里system消息是唯一可以稳定控制模型行为的位置。它不应该每轮重复而应该在整个会话生命周期里保持稳定。常见做法是把它写成“推理准则”明确告诉模型什么时候该用历史什么时候该说不知道。system_prompt 你是一个部署在生产环境的技术助手。 推理要求 1. 不要重复用户原文。 2. 回答依据只来自 system 指令和下方的对话历史。 3. 如果历史信息不足直接说“需要更多上下文”不要编造。 4. 涉及参数建议时给出具体数值和适用条件。 这段指令把模型的推理边界框在了对话历史里而不是让模型自由发挥。每一个规则都会占用 token所以只保留能约束输出的规则。如果你每轮都把 system 拼在 user 里不仅浪费预算还可能让模型分不清哪条指令更新。3.2 token 预算让 usage 告诉你还剩多少很多开发者只看界面上的“达到对话长度上限”其实这个限制是 prompt 和生成预留共同决定的。response.usage会返回三个字段prompt_tokens、completion_tokens、total_tokens。这里的total_tokens是已经消耗的真正决定下一次请求是否会失败的是context_budget hard_limit - reserve_for_answer其中hard_limit是你在应用内设的上下文上限reserve_for_answer是留给本次回复的最小空间。如果prompt_tokens已经超过context_budget就算把max_tokens设为 1请求也可能失败。def get_usage(response): return { prompt_tokens: response.usage.prompt_tokens, completion_tokens: response.usage.completion_tokens, total_tokens: response.usage.total_tokens, }我一般不会顶着模型窗口上限跑而是留出 10% 到 20% 的安全余量。输出长度不可控流式生成和重试都会让实际消耗偏离预期。下面是我常用的预算参数表参数默认建议含义hard_limit模型窗口的 80%整个请求允许的最大 token 数reserve_for_answer512 到 1024给回复预留的安全空间trim_ratio0.5一次截断要丢掉的过剩比例比如窗口上限 32000hard_limit设 25600reserve_for_answer设 1024那么 prompt 侧的警戒线是 24576。超过这条线就触发截断而不是等到请求报错。3.3 滑动窗口截断从最旧开始丢token 预算算清楚后最直接的降级方案是滑动窗口。它的原则是system永远保留历史消息按“从旧到新”的顺序丢弃直到 prompt 总量回到警戒线内。不要从最新消息开始丢那样模型会失去对当前问题的理解。def estimate_tokens(text): # 粗略估算中文按 1.5 个 token英文按 0.3 个 token zh sum(1 for c in text if \u4e00 c \u9fff) en len(text) - zh return int(zh * 1.5 en * 0.3) def trim_messages(messages, hard_limit, reserve_for_answer1024): budget hard_limit - reserve_for_answer kept [messages[0]] # 保留 system for msg in reversed(messages[1:]): cost estimate_tokens(msg[content]) current_cost sum(estimate_tokens(m[content]) for m in kept) if current_cost cost budget: break kept.insert(1, msg) return kept这里的关键是reversed(messages[1:])从最新历史开始扫描kept.insert(1, msg)把新扫描到的消息插在 system 后面。这样最后得到的kept依然是从旧到新的顺序只是丢掉了最早的一部分历史。current_cost每次重新求和是 O(n^2)对话轮次不多时可以接受轮次上千后维护一个累计 token 值会更高效。截断的优点是零额外 API 调用缺点是粗鲁。用户第一轮说过的目标、约束、偏好可能被当成最旧消息丢掉。要让长对话不失忆单靠截断不够还需要摘要和召回。4. 长对话的记忆摘要压缩与向量召回的双层方案4.1 截断的副作用上下文信息断层滑动窗口适合短会话但用户聊到第 30 轮时最早的目标和决策早就不在窗口里。模型并不是“忘记”而是根本没看到。这就是为什么要在截断之外增加记忆层第一层用摘要压缩全局信息第二层用检索召回局部细节二者互补。4.2 第一层用 DeepSeek 做摘要让历史“压缩”摘要的思路是在会话达到一定轮次后把旧对话送给 DeepSeek 生成一段结构化的摘要再把摘要放进 system同时清掉被摘要覆盖的旧消息。这样既保留了用户目标、明确决策和未完成事项又把 token 消耗从几千降到几百。def summarize_history(messages, budget800): text \n.join(f{m[role]}: {m[content]} for m in messages) prompt f你是上下文压缩器。把下面的对话压缩成中文摘要保留用户目标、明确决策和未完成事项。 对话 {text} 要求不超过{budget}字不要编造。 response client.chat.completions.create( modeldeepseek-chat, messages[{role: user, content: prompt}], temperature0.2, max_tokens1000, ) return response.choices[0].message.contenttemperature0.2是为了让摘要尽量稳定不要每次生成不一样。budget800是摘要正文的中文上限不是 token 上限。摘要本身也是一次模型调用会消耗额外 token所以触发阈值要保守比如超过 20 轮才执行。如果会话已经有过摘要下一次摘要应该把旧摘要连同新历史一起喂进去否则会出现“摘要的摘要”信息越缩越少。4.3 第二层用 SQLite 做轻量向量召回摘要擅长保留全局但不擅长保留具体数字、报错文本、代码片段。这些细节需要检索。生产环境通常会接 embedding 服务加向量库但在项目早期我用 SQLite 存对话片段查询时做关键词打分足以验证召回逻辑是否成立。import sqlite3 class MemoryStore: def __init__(self, db_path): self.conn sqlite3.connect(db_path) self.conn.execute( CREATE TABLE IF NOT EXISTS chunks (id INTEGER PRIMARY KEY, content TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP) ) def add_chunk(self, content): self.conn.execute(INSERT INTO chunks (content) VALUES (?), (content,)) self.conn.commit() def retrieve(self, query, top_k3): rows self.conn.execute(SELECT content FROM chunks).fetchall() scored [] qwords set(query.lower().split()) for (content,) in rows: cwords set(content.lower().split()) score len(qwords cwords) if score 0: scored.append((score, content)) scored.sort(keylambda x: x[0], reverseTrue) return [content for _, content in scored[:top_k]]这个实现里中文不能直接用split分词需要先接入 jieba 或者按字 n-gram 处理。为了让代码可运行我保留了英文式切词中文场景建议把切词结果用空格拼起来再入库。数据量超过几万条后关键词打分会慢那时再切到向量检索接口保持add_chunk和retrieve不变即可。下面这张表是两层记忆的定位差异记忆层覆盖范围代价适用场景摘要压缩全局每次摘要一次模型调用用户目标、偏好、未完成事项检索召回局部查询快需要分词或向量化具体数字、报错文本、代码片段4.4 合并策略摘要进 system召回进上下文有了摘要和召回结果下一步是把它们合并进最终的messages。为了避免出现连续两条 user 破坏多轮结构摘要和相关历史片段都放进system而不是插在 user 队列里。def build_context(system_prompt, summary, recalled, history): parts [] if summary: parts.append(f【全局摘要】\n{summary}) if recalled: parts.append(f【相关历史】\n \n.join(recalled)) if parts: system_prompt system_prompt \n\n \n\n.join(parts) return [{role: system, content: system_prompt}] historyhistory是当前会话中还没有被摘要覆盖的最近轮次。把召回片段放进 system 而不是 user是为了让模型把这些内容当作背景资料而不是新一轮提问。这样模型在推理当前问题时既能看到全局摘要又能看到与当前问题最相关的历史片段。5. 对话长度上限排错错误识别、降级与“开启新对话”的处理5.1 先分清超限的是 prompt 还是 completion遇到“达到对话长度上限请开启新对话”这类提示第一步不是改代码而是看错误发生在哪个阶段。如果请求还没送到模型就报 context length说明prompt_tokens已经超过预算如果生成中途断掉说明completion_tokens或max_tokens不够。常见的排查顺序是这样的先打印response.usage确认最近一次成功请求的 prompt token 趋势再检查hard_limit是否设置得过低最后看错误文本里有没有context length、rate limit等关键字。rate limit 是限流和上下文长度无关不能靠截断解决需要用退避重试。5.2 降级顺序截断、摘要、重置、新对话我在实现降级逻辑时严格按“越便宜越优先”的顺序先截断再摘要最后才让用户开新对话。下面是降级会话的完整骨架。class ContextLengthExceeded(Exception): pass class AutoContextSession: def __init__(self, client, system_prompt, hard_limit32000): self.client client self.system_prompt system_prompt self.hard_limit hard_limit self.history [] def _call(self): messages [ {role: system, content: self.system_prompt} ] self.history response self.client.chat.completions.create( modeldeepseek-chat, messagesmessages, max_tokens512, ) reply response.choices[0].message.content self.history.append({role: assistant, content: reply}) return reply def reply(self, user_message): self.history.append({role: user, content: user_message}) try: return self._call() except ContextLengthExceeded: full_messages [ {role: system, content: self.system_prompt} ] self.history trimmed trim_messages(full_messages, self.hard_limit) self.system_prompt trimmed[0][content] self.history trimmed[1:] try: return self._call() except ContextLengthExceeded: summary summarize_history(self.history[:-1]) self.history [ {role: user, content: user_message} ] self.system_prompt self.system_prompt \n\n【历史摘要】\n summary return self._call()第一级降级只做滑动窗口截断不产生额外模型调用。第二级降级才做摘要因为摘要本身要消耗一次请求。如果摘要后仍然超限说明单条 user 消息太长或者reserve_for_answer设置太小这时候应该返回业务错误提示用户“当前问题上下文过长请开启新对话”。5.3 参数表与日志下面这组参数是我在类似项目里常用的初始值接入时按实际模型窗口调整参数初始值作用hard_limit模型窗口的 80%触发截断的阈值reserve_for_answer512生成预留 tokensummary_trigger_round20历史轮次超过后触发摘要max_recall_items3召回片段数量上限每次降级都要记日志否则线上出了问题根本不知道是哪一层兜住的。JSON 日志比纯文本更适合后续分析和按字段聚合{ event: context_trimmed, before_tokens: 28654, after_tokens: 18320, dropped_messages: 6 }日志里至少要包含event、before_tokens、after_tokens和触发原因。有了这些数据才能回答“为什么用户频繁看到开启新对话”这类问题。6. 开发手册落地接口封装、自动测试与上下文指标监控6.1 统一 Session 接口把前面的截断、摘要、召回封装成一个对外接口业务层不需要关心底层上下文策略。class DeepSeekAppSession: def __init__(self, client, system_prompt, max_context_tokens32000): self.client client self.system_prompt system_prompt self.max_context_tokens max_context_tokens self.history [] self.stats {calls: 0, prompt_tokens: 0, completion_tokens: 0, trims: 0} def ask(self, text): self.history.append({role: user, content: text}) try: return self._complete() except ContextLengthExceeded: self._trim_history() self.stats[trims] 1 return self._complete() def _complete(self): messages [{role: system, content: self.system_prompt}] self.history response self.client.chat.completions.create( modeldeepseek-chat, messagesmessages, max_tokens512, ) reply response.choices[0].message.content self.history.append({role: assistant, content: reply}) self.stats[calls] 1 self.stats[prompt_tokens] response.usage.prompt_tokens self.stats[completion_tokens] response.usage.completion_tokens return reply def _trim_history(self): full_messages [ {role: system, content: self.system_prompt} ] self.history trimmed trim_messages(full_messages, self.max_context_tokens) self.system_prompt trimmed[0][content] self.history trimmed[1:]ask方法对外只接受用户文本内部自动处理上下文预算。stats里累计每次请求的 usage方便后续对接监控系统。_trim_history单独拆出来是为了在单元测试里直接触发截断而不需要真的把一个长对话跑完。6.2 最小自动测试多轮对话最容易出 bug 的地方是消息顺序和上下文丢失。用真实 API 做测试又慢又费钱所以我会用 FakeClient 代替网络调用只验证消息构造逻辑。class FakeClient: def __init__(self): self.last_messages None def chat(self): return self def completions(self): return self def create(self, **kwargs): self.last_messages kwargs[messages] class FakeResponse: class FakeChoice: class FakeMessage: content ok choices [FakeChoice()] class FakeUsage: prompt_tokens 10 completion_tokens 2 usage FakeUsage() return FakeResponse()然后在测试里断言最后一次请求仍然包含第一轮的用户内容def test_session_keeps_two_round_context(): client FakeClient() session DeepSeekAppSession(client, 你是助手, max_context_tokens1000) session.ask(第一轮记住数字7) session.ask(第二轮刚才的数字是多少) assert 记住数字7 in client.last_messages[-1][content]这个测试验证了多轮继承虽然第二轮问题里没有“数字7”但上下文仍然保留着。持续监控stats中的trims也很重要它代表用户真实遇到截断的次数。6.3 把 usage 变成监控指标最后一步是把 usage 接进日志和指标系统。我会在每次请求后输出一行结构化日志包含累计 prompt tokens、单次生成 tokens、截断次数。如果prompt_tokens在快速上升说明上下文管理策略没有生效如果trims频繁说明max_context_tokens设得太小或者摘要触发阈值太晚。import logging def log_session_stats(session): logging.info( calls%d prompt_tokens%d completion_tokens%d trims%d, session.stats[calls], session.stats[prompt_tokens], session.stats[completion_tokens], session.stats[trims], )开发环境里把这个函数放在每次ask之后用日志曲线代替肉眼猜测。我在实际项目中的第一步永远是打印 usage等prompt_tokens和completion_tokens的曲线出来了截断、摘要、召回的阈值才有调整依据而不是靠感觉拍参数。本文还有配套的精品资源点击获取 
长河君 · 2026/9/18 1:12:49 · 阅读 1.2k DeepSeek API多轮对话上下文管理：从token预算到摘要召回
简介《从零实现多轮对话DeepSeek上下文推理模式开发手册》是一份面向开发者的41页实操指南帮助读者基于DeepSeek模型从零搭建具备上下文推理能力的多轮对话系统。内容涵盖基础概念、模型原理与开发准备包括多轮对话定义与特点、上下文推理作用、DeepSeek模型架构优势、开发环境搭建、数据收集清洗与标注划分等。架构与算法部分是重点详细讲解意图识别、上下文推理、回复生成等模块设计并给出注意力机制与记忆网络的PyTorch实现思路从上下文编码、分词处理到模型训练、推理与调优逐步展开。后半部分还覆盖系统测试评估、部署上线、监控优化及常见问题排查包括数据预处理、模型压缩与加速、服务配置优化等细节形成从需求到落地的完整链路。资源包共1个PDF文件大小约2.27MB目录结构清晰已有145人学习下载适合具备Python和深度学习基础的中高级开发者参考。1. 多轮对话的“无状态”真相为什么先做上下文推理模式很多人第一次调 DeepSeek API 时以为多轮对话就是把上一轮的问答原样拼到下一轮。结果没跑几轮界面弹出一句“达到对话长度上限请开启新对话”。这其实不是模型坏了而是上下文窗口被塞满了。所谓上下文推理模式就是开发者主动管理每次请求中放入哪些历史消息、留多少生成空间、超限后怎么降级。这篇文章按一份开发手册的骨架来拆解从最小 API 调用开始到 token 预算、截断、摘要、召回和排错最后落到可测试的 Session 封装。适合正在把 DeepSeek 接入产品、想让对话真正继承上一个用户问题的工程师。2. DeepSeek API 调用从一条消息到多轮对话的最小实现2.1 环境准备用 OpenAI SDK 兼容的方式调 DeepSeek APIDeepSeek API 兼容 OpenAI 的接口协议所以不需要另起一套客户端。常见做法是安装openai库再把base_url指到 DeepSeek 的 API 端点。这样社区里现有的工具链、测试框架和监控脚本都能直接复用。pip install openai export DEEPSEEK_API_KEY你的API Key这里的环境变量名建议统一叫DEEPSEEK_API_KEY与代码解耦。生产环境不要用 export 写进 shell 历史应该放到密钥管理服务里进程启动时从环境变量注入。把base_url显式传入OpenAI构造函数是为了避免某些全局配置或历史环境变量把请求带到别处。2.2 第一行请求messages 决定模型看到什么单轮请求是最小单元所有多轮逻辑都建立在它之上。构建请求时messages是唯一你完全可控的输入结构模型只会依据这里的内容生成回复。from openai import OpenAI client OpenAI( api_key你的API Key, base_urlhttps://api.deepseek.com ) response client.chat.completions.create( modeldeepseek-chat, messages[ {role: system, content: 你是一个中文技术助手。}, {role: user, content: 如何计算上下文 token} ], temperature0.7, max_tokens512, streamFalse ) print(response.choices[0].message.content)这段代码里的messages数组有两种角色system定义模型的人设和行为边界user是当前输入。temperature控制随机性排错时往往调到 0.2 便于复现max_tokens是生成侧上限它不会出现在 prompt 里但会占用上下文总预算。streamFalse表示一次性拿完整结果开发阶段这样最简单。2.3 从单轮到多轮自己维护 session 数组DeepSeek API 本身是无状态的同一个 client 连续发两次请求模型不会自动记住第一次聊了什么。所谓“继承上一个对话”本质是把历史问答按顺序放进新的messages里。我会维护一个会话数组每次调用后自动把助手回复追加进去。class ChatSession: def __init__(self, system_prompt, client, modeldeepseek-chat): self.model model self.client client self.messages [{role: system, content: system_prompt}] def append(self, role, content): self.messages.append({role: role, content: content}) def call(self, temperature0.7, max_tokens512): response self.client.chat.completions.create( modelself.model, messagesself.messages, temperaturetemperature, max_tokensmax_tokens, ) reply response.choices[0].message.content self.append(assistant, reply) return reply, response.usage使用方式很简单session ChatSession(你是后端工程师助手回答要简洁不要重复用户的话。, client) session.append(user, 帮我写一个 Python 重试函数) reply1, _ session.call() session.append(user, 给这个函数加上指数退避参数) reply2, usage session.call()第二次请求时session.messages已经包含system - user - assistant - user四段历史。模型看到前一轮的 assistant 回答才能顺着话头继续。下面这张表是call方法的主要参数实际开发时建议显式传参不要依赖默认值。参数类型作用说明modelstr模型标识以账号开通的模型名为准messageslist完整上下文顺序不能乱否则推理会断temperaturefloat随机性排错用 0.2创意用 0.8max_tokensint生成长度上限要计入上下文总预算streambool是否流式返回开发阶段建议 False做到这里多轮对话的最小闭环已经成立。但messages只增不减对话一长就会撞上长度上限。接下来进入上下文推理模式的核心设计。3. 上下文推理模式消息结构、token 预算与截断策略3.1 system 指令上下文推理模式的“定音鼓”在多轮对话里system消息是唯一可以稳定控制模型行为的位置。它不应该每轮重复而应该在整个会话生命周期里保持稳定。常见做法是把它写成“推理准则”明确告诉模型什么时候该用历史什么时候该说不知道。system_prompt 你是一个部署在生产环境的技术助手。 推理要求 1. 不要重复用户原文。 2. 回答依据只来自 system 指令和下方的对话历史。 3. 如果历史信息不足直接说“需要更多上下文”不要编造。 4. 涉及参数建议时给出具体数值和适用条件。 这段指令把模型的推理边界框在了对话历史里而不是让模型自由发挥。每一个规则都会占用 token所以只保留能约束输出的规则。如果你每轮都把 system 拼在 user 里不仅浪费预算还可能让模型分不清哪条指令更新。3.2 token 预算让 usage 告诉你还剩多少很多开发者只看界面上的“达到对话长度上限”其实这个限制是 prompt 和生成预留共同决定的。response.usage会返回三个字段prompt_tokens、completion_tokens、total_tokens。这里的total_tokens是已经消耗的真正决定下一次请求是否会失败的是context_budget hard_limit - reserve_for_answer其中hard_limit是你在应用内设的上下文上限reserve_for_answer是留给本次回复的最小空间。如果prompt_tokens已经超过context_budget就算把max_tokens设为 1请求也可能失败。def get_usage(response): return { prompt_tokens: response.usage.prompt_tokens, completion_tokens: response.usage.completion_tokens, total_tokens: response.usage.total_tokens, }我一般不会顶着模型窗口上限跑而是留出 10% 到 20% 的安全余量。输出长度不可控流式生成和重试都会让实际消耗偏离预期。下面是我常用的预算参数表参数默认建议含义hard_limit模型窗口的 80%整个请求允许的最大 token 数reserve_for_answer512 到 1024给回复预留的安全空间trim_ratio0.5一次截断要丢掉的过剩比例比如窗口上限 32000hard_limit设 25600reserve_for_answer设 1024那么 prompt 侧的警戒线是 24576。超过这条线就触发截断而不是等到请求报错。3.3 滑动窗口截断从最旧开始丢token 预算算清楚后最直接的降级方案是滑动窗口。它的原则是system永远保留历史消息按“从旧到新”的顺序丢弃直到 prompt 总量回到警戒线内。不要从最新消息开始丢那样模型会失去对当前问题的理解。def estimate_tokens(text): # 粗略估算中文按 1.5 个 token英文按 0.3 个 token zh sum(1 for c in text if \u4e00 c \u9fff) en len(text) - zh return int(zh * 1.5 en * 0.3) def trim_messages(messages, hard_limit, reserve_for_answer1024): budget hard_limit - reserve_for_answer kept [messages[0]] # 保留 system for msg in reversed(messages[1:]): cost estimate_tokens(msg[content]) current_cost sum(estimate_tokens(m[content]) for m in kept) if current_cost cost budget: break kept.insert(1, msg) return kept这里的关键是reversed(messages[1:])从最新历史开始扫描kept.insert(1, msg)把新扫描到的消息插在 system 后面。这样最后得到的kept依然是从旧到新的顺序只是丢掉了最早的一部分历史。current_cost每次重新求和是 O(n^2)对话轮次不多时可以接受轮次上千后维护一个累计 token 值会更高效。截断的优点是零额外 API 调用缺点是粗鲁。用户第一轮说过的目标、约束、偏好可能被当成最旧消息丢掉。要让长对话不失忆单靠截断不够还需要摘要和召回。4. 长对话的记忆摘要压缩与向量召回的双层方案4.1 截断的副作用上下文信息断层滑动窗口适合短会话但用户聊到第 30 轮时最早的目标和决策早就不在窗口里。模型并不是“忘记”而是根本没看到。这就是为什么要在截断之外增加记忆层第一层用摘要压缩全局信息第二层用检索召回局部细节二者互补。4.2 第一层用 DeepSeek 做摘要让历史“压缩”摘要的思路是在会话达到一定轮次后把旧对话送给 DeepSeek 生成一段结构化的摘要再把摘要放进 system同时清掉被摘要覆盖的旧消息。这样既保留了用户目标、明确决策和未完成事项又把 token 消耗从几千降到几百。def summarize_history(messages, budget800): text \n.join(f{m[role]}: {m[content]} for m in messages) prompt f你是上下文压缩器。把下面的对话压缩成中文摘要保留用户目标、明确决策和未完成事项。 对话 {text} 要求不超过{budget}字不要编造。 response client.chat.completions.create( modeldeepseek-chat, messages[{role: user, content: prompt}], temperature0.2, max_tokens1000, ) return response.choices[0].message.contenttemperature0.2是为了让摘要尽量稳定不要每次生成不一样。budget800是摘要正文的中文上限不是 token 上限。摘要本身也是一次模型调用会消耗额外 token所以触发阈值要保守比如超过 20 轮才执行。如果会话已经有过摘要下一次摘要应该把旧摘要连同新历史一起喂进去否则会出现“摘要的摘要”信息越缩越少。4.3 第二层用 SQLite 做轻量向量召回摘要擅长保留全局但不擅长保留具体数字、报错文本、代码片段。这些细节需要检索。生产环境通常会接 embedding 服务加向量库但在项目早期我用 SQLite 存对话片段查询时做关键词打分足以验证召回逻辑是否成立。import sqlite3 class MemoryStore: def __init__(self, db_path): self.conn sqlite3.connect(db_path) self.conn.execute( CREATE TABLE IF NOT EXISTS chunks (id INTEGER PRIMARY KEY, content TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP) ) def add_chunk(self, content): self.conn.execute(INSERT INTO chunks (content) VALUES (?), (content,)) self.conn.commit() def retrieve(self, query, top_k3): rows self.conn.execute(SELECT content FROM chunks).fetchall() scored [] qwords set(query.lower().split()) for (content,) in rows: cwords set(content.lower().split()) score len(qwords cwords) if score 0: scored.append((score, content)) scored.sort(keylambda x: x[0], reverseTrue) return [content for _, content in scored[:top_k]]这个实现里中文不能直接用split分词需要先接入 jieba 或者按字 n-gram 处理。为了让代码可运行我保留了英文式切词中文场景建议把切词结果用空格拼起来再入库。数据量超过几万条后关键词打分会慢那时再切到向量检索接口保持add_chunk和retrieve不变即可。下面这张表是两层记忆的定位差异记忆层覆盖范围代价适用场景摘要压缩全局每次摘要一次模型调用用户目标、偏好、未完成事项检索召回局部查询快需要分词或向量化具体数字、报错文本、代码片段4.4 合并策略摘要进 system召回进上下文有了摘要和召回结果下一步是把它们合并进最终的messages。为了避免出现连续两条 user 破坏多轮结构摘要和相关历史片段都放进system而不是插在 user 队列里。def build_context(system_prompt, summary, recalled, history): parts [] if summary: parts.append(f【全局摘要】\n{summary}) if recalled: parts.append(f【相关历史】\n \n.join(recalled)) if parts: system_prompt system_prompt \n\n \n\n.join(parts) return [{role: system, content: system_prompt}] historyhistory是当前会话中还没有被摘要覆盖的最近轮次。把召回片段放进 system 而不是 user是为了让模型把这些内容当作背景资料而不是新一轮提问。这样模型在推理当前问题时既能看到全局摘要又能看到与当前问题最相关的历史片段。5. 对话长度上限排错错误识别、降级与“开启新对话”的处理5.1 先分清超限的是 prompt 还是 completion遇到“达到对话长度上限请开启新对话”这类提示第一步不是改代码而是看错误发生在哪个阶段。如果请求还没送到模型就报 context length说明prompt_tokens已经超过预算如果生成中途断掉说明completion_tokens或max_tokens不够。常见的排查顺序是这样的先打印response.usage确认最近一次成功请求的 prompt token 趋势再检查hard_limit是否设置得过低最后看错误文本里有没有context length、rate limit等关键字。rate limit 是限流和上下文长度无关不能靠截断解决需要用退避重试。5.2 降级顺序截断、摘要、重置、新对话我在实现降级逻辑时严格按“越便宜越优先”的顺序先截断再摘要最后才让用户开新对话。下面是降级会话的完整骨架。class ContextLengthExceeded(Exception): pass class AutoContextSession: def __init__(self, client, system_prompt, hard_limit32000): self.client client self.system_prompt system_prompt self.hard_limit hard_limit self.history [] def _call(self): messages [ {role: system, content: self.system_prompt} ] self.history response self.client.chat.completions.create( modeldeepseek-chat, messagesmessages, max_tokens512, ) reply response.choices[0].message.content self.history.append({role: assistant, content: reply}) return reply def reply(self, user_message): self.history.append({role: user, content: user_message}) try: return self._call() except ContextLengthExceeded: full_messages [ {role: system, content: self.system_prompt} ] self.history trimmed trim_messages(full_messages, self.hard_limit) self.system_prompt trimmed[0][content] self.history trimmed[1:] try: return self._call() except ContextLengthExceeded: summary summarize_history(self.history[:-1]) self.history [ {role: user, content: user_message} ] self.system_prompt self.system_prompt \n\n【历史摘要】\n summary return self._call()第一级降级只做滑动窗口截断不产生额外模型调用。第二级降级才做摘要因为摘要本身要消耗一次请求。如果摘要后仍然超限说明单条 user 消息太长或者reserve_for_answer设置太小这时候应该返回业务错误提示用户“当前问题上下文过长请开启新对话”。5.3 参数表与日志下面这组参数是我在类似项目里常用的初始值接入时按实际模型窗口调整参数初始值作用hard_limit模型窗口的 80%触发截断的阈值reserve_for_answer512生成预留 tokensummary_trigger_round20历史轮次超过后触发摘要max_recall_items3召回片段数量上限每次降级都要记日志否则线上出了问题根本不知道是哪一层兜住的。JSON 日志比纯文本更适合后续分析和按字段聚合{ event: context_trimmed, before_tokens: 28654, after_tokens: 18320, dropped_messages: 6 }日志里至少要包含event、before_tokens、after_tokens和触发原因。有了这些数据才能回答“为什么用户频繁看到开启新对话”这类问题。6. 开发手册落地接口封装、自动测试与上下文指标监控6.1 统一 Session 接口把前面的截断、摘要、召回封装成一个对外接口业务层不需要关心底层上下文策略。class DeepSeekAppSession: def __init__(self, client, system_prompt, max_context_tokens32000): self.client client self.system_prompt system_prompt self.max_context_tokens max_context_tokens self.history [] self.stats {calls: 0, prompt_tokens: 0, completion_tokens: 0, trims: 0} def ask(self, text): self.history.append({role: user, content: text}) try: return self._complete() except ContextLengthExceeded: self._trim_history() self.stats[trims] 1 return self._complete() def _complete(self): messages [{role: system, content: self.system_prompt}] self.history response self.client.chat.completions.create( modeldeepseek-chat, messagesmessages, max_tokens512, ) reply response.choices[0].message.content self.history.append({role: assistant, content: reply}) self.stats[calls] 1 self.stats[prompt_tokens] response.usage.prompt_tokens self.stats[completion_tokens] response.usage.completion_tokens return reply def _trim_history(self): full_messages [ {role: system, content: self.system_prompt} ] self.history trimmed trim_messages(full_messages, self.max_context_tokens) self.system_prompt trimmed[0][content] self.history trimmed[1:]ask方法对外只接受用户文本内部自动处理上下文预算。stats里累计每次请求的 usage方便后续对接监控系统。_trim_history单独拆出来是为了在单元测试里直接触发截断而不需要真的把一个长对话跑完。6.2 最小自动测试多轮对话最容易出 bug 的地方是消息顺序和上下文丢失。用真实 API 做测试又慢又费钱所以我会用 FakeClient 代替网络调用只验证消息构造逻辑。class FakeClient: def __init__(self): self.last_messages None def chat(self): return self def completions(self): return self def create(self, **kwargs): self.last_messages kwargs[messages] class FakeResponse: class FakeChoice: class FakeMessage: content ok choices [FakeChoice()] class FakeUsage: prompt_tokens 10 completion_tokens 2 usage FakeUsage() return FakeResponse()然后在测试里断言最后一次请求仍然包含第一轮的用户内容def test_session_keeps_two_round_context(): client FakeClient() session DeepSeekAppSession(client, 你是助手, max_context_tokens1000) session.ask(第一轮记住数字7) session.ask(第二轮刚才的数字是多少) assert 记住数字7 in client.last_messages[-1][content]这个测试验证了多轮继承虽然第二轮问题里没有“数字7”但上下文仍然保留着。持续监控stats中的trims也很重要它代表用户真实遇到截断的次数。6.3 把 usage 变成监控指标最后一步是把 usage 接进日志和指标系统。我会在每次请求后输出一行结构化日志包含累计 prompt tokens、单次生成 tokens、截断次数。如果prompt_tokens在快速上升说明上下文管理策略没有生效如果trims频繁说明max_context_tokens设得太小或者摘要触发阈值太晚。import logging def log_session_stats(session): logging.info( calls%d prompt_tokens%d completion_tokens%d trims%d, session.stats[calls], session.stats[prompt_tokens], session.stats[completion_tokens], session.stats[trims], )开发环境里把这个函数放在每次ask之后用日志曲线代替肉眼猜测。我在实际项目中的第一步永远是打印 usage等prompt_tokens和completion_tokens的曲线出来了截断、摘要、召回的阈值才有调整依据而不是靠感觉拍参数。本文还有配套的精品资源点击获取