Written with AI assistance from the source video. The video is the authority.
7:34Chase AI
Log in to leave a comment
No posts yet
كلما صدر نموذج جديد، فإن سبب تأخر اعتماده ليس درجات المقارنة المعيارية (Benchmarks). فالسبب الحقيقي هو البنية القديمة التي تدمج حزمة تطوير نرمجة (SDK) خاصة بمزود معين بعمق في منتصف منطق الأعمال، وفاتورة واجهة برمجة التطبيقات (API) التي لا يمكن تخمين مقدارها تماماً.
وعندما يُضاف إلى ذلك ملاحظة فريق الأمان القائلة: “إذا انتقلت بيانات العملاء إلى منطقة خارجية، فسوف ينتهك ذلك قانون حماية المعلومات الشخصية”، يضطر الموظفون للعمل بظهورهم كل ليلة لتعديل تقارير المراجعة. إن تغيير نقاط النهاية (Endpoints) هو الخطوة الأخيرة. من الآمن استخراج الأرقام من سجلات التشغيل الفعلية للحصول على موافقة الميزانية، وتصفية عمليات الحجب (Prompt Masking) لتجاوز عمليات التدقيق، ثم تبديل التعليمات البرمجية.
إذا قمت بتقديم مسودة بناءً على جدول الأسعار المعلن من قِبل المزود فقط، فستواجه فاتورة ضخمة عند تدفق حركة المرور. يجب عليك تحليل سجلات المكالمات الخاصة بالخادم قيد التشغيل لآخر 30 يوماً للتحقق من استهلاك الرموز (Tokens) الفعلي أولاً.
إذا كنت تستخدم CloudWatch Logs Insights، فيمكنك استخراج استهلاك الرموز لكل طلب والإجمالي اليومي في غضون 10 ثوانٍ باستخدام استعلام AWS CLI.
`bash
#!/usr/bin/env bash
set -euo pipefail
LOG_GROUP_NAME="/aws/backend/legacy-llm-service"
START_TIME=(date +%s)
QUERY_STRING='fields @timestamp, usage.prompt_tokens as p_tok, usage.completion_tokens as c_tok
| filter ispresent(p_tok) and ispresent(c_tok)
| stats count(@timestamp) as total_requests,
sum(p_tok) as sum_prompt_tokens,
sum(c_tok) as sum_completion_tokens,
avg(p_tok) as avg_prompt_tokens,
avg(c_tok) as avg_completion_tokens,
percentile(p_tok + c_tok, 95) as p95_total_tokens'
QUERY_ID=$(aws logs start-query
--log-group-name "$LOG_GROUP_NAME"
--start-time "$START_TIME"
--end-time "$END_TIME"
--query-string "$QUERY_STRING"
--output text --query 'queryId')
sleep 5
aws logs get-query-results --query-id "$QUERY_ID"
--output json | jq -r '.results[] | map({(.field): .value}) | add'
`
نموذج تسعير GPT-6 Astra هو معدل موحد قدره 10 دولارات لكل 10 ملايين رمز (1.00 دولار لكل مليون رمز) دون التمييز بين المدخلات والمخرجات. قم بمحاكاة الفاتورة الشهرية عن طريق ضرب عدد الطلبات اليومية المستخرجة ومتوسط الرموز لكل طلب.
إذا كان نموذج Fable الحالي يتقاضى متوسطاً مرجحاً قدره 5.00 دولارات لكل مليون رمز، فإن السعر ينخفض بنسبة 80% عند اعتماد Astra.
| تصنيف السيناريو | عدد الطلبات اليومية | متوسط الرموز للطلب الواحد | إجمالي الرموز الشهرية | سعر Astra (لكل 1 مليون رمز) | التكلفة الشهرية المتوقعة (دولار أمريكي) | المبلغ المحول للعملة المحلية |
|---|---|---|---|---|---|---|
| متحفظ (Low) | 35,000 | 1,200 | 1,260,000,000 | $1.00 | $1,260.00 | 1,701,000 وون |
| الأساسي (Base) | 50,000 | 1,200 | 1,800,000,000 | $1.00 | $1,800.00 | 2,430,000 وون |
| الذروة (Peak) | 75,000 | 1,500 | 3,375,000,000 | $1.00 | $3,375.00 | 4,556,250 وون |
افتح جدول البيانات، وأدخل عدد الطلبات اليومية (50000) في الخلية B1، ومتوسط رموز الإدخال (800) في B2، ومتوسط رموز الإخراج (400) في B3، وسعر الصرف الأساسي (1350) في B4. من خلال وضع الصيغة =30*B1*(B2+B3) في الخلية B5، و=(B5/10000000)*10 في B6، و=B6*B4 في B7، ستحصل فوراً على جدول الميزانية وفقاً لتغيرات حركة المرور.
عند إرفاق ورقة الحسابات التي تُظهر انخفاض تكلفة معالجة كل 1000 معاملة من 6.00 دولارات إلى 1.20 دولار، تتم موافقة الإدارة في غضون 30 دقيقة.
إذا كنت تستפיد من fable.Client() أو openai.Client() مباشرة داخل رمز الخدمة، فسيتعين عليك فتح عشرات الملفات واحدة تلو الأخرى وتعديل معالجة الاستثناءات في كل مرة تنتقل فيها إلى نموذج جديد. هذا العمل مثالي تماماً للتسبب في تعطل الخدمة.
من خلال تثبيت واجهة باستخدام typing.Protocol في لغة بايثون وتطبيق نمط محول الكائنات (Object Adapter Pattern)، يمكنك تغيير النموذج الخارجي دون المساس بمنطق الأعمال على الإطلاق.
`python
from typing import Protocol, List, Optional
from pydantic import BaseModel, Field
class LLMMessage(BaseModel):
role: str = Field(..., description="system, user, assistant")
content: str
class LLMUsage(BaseModel):
prompt_tokens: int
completion_tokens: int
total_tokens: int
class LLMResponse(BaseModel):
content: str
usage: LLMUsage
model_name: str
provider: str
class LLMClientAdapter(Protocol):
async def generate_completion(
self,
messages: List[LLMMessage],
temperature: float = 0.2,
max_tokens: Optional[int] = None
) -> LLMResponse:
...
`
قم بتغليف عميل Fable الحالي وعميل Astra الجديد كمحولات على التوالي. وأنشئ LLMFactory لإرجاع الكائن بناءً على قيمة متغير البيئة LLM_PROVIDER.
`python
import httpx
import os
from typing import List, Optional
from core.llm_protocol import LLMClientAdapter, LLMMessage, LLMResponse, LLMUsage
class AstraAdapter:
def init(self, api_key: str, base_url: str = "https://api.astra.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.client = httpx.AsyncClient(
base_url=self.base_url,
headers={"Authorization": f"Bearer {self.api_key}"},
timeout=httpx.Timeout(connect=5.0, read=90.0, write=10.0, pool=5.0)
)
async def generate_completion(
self,
messages: List[LLMMessage],
temperature: float = 0.2,
max_tokens: Optional[int] = None
) -> LLMResponse:
payload = {
"model": "gpt-6-astra",
"messages": [msg.model_dump() for msg in messages],
"temperature": temperature,
}
if max_tokens:
payload["max_tokens"] = max_tokens
resp = await self.client.post("/chat/completions", json=payload)
resp.raise_for_status()
data = resp.json()
return LLMResponse(
content=data["choices"][0]["message"]["content"],
usage=LLMUsage(
prompt_tokens=data["usage"]["prompt_tokens"],
completion_tokens=data["usage"]["completion_tokens"],
total_tokens=data["usage"]["total_tokens"]
),
model_name=data["model"],
provider="astra"
)
class FableAdapter:
def init(self, api_key: str, endpoint_url: str):
self.api_key = api_key
self.endpoint_url = endpoint_url
self.client = httpx.AsyncClient(
headers={"X-Fable-Key": self.api_key},
timeout=httpx.Timeout(connect=5.0, read=60.0, write=10.0, pool=5.0)
)
async def generate_completion(
self,
messages: List[LLMMessage],
temperature: float = 0.2,
max_tokens: Optional[int] = None
) -> LLMResponse:
fable_payload = {
"prompt_sequence": [{"speaker": m.role, "text": m.content} for m in messages],
"gen_params": {"temp": temperature, "limit": max_tokens or 1024}
}
resp = await self.client.post(self.endpoint_url, json=fable_payload)
resp.raise_for_status()
data = resp.json()
return LLMResponse(
content=data["result"]["generated_text"],
usage=LLMUsage(
prompt_tokens=data["meta"]["tokens_in"],
completion_tokens=data["meta"]["tokens_out"],
total_tokens=data["meta"]["tokens_in"] + data["meta"]["tokens_out"]
),
model_name="fable-legacy",
provider="fable"
)
class LLMFactory:
@staticmethod
def get_adapter() -> LLMClientAdapter:
provider = os.getenv("LLM_PROVIDER", "astra").lower()
if provider == "astra":
return AstraAdapter(
api_key=os.environ["ASTRA_API_KEY"],
base_url=os.getenv("ASTRA_BASE_URL", "https://api.astra.ai/v1")
)
elif provider == "fable":
return FableAdapter(
api_key=os.environ["FABLE_API_KEY"],
endpoint_url=os.environ["FABLE_ENDPOINT_URL"]
)
raise ValueError(f"Unsupported LLM provider: {provider}")
`
يجب اكتشاف الحوادث التي تتعلق بتلف تنسيق الاستجابة عند تغيير النموذج مسبقاً من خلال اختبارات الوحدة (Unit Tests). هذا اختبار يتحقق من مخططات Pydantic واستجابات المحاكاة (Mock).
`python
import pytest
from core.llm_protocol import LLMMessage, LLMResponse
from core.adapters import AstraAdapter
@pytest.mark.asyncio
async def test_astra_response_contract_compliance(monkeypatch):
mock_payload = {
"id": "chatcmpl-astra-001",
"model": "gpt-6-astra",
"choices": [
{"message": {"role": "assistant", "content": "نتائج طبيعية"}}
],
"usage": {
"prompt_tokens": 120,
"completion_tokens": 45,
"total_tokens": 165
}
}
class MockResponse:
def raise_for_status(self): pass
def json(self): return mock_payload
async def mock_post(*args, **kwargs):
return MockResponse()
adapter = AstraAdapter(api_key="test-key")
monkeypatch.setattr(adapter.client, "post", mock_post)
messages = [LLMMessage(role="user", content="طلب اختبار")]
result: LLMResponse = await adapter.generate_completion(messages)
assert isinstance(result, LLMResponse)
assert result.provider == "astra"
assert result.model_name == "gpt-6-astra"
assert result.content == "نتائج طبيعية"
assert result.usage.prompt_tokens == 120
assert result.usage.completion_tokens == 45
assert result.usage.total_tokens == 165
`
إذا قمت بربط هدف استدعاء وحدة نقطة دخول الأعمال بـ LLMFactory.get_adapter().generate_completion(...)، فل لن يستغرق عمل الانتقال حتى ساعة واحدة. حتى لو حدث خلل في النموذج الجديد، فإن إعادة ضبط متغير البيئة إلى LLM_PROVIDER=fable ينهي الأمر.
يجب التحقق من أرقام تقليل الهلوسة المكتوبة في عبارات الترويج الخاصة بالمزود باستخدام بيانات الشركة الداخلية. يتم تجميع 50 بيانات من سياسات الاسترداد أو شروط الدفع في مجموعة بيانات ذهبية ومقارنتها باستخدام مكتبة المصدر المفتوح DeepEval.
`python
import json
import asyncio
from deepeval.test_case import LLMTestCase
from deepeval.metrics import HallucinationMetric
from core.adapters import AstraAdapter, FableAdapter
from core.llm_protocol import LLMMessage
async def run_batch_evaluation():
with open("eval_dataset_50.json", "r", encoding="utf-8") as f:
cases = json.load(f)
astra = AstraAdapter(api_key="astra-key")
fable = FableAdapter(api_key="fable-key", endpoint_url="https://api.fable.internal/v1")
hallucination_metric = HallucinationMetric(threshold=0.1)
astra_scores, fable_scores = [], []
for case in cases:
msg = [LLMMessage(role="user", content=case["input"])]
astra_res = await astra.generate_completion(msg)
tc_astra = LLMTestCase(input=case["input"], actual_output=astra_res.content, retrieval_context=case["retrieval_context"])
hallucination_metric.measure(tc_astra)
astra_scores.append(hallucination_metric.score)
fable_res = await fable.generate_completion(msg)
tc_fable = LLMTestCase(input=case["input"], actual_output=fable_res.content, retrieval_context=case["retrieval_context"])
hallucination_metric.measure(tc_fable)
fable_scores.append(hallucination_metric.score)
avg_fable = sum(fable_scores) / len(fable_scores)
avg_astra = sum(astra_scores) / len(astra_scores)
delta = ((avg_fable - avg_astra) / avg_fable) * 100
print(f"Empirical Hallucination Reduction: {delta:.2f}%")
if name == "__main اصلی":
asyncio.run(run_batch_evaluation())
`
خاضت نتائج القياس اختبار t للأزواج (Paired t-test) باستخدام دالة scipy.stats.ttest_rel.
يُعتقد أن الهلوسة الفعلية قد انخفضت فقط عندما تنخفض القيمة الاحتمالية (p-value) إلى ما دون 0.05 بناءً على 50 عينة.
المشكلة الأكثر خطورة في الخدمة حقاً هي تأخير الاستجابة المتزايد مقارنة برموز الخطأ. إذا لم تكن هناك مهلة (Timeout)، فإن مجموعة اتصالات الواجهة الخلفية ستجف في ومضة عين. هذا هو رمز قاطع الدائرة (Circuit Breaker) الذي يقطع المكالمات فوراً ويتحول إلى النموذج السابق عند الفشل 5 مرات متتالية.
`python
import logging
from pybreaker import CircuitBreaker, CircuitBreakerError
from core.adapters import AstraAdapter, FableAdapter
from core.llm_protocol import LLMMessage, LLMResponse
logger = logging.getLogger(name)
astra_breaker = CircuitBreaker(fail_max=5, reset_timeout=30)
class ResilientLLMClient:
def init(self, primary: AstraAdapter, fallback: FableAdapter):
self.primary = primary
self.fallback = fallback
async def execute(self, messages: list[LLMMessage]) -> LLMResponse:
try:
return await self._call_primary(messages)
except (CircuitBreakerError, Exception) as exc:
logger.warning("Primary degraded [%s]. Executing fallback.", str(exc))
return await self._call_fallback(messages)
async def _call_primary(self, messages: list[LLMMessage]) -> LLMResponse:
if astra_breaker.current_state == "open":
raise CircuitBreakerError("Circuit breaker OPEN")
try:
response = await self.primary.generate_completion(messages)
astra_breaker.success()
return response
except Exception as err:
astra_breaker.fail()
raise err
async def _call_fallback(self, messages: list[LLMMessage]) -> LLMResponse:
return await self.fallback.generate_completion(messages)
`
إذا تجاوزت مهلة الاتصال (Connect) 5 ثوانٍ ومهلة القراءة (Read) 90 ثانية، فسيعتبر ذلك فشلاً دون تأخير. عندما تتراكم 5 حالات فشل، تفتح الدائرة وتحظر المكالمات الخارجية في 0 ثانية لمدة 30 ثانية. هذا هو خط الدفاع الأدنى لمنع التمسك بالمكالمات المتأخرة وقتل الخادم بأكمله.
إذا قمت بإرسال إدخال المستخدم كما هو إلى واجهة برمجة تطبيقات خارجية موجودة في منطقة خارجية، فسوف ينتهك ذلك المادة 28-8 من قانون حماية المعلومات الشخصية (النقل الدولي للمعلومات الشخصية). يجب إكمال الإخفاء في مرحلة الذاكرة (In-memory) قبل خروج البيانات خارج الشبكة. نظراً لأن التعبير النقدي البسيط المكون من 13 رقماً يمحو حتى أرقام الفواتير أو أرقام الطلبات، يتم تطبيق خوارزمية التحقق من الوزن (Checksum) معاً.
ext{Checksum} = 11 - left( left( sum_{i=1}^{12} d_i imes w_i ight) mod 11 ight)`python
import re
from typing import List
from core.llm_protocol import LLMMessage
class PIIMaskingPipeline:
EMAIL_REGEX = re.compile(r'[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+.[a-zA-Z0-9-.]+')
INTERNAL_IP_REGEX = re.compile(
r'\b(?:10.\d{1,3}.\d{1,3}.\d{1,3}|'
r'172.(?:1[6-9]|2\d|3[0-1]).\d{1,3}.\d{1,3}|'
r'192.168.\d{1,3}.\d{1,3})\b'
)
RRN_CANDIDATE_REGEX = re.compile(r'\b(\d{6})[- ]?(\d{7})\b')
@classmethod
def is_valid_rrn(cls, front: str, back: str) -> bool:
full = front + back
if len(full) != 13 or not full.isdigit():
return False
weights = [2, 3, 4, 5, 6, 7, 8, 9, 2, 3, 4, 5]
s = sum(int(full[i]) * weights[i] for i in range(12))
remainder = (11 - (s % 11)) % 10
return remainder == int(full[12])
@classmethod
def mask_rrn(cls, text: str) -> str:
def replace(match):
front, back = match.group(1), match.group(2)
if cls.is_valid_rrn(front, back):
return "[RESIDENT_ID_MASKED]"
return match.group(0)
return cls.RRN_CANDIDATE_REGEX.sub(replace, text)
@classmethod
def sanitize(cls, text: str) -> str:
text = cls.mask_rrn(text)
text = cls.EMAIL_REGEX.sub("[EMAIL_MASKED]", text)
text = cls.INTERNAL_IP_REGEX.sub("[INTERNAL_IP_MASKED]", text)
return text
@classmethod
def sanitize_messages(cls, messages: List[LLMMessage]) -> List[LLMMessage]:
return [
LLMMessage(role=m.role, content=cls.sanitize(m.content))
for m in messages
]
`
تأكد من وجود بند الاحتفاظ الصفرى بالبيانات (Zero Data Retention - ZDR) في عقد المزود والذي ينص على عدم استخدام مطالبات الإدخال لإعادة التدريب. امنع تخزين جسم المطالبة في سجلات CloudWatch واترك فقط trace_id، ووقت التأخير، وعدد الرموز.
قم بتسليم جدول تدفق البيانات أدناه إلى فريق الأمان أو مسؤول حماية المعلومات الشخصية (CPO).
| نطاق المعالجة | تدفق البيانات | البيانات المرسلة | جهاز التحكم الأمني | معيار الامتثال |
|---|---|---|---|---|
| النطاق 1 | العميل ← البوابة | المطالبة الأصلية | TLS 1.3, تفويض JWT | تشفير نطاق الاتصال |
| النطاق 2 | البرمجية الوسيطة للإخفاء في الذاكرة | الأصل ← إخفاء المعرفات | الاستبدال القائم على Checksum، تدمير الذاكرة الفوري | مبدأ الحد الأدنى لجمع المعلومات الشخصية |
| النطاق 3 | محول الواجهة الخلفية ← واجهة برمجة تطبيقات Astra | حمولة مكتملة الإخفاء | حقن مفتاح AWS Secrets Manager | المادة 28-8 من قانون حماية المعلومات الشخصية |
| النطاق 4 | واجهة برمجة تطبيقات Astra ← محول الواجهة الخلفية | نص منشأ بواسطة النموذج | ضمان الاحتفاظ الصفري بالبيانات (ZDR)، TLS | شروط اتفاقية معالجة البيانات المؤسسية (DPA) |
| النطاق 5 | مسجل الواجهة الخلفية ← سجلات CloudWatch | قياسات الرموز ووقت التأخير | حظر تسجيل النص الأساسي، يقتصر على البيانات الوصفية | المادة 29 من قانون حماية المعلومات الشخصية |
عند التقدم بطلب للمراجعة الأمنية، قم بتقديم الحزم التالية دفعة واحدة: 1) مواصفات إشعار المعالج في سياسة معالجة المعلومات الشخصية، 2) نسخة من عقد DPA، 3) نتائج اختبار البرمجية الوسيطة لإخفاء Checksum، 4) وثائق إعداد المهلة وقاطع الدائرة. المستندات المرفقة برموز دفاع البنية التحتية تقلل بشكل ملحوظ من فترة المراجعة التقنية.