diff --git "a/modules/ai_finetuning/model_finetuning.py" "b/modules/ai_finetuning/model_finetuning.py" deleted file mode 100644--- "a/modules/ai_finetuning/model_finetuning.py" +++ /dev/null @@ -1,2081 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- - -""" -وحدة تخصيص وضبط نماذج الذكاء الاصطناعي للمصطلحات التعاقدية المتخصصة -تتيح هذه الوحدة إمكانية تدريب نماذج الذكاء الاصطناعي على المصطلحات المتخصصة في مجال العقود والمناقصات -""" - -import os -import sys -import streamlit as st -import pandas as pd -import numpy as np -import json -import time -import datetime -from typing import List, Dict, Any, Optional, Tuple -import openai -import matplotlib.pyplot as plt -import tempfile -import csv -import re -import random -from pathlib import Path - -# إضافة مسار النظام للوصول للملفات المشتركة -sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../.."))) - -# استيراد مكونات واجهة المستخدم -from utils.components.header import render_header -from utils.components.credits import render_credits -from utils.helpers import format_number, format_currency, styled_button - - -class ModelFinetuning: - """فئة تخصيص وضبط نماذج الذكاء الاصطناعي""" - - def __init__(self): - """تهيئة وحدة تخصيص وضبط نماذج الذكاء الاصطناعي""" - # تهيئة مجلدات حفظ البيانات - self.data_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../data/finetuning")) - os.makedirs(self.data_dir, exist_ok=True) - - # تهيئة الملفات والمجلدات الفرعية - self.training_data_dir = os.path.join(self.data_dir, "training_data") - os.makedirs(self.training_data_dir, exist_ok=True) - - self.models_dir = os.path.join(self.data_dir, "models") - os.makedirs(self.models_dir, exist_ok=True) - - self.terminology_file = os.path.join(self.data_dir, "terminology.json") - - # تهيئة حالة الجلسة - if 'terminology_data' not in st.session_state: - if os.path.exists(self.terminology_file): - with open(self.terminology_file, 'r', encoding='utf-8') as f: - st.session_state.terminology_data = json.load(f) - else: - st.session_state.terminology_data = { - "terms": [], - "training_examples": [], - "models": [] - } - - if 'active_training_job' not in st.session_state: - st.session_state.active_training_job = None - - if 'training_results' not in st.session_state: - st.session_state.training_results = [] - - # ضبط API مفاتيح الذكاء الاصطناعي - self.api_key = os.environ.get("OPENAI_API_KEY") - self.anthropic_api_key = os.environ.get("ANTHROPIC_API_KEY") - - def render(self): - """عرض واجهة وحدة تخصيص وضبط نماذج الذكاء الاصطناعي""" - # عرض الشعار والعنوان الرئيسي - render_header("تخصيص وضبط نماذج الذكاء الاصطناعي للمصطلحات التعاقدية المتخصصة") - - # تبويبات الوحدة - tabs = st.tabs([ - "قاموس المصطلحات المتخصصة", - "إعداد بيانات التدريب", - "تدريب النموذج", - "اختبار النموذج", - "المساعد المتخصص" - ]) - - # تبويب قاموس المصطلحات المتخصصة - with tabs[0]: - self._render_terminology_dictionary() - - # تبويب إعداد بيانات التدريب - with tabs[1]: - self._render_training_data_setup() - - # تبويب تدريب النموذج - with tabs[2]: - self._render_model_training() - - # تبويب اختبار النموذج - with tabs[3]: - self._render_model_testing() - - # تبويب المساعد المتخصص - with tabs[4]: - self._render_specialized_assistant() - - # عرض حقوق النشر - render_credits() - - def _render_terminology_dictionary(self): - """عرض قاموس المصطلحات المتخصصة""" - st.markdown(""" -
-

📚 قاموس المصطلحات المتخصصة

-

أضف وحرر المصطلحات الفنية المتخصصة في مجال العقود والمناقصات باللغة العربية.

-

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

