Пример API DeepSeek на Python: чат, потоковая передача, инструменты

Установка и отправка первого запроса

Этот сервис не является официальным API DeepSeek. Это размещенный эндпоинт, совместимый с OpenAI, который использует одну модель с открытыми весами с идентификатором uncensored, управляемую независимым провайдером. Поскольку он следует формату OpenAI, стандартный Python SDK работает без изменений, кроме URL и ключа.

Copypip install openai

Получите ключ на странице API-ключ. После регистрации вы можете одним нажатием получить пробный баланс $0,50 без привязки карты, действительный в течение 7 дней. Храните ключ в переменной окружения, а не в файле исходного кода:

Copyexport API_KEY="YOUR_KEY"

Copyimport os
from openai import OpenAI

client = OpenAI(
    base_url="https://api.deepseekapi.top/v1",
    api_key=os.environ["API_KEY"],
)

resp = client.chat.completions.create(
    model="uncensored",
    messages=[
        {"role": "system", "content": "You are a concise assistant."},
        {"role": "user", "content": "Explain list comprehensions in two sentences."},
    ],
    temperature=0.7,
)

print(resp.choices[0].message.content)
print(resp.usage.prompt_tokens, resp.usage.completion_tokens)

Два показателя использования — это то, за что вам выставят счет. Умножьте их на цены, чтобы узнать стоимость любого вызова: входные токены × $0,15 и выходные токены × $0,50, каждое значение делится на 1 000 000.

Использование асинхронного клиента

Для веб-серверов и всего, что обрабатывает множество пользователей, используйте асинхронный клиент. Он имеет те же параметры, что и синхронный, и является готовой заменой в большинстве кода.

Copyimport asyncio
import os
from openai import AsyncOpenAI

client = AsyncOpenAI(
    base_url="https://api.deepseekapi.top/v1",
    api_key=os.environ["API_KEY"],
)

sem = asyncio.Semaphore(8)

async def ask(prompt):
    async with sem:
        r = await client.chat.completions.create(
            model="uncensored",
            messages=[{"role": "user", "content": prompt}],
        )
        return r.choices[0].message.content

async def main():
    prompts = ["Define idempotent.", "Define monotonic.", "Define pure function."]
    answers = await asyncio.gather(*(ask(p) for p in prompts))
    for a in answers:
        print(a)

asyncio.run(main())

Семафор установлен на 8, потому что это лимит параллельных запросов для полного ключа. Для пробного ключа используйте 2. Без ограничения серия задач приведет к ошибкам concurrency, даже если каждый запрос действителен.

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

Установите stream=True, и эндпоинт отправляет события, отправляемые сервером. Каждый чанк содержит небольшой фрагмент текста в delta.content. Запрос использования означает, что последний чанк сообщает о количестве токенов.

Copy# client = OpenAI(base_url="https://api.deepseekapi.top/v1", ...) as above
stream = client.chat.completions.create(
    model="uncensored",
    messages=[{"role": "user", "content": "Write a haiku about debugging."}],
    stream=True,
    stream_options={"include_usage": True},
)

for chunk in stream:
    if chunk.choices and chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)
    if chunk.usage:
        print("\n", chunk.usage.prompt_tokens, chunk.usage.completion_tokens)

Две детали часто вызывают трудности. Последний чанк использования может иметь пустой список choices, поэтому цикл сначала проверяет chunk.choices. И потоковый запрос удерживает слот параллельных запросов до конца потока, поэтому аккуратно выходите из циклов и закрывайте потоки, которые вы прерываете. Полный ключ позволяет 8 параллельных запросов; пробный ключ позволяет 2.

JSON-режим

Для вывода, который будет парсить программа, используйте response_format с json_object. Также укажите слово JSON в промпте и опишите желаемую структуру. Модели нужна эта инструкция; сам флаг не предоставляет схему.

Copy# client = OpenAI(base_url="https://api.deepseekapi.top/v1", ...) as above
import json

resp = client.chat.completions.create(
    model="uncensored",
    messages=[
        {"role": "system", "content": "Reply with JSON only: {\"title\": str, \"tags\": [str]}"},
        {"role": "user", "content": "Summarize: a guide to caching HTTP responses."},
    ],
    response_format={"type": "json_object"},
    temperature=0,
)

data = json.loads(resp.choices[0].message.content)
print(data["title"], data["tags"])

Оберните json.loads в try/except. Если ответ обрезан лимитом токенов, он не распарсится, поэтому избегайте малого max_tokens с JSON. Установка temperature в 0 и seed делает повторные запуски более воспроизводимыми, так как оба параметра передаются.

Вызов функций

API принимает tools и tool_choice, и возвращает tool_calls в сообщении ассистента. Вы выполняете функцию самостоятельно и отправляете результат обратно как сообщение с role: "tool".

Copy# client = OpenAI(base_url="https://api.deepseekapi.top/v1", ...) as above
import json

tools = [{
    "type": "function",
    "function": {
        "name": "add",
        "description": "Add two numbers.",
        "parameters": {
            "type": "object",
            "properties": {"a": {"type": "number"}, "b": {"type": "number"}},
            "required": ["a", "b"],
        },
    },
}]

messages = [{"role": "user", "content": "What is 17.5 plus 24.25?"}]
first = client.chat.completions.create(
    model="uncensored", messages=messages, tools=tools, tool_choice="auto"
)

