企业官网建设流程全解析

简介：《从零实现多轮对话：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 API

DeepSeek 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_url="https://api.deepseek.com" ) response = client.chat.completions.create( model="deepseek-chat", messages=[ {"role": "system", "content": "你是一个中文技术助手。"}, {"role": "user", "content": "如何计算上下文 token？"} ], temperature=0.7, max_tokens=512, stream=False ) print(response.choices[0].message.content)

这段代码里的messages数组有两种角色：system定义模型的人设和行为边界，user是当前输入。temperature控制随机性，排错时往往调到 0.2 便于复现；max_tokens是生成侧上限，它不会出现在 prompt 里，但会占用上下文总预算。stream=False表示一次性拿完整结果，开发阶段这样最简单。

2.3 从单轮到多轮：自己维护 session 数组

DeepSeek API 本身是无状态的，同一个 client 连续发两次请求，模型不会自动记住第一次聊了什么。所谓“继承上一个对话”，本质是把历史问答按顺序放进新的messages里。我会维护一个会话数组，每次调用后自动把助手回复追加进去。

class ChatSession: def __init__(self, system_prompt, client, model="deepseek-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, temperature=0.7, max_tokens=512): response = self.client.chat.completions.create( model=self.model, messages=self.messages, temperature=temperature, max_tokens=max_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方法的主要参数，实际开发时建议显式传参，不要依赖默认值。

做到这里，多轮对话的最小闭环已经成立。但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% 的安全余量。输出长度不可控，流式生成和重试都会让实际消耗偏离预期。下面是我常用的预算参数表：

比如窗口上限 32000，hard_limit设 25600，reserve_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_answer=1024): 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, budget=800): text = "\n".join(f"{m['role']}: {m['content']}" for m in messages) prompt = f"""你是上下文压缩器。把下面的对话压缩成中文摘要，保留用户目标、明确决策和未完成事项。 对话： {text} 要求：不超过{budget}字，不要编造。""" response = client.chat.completions.create( model="deepseek-chat", messages=[{"role": "user", "content": prompt}], temperature=0.2, max_tokens=1000, ) return response.choices[0].message.content

temperature=0.2是为了让摘要尽量稳定，不要每次生成不一样。budget=800是摘要正文的中文上限，不是 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_k=3): 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(key=lambda x: x[0], reverse=True) 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}] + history

history是当前会话中还没有被摘要覆盖的最近轮次。把召回片段放进 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_limit=32000): 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( model="deepseek-chat", messages=messages, max_tokens=512, ) 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 参数表与日志

下面这组参数是我在类似项目里常用的初始值，接入时按实际模型窗口调整：

每次降级都要记日志，否则线上出了问题根本不知道是哪一层兜住的。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_tokens=32000): 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( model="deepseek-chat", messages=messages, max_tokens=512, ) 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_tokens=1000) 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的曲线出来了，截断、摘要、召回的阈值才有调整依据，而不是靠感觉拍参数。

本文还有配套的精品资源，点击获取