DeepSeek API 流式聊天：分离思考与回答并统计 Token

写一个命令行聊天工具：流式输出，把思考过程和最终答案分开显示，支持多轮对话，显示每轮的 tokens 用量。同一份代码既能连 DeepSeek 官方 API，也能连本地 Ollama 上的 DeepSeek-R1 蒸馏模型，换环境变量就行。

先看清成品

• 测试： pytest 1 项通过，覆盖思考字段的兼容处理。
• 本地实测： 用本地 deepseek-r1-distill:1.5b 跑完一道鸡兔同笼题，接着追问一轮，两轮都答对了（兔 4、鸡 6），但第二轮没有遵守“只回答数字”。第一轮 24.5 秒、思考 1014 字；第二轮 10.4 秒。
• 调试示例： Ollama 和官方 API 的思考字段名不一样，导致思考内容全部丢失（见调试示例）。

没有在 box 上跑的部分： box 上没有 DeepSeek API Key，所以没有调用过 DeepSeek 官方 API。下面的实测结果都来自同一份代码连接本地 Ollama（同样是 OpenAI 兼容接口，模型是 DeepSeek-R1-Distill-Qwen-1.5B 的 Q4_K_M 量化版）。换成官方 API 只需要改环境变量，但效果和速度会和本地小模型差别很大。

合规提醒：API Key 只放在环境变量里，不要写进代码或提交到 Git。不要把身份证号、客户资料、公司机密发给云端 API；这类数据用 B 篇的本地部署处理。如果要把它做成面向公众的服务，需要按《生成式人工智能服务管理暂行办法》《人工智能生成合成内容标识办法》等规定履行相应义务。API 价格以 DeepSeek 官网“模型 & 价格”页为准，本文不列价格。

做完之后你会得到什么？

需要准备哪些环境？

截至本文写作时，官方文档的首页写明：base_url 是 https://api.deepseek.com，模型名是 deepseek-flash 和 deepseek-v4-pro。思考模式通过 thinking: {"type": "enabled"} 开启，思考内容在 reasoning_content 字段里。不带 tools 的多轮对话不需要回传 reasoning_content，传了也会被忽略。模型名以后可能会变，以文档为准。

文件结构

41-deepseek-A-src/
├── ds_chat.py
└── test_chat.py

第 1 步：代码

ds_chat.py：

"""命令行多轮对话：DeepSeek 官方 API 和本地 Ollama 通用（两者都兼容 OpenAI 接口）。
思考过程和最终答案分开显示；按官方文档，不带 tools 的多轮对话里，历史轮次只回传 content，不回传思考过程。
环境变量：
  官方 API：DEEPSEEK_API_KEY（必填），DS_MODEL 默认 deepseek-flash
  本地：    DS_BASE_URL=http://127.0.0.1:11434/v1  DS_MODEL=deepseek-r1-distill:1.5b（不需要 key）
用法：python ds_chat.py               交互模式，输入 /exit 退出
      python ds_chat.py "问题1" "问题2"  依次提问（方便脚本里测试）"""
import os, sys, time
from openai import OpenAI

BASE = os.environ.get("DS_BASE_URL", "https://api.deepseek.com")
MODEL = os.environ.get("DS_MODEL", "deepseek-flash")
client = OpenAI(base_url=BASE, api_key=os.environ.get("DEEPSEEK_API_KEY") or "local-no-key")

def reasoning_of(delta):
    # DeepSeek 官方返回 reasoning_content；Ollama 的 OpenAI 兼容接口返回 reasoning
    return getattr(delta, "reasoning_content", None) or getattr(delta, "reasoning", None) or ""

