استيراد بيانات التتبع من Langfuse Cloud

كيفية تحميل بيانات تتبع نماذج LLM المُصدَّرة من Langfuse Cloud إلى ClickHouse Cloud باستخدام ClickPipes.

يشرح هذا الدليل خطوة بخطوة كيفية تحميل بيانات تتبع نماذج LLM من Langfuse Cloud إلى ClickHouse Cloud لإجراء التحليلات في الوقت الفعلي.

​

التصدير من Langfuse Cloud

اتبع هذا الدليل لتهيئة تكامل Blob Storage في Langfuse Cloud بهدف إجراء عمليات تصدير مجدولة. يمكنك جدولة تصدير بيانات التتبع إلى التخزين الكائني بوتيرة تصل إلى مرة كل 20 minutes، أو وفق جدول hourly أو daily أو weekly. ويتضمن كل تصدير افتراضيًا observations الخاصة بك (مُثراةً بسمات التتبع) وscores.

​

بنية الملفات المُصدَّرة

يكتب تكامل Blob Storage في Langfuse Cloud البيانات إلى الحاوية (bucket) أو حاوية التخزين (storage container) الهدف وفق بنية الأدلة التالية:

{prefix}{project-id}/
├── observations_v2/   # observations enriched with trace attributes
├── scores/            # scores
└── manifests/         # one JSON file per completed run with export metadata

​

الاستيراد إلى ClickHouse Cloud

بعد تهيئة تكامل Blob Storage في Langfuse Cloud، ستُصدَّر بيانات التتبع دوريًا وفق جدول زمني إلى الحاوية (bucket) أو حاوية التخزين الهدف. ولاستيراد هذه البيانات إلى ClickHouse Cloud، يمكنك استخدام ClickPipes وموصلات التخزين الكائني المُدارة التي يوفرها:

مصدر البيانات	التفاصيل
Amazon S3 وخدمات التخزين الكائني المتوافقة مع S3	التكامل باستخدام S3 ClickPipes.
Google Cloud Storage	التكامل باستخدام GCS ClickPipes.
Azure Blob Storage	التكامل باستخدام ABS ClickPipes.

​

إنشاء ClickPipe

يستخدم هذا المثال S3 ClickPipe لاستيراد البيانات من حاوية S3 تستقبل عمليات تصدير مجدولة من Langfuse Cloud، ويمكن اعتماده نموذجًا لجميع وجهات التصدير إلى التخزين الكائني. وللاطلاع على إرشادات خاصة بكل مصدر بيانات، راجع وثائق ClickPipes.

1. اختر مصدر البيانات

   1. في ClickHouse Cloud، اختر Data sources من قائمة التنقل الرئيسية، ثم انقر على Create ClickPipe.

   

   2. انقر على بطاقة Amazon S3. ويمكنك أيضًا استخدام هذه البطاقة للاتصال بخدمات أخرى متوافقة مع S3 غير مدرجة في واجهة ClickPipes.

   
2. إعداد الاتصال

   1. املأ التفاصيل التي تحتاجها ClickPipes للاتصال بالحاوية التي تستقبل الملفات المُصدَّرة من Langfuse Cloud والمصادقة عليها.

   • Authentication method: يدعم S3 ClickPipe بيانات اعتماد IAM (Credentials) والمصادقة المستندة إلى دور IAM (IAM role). ولا تحتاج ClickPipes إلا إلى صلاحية القراءة على الحاوية. راجع الوثائق المرجعية للاطلاع على إرشادات المصادقة والأذونات.

   • S3 file path: وجّه ClickPipe إلى دليل فرعي، واستخدم حرف البدل * لمطابقة جميع الملفات المُصدَّرة. يجب إنشاء ClickPipe منفصل لكل دليل فرعي، نظرًا لاختلاف مخططات الملفات المُصدَّرة من دليل فرعي إلى آخر.

     الدليل الفرعي	مسار الملف
     observations_v2/	https://{bucket-name}.s3.{region-code}.amazonaws.com/{prefix}{project-id}/observations_v2/*
     scores/	https://{bucket-name}.s3.{region-code}.amazonaws.com/{prefix}{project-id}/scores/*

     راجع الوثائق المرجعية للاطلاع على إرشادات حول الأنماط المدعومة، بما في ذلك كيفية مطابقة الملفات عبر البادئات المتداخلة.

   2. اختر Continuous ingestion لإدخال الملفات الجديدة تلقائيًا فور كتابة Langfuse لها في حاوية الهدف.

   

   3. انقر على Incoming data. ستجلب ClickPipes البيانات الوصفية من حاويتك، وتستنتج مخطط جدول الهدف في الخطوة التالية.

3. التحقق من صحة البيانات الواردة

   1. ستتصل ClickPipes بحاوية التخزين الخاصة بك وتعرض الملفات الموجودة في المسار المحدد، مع استنتاج صيغة الملفات تلقائيًا. استخدمنا في هذا المثال صيغة ملفات التصدير الافتراضية في Langfuse Cloud ‏(Parquet).

   

   2. انقر على Parse information. ستستخدم ClickPipes ملفًا نموذجيًا لاستنتاج مخطط البيانات المُصدَّرة، ثم تربط حقول المصدر تلقائيًا بجدول الهدف في ClickHouse.

