كيمي-K2 في العمل: تطوير الوكلاء واستكشاف سيناريوهات التطبيق

المقدمة

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

مزايا وكيل كيمي-K2

1. قدرات استدعاء أدوات قوية

تم تحسين كيمي-K2 بشكل خاص لوظيفة استدعاء الأدوات أثناء التصميم، مما يمكّنه من فهم أوصاف الأدوات المعقدة وإجراء استدعاءات دقيقة:

import json
import torch
from transformers import AutoTokenizer, AutoModelForCausalLM

# تعريف وظائف الأدوات
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "الحصول على معلومات الطقس لمدينة محددة",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "اسم المدينة"
                    }
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function", 
        "function": {
            "name": "calculate",
            "description": "إجراء حسابات رياضية",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "التعبير الرياضي"
                    }
                },
                "required": ["expression"]
            }
        }
    }
]

# مثال على محادثة الوكيل
def chat_with_tools(model, tokenizer, user_input, tools):
    messages = [
        {"role": "system", "content": "أنت مساعد ذكي يمكنه استدعاء الأدوات لمساعدة المستخدمين في حل المشكلات."},
        {"role": "user", "content": user_input}
    ]
    
    # إضافة أوصاف الأدوات
    tool_prompt = f"الأدوات المتاحة: {json.dumps(tools, ensure_ascii=False, indent=2)}"
    messages[0]["content"] += f"\n\n{tool_prompt}"
    
    # توليد الاستجابة
    text = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
    inputs = tokenizer(text, return_tensors="pt")
    
    with torch.no_grad():
        outputs = model.generate(
            **inputs,
            max_new_tokens=512,
            temperature=0.7,
            do_sample=True
        )
    
    response = tokenizer.decode(outputs[0][inputs.input_ids.shape[-1]:], skip_special_tokens=True)
    return response

# مثال على الاستخدام
user_query = "كيف يكون الطقس في بكين اليوم؟ إذا كانت احتمالية المطر تتجاوز 70%، ساعدني في حساب كم سيكون أجرة التاكسي أغلى (عادة 15 يوان، زيادة 30% في السعر في الأيام الممطرة)"
response = chat_with_tools(model, tokenizer, user_query, tools)

2. ذاكرة سياق طويلة للغاية

تمكن طول السياق 128K كيمي-K2 من الحفاظ على تاريخ المحادثة على المدى الطويل:

class LongContextAgent:
    def __init__(self, model, tokenizer, max_context_length=128000):
        self.model = model
        self.tokenizer = tokenizer
        self.conversation_history = []
        self.max_context_length = max_context_length
    
    def add_message(self, role, content):
        self.conversation_history.append({"role": role, "content": content})
        self._trim_context()
    
    def _trim_context(self):
        # الحفاظ على حد طول السياق
        total_tokens = 0
        trimmed_history = []
        
        for message in reversed(self.conversation_history):
            message_tokens = len(self.tokenizer.encode(message["content"]))
            if total_tokens + message_tokens > self.max_context_length:
                break
            trimmed_history.insert(0, message)
            total_tokens += message_tokens
        
        self.conversation_history = trimmed_history
    
    def generate_response(self, user_input):
        self.add_message("user", user_input)
        
        # بناء تاريخ المحادثة الكامل
        text = self.tokenizer.apply_chat_template(
            self.conversation_history, 
            tokenize=False, 
            add_generation_prompt=True
        )
        
        inputs = self.tokenizer(text, return_tensors="pt")
        
        with torch.no_grad():
            outputs = self.model.generate(
                **inputs,
                max_new_tokens=512,
                temperature=0.6,
                do_sample=True
            )
        
        response = self.tokenizer.decode(
            outputs[0][inputs.input_ids.shape[-1]:], 
            skip_special_tokens=True
        )
        
        self.add_message("assistant", response)
        return response

3. مزايا التعاون بين الخبراء المتعددين

تمكن بنية MoE أنواعًا مختلفة من المهام من استدعاء الخبراء الأكثر ملاءمة:

