WaveSpeedAI

DeepSeek V4 API Python: Минимальные примеры кода с потоковой передачей

DeepSeek V4 API Python: Минимальные примеры кода с потоковой передачей

Привет! Я здесь Дора. Всё началось с небольшого раздражения: я всё время копировал одну и ту же шаблонную строку для чат-интерфейса между проектами, меняя базовые URL и названия моделей как этикетки на банках. Не сложная работа, просто та, что добавляет трение в твой день. Я давно видел DeepSeek и стал любопытен, поэтому я выделил несколько утренних часов в конце января 2026 года, чтобы подключить их API “V4” в мой стек Python и посмотреть, как это будет ощущаться в реальной работе.

Я не гнался за бенчмарками. Я хотел узнать: клиент не помешает мне, могу ли я надёжно потоковать, и ошибки ли падают так, чтобы было легко их понять? Вот что я попробовал, что меня спотыкнуло и что тихо сработало. Поехали!

Настройка окружения

Зависимости

Я держал настройку простой на macOS с Python 3.11. Можно делать это с использованием стандартной библиотеки, но три маленьких пакета облегчили жизнь:

  • requests (прямолинейный HTTP: достаточно хорош для большинства случаев)
  • httpx (асинхронность и таймауты, которые ведут себя хорошо)
  • python-dotenv (чтобы я не разбрасывал ключи) Если вы планируете потоковать с помощью Server-Sent Events, вы можете использовать requests и анализировать строки самостоятельно (что я и делал), или принесите помощник вроде sseclient-py. Я придерживался requests, меньше движущихся частей.

Установка

pip install requests httpx python-dotenv

Я также создал минимальное виртуальное окружение для каждого проекта. Это скучный совет, но он спасает вас от дрейфа зависимостей, когда вы вернётесь сюда через три месяца.

Конфигурация ключа API

Я сохранил ключ в переменной окружения. Ничего фантастического:

# .env
DEEPSEEK_API_KEY=your_key_here

Затем в Python:

from dotenv import load_dotenv
import os

load_dotenv()

API_KEY = os.getenv("DEEPSEEK_API_KEY")

if not API_KEY:
    raise RuntimeError("Missing DEEPSEEK_API_KEY")

Два небольших примечания из настройки:

  • Базовый URL и названия моделей меняются чаще, чем вы думаете. Я проверял официальную документацию DeepSeek API перед каждым запуском, чтобы подтвердить пути и доступные модели.
  • Я держал таймауты явными. Это привычка, которая окупается, когда вы попадаете на ограничения скорости или сетевой шум.

Базовый запрос чата

Ментальная модель знакома, если вы использовали чат-интерфейсы где-либо ещё. DeepSeek предоставляет конечную точку чата с messages=[{"role": "...", "content": "..."}]. Это полезно, потому что я не должен был переформатировать мои подсказки. Вот минимальный запрос, который я использовал с requests. Названия моделей варьируются в зависимости от аккаунта и региона, во время моих тестов я видел ссылки вроде deepseek-chat и deepseek-reasoner. Если ваша документация упоминает строку модели “V4”, используйте её. В противном случае выберите ближайшую модель общего назначения, указанную в вашей консоли.

import os
import requests

API_KEY = os.environ["DEEPSEEK_API_KEY"]
BASE_URL = "https://api.deepseek.com/v1/chat/completions"

payload = {
    "model": "deepseek-chat", # check docs/console for the exact model
    "messages": [
        {"role": "system", "content": "You are a concise assistant."},
        {"role": "user", "content": "Give me two bullet points on the value of clear commit messages."}
    ],
    "temperature": 0.3,
    "max_tokens": 200
}

resp = requests.post(
    BASE_URL,
    headers={
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    },
    json=payload,
    timeout=30
)

resp.raise_for_status()

data = resp.json()
content = data["choices"][0]["message"]["content"]
print(content)

