تنفيذ القواعد المخصصة

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

قاعدة مدخلات مخصصة

تنفيذ إجراء Python

# actions/input_safety.py
from nemoguardrails.actions import action
from nemoguardrails.actions.actions import ActionResult
import re

@action(name="check_financial_advice")
async def check_financial_advice(user_message: str) -> ActionResult:
    """
    التحقق إذا كان المستخدم يطلب نصيحة مالية.
    حظر الطلبات التي يمكن تفسيرها على أنها
    تسعى لتوصيات استثمارية شخصية.
    """
    financial_patterns = [
        r"should i (buy|sell|invest)",
        r"what stock",
        r"investment advice",
        r"financial recommendation",
        r"is .* a good investment",
    ]

    for pattern in financial_patterns:
        if re.search(pattern, user_message.lower()):
            return ActionResult(
                return_value=False,
                context_updates={"blocked_reason": "financial_advice"}
            )

    return ActionResult(return_value=True)


@action(name="check_competitor_mention")
async def check_competitor_mention(user_message: str) -> ActionResult:
    """حظر المناقشات حول المنافسين."""
    competitors = ["competitor_a", "competitor_b", "rival_corp"]

    message_lower = user_message.lower()
    for competitor in competitors:
        if competitor in message_lower:
            return ActionResult(
                return_value=False,
                context_updates={"blocked_reason": "competitor_mention"}
            )

    return ActionResult(return_value=True)


@action(name="extract_intent_entities")
async def extract_intent_entities(user_message: str) -> dict:
    """استخراج الكيانات والنية من رسالة المستخدم."""
    # في الإنتاج، استخدم NER أو مصنف النوايا
    entities = {
        "product_mentioned": None,
        "action_requested": None,
        "urgency": "normal"
    }

    if "urgent" in user_message.lower():
        entities["urgency"] = "high"

    return ActionResult(return_value=entities)

تدفق Colang

# rails/custom_input.co
define flow custom input checks
  # تشغيل جميع التحققات المخصصة للمدخلات
  $financial_ok = execute check_financial_advice(user_message=$user_message)

  if not $financial_ok
    bot decline financial advice
    stop

  $competitor_ok = execute check_competitor_mention(user_message=$user_message)

  if not $competitor_ok
    bot redirect away from competitor
    stop

  # استخراج الكيانات للاستخدام اللاحق
  $entities = execute extract_intent_entities(user_message=$user_message)
  # تخزين في السياق للتدفقات اللاحقة

define bot decline financial advice
  "لست قادراً على تقديم نصيحة مالية شخصية. لقرارات الاستثمار، يرجى استشارة مستشار مالي مرخص."

define bot redirect away from competitor
  "سأكون سعيداً بالتركيز على كيف يمكن لحلولنا مساعدتك. ما هي الاحتياجات المحددة التي تسعى لمعالجتها؟"

قاعدة مخرجات مخصصة

إجراء التحقق من الحقائق

# actions/output_safety.py
from nemoguardrails.actions import action
import httpx

@action(name="verify_factual_claims")
async def verify_factual_claims(bot_message: str) -> dict:
    """
    التحقق من الادعاءات الواقعية في الاستجابة.
    يُرجع نتيجة التحقق مع الادعاءات المُعلّمة.
    """
    # استخراج الادعاءات (في الإنتاج، استخدم NLP)
    claims = extract_claims(bot_message)

    results = {
        "verified": True,
        "flagged_claims": [],
        "confidence": 1.0
    }

    for claim in claims:
        # التحقق من قاعدة المعرفة
        verified = await check_against_kb(claim)
        if not verified:
            results["flagged_claims"].append(claim)
            results["verified"] = False
            results["confidence"] *= 0.5

    return results


@action(name="check_response_tone")
async def check_response_tone(bot_message: str) -> dict:
    """التأكد من أن الاستجابة تحافظ على نبرة مهنية."""
    informal_patterns = [
        r"lol", r"omg", r"gonna", r"wanna",
        r"!!+", r"\?\?+", r"😀|😊|🙂"  # فحص الرموز التعبيرية
    ]

    issues = []
    for pattern in informal_patterns:
        if re.search(pattern, bot_message.lower()):
            issues.append(f"نمط غير رسمي: {pattern}")

    return {
        "professional": len(issues) == 0,
        "issues": issues
    }


@action(name="sanitize_output")
async def sanitize_output(bot_message: str) -> str:
    """إزالة أو إخفاء المعلومات الحساسة من الاستجابة."""
    import re

    # إخفاء أي مراجع داخلية مسربة
    sanitized = re.sub(
        r'internal[_-]?id[:\s]*\w+',
        '[INTERNAL_REF_REMOVED]',
        bot_message,
        flags=re.IGNORECASE
    )

    # إزالة معلومات التصحيح
    sanitized = re.sub(
        r'\[DEBUG:.*?\]',
        '',
        sanitized
    )

    return sanitized

تدفق التحقق من المخرجات