class MultiExpertAgent:
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer
        self.task_types = {
            "coding": "البرمجة والمهام المتعلقة بالشيفرة",
            "math": "الحسابات الرياضية والتفكير المنطقي",
            "writing": "كتابة النصوص وتحريرها",
            "analysis": "تحليل البيانات وتلخيصها"
        }
    
    def classify_task(self, user_input):
        """منطق تصنيف المهام البسيط"""
        if any(keyword in user_input.lower() for keyword in ["code", "programming", "program", "algorithm"]):
            return "coding"
        elif any(keyword in user_input.lower() for keyword in ["calculate", "math", "formula", "reasoning"]):
            return "math"
        elif any(keyword in user_input.lower() for keyword in ["write", "article", "summary", "report"]):
            return "writing"
        elif any(keyword in user_input.lower() for keyword in ["analyze", "statistics", "data", "chart"]):
            return "analysis"
        else:
            return "general"
    
    def generate_specialized_response(self, user_input, task_type):
        system_prompts = {
            "coding": "أنت مساعد برمجة محترف، بارع في لغات البرمجة واللوغاريتمات المتعددة.",
            "math": "أنت خبير رياضيات، ماهر في حل المشكلات الرياضية المعقدة والتفكير المنطقي.",
            "writing": "أنت مساعد كتابة محترف، قادر على إنشاء وتحرير أنواع مختلفة من النصوص.",
            "analysis": "أنت خبير تحليل بيانات، ماهر في استخراج الرؤى والاتجاهات من البيانات.",
            "general": "أنت مساعد ذكاء اصطناعي متعدد الاستخدامات، قادر على التعامل مع أنواع مختلفة من المشكلات."
        }
        
        messages = [
            {"role": "system", "content": system_prompts.get(task_type, system_prompts["general"])},
            {"role": "user", "content": user_input}
        ]
        
        # منطق توليد الاستجابة...
        return self._generate_response(messages)

حالات التطبيق في العالم الحقيقي

الحالة 1: مساعد خدمة العملاء الذكي

class CustomerServiceAgent:
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer
        self.knowledge_base = {
            "refund_policy": "دعم الإرجاع بدون سبب لمدة 7 أيام، يجب أن تظل المنتجات في حالة جيدة...",
            "shipping_info": "عادة ما يتم التسليم في غضون 1-3 أيام عمل، التسليم في اليوم التالي متاح...",
            "product_warranty": "تأتي المنتجات الإلكترونية مع ضمان لمدة عام، قابل للتجديد لمدة 3 سنوات..."
        }
        self.conversation_state = {}
    
    def handle_query(self, user_id, query):
        # استرجاع المعرفة ذات الصلة
        relevant_info = self.search_knowledge(query)
        
        # بناء السياق
        context = f"المعلومات ذات الصلة: {relevant_info}\nسؤال المستخدم: {query}"
        
        messages = [
            {"role": "system", "content": "أنت مساعد خدمة عملاء محترف، تقدم إجابات مهذبة ودقيقة على أسئلة المستخدمين."},
            {"role": "user", "content": context}
        ]
        
        response = self._generate_response(messages)
        
        # تحديث حالة المحادثة
        self.conversation_state[user_id] = {
            "last_query": query,
            "last_response": response,
            "context": relevant_info
        }
        
        return response
    
    def search_knowledge(self, query):
        # منطق استرجاع المعرفة البسيط
        for key, value in self.knowledge_base.items():
            if any(keyword in query for keyword in key.split()):
                return value
        return "لم يتم العثور على معلومات ذات صلة، يرجى الاتصال بخدمة العملاء البشرية."

الحالة 2: مساعد مراجعة الشيفرة

class CodeReviewAgent:
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer
        self.review_criteria = [
            "صحة منطق الشيفرة",
            "اقتراحات تحسين الأداء", 
            "فحوصات الأمان",
            "أسلوب الشيفرة والمعايير",
            "آليات التعامل مع الأخطاء"
        ]
    
    def review_code(self, code, language="python"):
        # بناء نص معايير المراجعة
        criteria_text = "\n".join([f"- {criterion}" for criterion in self.review_criteria])
        
        # إنشاء مطالبة المراجعة
        review_prompt = f"يرجى إجراء مراجعة شاملة لشيفرة {language}"
        
        messages = [
            {"role": "system", "content": "أنت خبير مراجعة شيفرة كبير، قادر على تحديد مشكلات الشيفرة وتقديم اقتراحات احترافية."},
            {"role": "user", "content": review_prompt}
        ]
        
        return self._generate_response(messages)
    
    def suggest_improvements(self, code, issues):
        # إنشاء مطالبة التحسين
        improvement_prompt = "استنادًا إلى مشكلات مراجعة الشيفرة، يرجى تقديم شيفرة محسّنة"
        
        messages = [
            {"role": "system", "content": "يرجى تقديم شيفرة محسّنة وشرح أسباب التعديلات."},
            {"role": "user", "content": improvement_prompt}
        ]
        
        return self._generate_response(messages)

الحالة 3: مساعد التعليم

