تتبع تكلفة رموز وكيل AI في Python: usage والميزانية والتوقف

يعمل وكيلك لمدة عشر دقائق وتكون الفاتورة أكبر من المهمة. لمعرفة السبب، اقرأ كائن usage في كل استجابة API، واجمعه لكل مهمة واضربه في سعر النموذج لكل مليون رمز. لتحديد سقف، قارن هذا الإجمالي الجاري بميزانية بعد كل خطوة وأوقف الحلقة عند تجاوزها. فيما يلي فئة Python صغيرة تفعل هذا بالضبط.

يستخدم المثال نقطة النهاية المتوافقة مع OpenAI https://api.cheaptokens.dev/v1 ومعرّف النموذج deepseek-ai/DeepSeek-V4-Flash-0731 (سياق 380K، حتى 32,768 رمز إخراج لكل إجابة). إذا أردت تقدير تشغيل قبل بدئه، فهذه مهمة مختلفة، ويغطيها كيفية تقدير تكلفة وكيل AI. هذه المقالة عن القياس والتحديد أثناء سير التشغيل.

أين توجد الأرقام: كائن usage#

كل إكمال دردشة غير متدفق يعود بكائن usage. يسرد مرجع API الخاص بـ DeepSeek (تم التحقق في 26 أكتوبر 2026) هذه الحقول:

• prompt_tokens — كل ما أرسلته هذه المرة: موجه النظام، السجل، نتائج الأدوات.
• completion_tokens — ما كتبه النموذج ردًا.
• total_tokens — الاثنان معًا.

يوثق DeepSeek أيضًا إضافات اختيارية: prompt_cache_hit_tokens وprompt_cache_miss_tokens (يجمعان إلى prompt_tokens)، وcompletion_tokens_details.reasoning_tokens لنماذج التفكير. لا تعرض الصفحة تاريخ نشر، وهذه الإضافات امتدادات لتنسيق OpenAI، لذا يتعامل الكود أدناه معها كاختيارية ويعتمد فقط على الحقلين الأولين.

لماذا تسجل كل خطوة وليس الأخيرة فقط؟ لأن الوكيل يعيد إرسال سياقه المتنامي في كل مرة. قد ترسل الخطوة 1 2,000 رمز إدخال، وقد ترسل الخطوة 30 9,000، والفاتورة هي مجموعها جميعًا. السبب موضح في لماذا تكلف دردشة طويلة أكثر، وتقسيم الإدخال/الإخراج في رموز الإدخال مقابل الإخراج.

عداد يوقف الوكيل#

إليك كل شيء: عداد يجمع الرموز، ويطبع سطرًا واحدًا لكل خطوة ويرفع خطأ عند تجاوز الميزانية، بالإضافة إلى حلقة وكيل بسيطة حوله. ثبّت SDK باستخدام pip install openai وضع مفتاحك في CHEAPTOKENS_API_KEY.

import os
from openai import OpenAI

client = OpenAI(
    base_url='https://api.cheaptokens.dev/v1',
    api_key=os.environ['CHEAPTOKENS_API_KEY'],
)
MODEL = 'deepseek-ai/DeepSeek-V4-Flash-0731'

# example prices, USD per 1M tokens: use the current ones from the pricing table
PRICE_IN = 0.10
PRICE_OUT = 0.40


class BudgetExceeded(Exception):
    pass


class TaskMeter:
    def __init__(self, budget_usd):
        self.budget = budget_usd
        self.steps = 0
        self.tokens_in = 0
        self.tokens_out = 0
        self.reasoning = 0

    @property
    def cost(self):
        return (self.tokens_in * PRICE_IN + self.tokens_out * PRICE_OUT) / 1_000_000

    def add(self, usage):
        self.steps += 1
        self.tokens_in += usage.prompt_tokens
        self.tokens_out += usage.completion_tokens
        details = getattr(usage, 'completion_tokens_details', None)
        self.reasoning += getattr(details, 'reasoning_tokens', 0) or 0
        print(f'step {self.steps}: in={usage.prompt_tokens} '
              f'out={usage.completion_tokens} task=${self.cost:.4f}')
        if self.cost > self.budget:
            raise BudgetExceeded(
                f'budget ${self.budget:.2f} crossed at step {self.steps}: '
                f'${self.cost:.4f}')


def run_task(goal, budget_usd=0.50, max_steps=40):
    meter = TaskMeter(budget_usd)
    messages = [
        {'role': 'system', 'content': 'Work step by step. Reply DONE when finished.'},
        {'role': 'user', 'content': goal},
    ]
    for _ in range(max_steps):
        resp = client.chat.completions.create(
            model=MODEL, messages=messages, max_tokens=1024)
        meter.add(resp.usage)  # log first, then decide
        reply = resp.choices[0].message.content or ''
        messages.append({'role': 'assistant', 'content': reply})
        if 'DONE' in reply:
            break
        messages.append({'role': 'user', 'content': 'Continue.'})
    return meter


try:
    meter = run_task('Summarise the main risks of a database migration.')
    print(f'finished: {meter.steps} steps, ${meter.cost:.4f}')
except BudgetExceeded as err:
    print('stopped:', err)

ثلاثة خيارات هناك تستحق النسخ.

سجّل قبل أن تقرر. يسجل meter.add الخطوة أولاً ثم يتحقق من الميزانية فقط بعد ذلك، لذا يترك تشغيل متوقف أثرًا كاملاً. يحدث التحقق بعد الاستدعاء الذي تجاوز الحد، لذا يمكنك التجاوز بخطوة واحدة على الأكثر. يحد max_tokens=1024 من إخراج تلك الخطوة الأخيرة.

