ابنِ حلقة agent للذكاء الاصطناعي: أربعة حواجز حماية قبل النشر

الـ agent هو ما تحصل عليه حين تضع استخدام الأدوات داخل حلقة. نداء tool واحد يجيب عن سؤال واحد؛ أما الـ agent فينادي tool، ويقرأ النتيجة، ويقرّر ما يفعله تالياً، ويواصل حتى تُنجَز المهمة فعلاً — يبحث، ثم يقرأ أعلى نتيجة، ثم يستعلم عن سعر، ثم يكتب الإجابة. الآلية بسيطة، وهي الإيقاعات الأربعة ذاتها في كل مرة: النموذج يطلب tool، فتشغّله، فتعيد النتيجة، فتنادي النموذج مجدداً. الجزء الصعب ليس الحلقة. بل حواجز الحماية التي تمنعها من الدوران إلى الأبد أو تحميلك بصمت 40 دولاراً في تشغيل واحد.

هذه التدوينة تبني حلقة agent حقيقية وقابلة للتشغيل مقابل https://api.brievio.com/v1 باستخدام OpenAI Python SDK، ثم تغلّفها بالضوابط الأربعة التي تجعلها آمنة للنشر: سقف صارم للتكرارات، وتوزيع أدوات متحقَّق منه، وميزانية تكلفة لكل تشغيل تُقرأ من أعداد توكنات صادقة، ومعالجة سليمة للحالات التي يسيء فيها النموذج التصرّف. كل مقتطف يعمل كما هو؛ استبدل claude-sonnet-4-6 بـ gemini-2.5-pro وتقود الشيفرة نفسها نموذجاً مختلفاً.

الحلقة، ولماذا تحتاج إلى سقف

إليك المحرّك كاملاً. إنها حلقة استخدام الأدوات التي تعرفها أصلاً، مع إضافة واحدة تغيّر كل شيء: for step in range(MAX_ITERS) بدلاً من while True.

# حلقة الـ agent مع سقف صارم للتكرارات. النموذج -> tool_calls -> تشغيل ->
# إعادة تغذية النتائج -> تكرار، حتى يجيب النموذج نصاً أو نبلغ السقف
# الأعلى. السقف هو الفرق بين «agent» و«فاتورة منفلتة».
from openai import OpenAI
import json

client = OpenAI(
    api_key="sk-brievio-...",
    base_url="https://api.brievio.com/v1",
)

MAX_ITERS = 8   # معظم المهام تنتهي خلال 2-4 جولات؛ 8 هامش سخي.

def run_agent(question: str, model: str = "claude-sonnet-4-6") -> str:
    messages = [
        {"role": "system", "content": "You are a helpful research agent. "
         "Use the tools when you need live data. Answer directly when you "
         "already know enough."},
        {"role": "user", "content": question},
    ]

    for step in range(MAX_ITERS):
        resp = client.chat.completions.create(
            model=model,
            messages=messages,
            tools=TOOLS,
            tool_choice="auto",
        )
        msg = resp.choices[0].message

        # لم يُطلب أي tool -> هذه هي الإجابة النهائية. انتهى.
        if not msg.tool_calls:
            return msg.content

        # ألحق دور المساعد كما عاد بالضبط — فهو يحمل معرّفات
        # tool_call التي يجب أن تشير إليها الرسائل التالية.
        messages.append(msg)

        # شغّل كل نداء مطلوب وألحق رسالة tool واحدة لكل معرّف.
        for call in msg.tool_calls:
            result = dispatch(call)              # انظر المقتطف التالي
            messages.append({
                "role": "tool",
                "tool_call_id": call.id,          # يجب أن يطابق معرّف النداء
                "content": json.dumps(result),
            })
        # الحلقة: النموذج الآن يرى مخرجات الـ tool ويواصل.

    # بلغنا السقف دون حل. افشل بصوت عالٍ — لا تدر بصمت إلى الأبد.
    raise RuntimeError(f"agent did not finish within {MAX_ITERS} iterations")

