حلقة الوكيل

طريقان إلى نفس الفكرة

قبل أن تبدأ

الفكرة الجوهرية

كل وكيل هو نفس الحلقة. في هذا الدرس تكتبها بيدك، ثم تضيف كشف الدوران، ثم السجل، ثم تضع نموذجًا حقيقيًّا مكان النموذج المكتوب مسبقًا.

1الحلقة وحدّ الخطوات

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

انظر إلى الرسم: الرسائل تكبر مع كل خطوة. نتيجة كل أداة تُلحق بها، وهكذا «يعرف» النموذج ما حدث.

import json

def get_order_status(order_id):
    return {"48213": {"status": "delayed", "reason": "weather"}}.get(order_id, {"error": "not found"})

def get_refund_policy(reason):
    return {"policy": "Delays over 3 days due to weather: free shipping refund."}

TOOLS = {"get_order_status": get_order_status, "get_refund_policy": get_refund_policy}

def scripted_model(messages):
    """Stands in for the model so the loop runs here. A real model decides from the messages."""
    tool_results = [m for m in messages if m["role"] == "tool"]
    if len(tool_results) == 0:
        return {"tool": "get_order_status", "args": {"order_id": "48213"}}
    if len(tool_results) == 1:
        return {"tool": "get_refund_policy", "args": {"reason": "weather"}}
    return {"answer": "Your order 48213 is delayed by weather; you qualify for a shipping refund."}

def run_agent(question, max_steps=5):
    messages = [{"role": "user", "content": question}]
    for step in range(1, max_steps + 1):
        action = scripted_model(messages)
        if "answer" in action:
            print(f"step {step}: answer -> {action['answer']}")
            return action["answer"]
        result = TOOLS[action["tool"]](**action["args"])
        print(f"step {step}: {action['tool']}({json.dumps(action['args'])}) -> {json.dumps(result)}")
        messages.append({"role": "assistant", "content": json.dumps(action)})
        messages.append({"role": "tool", "content": json.dumps(result)})
    print(f"stopped: reached {max_steps} steps without an answer")
    return None

run_agent("Why is my order 48213 late, and do I get anything back?")

النموذج هنا دالة مكتوبة مسبقًا تقرّر الخطوات بدل نموذج حقيقي، حتى تعمل الحلقة في المتصفّح بلا مفتاح API. كود الحلقة نفسه لا يتغيّر حين تضع نموذجًا حقيقيًّا مكانها. الخطوة الأولى جلبت الحالة وسبب التأخير، والثانية استخدمت السبب لجلب السياسة، والثالثة أجابت. لاحظ أن كل نتيجة أداة أُلحقت بالرسائل، فهكذا «يعرف» النموذج ما حدث. وmax_steps=5 يضمن أن الحلقة تنتهي مهما فعل النموذج.

2كشف الدوران

الدوران تكرار نفس الاستدعاء بنفس المعاملات، غالبًا لأن الأداة لا تعطي النموذج ما يحتاجه. حد الخطوات يوقفه في النهاية، لكن بعد خطوات ضائعة.

انظر إلى الرسم: الحلقة تعدّ كل استدعاء متطابق. حين يصل العدد إلى 3، أي أكثر من max_repeats = 2، تتوقّف وتسلّم الأمر لإنسان.

أداة بحث لا تجد شيئًا، ونموذج يكرّر نفس البحث. بدون كشف، سيدور حتى الحد الأقصى.

import json

def search(query):
    return {"results": []}                       # the tool keeps finding nothing

def stuck_model(messages):
    return {"tool": "search", "args": {"query": "refund policy weather"}}

def run_agent(model, max_steps=8, max_repeats=2):
    messages, seen = [], {}
    for step in range(1, max_steps + 1):
        action = model(messages)
        key = json.dumps(action, sort_keys=True)
        seen[key] = seen.get(key, 0) + 1
        if seen[key] > max_repeats:
            return f"stopped at step {step}: same call repeated {max_repeats} times, handing over to a person"
        result = search(**action["args"])
        print(f"step {step}: search -> {result}")
        messages.append({"role": "tool", "content": json.dumps(result)})
    return f"stopped: step cap of {max_steps} reached"

print(run_agent(stuck_model))

الحلقة تحفظ كل استدعاء كنص JSON مرتّب المفاتيح، وتعدّ تكراره. عند التكرار الثالث تتوقّف وتسلّم لإنسان، بدل أن تحرق خمس خطوات أخرى بلا فائدة. في نظام حقيقي، أعد للنموذج أيضًا رسالة تقول «هذا البحث لم يجد شيئًا مرتين، جرّب صياغة أخرى أو اسأل المستخدم» قبل الإيقاف.

3السجل

السجل يحفظ كل خطوة: ماذا طلب النموذج، وبأي معاملات، وماذا أعادت الأداة، وكم استغرقت وكلّفت. بدونه لا تستطيع تشخيص أي فشل.

انظر إلى الرسم: سطر واحد لكل خطوة بخمسة حقول. هذا أول ما تقرؤه حين يخطئ الوكيل.

4الحلقة مع نموذج حقيقي

مع نموذج حقيقي يتغيّر سطر واحد: scripted_model يصير استدعاء API مع قائمة الأدوات. والرد إما نص هو الإجابة النهائية، أو طلب أداة باسمها ومعاملاتها.

def run_agent(client, question, tools, impl, max_steps=10):
    messages = [{"role": "user", "content": question}]
    for step in range(max_steps):
        reply = client.chat.completions.create(model=MODEL, messages=messages, tools=tools)
        msg = reply.choices[0].message
        if not msg.tool_calls:
            return msg.content                              # the final answer
        messages.append(msg)
        for call in msg.tool_calls:
            args = json.loads(call.function.arguments)
            result = impl[call.function.name](**args)
            log_step(step, call.function.name, args, result, reply.usage)
            messages.append({"role": "tool", "tool_call_id": call.id, "content": json.dumps(result)})
    return "I could not finish this within the step limit; a person will follow up."

نفس البنية بالضبط. الفرق أن النموذج قد يطلب عدّة أدوات في رد واحد، فنحلّق عليها، وكل نتيجة تُربط بمعرّف طلبها tool_call_id. وlog_step يكتب السجل مع استهلاك الرموز لكل خطوة.

كيف يعمل، خطوة بخطوة: دورة واحدة في حلقة الوكيل

مثال محلول

عميل يسأل لماذا تأخّر طلبه وهل يستحق تعويضًا. يحتاج الوكيل خطوتين قبل أن يجيب: حالة الطلب، ثم سياسة التعويض بحسب سبب التأخير.

جرّب بلا كود

خرافة مقابل حقيقة

جرّب بنفسك

بطاقات مراجعة سريعة

تحقق من فهمك