وكلاء على واجهة برمجة تطبيقات LLM غير الخاضعة للرقابة: استدعاء الأدوات، أطر العمل، ووسائل الحماية

تشريح حلقة استدعاء الأدوات

انزع أطر العمل وستلاحظ أن كل وكيل يقوم بالأربع خطوات نفسها. ترسل قائمة المحادثة بالإضافة إلى قائمة tools. يرد النموذج إما بنص، مما يعني أنه انتهى، أو بـ tool_calls، مما يعني أنه يريد تنفيذ شيء ما. تشغّل كل استدعاء وتضيف رسالة tool واحدة لكل استدعاء، مع الإشارة إلى id الخاص بالاستدعاء. ثم تطلب من النموذج مرة أخرى مع المحادثة الأطول.

إشارة التفرع هي finish_reason. عندما تساوي tool_calls، تحتوي رسالة المساعد على content مضبوطة على null ومصفوفة tool_calls، يحتوي كل عنصر على معرّف، واسم دالة، وسلسلة حجة بصيغة JSON. يعرض صفحة المرجع الشكل الدقيق. تفاصيلان يسببان مشاكل: الحجج هي سلسلة يجب تحليلها، ويجب أن تبقى رسالة المساعد التي تحتوي على الاستدعاءات في السجل قبل نتائج الأدوات الخاصة بك، بالترتيب.

حلقة كاملة من الصفر

يحدد السcript أدناه دالة واحدة، ويصفها بمخطط JSON، ويدور حتى ينتج النموذج نصًا عاديًا. يستخدم SDK الرسمي لـ Python الموجه إلى واجهة برمجة التطبيقات. كل حلقة تلصق رسالة المساعد، ثم رسالة أداة لكل استدعاء، وعدّاد خطوات يوقف السلوك الهارب.

Copyimport json, os
from openai import OpenAI

client = OpenAI(base_url="https://api.uncensoredapis.com/v1", api_key=os.environ["API_KEY"])

TICKETS = {"T-101": "open", "T-102": "closed"}

def ticket_status(ticket_id: str) -> str:
    return TICKETS.get(ticket_id, "unknown ticket")

TOOLS = [{
    "type": "function",
    "function": {
        "name": "ticket_status",
        "description": "Look up the status of a support ticket by id.",
        "parameters": {
            "type": "object",
            "properties": {"ticket_id": {"type": "string"}},
            "required": ["ticket_id"],
        },
    },
}]
REGISTRY = {"ticket_status": ticket_status}

messages = [{"role": "user", "content": "Are tickets T-101 and T-102 still open?"}]

for step in range(6):                      # hard cap on loop turns
    r = client.chat.completions.create(
        model="uncensored", messages=messages, tools=TOOLS, max_tokens=400)
    msg = r.choices[0].message
    messages.append(msg.model_dump(exclude_none=True))
    if r.choices[0].finish_reason != "tool_calls":
        print(msg.content)
        break
    for call in msg.tool_calls:
        fn = REGISTRY.get(call.function.name)
        try:
            args = json.loads(call.function.arguments)
            result = fn(**args) if fn else f"error: unknown tool {call.function.name}"
        except Exception as exc:
            result = f"error: {exc}"
        messages.append({"role": "tool", "tool_call_id": call.id, "content": str(result)})
else:
    print("stopped: step limit reached")

لاحظ ما تفعله الحلقة عند الفشل: تصبح الأخطاء نتائج دوال كنصوص. هذا يسمح للنموذج بقراءة المشكلة والمحاولة مرة أخرى، مثل تصحيح معرف تذكرة خاطئ، بدلاً من تعطل البرنامج. تُفعّل عبارة for ... else فقط إذا تم الوصول إلى الحد الأقصى دون إجابة نهائية، وهي الطريقة التي تكتشف بها أن النموذج عالق في حلقات.

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

حماية تنفيذ الأدوات

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

