تكامل LiteLLM

يوفر LiteLLM واجهة موحدة لأكثر من 100 مزود LLM. بالجمع مع Guardrails AI، تحصل على مخرجات متحققة عبر أي نموذج مع احتياطي تلقائي وموازنة الحمل.

إعداد LiteLLM

pip install litellm

import litellm

# تكوين مفاتيح API
import os
os.environ["OPENAI_API_KEY"] = "sk-..."
os.environ["ANTHROPIC_API_KEY"] = "sk-ant-..."
os.environ["AZURE_API_KEY"] = "..."
os.environ["AZURE_API_BASE"] = "https://your-resource.openai.azure.com"

LiteLLM الأساسي مع Guardrails

from guardrails import Guard
from pydantic import BaseModel, Field
import litellm

class AnalysisResult(BaseModel):
    summary: str = Field(description="ملخص التحليل")
    key_findings: list[str] = Field(description="النتائج الرئيسية")
    confidence: float = Field(ge=0, le=1, description="درجة الثقة")

guard = Guard.for_pydantic(AnalysisResult)

# استخدم أي نموذج مدعوم من LiteLLM
result = guard(
    model="gpt-4o",  # OpenAI
    messages=[{"role": "user", "content": "حلل تقرير أرباح Q3"}]
)

# تبديل المزودين بسهولة
result = guard(
    model="claude-3-5-sonnet-20241022",  # Anthropic
    messages=[{"role": "user", "content": "حلل تقرير أرباح Q3"}]
)

result = guard(
    model="azure/gpt-4o",  # Azure OpenAI
    messages=[{"role": "user", "content": "حلل تقرير أرباح Q3"}]
)

تكوين الاحتياطي

from guardrails import Guard
import litellm
from litellm import Router

# تكوين سلسلة الاحتياطي
router = Router(
    model_list=[
        {
            "model_name": "gpt-4",
            "litellm_params": {
                "model": "gpt-4o",
                "api_key": os.environ["OPENAI_API_KEY"]
            }
        },
        {
            "model_name": "gpt-4",
            "litellm_params": {
                "model": "azure/gpt-4o",
                "api_key": os.environ["AZURE_API_KEY"],
                "api_base": os.environ["AZURE_API_BASE"]
            }
        },
        {
            "model_name": "gpt-4",
            "litellm_params": {
                "model": "claude-3-5-sonnet-20241022",
                "api_key": os.environ["ANTHROPIC_API_KEY"]
            }
        }
    ],
    fallbacks=[
        {"gpt-4": ["azure/gpt-4o", "claude-3-5-sonnet-20241022"]}
    ],
    retry_after=5  # إعادة المحاولة بعد 5 ثواني
)

guard = Guard.for_pydantic(AnalysisResult)

# استخدم الموجه للاحتياطي التلقائي
async def analyze_with_fallback(content: str):
    try:
        response = await router.acompletion(
            model="gpt-4",
            messages=[{"role": "user", "content": content}]
        )
        return guard.parse(response.choices[0].message.content)
    except Exception as e:
        # الموجه يحاول الاحتياطيات تلقائياً
        raise

التوجيه المحسن للتكلفة

from litellm import Router
from guardrails import Guard

# التوجيه بناءً على التعقيد
router = Router(
    model_list=[
        # نماذج رخيصة للمهام البسيطة
        {
            "model_name": "simple",
            "litellm_params": {
                "model": "gpt-4o-mini",
                "api_key": os.environ["OPENAI_API_KEY"]
            }
        },
        # نماذج مكلفة للمهام المعقدة
        {
            "model_name": "complex",
            "litellm_params": {
                "model": "gpt-4o",
                "api_key": os.environ["OPENAI_API_KEY"]
            }
        }
    ]
)

class TaskRouter:
    def __init__(self, guard: Guard):
        self.guard = guard
        self.router = router

    async def route(self, task: str, complexity: str = "auto") -> dict:
        """توجيه المهمة للنموذج المناسب بناءً على التعقيد."""
        if complexity == "auto":
            complexity = self._estimate_complexity(task)

        model = "simple" if complexity == "low" else "complex"

        result = self.guard(
            model=self.router.model_list[0 if model == "simple" else 1]["litellm_params"]["model"],
            messages=[{"role": "user", "content": task}]
        )

        return {
            "result": result.validated_output,
            "model_used": model,
            "estimated_cost": self._estimate_cost(model, task)
        }

    def _estimate_complexity(self, task: str) -> str:
        """تقدير التعقيد البسيط."""
        complex_keywords = ["analyze", "compare", "synthesize", "evaluate"]
        return "high" if any(kw in task.lower() for kw in complex_keywords) else "low"

    def _estimate_cost(self, model: str, task: str) -> float:
        """تقدير التكلفة بناءً على النموذج وطول المدخل."""
        costs = {"simple": 0.00015, "complex": 0.005}  # لكل 1K رمز
        tokens = len(task.split()) * 1.3  # تقدير تقريبي للرموز
        return costs[model] * tokens / 1000

