تصحيح أخطاء الوكلاء

تصحيح أخطاء الوكلاء صعب بشكل فريد بسبب السلوك غير الحتمي وسلاسل الأدوات المعقدة. إليك استراتيجيات مثبتة لإيجاد وإصلاح المشاكل.

التسجيل الشامل

سجل كل ما يفعله الوكيل:

import logging
from datetime import datetime

class AgentLogger:
    def __init__(self, agent_id):
        self.agent_id = agent_id
        self.logger = logging.getLogger(f"agent.{agent_id}")
        self.trace = []

    def log_step(self, step_type, data):
        entry = {
            "timestamp": datetime.now().isoformat(),
            "agent_id": self.agent_id,
            "step_type": step_type,
            "data": data
        }
        self.trace.append(entry)
        self.logger.info(f"[{step_type}] {data}")

    def log_llm_call(self, prompt, response, tokens_used):
        self.log_step("llm_call", {
            "prompt_preview": prompt[:500],
            "response_preview": response[:500],
            "tokens": tokens_used
        })

    def log_tool_call(self, tool_name, params, result, duration_ms):
        self.log_step("tool_call", {
            "tool": tool_name,
            "params": params,
            "result": str(result)[:500],
            "duration_ms": duration_ms
        })

    def log_error(self, error_type, details, stack_trace):
        self.log_step("error", {
            "type": error_type,
            "details": details,
            "stack": stack_trace
        })

    def get_trace(self):
        return self.trace

تصور التتبع

إنشاء آثار تنفيذ قابلة للقراءة:

def visualize_trace(trace):
    """توليد أثر تنفيذ مقروء للبشر"""
    output = []
    indent = 0

    for entry in trace:
        step_type = entry["step_type"]
        data = entry["data"]

        if step_type == "llm_call":
            output.append(f"{'  ' * indent}🤖 استدعاء LLM ({data['tokens']} رمز)")
            output.append(f"{'  ' * indent}   الإدخال: {data['prompt_preview'][:100]}...")
            output.append(f"{'  ' * indent}   الإخراج: {data['response_preview'][:100]}...")

        elif step_type == "tool_call":
            output.append(f"{'  ' * indent}🔧 أداة: {data['tool']} ({data['duration_ms']}مللي ثانية)")
            output.append(f"{'  ' * indent}   المعلمات: {data['params']}")
            output.append(f"{'  ' * indent}   النتيجة: {data['result'][:100]}...")

        elif step_type == "error":
            output.append(f"{'  ' * indent}❌ خطأ: {data['type']}")
            output.append(f"{'  ' * indent}   {data['details']}")

    return "\n".join(output)

الإعادة والإعادة الإنتاجية

حفظ الجلسات للإعادة:

class SessionRecorder:
    def __init__(self, session_id):
        self.session_id = session_id
        self.events = []

    def record(self, event_type, data):
        self.events.append({
            "event_type": event_type,
            "data": data,
            "timestamp": datetime.now().isoformat()
        })

    def save(self, path):
        with open(path, 'w') as f:
            json.dump({
                "session_id": self.session_id,
                "events": self.events
            }, f)

    @classmethod
    def replay(cls, path, agent):
        """إعادة تشغيل جلسة مسجلة"""
        with open(path) as f:
            session = json.load(f)

        for event in session["events"]:
            if event["event_type"] == "user_input":
                response = agent.process(event["data"]["message"])
                print(f"الأصلي: {event['data'].get('original_response')}")
                print(f"المُعاد: {response}")

سيناريوهات التصحيح الشائعة

1. حلقات الوكيل

def diagnose_loop(trace):
    """كشف الإجراءات المتكررة في التتبع"""
    actions = [e["data"]["tool"] for e in trace if e["step_type"] == "tool_call"]

    # إيجاد الأنماط المتكررة
    for window_size in range(2, len(actions) // 2):
        for i in range(len(actions) - window_size * 2):
            window1 = actions[i:i + window_size]
            window2 = actions[i + window_size:i + window_size * 2]
            if window1 == window2:
                return f"تم كشف حلقة: {window1} متكررة في الموضع {i}"

    return None

2. فشل الأدوات

def analyze_tool_failures(trace):
    """تلخيص أنماط فشل الأدوات"""
    failures = {}

    for entry in trace:
        if entry["step_type"] == "error" and "tool" in entry["data"]:
            tool = entry["data"]["tool"]
            error = entry["data"]["type"]

            if tool not in failures:
                failures[tool] = {}
            if error not in failures[tool]:
                failures[tool][error] = 0
            failures[tool][error] += 1

    return failures

3. تجاوز السياق

def check_context_usage(trace):
    """تتبع استخدام الرموز عبر الزمن"""
    total_tokens = 0
    max_tokens = 128000  # التعديل حسب النموذج

    for entry in trace:
        if entry["step_type"] == "llm_call":
            total_tokens = entry["data"]["tokens"]

            if total_tokens > max_tokens * 0.9:
                return {
                    "warning": "السياق ممتلئ تقريباً",
                    "usage": f"{total_tokens}/{max_tokens}",
                    "timestamp": entry["timestamp"]
                }

    return {"status": "ok", "final_usage": total_tokens}

تكامل أدوات التصحيح

# تكامل LangSmith لـ LangChain
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "my-agent-debug"

# Phoenix للمراقبة
from phoenix.trace import langchain as phoenix_langchain
phoenix_langchain.instrument()

قائمة فحص التصحيح

الفحص	الأداة	متى
أثر التنفيذ	المُسجل	كل تشغيل
استخدام الرموز	عداد الرموز	قبل النشر
معدل نجاح الأدوات	محلل الفشل	أسبوعياً
جودة الاستجابة	مجموعة التقييم	بعد التغييرات
التأخير	مراقب الأداء	مستمر

أفضل الممارسات

• سجل المطالبات والاستجابات (مع حدود الحجم)
• سجل جميع استدعاءات الأدوات مع التوقيت
• مكّن الإعادة لمشاكل الإنتاج
• أعد التنبيهات لارتفاعات الأخطاء
• استخدم منصات المراقبة (LangSmith، Phoenix، Arize)

اختبر معرفتك بمعالجة الأخطاء في اختبار الوحدة! :::