Home
Blog
ما بعد تحميل السياق الزائد: كيف يجلب Parlant × ميلفوس التحكم والوضوح لسلوك وكيل LLM

ما بعد تحميل السياق الزائد: كيف يجلب Parlant × ميلفوس التحكم والوضوح لسلوك وكيل LLM

Tutorials

November 05, 2025

Min Yin

تخيل أنه طُلب منك إكمال مهمة تتضمن 200 قاعدة عمل و50 أداة و30 عرضًا توضيحيًا، وليس لديك سوى ساعة واحدة للقيام بذلك. هذا ببساطة مستحيل. ومع ذلك فإننا غالبًا ما نتوقع من النماذج اللغوية الكبيرة أن تفعل ذلك بالضبط عندما نحولها إلى "وكلاء" ونثقلها بالتعليمات.

في الممارسة العملية، سرعان ما ينهار هذا النهج. حيث تقوم أطر عمل الوكلاء التقليدية، مثل LangChain أو LlamaIndex، بحقن جميع القواعد والأدوات في سياق النموذج في آن واحد، مما يؤدي إلى تعارض القواعد، وتحميل السياق فوق طاقته، وسلوك غير متوقع في الإنتاج.

ولمعالجة هذه المشكلة، اكتسب إطار عمل الوكيل مفتوح المصدر المسمى Parlant مؤخرًا زخمًا على GitHub. وهو يقدم نهجًا جديدًا يسمى نمذجة المحاذاة، إلى جانب آلية إشرافية وانتقالات مشروطة تجعل سلوك الوكيل أكثر قابلية للتحكم والتفسير.

عند إقرانها مع Milvus، وهي قاعدة بيانات متجهة مفتوحة المصدر، تصبح Parlant أكثر قدرة. تضيف Milvus ذكاءً دلاليًا، مما يسمح للوكلاء باسترداد القواعد والسياق الأكثر صلة بشكل ديناميكي في الوقت الفعلي - مما يجعلها دقيقة وفعالة وجاهزة للإنتاج.

في هذا المنشور، سنستكشف في هذا المقال كيفية عمل Parlant تحت الغطاء - وكيف يتيح تكامله مع Milvus إمكانية الوصول إلى مستوى الإنتاج.

لماذا تنهار أطر عمل الوكيل التقليدية

تحب أطر عمل الوكلاء التقليدية أن تكون كبيرة: مئات القواعد، وعشرات الأدوات، وحفنة من العروض التوضيحية - وكلها محشورة في موجه واحد مكتظ. قد يبدو الأمر رائعًا في عرض توضيحي أو اختبار صندوق رمل صغير، ولكن بمجرد أن تدفعه إلى الإنتاج، تبدأ الشقوق في الظهور بسرعة.

القواعد المتضاربة تجلب الفوضى: عندما تُطبَّق قاعدتان أو أكثر في نفس الوقت، لا تملك هذه الأطر طريقة مدمجة لتحديد أيهما يفوز. أحيانًا تختار واحدة. وأحيانًا تمزج الاثنين معًا. وأحيانًا تفعل شيئًا لا يمكن التنبؤ به تمامًا.
حالات الحافة تكشف الثغرات: لا يمكنك التنبؤ بكل ما قد يقوله المستخدم. وعندما يصطدم نموذجك بشيء خارج بيانات التدريب الخاصة به، فإنه يتجه افتراضيًا إلى إجابات عامة وغير ملتزمة.
تصحيح الأخطاء مؤلم ومكلف: عندما يسيء الوكيل التصرف، يكاد يكون من المستحيل تحديد القاعدة التي تسببت في المشكلة. نظرًا لأن كل شيء يعيش داخل موجه نظام عملاق واحد، فإن الطريقة الوحيدة لإصلاحه هي إعادة كتابة الموجه وإعادة اختبار كل شيء من الصفر.

ما هو "بارلانت" وكيف يعمل

Parlant هو محرك محاذاة مفتوح المصدر لوكلاء LLM. يمكنك التحكم بدقة في كيفية تصرف الوكيل عبر سيناريوهات مختلفة من خلال نمذجة عملية اتخاذ القرار الخاصة به بطريقة منظمة وقائمة على القواعد.