Заметки в поле

  • Первый запуск был без происшествий (облегчение). Структура соответствовала тому, что я ожидал, что ускорило миграцию небольшой библиотеки подсказок.
  • Я держал температуру низкой для повторяемых ответов. Это звучит очевидно, но я всё ещё забываю, когда я борюсь с ошибками.
  • Если вам нужны детерминированные запуски, также установите top_p и seed, если API это поддерживает. Когда документация молчит, я предполагаю недетерминированность.

Если вы сравниваете провайдеров, преимущество здесь в низком трении. Недостаток в том, что различия скрываются на краях, в ошибочных нагрузках, подсчёте токенов и форме потока. Эти края — это то, где ваша интеграция либо ощущается прочной, либо раздражающей.

Пример генерации кода

Я не прошу модели писать полные модули. Это становится работой по уборке. Но для небольших помощников, вроде “разбери этот формат отметки времени” или “набросай SQL с заполнителями”, это удобно. Я использовал узкую подсказку, чёткий контракт и небольшие лимиты вывода. Это помешало модели блуждать и облегчило просмотр различий.

import requests, os

API_KEY = os.environ["DEEPSEEK_API_KEY"]
BASE_URL = "https://api.deepseek.com/v1/chat/completions"

messages = [
    {"role": "system", "content": (
        "You generate small, safe Python helpers. "
        "Return only code inside one block."
    )},
    {"role": "user", "content": (
        "Write a Python function `parse_yyyymmdd` that takes a string like '2026-01-31' "
        "and returns a datetime.date. If invalid, return None. No external deps."
    )}
]

resp = requests.post(
    BASE_URL,
    headers={
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    },
    json={
        "model": "deepseek-chat", # or your V4-capable model
        "messages": messages,
        "temperature": 0,
        "max_tokens": 250
    },
    timeout=30
)

resp.raise_for_status()
code = resp.json()["choices"][0]["message"]["content"]
print(code)

Что помогло на практике

  • Я всегда говорю ему вернуть только код. Если я пропущу это, я получу заключительные предложения, которые мне не нужны.
  • Температура 0 снижает утомительные правки.
  • Я всё равно прочитал логику. В моём запуске он обработал ValueError, но я всё ещё добавил дополнительный тест на пробелы. Два дополнительных минут сейчас экономит часы сюрпризов позже.

Это не сэкономило время с первого раза. После трёх или четырёх небольших помощников я заметил, что это снизило умственные усилия: меньше переключений вкладок, меньше “какой снова точный код strptime?” моментов. Этого для меня достаточно.

Потоковые ответы

Мне нравится потоковать для любой подсказки, которая может вырасти. Это позволяет мне вывалиться рано, если ответ уходит в сторону, и это делает длинные ответы менее тяжёлыми. Потоковая передача DeepSeek использовала обычный паттерн в моих тестах: установите stream=true и читайте строки данных до [DONE]. Мне не нужен был специальный клиент, requests с iter_lines был достаточен.

import os, json, requests

API_KEY = os.environ["DEEPSEEK_API_KEY"]
BASE_URL = "https://api.deepseek.com/v1/chat/completions"

payload = {
    "model": "deepseek-chat",
    "messages": [
        {"role": "system", "content": "Be brief."},
        {"role": "user", "content": "Summarize this: Streaming keeps the UI responsive and lets me stop early."}
    ],
    "stream": True,
    "temperature": 0.2,
}

with requests.post(
    BASE_URL,
    headers={
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    },
    json=payload,
    stream=True,
    timeout=60
) as r:
    r.raise_for_status()
    for line in r.iter_lines(decode_unicode=True):
        if not line:
            continue
        if line.startswith("data: "):
            chunk = line[len("data: "):]
            if chunk == "[DONE]":
                break
            try:
                obj = json.loads(chunk)
                delta = obj["choices"][0]["delta"].get("content", "")
                if delta:
                    print(delta, end="", flush=True)
            except json.JSONDecodeError:
                # I keep a small log when this happens: usually network blips
                pass

print()

Два небольших поведения, которые мне понравились:

  • Ранние токены прибывали быстро (секунда или две на чистом соединении). Не научно, просто достаточно, чтобы ощущаться резко, когда я подключил это к инструменту CLI.
  • Маркер [DONE] появлялся надёжно. Это звучит тривиально, пока не произойдёт, отсутствующие маркеры завершения зависают интерфейсы.