-
- """, unsafe_allow_html=True) - - # إضا��ة مصطلح جديد - st.markdown("### إضافة مصطلح جديد") - - col1, col2 = st.columns(2) - - with col1: - term = st.text_input("المصطلح", key="new_term") - category = st.selectbox( - "الفئة", - options=[ - "شروط تعاقدية", "مواصفات فنية", "مستندات مناقصة", - "بنود مالية", "جداول كميات", "ضمانات", "مصطلحات قانونية", - "محتوى محلي", "أخرى" - ], - key="new_term_category" - ) - - with col2: - english_term = st.text_input("المصطلح بالإنجليزية (اختياري)", key="new_term_english") - importance = st.slider("مستوى الأهمية", 1, 5, 3, key="new_term_importance") - - definition = st.text_area("التعريف", key="new_term_definition") - examples = st.text_area("أمثلة على استخدام المصطلح (فصل بين الأمثلة بسطر جديد)", key="new_term_examples") - - # زر إضافة المصطلح - if styled_button("إضافة المصطلح", key="add_term", type="primary", icon="➕"): - if not term or not definition: - st.error("يرجى تعبئة المصطلح والتعريف على الأقل.") - else: - # إنشاء كائن المصطلح - new_term = { - "term": term, - "definition": definition, - "category": category, - "english_term": english_term, - "importance": importance, - "examples": [ex.strip() for ex in examples.split("\n") if ex.strip()], - "added_at": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") - } - - # إضافة المصطلح للقائمة - st.session_state.terminology_data["terms"].append(new_term) - - # حفظ البيانات - self._save_terminology_data() - - st.success(f"تمت إضافة المصطلح '{term}' بنجاح!") - st.rerun() - - # عرض المصطلحات الموجودة - st.markdown("### المصطلحات الموجودة") - - terms = st.session_state.terminology_data.get("terms", []) - - if not terms: - st.info("لا توجد مصطلحات مضافة. يرجى إضافة مصطلحات جديدة.") - else: - # تصفية المصطلحات - filter_col1, filter_col2 = st.columns(2) - - with filter_col1: - filter_category = st.selectbox( - "تصفية حسب الفئة", - options=["الكل"] + list(set(t.get("category") for t in terms)), - key="filter_term_category" - ) - - with filter_col2: - search_query = st.text_input("بحث", key="search_term") - - # تطبيق التصفية - filtered_terms = terms - if filter_category != "الكل": - filtered_terms = [t for t in filtered_terms if t.get("category") == filter_category] - - if search_query: - filtered_terms = [ - t for t in filtered_terms - if search_query.lower() in t.get("term", "").lower() or - search_query.lower() in t.get("definition", "").lower() or - search_query.lower() in t.get("english_term", "").lower() - ] - - # عرض المصطلحات المصفاة - if not filtered_terms: - st.warning("لا توجد مصطلحات تطابق معايير التصفية.") - else: - # إعداد بيانات للعرض - for i, term in enumerate(filtered_terms): - with st.expander(f"{term.get('term')} ({term.get('english_term', '')})", expanded=i==0 and len(filtered_terms)<5): - term_col1, term_col2 = st.columns([3, 1]) - - with term_col1: - st.markdown(f"**التعريف:** {term.get('definition')}") - st.markdown(f"**الفئة:** {term.get('category')}") - st.markdown(f"**المصطلح بالإنجليزية:** {term.get('english_term', '-')}") - - if "examples" in term and term["examples"]: - st.markdown("**أمثلة:**") - for ex in term["examples"]: - st.markdown(f"- {ex}") - - with term_col2: - st.markdown(f"**مستوى الأهمية:** {'⭐' * term.get('importance', 3)}") - st.markdown(f"**تاريخ الإضافة:** {term.get('added_at', '-')}") - - # أزرار التحرير والحذف - if styled_button("تحرير", key=f"edit_term_{i}", type="secondary", icon="✏️"): - st.session_state.term_to_edit = i - - if styled_button("حذف", key=f"delete_term_{i}", type="danger", icon="🗑️"): - st.session_state.term_to_delete = i - - # معالجة تحرير أو حذف المصطلح - if "term_to_edit" in st.session_state: - self._render_edit_term_form(st.session_state.term_to_edit, filtered_terms) - - if "term_to_delete" in st.session_state: - if st.warning(f"هل أنت متأكد من حذف المصطلح '{filtered_terms[st.session_state.term_to_delete].get('term')}'؟"): - if styled_button("تأكيد الحذف", key="confirm_delete", type="danger", icon="🗑️"): - # حذف المصطلح - term_index = terms.index(filtered_terms[st.session_state.term_to_delete]) - del st.session_state.terminology_data["terms"][term_index] - - # حفظ البيانات - self._save_terminology_data() - - # إعادة ضبط حالة الحذف - del st.session_state.term_to_delete - - st.success("تم حذف المصطلح بنجاح!") - st.rerun() - - if styled_button("إلغاء", key="cancel_delete", type="secondary", icon="❌"): - del st.session_state.term_to_delete - st.rerun() - - # تصدير المصطلحات - st.markdown("### تصدير وتوريد المصطلحات") - - col1, col2 = st.columns(2) - - with col1: - if styled_button("تصدير المصطلحات إلى CSV", key="export_terms", type="primary", icon="📤"): - self._export_terms_to_csv() - - with col2: - uploaded_file = st.file_uploader("استيراد المصطلحات من ملف CSV", type=["csv"], key="import_terms_file") - - if uploaded_file is not None: - if styled_button("استيراد المصطلحات", key="import_terms", type="success", icon="📥"): - self._import_terms_from_csv(uploaded_file) - - def _render_edit_term_form(self, term_index, terms_list): - """عرض نموذج تحرير المصطلح""" - term = terms_list[term_index] - - st.markdown("### تحرير المصطلح") - - col1, col2 = st.columns(2) - - with col1: - edited_term = st.text_input("المصطلح", value=term.get("term", ""), key="edit_term_name") - edited_category = st.selectbox( - "الفئة", - options=[ - "شروط تعاقدية", "مواصفات فنية", "مستندات مناقصة", - "بنود مالية", "جداول كميات", "ضمانات", "مصطلحات قانونية", - "محتوى محلي", "أخرى" - ], - index=["شروط تعاقدية", "مواصفات فنية", "مستندات مناقصة", "بنود مالية", "جداول كميات", "ضمانات", "مصطلحات قانونية", "محتوى محلي", "أخرى"].index(term.get("category", "أخرى")), - key="edit_term_category" - ) - - with col2: - edited_english_term = st.text_input("المصطلح بالإنجليزية (اختياري)", value=term.get("english_term", ""), key="edit_term_english") - edited_importance = st.slider("مستوى الأهمية", 1, 5, term.get("importance", 3), key="edit_term_importance") - - edited_definition = st.text_area("التعريف", value=term.get("definition", ""), key="edit_term_definition") - edited_examples = st.text_area("أمثلة على استخدام المصطلح (فصل بين الأمثلة بسطر جديد)", value="\n".join(term.get("examples", [])), key="edit_term_examples") - - col1, col2 = st.columns(2) - - with col1: - if styled_button("حفظ التغييرات", key="save_edited_term", type="primary", icon="💾"): - if not edited_term or not edited_definition: - st.error("يرجى تعبئة المصطلح والتعريف على الأقل.") - else: - # تحديث المصطلح - updated_term = { - "term": edited_term, - "definition": edited_definition, - "category": edited_category, - "english_term": edited_english_term, - "importance": edited_importance, - "examples": [ex.strip() for ex in edited_examples.split("\n") if ex.strip()], - "added_at": term.get("added_at", datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")), - "updated_at": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") - } - - # الحصول على المؤشر الفعلي في القائمة الكاملة - all_terms = st.session_state.terminology_data["terms"] - actual_index = all_terms.index(term) - - # تحديث المصطلح - st.session_state.terminology_data["terms"][actual_index] = updated_term - - # حفظ البيانات - self._save_terminology_data() - - # إعادة ضبط حالة التحرير - del st.session_state.term_to_edit - - st.success(f"تم تحديث المصطلح '{edited_term}' بنجاح!") - st.rerun() - - with col2: - if styled_button("إلغاء", key="cancel_edit_term", type="secondary", icon="❌"): - del st.session_state.term_to_edit - st.rerun() - - def _export_terms_to_csv(self): - """تصدير المصطلحات إلى ملف CSV""" - terms = st.session_state.terminology_data.get("terms", []) - - if not terms: - st.error("لا توجد مصطلحات للتصدير.") - return - - # إنشاء ملف CSV مؤقت - with tempfile.NamedTemporaryFile(mode='w+', suffix='.csv', newline='', encoding='utf-8', delete=False) as f: - writer = csv.writer(f) - - # كتابة الترويسة - writer.writerow([ - 'المصطلح', 'التعريف', 'الفئة', 'المصطلح بالإنجليزية', - 'مستوى الأهمية', 'الأمثلة', 'تاريخ الإضافة' - ]) - - # كتابة المصطلحات - for term in terms: - writer.writerow([ - term.get('term', ''), - term.get('definition', ''), - term.get('category', ''), - term.get('english_term', ''), - term.get('importance', 3), - '|'.join(term.get('examples', [])), - term.get('added_at', '') - ]) - - # الحصول على مسار الملف - csv_path = f.name - - # قراءة الملف وتقديمه للتنزيل - with open(csv_path, 'r', encoding='utf-8') as f: - csv_data = f.read() - - # تقديم الملف للتنزيل - st.download_button( - label="تنزيل ملف CSV", - data=csv_data, - file_name="terminology_dictionary.csv", - mime="text/csv" - ) - - # حذف الملف المؤقت - os.unlink(csv_path) - - def _import_terms_from_csv(self, uploaded_file): - """استيراد المصطلحات من ملف CSV""" - try: - # قراءة الملف - df = pd.read_csv(uploaded_file, encoding='utf-8') - - # التحقق من وجود الأعمدة المطلوبة - required_columns = ['المصطلح', 'التعريف'] - missing_columns = [col for col in required_columns if col not in df.columns] - - if missing_columns: - st.error(f"الملف لا يحتوي على الأعمدة التالية: {', '.join(missing_columns)}") - return - - # إضافة المصطلحات - terms_added = 0 - terms_updated = 0 - - for _, row in df.iterrows(): - term = row['المصطلح'] - - # البحث عن المصطلح الموجود - existing_term = next((t for t in st.session_state.terminology_data["terms"] if t.get("term") == term), None) - - # تحضير كائن المصطلح - term_obj = { - "term": term, - "definition": row.get('التعريف', ''), - "category": row.get('الفئة', 'أخرى'), - "english_term": row.get('المصطلح بالإنجليزية', ''), - "importance": int(row.get('مستوى الأهمية', 3)), - "examples": row.get('الأمثلة', '').split('|') if 'الأمثلة' in row else [], - "added_at": row.get('تاريخ الإضافة', datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")) - } - - if existing_term: - # تحديث المصطلح الموجود - index = st.session_state.terminology_data["terms"].index(existing_term) - st.session_state.terminology_data["terms"][index] = term_obj - terms_updated += 1 - else: - # إضافة مصطلح جديد - st.session_state.terminology_data["terms"].append(term_obj) - terms_added += 1 - - # حفظ البيانات - self._save_terminology_data() - - st.success(f"تم استيراد المصطلحات بنجاح! (تمت إضافة {terms_added} مصطلح جديد، وتحديث {terms_updated} مصطلح موجود)") - st.rerun() - - except Exception as e: - st.error(f"حدث خطأ أثناء استيراد المصطلحات: {str(e)}") - - def _render_training_data_setup(self): - """عرض إعداد بيانات التدريب""" - st.markdown(""" -
-

🔬 إعداد بيانات التدريب

-

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

-

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