توقفان صارمان، وليس واحدًا. يلتقط max_steps وكيلاً يتكرر دون إنفاق كثير؛ وتلتقط ميزانية الدولار وكيلاً ينفق كثيرًا لكل خطوة. تريد كليهما.

الأسعار في مكان واحد. PRICE_IN وPRICE_OUT أرقام عنصر نائب. انسخ القيم الحالية لنموذجك من صفحة الأسعار؛ حاليًا DeepSeek-V4-Flash هو $0.015 لكل مليون رمز إدخال و$0.10 لكل مليون رمز إخراج.

ما يجمعه تشغيل من 40 خطوة#

لنفترض أن وكيلك يستغرق 40 خطوة، ولأن السياق ينمو، ترسل الخطوة المتوسطة 6,000 رمز إدخال وتحصل على 400 رمز ردًا. هذا 40 × 6,000 = 240,000 رمز إدخال و40 × 400 = 16,000 رمز إخراج للمهمة.

على DeepSeek-V4-Flash يكلف هذا $0.0052. يشغّل الجدول نفس الحمل على النماذج الأخرى ويضع سعر OpenRouter المرجعي بجانب سعرنا:

إجمالاً، مع احتساب الإدخال والإخراج معًا، يخرج DeepSeek-V4-Flash أرخص من OpenRouter لهذه المهمة. لاحظ أيضًا أين توجد الرموز: 240,000 من 256,000 هي إدخال. السياق الذي يسحبه وكيلك هو الفاتورة، ولهذا فإن رقم in= لكل خطوة في سجلك هو ما يجب مراقبته. ميزانية 0.50 دولار لكل مهمة في المثال أعلاه هي سقف لانفلات، وليست هدفًا.

تبديل النماذج يعني فقط تغيير MODEL. اختر المعرّف من وثائق CheapTokens: تختلف الأسماء بين المزودين، ولن يعمل معرّف قصير من تسمية سعرية في API. لحلقات الوكلاء الطويلة، لدى DeepSeek-V4-Flash سعر واحد في أي ساعة من اليوم، لذا تكلف المهمة نفسها عند الظهر وعند الثالثة صباحًا. راجع صفحة DeepSeek لحدودها.

عندما يفكر النموذج#

إذا كانت الإجابة أطول بكثير في الرموز من النص الذي تراه، فانظر إلى reasoning_tokens. تُحتسب رموز التفكير كإخراج، لذا قد تنتج خطوة كثيفة الاستدلال رموز إخراج أكثر بكثير مما يوحي به النص المرئي، وتُفوتر كإخراج. يجمعها العداد أعلاه في meter.reasoning عندما يعيد API الحقل ويتركها 0 عندما لا يفعل، لذا لا يتعطل أبدًا بسبب مفتاح مفقود. اطبعه بجانب out= إذا أردت رؤية مقدار الإخراج الذي كان تفكيرًا. يشرح جهد الاستدلال ورموز التفكير السبب.

البث: usage يأتي أخيرًا#

إذا كان وكيلك يبث، فلا يوجد usage في الأجزاء العادية. في تنسيق OpenAI تطلبه باستخدام stream_options={'include_usage': True}. يقول مرجع DeepSeek (تم التحقق في 26 أكتوبر 2026) إن الخيار يعمل فقط مع stream=True (وإلا يجيب API بالرمز 400)، وإن usage هو null في كل جزء إلا واحدًا، وإن الجزء الأخير قبل data: [DONE] يحمل الإجماليات للطلب كله. لذا في حلقة بث، استدعِ meter.add مرة واحدة، على الجزء الذي لا يكون فيه usage فارغًا. الحلقة غير المتدفقة أعلاه هي أبسط مكان للبدء؛ البث في Python يحتوي على الحلقة نفسها.

إذا بدت الأرقام خاطئة#

• usage هو None. أنت تبث بدون include_usage. أضفه، أو انتقل إلى استدعاء غير متدفق.
• الإجمالي أقل من انخفاض الرصيد. تحقق من أن كل استدعاء يمر عبر العداد: إعادات المحاولة، أو استدعاء تلخيص أو استدعاء اختيار أداة يتم خارج run_task يسهل تفويته. غلّف client.chat.completions.create في دالة واحدة ووجّه كل شيء عبرها.
• تكلفة الخطوة تستمر في الارتفاع. هذا طبيعي لحلقة تضيف كل شيء إلى messages. قلّم إخراج الأدوات القديم أو لخّص السجل؛ إخراج الأدوات خارج السياق يوضح الفكرة.
• BudgetExceeded يطلق عند الخطوة 2. ميزانيتك أصغر من خطوة واحدة. تحقق من أن PRICE_IN لكل مليون رمز، وليس لكل ألف.

لتخزين الأرقام، اكتب meter.steps، وmeter.tokens_in، وmeter.tokens_out وmeter.cost كسطر JSON واحد لكل مهمة. أسبوع من هذه السطور يخبرك بالرموز الحقيقية لكل خطوة لوكيلك، وهو الرقم الذي يجب وضعه في كل تقدير بعد ذلك. النماذج والأسعار والطلب الأول موجودة في الوثائق؛ وتُغطى استدعاءات الأدوات في استدعاء الدوال في Python.