ذلك الـ for المحدود هو أهمّ سطر منفرد في أي agent. النموذج القدير على مهمة محدّدة النطاق ينتهي خلال جولتين إلى أربع. لكن النماذج تختلط عليها الأمور: تنادي البحث نفسه مرتين، أو تلاحق طريقاً مسدوداً، أو — وهو الإخفاق الكلاسيكي — تنادي tool، فتحصل على نتيجة لا تعجبها، فتعيد نداءه بوسائط شبه مطابقة، إلى الأبد. الـ while True يحوّل ذلك إلى فاتورة بلا حدود وطلب معلّق. السقف يحوّل «الدوران إلى الأبد» إلى «الفشل بعد 8 محاولات مع خطأ واضح»، وهو شيء يمكنك التقاطه وتسجيله والتعافي منه. اختر الرقم من مهمتك: استعلام لمرة واحدة يحتاج 2، وagent بحثي متعدّد الخطوات ربما 10. حدّده بتعمّد؛ لا تتركه بلا حدّ.

لاحظ الحاجز الآخر المختبئ على مرأى الجميع: عالج حالة عدم نداء النموذج لأي tool. حين يكون msg.tool_calls فارغاً، فذلك يعني أن النموذج قرّر أن لديه ما يكفي للإجابة — هذا هو مخرجك، لا خطأ. الحلقة التي تفترض أن كل دورة تنتج نداء tool ستُسقط أو لن تنتهي أبداً. تفرّع على كلا النتيجتين في كل تكرار.

توزيع الأدوات: النموذج يقترح، وشيفرتك تتصرّف

النموذج لا يلمس أنظمتك أبداً. إنه يصدر اسم دالة وسلسلة JSON من الوسائط ثم يتوقّف؛ وشيفرتك تقرّر هل تستجيب لذلك. ذلك الحدّ هو قصة الأمان بأكملها لأي agent، لذا فإن دالة التوزيع هي حيث يعيش التحقّق — لا بوصفه ترفاً، بل لأن كل وسيط هو مخرَج نموذج غير موثوق، تماماً كحقل نموذج ملأه غريب.

# توزيع الأدوات مع تحقّق. النموذج يقترح النداء؛ شيفرتك
# تتصرّف فيه. كل وسيط هو مخرَج نموذج غير موثوق — حلّله،
# وتحقّق أن الاسم من تلك التي سجّلتها، وتأكّد من الأنواع قبل التشغيل.
def get_weather(city: str, unit: str = "celsius") -> dict:
    if not isinstance(city, str) or not city.strip():
        raise ValueError("city must be a non-empty string")
    if unit not in ("celsius", "fahrenheit"):
        raise ValueError(f"unsupported unit: {unit!r}")
    return {"city": city, "temp": 18, "unit": unit, "sky": "clear"}

# قائمة بيضاء: لا يمكن للنموذج استدعاء إلا ما سجّلته صراحةً.
TOOL_IMPLS = {"get_weather": get_weather}

TOOLS = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "Get current weather for a city.",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {"type": "string"},
                "unit": {"type": "string",
                         "enum": ["celsius", "fahrenheit"]},
            },
            "required": ["city"],
        },
    },
}]

def dispatch(call) -> dict:
    name = call.function.name
    fn = TOOL_IMPLS.get(name)
    if fn is None:
        # هلوس النموذج tool غير موجود. لا تُسقط الحلقة — أعد
        # الخطأ كنتيجة tool ليصحّح النموذج نفسه.
        return {"error": f"unknown tool: {name}"}

    try:
        args = json.loads(call.function.arguments)   # دائماً سلسلة JSON
    except json.JSONDecodeError:
        return {"error": "arguments were not valid JSON"}

    try:
        return fn(**args)
    except (TypeError, ValueError) as e:
        # وسائط سيئة (نوع خاطئ، حقل ناقص، خارج المدى). أعد
        # الرسالة؛ عادةً يعيد النموذج المحاولة بنداء مصحّح.
        return {"error": str(e)}