-
- """, unsafe_allow_html=True) - - # تبويبات إعداد البيانات - training_tabs = st.tabs(["أمثلة التدريب الحالية", "إنشاء أمثلة يدوياً", "توليد أمثلة تلقائياً", "استيراد وتصدير البيانات"]) - - # عرض أمثلة التدريب الحالية - with training_tabs[0]: - self._render_existing_training_examples() - - # إنشاء أمثلة يدوياً - with training_tabs[1]: - self._render_manual_example_creation() - - # توليد أمثلة تلقائياً - with training_tabs[2]: - self._render_automatic_example_generation() - - # استيراد وتصدير البيانات - with training_tabs[3]: - self._render_import_export_training_data() - - def _render_existing_training_examples(self): - """عرض أمثلة التدريب الحالية""" - st.markdown("### أمثلة التدريب الحالية") - - examples = st.session_state.terminology_data.get("training_examples", []) - - if not examples: - st.info("لا توجد أمثلة تدريب. يرجى إنشاء أمثلة جديدة.") - return - - # عرض إحصائيات البيانات - st.markdown("#### إحصائيات البيانات") - - total_examples = len(examples) - categories = {} - terms_used = set() - - for ex in examples: - cat = ex.get("category", "غير مصنف") - categories[cat] = categories.get(cat, 0) + 1 - - for term in ex.get("terms", []): - terms_used.add(term) - - col1, col2, col3 = st.columns(3) - - with col1: - st.metric("إجمالي الأمثلة", total_examples) - - with col2: - st.metric("عدد المصطلحات المستخدمة", len(terms_used)) - - with col3: - st.metric("عدد الفئات", len(categories)) - - # عرض توزيع الفئات - st.markdown("#### توزيع الأمثلة حسب الفئة") - - categories_df = pd.DataFrame({ - "الفئة": list(categories.keys()), - "عدد الأمثلة": list(categories.values()) - }) - - fig, ax = plt.subplots(figsize=(10, 6)) - ax.bar(categories_df["الفئة"], categories_df["عدد الأمثلة"]) - ax.set_title("توزيع أمثلة التدريب حسب الفئة") - ax.set_xlabel("الفئة") - ax.set_ylabel("عدد الأمثلة") - - # تدوير أسماء الفئات لتسهيل القراءة - plt.xticks(rotation=45, ha='right') - plt.tight_layout() - - st.pyplot(fig) - - # تصفية الأمثلة - st.markdown("#### تصفية الأمثلة") - - filter_col1, filter_col2 = st.columns(2) - - with filter_col1: - filter_category = st.selectbox( - "تصفية حسب الفئة", - options=["الكل"] + list(categories.keys()), - key="filter_example_category" - ) - - with filter_col2: - search_query = st.text_input("بحث في النص", key="search_example") - - # تطبيق التصفية - filtered_examples = examples - if filter_category != "الكل": - filtered_examples = [ex for ex in filtered_examples if ex.get("category") == filter_category] - - if search_query: - filtered_examples = [ - ex for ex in filtered_examples - if search_query.lower() in ex.get("input", "").lower() or - search_query.lower() in ex.get("output", "").lower() - ] - - # عرض الأمثلة المصفاة - if not filtered_examples: - st.warning("لا توجد أمثلة تطابق معايير التصفية.") - else: - # عرض عدد محدود من الأمثلة في كل صفحة - examples_per_page = 10 - total_pages = (len(filtered_examples) - 1) // examples_per_page + 1 - - # التنقل بين الصفحات - col1, col2, col3 = st.columns([1, 3, 1]) - - with col2: - page = st.slider("الصفحة", 1, max(1, total_pages), 1, key="examples_page") - - start_idx = (page - 1) * examples_per_page - end_idx = min(start_idx + examples_per_page, len(filtered_examples)) - - page_examples = filtered_examples[start_idx:end_idx] - - # عرض الأمثلة - for i, example in enumerate(page_examples): - example_idx = start_idx + i - with st.expander(f"مثال #{example_idx+1} - {example.get('category', 'غير مصنف')}", expanded=i==0 and len(page_examples)<5): - ex_col1, ex_col2 = st.columns([3, 1]) - - with ex_col1: - st.markdown("**النص المدخل:**") - st.markdown(f"```\n{example.get('input', '')}\n```") - - st.markdown("**النص المتوقع:**") - st.markdown(f"```\n{example.get('output', '')}\n```") - - with ex_col2: - st.markdown("**الفئة:** " + example.get('category', 'غير مصنف')) - st.markdown("**المصطلحات المستخدمة:**") - for term in example.get("terms", []): - st.markdown(f"- {term}") - - # تاريخ الإنشاء - if "created_at" in example: - st.markdown(f"**تاريخ الإنشاء:** {example['created_at']}") - - # أزرار التحرير والحذف - if styled_button("تحرير", key=f"edit_example_{example_idx}", type="secondary", icon="✏️"): - st.session_state.example_to_edit = example_idx - - if styled_button("حذف", key=f"delete_example_{example_idx}", type="danger", icon="🗑️"): - st.session_state.example_to_delete = example_idx - - # معالجة تحرير أو حذف مثال - if "example_to_edit" in st.session_state: - self._render_edit_example_form(st.session_state.example_to_edit, filtered_examples) - - if "example_to_delete" in st.session_state: - if st.warning(f"هل أنت متأكد من حذف المثال #{st.session_state.example_to_delete+1}؟"): - if styled_button("تأكيد الحذف", key="confirm_delete_example", type="danger", icon="🗑️"): - # حذف المثال - example_index = examples.index(filtered_examples[st.session_state.example_to_delete]) - del st.session_state.terminology_data["training_examples"][example_index] - - # حفظ البيانات - self._save_terminology_data() - - # إعادة ضبط حالة الحذف - del st.session_state.example_to_delete - - st.success("تم حذف المثال بنجاح!") - st.rerun() - - if styled_button("إلغاء", key="cancel_delete_example", type="secondary", icon="❌"): - del st.session_state.example_to_delete - st.rerun() - - def _render_edit_example_form(self, example_index, examples_list): - """عرض نموذج تحرير مثال التدريب""" - example = examples_list[example_index] - - st.markdown("### تحرير مثال التدريب") - - # اختيار المصطلحات المستخدمة - all_terms = [term.get("term") for term in st.session_state.terminology_data.get("terms", [])] - selected_terms = st.multiselect( - "المصطلحات المستخدمة في المثال", - options=all_terms, - default=example.get("terms", []), - key="edit_example_terms" - ) - - # إدخال النص المدخل والمتوقع - input_text = st.text_area("النص المدخل", value=example.get("input", ""), key="edit_example_input", height=150) - output_text = st.text_area("النص المتوقع", value=example.get("output", ""), key="edit_example_output", height=150) - - # اختيار الفئة - category = st.selectbox( - "الفئة", - options=["شروط تعاقدية", "مواصفات فنية", "مستندات مناقصة", "بنود مالية", "جداول كميات", "ضمانات", "مصطلحات قانونية", "محتوى محلي", "أخرى"], - index=["شروط تعاقدية", "مواصفات فنية", "مستندات مناقصة", "بنود مالية", "جداول كميات", "ضمانات", "مصطلحات قانونية", "محتوى محلي", "أخرى"].index(example.get("category", "أخرى")), - key="edit_example_category" - ) - - col1, col2 = st.columns(2) - - with col1: - if styled_button("حفظ التغييرات", key="save_edited_example", type="primary", icon="💾"): - if not input_text or not output_text: - st.error("يرجى تعبئة النص المدخل والنص المتوقع.") - else: - # تحديث المثال - updated_example = { - "input": input_text, - "output": output_text, - "category": category, - "terms": selected_terms, - "created_at": example.get("created_at", datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")), - "updated_at": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") - } - - # الحصول على المؤشر الفعلي في القائمة الكاملة - all_examples = st.session_state.terminology_data["training_examples"] - actual_index = all_examples.index(example) - - # تحديث المثال - st.session_state.terminology_data["training_examples"][actual_index] = updated_example - - # حفظ البيانات - self._save_terminology_data() - - # إعادة ضبط حالة التحرير - del st.session_state.example_to_edit - - st.success("تم تحديث مثال التدريب بنجاح!") - st.rerun() - - with col2: - if styled_button("إلغاء", key="cancel_edit_example", type="secondary", icon="❌"): - del st.session_state.example_to_edit - st.rerun() - - def _render_manual_example_creation(self): - """عرض نموذج إنشاء أمثلة يدوياً""" - st.markdown("### إنشاء مثال تدريب جديد") - - # اختيار المصطلحات المستخدمة - all_terms = [term.get("term") for term in st.session_state.terminology_data.get("terms", [])] - selected_terms = st.multiselect( - "المصطلحات المستخدمة في المثال", - options=all_terms, - key="new_example_terms" - ) - - # اختيار الفئة - category = st.selectbox( - "الفئة", - options=["شروط تعاقدية", "مواصفات فنية", "مستندات مناقصة", "بنود مالية", "جداول كميات", "ضمانات", "مصطلحات قانونية", "محتوى محلي", "أخرى"], - key="new_example_category" - ) - - # إدخال النص المدخل والمتوقع - st.markdown("**النص المدخل** (نص السؤال أو الطلب)") - input_text = st.text_area("", key="new_example_input", height=150, placeholder="مثال: قم بشرح معنى مصطلح 'محتوى محلي' وكيفية حسابه في المشاريع الحكومية.") - - st.markdown("**النص المتوقع** (الإجابة المثالية التي يجب أن يقدمها النموذج)") - output_text = st.text_area("", key="new_example_output", height=150, placeholder="مثال: المحتوى المحلي (Local Content) هو نسبة القيمة المحلية المضافة في المنتجات والخدمات المقدمة في المشروع...") - - # عرض تعريفات المصطلحات المختارة للمساعدة - if selected_terms: - with st.expander("تعريفات المصطلحات المختارة", expanded=True): - for term_name in selected_terms: - term = next((t for t in st.session_state.terminology_data.get("terms", []) if t.get("term") == term_name), None) - if term: - st.markdown(f"**{term_name}**: {term.get('definition', '')}") - - # زر إضافة المثال - if styled_button("إضافة مثال التدريب", key="add_example", type="primary", icon="➕"): - if not input_text or not output_text: - st.error("يرجى تعبئة النص المدخل والنص المتوقع.") - else: - # إنشاء كائن المثال - new_example = { - "input": input_text, - "output": output_text, - "category": category, - "terms": selected_terms, - "created_at": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") - } - - # إضافة المثال للقائمة - if "training_examples" not in st.session_state.terminology_data: - st.session_state.terminology_data["training_examples"] = [] - - st.session_state.terminology_data["training_examples"].append(new_example) - - # حفظ البيانات - self._save_terminology_data() - - st.success("تم إضافة مثال التدريب بنجاح!") - st.rerun() - - def _render_automatic_example_generation(self): - """عرض واجهة توليد أمثلة تلقائياً""" - st.markdown("### توليد أمثلة تدريب تلقائياً") - - # التحقق من وجود مفاتيح API - if not self.api_key and not self.anthropic_api_key: - st.warning("لم يتم العثور على مفاتيح API للذكاء الاصطناعي. يرجى إضافة OPENAI_API_KEY أو ANTHROPIC_API_KEY إلى المتغيرات البيئية.") - return - - # اختيار نموذج الذكاء الاصطناعي - ai_models = [] - - if self.api_key: - ai_models.extend(["gpt-4o", "gpt-3.5-turbo"]) - - if self.anthropic_api_key: - ai_models.extend(["claude-3-7-sonnet-20250219"]) - - selected_model = st.selectbox( - "اختر نموذج الذكاء الاصطناعي", - options=ai_models, - key="auto_gen_model" - ) - - # اختيار المصطلحات لتوليد أمثلة حولها - all_terms = [term.get("term") for term in st.session_state.terminology_data.get("terms", [])] - selected_terms = st.multiselect( - "اختر المصطلحات لتوليد أمثلة حولها", - options=all_terms, - key="auto_gen_terms" - ) - - # اختيار عدد الأمثلة المراد توليدها - num_examples = st.slider("عدد الأمثلة لكل مصطلح", 1, 5, 2, key="auto_gen_count") - - # اختيار الفئات المرغوبة - selected_categories = st.multiselect( - "اختر الفئات المرغوبة للأمثلة", - options=["شروط تعاقدية", "مواصفات فنية", "مستندات مناقصة", "بنود مالية", "جداول كميات", "ضمانات", "مصطلحات قانونية", "محتوى محلي", "أخرى"], - default=["شروط تعاقدية", "مستندات مناقصة", "مصطلحات قانونية"], - key="auto_gen_categories" - ) - - # زر توليد الأمثلة - if styled_button("توليد الأمثلة", key="generate_examples", type="primary", icon="✨"): - if not selected_terms: - st.error("يرجى اختيار مصطلح واحد على الأقل.") - elif not selected_categories: - st.error("يرجى اختيار فئة واحدة على الأقل.") - else: - # عرض شريط التقدم - progress_bar = st.progress(0) - status_text = st.empty() - - # تجهيز المصطلحات وتعريفاتها - terms_with_definitions = {} - for term_name in selected_terms: - term = next((t for t in st.session_state.terminology_data.get("terms", []) if t.get("term") == term_name), None) - if term: - terms_with_definitions[term_name] = term.get('definition', '') - - # توليد الأمثلة - generated_examples = [] - total_iterations = len(selected_terms) * len(selected_categories) - current_iteration = 0 - - for term_name, definition in terms_with_definitions.items(): - for category in selected_categories: - current_iteration += 1 - progress = current_iteration / total_iterations - progress_bar.progress(progress) - status_text.text(f"جاري توليد أمثلة للمصطلح '{term_name}' في الفئة '{category}'...") - - # توليد أمثلة لهذا المصطلح والفئة - examples = self._generate_examples_with_ai( - term_name, - definition, - category, - num_examples, - selected_model - ) - - generated_examples.extend(examples) - - # إضافة الأمثلة المولدة إلى البيانات - if "training_examples" not in st.session_state.terminology_data: - st.session_state.terminology_data["training_examples"] = [] - - st.session_state.terminology_data["training_examples"].extend(generated_examples) - - # حفظ البيانات - self._save_terminology_data() - - # إكمال شريط التقدم - progress_bar.progress(1.0) - status_text.text(f"تم توليد {len(generated_examples)} مثال بنجاح!") - - st.success(f"تم توليد {len(generated_examples)} مثال بنجاح!") - st.rerun() - - def _generate_examples_with_ai(self, term_name, definition, category, num_examples, model): - """توليد أمثلة باستخدام الذكاء الاصطناعي""" - # تحضير الرسالة - prompt = f""" - أنت خبير في توليد أمثلة تدريب لضبط نماذج الذكاء الاصطناعي على المصطلحات التعاقدية والهندسية المتخصصة باللغة العربية. - - أريد منك توليد {num_examples} مثال تدريب للمصطلح التالي: - - المصطلح: {term_name} - التعريف: {definition} - الفئة: {category} - - لكل مثال، قم بتوليد: - 1. نص المدخل (سؤال أو طلب حول المصطلح) - 2. نص المخرج المتوقع (الإجابة المثالية التي يجب أن يقدمها النموذج) - - تأكد من: - - جعل الأمثلة متنوعة وواقعية - - تضمين سياقات مختلفة لاستخدام المصطلح - - استخدام أسلوب مناسب لوثائق المناقصات والعقود - - تضمين تفاصيل تقنية دقيقة عند الحاجة - - قم بإرجاع النتائج بتنسيق JSON كما يلي: - - ```json - [ - { - "input": "نص المدخل للمثال الأول", - "output": "نص المخرج المتوقع للمثال الأول" - }, - { - "input": "نص المدخل للمثال الثاني", - "output": "نص المخرج المتوقع للمثال الثاني" - }, - ... - ] - ``` - - أرجع البيانات بتنسيق JSON فقط. - """ - - try: - # استدعاء API المناسب حسب النموذج المختار - if "gpt" in model and self.api_key: - # استخدام OpenAI API - openai.api_key = self.api_key - - response = openai.chat.completions.create( - model=model, - messages=[ - {"role": "system", "content": "أنت مساعد محترف متخصص في توليد بيانات تدريب لضبط نماذج الذكاء الاصطناعي."}, - {"role": "user", "content": prompt} - ], - temperature=0.7, - response_format={"type": "json_object"} - ) - - # استخراج النتيجة - result_text = response.choices[0].message.content - - # تنظيف النص واستخراج JSON - json_match = re.search(r'```json\s*(.*?)\s*```', result_text, re.DOTALL) - if json_match: - result_json = json_match.group(1) - else: - result_json = result_text - - # تحليل JSON - examples_data = json.loads(result_json) - - # إذا كان الناتج كائن JSON بخاصية examples - if isinstance(examples_data, dict) and "examples" in examples_data: - examples_data = examples_data["examples"] - - elif "claude" in model and self.anthropic_api_key: - # استخدام Anthropic API - from anthropic import Anthropic - - anthropic_client = Anthropic(api_key=self.anthropic_api_key) - - response = anthropic_client.messages.create( - model=model, - max_tokens=4000, - messages=[ - {"role": "user", "content": prompt} - ], - temperature=0.7 - ) - - # استخراج النتيجة - result_text = response.content[0].text - - # تنظيف النص واستخراج JSON - json_match = re.search(r'```json\s*(.*?)\s*```', result_text, re.DOTALL) - if json_match: - result_json = json_match.group(1) - else: - result_json = result_text - - # تحليل JSON - examples_data = json.loads(result_json) - - # إذا كان الناتج كائن JSON بخاصية examples - if isinstance(examples_data, dict) and "examples" in examples_data: - examples_data = examples_data["examples"] - - else: - # في حالة عدم توفر النموذج المطلوب - return [] - - # تحويل البيانات إلى الصيغة المطلوبة للأمثلة - formatted_examples = [] - - for example in examples_data: - formatted_examples.append({ - "input": example["input"], - "output": example["output"], - "category": category, - "terms": [term_name], - "created_at": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), - "generated_by": model - }) - - return formatted_examples - - except Exception as e: - st.error(f"حدث خطأ أثناء توليد الأمثلة: {str(e)}") - return [] - - def _render_import_export_training_data(self): - """عرض واجهة استيراد وتصدير بيانات التدريب""" - st.markdown("### استيراد وتصدير بيانات التدريب") - - col1, col2 = st.columns(2) - - with col1: - st.markdown("#### تصدير بيانات التدريب") - - export_format = st.selectbox( - "صيغة التصدير", - options=["JSON", "JSONL", "CSV"], - key="export_format" - ) - - if styled_button("تصدير البيانات", key="export_data", type="primary", icon="📤"): - self._export_training_data(export_format) - - with col2: - st.markdown("#### استيراد بيانات التدريب") - - import_format = st.selectbox( - "صيغة الاستيراد", - options=["JSON", "JSONL", "CSV"], - key="import_format" - ) - - uploaded_file = st.file_uploader("استيراد بيانات التدريب", type=["json", "jsonl", "csv"], key="import_data_file") - - if uploaded_file is not None: - if styled_button("استيراد البيانات", key="import_data", type="success", icon="📥"): - self._import_training_data(uploaded_file, import_format) - - def _export_training_data(self, format): - """تصدير بيانات التدريب إلى ملف""" - examples = st.session_state.terminology_data.get("training_examples", []) - - if not examples: - st.error("لا توجد بيانات تدريب للتصدير.") - return - - try: - if format == "JSON": - # تصدير إلى ملف JSON - with tempfile.NamedTemporaryFile(mode='w+', suffix='.json', encoding='utf-8', delete=False) as f: - json.dump(examples, f, ensure_ascii=False, indent=2) - json_path = f.name - - # قراءة الملف وتقديمه للتنزيل - with open(json_path, 'r', encoding='utf-8') as f: - json_data = f.read() - - st.download_button( - label="تنزيل ملف JSON", - data=json_data, - file_name="training_data.json", - mime="application/json" - ) - - # حذف الملف المؤقت - os.unlink(json_path) - - elif format == "JSONL": - # تصدير إلى ملف JSONL - jsonl_content = "" - for example in examples: - jsonl_content += json.dumps(example, ensure_ascii=False) + "\n" - - st.download_button( - label="تنزيل ملف JSONL", - data=jsonl_content, - file_name="training_data.jsonl", - mime="application/jsonl" - ) - - elif format == "CSV": - # تصدير إلى ملف CSV - with tempfile.NamedTemporaryFile(mode='w+', suffix='.csv', newline='', encoding='utf-8', delete=False) as f: - writer = csv.writer(f) - - # كتابة الترويسة - writer.writerow([ - 'النص المدخل', 'النص المتوقع', 'الفئة', 'المصطلحات', 'تاريخ الإنشاء', 'تم التوليد بواسطة' - ]) - - # كتابة البيانات - for example in examples: - writer.writerow([ - example.get('input', ''), - example.get('output', ''), - example.get('category', ''), - '|'.join(example.get('terms', [])), - example.get('created_at', ''), - example.get('generated_by', 'يدوي') - ]) - - # الحصول على مسار الملف - csv_path = f.name - - # قراءة الملف وتقديمه للتنزيل - with open(csv_path, 'r', encoding='utf-8') as f: - csv_data = f.read() - - st.download_button( - label="تنزيل ملف CSV", - data=csv_data, - file_name="training_data.csv", - mime="text/csv" - ) - - # حذف الملف المؤقت - os.unlink(csv_path) - - except Exception as e: - st.error(f"حدث خطأ أثناء تصدير البيانات: {str(e)}") - - def _import_training_data(self, uploaded_file, format): - """استيراد بيانات التدريب من ملف""" - try: - examples = [] - - if format == "JSON": - # استيراد من ملف JSON - content = uploaded_file.read().decode('utf-8') - examples = json.loads(content) - - elif format == "JSONL": - # استيراد من ملف JSONL - content = uploaded_file.read().decode('utf-8') - - for line in content.strip().split('\n'): - if line.strip(): - examples.append(json.loads(line)) - - elif format == "CSV": - # استيراد من ملف CSV - df = pd.read_csv(uploaded_file, encoding='utf-8') - - for _, row in df.iterrows(): - example = { - "input": row.get('النص المدخل', ''), - "output": row.get('النص المتوقع', ''), - "category": row.get('الفئة', 'أخرى'), - "terms": row.get('المصطلحات', '').split('|') if pd.notna(row.get('المصطلحات', '')) else [], - "created_at": row.get('تاريخ الإنشاء', datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")), - "generated_by": row.get('تم التوليد بواسطة', 'مستورد') - } - - examples.append(example) - - # إضافة الأمثلة المستوردة - if "training_examples" not in st.session_state.terminology_data: - st.session_state.terminology_data["training_examples"] = [] - - # فلترة الأمثلة الصحيحة - valid_examples = [] - for ex in examples: - if "input" in ex and "output" in ex: - valid_examples.append(ex) - - # إضافة الأمثلة وحفظ البيانات - if valid_examples: - st.session_state.terminology_data["training_examples"].extend(valid_examples) - self._save_terminology_data() - - st.success(f"تم استيراد {len(valid_examples)} مثال بنجاح!") - st.rerun() - else: - st.error("لم يتم العثور على أمثلة صالحة في الملف.") - - except Exception as e: - st.error(f"حدث خطأ أثناء استيراد البيانات: {str(e)}") - - def _render_model_training(self): - """عرض واجهة تدريب النموذج""" - st.markdown(""" -
-