• السماح بالاسم. ابحث عن الدالة في سجل قمت بإنشائه. لا تستخدم النص أبدًا لاستيراد أو تقييم الكود.
• افصل القراءة عن الكتابة. تعمل أدوات القراءة فقط بحرية. أي شيء يرسل بريدًا إلكترونيًا، أو ينفق أموالًا، أو يحذف بيانات، أو يلمس الملفات يحتاج إلى موافقة صريحة من الإنسان.
• تحقق من الحجج مقابل المخطط الذي أعلنته: المفاتيح المطلوبة، لا مفاتيح إضافية، الأنواع وحدود الطول.
• حدّد الوقت والمخرجات. طبّق مهلات زمنية، واقتطع النتائج قبل إعادة تغذيتها حتى لا تستهلك مخرجات الدلة الضخمة نافذة السياق الخاصة بك التي تتكون من 100,000 رمز.
• حدّ التكرارات والتكلفة. قيّد عدد الخطوات لكل مهمة ورمز لكل جلسة، باستخدام كتلة usage.
• سجل كل استدعاء بالاسم، الحجج والنتيجة حتى تتمكن من تدقيق السلوك.

Copyimport json
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutTimeout

ALLOWED = {"ticket_status", "lookup_invoice"}      # read-only tools only
NEEDS_APPROVAL = {"refund_order"}                  # side effects wait for a human
POOL = ThreadPoolExecutor(max_workers=4)

def validate(args: dict, schema: dict) -> None:
    props = schema["properties"]
    for key in schema.get("required", []):
        if key not in args:
            raise ValueError(f"missing argument: {key}")
    for key, val in args.items():
        if key not in props:
            raise ValueError(f"unexpected argument: {key}")
        if props[key]["type"] == "string" and not isinstance(val, str):
            raise ValueError(f"{key} must be a string")
        if isinstance(val, str) and len(val) > 200:
            raise ValueError(f"{key} too long")

def run_tool(name, raw_args, registry, schemas, approve=lambda n, a: False):
    if name in NEEDS_APPROVAL:
        args = json.loads(raw_args)
        if not approve(name, args):
            return "denied: this action needs human approval"
    elif name not in ALLOWED:
        return f"error: tool {name} is not permitted"
    args = json.loads(raw_args)
    validate(args, schemas[name]["parameters"])
    future = POOL.submit(registry[name], **args)
    try:
        return str(future.result(timeout=5))[:2000]    # time and size limits
    except FutTimeout:
        return "error: tool timed out"

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

كتابة مخططات الأدوات التي يمكن للنموذج استخدامها

المخطط هو التوثيق الوحيد الذي يراه النموذج، لذا يستحق نفس العناية التي توليها لواجهة برمجة تطبيقات عامة. سمِّ الدوال بأفعال وأسماء تتطابق مع المهمة، مثل ticket_status أو lookup_invoice، بدلاً من العناوين العامة مثل run أو helper. اكتب description كتعليمات: ما تفعله الدالة، وما الذي تعيده، ومتى لا تستخدمها.

• حافظ على قلة المسطّحات في المعاملات. حقلان أو ثلاثة حقول بسيطة تُنادى بشكل صحيح في كثير من الأحيان أكثر من كائن متداخل يحتوي على عشرة حقول.
• استخدم القوائم المحددة للمجموعات المغلقة، على سبيل المثال "enum": ["open", "closed"]، حتى لا يخترع النموذج قيمة.
• صف كل خاصية بوحدة أو مثال: "معرف تذكرة مثل T-101" يتفوق على سلسلة عارية.
• أعد نتائج مدمجة. تعمل النصوص أو كائنات JSON الصغيرة بشكل أفضل؛ النصوص الطويلة تستهلك السياق مقابل مكسب قليل.
• فضّل عدة أدوات متخصصة على أداة شاملة واحدة تحتوي على علم وضع.

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

إعداد LangChain

تقبل فئة ChatOpenAI في LangChain base_url، لذا لا حاجة لحزمة مزود مخصصة. اضبط model على uncensored، مرر المفتاح من بيئتك، واربط الأدوات بـ bind_tools. ثبّت langchain-openai وlangchain-core.

Copyimport os
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

@tool
def ticket_status(ticket_id: str) -> str:
    """Return the status of a support ticket."""
    return {"T-101": "open", "T-102": "closed"}.get(ticket_id, "unknown ticket")

llm = ChatOpenAI(
    model="uncensored",
    base_url="https://api.uncensoredapis.com/v1",
    api_key=os.environ["API_KEY"],
    temperature=0.3,
    max_tokens=400,
)
llm_with_tools = llm.bind_tools([ticket_status])

ai = llm_with_tools.invoke("What is the status of ticket T-101?")
for call in ai.tool_calls:
    print(call["name"], call["args"], "->", ticket_status.invoke(call["args"]))

