接 DeepSeek 做智能问答核心就四件事存 API Key、流式响应不要让客户等、会话上下文按客户隔离存、价格控制避免账单失控。这篇给一个能直接跑的最小对接方案。一、API Key 与基础调用deepseek-chat 模型DeepSeek V3 用 /chat/completions 接口模型名 deepseek-chat入参 messages 数组role 为 system/user/assistant。微信回调拿到消息后构造 messages 数组历史滑窗 当前 user 消息调接口拿到回复后 POST 回微信。API Key 不要硬编码在代码里——放环境变量或本地加密配置文件。二、流式响应边生成边回不让客户等DeepSeek 接口支持 streamtrue 参数返回 SSEServer-Sent Events流。做法是拿到流后一边解析一边构造回复文本解析完成后一次性 POST 回微信微信接口目前不支持流式发送只能整包发。流式处理能显著降低客户等待感知——虽然最终还是整包发但程序侧能更及时拿到完整回复避免接口超时截断。三、会话上下文按客户隔离存滑窗每个客户 wxid 独立维护自己的滑窗最近 8 轮对话微信回调进来先查该客户的滑窗历史、拼当前消息、再调 DeepSeek。上下文存 Redis按 wxid 存、TTL 30 分钟30 分钟没互动就清空当成新会话。别把多个客户的上下文混在一起——每个 wxid 独立一条记录避免 A 的历史污染 B 的回复。四、价格控制每日上限 单轮 token 限DeepSeek 按 token 计费V3 约 1 元 100 万 tokens 输入用量不大但还是要控。做法是两个限制每日总 token 上限比如 50 万 token/天达到就降级成 FAQ 关键词回复不再调模型单轮 prompt response 总 token 不超 2000避免单次调用被刷爆。token 统计从 DeepSeek 响应里的 usage 字段拿每天零点重置。DeepSeek 对接对照环节做什么关键点常见坑基础调用/chat/completions deepseek-chatKey 存环境变量硬编码泄露流式响应streamtrue 解析 SSE微信端整包发超时截断上下文隔离每 wxid 独立 Redis 滑窗TTL 30min 自动清多客户混上下文价格控制每日 50 万 token 上限超限降级 FAQ账单失控DeepSeek 对接实现import os, time, redis, json, requests class DeepSeekWechatBot: DEEPSEEK_URL https://api.deepseek.com/v1/chat/completions MODEL deepseek-chat MAX_DAILY_TOKENS 500_000 MAX_TURN_TOKENS 2000 CONTEXT_TTL 1800 # 30min CONTEXT_WINDOW 8 # 最近 8 轮 def __init__(self, wx_api): self.wx wx_api self.key os.environ[DEEPSEEK_KEY] self.rdb redis.Redis() def on_message(self, event): if event.get(from_me): return wxid event[from] content event[content] # 1. 价格闸门今日 token 余额 used int(self.rdb.get(ds:today_tokens) or 0) if used self.MAX_DAILY_TOKENS: self._fallback_faq(wxid, content) return # 2. 取该客户的上下文滑窗 history self.rdb.lrange(fctx:{wxid}, 0, -1) messages [{role: system, content: 你是一个专业的客服助手}] for h in history: messages.append(json.loads(h)) messages.append({role: user, content: content}) # 3. 调 DeepSeek reply, usage self._call_deepseek(messages) if not reply: self._fallback_faq(wxid, content); return # 4. 发回微信 self.wx.post(/send, { receiver: wxid, content: reply[:500]}).json() # 5. 更新上下文 扣 token self._update_context(wxid, content, reply) self.rdb.incrby(ds:today_tokens, usage.get(total_tokens, 0)) def _call_deepseek(self, messages): # 流式调用解析 SSE headers {Authorization: fBearer {self.key}, Content-Type: application/json} payload {model: self.MODEL, messages: messages, stream: True, max_tokens: self.MAX_TURN_TOKENS} r requests.post(self.DEEPSEEK_URL, headersheaders, jsonpayload, streamTrue, timeout30) reply_parts, last_usage [], {} for line in r.iter_lines(decode_unicodeTrue): if not line or not line.startswith(data:): continue data line[5:].strip() if data [DONE]: break chunk json.loads(data) delta chunk[choices][0].get(delta, {}) if content in delta and delta[content]: reply_parts.append(delta[content]) if chunk.get(usage): last_usage chunk[usage] return .join(reply_parts), last_usage def _update_context(self, wxid, user_text, bot_text): key fctx:{wxid} self.rdb.rpush(key, json.dumps({role: user, content: user_text})) self.rdb.rpush(key, json.dumps({role: assistant, content: bot_text})) # 滑窗裁剪 while self.rdb.llen(key) self.CONTEXT_WINDOW * 2: self.rdb.lpop(key); self.rdb.lpop(key) self.rdb.expire(key, self.CONTEXT_TTL) def _fallback_faq(self, wxid, content): ...落地建议接 DeepSeek 做智能问答四件事别省API Key 存环境变量不硬编码、流式响应解析 SSE 流微信端整包发、每个客户 wxid 独立存滑窗上下文避免混、每日 token 上限 50 万超了降级 FAQ。每日零点记得重置 token 计数。微信侧消息收发由Eyun这类平台提供DeepSeek 对接、上下文隔离、价格控制逻辑在自建服务实现接口字段以平台开发文档为准。
接 DeepSeek 做智能问答核心就四件事存 API Key、流式响应不要让客户等、会话上下文按客户隔离存、价格控制避免账单失控。这篇给一个能直接跑的最小对接方案。一、API Key 与基础调用deepseek-chat 模型DeepSeek V3 用 /chat/completions 接口模型名 deepseek-chat入参 messages 数组role 为 system/user/assistant。微信回调拿到消息后构造 messages 数组历史滑窗 当前 user 消息调接口拿到回复后 POST 回微信。API Key 不要硬编码在代码里——放环境变量或本地加密配置文件。二、流式响应边生成边回不让客户等DeepSeek 接口支持 streamtrue 参数返回 SSEServer-Sent Events流。做法是拿到流后一边解析一边构造回复文本解析完成后一次性 POST 回微信微信接口目前不支持流式发送只能整包发。流式处理能显著降低客户等待感知——虽然最终还是整包发但程序侧能更及时拿到完整回复避免接口超时截断。三、会话上下文按客户隔离存滑窗每个客户 wxid 独立维护自己的滑窗最近 8 轮对话微信回调进来先查该客户的滑窗历史、拼当前消息、再调 DeepSeek。上下文存 Redis按 wxid 存、TTL 30 分钟30 分钟没互动就清空当成新会话。别把多个客户的上下文混在一起——每个 wxid 独立一条记录避免 A 的历史污染 B 的回复。四、价格控制每日上限 单轮 token 限DeepSeek 按 token 计费V3 约 1 元 100 万 tokens 输入用量不大但还是要控。做法是两个限制每日总 token 上限比如 50 万 token/天达到就降级成 FAQ 关键词回复不再调模型单轮 prompt response 总 token 不超 2000避免单次调用被刷爆。token 统计从 DeepSeek 响应里的 usage 字段拿每天零点重置。DeepSeek 对接对照环节做什么关键点常见坑基础调用/chat/completions deepseek-chatKey 存环境变量硬编码泄露流式响应streamtrue 解析 SSE微信端整包发超时截断上下文隔离每 wxid 独立 Redis 滑窗TTL 30min 自动清多客户混上下文价格控制每日 50 万 token 上限超限降级 FAQ账单失控DeepSeek 对接实现import os, time, redis, json, requests class DeepSeekWechatBot: DEEPSEEK_URL https://api.deepseek.com/v1/chat/completions MODEL deepseek-chat MAX_DAILY_TOKENS 500_000 MAX_TURN_TOKENS 2000 CONTEXT_TTL 1800 # 30min CONTEXT_WINDOW 8 # 最近 8 轮 def __init__(self, wx_api): self.wx wx_api self.key os.environ[DEEPSEEK_KEY] self.rdb redis.Redis() def on_message(self, event): if event.get(from_me): return wxid event[from] content event[content] # 1. 价格闸门今日 token 余额 used int(self.rdb.get(ds:today_tokens) or 0) if used self.MAX_DAILY_TOKENS: self._fallback_faq(wxid, content) return # 2. 取该客户的上下文滑窗 history self.rdb.lrange(fctx:{wxid}, 0, -1) messages [{role: system, content: 你是一个专业的客服助手}] for h in history: messages.append(json.loads(h)) messages.append({role: user, content: content}) # 3. 调 DeepSeek reply, usage self._call_deepseek(messages) if not reply: self._fallback_faq(wxid, content); return # 4. 发回微信 self.wx.post(/send, { receiver: wxid, content: reply[:500]}).json() # 5. 更新上下文 扣 token self._update_context(wxid, content, reply) self.rdb.incrby(ds:today_tokens, usage.get(total_tokens, 0)) def _call_deepseek(self, messages): # 流式调用解析 SSE headers {Authorization: fBearer {self.key}, Content-Type: application/json} payload {model: self.MODEL, messages: messages, stream: True, max_tokens: self.MAX_TURN_TOKENS} r requests.post(self.DEEPSEEK_URL, headersheaders, jsonpayload, streamTrue, timeout30) reply_parts, last_usage [], {} for line in r.iter_lines(decode_unicodeTrue): if not line or not line.startswith(data:): continue data line[5:].strip() if data [DONE]: break chunk json.loads(data) delta chunk[choices][0].get(delta, {}) if content in delta and delta[content]: reply_parts.append(delta[content]) if chunk.get(usage): last_usage chunk[usage] return .join(reply_parts), last_usage def _update_context(self, wxid, user_text, bot_text): key fctx:{wxid} self.rdb.rpush(key, json.dumps({role: user, content: user_text})) self.rdb.rpush(key, json.dumps({role: assistant, content: bot_text})) # 滑窗裁剪 while self.rdb.llen(key) self.CONTEXT_WINDOW * 2: self.rdb.lpop(key); self.rdb.lpop(key) self.rdb.expire(key, self.CONTEXT_TTL) def _fallback_faq(self, wxid, content): ...落地建议接 DeepSeek 做智能问答四件事别省API Key 存环境变量不硬编码、流式响应解析 SSE 流微信端整包发、每个客户 wxid 独立存滑窗上下文避免混、每日 token 上限 50 万超了降级 FAQ。每日零点记得重置 token 计数。微信侧消息收发由Eyun这类平台提供DeepSeek 对接、上下文隔离、价格控制逻辑在自建服务实现接口字段以平台开发文档为准。
` 这一行漏掉了一个临时变量)
