DeepSeek V4 API Python: ストリーミング付きの最小限のコード例
こんにちは、皆さん!Doraです。小さなイライラから始まったんです:プロジェクト間で同じチャット補完のボイラープレートを何度もコピーして、ベースURLとモデル名を瓶のラベルのように入れ替えていました。難しい作業ではなく、ただ日々に砂を加えるようなものです。2026年1月末に何度かDeepSeekが目に入るようになったので、彼らの「V4」APIをPythonスタックに組み込んで、実際に使ってみるのはどんな感じなのかを知りたくなり、数朝を費やしました。
ベンチマークを追い求めていたわけではありません。知りたかったのは:クライアントは邪魔になりませんか、確実にストリーミングできますか、そしてエラーは推論しやすい方法で失敗しますか、ということです。試したこと、つまずいたこと、そして静かに機能したことをここに述べます。さあ、行きましょう!
環境セットアップ
依存関係
macOS上のPython 3.11でセットアップを簡潔に保ちました。標準ライブラリでもできますが、3つの小さなパッケージが生活を楽にしてくれました:
- requests(シンプルなHTTP:ほとんどの場合に十分)
- httpx(非同期とうまく機能するタイムアウト)
- python-dotenv(キーをあちこちに貼り付けないようにするため)
サーバー送信イベント(Server-Sent Events)でストリーミングする予定がある場合は、requestsを使って自分で行をパースできます(私がしたこと)。または、sseclient-pyのようなヘルパーを持ってこられます。requestsに固執して、動く部分を減らしました。
インストール
pip install requests httpx python-dotenv
プロジェクトごとに最小限の仮想環境も作成しました。つまらないアドバイスですが、3ヶ月後にこれに戻ってくるときに依存関係の漂流から救ってくれます。
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")
セットアップからの2つの小さな注記:
- ベース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", # ドキュメント/コンソールで正確なモデルを確認してください
"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)
フィールドノート
- 最初の実行は問題なく済みました(ほっとしました)。構造は期待通りでしたが、小さなプロンプトライブラリの移行が速くなりました。
- 反復可能な回答のために温度を低く保ちました。明白に聞こえますが、トラブルシューティング中はまだ忘れます。
- 決定論的な実行が必要な場合は、APIがサポートしていれば、top_pとseedも固定してください。ドキュメントが沈黙している場合は、非決定論的と想定します。
プロバイダーを比較している場合、ここの利点は摩擦が低いことです。欠点は、違いがエッジ、エラーペイロード、トークン会計、ストリーミング形状に隠れていることです。それらのエッジは、統合が堅牢に感じられるか、それとも厄介に感じられるかです。
コード生成の例
モデルに完全なモジュールを書かせません。それはクリーンアップの仕事になります。しかし、「このタイムスタンプ形式を解析する」や「プレースホルダーでSQLをドラフトする」のような小さなヘルパーの場合は、便利です。 狭いプロンプト、明確な契約、小さな出力制限を使用しました。これはモデルがさまよくのを防ぎ、diffの確認を簡単にしました。
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", # またはあなたのV4対応モデル
"messages": messages,
"temperature": 0,
"max_tokens": 250
},
timeout=30
)
resp.raise_for_status()
code = resp.json()["choices"][0]["message"]["content"]
print(code)
実際に役立ったもの
- コードだけを返すよう常に指示します。それをスキップすると、必要ないまとめの文が得られます。
- 温度0は厄介な編集を減らします。
- 論理を読み直しました。実行ではValueErrorを処理していましたが、空白用に追加のテストを追加しました。今は2分余分、後で数時間の驚きを救います。
最初のショットで時間を節約しませんでした。3つまたは4つの小さなヘルパーの後で、精神的な努力が減ったことに気付きました:より少ないタブスイッチ、より少ない「正確な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:
# これが起こるとき、小さなログを保ちます:通常はネットワークの問題です
pass
print()
2つの小さな振る舞いが好きでした:
- 早期トークンは素早く到着しました(きれいな接続で1~2秒)。科学的ではなく、CLIツールに配線したときにキビキビ感じるのに十分です。
[DONE]マーカーは確実に表示されました。完全になるまで重要に聞こえますが、終了マーカーの欠落はUIをハング状態にします。
Webアプリにストリーミングする必要がある場合は、イベントを正規化するためにその間に薄いサーバーレイヤーを配置します。1つの余分なステップですが、フロントエンドをシンプルに保ちます。
サーバー送信イベント
フードの下では、基本的にサーバー送信イベント(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~150ms)も追加して、雷群を避けました。
- タイムアウトハイジーン:迅速なチェック(5~10秒)の場合は短い接続/読み取りタイムアウトを設定し、大きなプロンプトの場合は長いものを設定しました。タイムアウトはすべてデフォルトで30秒であってはいけません:問題を隠します。
- エラーペイロード:何かが失敗したとき、JSONボディにはログに表示できるメッセージが含まれていました。ログに達するもの、UIに到達するものを制御できるように、独自の例外でラップします。
コードベースがOpenAIスタイルスキーマを話す場合、これは管理可能です:同じメッセージ形状、少し異なるエッジ。重要なのは、モデル名に厳密であり、非2xxで完全なレスポンスボディをログに記録して、推測しないことです。 ドキュメント的には、パラメータ名とストリーミング形状について公式のDeepSeek APIドキュメントに頼りました。プロバイダーが親しみ深いエンドポイントを使用する場合、パリティを想定するのは誘惑されます。ドキュメントを最初にチェックして、クライアント間で考えることができるより少なくコピーすることを学びました。
これが好きな人

- チャット補完用の既存のPythonラッパーがある場合は、移行パスは緩やかです。
- ストリーミングとシンプルな再試行に気をつけている場合は、予測可能に動作します。
- 非常に特定のツール(関数呼び出しスキーマ、推論トークン、またはバッチジョブ)が必要な場合は、ドキュメントを詳しく読んで、コミットする前に1つの狭いタスクでプロトタイプを作成する必要があります。
長い複数段階のエージェントをオーケストレートしようとしませんでした。摩擦を削減する小さな日常的なプロンプトに焦点を当てました。DeepSeek V4 APIとPythonは十分に安定していて、保持しておく価値があるという感じでした。





