Ví dụ Python API DeepSeek: Chat, Streaming, Công cụ

Cài đặt và gửi yêu cầu đầu tiên

Dịch vụ này không phải là API DeepSeek chính thức. Đây là một endpoint tương thích OpenAI được lưu trữ, chạy một mô hình trọng số mở, ID uncensored, do một nhà cung cấp độc lập vận hành. Vì nó tuân theo định dạng OpenAI, SDK Python tiêu chuẩn hoạt động mà không cần thay đổi ngoài URL và khóa.

Copypip install openai

Lấy khóa từ trang khóa API. Sau khi đăng ký, bạn có thể nhận $0.50 tín dụng dùng thử chỉ với một cú nhấp chuột, không cần thẻ, có hiệu lực trong 7 ngày. Lưu khóa vào biến môi trường thay vì trong tệp nguồn:

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)

Hai số liệu sử dụng là cơ sở để bạn tính phí. Nhân chúng với giá để xem chi phí của bất kỳ yêu cầu nào: token đầu vào nhân $0.15 và token đầu ra nhân $0.50, mỗi số chia cho 1,000,000.

Sử dụng client không đồng bộ

Đối với máy chủ web và bất kỳ thứ gì xử lý nhiều người dùng, hãy sử dụng client bất đồng bộ. Nó có cùng các tham số với client đồng bộ và là một thay thế trực tiếp trong hầu hết các đoạn mã.

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

Semaphore được đặt thành 8 vì đó là concurrency của một khóa đầy đủ. Với khóa dùng thử, bạn hãy dùng 2. Nếu không có giới hạn, một đợt bùng nổ các tác vụ sẽ tạo ra lỗi concurrency mặc dù mỗi yêu cầu đều hợp lệ.

Phản hồi streaming

Đặt stream=True và endpoint sẽ gửi các sự kiện do máy chủ gửi. Mỗi chunk mang một đoạn văn bản nhỏ trong delta.content. Việc yêu cầu sử dụng có nghĩa là chunk cuối cùng báo cáo số lượng token.

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)

Hai chi tiết thường gây khó khăn cho mọi người. Chunk sử dụng cuối cùng có thể có danh sách choices rỗng, đó là lý do tại sao vòng lặp kiểm tra chunk.choices trước tiên. Và một yêu cầu streaming giữ slot concurrency của nó cho đến khi stream kết thúc, vì vậy bạn hãy thoát khỏi các vòng lặp một cách sạch sẽ và đóng các stream bạn bỏ dở. Một khóa đầy đủ cho phép 8 yêu cầu đồng thời; một khóa dùng thử cho phép 2.

Chế độ JSON

Đối với đầu ra mà chương trình sẽ phân tích cú pháp, bạn hãy sử dụng response_format với json_object. Bạn cũng hãy đề cập từ JSON trong prompt và mô tả hình dạng bạn muốn. Mô hình cần hướng dẫn đó; riêng tham số này không cung cấp lược đồ.

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

Bạn hãy bọc json.loads trong try/except. Nếu một phản hồi bị cắt ngang bởi giới hạn token, nó sẽ không phân tích cú pháp được, vì vậy hãy tránh một max_tokens nhỏ với JSON. Việc đặt temperature thành 0 và một seed giúp các lần chạy lại có thể lặp lại hơn, vì cả hai đều được truyền qua.

Gọi hàm

API chấp nhận tools và tool_choice, và trả về tool_calls trong tin nhắn trợ lý. Bạn tự chạy hàm và gửi kết quả lại dưới dạng tin nhắn với 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)

Luôn phân tích arguments một cách thận trọng. Đó là một chuỗi JSON do mô hình tạo ra và có thể gặp chuỗi sai định dạng. Hãy xác thực nó trước khi hành động dựa trên nó, đặc biệt nếu hàm chạm vào tệp hoặc cơ sở dữ liệu.

Lỗi, giới hạn và thử lại

Một yêu cầu vượt quá giới hạn sẽ trả về HTTP 429. Mã lỗi cho bạn biết giới hạn nào: rate_limited cho yêu cầu mỗi phút (300 trên khóa đầy đủ, 60 trên khóa dùng thử) và concurrency cho quá nhiều cùng một lúc. Khi quá tải tạm thời, bạn cũng nhận được 429 với Retry-After: 5. SDK sẽ ném ra 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."))

