DeepSeek API Python例：チャット、ストリーミング、ツールグ、ツール

インストールと最初のリクエストの送信

このサービスは公式のDeepSeek APIではありません。独立したプロバイダが運営するホスト型エンドポイントで、ID uncensoredのオープンウェイトモデルを実行しています。OpenAI形式に従っているため、URLとキーの変更以外の追加変更なしで標準Python SDKが動作します。

Copypip install openai

APIキーページからキーを取得してください。サインアップ後、カード不要で1クリックするだけで、7日間有効な$0.50の無料トライアルクレジットを請求できます。キーをソースファイルではなく環境変数に保存してください。

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)

2つの使用量数字は請求の対象となります。価格を乗算して、あらゆる呼び出しのコストを確認してください：入力トークン数×$0.15および出力トークン数×$0.50をそれぞれ1,000,000で割ります。

非同期クライアントの使用

Webサーバーや多数のユーザーを処理するものには、非同期クライアントを使用してください。同期クライアントと同じパラメータを持ち、ほとんどのコードでそのまま置き換え可能です。

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)

つまずきやすい点が2つあります。最後の使用量チャンクは空の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で囲んでください。トークン制限で応答が途中で切れると解析できないため、JSONでは小さすぎるmax_tokensを避けてください。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を返します。エラーコードはどの制限かを示します：毎分のリクエスト数の制限（フルキーで300、試用キーで60）はrate_limited、同時リクエストが多すぎる場合はconcurrencyです。一時的な過負荷の場合も、Retry-After: 5付きで429を受け取ります。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()

すべての失敗にリクエストIDをログに記録すると、サポートの質問がはるかに答えやすくなります。失敗または拒否されたリクエストには請求されません。

会話を続ける

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のドキュメントで説明されているように動作します。一度に1つ変更して比較してください。

コストとコンテキストの計画

コンテキストウィンドウは100,000トークン（プロンプトと補完を含む）、リクエストボディの上限は8 MBです。補完はウィンドウの残りを何でも使用でき、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の整数額で、$50以上で+5%、$100以上で+10%、$200以上で+15%のボーナスがあり、カードまたは暗号資産で支払えます。詳細は料金ページ、curlとNodeの例はクイックスタートドキュメントをご覧ください。キーの場所のみが必要な場合は、APIキーの概要をご覧ください。

プロンプトはトレーニングには使用されません。サービスはテキストのみ：画像、オーディオ、埋め込み、ファインチューニングはないため、それらのエンドポイントを中心に計画しないでください。

その他のガイド

• CursorでDeepSeek APIを使用する方法：カスタムベースURLの設定
• SillyTavernでDeepSeek APIを使用する方法：カスタムエンドポイント
• DeepSeek APIキー：無検閲LLM APIアクセス