تُعالَج هنا ثلاث حالات إخفاق، وكلها تعيد الخطأ إلى النموذج بدلاً من إسقاط الحلقة. اسم tool مهلوس اخترعه النموذج لكنك لم تسجّله قط — القائمة البيضاء تلتقطه. JSON مشوّه في سلسلة الوسائط — نادر على نموذج رائد أصيل، لكنك تحلّله دفاعياً على أي حال. وقيم وسائط سيئة — نوع خاطئ، أو حقل مطلوب ناقص، أو enum اختلقه النموذج. في كل حالة، إعادة {"error": "..."} كنتيجة tool أفضل من إطلاق استثناء، لأن النموذج يقرأ تلك الرسالة في الدورة التالية وعادةً يصلح نداءه بنفسه. الـ agent القادر على التعافي من أخطائه أمتن بكثير من الذي يموت عند أول وسيط سيئ.

أبقِ القائمة البيضاء محكمة. TOOL_IMPLS.get(name) يعني أن النموذج — أصيلاً كان أو غير ذلك — لا يمكنه أبداً استدعاء إلا الدوال التي سجّلتها صراحةً. ذلك القاموس المنفرد هو نطاق انفجارك. إن كان tool يحذف بيانات، أو يخصم من بطاقة، أو يرسل بريداً، فضعه خلف تأكيد صريح بدلاً من ترك الحلقة تطلقه تلقائياً.

حارس الميزانية: الحلقات تعيد إرسال سياق متنامٍ

سقف التكرارات يحدّ كم مرة تنادي النموذج. لكنه لا يحدّ تكلفة كل نداء — وفي الحلقة، تتسلّق التكلفة كل جولة. السبب بنيوي: كل دورة تعيد إرسال المحادثة بأكملها حتى الآن، إضافة إلى كل نتيجة tool أُلحقت بها. الدورة الأولى قد تكون 800 توكن مدخلات؛ والدورة السادسة، بعد أن تتكدّس خمس مخرجات tool، قد تكون 6000. ثماني جولات رخيصة تتراكم بهدوء إلى تشغيل غير رخيص. الحلّ سقف ثانٍ مستقلّ على الإنفاق، محسوب من أعداد التوكنات الحقيقية التي يعيدها كل نداء:

# حارس ميزانية تكلفة/توكنات لكل تشغيل. كل دورة في الحلقة تعيد إرسال
# سياق متنامٍ (السجل + مخرجات الأدوات)، فترتفع التكلفة كل جولة. اقرأ
# كائن usage الصادق بعد كل نداء، سعّره، وتوقّف حين يتجاوز
# التشغيل ميزانيته — بمعزل عن سقف التكرارات.
from decimal import Decimal

# أسعار Brievio المنشورة، دولار لكل مليون توكن (~15% دون القائمة الرسمية).
RATES = {
    "claude-sonnet-4-6": {"in": Decimal("2.55"), "out": Decimal("12.75")},
    "claude-haiku-4-5":  {"in": Decimal("0.85"), "out": Decimal("4.25")},
}

def call_cost(model: str, usage) -> Decimal:
    r = RATES[model]
    m = Decimal("1000000")
    return usage.prompt_tokens * r["in"] / m + usage.completion_tokens * r["out"] / m

RUN_BUDGET = Decimal("0.10")   # 10 سنتات لكل تشغيل agent، سقف صارم.

def run_agent_budgeted(question: str, model: str = "claude-sonnet-4-6") -> str:
    messages = [{"role": "user", "content": question}]
    spent = Decimal("0")

    for step in range(MAX_ITERS):
        resp = client.chat.completions.create(
            model=model, messages=messages, tools=TOOLS, tool_choice="auto",
        )
        spent += call_cost(model, resp.usage)   # أحصِ التوكنات الحقيقية، كل دورة
        if spent > RUN_BUDGET:
            raise RuntimeError(f"run exceeded ${RUN_BUDGET} (spent ${spent:.4f})")

        msg = resp.choices[0].message
        if not msg.tool_calls:
            return msg.content

        messages.append(msg)
        for call in msg.tool_calls:
            messages.append({"role": "tool", "tool_call_id": call.id,
                             "content": json.dumps(dispatch(call))})

    raise RuntimeError(f"agent did not finish within {MAX_ITERS} iterations")