msg = first.choices[0].message
if msg.tool_calls:
    messages.append(msg)
    for call in msg.tool_calls:
        args = json.loads(call.function.arguments)
        result = args["a"] + args["b"]
        messages.append({
            "role": "tool",
            "tool_call_id": call.id,
            "content": str(result),
        })
    final = client.chat.completions.create(
        model="uncensored", messages=messages, tools=tools
    )
    print(final.choices[0].message.content)
else:
    print(msg.content)

Всегда парсите arguments с осторожностью. Это JSON-строка, сгенерированная моделью, и возможна некорректная строка. Проверьте ее перед использованием, особенно если функция затрагивает файлы или базу данных.

Ошибки, лимиты и повторные попытки

Запрос, превышающий лимит, возвращает HTTP 429. Код ошибки указывает, какой именно лимит: rate_limited для запросов в минуту (300 для полного ключа, 60 для пробного) и concurrency для одновременных запросов. При временной перегрузке вы также получаете 429 с Retry-After: 5. SDK вызывает исключение openai.RateLimitError.

Copy# client = OpenAI(base_url="https://api.deepseekapi.top/v1", ...) as above
import time
import openai

def ask(prompt, tries=4):
    for attempt in range(tries):
        try:
            r = client.chat.completions.create(
                model="uncensored",
                messages=[{"role": "user", "content": prompt}],
            )
            return r.choices[0].message.content
        except openai.RateLimitError as e:
            wait = int(e.response.headers.get("Retry-After", 5))
            time.sleep(wait * (attempt + 1))
    raise RuntimeError("still rate limited")

print(ask("Name three Python web frameworks."))

Если вы отправляете много запросов, ограничьте собственную параллельность семафором или пулом потоков из 8 рабочих, или 2 для пробного ключа. Это избежит ошибок concurrency вместо их обработки через повторные попытки.

Чтение заголовков

Каждый ответ содержит X-Request-Id, X-Balance-USD, X-RateLimit-Limit-Requests и X-RateLimit-Limit-Concurrency. SDK предоставляет доступ к ним через сырой ответ:

Copy# client = OpenAI(base_url="https://api.deepseekapi.top/v1", ...) as above
raw = client.chat.completions.with_raw_response.create(
    model="uncensored",
    messages=[{"role": "user", "content": "ping"}],
)
print(raw.headers.get("X-Balance-USD"), raw.headers.get("X-Request-Id"))
completion = raw.parse()

Логирование идентификатора запроса при каждой ошибке значительно упрощает поддержку. Неудачные или отклонённые запросы не оплачиваются.

Продолжение разговора

API не сохраняет состояние. Он не запоминает предыдущие запросы, поэтому для многооборотного диалога нужно каждый раз отправлять всю историю. Ведите список и добавляйте в него новые сообщения:

Copy# client = OpenAI(base_url="https://api.deepseekapi.top/v1", ...) as above
history = [{"role": "system", "content": "You are a helpful tutor."}]

def chat(user_text):
    history.append({"role": "user", "content": user_text})
    r = client.chat.completions.create(model="uncensored", messages=history)
    reply = r.choices[0].message.content
    history.append({"role": "assistant", "content": reply})
    return reply

print(chat("What is a closure?"))
print(chat("Show me one in Python."))

Каждый ход отправляет все заново, поэтому входные токены растут с каждым ходом. Удаляйте старые сообщения, когда список становится большим, или просите модель резюмировать раннюю часть и заменять ее. С контекстным окном на 100 000 токенов у вас много места, но счет растет вместе с историей, поэтому обрезанный список дешевле.

Выбор значений сэмплирования

Используйте temperature от 0 до 0,3 для кода и извлечения, около 0,7 для общего чата и выше для художественной литературы. top_p, stop, seed и штрафы за частоту и присутствие передаются, поэтому они ведут себя так, как описано в документации OpenAI. Изменяйте по одному и сравнивайте.

Планирование стоимости и контекста

Контекстное окно составляет 100 000 токенов, включая промпт и завершение, а лимит тела запроса составляет 8 МБ. Завершение может использовать все оставшееся место окна, и max_tokens является необязательным. Установка этого значения все же рекомендуется для пакетных заданий.

Пример расчета: пакет из 1 000 кратких сводок, каждое с 1 500 входными токенами и 200 выходными токенами.

• Вход: 1 000 × 1 500 = 1 500 000 токенов, что составляет 1,5 × $0,15 = $0,225
• Выход: 1 000 × 200 = 200 000 токенов, что составляет 0,2 × $0,50 = $0,10
• Итого: $0,325

Ценообразование — оплата по факту из предоплаченного баланса, без подписки, и кредит не сгорает. Пополнения от $20 до $500 целыми суммами, с +5% от $50, +10% от $100 и +15% от $200, оплачиваются картой или криптовалютой. См. страницу ценообразования и документацию быстрого старта для примеров curl и Node. Если вам нужно только найти свой ключ, раздел обзор API-ключа покрывает это.

Промпты не используются для обучения. Сервис работает только с текстом: нет изображений, аудио, эмбеддингов или дообучения, поэтому не планируйте работу с этими эндпоинтами.

Дополнительные руководства

• Как использовать API DeepSeek в Cursor: настройка базового URL
• Как использовать API DeepSeek в SillyTavern: пользовательский эндпоинт
• Ключ API DeepSeek: доступ к API LLM без цензуры