def ask(messages, show_thinking=True):
    extra = {} if "11434" in BASE else {"thinking": {"type": "enabled"}}
    t0 = time.time(); think, answer, usage = [], [], None
    stream = client.chat.completions.create(model=MODEL, messages=messages, stream=True,
                                            stream_options={"include_usage": True}, extra_body=extra)
    for chunk in stream:
        if chunk.usage: usage = chunk.usage
        if not chunk.choices: continue
        d = chunk.choices[0].delta
        r = reasoning_of(d)
        if r:
            if not think and show_thinking: print("\033[2m[思考] ", end="")
            think.append(r); show_thinking and print(r, end="", flush=True)
        if d.content:
            if think and not answer and show_thinking: print("\033[0m")
            answer.append(d.content); print(d.content, end="", flush=True)
    print(f"\n\033[2m（{time.time()-t0:.1f}s；思考 {len(''.join(think))} 字；"
          f"tokens：输入 {usage.prompt_tokens if usage else '?'} / 输出 {usage.completion_tokens if usage else '?'}）\033[0m")
    return "".join(answer).strip()

def main():
    messages = [{"role": "system", "content": "你是一个简洁的中文助手。"}]
    questions = sys.argv[1:] or iter(lambda: input("\n你> "), "/exit")
    for q in questions:
        if not q.strip(): continue
        if sys.argv[1:]: print(f"\n你> {q}")
        messages.append({"role": "user", "content": q})
        messages.append({"role": "assistant", "content": ask(messages)})   # 只回传最终答案

if __name__ == "__main__":
    main()

test_chat.py：

from types import SimpleNamespace as N
import os; os.environ.setdefault("DS_BASE_URL", "http://127.0.0.1:11434/v1")
from ds_chat import reasoning_of

def test_reasoning_field_both_providers():
    assert reasoning_of(N(reasoning_content="官方", content=None)) == "官方"
    assert reasoning_of(N(reasoning="本地", content=None)) == "本地"
    assert reasoning_of(N(content="答案")) == ""

第 2 步：运行

# 官方 API
$ export DEEPSEEK_API_KEY=sk-...
$ python ds_chat.py
# 本地 Ollama（box 上实测的就是这种方式）
$ DS_BASE_URL=http://127.0.0.1:11434/v1 DS_MODEL=deepseek-r1-distill:1.5b python ds_chat.py \
    "一个笼子里有鸡和兔共 10 只，脚共 28 只，兔有几只？" "那鸡有几只？只回答数字"

你> 一个笼子里有鸡和兔共 10 只，脚共 28 只，兔有几只？
[思考] 嗯，今天遇到了一个有趣的问题……设兔子的数量为R，那么鸡的数量就是10 - R……2R = 8……所以，答案应该是兔子有4只。
因此，兔子的数量是 **4只**。
（24.5s；思考 1014 字；tokens：输入 39 / 输出 899）

你> 那鸡有几只？只回答数字
[思考] 嗯，好的，我现在得解决这个鸡兔同笼的问题……所以，答案应该是鸡有6只，兔子有4只。
鸡有6只，兔子有4只。
（10.4s；思考 580 字；tokens：输入 237 / 输出 429）

检查点：第二轮输入的 tokens 从 39 涨到了 237，说明第一轮的答案被带进了上下文，但思考内容没有带进去（只回传 content）。

调试示例：思考内容“消失”了

第一版只读官方文档里的 reasoning_content 字段。连接本地 Ollama 时，灰色的思考部分一直是空的。我写了个小脚本统计流式返回的字段：

chunks=830 reasoning_content字数=0 reasoning字数=2300

原来 Ollama 的 OpenAI 兼容接口把思考内容放在 reasoning 字段里，DeepSeek 官方 API 用的是 reasoning_content。reasoning_of() 两个字段都读，问题就解决了。经验是：说是“兼容 OpenAI 接口”，扩展字段也可能各不相同，换一家服务就先打印一个 chunk 看看实际结构。

常见问题

模板规则

• 思考内容只显示、不回传（不带 tools 时）。
• 换服务商先打印一个原始 chunk，确认字段名。
• 每轮都显示 tokens，费用心里有数。

项目资源

学会把思考过程和答案分开处理之后，再换成其他推理模型（包括 B 篇的本地模型）也是同样的写法。