المفتاح أن resp.usage على Brievio يحمل أعداد توكنات المدخلات والمخرجات الصادقة التي عالجها النموذج الأصيل فعلاً — فيكون المجموع الجاري مالاً حقيقياً، لا تخميناً. قراءة usage بعد كل دورة والتوقّف عند RUN_BUDGET يعني أن agent مرتبكاً كان سيحرق لولا ذلك ثماني جولات باهظة يُقطَع عنه التيار لحظة تجاوزه عشرة سنتات، بصرف النظر عن عدد التكرارات التي استغرقها. سقفان، وحالتا إخفاق مختلفتان مغطّاتان: سقف التكرارات يوقف الحلقات اللانهائية، والميزانية توقف الباهظة. تريد كليهما، لأن الحلقة قد تكون قصيرة باهظة أو طويلة رخيصة، ولا يحميك أحدهما وحده من الآخر.

مما يجدر معرفته للحساب: النداءات الفاشلة 4xx/5xx لا تُحتسَب على Brievio، فإعادة المحاولة على tool متقلّب أو خطأ upstream عابر لا تستنزف ميزانية التشغيل — فأنت تحصي التكلفة فقط للنداءات التي أعادت نتيجة فعلاً. وهذا يبقي منحنى الإنفاق متتبّعاً للعمل المنجَز، لا للأخطاء المُمتصّة. والنمط الكامل لتحديد الإنفاق لكل نداء ولكل مستخدم موجود في دليل تحديد سقف إنفاق الـ API.

إبقاء فاتورة التوكنات منخفضة مع نمو الحلقة

تحديد التكلفة شيء؛ وخفضها شيء آخر. ولأن كل دورة تعيد إرسال بادئة متنامية، يُدفَع ثمن السياق نفسه مراراً وتكراراً — وهذا تحديداً الشكل الذي بُني له تخزين الـ prompt المؤقت. علّم الأجزاء الثابتة من الطلب (الـ system prompt، وتعريفات الأدوات) على أنها قابلة للتخزين المؤقت، ومن الدورة الثانية فصاعداً تدفع جزءاً يسيراً من سعر المدخلات على كل ما لم يتغيّر. في حلقة تعيد إرسال كتالوج أدوات وsystem prompt بحجم آلاف التوكنات في كل جولة على حدة، يكون ذلك أكبر رافعة منفردة على الفاتورة.

تساعد أيضاً عادتان عمليتان. أبقِ الـ system prompt وتعريفات الأدوات مستقرّة عبر التشغيل — فإضافة tool في منتصف الحلقة أو طابع زمني في الـ system prompt يُبطل الـ cache ويضاعف بهدوء تكلفة مدخلاتك. وإن كان tool قد يعيد جداراً من البيانات (صفحة ويب كاملة، استعلاماً بألف صف)، فلخّص النتيجة أو اقتطعها قبل إلحاقها بـ messages؛ فالنموذج نادراً ما يحتاجها كلها، وكل بايت تلحقه يُعاد إرساله في كل دورة لاحقة. حلقة الـ agent حسّاسة بشكل فريد لتضخّم السياق لأن السياق يُعاد إرساله N مرة، لا مرة واحدة.

ذاكرة المحادثة: ما الذي تحمله بين التشغيلات

كل ما سبق هو ذاكرة داخل تشغيل واحد — قائمة messages هي ذاكرة العمل لدى الـ agent، وإلحاقها بها هو كيف يتذكّر النموذج ما استعلم عنه أصلاً. أما agent متعدّد الأدوار يتحدّث إلى مستخدم عبر عدة طلبات، فتحمل تلك القائمة قدماً: خزّن messages لكل جلسة (Redis، أو عمود قاعدة بيانات، أو حيثما كان)، وأعد تحميلها في الطلب التالي، وألحق دور المستخدم الجديد. الحلقة متطابقة؛ تتغيّر حالة البدء فقط.

