ตัวอย่าง Python API DeepSeek: แชท สตรีมมิง และเครื่องมือ

ติดตั้งและส่งคำขอแรก

บริการนี้ไม่ใช่ DeepSeek API ทางการ เป็นเอนด์พอยต์ที่รองรับรูปแบบ OpenAI ซึ่งรันโมเดล open-weight model เดียว ID คือ uncensored โดยผู้ให้บริการอิสระ เนื่องจากใช้รูปแบบ OpenAI SDK มาตรฐานของ Python จึงใช้งานได้ทันทีโดยไม่ต้องแก้ไข ยกเว้นการเปลี่ยน URL และคีย์

Copypip install openai

รับคีย์จากหน้า API key หลังจากสมัครสมาชิก คุณสามารถรับเครดิตทดลองใช้ $0.50 ได้ด้วยการคลิกเพียงครั้งเดียว ไม่ต้องใส่บัตรเครดิต ใช้ได้ 7 วัน เก็บคีย์ไว้ในตัวแปรสภาพแวดล้อมแทนที่จะเก็บในไฟล์ซอร์สโค้ด:

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)

ตัวเลขการใช้งานสองค่าคือสิ่งที่เราคิดเงินจากคุณ คูณตัวเลขเหล่านี้กับราคาเพื่อดูค่าใช้จ่ายของการเรียกใช้ใดๆ: input tokens คูณด้วย $0.15 และ output tokens คูณด้วย $0.50 โดยแต่ละค่าหารด้วย 1,000,000

การใช้ async client

สำหรับเว็บเซิร์ฟเวอร์และแอปพลิเคชันที่จัดการผู้ใช้จำนวนมาก ให้ใช้ไคลเอนต์แบบ async ไคลเอนต์นี้มีพารามิเตอร์เดียวกันกับไคลเอนต์แบบ sync และการเปลี่ยนแปลงเป็นแบบ drop-in ในโค้ดส่วนใหญ่

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) ตั้งค่าไว้ที่ 8 เพราะนั่นคือขีดจำกัดการทำงานพร้อมกัน (concurrency) ของคีย์เต็ม สำหรับคีย์ทดลองใช้ให้ใช้ค่า 2 หากไม่มีขีดจำกัด การทำงานพร้อมกันจำนวนมากจะเกิดข้อผิดพลาด concurrency แม้แต่ละคำขอจะถูกต้องก็ตาม

การตอบกลับแบบสตรีมมิง

ตั้งค่า stream=True แล้วเอนด์พอยต์จะส่ง server-sent events แต่ละ chunk จะมีข้อความชิ้นเล็กๆ ใน delta.content การขอข้อมูลการใช้งานหมายถึง chunk สุดท้ายจะรายงานจำนวนโทเคน

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)

สองรายละเอียดที่ทำให้คนสับสน chunk การใช้ข้อมูลสุดท้ายอาจมีรายการ choices ว่างเปล่า นั่นคือเหตุผลที่ลูปต้องตรวจสอบ chunk.choices ก่อน และคำขอแบบสตรีมมิงจะคงตำแหน่ง concurrency ไว้จนกว่าสตรีมจะสิ้นสุด ดังนั้นจึงควรออกจากลูปอย่างสะอาดและปิดสตรีมที่ถูกยกเลิก คีย์เต็มอนุญาตให้ทำคำขอพร้อมกัน (concurrent requests) ได้ 8 คำขอ ส่วนคีย์ทดลองใช้ได้ 2 คำขอ

โหมด JSON

สำหรับเอาต์พุตที่โปรแกรมจะแยกวิเคราะห์ ให้ใช้ response_format พร้อม json_object นอกจากนี้ให้ใส่คำว่า JSON ในพรอมต์และอธิบายโครงสร้างที่ต้องการ โมเดลต้องการคำแนะนำนั้น; การตั้งค่าเพียงอย่างเดียวไม่ได้ให้ schema

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 หากข้อความตอบกลับถูกตัดออกเนื่องจากข้อจำกัดของโทเคน มันจะไม่ทำการ parse ดังนั้นหลีกเลี่ยงการตั้งค่า max_tokens ที่ต่ำเกินไปเมื่อใช้ JSON การตั้งค่า temperature เป็น 0 และ seed ทำให้การรันซ้ำมีความสม่ำเสมอมากขึ้น เนื่องจากทั้งสองค่าจะถูกส่งผ่านไปด้วย

การเรียกใช้ฟังก์ชัน