class EducationAgent:
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer
        self.student_progress = {}
    
    def adaptive_tutoring(self, student_id, subject, question, difficulty="medium"):
        # الحصول على أداء الطالب التاريخي
        progress = self.student_progress.get(student_id, {"correct": 0, "total": 0})
        success_rate = progress["correct"] / max(progress["total"], 1)
        
        # تعديل استراتيجية التعليم بناءً على معدل النجاح
        if success_rate > 0.8:
            teaching_style = "يمكن تجربة محتوى صعب أعلى، تقديم مسائل تحدي"
        elif success_rate > 0.6:
            teaching_style = "الحفاظ على الصعوبة الحالية، تقديم شروحات مفصلة"
        else:
            teaching_style = "يحتاج إلى شروحات أساسية أكثر، توجيه خطوة بخطوة"
        
        prompt = f"""
سؤال الطالب: {question}
الموضوع: {subject}
مستوى الصعوبة: {difficulty}
استراتيجية التعليم: {teaching_style}
معدل نجاح الطالب: {success_rate:.2%}

يرجى التصرف كمعلم محترف للإجابة على السؤال، باستخدام أساليب تعليم مناسبة.
"""
        
        messages = [
            {"role": "system", "content": "أنت معلم ذو خبرة، جيد في تخصيص التعليم لاحتياجات الأفراد وتعديل أساليب التعليم بناءً على مستويات الطلاب."},
            {"role": "user", "content": prompt}
        ]
        
        response = self._generate_response(messages)
        return response
    
    def generate_practice_problems(self, subject, topic, difficulty, count=3):
        # إنشاء مطالبة مسائل الممارسة
        prompt = f"يرجى إنشاء مسائل ممارسة حول {topic} في {subject}"
        
        messages = [
            {"role": "system", "content": "أنت مصمم مسائل محترف، قادر على إنشاء مسائل ممارسة عالية الجودة."},
            {"role": "user", "content": prompt}
        ]
        
        return self._generate_response(messages)

نصائح تحسين الأداء

1. استراتيجية التخزين المؤقت الذكي

import hashlib
import pickle
from functools import lru_cache

class CachedAgent:
    def __init__(self, model, tokenizer, cache_size=1000):
        self.model = model
        self.tokenizer = tokenizer
        self.response_cache = {}
        self.cache_size = cache_size
    
    def _hash_input(self, messages):
        # توليد هاش للإدخال
        content = str(messages)
        return hashlib.md5(content.encode()).hexdigest()
    
    def generate_with_cache(self, messages):
        cache_key = self._hash_input(messages)
        
        if cache_key in self.response_cache:
            return self.response_cache[cache_key]
        
        response = self._generate_response(messages)
        
        # إدارة التخزين المؤقت
        if len(self.response_cache) >= self.cache_size:
            # حذف أقدم إدخال في التخزين المؤقت
            oldest_key = next(iter(self.response_cache))
            del self.response_cache[oldest_key]
        
        self.response_cache[cache_key] = response
        return response

2. المعالجة غير المتزامنة

import asyncio
import aiohttp

class AsyncAgent:
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer
        self.session = None
    
    async def process_multiple_queries(self, queries):
        tasks = []
        for query in queries:
            task = asyncio.create_task(self.process_single_query(query))
            tasks.append(task)
        
        results = await asyncio.gather(*tasks)
        return results
    
    async def process_single_query(self, query):
        # محاكاة المعالجة غير المتزامنة
        await asyncio.sleep(0.1)  # تجنب الحظر
        
        messages = [
            {"role": "user", "content": query}
        ]
        
        return self._generate_response(messages)
    
    async def external_api_call(self, url, data):
        if not self.session:
            self.session = aiohttp.ClientSession()
        
        async with self.session.post(url, json=data) as response:
            return await response.json()

توصيات الممارسات الجيدة

1. إدارة حالة المحادثة

from enum import Enum
from dataclasses import dataclass
from typing import Dict, List, Optional

class ConversationState(Enum):
    GREETING = "greeting"
    COLLECTING_INFO = "collecting_info"
    PROCESSING = "processing"
    CLARIFYING = "clarifying"
    COMPLETED = "completed"

@dataclass
class UserContext:
    user_id: str
    state: ConversationState
    collected_info: Dict
    preferences: Dict
    history: List[Dict]