لمعالجة المشاكل الموجودة في أطر عمل الوكلاء التقليدية، يقدم Parlant نهجًا قويًا جديدًا: نمذجة المحاذاة. وتتمثل فكرتها الأساسية في فصل تعريف القواعد عن تنفيذ القواعد، مما يضمن إدخال القواعد الأكثر صلة فقط في سياق الوكيل في أي وقت معين.

الإرشادات الدقيقة: جوهر نمذجة المواءمة

يقع مفهوم المبادئ التوجيهية الحبيبية في صميم نموذج المواءمة في بارلانت. فبدلاً من كتابة موجه نظام عملاق واحد مليء بالقواعد، يمكنك تحديد إرشادات صغيرة ومعيارية - كل منها يصف كيف يجب على الوكيل التعامل مع نوع معين من المواقف.

يتكون كل مبدأ توجيهي من ثلاثة أجزاء:

الشرط - وصف بلغة طبيعية للوقت الذي يجب أن تنطبق فيه القاعدة. يقوم Parlant بتحويل هذا الشرط إلى متجه دلالي ويطابقه مع مدخلات المستخدم لمعرفة ما إذا كان ذا صلة.
الإجراء - تعليمات واضحة تحدد كيفية استجابة الوكيل بمجرد استيفاء الشرط. يتم حقن هذا الإجراء في سياق LLM فقط عند تشغيله.
الأدوات - أي وظائف خارجية أو واجهات برمجة تطبيقات مرتبطة بتلك القاعدة المحددة. يتم كشفها للوكيل فقط عندما يكون المبدأ التوجيهي نشطًا، مما يجعل استخدام الأداة خاضعًا للتحكم ومدركًا للسياق.

await agent.create_guideline(
    condition="The user asks about a refund and the order amount exceeds 500 RMB",
    action="First call the order status check tool to confirm whether the refund conditions are met, then provide a detailed explanation of the refund process",
    tools=[check_order_status, calculate_refund_amount]
)

في كل مرة يتفاعل فيها المستخدم مع الوكيل، يقوم Parlant بتشغيل خطوة مطابقة خفيفة للعثور على أكثر ثلاثة إلى خمسة إرشادات ذات صلة. يتم حقن تلك القواعد فقط في سياق النموذج، مما يحافظ على إيجاز وتركيز الإرشادات مع ضمان اتباع الوكيل للقواعد الصحيحة باستمرار.

آلية الإشراف على الدقة والاتساق

للحفاظ على مزيد من الدقة والاتساق، يقدم Parlant آلية إشراف تعمل كطبقة ثانية من مراقبة الجودة. تتكشف العملية في ثلاث خطوات:

1. إنشاء رد مرشح - ينشئ الوكيل ردًا أوليًا بناءً على الإرشادات المتطابقة وسياق المحادثة الحالي.

2. التحقق من الامتثال - تتم مقارنة الرد بالإرشادات النشطة للتحقق من اتباع كل التعليمات بشكل صحيح.

3. المراجعة أو التأكيد - إذا تم العثور على أي مشاكل، يقوم النظام بتصحيح الإخراج؛ إذا تم التحقق من كل شيء، تتم الموافقة على الرد وإرساله إلى المستخدم.

تضمن آلية الإشراف هذه أن الوكيل لا يفهم القواعد فحسب، بل يلتزم بها فعليًا قبل الرد - مما يحسّن الموثوقية والتحكم على حد سواء.

الانتقالات الشرطية للتحكم والسلامة

في أطر عمل الوكيل التقليدية، تكون كل أداة متاحة مكشوفة للوكيل في جميع الأوقات. غالبًا ما يؤدي نهج "كل شيء على الطاولة" هذا إلى مطالبات مثقلة واستدعاءات غير مقصودة للأداة. يحل Parlant هذا الأمر من خلال الانتقالات الشرطية. على غرار كيفية عمل آلات الحالة، لا يتم تشغيل إجراء أو أداة إلا عند استيفاء شرط معين. ترتبط كل أداة ارتباطًا وثيقًا بالمبدأ التوجيهي المقابل لها، ولا تصبح متاحة إلا عند تفعيل شرط ذلك المبدأ التوجيهي.

