用原生 Python 完成两轮 DeepSeek 对话

2026-09-09 00:00    #AI   #DeepSeek   #Python   #LLM  

这一篇的目标很具体:不装任何 SDK,只用 Python 标准库向 DeepSeek 的 /chat/completions 发请求,完成一次「提问 → 回答 → 追问 → 回答」的两轮对话,并搞清楚多轮到底是怎么拼出来的。

先把结论写在前面:

核心结论

/chat/completions无状态接口,服务端不保存你的上下文。 多轮对话的本质是:客户端自己维护一个 messages 数组,每轮把「历史全部 + 新提问」一起发过去。 所以「第二轮请求体比第一轮长」不是 bug,而是多轮对话的全部机制。

接口的形状

方法 / 路径POST /chat/completions
Base URL(OpenAI 格式)https://api.deepseek.com
完整地址https://api.deepseek.com/chat/completions
认证Authorization: Bearer ${DEEPSEEK_API_KEY}
Content-Typeapplication/json
模型deepseek-flashdeepseek-v4-pro

请求体最小可用形态只有两个必填字段:

1{
2  "model": "deepseek-flash",
3  "messages": [{ "role": "user", "content": "你好" }]
4}

messages 里每条消息是一个对象,rolesystem / user / assistant / toolcontent 是文本(或内容块数组,用于图片)。

剩下的字段都是可选的,本篇会用到三个:

字段作用本篇取值
thinking思考模式开关,{"type": "enabled"|"disabled"}enabled
reasoning_effort思考强度,none/low/high/maxhigh
stream是否用 SSE 流式返回false
thinking 放哪取决于调用方式

直接发 HTTP 时,thinking请求体的顶层字段。 如果用的是 OpenAI SDK,则要放进 extra_body,因为 SDK 不认识这个非标准字段(官方文档里那句 extra_body={"thinking": {...}} 就是这个原因)。 另外 reasoning_effort 只在思考模式下有意义;思考模式下 temperaturefrequency_penaltypresence_penalty 都不生效,设置了也不报错,只是被忽略。

第一轮:一次请求

urllib 发请求,核心就是把 dict 序列化成 JSON 再 POST 出去。注意错误处理必须读 exc.read():DeepSeek 的真实错误原因在响应体里,只看状态码会一头雾水。

 1import json, os, sys, urllib.error, urllib.request
 2
 3API_URL = "https://api.deepseek.com/chat/completions"
 4MODEL = "deepseek-flash"
 5
 6
 7def chat(messages, *, model=MODEL, thinking=True, reasoning_effort="high",
 8         max_tokens=None, timeout=120):
 9    api_key = os.environ.get("DEEPSEEK_API_KEY")
10    if not api_key:
11        sys.exit("缺少环境变量 DEEPSEEK_API_KEY")
12
13    body = {
14        "model": model,
15        "messages": messages,
16        # raw HTTP 下 thinking 是顶层字段,不需要 SDK 的 extra_body
17        "thinking": {"type": "enabled" if thinking else "disabled"},
18        "stream": False,
19    }
20    if thinking:
21        body["reasoning_effort"] = reasoning_effort  # 只有思考模式才接受
22    if max_tokens is not None:
23        body["max_tokens"] = max_tokens
24
25    request = urllib.request.Request(
26        API_URL,
27        data=json.dumps(body).encode("utf-8"),
28        headers={
29            "Content-Type": "application/json",
30            "Authorization": f"Bearer {api_key}",
31        },
32        method="POST",
33    )
34
35    try:
36        with urllib.request.urlopen(request, timeout=timeout) as resp:
37            return json.loads(resp.read().decode("utf-8"))
38    except urllib.error.HTTPError as exc:
39        # 4xx/5xx 的正文里才有真正的错误信息,不要只看状态码
40        detail = exc.read().decode("utf-8", "replace")
41        raise SystemExit(f"HTTP {exc.code} {exc.reason}\n{detail}") from exc
42    except urllib.error.URLError as exc:
43        raise SystemExit(f"网络错误:{exc.reason}") from exc

第一轮只要一条 user 消息:

1messages = [{"role": "user", "content": "世界上最高的山是哪一座?"}]
2resp = chat(messages)