class StatefulAgent:
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer
        self.user_contexts = {}
    
    def get_or_create_context(self, user_id):
        if user_id not in self.user_contexts:
            self.user_contexts[user_id] = UserContext(
                user_id=user_id,
                state=ConversationState.GREETING,
                collected_info={},
                preferences={},
                history=[]
            )
        return self.user_contexts[user_id]
    
    def handle_message(self, user_id, message):
        context = self.get_or_create_context(user_id)
        
        # التعامل مع الرسالة بناءً على الحالة الحالية
        if context.state == ConversationState.GREETING:
            return self.handle_greeting(context, message)
        elif context.state == ConversationState.COLLECTING_INFO:
            return self.handle_info_collection(context, message)
        # معالجة الحالات الأخرى...
        
        return self.generate_default_response(context, message)

2. التعامل مع الأخطاء والتدهور اللائق

class RobustAgent:
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer
        self.fallback_responses = {
            "generation_failed": "عذرًا، لا أستطيع توليد استجابة في الوقت الحالي، يرجى المحاولة لاحقًا.",
            "context_too_long": "تاريخ المحادثة طويل جدًا، دعنا نبدأ من جديد.",
            "tool_call_failed": "فشل استدعاء الأداة، سأجيب باستخدام طرق بديلة."
        }
    
    def safe_generate(self, messages, max_retries=3):
        for attempt in range(max_retries):
            try:
                return self._generate_response(messages)
            except torch.cuda.OutOfMemoryError:
                torch.cuda.empty_cache()
                # تقليل طول الإدخال
                messages = self._trim_messages(messages)
            except Exception as e:
                if attempt == max_retries - 1:
                    return self.fallback_responses["generation_failed"]
                continue
        
        return self.fallback_responses["generation_failed"]
    
    def _trim_messages(self, messages, max_length=4096):
        # الحفاظ على الرسائل النظامية وأحدث رسائل المستخدم
        system_msgs = [msg for msg in messages if msg["role"] == "system"]
        user_msgs = [msg for msg in messages if msg["role"] == "user"]
        
        if user_msgs:
            return system_msgs + [user_msgs[-1]]
        return system_msgs

النشر والمراقبة

1. مراقبة الأداء

import time
import logging
from dataclasses import dataclass
from typing import Dict

@dataclass
class AgentMetrics:
    total_requests: int = 0
    successful_requests: int = 0
    failed_requests: int = 0
    average_response_time: float = 0.0
    peak_memory_usage: float = 0.0

class MonitoredAgent:
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer
        self.metrics = AgentMetrics()
        self.logger = logging.getLogger(__name__)
    
    def generate_with_monitoring(self, messages):
        start_time = time.time()
        self.metrics.total_requests += 1
        
        try:
            response = self._generate_response(messages)
            self.metrics.successful_requests += 1
            
            # تحديث متوسط وقت الاستجابة
            elapsed = time.time() - start_time
            self.metrics.average_response_time = (
                (self.metrics.average_response_time * (self.metrics.successful_requests - 1) + elapsed) 
                / self.metrics.successful_requests
            )
            
            self.logger.info(f"اكتمل الطلب في {elapsed:.2f}s")
            return response
            
        except Exception as e:
            self.metrics.failed_requests += 1
            self.logger.error(f"فشل الطلب: {str(e)}")
            raise
    
    def get_metrics_summary(self):
        success_rate = (
            self.metrics.successful_requests / max(self.metrics.total_requests, 1) * 100
        )
        
        return {
            "total_requests": self.metrics.total_requests,
            "success_rate": f"{success_rate:.2f}%",
            "average_response_time": f"{self.metrics.average_response_time:.2f}s",
            "failure_count": self.metrics.failed_requests
        }

الخاتمة

يوفر كيمي-K2 أساسًا تقنيًا قويًا لتطوير الوكلاء. تمكّن قدرات التخصص الخبيري لبنية MoE، وذاكرة السياق الطويلة للغاية، ووظيفة استدعاء الأدوات الممتازة المطورين من بناء تطبيقات ذكية وعملية للغاية.

من خلال الحالات والممارسات الجيدة في هذه المقالة، يمكن للمطورين:

• الاستفادة من قدرات استدعاء الأدوات لبناء وكلاء غنيين بالميزات
• تنفيذ حوار مستمر من خلال ذاكرة السياق الطويلة
• التعامل مع المهام المعقدة باستخدام التعاون بين الخبراء المتعددين
• اعتماد الممارسات الجيدة لضمان استقرار النظام وموثوقيته

مع استمرار تطور التكنولوجيا، سيستمر كيمي-K2 في دفع الابتكار في تطبيقات الوكلاء، مما يفتح المزيد من الاحتمالات لمختلف الصناعات.