الأمر الذي يجب إدارته هو النمو بلا حدود. الجلسة طويلة العمر تراكم سجلاً حتى تصبح باهظة في كل نداء وتتجاوز في النهاية نافذة السياق. استراتيجيتان شائعتان: احتفظ بنافذة متدحرجة لآخر N دور وأسقط الأقدم، أو لخّص دورياً السجل الأقدم في ملاحظة مدمجة واستبدل بها الأدوار الخام. كلاهما يقايض بعض الدقة بحجم سياق محدود ومتوقَّع. وأياً اخترت، يبقى حارس ميزانية التشغيل أعلاه سارياً — فهو شبكة الأمان التي تلتقط جلسة نمت أكبر مما خطّطت له.

مفتاح واحد لـ agent يتصاعد

خاصية مفيدة لبناء هذا خلف Brievio: يمكن للـ agent أن يغيّر النموذج الذي يستخدمه في منتصف المهمة دون تغيير أي شيء آخر. شغّل الجولات الرخيصة على نموذج أصغر وتصاعد إلى نموذج رائد فقط حين تكون المهمة صعبة — وجّه توزيع الأدوات السهل عبر Haiku 4.5 بسعر 0.85 دولار مدخلات / 4.25 دولار مخرجات، وارجع إلى Sonnet أو عائلة مختلفة للإجابة النهائية كثيفة الاستدلال. ولأن مفتاحاً واحداً يغطّي كل نموذج خلف base_url واحد، فإن ذلك التصاعد تغيير من سطر واحد في سلسلة model داخل الحلقة — بلا SDK ثانٍ، ولا مخطّط مصادقة ثانٍ، ولا علاقة فوترة ثانية. العقد الكامل للطلب/الاستجابة، بما فيه حقول الأدوات، موجود في وثائق Chat Completions، وقائمة النماذج الحيّة بمعرّفاتها الدقيقة على صفحة النماذج.

ولا يهمّ ذلك، بالطبع، إلا إذا كان النموذج على الطرف الآخر أصيلاً: حلقة الـ agent لا تتسامح مع بديل مُنزَّل المرتبة، لأن نموذجاً يتعثّر في وسائط الأدوات أو يتجاهل tool سيحرق تكرارات وإنفاقاً وهو يلاحق أخطاءه. يقدّم Brievio النماذج الأصلية من الطرف الأول، ويحترم نداء الأدوات الأصيل، ويبلّغ عن أعداد توكنات صادقة — وهذا ما يجعل الحلقة وحساب الميزانية يعملان فعلاً.

الخلاصة: أربعة حواجز، ثم انشر

الحلقة نفسها دزينة أسطر. ما يجعلها جاهزة للإنتاج هو الحدّ المحيط بها:

• حدّد سقف التكرارات. for محدود بدلاً من while True. افشل بصوت عالٍ عند السقف بدلاً من الدوران إلى الأبد.
• عالج حالة عدم وجود tool. الـ tool_calls الفارغ هو المخرج، لا خطأ. تفرّع عليه كل دورة.
• تحقّق من كل توزيع. أدرج أسماء الأدوات في قائمة بيضاء، وحلّل الوسائط، وتأكّد من الأنواع — وأعد الأخطاء إلى النموذج بدلاً من الإسقاط.
• ضع ميزانية للتشغيل. اقرأ usage كل دورة، وسعّرها مقابل الأسعار المنشورة، وتوقّف عند سقف إنفاق صارم. راقب السياق المتنامي، وخزّن البادئة الثابتة مؤقتاً لإبقاء تكلفة إعادة الإرسال منخفضة.

أتقن هذه الأربعة وتحصل على agent يؤدّي عملاً حقيقياً متعدّد الخطوات، ويتعافى من أخطائه، وله تكلفة في أسوأ الأحوال اخترتها أنت بدلاً من اكتشافها على فاتورة. ابدأ من دليل استخدام الأدوات إن احتجت آلية النداء المفرد أولاً، ثم غلّفها بالحلقة وحواجز الحماية الأربعة أعلاه.