Nếu bạn gửi nhiều yêu cầu, bạn hãy giới hạn độ song song của chính mình bằng một semaphore hoặc một nhóm luồng có 8 worker, hoặc 2 trên khóa dùng thử. Điều đó tránh lỗi concurrency thay vì thử lại qua chúng.

Đọc tiêu đề

Mỗi phản hồi đều mang theo X-Request-Id, X-Balance-USD, X-RateLimit-Limit-Requests và X-RateLimit-Limit-Concurrency. SDK hiển thị chúng thông qua phản hồi thô:

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

Việc ghi lại ID yêu cầu với mọi lỗi sẽ giúp các câu hỏi hỗ trợ dễ trả lời hơn nhiều. Các yêu cầu bị lỗi hoặc bị từ chối không bị tính phí.

Duy trì hội thoại

API là không trạng thái. Nó không nhớ các yêu cầu trước đó, vì vậy một cuộc trò chuyện nhiều lượt có nghĩa là bạn gửi toàn bộ lịch sử mỗi lần. Bạn hãy giữ một danh sách và thêm vào đó:

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

Mỗi lượt gửi lại mọi thứ, vì vậy số token đầu vào tăng dần theo từng lượt. Cắt ngắn các tin nhắn cũ khi danh sách trở nên dài, hoặc yêu cầu mô hình tóm tắt phần đầu và thay thế nó. Với cửa sổ ngữ cảnh 100,000 token, bạn có rất nhiều chỗ, nhưng hóa đơn tăng theo lịch sử, vì vậy một danh sách đã cắt ngắn sẽ rẻ hơn.

Chọn các giá trị lấy mẫu

Sử dụng temperature từ 0 đến 0.3 cho mã và trích xuất, khoảng 0.7 cho trò chuyện chung, và cao hơn cho viết truyện hư cấu. top_p, stop, seed và các hình phạt về tần suất và sự hiện diện được chuyển tiếp, vì vậy chúng hoạt động như tài liệu OpenAI mô tả. Thay đổi từng cái một và so sánh.

Lập kế hoạch chi phí và ngữ cảnh

Cửa sổ ngữ cảnh là 100,000 token, bao gồm prompt và phần hoàn thành, và giới hạn body request là 8 MB. Phần hoàn thành có thể sử dụng bất kỳ phần còn lại nào của cửa sổ, và max_tokens là tùy chọn. Việc đặt nó vẫn là một thực hành tốt cho các công việc theo lô.

Một ví dụ minh họa: một lô 1,000 bản tóm tắt, mỗi bản có 1,500 token đầu vào và 200 token đầu ra.

• Đầu vào: 1,000 × 1,500 = 1,500,000 token, tương đương 1.5 × $0.15 = $0.225
• Đầu ra: 1,000 × 200 = 200,000 token, tương đương 0.2 × $0.50 = $0.10
• Tổng cộng: $0.325

Giá cả theo hình thức trả tiền theo từng lần sử dụng từ tín dụng trả trước, không có gói đăng ký, và tín dụng không bao giờ hết hạn. Các khoản nạp tiền chạy từ $20 đến $500 với các số nguyên, với +5% từ $50, +10% từ $100 và +15% từ $200, có thể thanh toán bằng thẻ hoặc tiền mã hóa. Xem trang giá cả, và tài liệu bắt đầu nhanh cho các ví dụ về curl và Node. Nếu bạn chỉ cần tìm khóa của mình, tổng quan về khóa API sẽ nói về việc này.

Các prompt không được sử dụng để huấn luyện. Dịch vụ chỉ dành cho văn bản: không có hình ảnh, âm thanh, embedding hoặc fine-tuning, vì vậy bạn đừng lập kế hoạch xung quanh các endpoint đó.

Hướng dẫn khác

• Cách sử dụng DeepSeek API trong Cursor: Thiết lập Base URL tùy chỉnh
• Cách sử dụng API DeepSeek trong SillyTavern: Endpoint tùy chỉnh
• Khóa DeepSeek API: Truy cập LLM API không kiểm duyệt