# The balance inquiry tool is exposed only when the condition "the user wants to make a transfer" is met
await agent.create_guideline(
    condition="The user wants to make a transfer",
    action="First check the account balance. If the balance is below 500 RMB, remind the user that an overdraft fee may apply.",
    tools=[get_user_account_balance]
)

تعمل هذه الآلية على تحويل استدعاء الأداة إلى انتقال مشروط - تنتقل الأدوات من "غير نشط" إلى "نشط" فقط عند استيفاء شروط التفعيل الخاصة بها. من خلال هيكلة التنفيذ بهذه الطريقة، يضمن Parlant أن كل إجراء يحدث بشكل مدروس وفي سياقه، مما يمنع إساءة الاستخدام مع تحسين كل من الكفاءة وسلامة النظام.

كيف يقوم ميلفوس بتشغيل بارلانت

عندما ننظر تحت غطاء عملية مطابقة المبادئ التوجيهية في بارلانت، يتضح لنا تحدٍ تقني أساسي: كيف يمكن للنظام العثور على أكثر ثلاث إلى خمس قواعد ذات صلة من بين مئات - أو حتى آلاف - الخيارات في بضعة أجزاء من الثانية؟ وهنا بالضبط يأتي دور قاعدة بيانات المتجهات. الاسترجاع الدلالي هو ما يجعل ذلك ممكناً.

كيف يدعم ميلفوس عملية مطابقة المبادئ التوجيهية في بارلانت

تعمل مطابقة المبادئ التوجيهية من خلال التشابه الدلالي. يتم تحويل حقل حالة كل مبدأ توجيهي إلى تضمين متجه، بحيث يتم التقاط معناه بدلاً من مجرد نصه الحرفي. عندما يرسل المستخدم رسالة، يقارن Parlant دلالات تلك الرسالة مع جميع التضمينات الإرشادية المخزنة للعثور على أكثرها صلة.

إليك كيفية عمل العملية خطوة بخطوة:

1. ترميز الاستعلام - يتم تحويل رسالة المستخدم وسجل المحادثات الأخيرة إلى متجه استعلام.

2. البحث عن التشابه - يُجري النظام بحثًا عن التشابه داخل مخزن متجه المبادئ التوجيهية للعثور على أقرب التطابقات.

3. استرجاع أفضل النتائج - يتم إرجاع أفضل ثلاثة إلى خمسة إرشادات ذات صلة دلالية.

4. الحقن في السياق - يتم بعد ذلك إدراج هذه الإرشادات المتطابقة ديناميكيًا في سياق LLM حتى يتمكن النموذج من العمل وفقًا للقواعد الصحيحة.

لجعل سير العمل هذا ممكناً، يجب أن توفر قاعدة بيانات المتجهات ثلاث قدرات حاسمة: البحث عن أقرب جار تقريبي (ANN) عالي الأداء، وتصفية البيانات الوصفية المرنة، وتحديثات المتجهات في الوقت الفعلي. توفر Milvus، قاعدة بيانات المتجهات مفتوحة المصدر، وهي قاعدة بيانات المتجهات السحابية الأصلية مفتوحة المصدر، أداءً على مستوى الإنتاج في جميع المجالات الثلاثة.

لفهم كيفية عمل Milvus في سيناريوهات حقيقية، دعنا ننظر إلى وكيل خدمات مالية كمثال.