响应结构里我们关心两块:choices[0].messageusage。下面是一份真实字段结构的示例(字段名与层级来自官方文档,具体数值仅为示意):

 1{
 2  "id": "…",
 3  "object": "chat.completion",
 4  "created": 1758268800,
 5  "model": "deepseek-flash",
 6  "choices": [
 7    {
 8      "index": 0,
 9      "finish_reason": "stop",
10      "message": {
11        "role": "assistant",
12        "content": "世界上最高的山是珠穆朗玛峰……",
13        "reasoning_content": "用户问的是世界最高峰,这属于地理常识……"
14      }
15    }
16  ],
17  "usage": {
18    "prompt_tokens": 12,
19    "completion_tokens": 260,
20    "total_tokens": 272,
21    "prompt_tokens_details": {
22      "cached_tokens": 0,
23      "prompt_cache_hit_tokens": 0,
24      "prompt_cache_miss_tokens": 12
25    },
26    "completion_tokens_details": { "reasoning_tokens": 128 }
27  }
28}

几个容易忽略的点:

第二轮:把回答塞回历史

这里是新手最容易翻车的地方:不能只发新的那一句,必须把第一轮的 assistant 回答也带上,否则模型根本不知道你在追问什么。

choices[0].message 本身就是一条合法的 assistant 消息(它带着 role: assistant),但里面还有 reasoning_contenttool_calls 等字段。这里显式只取 rolecontent 构造干净的历史项,行为更可控:

 1def take_assistant_message(response):
 2    """从响应里取出 assistant 消息。
 3
 4    返回两项:可直接 append 回 messages 的历史项,以及原始 message
 5    (后者用于打印 reasoning_content 和 content)。
 6    """
 7    message = response["choices"][0]["message"]
 8    history_entry = {
 9        "role": "assistant",
10        "content": message.get("content"),
11    }
12    return history_entry, message
为什么不用 messages.append(response.choices[0].message)

官方样例常用这种简写,它会把 reasoning_content 等字段一并塞进历史(虽然不带 tools 时会被忽略)。 本文明式构造 {"role", "content"} 两项,历史更干净,以后要加字段也清楚该加在哪里。

于是两轮的主流程只有 8 行关键代码:

 1# 第一轮:只带一条 user 消息
 2messages = [{"role": "user", "content": "世界上最高的山是哪一座?"}]
 3
 4resp = chat(messages)
 5assistant_entry, message = take_assistant_message(resp)
 6messages.append(assistant_entry)   # 关键:把模型回答写回历史
 7
 8# 第二轮:第一轮的历史 + 新的追问,一起发过去
 9messages.append({"role": "user", "content": "那第二高的是哪一座?"})
10
11resp = chat(messages)
12assistant_entry, message = take_assistant_message(resp)
13messages.append(assistant_entry)

两轮结束后,messages 的长度是 4:

1[0] user      世界上最高的山是哪一座?
2[1] assistant 世界上最高的山是珠穆朗玛峰……
3[2] user      那第二高的是哪一座?
4[3] assistant 世界第二高峰是乔戈里峰(K2)……

对应到网络层,两次请求的 body 分别是:

1Round 1 → messages 长度 1
2  [{ user: 世界上最高的山是哪一座? }]
3
4Round 2 → messages 长度 3     ← 多出的两条就是第一轮的往返
5  [{ user:     世界上最高的山是哪一座? },
6   { assistant: 世界上最高的山是珠穆朗玛峰…… },
7   { user:     那第二高的是哪一座? }]
为什么大模型"记得"上下文

模型本身没有记忆。「记得」的错觉来自每一轮都把整段历史重新喂了一遍。 代价是 token 随轮数线性增长(第二轮输入 ≈ 第一轮输入 + 第一轮输出 + 新提问), 好处是这段前缀不变,能稳定命中上下文缓存,实际计费比表面上便宜。

关于 reasoning_content 要不要回传

这是个值得单独记住的规则,搞错了会直接报错:

请求是否携带 tools历史轮次的 reasoning_content
不带 tools(本篇情况)不用回传;即使传了也会被忽略,不进入上下文
tools(工具调用)必须完整回传,否则 API 返回 400

所以本篇的 take_assistant_message 丢掉 reasoning_content 是完全正确的做法——不浪费 token,也不改变模型行为。一旦以后要写工具调用,就必须改成回传完整 message。

另外注意:思考模式下的 token 消耗包含思维链。usage.completion_tokens_details.reasoning_tokens 就是思维链部分,它同样按输出 token 计费。所以 reasoning_efforthigh 换成 max 不只是变慢,是真的更贵。

完整可运行文件