🧠 تدريب النموذج

-

قم بتدريب نموذج الذكاء الاصطناعي على المصطلحات المتخصصة باستخدام أمثلة التدريب.

-

يمكنك اختيار النموذج الأساسي والإعدادات المناسبة لعملية التدريب.

-
- """, unsafe_allow_html=True) - - # التحقق من وجود بيانات تدريب كافية - examples = st.session_state.terminology_data.get("training_examples", []) - if len(examples) < 10: - st.warning(f"عدد أمثلة التدريب الحالية ({len(examples)}) غير كافٍ للتدريب. يُنصح بوجود 10 أمثلة على الأقل.") - - # تبويبات تدريب النموذج - training_tabs = st.tabs(["إعداد التدريب", "نماذج سابقة", "وظائف التدريب النشطة"]) - - # تبويب إعداد التدريب - with training_tabs[0]: - self._render_training_setup() - - # تبويب النماذج السابقة - with training_tabs[1]: - self._render_previous_models() - - # تبويب وظائف التدريب النشطة - with training_tabs[2]: - self._render_active_training_jobs() - - def _render_training_setup(self): - """عرض إعدادات تدريب النموذج""" - st.markdown("### إعداد عملية التدريب") - - # التحقق من وجود مفاتيح API - if not self.api_key and not self.anthropic_api_key: - st.warning("لم يتم العثور على مفاتيح API للذكاء الاصطناعي. يرجى إضافة OPENAI_API_KEY أو ANTHROPIC_API_KEY إلى المتغيرات البيئية.") - return - - # اختيار نموذج الذكاء الاصطناعي الأساسي - provider_options = [] - if self.api_key: - provider_options.append("OpenAI") - if self.anthropic_api_key: - provider_options.append("Anthropic") - - provider = st.selectbox( - "مزود الذكاء الاصطناعي", - options=provider_options, - key="training_provider" - ) - - # الإعدادات حسب المزود - if provider == "OpenAI": - # نماذج OpenAI المتاحة للضبط - base_model = st.selectbox( - "النموذج الأساسي", - options=["gpt-3.5-turbo-0125", "gpt-4o-mini"], - key="openai_base_model" - ) - - # إعدادات التدريب - col1, col2 = st.columns(2) - - with col1: - n_epochs = st.slider("عدد الحقب (Epochs)", 1, 4, 2, key="openai_epochs") - batch_size = st.selectbox("حجم الدفعة (Batch Size)", options=[1, 2, 4, 8], index=1, key="openai_batch_size") - - with col2: - learning_rate_multiplier = st.slider("مضاعف معدل التعلم", 0.1, 2.0, 1.0, 0.1, key="openai_lr") - suffix = st.text_input("لاحقة اسم النموذج", value="arabic-contracts-expert", key="openai_suffix") - - # زر بدء التدريب - if styled_button("بدء التدريب", key="start_openai_training", type="primary", icon="🚀"): - # التحقق من وجود بيانات كافية - examples = st.session_state.terminology_data.get("training_examples", []) - if len(examples) < 10: - st.error("عدد أ��ثلة التدريب الحالية قليل جداً. يُفضل وجود على الأقل 10 أمثلة للتدريب.") - else: - # التأكيد على بدء التدريب - confirm = st.warning(f"سيتم بدء عملية تدريب نموذج {base_model} باستخدام {len(examples)} مثال. هل أنت متأكد؟") - if styled_button("تأكيد بدء التدريب", key="confirm_openai_training", type="success", icon="✅"): - # توجيه بيانات التدريب لصيغة OpenAI - formatted_data = self._format_training_data_for_openai(examples) - - # بدء التدريب - self._start_openai_training( - base_model=base_model, - training_data=formatted_data, - n_epochs=n_epochs, - batch_size=batch_size, - learning_rate_multiplier=learning_rate_multiplier, - suffix=suffix - ) - - elif provider == "Anthropic": - st.info("ضبط نماذج Anthropic غير متاح حالياً في واجهة البرمجة العامة. يمكنك استخدام أمثلة التدريب مع المساعد المتخصص.") - - def _format_training_data_for_openai(self, examples): - """تنسيق بيانات التدريب لواجهة برمجة OpenAI""" - formatted_examples = [] - - for example in examples: - formatted_examples.append({ - "messages": [ - {"role": "user", "content": example.get("input", "")}, - {"role": "assistant", "content": example.get("output", "")} - ] - }) - - return formatted_examples - - def _start_openai_training(self, base_model, training_data, n_epochs, batch_size, learning_rate_multiplier, suffix): - """بدء عملية تدريب نموذج OpenAI""" - try: - # تهيئة واجهة برمجة OpenAI - openai.api_key = self.api_key - - # إنشاء ملف تدريب - training_file_path = os.path.join(self.training_data_dir, f"training_data_{int(time.time())}.jsonl") - - with open(training_file_path, 'w', encoding='utf-8') as f: - for example in training_data: - f.write(json.dumps(example, ensure_ascii=False) + "\n") - - # رفع ملف التدريب إلى OpenAI - with open(training_file_path, 'rb') as f: - response = openai.files.create( - file=f, - purpose="fine-tune" - ) - - file_id = response.id - - # بدء وظيفة التدريب - response = openai.fine_tuning.jobs.create( - training_file=file_id, - model=base_model, - hyperparameters={ - "n_epochs": n_epochs, - "batch_size": batch_size, - "learning_rate_multiplier": learning_rate_multiplier - }, - suffix=suffix - ) - - job_id = response.id - - # تخزين معلومات وظيفة التدريب - training_job = { - "job_id": job_id, - "provider": "OpenAI", - "base_model": base_model, - "n_epochs": n_epochs, - "batch_size": batch_size, - "learning_rate_multiplier": learning_rate_multiplier, - "suffix": suffix, - "status": "running", - "file_id": file_id, - "file_path": training_file_path, - "examples_count": len(training_data), - "started_at": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), - "finished_at": None, - "fine_tuned_model": None - } - - # إضافة الوظيفة إلى حالة الجلسة - st.session_state.active_training_job = training_job - - # إضافة الوظيفة إلى قائمة وظائف التدريب - if "training_jobs" not in st.session_state.terminology_data: - st.session_state.terminology_data["training_jobs"] = [] - - st.session_state.terminology_data["training_jobs"].append(training_job) - - # حفظ البيانات - self._save_terminology_data() - - st.success(f"تم بدء وظيفة التدريب بنجاح! معرف الوظيفة: {job_id}") - st.info("يمكنك متابعة حالة التدريب من تبويب 'وظائف التدريب النشطة'.") - - except Exception as e: - st.error(f"حدث خطأ أثناء بدء عملية التدريب: {str(e)}") - - def _render_previous_models(self): - """عرض النماذج المدربة سابقاً""" - st.markdown("### النماذج المدربة سابقاً") - - # الحصول على النماذج المدربة - models = st.session_state.terminology_data.get("models", []) - - if not models: - st.info("لا توجد نماذج مدربة سابقاً.") - return - - # عرض النماذج - for i, model in enumerate(models): - with st.expander(f"{model.get('model_id')} - {model.get('base_model')}", expanded=i==0): - col1, col2 = st.columns([3, 1]) - - with col1: - st.markdown(f"**معرف النموذج:** {model.get('model_id')}") - st.markdown(f"**النموذج الأساسي:** {model.get('base_model')}") - st.markdown(f"**عدد أمثلة التدريب:** {model.get('examples_count')}") - st.markdown(f"**تاريخ الإنشاء:** {model.get('created_at')}") - - # عرض مؤشرات الأداء إن وجدت - if "metrics" in model: - st.markdown("#### مؤشرات الأداء") - - metrics = model.get("metrics", {}) - for metric_name, metric_value in metrics.items(): - st.markdown(f"**{metric_name}:** {metric_value}") - - with col2: - # أزرار الاستخدام والحذف - if styled_button("استخدام النموذج", key=f"use_model_{i}", type="primary", icon="✅"): - st.session_state.selected_model = model.get('model_id') - st.success(f"تم اختيار النموذج {model.get('model_id')} للاستخدام.") - - if styled_button("حذف النموذج", key=f"delete_model_{i}", type="danger", icon="🗑️"): - st.session_state.model_to_delete = i - - # عرض الوصف والملاحظات - st.markdown(f"**الوصف:** {model.get('description', 'لا يوجد وصف.')}") - - # عرض النماذج المستخدمة في التدريب - if "examples_preview" in model and model["examples_preview"]: - with st.expander("عينة من أمثلة التدريب"): - for j, example in enumerate(model["examples_preview"]): - st.markdown(f"**مثال #{j+1}**") - st.markdown(f"**المدخل:** {example.get('input')}") - st.markdown(f"**المخرج:** {example.get('output')}") - st.markdown("---") - - # معالجة حذف النموذج - if "model_to_delete" in st.session_state: - if st.warning(f"هل أنت متأكد من حذف النموذج '{models[st.session_state.model_to_delete].get('model_id')}'؟"): - if styled_button("تأكيد الحذف", key="confirm_delete_model", type="danger", icon="🗑️"): - # حذف النموذج - del st.session_state.terminology_data["models"][st.session_state.model_to_delete] - - # حفظ البيانات - self._save_terminology_data() - - # إعادة ضبط حالة الحذف - del st.session_state.model_to_delete - - st.success("تم حذف النموذج بنجاح!") - st.rerun() - - if styled_button("إلغاء", key="cancel_delete_model", type="secondary", icon="❌"): - del st.session_state.model_to_delete - st.rerun() - - def _render_active_training_jobs(self): - """عرض وظائف التدريب النشطة""" - st.markdown("### وظائف التدريب النشطة") - - # الحصول على وظائف التدريب - jobs = st.session_state.terminology_data.get("training_jobs", []) - - # فرز الوظائف حسب الحالة - active_jobs = [job for job in jobs if job.get("status") in ["running", "validating_files", "queued"]] - completed_jobs = [job for job in jobs if job.get("status") == "succeeded"] - failed_jobs = [job for job in jobs if job.get("status") in ["failed", "cancelled"]] - - # زر تحديث حالة الوظائف - if styled_button("تحديث حالة الوظائف", key="refresh_jobs", type="primary", icon="🔄"): - self._refresh_training_jobs_status() - - # عرض الوظائف النشطة - if active_jobs: - st.markdown("#### الوظائف النشطة") - - for i, job in enumerate(active_jobs): - with st.expander(f"{job.get('job_id')} - {job.get('base_model')}", expanded=True): - col1, col2 = st.columns([3, 1]) - - with col1: - st.markdown(f"**معرف الوظيفة:** {job.get('job_id')}") - st.markdown(f"**النموذج الأساسي:** {job.get('base_model')}") - st.markdown(f"**الحالة:** {job.get('status')}") - st.markdown(f"**تاريخ البدء:** {job.get('started_at')}") - - # عرض تقدم التدريب إن وجد - if "progress" in job: - progress = job.get("progress", 0) - st.progress(progress) - st.markdown(f"**التقدم:** {progress*100:.1f}%") - - with col2: - # زر إلغاء الوظيفة - if styled_button("إلغاء الوظيفة", key=f"cancel_job_{i}", type="danger", icon="⛔"): - st.session_state.job_to_cancel = i - - # عرض معلومات إضافية - st.markdown(f"**عدد الحقب:** {job.get('n_epochs')}") - st.markdown(f"**حجم الدفعة:** {job.get('batch_size')}") - st.markdown(f"**مضاعف معدل التعلم:** {job.get('learning_rate_multiplier')}") - else: - st.info("لا توجد وظائف تدريب نشطة حالياً.") - - # عرض الوظائف المكتملة - if completed_jobs: - st.markdown("#### الوظائف المكتملة") - - for i, job in enumerate(completed_jobs): - with st.expander(f"{job.get('job_id')} - {job.get('base_model')}", expanded=False): - col1, col2 = st.columns([3, 1]) - - with col1: - st.markdown(f"**معرف الوظيفة:** {job.get('job_id')}") - st.markdown(f"**النموذج الأساسي:** {job.get('base_model')}") - st.markdown(f"**تاريخ البدء:** {job.get('started_at')}") - st.markdown(f"**تاريخ الانتهاء:** {job.get('finished_at')}") - st.markdown(f"**النموذج المدرب:** {job.get('fine_tuned_model')}") - - with col2: - # زر استخدام النموذج المدرب - if styled_button("استخدام النموذج", key=f"use_trained_model_{i}", type="primary", icon="✅"): - st.session_state.selected_model = job.get('fine_tuned_model') - st.success(f"تم اختيار النموذج {job.get('fine_tuned_model')} للاستخدام.") - - # زر حذف الوظيفة - if styled_button("حذف الوظيفة", key=f"delete_completed_job_{i}", type="danger", icon="🗑️"): - st.session_state.completed_job_to_delete = len(active_jobs) + i - - # عرض الوظائف الفاشلة - if failed_jobs: - st.markdown("#### الوظائف الفاشلة") - - for i, job in enumerate(failed_jobs): - with st.expander(f"{job.get('job_id')} - {job.get('base_model')}", expanded=False): - col1, col2 = st.columns([3, 1]) - - with col1: - st.markdown(f"**معرف الوظيفة:** {job.get('job_id')}") - st.markdown(f"**النموذج الأساسي:** {job.get('base_model')}") - st.markdown(f"**الحالة:** {job.get('status')}") - st.markdown(f"**تاريخ البدء:** {job.get('started_at')}") - - # عرض سبب الفشل إن وجد - if "error" in job: - st.error(f"سبب الفشل: {job.get('error')}") - - with col2: - # زر حذف الوظيفة - if styled_button("حذف الوظيفة", key=f"delete_failed_job_{i}", type="danger", icon="🗑️"): - st.session_state.failed_job_to_delete = len(active_jobs) + len(completed_jobs) + i - - # معالجة إلغاء الوظيفة - if "job_to_cancel" in st.session_state: - if st.warning(f"هل أنت م��أكد من إلغاء وظيفة التدريب '{active_jobs[st.session_state.job_to_cancel].get('job_id')}'؟"): - if styled_button("تأكيد الإلغاء", key="confirm_cancel_job", type="danger", icon="🗑️"): - # إلغاء الوظيفة - self._cancel_training_job(active_jobs[st.session_state.job_to_cancel]) - - # إعادة ضبط حالة الإلغاء - del st.session_state.job_to_cancel - - st.success("تم إلغاء وظيفة التدريب بنجاح!") - st.rerun() - - if styled_button("إلغاء", key="cancel_job_cancellation", type="secondary", icon="❌"): - del st.session_state.job_to_cancel - st.rerun() - - # معالجة حذف الوظائف المكتملة - if "completed_job_to_delete" in st.session_state: - idx = st.session_state.completed_job_to_delete - if 0 <= idx < len(jobs): - if st.warning(f"هل أنت متأكد من حذف وظيفة التدريب '{jobs[idx].get('job_id')}'؟"): - if styled_button("تأكيد الحذف", key="confirm_delete_completed_job", type="danger", icon="🗑️"): - # حذف الوظيفة - del st.session_state.terminology_data["training_jobs"][idx] - - # حفظ البيانات - self._save_terminology_data() - - # إعادة ضبط حالة الحذف - del st.session_state.completed_job_to_delete - - st.success("تم حذف وظيفة التدريب بنجاح!") - st.rerun() - - if styled_button("إلغاء", key="cancel_completed_job_deletion", type="secondary", icon="❌"): - del st.session_state.completed_job_to_delete - st.rerun() - - # معالجة حذف الوظائف الفاشلة - if "failed_job_to_delete" in st.session_state: - idx = st.session_state.failed_job_to_delete - if 0 <= idx < len(jobs): - if st.warning(f"هل أنت متأكد من حذف وظيفة التدريب '{jobs[idx].get('job_id')}'؟"): - if styled_button("تأكيد الحذف", key="confirm_delete_failed_job", type="danger", icon="🗑️"): - # حذف الوظيفة - del st.session_state.terminology_data["training_jobs"][idx] - - # حفظ البيانات - self._save_terminology_data() - - # إعادة ضبط حالة الحذف - del st.session_state.failed_job_to_delete - - st.success("تم حذف وظيفة التدريب بنجاح!") - st.rerun() - - if styled_button("إلغاء", key="cancel_failed_job_deletion", type="secondary", icon="❌"): - del st.session_state.failed_job_to_delete - st.rerun() - - def _refresh_training_jobs_status(self): - """تحديث حالة وظائف التدريب""" - jobs = st.session_state.terminology_data.get("training_jobs", []) - - # فلترة الوظائف النشطة - active_jobs = [job for job in jobs if job.get("status") in ["running", "validating_files", "queued"]] - - if not active_jobs: - st.info("لا توجد وظائف تدريب نشطة للتحديث.") - return - - try: - # تحديث حالة كل وظيفة نشطة - for job in active_jobs: - if job.get("provider") == "OpenAI" and self.api_key: - # تحديث حالة وظيفة OpenAI - job_id = job.get("job_id") - - # استعلام عن حالة الوظيفة - openai.api_key = self.api_key - response = openai.fine_tuning.jobs.retrieve(job_id) - - # تحديث حالة الوظيفة - job["status"] = response.status - - # تحديث التقدم إن وجد - if hasattr(response, "progress") and response.progress is not None: - job["progress"] = response.progress - - # إذا اكتملت الوظيفة، تحديث معلومات النموذج المدرب - if response.status == "succeeded": - job["finished_at"] = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") - job["fine_tuned_model"] = response.fine_tuned_model - - # إضافة النموذج المدرب إلى قائمة النماذج - self._add_trained_model(job, response) - - # إذا فشلت الوظيفة، تسجيل سبب الفشل - elif response.status == "failed" and hasattr(response, "error"): - job["error"] = response.error - - # حفظ البيانات - self._save_terminology_data() - - st.success("تم تحديث حالة وظائف التدريب بنجاح!") - - except Exception as e: - st.error(f"حدث خطأ أثناء تحديث حالة وظائف التدريب: {str(e)}") - - def _cancel_training_job(self, job): - """إلغاء وظيفة تدريب""" - try: - if job.get("provider") == "OpenAI" and self.api_key: - # إلغاء وظيفة OpenAI - job_id = job.get("job_id") - - # استدعاء واجهة برمجة OpenAI - openai.api_key = self.api_key - openai.fine_tuning.jobs.cancel(job_id) - - # تحديث حالة الوظيفة - job["status"] = "cancelled" - job["finished_at"] = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") - - # حفظ البيانات - self._save_terminology_data() - - return True - - return False - - except Exception as e: - st.error(f"حدث خطأ أثناء إلغاء وظيفة التدريب: {str(e)}") - return False - - def _add_trained_model(self, job, response): - """إضافة النموذج المدرب إلى قائمة النماذج""" - # إنشاء كائن النموذج - model = { - "model_id": response.fine_tuned_model, - "base_model": job.get("base_model"), - "provider": job.get("provider"), - "training_job_id": job.get("job_id"), - "description": f"النموذج المدرب على المصطلحات المتخصصة في {job.get('suffix')}", - "examples_count": job.get("examples_count"), - "created_at": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), - "metrics": {} - } - - # إضافة مؤشرات الأداء إن وجدت - if hasattr(response, "result_files") and response.result_files: - # تنزيل ملف النتائج وقراءة مؤشرات الأداء - pass - - # إضافة عينة من أمثلة التدريب - examples = st.session_state.terminology_data.get("training_examples", []) - if examples: - # أخذ 5 أمثلة كعينة - sample_examples = random.sample(examples, min(5, len(examples))) - model["examples_preview"] = sample_examples - - # إضافة النموذج إلى قائمة النماذج - if "models" not in st.session_state.terminology_data: - st.session_state.terminology_data["models"] = [] - - st.session_state.terminology_data["models"].append(model) - - def _render_model_testing(self): - """عرض واجهة اختبار النموذج""" - st.markdown(""" -
-