لنفترض أن النظام يحدد 800 إرشادات عمل تغطي مهام مثل الاستفسارات عن الحسابات، وتحويلات الأموال، واستشارات منتجات إدارة الثروات. في هذا الإعداد، يعمل نظام Milvus كطبقة تخزين واسترجاع لجميع بيانات المبادئ التوجيهية.

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType
import parlant.sdk as p
# Connect to Milvus
connections.connect(host=“localhost”, port=“19530”)
# Define the schema for the guideline collection
fields = [
FieldSchema(name=“guideline_id”, dtype=DataType.VARCHAR, max_length=100, is_primary=True),
FieldSchema(name=“condition_vector”, dtype=DataType.FLOAT_VECTOR, dim=768),
FieldSchema(name=“condition_text”, dtype=DataType.VARCHAR, max_length=1000),
FieldSchema(name=“action_text”, dtype=DataType.VARCHAR, max_length=2000),
FieldSchema(name=“priority”, dtype=DataType.INT64),
FieldSchema(name=“business_domain”, dtype=DataType.VARCHAR, max_length=50)
]
schema = CollectionSchema(fields=fields, description=“Agent Guidelines”)
guideline_collection = Collection(name=“agent_guidelines”, schema=schema)
# Create an HNSW index for high-performance retrieval
index_params = {
“index_type”: “HNSW”,
“metric_type”: “COSINE”,
“params”: {“M”: 16, “efConstruction”: 200}
}
guideline_collection.create_index(field_name=“condition_vector”, index_params=index_params)

والآن، عندما يقول المستخدم "أريد تحويل 100,000 رنمينبي إلى حساب والدتي"، فإن تدفق وقت التشغيل يكون

1. إعادة توجيه الاستعلام - تحويل مدخلات المستخدم إلى متجه ذي 768 بُعدًا.

2. الاسترجاع الهجين - تشغيل بحث تشابه متجه في Milvus مع تصفية البيانات الوصفية (على سبيل المثال، business_domain="transfer").

3. ترتيب النتائج - رتب الإرشادات المرشحة بناءً على درجات التشابه مع قيم أولويتها.

4. حقن السياق - حقن أفضل 3 إرشادات متطابقة action_text في سياق وكيل Parlant.

في هذا التكوين، يوفر Milvus زمن انتقال P99 أقل من 15 مللي ثانية، حتى عندما يصل حجم مكتبة المبادئ التوجيهية إلى 100,000 مدخل. وعلى سبيل المقارنة، يؤدي استخدام قاعدة بيانات علائقية تقليدية مع مطابقة الكلمات الرئيسية عادةً إلى زمن انتقال يتجاوز 200 مللي ثانية ودقة مطابقة أقل بكثير.

كيفية تمكين Milvus للذاكرة طويلة المدى والتخصيص

يقوم Milvus بأكثر من المطابقة الإرشادية. في السيناريوهات التي يحتاج فيها الوكلاء إلى ذاكرة طويلة الأجل واستجابات مخصصة، يمكن أن يعمل Milvus كطبقة ذاكرة تخزن وتسترجع تفاعلات المستخدمين السابقة على شكل تضمينات متجهة، مما يساعد الوكيل على تذكر ما تمت مناقشته من قبل.

# store user’s past interactions
user_memory_fields = [
    FieldSchema(name="interaction_id", dtype=DataType.VARCHAR, max_length=100, is_primary=True),
    FieldSchema(name="user_id", dtype=DataType.VARCHAR, max_length=50),
    FieldSchema(name="interaction_vector", dtype=DataType.FLOAT_VECTOR, dim=768),
    FieldSchema(name="interaction_summary", dtype=DataType.VARCHAR, max_length=500),
    FieldSchema(name="timestamp", dtype=DataType.INT64)
]
memory_collection = Collection(name="user_memory", schema=CollectionSchema(user_memory_fields))

عندما يعود المستخدم نفسه، يمكن للوكيل استرجاع التفاعلات التاريخية الأكثر صلة من Milvus واستخدامها لتوليد تجربة أكثر ترابطاً وشبيهة بالبشر. على سبيل المثال، إذا سأل أحد المستخدمين عن صندوق استثماري في الأسبوع الماضي، يمكن للوكيل استرجاع هذا السياق والرد بشكل استباقي: "مرحباً بعودتك! هل لا تزال لديك أسئلة حول الصندوق الذي ناقشناه في المرة السابقة؟

كيفية تحسين الأداء لأنظمة الوكلاء التي تعمل بنظام الوكيل المدعوم من ميلفوس