# rails/custom_output.co
define flow custom output checks
  # التحقق من الدقة الواقعية
  $fact_check = execute verify_factual_claims(bot_message=$bot_message)

  if not $fact_check.verified
    # إعادة التوليد مع تحفظات
    $bot_message = execute add_uncertainty_caveats(
      message=$bot_message,
      claims=$fact_check.flagged_claims
    )

  # التحقق من النبرة المهنية
  $tone_check = execute check_response_tone(bot_message=$bot_message)

  if not $tone_check.professional
    $bot_message = execute rephrase_professionally(message=$bot_message)

  # التنظيف النهائي
  $bot_message = execute sanitize_output(bot_message=$bot_message)

define flow add uncertainty caveats
  """إضافة لغة تحفظية للادعاءات غير المتحققة."""
  bot say "بناءً على فهمي، على الرغم من أنني أوصي بالتحقق: "
  bot say $bot_message

قاعدة حوار مخصصة

حواجز المحادثة

# actions/dialog_controls.py
from nemoguardrails.actions import action
from datetime import datetime, timedelta

# تتبع حالة المحادثة
conversation_state = {}

@action(name="check_rate_limit")
async def check_rate_limit(user_id: str) -> bool:
    """تحديد معدل المستخدمين لمنع الإساءة."""
    now = datetime.now()
    window = timedelta(minutes=1)

    if user_id not in conversation_state:
        conversation_state[user_id] = {"messages": [], "warnings": 0}

    state = conversation_state[user_id]

    # إزالة الرسائل القديمة
    state["messages"] = [
        ts for ts in state["messages"]
        if now - ts < window
    ]

    # التحقق من الحد (10 رسائل في الدقيقة)
    if len(state["messages"]) >= 10:
        state["warnings"] += 1
        return False

    state["messages"].append(now)
    return True


@action(name="check_topic_allowed")
async def check_topic_allowed(
    topic: str,
    user_tier: str = "free"
) -> bool:
    """التحقق إذا كان مستوى المستخدم يسمح بالوصول للموضوع."""
    topic_permissions = {
        "free": ["general", "faq", "basic_support"],
        "pro": ["general", "faq", "basic_support", "advanced_support"],
        "enterprise": ["*"]  # جميع المواضيع
    }

    allowed = topic_permissions.get(user_tier, [])
    return "*" in allowed or topic in allowed


@action(name="track_conversation_metrics")
async def track_conversation_metrics(
    user_id: str,
    message: str,
    response: str
) -> None:
    """تتبع المقاييس للتحليلات."""
    # إرسال لنظام التحليلات
    await log_to_analytics({
        "user_id": user_id,
        "message_length": len(message),
        "response_length": len(response),
        "timestamp": datetime.now().isoformat()
    })

تدفق التحكم في الحوار

# rails/custom_dialog.co
define flow controlled conversation
  # تحديد المعدل
  $rate_ok = execute check_rate_limit(user_id=$user_id)

  if not $rate_ok
    bot rate limit warning
    stop

  # التحكم في الوصول للموضوع
  $topic = execute classify_topic(text=$user_message)
  $user_tier = execute get_user_tier(user_id=$user_id)
  $topic_allowed = execute check_topic_allowed(
    topic=$topic,
    user_tier=$user_tier
  )

  if not $topic_allowed
    bot suggest upgrade
    stop

  # المتابعة مع التدفق العادي
  # ... بقية منطق المحادثة

define bot rate limit warning
  "أنت ترسل رسائل بسرعة كبيرة. يرجى الانتظار لحظة قبل المتابعة."

define bot suggest upgrade
  "هذا الموضوع متاح مع خطة Pro الخاصة بنا. هل تريد معرفة المزيد عن الترقية؟"

دمج القواعد المخصصة

التكوين

# config/config.yml
models:
  - type: main
    engine: openai
    model: gpt-4o

rails:
  input:
    flows:
      - self check input
      - custom input checks    # قاعدتنا المخصصة

  output:
    flows:
      - self check output
      - custom output checks   # قاعدتنا المخصصة

  dialog:
    flows:
      - controlled conversation  # قاعدة الحوار المخصصة لدينا

# تسجيل وحدات الإجراءات
actions:
  - actions.input_safety
  - actions.output_safety
  - actions.dialog_controls

التطبيق الرئيسي

# main.py
from nemoguardrails import LLMRails, RailsConfig

# استيراد الإجراءات المخصصة لتسجيلها
import actions.input_safety
import actions.output_safety
import actions.dialog_controls

config = RailsConfig.from_path("./config")
rails = LLMRails(config)

async def process_message(user_id: str, message: str) -> str:
    """معالجة رسالة المستخدم مع جميع القواعد المخصصة."""
    response = await rails.generate_async(
        messages=[{"role": "user", "content": message}],
        context={"user_id": user_id}
    )

    # تتبع المقاييس
    await actions.dialog_controls.track_conversation_metrics(
        user_id=user_id,
        message=message,
        response=response["content"]
    )

    return response["content"]

نصيحة التنفيذ: حافظ على الإجراءات مركزة وقابلة للاختبار. يجب أن يقوم كل إجراء بشيء واحد جيداً. ركّب السلوك المعقد في تدفقات Colang بدلاً من إجراءات Python.

التالي: تنفيذ قواعد استرجاع RAG للردود المبنية على المعرفة. :::