代码放在 deepseek_dialog.py,除了上面讲的内容,还加了:

  1#!/usr/bin/env python3
  2"""用 Python 标准库调用 DeepSeek Chat Completions API,完成两轮对话。
  3
  4不依赖 openai SDK,只用 urllib + json。
  5运行前先导出 API key:
  6
  7    export DEEPSEEK_API_KEY="sk-..."
  8    python3 deepseek_dialog.py
  9"""
 10
 11import json
 12import os
 13import sys
 14import urllib.error
 15import urllib.request
 16
 17API_URL = "https://api.deepseek.com/chat/completions"
 18MODEL = "deepseek-flash"
 19
 20
 21def chat(messages, *, model=MODEL, thinking=True, reasoning_effort="high",
 22         max_tokens=None, timeout=120):
 23    """发一次 /chat/completions 请求,返回解析后的 JSON 字典。
 24
 25    messages 是本轮为止的完整对话历史;API 无状态,历史由调用方维护。
 26    """
 27    api_key = os.environ.get("DEEPSEEK_API_KEY")
 28    if not api_key:
 29        sys.exit("缺少环境变量 DEEPSEEK_API_KEY")
 30
 31    body = {
 32        "model": model,
 33        "messages": messages,
 34        # 思考模式:raw HTTP 下 thinking 是顶层字段,不需要 SDK 的 extra_body
 35        "thinking": {"type": "enabled" if thinking else "disabled"},
 36        "stream": False,
 37    }
 38    if thinking:
 39        # 只有思考模式才接受 reasoning_effort;强度 none/low/high/max
 40        body["reasoning_effort"] = reasoning_effort
 41    if max_tokens is not None:
 42        body["max_tokens"] = max_tokens
 43
 44    request = urllib.request.Request(
 45        API_URL,
 46        data=json.dumps(body).encode("utf-8"),
 47        headers={
 48            "Content-Type": "application/json",
 49            "Authorization": f"Bearer {api_key}",
 50        },
 51        method="POST",
 52    )
 53
 54    try:
 55        with urllib.request.urlopen(request, timeout=timeout) as resp:
 56            return json.loads(resp.read().decode("utf-8"))
 57    except urllib.error.HTTPError as exc:
 58        # 4xx/5xx 的正文里才有真正的错误信息,不要只看状态码
 59        detail = exc.read().decode("utf-8", "replace")
 60        raise SystemExit(f"HTTP {exc.code} {exc.reason}\n{detail}") from exc
 61    except urllib.error.URLError as exc:
 62        raise SystemExit(f"网络错误:{exc.reason}") from exc
 63
 64
 65def take_assistant_message(response):
 66    """从响应里取出 assistant 消息,转成可直接 append 回 messages 的 dict。"""
 67    message = response["choices"][0]["message"]
 68    history_entry = {
 69        "role": "assistant",
 70        "content": message.get("content"),
 71    }
 72    return history_entry, message
 73
 74
 75def print_round(label, message, usage):
 76    reasoning = message.get("reasoning_content")
 77    if reasoning:
 78        print(f"--- {label} 思维链(reasoning_content)---")
 79        print(reasoning)
 80    print(f"--- {label} 回答(content)---")
 81    print(message.get("content"))
 82    print(
 83        f"[usage] prompt={usage['prompt_tokens']} "
 84        f"cached={usage['prompt_tokens_details']['cached_tokens']} "
 85        f"completion={usage['completion_tokens']} "
 86        f"total={usage['total_tokens']}"
 87    )
 88    print()
 89
 90
 91def main():
 92    # 第一轮:只带一条 user 消息
 93    messages = [
 94        {"role": "user", "content": "世界上最高的山是哪一座?"},
 95    ]
 96
 97    resp = chat(messages)
 98    assistant_entry, message = take_assistant_message(resp)
 99    messages.append(assistant_entry)  # 关键:把模型回答写回历史
100    print_round("Round 1", message, resp["usage"])
101
102    # 第二轮:第一轮的历史 + 新的追问,一起发过去
103    messages.append({"role": "user", "content": "那第二高的是哪一座?"})
104
105    resp = chat(messages)
106    assistant_entry, message = take_assistant_message(resp)
107    messages.append(assistant_entry)
108    print_round("Round 2", message, resp["usage"])
109
110    # 两轮结束后 messages 里有 4 条消息
111    print("最终 messages 结构:")
112    for i, m in enumerate(messages):
113        preview = (m.get("content") or "")[:30].replace("\n", " ")
114        print(f"  [{i}] {m['role']:<9} {preview}...")
115    print(f"\nmessages 长度 = {len(messages)}")
116
117
118if __name__ == "__main__":
119    main()

运行方式:

1export DEEPSEEK_API_KEY="sk-..."
2python3 deepseek_dialog.py

只需要标准库,Python 3.8+ 都能跑,不需要 pip install openai

验证方式

本机没有配置 API key,无法请求真实端点,因此用一个本地 mock server 验证了请求形状与两轮拼接逻辑,断言如下:

mock 校验通过后,接入真实 key 时只需确认 API Key 有效即可,代码路径不变。

小结