🧪 اختبار النموذج

-

اختبر نموذج الذكاء الاصطناعي المدرب على المصطلحات المتخصصة.

-

يمكنك تجريب أسئلة مختلفة ومقارنة النتائج مع النماذج الأخرى.

-
- """, unsafe_allow_html=True) - - # التحقق من وجود مفاتيح API - if not self.api_key and not self.anthropic_api_key: - st.warning("لم يتم العثور على مفاتيح API للذكاء الاصطناعي. يرجى إضافة OPENAI_API_KEY أو ANTHROPIC_API_KEY إلى المتغيرات البيئية.") - return - - # الحصول على قائمة النماذج المتاحة - available_models = [] - - # OpenAI - if self.api_key: - available_models.extend(["gpt-4o", "gpt-3.5-turbo"]) - - # إضافة النماذج المدربة إن وجدت - for model in st.session_state.terminology_data.get("models", []): - if model.get("provider") == "OpenAI": - available_models.append(model.get("model_id")) - - # Anthropic - if self.anthropic_api_key: - available_models.extend(["claude-3-7-sonnet-20250219"]) - - # اختيار النموذج - selected_model = st.selectbox( - "اختر النموذج", - options=available_models, - key="test_model" - ) - - # إدخال النص للاختبار - test_input = st.text_area( - "أدخل نص الاختبار", - value="ما هو مفهوم المحتوى المحلي في المشاريع الحكومية وكيف يتم حسابه؟", - height=150, - key="test_input" - ) - - # خيارات متقدمة - with st.expander("خيارات متقدمة"): - temperature = st.slider("درجة الإبداعية (Temperature)", 0.0, 1.0, 0.7, 0.1, key="test_temperature") - max_tokens = st.slider("الحد الأقصى للرموز (Max Tokens)", 100, 2000, 500, 100, key="test_max_tokens") - - # تحميل المصطلحات والتعريفات - terms_with_definitions = {} - for term in st.session_state.terminology_data.get("terms", []): - terms_with_definitions[term.get("term")] = term.get("definition") - - # زر إجراء الاختبار - if styled_button("إجراء الاختبار", key="run_test", type="primary", icon="🧪"): - if not test_input: - st.error("يرجى إدخال نص للاختبار.") - else: - # عرض شريط التقدم - with st.spinner("جاري معالجة النص..."): - # إجراء الاختبار - response = self._test_model( - model=selected_model, - input_text=test_input, - temperature=temperature, - max_tokens=max_tokens, - terms_with_definitions=terms_with_definitions - ) - - # عرض النتيجة - st.markdown("### نتيجة الاختبار") - st.markdown(response) - - # تحليل الاستجابة لاكتشاف المصطلحات المستخدمة - used_terms = [] - for term in terms_with_definitions: - if term in response: - used_terms.append(term) - - if used_terms: - st.markdown("### المصطلحات المكتشفة في الاستجابة") - for term in used_terms: - st.markdown(f"- **{term}**: {terms_with_definitions[term]}") - - def _test_model(self, model, input_text, temperature, max_tokens, terms_with_definitions): - """اختبار النموذج""" - try: - # تجهيز المحتوى النظامي - system_prompt = "أنت مساعد متخصص في عقود المقاولات والمناقصات باللغة العربية. قم بالإجابة بدقة على الأسئلة والطلبات مع مراعاة المصطلحات الفنية المتخصصة." - - # إضافة المصطلحات إلى المحتوى النظامي - if terms_with_definitions: - system_prompt += "\n\nفيما يلي قائمة بالمصطلحات المتخصصة وتعريفاتها:\n\n" - for term, definition in terms_with_definitions.items(): - system_prompt += f"- {term}: {definition}\n" - - # OpenAI - if "gpt" in model or any(model_data.get("model_id") == model for model_data in st.session_state.terminology_data.get("models", [])): - # استخدام OpenAI API - openai.api_key = self.api_key - - response = openai.chat.completions.create( - model=model, - messages=[ - {"role": "system", "content": system_prompt}, - {"role": "user", "content": input_text} - ], - temperature=temperature, - max_tokens=max_tokens - ) - - return response.choices[0].message.content - - # Anthropic - elif "claude" in model and self.anthropic_api_key: - # استخدام Anthropic API - from anthropic import Anthropic - - anthropic_client = Anthropic(api_key=self.anthropic_api_key) - - response = anthropic_client.messages.create( - model=model, - max_tokens=max_tokens, - messages=[ - {"role": "system", "content": system_prompt}, - {"role": "user", "content": input_text} - ], - temperature=temperature - ) - - return response.content[0].text - - else: - return "النموذج المختار غير مدعوم حالياً." - - except Exception as e: - return f"حدث خطأ أثناء اختبار النموذج: {str(e)}" - - def _render_specialized_assistant(self): - """عرض واجهة المساعد المتخصص""" - st.markdown(""" -
-

