حلقة الاستدعاء لوكلاء الذكاء الاصطناعي المخصصين بعد انتهاء العرض التوضيحي وطرق التحكم في تكاليف الرموز (Tokens)
makedream28 de julho de 2026
0
Computing/Software
Comments (0)
Log in to leave a comment
No posts yet
makedreammakedreamLog in to leave a comment
No posts yet
تجري عملية استيراد أطر العمل وبناء وكيل ذكاء اصطناعي (AI Agent) باستخدام البيانات الداخلية للشركة بسلاسة حتى مرحلة النموذج الأولي (Prototype). لكن المشكلة الحقيقية تبدأ في اللحظة التي يتم فيها إطلاق هذا الكود في بيئة الإنتاج الفعلية. فالوكيل الذي كان يعمل بشكل ممتاز داخل الشاشة، قد يعلق في إدخال غير متوقع ويكرر الاستدعاء في حلقة لا نهائية، مما قد يتسبب في فرض فواتير واجهة برمجية التطبيقات (API) بآلاف الدولارات خلال عطلة نهاية الأسبوع.
إن معظم الأعطال التي تحدث في بيئة العمل الحقيقية ترجع إلى الفشل في إدارة الحالة وغياب التحكم الخارجي، أكثر من كونها قيودًا في نموذج الذكاء الاصطناعي (LLM) نفسه. لقد قمت بإنشاء أداة مخصصة بناءً على تعليمات الإدارة، ولكن يجب إنهاء هذا الوضع الذي تقضي فيه اليوم بطوله في تصحيح أخطاء الهلوسة للوكيل التي تظهر بشكل غير متوقع، والتعامل مع إيصالات الرموز المتراكمة، بدلاً من التركيز على عملك الأساسي.
بمجرد خروج الوكيل من بيئة العرض التوضيحي، فإنه يصطدم بثلاثة جدران: العمليات المظلمة (Black-box operations)، والحلقات غير المحددة (Non-deterministic loops)، ومخاطر تسريب البيانات.
إذا لم تسجل المدخلات والمخرجات في كل مرحلة من مراحل التفكير (Chain-of-Thought)، واستدعاء الأدوات الخارجية (Tool Call)، والبحث في قاعدة بيانات المتجهات (Vector Database)، فلن تتمكن من معرفة السبب عند حدوث مشكلة. وسوف تستهلك يومًا كاملاً لمجرد تتبع المكان الذي ارتبكت فيه المطالبة (Prompt).
والأمر الأكثر خطورة هو الحلقة اللانهائية. فالوكيل الذي يعتمد على هيكلية ReAct (Reasoning + Acting) يعيد إرسال الطلب نفسه مرارًا وتكرارًا إذا كانت نتيجة تنفيذ الأداة غموضًا.
┌─────────────────────────────────────────────────────────────────────────┐ │ ReAct Architecture Loop │ │ │ │ ┌────────────┐ User Query ┌────────────┐ Tool Call Request │ │ │ User │ ──────────────> │ Main LLM │ ────────────────────┐ │ │ └────────────┘ └────────────┘ │ │ │ ▲ ▼ │ │ │ Observation ┌──────┐ │ │ │ (Ambiguous/Failed) │ Tool │ │ │ └─────────────────────── │ A │ │ │ └──────┘ │ │ * Problem: When Observation fails, LLM retries Tool A endlessly. │ └─────────────────────────────────────────────────────────────────────────┘
في اللحظة التي يفشل فيها الوكيل في الخروج من شرط الانتهاء ويستمر في ضرب الـ API بنفس الوسائط، تستنفد ميزانية الرموز المخصصة للشهر بأكمله في غضون دقائق معدودة. وتعد الفشل المتسلسل الناجم عن استدعاء الوكيل الرئيسي والوكلاء الفرعيين لبعضهم البعض وإعادة المحاولة مسؤولاً عن أكثر من 30% من إجمالي أعطال النظام.
كما تظهر مشكلة تراجع أداء النموذج نفسه عندما تتراكم سجلات المحادثة وتملأ نافذة السياق (Context Window). بالإضافة إلى ذلك، فإن تضمين المطالبات كنصوص ثابته (Hardcoded) داخل كود Python يجعل تغيير عبارة بسيطة يتطلب إعادة بناء النظام بالكامل وإعادة نشره.
يجب فصل حلول تصحيح الأخطاء المحلية عن أدوات الملاحظة والرقابة في بيئة الإنتاج لتحديد نقاط التجميع. في مرحلة التطوير، نتحقق من جودة تضمين RAG واستدعاءات الأدوات باستخدام Arize Phoenix المعتمد على OpenTelemetry. أما في بيئة الخدمة الفعلية، فنقوم بربط Langfuse الذي يستخدم ClickHouse كخلفية (Backend) لمراقبة سجلات التتبع (Traces) واستهلاك الرموز في الوقت الفعلي.
┌─────────────────────────────────────────────────────────────────────────┐ │ Semantic Caching & Routing Flow │ │ │ │ Client Query │ │ │ │ │ ▼ │ │ ┌───────────┐ Similarity >= 0.92? ┌─────────────────────────────┐ │ │ │ Redis Vector │ ─────────────────────────> │ Return Cached Response │ │ │ │ Cache │ (Cache Hit) │ (Latency -88%, Cost -86%) │ │ │ └───────────┘ └─────────────────────────────┘ │ │ │ │ │ │ (Cache Miss) │ │ ▼ │ │ ┌───────────┐ Execution & Save Cache ┌─────────────────────────────┐ │ │ │ External │ ────────────────────────> │ Store Result as Vector in │ │ │ │ LLM API │ │ Backend Database │ │ │ └───────────┘ └─────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────────────┘
تتم حماية تكاليف API الناتجة عن تكرار نفس الاستعلامات باستخدام طبقة التخزين المؤقت الدلالي (Semantic Caching).
وفقًا لبيانات تحليل 60,000 استعلام نشرتها AWS، عند تطبيق التخزين المؤقت الدلالي باستعمال معيار تشابه جيب التمام (Cosine Similarity)، انخفضت تكلفة استنتاج LLM بنسبة تصل إلى 86% وتحسن زمن الاستجابة بنسبة 88%. كما خفض إطار العمل RouteLLM من LMSYS التكاليف بنسبة 85% عن طريق تقسيم الطلبات بين النماذج عالية التكلفة ومنخفضة التكلفة بناءً على صعوبة السؤال.
يتم بناء التخزين المؤقت الدلالي وفقًا للخطوات التالية:
ولحظر الحلقات اللانهائية بشكل جذري، يجب إدراج نمط قاطع الدائرة (Circuit Breaker) الذي يتحقق من حالة الوكيل مباشرة في تدفق التحكم في الأنبوب (Pipeline). الاعتماد على خيارات إعادة المحاولة الافتراضية التي يوفرها إطار العمل قد يؤدي إلى انهيار الخدمة بأكملها عند حدوث استثناء. من الآمن تشفير اسم الأداة وقيم الوسائط باستخدام SHA-256 ووضعها في قائمة، وإذا تراكم نفس التشفير 3 مرات متتالية، يتم توجيه التدفق قسرًا إلى عقدة أخرى.
`python
import hashlib
from typing import TypedDict, Annotated, List
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
messages: Annotated[List[BaseMessage], add_messages]
steps: int
tool_hashes: List[str]
def hash_tool_call(tool_name: str, tool_args: str) -> str:
raw_str = f"{tool_name}:{tool_args}"
return hashlib.sha256(raw_str.encode('utf-8')).hexdigest()
def agent_circuit_breaker_router(state: AgentState) -> str:
# 1. إذا تجاوز إجمالي عدد مرات التنفيذ 5 مرات، انتقل إلى عقدة التراجع المحددة
if state["steps"] > 5:
return "fallback_graceful_node"
# 2. إذا تم استدعاء نفس الأداة والوسيط 3 مرات متتالية، يتم الحظر فورًا
hashes = state.get("tool_hashes", [])
if len(hashes) >= 3 and hashes[-1] == hashes[-2] == hashes[-3]:
return "fallback_graceful_node"
# 3. التحقق من الكلمة المفتاحية للإنهاء الطبيعي
last_message = state["messages"][-1]
if "FINAL_ANSWER" in last_message.content:
return "end"
return "continue_tools"
`
بغض النظر عن حالة المخرجات غير المحددة لـ LLM، فإن هذا الموجه (Router) يكتشف الجمل الشرطية في بيئة تشغيل Python، مما يمنع فيزيائيًا الدخول في حلقة ناتجة عن الهلوسة.
إذا تركت المطالبات داخل الكود، فلن تتمكن من اكتشاف أعطال التراجع (Regression Failures) حيث تتوقف الميزات التي كانت تعمل سابقًا بعد تعديل النموذج. يتم إخراج المطالبات من كود Python وإدارتها في ملفات YAML مستقلة.
`yaml
name: "agent_reasoning"
version: "1.2.0"
model: "gpt-4o"
temperature: 0.1
messages:
يتم اختبار المطالبات المنفصلة بهذه الطريقة تلقائيًا في خط أنابيب CI/CD عن طريق دمج إطار التقييم مفتوح المصدر DeepEval مع Pytest. نحدد مقياس G-Eval كنقطة مرجعية ونتحقق من أرقام استجابة النموذج قبل النشر.
`python
import pytest
from deepeval import assert_test
from deepeval.metrics import GEval, TaskCompletionMetric
from deepeval.test_case import LLMTestCase, SingleTurnParams
correctness_metric = GEval(
name="الدقة والامتثال للمخطط",
criteria="هل تجيب استجابة LLM على السؤال بدقة، وهل تمتثل تمامًا لتنسيق JSON المطلوب؟",
evaluation_params=[SingleTurnParams.ACTUAL_OUTPUT, SingleTurnParams.EXPECTED_OUTPUT],
threshold=0.7
)
@pytest.mark.parametrize(
"user_input, expected_output",
[
("ملخص بيانات مبيعات الربع الأول من عام 2024.", "إجمالي مبيعات الربع الأول هو 5 مليارات وون."),
("أخبرني بلائحة حساب مكافأة نهاية الخدمة.", "مكافأة نهاية الخدمة هي متوسط أجر 30 يومًا أو أكثر عن كل سنة خدمة مستمرة.")
]
)
def test_agent_regression(user_input, expected_output):
actual_output = run_in_house_agent(user_input)
test_case = LLMTestCase(
input=user_input,
actual_output=actual_output,
expected_output=expected_output
)
# إيقاف البناء إذا كانت النتيجة أقل من المعيار المحدد
assert_test(test_case, [correctness_metric, TaskCompletionMetric(threshold=0.8)])
`
يتكامل نظام التقييم في ثلاث مراحل:
deepeval test run تلقائيًا عند رفع طلب سحب (PR) بعد تعديل الكود.`yaml
name: AI Agent Evaluation Gate
on:
pull_request:
branches: [ main ]
jobs:
eval-gate:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Install Dependencies
run: |
pip install poetry
poetry install
- name: Run DeepEval Suite
env:
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
run: |
poetry run deepeval test run tests/test_evals.py
`
من أجل منع تسرب البيانات الداخلية للشركة، هناك حاجة إلى درابزين حماية (Guardrails) لإلغاء تحديد الهوية الشخصية (PII) قبل خروج طلب الـ API إلى الخارج. تتيح إضافة محرك Microsoft Presidio إلى البوابة إخفاء أرقام السجل السكاني، والبريد الإلكتروني، وأرقام الهواتف، وأرقام الموظفين تلقائيًا.
`python
from presidio_analyzer import AnalyzerEngine, PatternRecognizer
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig
analyzer = AnalyzerEngine()
employee_id_recognizer = PatternRecognizer(
supported_entity="EMPLOYEE_ID",
regex="EMP-[0-9]{6}",
score=0.95
)
analyzer.registry.add_recognizer(employee_id_recognizer)
anonymizer = AnonymizerEngine()
def sanitize_user_prompt(raw_prompt: str) -> str:
results = analyzer.analyze(
text=raw_prompt,
entities=["PERSON", "PHONE_NUMBER", "EMAIL_ADDRESS", "EMPLOYEE_ID"],
language="en"
)
anonymized_result = anonymizer.anonymize(
text=raw_prompt,
analyzer_results=results,
operators={
"DEFAULT": OperatorConfig("replace", {"new_value": "<REDACTED>"}),
"EMPLOYEE_ID": OperatorConfig("mask", {"chars_to_mask": 6, "masking_char": "*", "from_end": True})
}
)
return anonymized_result.text
`
عند ربط بحث RAG أيضًا، يجب اتخاذ تدابير لجلب المستندات التي تتوافق فقط مع مستوى صلاحيات المستخدم. تتعامل قواعد بيانات المتجهات مثل Qdrant أو Pinecone مع فصل الصلاحيات باستخدام تصفية البيانات البيانات الوصفية (Metadata Filtering) أثناء إجراء بحث التشابه.
`python
from qdrant_client import QdrantClient
from qdrant_client.http import models
client = QdrantClient(host="localhost", port=6333)
def search_documents_with_rbac(query_vector: list, user_department: str, user_clearance_level: int):
search_result = client.search(
collection_name="enterprise_knowledge_base",
query_vector=query_vector,
query_filter=models.Filter(
must=[
models.FieldCondition(
key="department",
match=models.MatchValue(value=user_department)
),
models.FieldCondition(
key="security_level",
range=models.Range(lte=user_clearance_level)
)
]
),
limit=5
)
return search_result
`
إن المبدأ الأساسي للوقاية من حقن المطالبات (Prompt Injection) هو الفصل التام بين قناة مطالبة النظام وقناة إدخال المستخدم. ولضمان الأمان، يجب أن تخضع المهام الخطيرة مثل تعديل قاعدة البيانات أو استدعاء APIs خارجية لإجراءات الموافقة البشرية (Human-in-the-Loop) أو تقييد صلاحيات التنفيذ داخل بيئة معزولة (Sandbox).
لاستبدال وحدات SaaS والاستمرار في تشغيلها كنظام داخلي، هناك حاجة إلى روتين فحص دوري.
| الدورية | عنصر فحص التشغيل | التفاصيل |
|---|---|---|
| يوميًا | معدل الأخطاء واستخدام الرموز | التحقق من معدل أخطاء HTTP 5xx واستهلاك الرموز حسب القسم في لوحة قيادة Langfuse |
| يوميًا | سجل حظر قاطع الدائرة | جمع اسم الأداة وأنماط الوسائط للجلسات المقطوعة بسبب اكتشاف الحلقة ثم إصلاحها |
| أسبوعيًا | استخراج استعلامات RAG الفاشلة | تصفية الطلبات ذات درجة التشابه أقل من 0.6 وإضافتها إلى مجموعة الإجابات الصحيحة للاختبار |
| أسبوعيًا | التحقق من الإيجابيات الكاذبة لقناع PII | أخذ عينات من سجلات معالجة Presidio للتحقق من وجود أي نقص في القناع |
| شهريًا | تحديث معايير تقييم CI/CD | تعديل حالات الاختبار الآلي لتعكس تغييرات الأعمال |
إذا كنت تفكر بين تقديم النموذج ذاتيًا (بناءً على vLLM) والاشتراك في الـ API الخارجي، يمكنك حساب ذلك بناءً على حركة المرور اليومية.
تقتطع تكلفة الاستخدام الشهري لمثيل AWS EC2 g5.2xlarge المزود ببطاقة GPU واحدة من نوع NVIDIA A10G حوالي $880. في المقابل، تبلغ تكلفة gpt-4o API حوالي 10.00 لكل مليون رمز إخراج. إذا كانت حركة المرور اليومية أقل من 50 مليون رمز، فإن طريقة الاشتراك في الـ API تكون أكثر فائدة بالنظر إلى جهود إدارة مهندسي DevOps والتكاليف الثابتة لـ GPU. فقط عندما تتجاوز حركة المرور اليومية 100 مليون رمز أو عندما يكون العزل التام داخل الشبكة الداخلية أمرًا ضروريًا، يكون الانتقال إلى التقديم الذاتي القائم على vLLM هو الخيار الصحيح.
للتحضير لأعطال النموذج الرئيسي، نقوم بإنشاء هيكلية توجيه احتياطية ذات 4 مستويات.
┌─────────────────────────────────────────────────────────────────────────┐ │ 4-Tier Graceful Degradation Architecture │ │ │ │ [Tier 1] Primary High-Performance Model (e.g., GPT-4o) │ │ │ │ │ ▼ (API Failure / Timeout / Circuit Breaker) │ │ [Tier 2] Lightweight Routing Model (e.g., GPT-4o-mini / On-Prem vLLM) │ │ │ │ │ ▼ (Continuous Outage) │ │ [Tier 3] Deterministic Regex & SQL Rule Engine │ │ │ │ │ ▼ (Unrecoverable Error) │ │ [Tier 4] Static Error Message & Async Admin Ticket Generation │ └─────────────────────────────────────────────────────────────────────────┘
تتم عملية تحسين نظام الوكيل المخصص على فترات زمنية مدتها 90 يومًا.
| الفترة | هدف التطبيق | تفاصيل العمل |
|---|---|---|
| الأيام 1~30 | الرؤية وحماية البيانات | تثبيت محرك قناع Presidio وزراعة SDK لتسجيل Langfuse |
| الأيام 31~60 | حظر الحلقات وتقليل التكاليف | تطبيق التخزين المؤقت الدلالي لـ Redis وربط قاطع الدائرة المعتمد على Python |
| الأيام 61~90 | أتمتة التحقق والتحكم في الصلاحيات | إدارة المطالبات عبر Git، وربط DeepEval CI/CD، وتطبيق بيانات Qdrant RBAC الوصفية |
في الشهر الأول، ننشئ حل ملاحظة لمنع تسرب المعلومات الشخصية ولتأكيد تسجيل جميع الطلبات. في الشهر الثاني، نربط التخزين المؤقت للمطالبات المكررة وموجه حظر الحلقات لقطع النفقات غير المتوقعة. وفي الشهر الأخير، عند اكتمال إدارة إصدارات المطالبات ونظام التقييم والاختبار الآلي، يصبح التشغيل المستقر متاحًا دون القلق بشأن خلل الوكيل الذي يحدث مع كل عملية نشر.