مثال Python لواجهة DeepSeek API: الدردشة، البث المتدفق، الأدوات

ثبّت وأرسل الطلب الأول

هذه الخدمة ليست واجهة برمجة التطبيقات الرسمية من DeepSeek. إنها نقطة نهاية متوافقة مع OpenAI مستضافة وتشغّل نموذجًا واحدًا مفتوح الأوزان، معرفه uncensored، تديره جهة مستقلة. نظرًا لأنه يتبع تنسيق OpenAI، فإن مكتبة SDK الخاصة بـ Python تعمل دون تغييرات سوى تعديل URL ومفتاح API.

Copypip install openai

احصل على مفتاح من صفحة مفتاح API. بعد التسجيل، يمكنك الحصول على رصيد تجريبي بقيمة $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)

رقما الاستخدام هما الأساس الذي يُحسب عليك بناءً عليه. اضربهما في الأسعار لمعرفة تكلفة أي طلب: رموز الإدخال مضروبة في $0.15 ورموز الإخراج مضروبة في $0.50، وكل منهما مقسوم على 1,000,000.

استخدام العميل غير المتزامن

لخوادم الويب وأي شيء يتعامل مع العديد من المستخدمين، استخدم العميل غير المتزامن. يحتوي على نفس المعاملات مثل العميل المتزامن وهو تغيير مباشر في معظم الأكواد.

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 وترسل نقطة النهاية أحداثًا مرسلة من الخادم. يحمل كل جزء قطعة صغيرة من النص في 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)

تفاصيلان يسببان مشاكل. يمكن أن يحتوي الجزء النهائي للاستخدام على قائمة choices فارغة، ولهذا تتحقق الحلقة من chunk.choices أولاً. ويحتفظ الطلب المتدفق بخانة التزامن (slot) الخاصة به حتى ينتهي البث، لذا اخرج من الحلقات بنظافة وأغلق البث الذي تتخلى عنه. يسمح المفتاح الكامل بـ 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"])

ضع try/except حول json.loads. إذا تم قطع الرد بواسطة حد الرموز فلن يتم تحليله، لذا تجنب استخدام max_tokens صغير مع JSON. تعيين temperature إلى 0 وseed يجعل إعادة التشغيل أكثر قابلية للتكرار، حيث يتم تمرير كليهما.

استدعاء الدوال

تقبل واجهة برمجة التطبيقات 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. يوضح رمز الخطأ الحد الذي تم تجاوزه: rate_limited للطلبات في الدقيقة الواحدة (300 على مفتاح كامل، 60 على مفتاح تجريبي) وconcurrency للكثير في وقت واحد. في حالة الحمل الزائد المؤقت، تحصل أيضًا على 429 مع Retry-After: 5. تثير مكتبة 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."))

إذا أرسلت العديد من الطلبات، حدّ من التزامن لديك باستخدام سيمافور (semaphore) أو مجموعة خيوط بـ 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()

تسجيل معرف الطلب مع كل فشل يجعل الإجابة على أسئلة الدعم أسهل بكثير. الطلبات الفاشلة أو المرفوضة لا تُحسب.

إدامة المحادثة

واجهة برمجة التطبيقات عديمة الحالة. لا تتذكر الطلبات السابقة، لذا فإن دردشة متعددة المراحل تعني أنك ترسل كل السجل في كل مرة. احتفظ بقائمة وأضف إليها:

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. غيّر واحدة في كل مرة وقارن.

تخطيط التكلفة والنافذة السياقية

تبلغ نافذة السياق 100,000 رمز (token)، وتشمل الموجّه والإكمال، بينما حدّ حجم جسم الطلب هو 8 MB. يمكن أن يستخدم الإكمال المساحة المتبقية من النافذة، وmax_tokens اختياري. يُعدّ ضبطه ممارسة جيدة لعمليات الدفعات.

مثال عملي: دفعة من 1,000 ملخص، يحتوي كل منها على 1,500 رمز (token) للإدخال و200 رمز (token) للإخراج.

• الإدخال: 1,000 × 1,500 = 1,500,000 رمز (token)، وهو ما يعادل 1.5 × $0.15 = $0.225
• الإخراج: 1,000 × 200 = 200,000 رمز (token)، وهو ما يعادل 0.2 × $0.50 = $0.10
• المجموع: $0.325

التسعير يعتمد على الدفع حسب الاستخدام من رصيد مسبق الدفع، بدون اشتراك، ولا ينتهي الرصيد. تتراوح عمليات الشحن من $20 إلى $500 بمبالغ صحيحة، مع إضافة +5% من $50، و+10% من $100، و+15% من $200، ويمكن الدفع بالبطاقة أو العملات المشفرة. انظر صفحة التسعير، ودليل البدء السريع للحصول على أمثلة curl وNode.js. إذا كنت بحاجة فقط إلى العثور على مفتاحك، يغطي نظرة عامة على مفتاح API ذلك.

لا تُستخدم الموجّهات للتدريب. الخدمة مخصصة للنصوص فقط: لا صور، ولا صوت، ولا تضمينات (embeddings)، ولا ضبط دقيق (fine-tuning)، لذا لا تخطط حول نقاط النهاية الخاصة بها.

المزيد من الأدلة

• كيفية استخدام DeepSeek API في Cursor: إعداد Base URL مخصص
• كيفية استخدام DeepSeek API في SillyTavern: نقطة نهاية مخصصة
• مفتاح DeepSeek API: الوصول إلى LLM بدون رقابة