🤖 المساعد المتخصص

-

استخدم المساعد الذكي المتخصص في المصطلحات التعاقدية الهندسية.

-

يمكنك طرح أسئلة حول المصطلحات وتفسيراتها واستخداماتها.

-
- """, unsafe_allow_html=True) - - # التحقق من وجود مفاتيح API - if not self.api_key and not self.anthropic_api_key: - st.warning("لم يتم العثور على مفاتيح API للذكاء الاصطناعي. يرجى إضافة OPENAI_API_KEY أو ANTHROPIC_API_KEY إلى المتغيرات البيئية.") - return - - # الحصول على قائمة النماذج المتاحة - available_models = [] - - # OpenAI - if self.api_key: - available_models.extend(["gpt-4o", "gpt-3.5-turbo"]) - - # إضافة النماذج المدربة إن وجدت - for model in st.session_state.terminology_data.get("models", []): - if model.get("provider") == "OpenAI": - available_models.append(model.get("model_id")) - - # Anthropic - if self.anthropic_api_key: - available_models.extend(["claude-3-7-sonnet-20250219"]) - - # تهيئة حالة المحادثة - if "chat_history" not in st.session_state: - st.session_state.chat_history = [] - - if "assistant_model" not in st.session_state: - st.session_state.assistant_model = available_models[0] if available_models else "" - - # اختيار النموذج - selected_model = st.selectbox( - "اختر نموذج المساعد", - options=available_models, - index=available_models.index(st.session_state.assistant_model) if st.session_state.assistant_model in available_models else 0, - key="assistant_model_selector" - ) - - # تحديث النموذج المختار - if selected_model != st.session_state.assistant_model: - st.session_state.assistant_model = selected_model - st.rerun() - - # عرض المحادثة - st.markdown("### المحادثة") - - for message in st.session_state.chat_history: - if message["role"] == "user": - st.markdown(f""" -
- أنت: {message["content"]} -
- """, unsafe_allow_html=True) - else: - st.markdown(f""" -
- المساعد: {message["content"]} -
- """, unsafe_allow_html=True) - - # إدخال رسالة جديدة - user_input = st.text_area("اكتب رسالتك هنا", key="assistant_input", height=100) - - # أزرار التحكم - col1, col2, col3 = st.columns([1, 1, 1]) - - with col1: - if styled_button("إرسال", key="send_message", type="primary", icon="✉️"): - if not user_input: - st.error("يرجى كتابة رسالة للإرسال.") - else: - # إضافة رسالة المستخدم إلى المحادثة - st.session_state.chat_history.append({ - "role": "user", - "content": user_input - }) - - # الحصول على الرد من النموذج - with st.spinner("المساعد يفكر..."): - # تحميل المصطلحات والتعريفات - terms_with_definitions = {} - for term in st.session_state.terminology_data.get("terms", []): - terms_with_definitions[term.get("term")] = term.get("definition") - - # الحصول على رد المساعد - response = self._get_assistant_response( - model=st.session_state.assistant_model, - chat_history=st.session_state.chat_history, - terms_with_definitions=terms_with_definitions - ) - - # إضافة رد المساعد إلى المحادثة - st.session_state.chat_history.append({ - "role": "assistant", - "content": response - }) - - # إعادة تشغيل لتحديث واجهة المستخدم - st.rerun() - - with col2: - if styled_button("مسح المحادثة", key="clear_chat", type="danger", icon="🗑️"): - st.session_state.chat_history = [] - st.rerun() - - with col3: - if styled_button("اقتراح أسئلة", key="suggest_questions", type="secondary", icon="💡"): - # عرض أسئلة مقترحة - st.markdown("### أسئلة مقترحة") - - # الحصول على المصطلحات المتاحة - terms = [term.get("term") for term in st.session_state.terminology_data.get("terms", [])] - - # إنشاء أسئلة مقترحة - suggested_questions = [ - f"ما هو تعريف مصطلح {term}؟" for term in random.sample(terms, min(3, len(terms))) - ] - - suggested_questions.extend([ - "كيف يتم حساب المحتوى المحلي في مشاريع البنية التحتية؟", - "ما هي أهم البنود التي يجب الانتباه لها في العقود الهندسية؟", - "ما الفرق بين الضمان الابتدائي والضمان النهائي؟", - "كيف يمكن تقييم المخاطر في مشاريع البناء؟" - ]) - - # عرض الأسئلة المقترحة - for i, question in enumerate(suggested_questions): - if styled_button(question, key=f"suggested_q_{i}", type="info", icon="❓"): - # إضافة السؤال المقترح إلى المحادثة - st.session_state.chat_history.append({ - "role": "user", - "content": question - }) - - # الحصول على الرد من النموذج - with st.spinner("المساعد يفكر..."): - # تحميل المصطلحات والتعريفات - terms_with_definitions = {} - for term in st.session_state.terminology_data.get("terms", []): - terms_with_definitions[term.get("term")] = term.get("definition") - - # الحصول على رد المساعد - response = self._get_assistant_response( - model=st.session_state.assistant_model, - chat_history=st.session_state.chat_history, - terms_with_definitions=terms_with_definitions - ) - - # إضافة رد المساعد إلى المحادثة - st.session_state.chat_history.append({ - "role": "assistant", - "content": response - }) - - # إعادة تشغيل لتحديث واجهة المستخدم - st.rerun() - - def _get_assistant_response(self, model, chat_history, terms_with_definitions): - """الحصول على رد المساعد المتخصص""" - try: - # تجهيز المحتوى النظامي - system_prompt = """أنت مساعد متخصص في عقود المقاولات والمناقصات باللغة العربية. - مهمتك هي تقديم معلومات دقيقة ومفصلة حول المصطلحات التعاقدية والهندسية المتخصصة. - قدم شرحاً واضحاً ومفهوماً للمصطلحات، مع أمثلة عملية عند الإمكان. - استخدم لغة مهنية ودقيقة، مع مراعاة السياق الهندسي والقانوني للمصطلحات.""" - - # إضافة المصطلحات إلى المحتوى النظامي - if terms_with_definitions: - system_prompt += "\n\nفيما يلي قائمة بالمصطلحات المتخصصة وتعريفاتها التي يجب عليك استخدامها في إجاباتك:\n\n" - for term, definition in terms_with_definitions.items(): - system_prompt += f"- {term}: {definition}\n" - - # تحويل محادثة streamlit إلى صيغة مناسبة للـ API - messages = [{"role": "system", "content": system_prompt}] - - for msg in chat_history: - messages.append({"role": msg["role"], "content": msg["content"]}) - - # OpenAI - if "gpt" in model or any(model_data.get("model_id") == model for model_data in st.session_state.terminology_data.get("models", [])): - # استخدام OpenAI API - openai.api_key = self.api_key - - response = openai.chat.completions.create( - model=model, - messages=messages, - temperature=0.7 - ) - - return response.choices[0].message.content - - # Anthropic - elif "claude" in model and self.anthropic_api_key: - # استخدام Anthropic API - from anthropic import Anthropic - - # تعديل الرسائل لتتناسب مع صيغة Anthropic - anthropic_messages = [] - for msg in messages[1:]: # تخطي رسالة النظام - anthropic_messages.append({"role": msg["role"], "content": msg["content"]}) - - anthropic_client = Anthropic(api_key=self.anthropic_api_key) - - response = anthropic_client.messages.create( - model=model, - max_tokens=2000, - system=system_prompt, - messages=anthropic_messages, - temperature=0.7 - ) - - return response.content[0].text - - else: - return "النموذج المختار غير مدعوم حالياً." - - except Exception as e: - return f"عذراً، حدث خطأ أثناء معالجة طلبك: {str(e)}" - - def _save_terminology_data(self): - """حفظ بيانات المصطلحات""" - try: - # التأكد من وجود المجلد - os.makedirs(os.path.dirname(self.terminology_file), exist_ok=True) - - # حفظ البيانات - with open(self.terminology_file, 'w', encoding='utf-8') as f: - json.dump(st.session_state.terminology_data, f, ensure_ascii=False, indent=2) - - except Exception as e: - st.error(f"حدث خطأ أثناء حفظ البيانات: {str(e)}") - - -# تشغيل النموذج بشكل مستقل -def main(): - """تشغيل وحدة تخصيص وضبط نماذج الذكاء الاصطناعي بشكل مستقل""" - # تهيئة الواجهة - st.set_page_config( - page_title="تخصيص وضبط نماذج الذكاء الاصطناعي | WAHBi AI", - page_icon="🧠", - layout="wide", - initial_sidebar_state="expanded", - menu_items={ - 'Get Help': 'mailto:support@wahbi-ai.com', - 'Report a bug': 'mailto:support@wahbi-ai.com', - 'About': 'وحدة تخصيص وضبط نماذج الذكاء الاصطناعي للمصطلحات التعاقدية المتخصصة - جزء من نظام WAHBi AI لتحليل المناقصات' - } - ) - - # تهيئة وحدة الضبط - model_finetuning = ModelFinetuning() - - # عرض واجهة الوحدة - model_finetuning.render() - -# تشغيل النموذج عند استدعاء الملف مباشرة -if __name__ == "__main__": - main() \ No newline at end of file