عند نشر نظام وكيل مدعوم من Milvus في بيئة إنتاج، يصبح ضبط الأداء أمرًا بالغ الأهمية. لتحقيق زمن انتقال منخفض وإنتاجية عالية، هناك العديد من المعلمات الرئيسية التي تحتاج إلى الاهتمام:

1. اختيار نوع الفهرس المناسب

من المهم اختيار بنية الفهرس المناسبة. على سبيل المثال، يعد HNSW (العالم الصغير القابل للتنقل الهرمي) مثاليًا للسيناريوهات عالية الاستدعاء مثل التمويل أو الرعاية الصحية، حيث تكون الدقة أمرًا بالغ الأهمية. يعمل IVF_FLAT بشكل أفضل للتطبيقات واسعة النطاق مثل توصيات التجارة الإلكترونية، حيث يكون الاستدعاء الأقل قليلاً مقبولاً مقابل أداء أسرع واستخدام أقل للذاكرة.

2. استراتيجية التجزئة

عندما يتجاوز عدد الإرشادات المخزنة مليون مدخل، يوصى باستخدام التقسيم لتقسيم البيانات حسب مجال العمل أو حالة الاستخدام. يقلل التقسيم من مساحة البحث لكل استعلام، مما يحسن سرعة الاسترجاع ويحافظ على ثبات زمن الاستجابة حتى مع نمو مجموعة البيانات.

3. تكوين ذاكرة التخزين المؤقت

بالنسبة للمبادئ التوجيهية التي يتم الوصول إليها بشكل متكرر مثل استعلامات العملاء القياسية أو عمليات سير العمل ذات حركة المرور العالية، يمكنك استخدام التخزين المؤقت لنتائج استعلام Milvus. يسمح ذلك للنظام بإعادة استخدام النتائج السابقة، مما يقلل من زمن الاستجابة إلى أقل من 5 مللي ثانية لعمليات البحث المتكررة.

عرض توضيحي عملي: كيفية بناء نظام ذكي للأسئلة والأجوبة باستخدام Parlant و Milvus Lite

MilvusLite هو نسخة خفيفة الوزن من Milvus - مكتبة بايثون التي يمكن تضمينها بسهولة في تطبيقاتك. وهي مثالية للنماذج الأولية السريعة في بيئات مثل Jupyter Notebooks أو للتشغيل على الأجهزة المتطورة والذكية ذات الموارد الحوسبية المحدودة. على الرغم من صغر حجمه، إلا أن Milvus Lite يدعم نفس واجهات برمجة التطبيقات التي تدعمها عمليات نشر Milvus الأخرى. هذا يعني أن التعليمات البرمجية من جانب العميل التي تكتبها لـ Milvus Lite يمكن أن تتصل بسلاسة بمثيل Milvus أو Zilliz Cloud الكامل لاحقًا - دون الحاجة إلى إعادة الهيكلة.

في هذا العرض التوضيحي، سنستخدم Milvus Lite مع Parlant لتوضيح كيفية بناء نظام أسئلة وأجوبة ذكي يوفر إجابات سريعة ومدركة للسياق بأقل قدر من الإعداد.

المتطلبات الأساسية ：

1- بارلانت جيثب: https://github.com/emcie-co/parlant

2- وثائق بارلانت: https://parlant.io/docs

3.python3.10+

4.OpenAI_key

5.MlivusLite

الخطوة 1: تثبيت التبعيات

# Install required Python packages
pip install pymilvus parlant openai
# Or, if you’re using a Conda environment:
conda activate your_env_name
pip install pymilvus parlant openai

الخطوة 2: تكوين متغيرات البيئة

# Set your OpenAI API key
export OPENAI_API_KEY="your_openai_api_key_here"
# Verify that the variable is set correctly
echo $OPENAI_API_KEY

الخطوة 3: تنفيذ الكود الأساسي

إنشاء مضمن OpenAI مخصص

class OpenAIEmbedder(p.Embedder):
    # Converts text into vector embeddings with built-in timeout and retry
    # Dimension: 1536 (text-embedding-3-small)
    # Timeout: 60 seconds; Retries: up to 2 times

