Przykład API DeepSeek w Pythonie: czat, strumieniowanie, narzędzia, narzędzia

Zainstaluj i wyślij pierwsze zapytanie

Usługa ta nie jest oficjalnym API DeepSeek. To hostowany endpoint zgodny z OpenAI, który uruchamia jeden model o otwartych wagach o ID uncensored, obsługiwany przez niezależnego dostawcę. Ponieważ korzysta z formatu OpenAI, standardowe Python SDK działa bez zmian poza adresem URL i kluczem.

Copypip install openai

Pobierz klucz ze strony klucza API. Po rejestracji możesz jednym kliknięciem uzyskać darmowy kredyt próbny w wysokości $0,50 bez karty, ważny przez 7 dni. Przechowuj klucz w zmiennej środowiskowej, a nie w pliku źródłowym:

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)

Dwie liczby użycia są podstawą do rozliczeń. Pomnóż je przez ceny, aby zobaczyć koszt dowolnego zapytania: tokeny wejściowe razy $0,15 i tokeny wyjściowe razy $0,50, każde dzielone przez 1 000 000.

Korzystanie z klienta asynchronicznego

Dla serwerów sieciowych i wszystkiego, co obsługuje wielu użytkowników, użyj klienta asynchronicznego. Ma te same parametry co synchroniczny i jest łatwą zmianą w większości kodu.

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())

Semafor jest ustawiony na 8, ponieważ to limit współbieżności pełnego klucza. Przy kluczu próbnym użyj 2. Bez ograniczeń nagły wzrost zadań spowoduje błędy concurrency, mimo że każde zapytanie jest poprawne.

Odpowiedzi strumieniowe

Ustaw stream=True, a endpoint wysyła SSE. Każdy fragment zawiera małą część tekstu w delta.content. Żądanie użycia oznacza, że ostatni fragment raportuje liczbę tokenów.

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)

Dwa szczegóły sprawiają problemy. Końcowy fragment użycia może mieć pustą listę choices, dlatego pętla sprawdza chunk.choices najpierw. A strumieniowane zapytanie trzyma swój slot współbieżności do końca strumienia, więc wychodź z pętli czysto i zamykaj strumienie, które porzucasz. Pełny klucz pozwala na 8 równoległych zapytań; klucz próbny pozwala na 2.

Tryb JSON

Dla danych wyjściowych do parsowania przez program użyj response_format z json_object. Dodaj również słowo JSON w prompcie i opisz żądaną strukturę. Model potrzebuje tej instrukcji; sama flaga nie dostarcza schematu.

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"])

Owiń json.loads blokiem try/except. Jeśli odpowiedź zostanie ucięta przez limit tokenów, nie zostanie sparsowana, więc unikaj małego max_tokens z JSON. Ustawienie temperature na 0 i seed sprawia, że ponowne uruchomienia są bardziej powtarzalne, ponieważ są one przekazywane dalej.

Wywoływanie funkcji

API akceptuje tools i tool_choice, a zwraca tool_calls w wiadomości asystenta. Sam uruchamiasz funkcję i wysyłasz wynik z powrotem jako wiadomość z 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)

Zawsze parsuj arguments ostrożnie. Jest to ciąg JSON wygenerowany przez model, a jego niepoprawny format jest możliwy. Zweryfikuj go przed wykonaniem akcji, zwłaszcza jeśli funkcja odnosi się do plików lub bazy danych.

Błędy, limity i ponawianie

Zapytanie powyżej limitu zwraca HTTP 429. Kod błędu informuje, który limit: rate_limited dla zapytań na minutę (300 na pełnym kluczu, 60 na kluczu próbnym) i concurrency dla zbyt wielu naraz. Przy tymczasowym przeciążeniu otrzymujesz również 429 z Retry-After: 5. SDK zgłasza wyjątek 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."))

Jeśli wysyłasz wiele zapytań, ogranicz własną równoległość semaforem lub pulą 8 wątków (lub 2 przy kluczu próbnym). Unikniesz błędów concurrency zamiast ponawiać próbę przez nie.

Odczytywanie nagłówków

Każda odpowiedź zawiera X-Request-Id, X-Balance-USD, X-RateLimit-Limit-Requests i X-RateLimit-Limit-Concurrency. SDK udostępnia je poprzez surową odpowiedź:

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()

Logowanie identyfikatora zapytania przy każdej awarii znacznie ułatwia odpowiadanie na pytania wsparcia. Nieudane lub odrzucone zapytania nie są rozliczane.

Kontynuowanie konwersacji

API jest bezstanowe. Nie pamięta wcześniejszych zapytań, więc czat wieloetapowy oznacza wysyłanie całej historii za każdym razem. Zachowaj listę i dodawaj do niej:

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."))

Każda tura wysyła ponownie wszystko, więc liczba tokenów wejściowych rośnie z turą. Skróć stare wiadomości, gdy lista stanie się długa, lub poproś model o podsumowanie początkowej części i zastąp ją. Przy oknie kontekstu 100 000 tokenów masz dużo miejsca, ale rachunek rośnie wraz z historią, więc skrócona lista jest tańsza.

Wybieranie wartości próbkowania

Użyj temperature 0 do 0,3 dla kodu i ekstrakcji, około 0,7 dla ogólnego czatu i wyższych dla fikcji. top_p, stop, seed i kary częstotliwości i obecności są przekazywane, więc zachowują się zgodnie z dokumentacją OpenAI. Zmień jeden na raz i porównaj.

Planowanie kosztów i kontekstu

Okno kontekstu ma 100 000 tokenów, prompt plus uzupełnienie, a limit rozmiaru treści żądania to 8 MB. Uzupełnienie może użyć tego, co zostało z okna, a max_tokens jest opcjonalne. Ustawienie go jest nadal dobrą praktyką dla zadań wsadowych.

Przykład z życia wzięty: partia 1 000 podsumowań, każde z 1 500 tokenami wejściowymi i 200 tokenami wyjściowymi.

• Wejście: 1 000 × 1 500 = 1 500 000 tokenów, co jest 1,5 × $0,15 = $0,225
• Wyjście: 1 000 × 200 = 200 000 tokenów, co jest 0,2 × $0,50 = $0,10
• Razem: $0,325

Cennik oparty jest na modelu pay-as-you-go z przedpłaconego kredytu, bez subskrypcji, kredyt nie wygasa. Doładowania od $20 do $500 w całości, z bonusem +5% od $50, +10% od $100 i +15% od $200, płatne kartą lub kryptowalutami. Zobacz stronę z cenami oraz dokumentację szybkiego startu z przykładami curl i Node. Jeśli potrzebujesz tylko znaleźć klucz, przegląd kluczy API to wyjaśnia.

Prompty nie są używane do trenowania. Usługa jest tylko tekstowa: bez obrazów, audio, embeddingów ani dostrajania, więc nie planuj wokół tych endpointów.

Więcej przewodników

• Jak korzystać z API DeepSeek w Cursor: Konfiguracja niestandardowego Base URL
• ⟦a0⟾Jak używać API DeepSeek w SillyTavern: niestandardowy endpoint/a0⟾
• ⟦a0⟾Klucz API DeepSeek: dostęp do API LLM bez cenzury/a0⟾