API รองรับ tools และ tool_choice และจะส่งคืน tool_calls ในข้อความของ assistant คุณเรียกใช้ฟังก์ชันเองและส่งผลลัพธ์กลับเป็นข้อความที่มี 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 รหัสข้อผิดพลาดจะบอกคุณว่าขีดจำกัดใด: rate_limited สำหรับคำขอต่อนาที (300 บนคีย์เต็ม, 60 บนคีย์ทดลอง) และ concurrency สำหรับคำขอพร้อมกันมากเกินไป เมื่อเกิดโอเวอร์โหลดชั่วคราวคุณยังได้รับ 429 พร้อม Retry-After: 5 SDK จะ raise 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."))

หากคุณส่งคำขอจำนวนมาก ให้จำกัดการทำงานแบบขนานของคุณด้วย semaphore หรือ thread pool ที่มี worker 8 ตัว หรือ 2 ตัวบนคีย์ทดลอง สิ่งนี้จะหลีกเลี่ยงข้อผิดพลาด concurrency แทนที่จะทำ retry ผ่านข้อผิดพลาดเหล่านั้น

การอ่าน headers

แต่ละการตอบกลับมี X-Request-Id, X-Balance-USD, X-RateLimit-Limit-Requests และ X-RateLimit-Limit-Concurrency SDK เปิดเผยค่าเหล่านี้ผ่าน response ดิบ:

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

การบันทึก request ID เมื่อเกิดข้อผิดพลาดทุกครั้งทำให้การตอบคำถามจากฝ่ายสนับสนุนง่ายขึ้นมาก คำขอที่ล้มเหลวหรือถูกปฏิเสธจะไม่ถูกเรียกเก็บเงิน

การสนทนาต่อเนื่อง

API เป็น stateless มันไม่จำคำขอในอดีต ดังนั้นการแชทหลายรอบหมายถึงคุณต้องส่งประวัติทั้งหมดทุกครั้ง เก็บรายการและเพิ่มข้อความใหม่เข้าไป:

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

แต่ละรอบจะส่งข้อมูลทั้งหมดอีกครั้ง ดังนั้น input tokens จะเพิ่มขึ้นตามรอบ ควรตัดข้อความเก่าออกเมื่อรายการยาวขึ้น หรือขอให้โมเดลสรุปส่วนต้นและแทนที่ด้วยข้อความนั้น ด้วยหน้าต่างบริบท 100,000 โทเคน คุณมีพื้นที่เพียงพอ แต่บิลจะเพิ่มขึ้นตามประวัติ ดังนั้นรายการที่ตัดแล้วจึงมีค่าใช้จ่ายน้อยกว่า

การเลือกค่า sampling

ใช้ temperature 0 ถึง 0.3 สำหรับโค้ดและการดึงข้อมูล ใช้ประมาณ 0.7 สำหรับแชททั่วไป และค่าที่สูงกว่าสำหรับงานวรรณกรรม top_p, stop, seed และค่าปรับความถี่และการปรากฏจะถูกส่งผ่าน ดังนั้นจึงทำงานตามที่เอกสารของ OpenAI ระบุ เปลี่ยนค่าทีละตัวแล้วเปรียบเทียบ

การวางแผนต้นทุนและบริบท

หน้าต่างบริบทมีขนาด 100,000 โทเคน (รวมพรอมต์และการสร้างผลลัพธ์) และขีดจำกัดขนาด request bodyคือ 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 ในจำนวนเต็ม พร้อมโบนัส +5% เมื่อเติม $50, +10% เมื่อเติม $100 และ +15% เมื่อเติม $200 สามารถชำระเงินด้วยบัตรหรือคริปโต ดูรายละเอียดใน หน้าราคา และเอกสาร เริ่มต้นใช้งาน สำหรับตัวอย่าง curl และ Node หากต้องการค้นหาคีย์ API ของคุณ ดูที่ ภาพรวมคีย์ API

พรอมต์ไม่ถูกนำไปใช้ในการฝึกโมเดล บริการนี้รองรับเฉพาะข้อความ: ไม่มีรูปภาพ, เสียง, embedding หรือ fine-tuning ดังนั้นอย่าวางแผนใช้งานตามเอนด์พอยต์เหล่านั้น

คู่มือเพิ่มเติม

• วิธีใช้ DeepSeek API ใน Cursor: การตั้งค่า Base URL แบบกำหนดเอง
• วิธีใช้ API ของ DeepSeek ใน SillyTavern: เอนด์พอยต์แบบกำหนดเอง
• คีย์ API ของ DeepSeek: เข้าถึง API LLM แบบไม่เซ็นเซอร์