تهيئة القاعدة المعرفية

1- قم بإنشاء مجموعة Milvus باسم kb_articles.

2- إدراج بيانات نموذجية (مثل سياسة الاسترداد وسياسة الاستبدال ووقت الشحن).

3- إنشاء فهرس HNSW لتسريع عملية الاسترجاع.

بناء أداة البحث المتجه

@p.tool
async def vector_search(query: str, top_k: int = 5, min_score: float = 0.35):
    # 1. Convert user query into a vector
    # 2. Perform similarity search in Milvus
    # 3. Return results with relevance above threshold

تكوين وكيل بارلانت

المبدأ التوجيهي 1: بالنسبة للأسئلة المتعلقة بالوقائع أو السياسات، يجب على الوكيل أولاً إجراء بحث متجه.

المبدأ التوجيهي 2: عند العثور على دليل، يجب على الوكيل الرد باستخدام قالب منظم (ملخص + نقاط رئيسية + مصادر).

# Guideline 1: Run vector search for factual or policy-related questions
await agent.create_guideline(
            condition="User asks a factual question about policy, refund, exchange, or shipping",
            action=(
                "Call vector_search with the user's query. "
                "If evidence is found, synthesize an answer by quoting key sentences and cite doc_id/title. "
                "If evidence is insufficient, ask a clarifying question before answering."
            ),
            tools=[vector_search],
# Guideline 2: Use a standardized, structured response when evidence is available
await agent.create_guideline(
condition=“Evidence is available”,
action=(
“Answer with the following template:\n”
“Summary: provide a concise conclusion.\n”
“Key points: 2-3 bullets distilled from evidence.\n”
“Sources: list doc_id and title.\n”
“Note: if confidence is low, state limitations and ask for clarification.”
),
tools=[],
)
tools=[],

)

اكتب الرمز الكامل

import os
import asyncio
import json
from typing import List, Dict, Any
import parlant.sdk as p
from pymilvus import MilvusClient, DataType
# 1) Environment variables: using OpenAI (as both the default generation model and embedding service)
# Make sure the OPENAI_API_KEY is set
OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY")
if not OPENAI_API_KEY:
    raise RuntimeError("Please set OPENAI_API_KEY environment variable")
# 2) Initialize Milvus Lite (runs locally, no standalone service required)
# MilvusClient runs in Lite mode using a local file path (requires pymilvus >= 2.x)
client = MilvusClient("./milvus_demo.db")  # Lite mode uses a local file path
COLLECTION = "kb_articles"
# 3) Example data: three policy or FAQ entries (in practice, you can load and chunk data from files)
DOCS = [
    {"doc_id": "POLICY-001", "title": "Refund Policy", "chunk": "Refunds are available within 30 days of purchase if the product is unused."},
    {"doc_id": "POLICY-002", "title": "Exchange Policy", "chunk": "Exchanges are permitted within 15 days; original packaging required."},
    {"doc_id": "FAQ-101", "title": "Shipping Time", "chunk": "Standard shipping usually takes 3–5 business days within the country."},
]
# 4) Generate embeddings using OpenAI (you can replace this with another embedding service)
# Here we use Parlant’s built-in OpenAI embedder for simplicity, but you could also call the OpenAI SDK directly.
class OpenAIEmbedder(p.Embedder):
    async def embed(self, texts: List[str], hints: Dict[str, Any] = {}) -> p.EmbeddingResult:
        # Generate text embeddings using the OpenAI API, with timeout and retry handling
        import openai
        try:
            client = openai.AsyncOpenAI(
                api_key=OPENAI_API_KEY,
                timeout=60.0,  # 60-second timeout
                max_retries=2  # Retry up to 2 times
            )
            print(f"Generating embeddings for {len(texts)} texts...")
            response = await client.embeddings.create(
                model="text-embedding-3-small",
                input=texts
            )
            vectors = [data.embedding for data in response.data]
            print(f"Successfully generated {len(vectors)} embeddings.")
            return p.EmbeddingResult(vectors=vectors)
        except Exception as e:
            print(f"OpenAI API call failed: {e}")
            # Return mock vectors for testing Milvus connectivity
            print("Using mock vectors for testing...")
            import random
            vectors = [[random.random() for _ in range(1536)] for _ in texts]
            return p.EmbeddingResult(vectors=vectors)
    @property
    def id(self) -> str:
        return "text-embedding-3-small"
    @property
    def max_tokens(self) -> int:
        return 8192
    @property
    def tokenizer(self) -> p.EstimatingTokenizer:
        from parlant.core.nlp.tokenization import ZeroEstimatingTokenizer
        return ZeroEstimatingTokenizer()
    @property
    def dimensions(self) -> int:
        return 1536