نتيجة استدعاء invoke هي رسالة AIMessage تحتوي قائمة tool_calls على الأسماء والمعاملات محللة، لذا تتخطى خطوة JSON اليدوية. للحصول على حلقة مستقلة كاملة، مر كائن llm نفسه إلى منشئي الوكلاء في LangChain؛ نقطة النهاية يجب أن تحترم فقط استدعاء الدوال، وهو ما تفعله. إذا بدا أن الوكيل يتجاهل أدواته، تحقق من وجود نص التوثيق وتلميحات الأنواع، لأنها تصبح المخطط الذي يراه النموذج.

إعداد LlamaIndex

تأتي LlamaIndex مع OpenAILike للخوادم التي تتبع تنسيق إكمال الدردشة. ثلاث إعدادات مهمة. يشير api_base إلى واجهة برمجة التطبيقات، ويحدد is_chat_model=True مسار الدردشة، ويخبر is_function_calling_model=True الإطار بأنه قد يرسل أدوات. اضبط context_window على 100000 حتى تقوم مقسّمات النص ومخازن الذاكرة بتحديد أحجامها بشكل صحيح. ثبّت llama-index-llms-openai-like.

Copyimport os
from llama_index.llms.openai_like import OpenAILike
from llama_index.core.tools import FunctionTool

def ticket_status(ticket_id: str) -> str:
    """Return the status of a support ticket."""
    return {"T-101": "open", "T-102": "closed"}.get(ticket_id, "unknown ticket")

llm = OpenAILike(
    model="uncensored",
    api_base="https://api.uncensoredapis.com/v1",
    api_key=os.environ["API_KEY"],
    is_chat_model=True,
    is_function_calling_model=True,
    context_window=100000,
    max_tokens=400,
)

tool = FunctionTool.from_defaults(fn=ticket_status)
response = llm.predict_and_call([tool], "Check ticket T-102 for me.")
print(response)

من هنا يمكنك تمرير نفس كائن llm إلى عامل وكيل أو محرك استعلام. تذكر أن خطوط استرداد البيانات يمكن أن تملأ موجزًا واحدًا بالعديد من الأجزاء؛ مع نافذة تتكون من 100,000 رمز مشتركة بين الموجّه والإكمال، قيّد عدد الأجزاء المسترجعة واحتفظ بـ max_tokens معتدلًا.

تشغيل الوكلاء بميزانية

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

ينمو السياق مع كل جولة لأن كل نتيجة أداة تُضاف. لخص أو احذف مخرجات الأدوات القديمة بمجرد استخدامها، وراقب usage.prompt_tokens؛ عندما يقترب من 50,000، اضغط التاريخ. بالنسبة للتكلفة، افترض مهمة توضيحية مكونة من 6 استدعاءات بمتوسط 3,000 رمز موجّه و200 رمز إكمال: هذا يعني 18,000 رمز إدخال (حوالي $0.0045 بسعر $0.25 لكل مليون) و1,200 رمز إخراج (حوالي $0.0012 بسعر $1.00 لكل مليون). هذه الأرقام هي افتراضات للمجموعات، وليست قياسات. إذا كنت جديدًا على نقطة النهاية، ابدأ بـ المستندات ورصيد التجريب المجاني.

اختبار وتصحيح خطأ وكيل

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

1. زيف النموذج. استبدل العميل بحيث يعيد رسائل tool_calls مبرمجة مسبقًا. هذا يختبر سجلك والتحقق من الصحة وإنهاء الحلقة بدون شبكة وبدون إنفاق.
2. ثبت العينة. استخدم درجة temperature منخفضة، مثل 0.2، للمهام التي تعتمد على الأدوات. تقلل العشوائية الأقل من تنوع JSON للمعاملات وتعطي JSON معاملات أكثر تكرارًا، ويمكنك رفعها لاحقًا للخطوات التي تعتمد على الكتابة.
3. فرض على المسارات. لمجموعة صغيرة من الموجّهات الثابتة، تحقق من الأدوات التي تم استدعاؤها وبأي ترتيب، وليس فقط النص النهائي.

راقب finish_reason أيضًا. قيمة length في منتصف استدعاء أداة تعني أن المعاملات قُطعت؛ ارفع max_tokens (حتى 32,000) وستكتمل JSON. يعمل البث المتدفق مع الأدوات أيضًا، لكن اجمع أجزاء المعاملات لكل فهرس استدعاء وشرّح فقط بعد إغلاق البث. للحصول على مراجعة سريعة حول تلك الأجزاء، انظر إلى قسم البث المتدفق في المرجع.