Если вам нужно потоковать в веб-приложение, я бы поместил тонкий слой сервера между ними для нормализации событий. Это один дополнительный шаг, но он держит ваш фронтенд простым.

Server-Sent Events

Под капотом вы фактически читаете Server-Sent Events. Если вы предпочитаете помощника, sseclient-py работает, но писать свой здесь в порядке, пока вы защищаетесь от частичных строк и таймаутов. Страница документации о потоковой передаче в документации DeepSeek API было достаточно, чтобы заставить это работать без сюрпризов.

Обработка ошибок

Большинство моих ошибок были предсказуемы: отсутствующий ключ, неправильное название модели или таймауты, когда я дросселировал мою сеть, чтобы имитировать путешествие Wi-Fi. Небольшой паттерн, который я переиспользую:

import httpx, time, os

API_KEY = os.environ["DEEPSEEK_API_KEY"]
BASE_URL = "https://api.deepseek.com/v1/chat/completions"

RETRIABLE = {408, 409, 425, 429, 500, 502, 503, 504}

async def chat_once(client, messages):
    resp = await client.post(
        BASE_URL,
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={
            "model": "deepseek-chat",
            "messages": messages,
            "temperature": 0.2,
            "max_tokens": 300,
        },
        timeout=30,
    )
    if resp.status_code == 401:
        raise RuntimeError("Unauthorized. Check DEEPSEEK_API_KEY and account access.")
    if resp.status_code == 404:
        raise RuntimeError("Endpoint or model not found. Confirm model name in console/docs.")
    if resp.status_code in RETRIABLE:
        raise RuntimeError(f"Retryable status: {resp.status_code}")
    resp.raise_for_status()
    return resp.json()

async def chat_with_retries(messages, attempts=4):
    backoff = 0.5
    async with httpx.AsyncClient() as client:
        for i in range(attempts):
            try:
                return await chat_once(client, messages)
            except RuntimeError as e:
                msg = str(e)
                if "Retryable status" in msg and i < attempts - 1:
                    time.sleep(backoff)
                    backoff *= 2
                    continue
                raise

Несколько практических примечаний:

  • Ограничения скорости: я видел 429, когда я запускал параллельные тесты. Экспоненциальный отступ помог, но я также добавил небольшую дрожь (случайные 50–150мс), чтобы избежать грохочущих стад.
  • Гигиена таймаутов: я установил более короткие таймауты подключения/чтения для быстрых проверок (5–10с) и более длинные для больших подсказок. Таймауты не должны быть все 30 сек по умолчанию: это скрывает проблемы.
  • Ошибочные нагрузки: когда вещи не удавались, тело JSON включало сообщение, которое я мог поверхностно отнести к логу. Я всё ещё оборачиваю его в свои собственные исключения, поэтому я контролирую, что достигает интерфейса.

Если ваша кодовая база уже говорит на схеме в стиле OpenAI, это управляемо: та же форма сообщения, немного разные края. Главное — быть строгим в отношении названий моделей и логировать полное тело ответа на не-2xx, чтобы вы не угадывали. В отношении документации я полагался на официальную документацию DeepSeek API для названий параметров и форму потока. Всякий раз, когда провайдер использует знакомые конечные точки, соблазнительно предположить паритет. Я научился проверять документацию в первую очередь и копировать меньше между клиентами, чем я думаю, что я могу.

Кому может понравиться это

  • Если у вас уже есть обёртка Python для завершения чата, путь миграции мягок.
  • Если вам небезразлична потоковая передача и простые повторы, это ведёт себя предсказуемо.
  • Если вам нужны очень специфические инструменты (схемы вызова функций, маркеры рассуждений или пакетные задания), вам нужно будет внимательно прочитать документацию и создать прототип с одной узкой задачей, прежде чем вы будете обязаны.

Я не пытался организовать длинные, многошаговые агенты здесь. Я сосредоточился на небольших, ежедневных подсказках, тех, которые снимают трение. Вот где DeepSeek V4 API с Python ощущался достаточно стабильным, чтобы держать.