Ollama 的 DeepSeek API 怎么接入 Python？保留多轮消息并读取流式回答

Python 接入 Ollama 本地 DeepSeek，使用的是本机 /api/chat 接口。多轮聊天需要客户端保存并重新发送消息历史；流式回答则按每行一个 JSON 对象读取。不能把多个 JSON 块直接交给一次 json.load，也不能把思考字段当成最终回答正文。

确认本地服务和实际模型

准备 Python 3.10 或更新版本，先在终端执行 ollama ls，确认已有 deepseek-r1:7b，且本地 Ollama 正在运行。未安装模型时先完成安装和下载；本文重点是已有模型的 Python 接入，不重复基础安装步骤。

模型名称可在 官方模型库 核对。Chat API 文档 规定 model、messages 和 stream 等请求字段；这里使用本地模型，不填写远端 DeepSeek 平台的 API Key。

先用非流式请求检查字段

在 PowerShell 发一条最小请求，先确认是接口连通问题还是客户端解析问题：

$payload = @{
  model = 'deepseek-r1:7b'
  messages = @(@{role='user'; content='只回答一个数字：17 加 25 等于多少？'})
  stream = $false
} | ConvertTo-Json -Depth 5
$r = Invoke-RestMethod -Method Post -Uri 'http://127.0.0.1:11434/api/chat' `
  -ContentType 'application/json; charset=utf-8' `
  -Body ([Text.Encoding]::UTF8.GetBytes($payload))
$r.done
$r.message.content

验收条件是 done=True，message.content 有答案，并且最终答案可核对。Chat 接口读取 message.content；不要照搬 Generate 接口的顶层 response 字段。

保存两轮消息，并读取流式正文

把下面代码保存为 chat_client.py，执行 python chat_client.py。它只用 Python 标准库，不需要另外安装 SDK。代码中的青禾是人为构造的会话验证例子，不能当成真实业务记录。

"""Article example: Ollama NDJSON chat with bounded history."""
import json
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

MODEL = 'deepseek-r1:7b'
URL = 'http://127.0.0.1:11434/api/chat'

def read_reply(lines):
    content, thinking = [], []
    finished = False
    for line in lines:
        if not line.strip():
            continue
        item = json.loads(line.decode('utf-8'))
        if item.get('error'):
            raise RuntimeError(item['error'])
        message = item.get('message', {})
        text = message.get('content', '')
        if text:
            print(text, end='', flush=True)
            content.append(text)
        if message.get('thinking'):
            thinking.append(message['thinking'])
        if item.get('done') is True:
            if item.get('done_reason') == 'length':
                raise RuntimeError('达到输出上限，回答截断，未记入会话')
            finished = True
            break
    print()
    answer = ''.join(content)
    if not finished or not answer.strip():
        raise RuntimeError('回答中断或最终正文为空，未记入会话')
    reply = {'role': 'assistant', 'content': answer}
    if thinking:
        reply['thinking'] = ''.join(thinking)
    return reply

def chat(history, question):
    pending = history + [{'role': 'user', 'content': question}]
    request = Request(
        URL, data=json.dumps({'model': MODEL, 'messages': pending,
                              'stream': True}).encode('utf-8'),
        headers={'Content-Type': 'application/json'}, method='POST')
    with urlopen(request, timeout=300) as response:
        reply = read_reply(response)
    return pending + [reply]

if __name__ == '__main__':
    messages = []
    try:
        messages = chat(messages, '演示：我的项目代号是青禾。请复述代号。')
        messages = chat(messages, '上一轮的项目代号是什么？只回答代号。')
        print('保存的消息数：', len(messages))
    except (HTTPError, URLError, TimeoutError, ValueError, RuntimeError) as exc:
        print('请求失败：', exc)
        raise SystemExit(1)

依据 Streaming 文档，默认流式返回使用 NDJSON，也就是每行一个 JSON。示例把普通正文块拼接，思考内容另存；只有读到 done=true、未因输出上限截断且最终正文非空，才把这一轮记入历史。官方服务实现把输出长度上限对应的结束原因表示为 length；它也会结束生成，不能只看 done。

第二轮会把第一轮的用户消息与助手回复一起发送。正确完成两轮时，保存的消息数为 4；第二轮应根据历史回答“青禾”。这验证的是应用确实传递了上下文，不证明模型永远能正确记住长对话。

处理失败和长会话

• 连接失败：确认 Ollama 后台已启动，地址和端口一致；此代码只连接本机回环地址。
• 模型不存在：对照 ollama ls 的完整标签，先修正配置，不能用“DeepSeek”泛称替代。
• 已经打印部分文字却中断：示例会报错，失败轮次不会追加到历史。界面应标记该回答未完成，不能把半段文字当成功结果。
• HTTP 错误或 NDJSON 中的 error：保存脱敏错误原文后处理。没有自动重发，避免把同一问题重复入账或重复显示。

这里的 300 秒是网络读操作超时设置，不是整次生成的总时限。长会话不能无限追加历史：根据实际上下文预算裁剪较早的完整轮次，或生成经用户核对的摘要；不要只删用户消息而保留孤立的助手回复。需要持久化时还要区分用户和会话，避免不同人的历史混在一起。

示例验证范围

本文的流解析函数用构造的 NDJSON 数据验证了中文分块拼接、错误块、缺失结束块、输出上限截断以及只有思考没有最终正文的处理。这些是客户端解析检查，不是 DeepSeek 模型实测；未在你的电脑上调用本地模型。接口资料核验于 2026 年 10 月 1 日，实际回答与硬件可运行性需自行验证。