embedder = OpenAIEmbedder()
async def ensure_collection_and_load():
    # Create the collection (schema: primary key, vector field, additional fields)
    if not client.has_collection(COLLECTION):
        client.create_collection(
            collection_name=COLLECTION,
            dimension=len((await embedder.embed(["dimension_probe"])).vectors[0]),
            # Default metric: COSINE (can be changed with metric_type="COSINE")
            auto_id=True,
        )
        # Create an index to speed up retrieval (HNSW used here as an example)
        client.create_index(
            collection_name=COLLECTION,
            field_name="vector",
            index_type="HNSW",
            metric_type="COSINE",
            params={"M": 32, "efConstruction": 200}
        )
    # Insert data (skip if already exists; simple idempotent logic for the demo)
    # Generate embeddings
    chunks = [d["chunk"] for d in DOCS]
    embedding_result = await embedder.embed(chunks)
    vectors = embedding_result.vectors
    # Check if the same doc_id already exists; this is for demo purposes only — real applications should use stricter deduplication
    # Here we insert directly. In production, use an upsert operation or an explicit primary key
    client.insert(
        COLLECTION,
        data=[
            {"vector": vectors[i], "doc_id": DOCS[i]["doc_id"], "title": DOCS[i]["title"], "chunk": DOCS[i]["chunk"]}
            for i in range(len(DOCS))
        ],
    )
    # Load into memory
    client.load_collection(COLLECTION)
# 5) Define the vector search tool (Parlant Tool)
@p.tool
async def vector_search(context: p.ToolContext, query: str, top_k: int = 5, min_score: float = 0.35) -> p.ToolResult:
    # 5.1 Generate the query vector
    embed_res = await embedder.embed([query])
    qvec = embed_res.vectors[0]
    # 5.2 Search Milvus
    results = client.search(
        collection_name=COLLECTION,
        data=[qvec],
        limit=top_k,
        output_fields=["doc_id", "title", "chunk"],
        search_params={"metric_type": "COSINE", "params": {"ef": 128}},
    )
    # 5.3 Assemble structured evidence and filter by score threshold
    hits = []
    for hit in results[0]:
        score = hit["distance"] if "distance" in hit else hit.get("score", 0.0)
        if score >= min_score:
            hits.append({
                "doc_id": hit["entity"]["doc_id"],
                "title": hit["entity"]["title"],
                "chunk": hit["entity"]["chunk"],
                "score": float(score),
            })
    return p.ToolResult({"evidence": hits})
# 6) Run Parlant Server and create the Agent + Guidelines
async def main():
    await ensure_collection_and_load()
    async with p.Server() as server:
        agent = await server.create_agent(
            name="Policy Assistant",
            description="Rule-controlled RAG assistant with Milvus Lite",
        )
        # Example variable: current time (can be used in templates or logs)
        @p.tool
        async def get_datetime(context: p.ToolContext) -> p.ToolResult:
            from datetime import datetime
            return p.ToolResult({"now": datetime.now().isoformat()})
        await agent.create_variable(name="current-datetime", tool=get_datetime)
        # Core Guideline 1: Run vector search for factual or policy-related questions
        await agent.create_guideline(
            condition="User asks a factual question about policy, refund, exchange, or shipping",
            action=(
                "Call vector_search with the user's query. "
                "If evidence is found, synthesize an answer by quoting key sentences and cite doc_id/title. "
                "If evidence is insufficient, ask a clarifying question before answering."
            ),
            tools=[vector_search],
        )
        # Core Guideline 2: Use a standardized, structured response when evidence is available
        await agent.create_guideline(
            condition="Evidence is available",
            action=(
                "Answer with the following template:\\n"
                "Summary: provide a concise conclusion.\\n"
                "Key points: 2-3 bullets distilled from evidence.\\n"
                "Sources: list doc_id and title.\\n"
                "Note: if confidence is low, state limitations and ask for clarification."
            ),
            tools=[],
        )
        # Hint: Local Playground URL
        print("Playground: <http://localhost:8800>")