تحديد المعدل والحصص

from litellm import Router
import asyncio
from datetime import datetime, timedelta

class RateLimitedRouter:
    """موجه مع تحديد المعدل لكل نموذج."""

    def __init__(self, guard: Guard):
        self.guard = guard
        self.request_counts = {}
        self.rate_limits = {
            "gpt-4o": 60,        # 60 طلب في الدقيقة
            "gpt-4o-mini": 200,  # 200 طلب في الدقيقة
            "claude-3-5-sonnet-20241022": 40   # 40 طلب في الدقيقة
        }
        self.router = Router(
            model_list=[
                {"model_name": "primary", "litellm_params": {"model": "gpt-4o"}},
                {"model_name": "fallback", "litellm_params": {"model": "gpt-4o-mini"}},
            ]
        )

    async def completion(self, model: str, messages: list) -> dict:
        """إكمال مع الوعي بحد المعدل."""
        await self._wait_for_rate_limit(model)

        result = self.guard(
            model=model,
            messages=messages
        )

        self._record_request(model)
        return result

    async def _wait_for_rate_limit(self, model: str):
        """انتظار إذا كان سيتم تجاوز حد المعدل."""
        limit = self.rate_limits.get(model, 60)
        current = self._get_request_count(model)

        if current >= limit:
            wait_time = 60 - (datetime.now().second)
            await asyncio.sleep(wait_time)

    def _record_request(self, model: str):
        """تسجيل طلب لتحديد المعدل."""
        now = datetime.now()
        minute_key = now.strftime("%Y%m%d%H%M")

        if model not in self.request_counts:
            self.request_counts[model] = {}

        self.request_counts[model][minute_key] = \
            self.request_counts[model].get(minute_key, 0) + 1

    def _get_request_count(self, model: str) -> int:
        """الحصول على عدد الطلبات للدقيقة الحالية."""
        minute_key = datetime.now().strftime("%Y%m%d%H%M")
        return self.request_counts.get(model, {}).get(minute_key, 0)

البث مع Guardrails

from guardrails import Guard
import litellm

guard = Guard.for_pydantic(AnalysisResult)

async def stream_with_validation(prompt: str):
    """بث الاستجابة والتحقق في النهاية."""
    full_response = ""

    async for chunk in await litellm.acompletion(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        stream=True
    ):
        if chunk.choices[0].delta.content:
            content = chunk.choices[0].delta.content
            full_response += content
            yield {"type": "chunk", "content": content}

    # التحقق من الاستجابة الكاملة
    try:
        validated = guard.parse(full_response)
        yield {"type": "validated", "result": validated}
    except Exception as e:
        yield {"type": "error", "error": str(e)}

مغلف الإنتاج

from guardrails import Guard
from pydantic import BaseModel
import litellm
from typing import TypeVar, Type
import logging

T = TypeVar("T", bound=BaseModel)

class GuardedLLM:
    """مغلف LLM جاهز للإنتاج مع Guardrails."""

    def __init__(
        self,
        primary_model: str = "gpt-4o",
        fallback_model: str = "gpt-4o-mini",
        max_retries: int = 3
    ):
        self.primary_model = primary_model
        self.fallback_model = fallback_model
        self.max_retries = max_retries
        self.logger = logging.getLogger(__name__)

    async def generate(
        self,
        schema: Type[T],
        prompt: str,
        system_prompt: str = None,
        num_reasks: int = 2
    ) -> T:
        """توليد استجابة متحققة."""
        guard = Guard.for_pydantic(schema)

        messages = []
        if system_prompt:
            messages.append({"role": "system", "content": system_prompt})
        messages.append({"role": "user", "content": prompt})

        # جرب النموذج الأساسي
        try:
            result = guard(
                model=self.primary_model,
                messages=messages,
                num_reasks=num_reasks
            )

            if result.validated_output:
                return result.validated_output

        except Exception as e:
            self.logger.warning(f"فشل النموذج الأساسي: {e}")

        # الاحتياطي
        try:
            result = guard(
                model=self.fallback_model,
                messages=messages,
                num_reasks=num_reasks + 1  # إعادة محاولة إضافية للاحتياطي
            )

            if result.validated_output:
                return result.validated_output

        except Exception as e:
            self.logger.error(f"فشل النموذج الاحتياطي: {e}")
            raise

        raise ValueError("فشل في توليد استجابة صالحة")

# الاستخدام
llm = GuardedLLM()

result = await llm.generate(
    schema=AnalysisResult,
    prompt="حلل اتجاهات السوق للربع الرابع",
    system_prompt="أنت محلل مالي."
)

نصيحة LiteLLM: استخدم بادئات متغيرات البيئة للتكوينات الخاصة بالمزود: OPENAI_API_KEY، ANTHROPIC_API_KEY، AZURE_API_KEY، إلخ. يلتقطها LiteLLM تلقائياً.

التالي: أنماط نشر الإنتاج والمراقبة. :::