DeepSeek V4 API Python:最小化程式碼示例與串流處理

DeepSeek V4 API Python:最小化程式碼示例與串流處理

大家好!我是 Dora。事情開始於一個小小的煩惱:我一直在項目之間複製相同的聊天完成樣板,像換罐子上的標籤一樣交換基礎 URL 和模型名稱。這不是困難的工作,只是那種會給你的一天增加磨擦的工作。我已經看到 DeepSeek 出現得足夠頻繁,引起了我的好奇心,所以我在 2026 年 1 月下旬留出幾個上午,將他們的「V4」API 接入我的 Python 堆棧,看看在實際使用中感覺如何。

我不是在追逐基準。我想知道:客戶端是否不妨礙我,我能否可靠地流式傳輸,以及錯誤是否以易於推理的方式失敗?以下是我嘗試的內容、困擾我的內容和悄悄發揮作用的內容。開始吧!

環境設置

依賴項

我在 macOS 上使用 Python 3.11 保持設置簡單。你可以用標準庫做到這一點,但三個小包使生活更輕鬆:

  • requests(直接的 HTTP:適用於大多數情況)
  • httpx(非同步和表現良好的超時)
  • python-dotenv(所以我不在四周粘貼金鑰) 如果你計劃使用伺服器發送事件進行流式傳輸,你可以使用 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)

現場筆記

  • 第一次運行很平靜(令人欣慰)。該結構與我的預期相符,這使得遷移一個小的提示庫變得很快。
  • 我為了可重複的答案保持溫度低。這聽起來很明顯,但當我進行故障排除時,我仍然忘記。
  • 如果你需要確定性運行,如果 API 支持,也可以固定 top_p 和 seed。當文檔沉默時,我假設非確定性。

如果你正在比較提供商,這裡的優勢是低摩擦力。缺點是差異隱藏在邊緣、錯誤有效載荷、令牌會計和流式傳輸形狀中。這些邊緣是你的集成要麼感覺堅固,要麼令人厭煩的地方。

代碼生成示例

我不要求模型編寫完整的模組。它變成了一項清理工作。但對於小幫手,如「解析此時間戳格式」或「起草帶有佔位符的 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]。我不需要特殊客戶端,requestsiter_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] 標記可靠地出現。這聽起來微不足道,直到它沒有,遺漏的終止符使 UI 掛起。

如果你需要流式傳輸到 Web 應用程序,我會在其間放置一個薄伺服器層來規範化事件。這是額外的一步,但它可以簡化你的前端。

伺服器發送事件

在引擎蓋下,你實際上是在閱讀伺服器發送事件。如果你更喜歡幫手,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 正文包括我可以表面到日誌的消息。我仍然將其包裝在我自己的異常中,以便我控制到達 UI 的內容。

如果你的程式碼庫已經使用 OpenAI 風格的架構,這是可管理的:相同的消息形狀,邊緣略有不同。主要的事情是對模型名稱嚴格,並在非 2xx 時記錄完整的回應正文,以便你不會猜測。
文檔方面,我依靠官方 DeepSeek API 文檔來查看參數名稱和流式傳輸形狀。每當提供商使用熟悉的端點時,假設奇偶校驗很誘人。我已經學到了先檢查文檔,並在客戶端之間複製少於我認為可以的內容。

誰可能喜歡這個

  • 如果你有針對聊天完成的現有 Python 包裝程序,遷移路徑是溫和的。
  • 如果你關心流式傳輸和簡單重試,它的表現可預測。
  • 如果你需要非常具體的工具(函數調用架構、推理令牌或批量作業),你會想要仔細閱讀文檔,並在提交前使用一項狹隘的任務進行原型設計。

我沒有嘗試在這裡協調長時間、多步驟的代理。我專注於小的、日常使用的提示,那種逐漸消除摩擦的。這就是 DeepSeek V4 API 與 Python 感到足夠穩定以保持的地方。