if __name__ == "__main__":
    asyncio.run(main())

الخطوة 4: تشغيل الكود

# Run the main program
python main.py

قم بزيارة ساحة اللعب:

<http://localhost:8800>

لقد نجحت الآن في بناء نظام أسئلة وأجوبة ذكي باستخدام بارلانت وميلفوس.

Parlant مقابل LangChain/LlamaIndex: كيف يختلفان وكيف يعملان معاً

مقارنةً بأطر عمل الوكلاء الحاليين مثل LangChain أو LlamaIndex، كيف يختلف بارلانت عن أطر عمل الوكلاء الحاليين مثل LangChain أو LlamaIndex؟

إن LangChain و LlamaIndex هما إطارا عمل للأغراض العامة. فهي توفر مجموعة واسعة من المكونات والتكاملات، مما يجعلها مثالية للنماذج الأولية السريعة والتجارب البحثية. ومع ذلك، عندما يتعلق الأمر بالنشر في الإنتاج، غالبًا ما يحتاج المطورون إلى بناء طبقات إضافية بأنفسهم - مثل إدارة القواعد، والتحقق من الامتثال، وآليات الموثوقية - للحفاظ على اتساق الوكلاء وجدارتهم بالثقة.

يوفر Parlant إدارة المبادئ التوجيهية المدمجة، وآليات النقد الذاتي، وأدوات التوضيح التي تساعد المطورين على إدارة كيفية تصرف الوكيل واستجابته وأسبابه. هذا يجعل Parlant مناسبًا بشكل خاص لحالات الاستخدام عالية المخاطر التي تواجه العملاء حيث تكون الدقة والمساءلة مهمة، مثل الخدمات المالية والرعاية الصحية والخدمات القانونية.

في الواقع، يمكن أن تعمل هذه الأطر معًا:

استخدم LangChain لإنشاء خطوط أنابيب معقدة لمعالجة البيانات أو تدفقات عمل الاسترجاع.
استخدم Parlant لإدارة طبقة التفاعل النهائية، مما يضمن أن المخرجات تتبع قواعد العمل وتظل قابلة للتفسير.
استخدم Milvus كأساس لقاعدة بيانات المتجهات لتقديم بحث دلالي في الوقت الفعلي وذاكرة واسترجاع المعرفة عبر النظام.

الخاتمة

مع انتقال وكلاء LLM من مرحلة التجريب إلى مرحلة الإنتاج، لم يعد السؤال الرئيسي هو ما يمكنهم القيام به - بل كيف يمكنهم القيام بذلك بشكل موثوق وآمن. توفر Parlant الهيكلية والتحكم لهذه الموثوقية، بينما توفر Milvus البنية التحتية القابلة للتطوير التي تحافظ على سرعة كل شيء وتراعي السياق.

ويسمحان معًا للمطورين ببناء وكلاء ذكاء اصطناعي ليسوا فقط قادرين على القيام بذلك، بل جديرين بالثقة وقابلين للتفسير وجاهزين للإنتاج.

🚀 تحقق من Parlant على GitHub وقم بدمجه مع Milvus لبناء نظام الوكيل الذكي الخاص بك القائم على القواعد.

هل لديك أسئلة أو تريد التعمق في أي ميزة؟ انضم إلى قناة Discord الخاصة بنا أو قم بتسجيل المشكلات على GitHub. يمكنك أيضًا حجز جلسة فردية مدتها 20 دقيقة للحصول على رؤى وإرشادات وإجابات عن أسئلتك من خلال ساعات عمل Milvus المكتبية.