4. تهيئة الهدف

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

   

   ضمن Upload data to، أبقِ الخيار New table محددًا واضبط ما يلي:

   • Database وName: قاعدة بيانات مخصصة لبيانات Langfuse (مثل langfuse)، واسم واضح للجدول الوجهة (مثل scores). تستخدم ClickPipes افتراضيًا الاسم default.s3-<uuid> لتجنب تعارض الأسماء.

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

     الدليل الفرعي	الجدول الوجهة	مفتاح الفرز المقترح
     observations_v2/	observations_v2	environment, start_time, trace_id, id
     scores/	scores	environment, name, timestamp, trace_id, observation_id, id

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

   • Partition by: اتركه فارغًا ما لم تكن تنوي حذف بيانات التتبّع القديمة بعد انتهاء صلاحيتها. يُستخدم التقسيم في ClickHouse لإدارة البيانات لا لتحسين الاستعلامات؛ إذ إن مفتاح الفرز المقترح يقلّص أصلًا نطاق البيانات المقروءة في استعلامات الفترات الزمنية. أما إذا كنت بحاجة إلى سياسة احتفاظ، فقسّم حسب عمود الوقت (toYYYYMM(start_time)، أو toYYYYMM(timestamp) بالنسبة إلى scores)، بحيث يصبح حذف شهر كامل من بيانات التتبّع عملية DROP PARTITION واحدة.

   2. بعد ذلك، وسّع Advanced settings واضبط Engine على ReplacingMergeTree لضمان إزالة التكرار، مع استخدام updated_at عمودًا لـ Version.

   

   تشمل نوافذ التصدير في تكامل Langfuse Cloud Blob Storage حدَّيها كليهما، لذا قد يظهر السجل نفسه في أكثر من ملف مُصدَّر. ما لم تكن أنماط استعلاماتك التحليلية متساوية القوى (مثل uniq() وmax() وmin())، يجب عليك معالجة إزالة التكرار ضمن نموذج البيانات اللاحق، باستخدام ReplacingMergeTree محركًا لجدول الهدف.

   إزالة التكرار باستخدام ReplacingMergeTree

   يدمج محرك الجدول هذا السجلات المكررة عبر عمليات الدمج في الخلفية. ولإزالة التكرار بالكامل عند القراءة (أي اعتماد مبدأ الدمج عند القراءة)، يجب استخدام المُعدِّل FINAL (أو تجميعات من نوع argMax()) عند تنفيذ الاستعلام.

5. تهيئة الأذونات

   تُنشئ ClickPipes مستخدمًا مخصصًا لكتابة البيانات في جدول الوجهة. اختر دورًا لهذا المستخدم الداخلي، إما دورًا مخصصًا أو أحد الأدوار المُعرَّفة مسبقًا:

   • Full access: يمنح صلاحية وصول كاملة إلى العنقود. وهو مطلوب إذا كنت تستخدم عرضًا مُجسَّدًا أو قاموسًا مع جدول الوجهة.
   • Only destination table: يمنح صلاحيات INSERT على جدول الوجهة فقط.

   

   انقر على Complete setup لإنشاء ClickPipe.

6. أكمل الإعداد

   وبهذا يكون كل شيء جاهزًا! ستبدأ ClickPipes أولاً بتعبئة البيانات التاريخية (backfill) لجميع الملفات الموجودة في المسار المحدد، ثم تنتقل إلى استيعاب الملفات الجديدة فور وصولها إلى الحاوية.

   

   كرّر الخطوات الواردة في هذا الشرح لكل دليل فرعي تريد استيراده إلى ClickHouse Cloud.

​

الاستعلام عن بيانات التتبع

بعد أن تبدأ ClickPipes بالعمل، يمكنك الاستعلام عن بيانات التتبع الخاصة بك في Langfuse Cloud مباشرةً من ClickHouse Cloud. تتناول الأمثلة أدناه بعض أنماط الاستعلام الشائعة: التجميعات الإجمالية للتكلفة وزمن الاستجابة، وإعادة بناء تتبع واحد، وربط الدرجات بالنماذج والموجّهات.

​

تجميعات المقاييس ضمن فترات زمنية

تُصدَّر بعض الحقول، مثل usage_details وcost_details، كأعمدة من نوع Map. للوصول إلى قيمة واحدة، ابحث عنها بمفتاحها (مثلًا: sum(cost_details['total'])).

SELECT
    toStartOfHour(start_time) AS t,
    provided_model_name,
    count()                        AS spans,
    quantile(0.95)(latency)        AS p95_latency,
    sum(total_cost)                AS cost,
    sum(usage_details['total'])    AS tokens
FROM observations_v2 FINAL
WHERE start_time >= now() - INTERVAL 7 DAY
-- AND environment = '<environment>'
GROUP BY t, provided_model_name
ORDER BY t;

​

إعادة بناء التتبّع

SELECT *
FROM observations_v2 FINAL
WHERE trace_id = '<trace_id>'
ORDER BY start_time;

​

تحليل الجودة

SELECT
    o.provided_model_name,
    s.name,
    avg(s.value) AS avg_score
FROM "s3-26237f63-84dc-4a7d-a0e1-7361cf55ee0b" AS s FINAL
JOIN "s3-80899034-ce1e-4dc9-a4e6-b059ddec4175" AS o FINAL ON s.trace_id = o.trace_id
--WHERE s.environment = '<environment>'
GROUP BY o.provided_model_name, s.name
ORDER BY avg_score DESC;

​

انظر أيضًا

• Langfuse وClickHouse
• التصدير من Langfuse Cloud إلى Blob Storage
• ClickPipes للتخزين الكائني