接 DeepSeek 做智能问答，核心就四件事：存 API Key、流式响应不要让客户等、会话上下文按客户隔离存、价格控制避免账单失控。这篇给一个能直接跑的最小对接方案。
一、API Key 与基础调用：deepseek-chat 模型
DeepSeek V3 用 /chat/completions 接口，模型名 deepseek-chat，入参 messages 数组（role 为 system/user/assistant）。微信回调拿到消息后，构造 messages 数组（历史滑窗 + 当前 user 消息）调接口，拿到回复后 POST 回微信。
API Key 不要硬编码在代码里——放环境变量或本地加密配置文件。
二、流式响应：边生成边回，不让客户等
DeepSeek 接口支持 stream=true 参数返回 SSE（Server-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-chat | Key 存环境变量 | 硬编码泄露 |
流式响应 | stream=true 解析 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(f"ctx:{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": f"Bearer {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, headers=headers, json=payload, stream=True, timeout=30) reply_parts, last_usage = [], {} for line in r.iter_lines(decode_unicode=True): 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 = f"ctx:{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 对接、上下文隔离、价格控制逻辑在自建服务实现，接口字段以平台开发文档为准。
