LLMChat
v1.1.3.4.1
مرحبًا بك في مستودع LLMCHAT ، وهو تطبيق كامل لخادم API تم تصميمه باستخدام Python fastapi ، وواجهة أمامية جميلة مدعومة بالرفرف. تم تصميم هذا المشروع لتقديم تجربة دردشة سلسة مع طرازات ChatGPT المتقدمة وغيرها من طرز LLM. ؟ تقديم بنية تحتية حديثة يمكن تمديدها بسهولة عندما تتوفر ميزات GPT-4 المتعددة الوسائط والمكونات الإضافية. استمتع بإقامتك!
mobile PC .Markdown أيضًا ، بحيث يمكنك استخدامها لتنسيق رسائلك.يمكنك استخدام محرك بحث DuckDuckgo للعثور على المعلومات ذات الصلة على الويب. فقط قم بتفعيل زر تبديل "تصفح"!
شاهد الفيديو التجريبي للشرف الكامل: https://www.youtube.com/watch؟v=MJ_CVRWRS08
باستخدام أمر /embed ، يمكنك تخزين النص إلى أجل غير مسمى في قاعدة بيانات المتجه الخاصة بك والاستعلام عنها لاحقًا ، في أي وقت. إذا كنت تستخدم أمر /share ، يتم تخزين النص في قاعدة بيانات المتجهات العامة التي يمكن للجميع مشاركتها. يساعد تمكين زر تبديل Query أو /query أمر AI على إنشاء إجابات سياقية من خلال البحث عن أوجه التشابه في النص في قواعد البيانات العامة والخاصة. هذا يحل واحدة من أكبر قيود نماذج اللغة: الذاكرة .
يمكنك تضمين ملف PDF من خلال النقر على Embed Document في أسفل اليسار. في بضع ثوان ، سيتم تحويل محتويات النصوص من PDF إلى متجهات ومضمنة على ذاكرة التخزين المؤقت Redis.
LLMModels الموجود في app/models/llms.py بالنسبة إلى LLALAM LLMS المحلي ، يُفترض أنه يعمل فقط في البيئة المحلية ويستخدم http://localhost:8002/v1/completions نقطة نهاية. إنه يتحقق باستمرار من حالة خادم LLAMA API من خلال الاتصال بـ http://localhost:8002/health مرة واحدة واحدة لمعرفة ما إذا كان يتم إرجاع استجابة 200 OK ، وإذا لم يكن الأمر كذلك ، فإنه يقوم تلقائيًا بتشغيل عملية منفصلة لإنشاء خادم API.
الهدف الرئيسي من llama.cpp هو تشغيل نموذج LLAMA باستخدام كمية GGML 4 بت مع تنفيذ C/C ++ العادي دون تبعيات. يجب عليك تنزيل ملف GGML bin من Huggingface ووضعه في مجلد llama_models/ggml ، وتحديد Llmmodel في app/models/llms.py . هناك أمثلة قليلة ، بحيث يمكنك بسهولة تحديد النموذج الخاص بك. ارجع إلى مستودع llama.cpp لمزيد من المعلومات: https://github.com/ggerganov/llama.cpp
تنفيذ Python/C ++/CUDA المستقل من LLAMA للاستخدام مع أوزان GPTQ 4 بت ، مصمم ليكون سريعًا وفعالًا للذاكرة على وحدات معالجة الرسومات الحديثة. ويستخدم pytorch و sentencepiece لتشغيل النموذج. من المفترض أن تعمل فقط في البيئة المحلية ، ويحتاج NVIDIA CUDA GPU على الأقل. يجب عليك تنزيل ملفات Tokenizer و Config و GPTQ من Huggingface ووضعها في مجلد llama_models/gptq/YOUR_MODEL_FOLDER ، وتحديد Llmmodel في app/models/llms.py . هناك أمثلة قليلة ، بحيث يمكنك بسهولة تحديد النموذج الخاص بك. ارجع إلى مستودع exllama للحصول على معلومات أكثر تفصيلاً: https://github.com/turboderp/exllama
web framework عالي الأداء لبناء واجهات برمجة التطبيقات مع Python.Webapp Frontend مع واجهة مستخدم جميلة ومجموعة غنية من الأدوات القابلة للتخصيص.OpenAI API لتوليد النص وإدارة الرسائل.LlamaCpp و Exllama .Real-time ، والاتصال ثنائي الاتجاه مع ChatGPT ، ونماذج LLM الأخرى ، مع Flutter Frontend WebApp.Redis و Langchain ، تخزين واسترداد التضمينات المتجهات للبحث عن التشابه. سوف يساعد الذكاء الاصطناعي على توليد المزيد من الاستجابات ذات الصلة.Duckduckgo ، وتصفح الويب والعثور على المعلومات ذات الصلة.async / await للتزامن والتوازي.MySQL . قم بسهولة بإنشاء الإجراءات والقراءة والتحديث وحذفها ، باستخدام sqlalchemy.asyncioRedis مع Aioredis. قم بسهولة بإنشاء الإجراءات والقراءة والتحديث وحذفها باستخدام aioredis . لإعداد الجهاز المحلي الخاص بك ، اتبع هذه الخطوات البسيطة. قبل أن تبدأ ، تأكد من تثبيت docker و docker-compose على جهازك. إذا كنت ترغب في تشغيل الخادم بدون Docker ، فيجب عليك تثبيت Python 3.11 بالإضافة إلى ذلك. على الرغم من أنك تحتاج إلى Docker لتشغيل خوادم DB.
لاستنساخ النطاق الفرعي بشكل متكرر لاستخدام نماذج Exllama أو llama.cpp ، استخدم الأمر التالي:
git clone --recurse-submodules https://github.com/c0sogi/llmchat.gitتريد فقط استخدام الميزات الأساسية (Openai) ، استخدم الأمر التالي:
git clone https://github.com/c0sogi/llmchat.git cd LLMChat.env قم بإعداد ملف ENV ، في إشارة إلى ملف .env-sample . أدخل معلومات قاعدة البيانات لإنشاء مفتاح API Openai ، وغيرها من التكوينات اللازمة. لا يلزم الاختيارات ، فقط اتركهم كما هم.
تنفيذ هذه. قد يستغرق الأمر بضع دقائق لبدء الخادم لأول مرة:
docker-compose -f docker-compose-local.yaml updocker-compose -f docker-compose-local.yaml down يمكنك الآن الوصول إلى الخادم على http://localhost:8000/docs وقاعدة البيانات على db:3306 أو cache:6379 . يمكنك أيضًا الوصول إلى التطبيق على http://localhost:8000/chat .
لتشغيل الخادم بدون Docker إذا كنت ترغب في تشغيل الخادم بدون Docker ، يجب عليك تثبيت Python 3.11 بالإضافة إلى ذلك. على الرغم من أنك تحتاج إلى Docker لتشغيل خوادم DB. قم بإيقاف تشغيل خادم API الذي يعمل بالفعل باستخدام docker-compose -f docker-compose-local.yaml down api . لا تنس تشغيل خوادم DB أخرى على Docker! ثم ، قم بتشغيل الأوامر التالية:
python -m main يجب أن يكون الخادم الخاص بك الآن قيد التشغيل على http://localhost:8001 في هذه الحالة.
تم ترخيص هذا المشروع بموجب ترخيص MIT ، والذي يسمح بالاستخدام المجاني والتعديل والتوزيع ، طالما يتم تضمين حقوق الطبع والنشر الأصلية وإشعار الترخيص في أي نسخة أو جزء كبير من البرنامج.
FastAPI هو إطار عمل حديث على شبكة الإنترنت لبناء واجهات برمجة التطبيقات مع Python. ؟ إنه ذو أداء عالي وسهل التعلم وسريع إلى رمز وجاهز للإنتاج. ؟ واحدة من الميزات الرئيسية لـ FastAPI هي أنه يدعم بناء الجملة async / await . ؟ هذا يعني أنه يمكنك كتابة التعليمات البرمجية التي يمكنها التعامل مع مهام متعددة في نفس الوقت دون منع بعضها البعض ، خاصة عند التعامل مع عمليات الإدخال/الإخراج ، مثل طلبات الشبكة ، واستعلامات قاعدة البيانات ، وعمليات الملفات ، إلخ.
Flutter عبارة عن مجموعة أدوات واجهة المستخدم مفتوحة المصدر تم تطويرها بواسطة Google لإنشاء واجهات المستخدم الأصلية للهاتف المحمول والواقع وسطح المكتب من قاعدة كود واحدة. يستخدم Dart ، وهي لغة برمجة موجهة نحو الكائنات الحديثة ، وتوفر مجموعة غنية من الأدوات المصغرة التي يمكن التخصيص التي يمكن أن تتكيف مع أي تصميم.
يمكنك الوصول إلى ChatGPT أو LlamaCpp من خلال اتصال WebSocket باستخدام وحدتين: app/routers/websocket و app/utils/chat/chat_stream_manager . تسهل هذه الوحدات التواصل بين عميل Flutter ونموذج الدردشة من خلال WebSocket. مع WebSocket ، يمكنك إنشاء قناة اتصال في الوقت الفعلي للتفاعل مع LLM.
لبدء محادثة ، قم بالتواصل مع WebSocket Route /ws/chat/{api_key} باستخدام مفتاح API صالح مسجل في قاعدة البيانات. لاحظ أن مفتاح API هذا ليس هو نفسه مفتاح Openai API ، ولكنه متاح فقط لخادمك للتحقق من صحة المستخدم. بمجرد الاتصال ، يمكنك إرسال الرسائل والأوامر للتفاعل مع طراز LLM. سيرسل WebSocket ردود الدردشة في الوقت الفعلي. يتم إنشاء اتصال WebSocket عبر تطبيق Flutter ، والذي يمكن الوصول إليه باستخدام نقطة النهاية /chat .
websocket.py مسؤول عن إعداد اتصال WebSocket ومعالجة مصادقة المستخدم. إنه يحدد WebSocket Route /chat/{api_key} الذي يقبل WebSocket ومفتاح API كمعلمات.
عندما يتصل العميل بـ WebSocket ، يقوم أولاً بالتحقق من مفتاح API لمصادقة المستخدم. إذا كان مفتاح API صالحًا ، فسيتم استدعاء وظيفة begin_chat() من وحدة stream_manager.py لبدء المحادثة.
في حالة وجود مفتاح API غير المسجل أو خطأ غير متوقع ، يتم إرسال رسالة مناسبة إلى العميل ويتم إغلاق الاتصال.
@ router . websocket ( "/chat/{api_key}" )
async def ws_chat ( websocket : WebSocket , api_key : str ):
... stream_manager.py مسؤول عن إدارة محادثة ومعالجة رسائل المستخدم. إنه يحدد وظيفة begin_chat() ، التي تأخذ websocket ، معرف المستخدم كمعلمات.
تهيئة الوظيفة أولاً سياق دردشة المستخدم من مدير ذاكرة التخزين المؤقت. بعد ذلك ، يرسل سجل الرسالة الأولي إلى العميل من خلال WebSocket.
تستمر المحادثة في حلقة حتى يتم إغلاق الاتصال. أثناء المحادثة ، تتم معالجة رسائل المستخدم ويتم إنشاء استجابات GPT وفقًا لذلك.
class ChatStreamManager :
@ classmethod
async def begin_chat ( cls , websocket : WebSocket , user : Users ) -> None :
... يتم استخدام فئة SendToWebsocket لإرسال الرسائل والتدفقات إلى WebSocket. لديها طريقتان: message() و stream() . ترسل طريقة message() رسالة كاملة إلى WebSocket ، بينما ترسل طريقة stream() دفقًا إلى WebSocket.
class SendToWebsocket :
@ staticmethod
async def message (...):
...
@ staticmethod
async def stream (...):
... تتولى فئة MessageHandler أيضًا استجابات الذكاء الاصطناعي. ترسل طريقة ai() استجابة الذكاء الاصطناعي إلى WebSocket. إذا تم تمكين الترجمة ، يتم ترجمة الاستجابة باستخدام Google Translate API قبل إرسالها إلى العميل.
class MessageHandler :
...
@ staticmethod
async def ai (...):
... تتم معالجة رسائل المستخدم باستخدام فئة HandleMessage . إذا بدأت الرسالة بـ / ، مثل /YOUR_CALLBACK_NAME . يتم التعامل معها كأمر ويتم إنشاء استجابة الأوامر المناسبة. خلاف ذلك ، تتم معالجة رسالة المستخدم وإرسالها إلى نموذج LLM لإنشاء استجابة.
يتم التعامل مع الأوامر باستخدام فئة ChatCommands . ينفذ وظيفة رد الاتصال المقابلة اعتمادًا على الأمر. يمكنك إضافة أوامر جديدة عن طريق إضافة رد اتصال في فئة ChatCommands من app.utils.chat.chat_commands .
باستخدام redis لتخزين تضمينات المتجهات للمحادثات؟ ️ هل يمكن أن يساعد نموذج ChatGPT؟ بعدة طرق ، مثل الاسترجاع الفعال والسريع لسياق المحادثة ♀ ، التعامل مع كميات كبيرة من البيانات ، وتوفير استجابات أكثر صلة من خلال البحث في المتجه؟
بعض الأمثلة الممتعة حول كيفية عمل هذا في الممارسة العملية:
/embed عندما يدخل المستخدم أمرًا في نافذة الدردشة مثل /embed <text_to_embed> ، يتم استدعاء طريقة VectorStoreManager.create_documents . تقوم هذه الطريقة بتحويل نص الإدخال إلى متجه باستخدام نموذج Openai text-embedding-ada-002 ويخزنه في Redis VectorStore.
@ staticmethod
@ command_response . send_message_and_stop
async def embed ( text_to_embed : str , / , buffer : BufferedUserContext ) -> str :
"""Embed the text and save its vectors in the redis vectorstore. n
/embed <text_to_embed>"""
.../query عندما يدخل المستخدم إلى الأمر /query <query> ، يتم استخدام وظيفة asimilarity_search للعثور على ما يصل إلى ثلاث نتائج مع أعلى تشابه متجه مع البيانات المدمجة في Redis VectorStore. يتم تخزين هذه النتائج مؤقتًا في سياق الدردشة ، مما يساعد منظمة العفو الدولية على الإجابة على الاستعلام من خلال الإشارة إلى هذه البيانات.
@ staticmethod
async def query ( query : str , / , buffer : BufferedUserContext , ** kwargs ) -> Tuple [ str | None , ResponseType ]:
"""Query from redis vectorstore n
/query <query>"""
... عند تشغيل وظيفة begin_chat ، إذا قام المستخدم بتحميل ملف يحتوي على نص (على سبيل المثال ، ملف PDF أو TXT) ، يتم استخراج النص تلقائيًا من الملف ، ويتم حفظ التضمين المتجه الخاص به على redis.
@ classmethod
async def embed_file_to_vectorstore ( cls , file : bytes , filename : str , collection_name : str ) -> str :
# if user uploads file, embed it
...commands.py في ملف commands.py ، هناك العديد من المكونات المهمة:
command_response : يتم استخدام هذه الفئة لتعيين ديكور على طريقة الأوامر لتحديد الإجراء التالي. يساعد على تحديد أنواع الاستجابة المختلفة ، مثل إرسال رسالة وإيقاف ، وإرسال رسالة ، ومعالجة إدخال المستخدم ، والتعامل مع ردود الذكاء الاصطناعي ، والمزيد.command_handler : هذه الوظيفة مسؤولة عن تنفيذ طريقة رد الاتصال على الأمر بناءً على النص الذي أدخله المستخدم.arguments_provider : توفر هذه الوظيفة تلقائيًا الوسائط المطلوبة بواسطة طريقة الأمر بناءً على نوع التعليقات التوضيحية لطريقة الأمر.تشغيل المهمة : يتم تنشيط هذه الميزة كلما قام المستخدم بتكوين رسالة أو يستجيب الذكاء الاصطناعي برسالة. في هذه المرحلة ، يتم إنشاء مهمة تلخيص تلقائي لتكثيف محتوى النص.
تخزين المهمة : يتم تخزين مهمة Summarization Auto في سمة task_list من BufferUserChatContext . هذا بمثابة قائمة انتظار لإدارة المهام المرتبطة بسياق دردشة المستخدم.
حصاد المهام : بعد الانتهاء من دورة سؤال ودورة الإجابة عن المستخدم من قِبل MessageHandler ، يتم استدعاء وظيفة harvest_done_tasks . تجمع هذه الوظيفة نتائج مهمة التلخيص ، مع التأكد من عدم استبعاد أي شيء.
تطبيق تلخيص : بعد عملية الحصاد ، تحل النتائج الملخصة الرسالة الفعلية عندما يطلب chatbot لدينا إجابات من نماذج تعلم اللغة (LLMS) ، مثل Openai و Llama_CPP. من خلال القيام بذلك ، نحن قادرون على إرسال مطالبات موجزة أكثر من الرسالة الطويلة الأولية.
تجربة المستخدم : الأهم من ذلك ، من وجهة نظر المستخدم ، يرون فقط الرسالة الأصلية. لا يتم عرض النسخة الملخصة من الرسالة لهم ، مع الحفاظ على الشفافية وتجنب الالتباس المحتمل.
المهام المتزامنة : ميزة رئيسية أخرى لمهمة التحويل التلقائي هذه هي أنها لا تعرقل المهام الأخرى. بمعنى آخر ، في حين أن chatbot مشغول بتلخيص النص ، لا يزال من الممكن تنفيذ المهام الأخرى ، وبالتالي تحسين الكفاءة الإجمالية لـ chatbot لدينا.
ChatConfig . يحتوي هذا المستودع على نماذج LLM مختلفة ، محددة في llms.py ترث كل فئة من طراز LLM من الفئة الأساسية LLMModel . تعداد LLMModels هي مجموعة من هذه LLMS.
يتم التعامل مع جميع العمليات بشكل غير متزامن دون تدخل الخيط الرئيسي. ومع ذلك ، لا تتمكن LLMs المحلية من التعامل مع طلبات متعددة في نفس الوقت ، لأنها باهظة الثمن للغاية. لذلك ، يتم استخدام Semaphore للحد من عدد الطلبات إلى 1.
نموذج LLM الافتراضي الذي يستخدمه المستخدم عبر UserChatContext.construct_default هو gpt-3.5-turbo . يمكنك تغيير الافتراضي لهذه الوظيفة.
يقوم OpenAIModel بإنشاء نص غير متزامن عن طريق طلب إكمال الدردشة من خادم Openai. يتطلب مفتاح API Openai.
يقرأ LlamaCppModel نموذج GGML مخزّن محليًا. يجب وضع نموذج llama.cpp ggml في مجلد llama_models/ggml كملف .bin . على سبيل المثال ، إذا قمت بتنزيل نموذج كمي Q4_0 من "https://huggingface.co/thebloke/robin-7b-v2-ggml" ، يجب أن يكون مسار النموذج "robin-7b.ggmlv3.q4_0.bin".
ExllamaModel قراءة نموذج GPTQ مخزّن محليًا. يجب وضع نموذج Exllama GPTQ في مجلد llama_models/gptq كمجلد. على سبيل المثال ، إذا قمت بتنزيل 3 ملفات من "https://huggingface.co/thebloke/orca_mini_7b-gptq/tree/main":
ثم تحتاج إلى وضعها في مجلد. يجب أن يكون مسار النموذج هو اسم المجلد. دعنا نقول ، "orca_mini_7b" ، الذي يحتوي على الملفات الثلاثة.
التعامل مع الاستثناءات التي قد تحدث أثناء توليد النص. إذا تم طرح ChatLengthException ، فإنه يقوم تلقائيًا بإجراء روتين لإعادة تحديد الرسالة إلى عدد الرموز المميزة المحدودة بواسطة وظيفة cutoff_message_histories ، وإعادة تقديمها. هذا يضمن أن لدى المستخدم تجربة دردشة سلسة بغض النظر عن الحد المميز.
يهدف هذا المشروع إلى إنشاء واجهة برمجة تطبيقات لتمكين خدمة chatbot الطراز الكبير. إنه يستخدم مدير ذاكرة التخزين المؤقت لتخزين الرسائل وملفات تعريف المستخدمين في Redis ، ومدير الرسائل إلى رسائل ذاكرة التخزين المؤقت بأمان بحيث لا يتجاوز عدد الرموز الحد الأقصى المقبول.
مدير ذاكرة التخزين المؤقت ( CacheManager ) مسؤول عن التعامل مع معلومات سياق المستخدم وتاريخ الرسائل. يقوم بتخزين هذه البيانات في redis ، مما يسمح بسهولة الاسترجاع والتعديل. يوفر المدير عدة طرق للتفاعل مع ذاكرة التخزين المؤقت ، مثل:
read_context_from_profile : يقرأ سياق دردشة المستخدم من redis ، وفقًا لملف تعريف المستخدم.create_context : يقوم بإنشاء سياق دردشة مستخدم جديد في Redis.reset_context : إعادة تعيين سياق دردشة المستخدم إلى القيم الافتراضية.update_message_histories : يقوم بتحديث تاريخ الرسائل لدور محدد (المستخدم أو الذكاء الاصطناعي أو النظام).lpop_message_history / rpop_message_history : يزيل وإرجاع سجل الرسالة من الطرف الأيسر أو الأيمن من القائمة.append_message_history : قم بإلحاق سجل الرسالة إلى نهاية القائمة.get_message_history : استرجع تاريخ الرسالة لدور محدد.delete_message_history : حذف سجل الرسالة لدور محدد.set_message_history : يعين سجل رسالة محدد لدور وفهرس. يضمن مدير الرسائل ( MessageManager ) أن عدد الرموز في تاريخ الرسائل لا يتجاوز الحد المحدد. إنه يتعامل بأمان لإضافة تاريخ الرسائل وإزالته وإعداده في سياق دردشة المستخدم مع الحفاظ على حدود الرمز المميز. يوفر المدير عدة طرق للتفاعل مع تاريخ الرسائل ، مثل:
add_message_history_safely : يضيف سجل رسالة إلى سياق دردشة المستخدم ، مما يضمن عدم تجاوز حد الرمز المميز.pop_message_history_safely : يزيل وإرجاع سجل الرسالة من الطرف الأيمن من القائمة أثناء تحديث عدد الرمز المميز.set_message_history_safely : يعين سجل رسالة محدد في سياق دردشة المستخدم ، وتحديث عدد الرمز المميز والتأكد من عدم تجاوز حد الرمز المميز. لاستخدام مدير ذاكرة التخزين المؤقت ومدير الرسائل في مشروعك ، استيرادها على النحو التالي:
from app . utils . chat . managers . cache import CacheManager
from app . utils . chat . message_manager import MessageManagerبعد ذلك ، يمكنك استخدام أساليبهم للتفاعل مع ذاكرة التخزين المؤقت Redis وإدارة تاريخ الرسائل وفقًا لمتطلباتك.
على سبيل المثال ، لإنشاء سياق دردشة المستخدم الجديد:
user_id = "[email protected]" # email format
chat_room_id = "example_chat_room_id" # usually the 32 characters from `uuid.uuid4().hex`
default_context = UserChatContext . construct_default ( user_id = user_id , chat_room_id = chat_room_id )
await CacheManager . create_context ( user_chat_context = default_context )لإضافة سجل رسالة بأمان إلى سياق دردشة المستخدم:
user_chat_context = await CacheManager . read_context_from_profile ( user_chat_profile = UserChatProfile ( user_id = user_id , chat_room_id = chat_room_id ))
content = "This is a sample message."
role = ChatRoles . USER # can be enum such as ChatRoles.USER, ChatRoles.AI, ChatRoles.SYSTEM
await MessageManager . add_message_history_safely ( user_chat_context , content , role ) يستخدم هذا المشروع token_validator الوسيطة وغيرها من الأدوات الوسطى المستخدمة في تطبيق Fastapi. هذه الأدوات المتوسطة هي المسؤولة عن التحكم في الوصول إلى واجهة برمجة التطبيقات ، مما يضمن معالجة الطلبات المعتمدة والمصادقة فقط.
تتم إضافة الأدوات الوسطى التالية إلى تطبيق Fastapi:
يتم تعريف البرامج الوسيطة للتحكم في الوصول في ملف token_validator.py . وهي مسؤولة عن التحقق من صحة مفاتيح API ورموز JWT.
يتم استخدام فئة StateManager لتهيئة متغيرات حالة الطلب. يقوم بتعيين وقت الطلب ووقت البدء وعنوان IP ورمز المستخدم.
تحتوي فئة AccessControl على طريقتين ثابتتين للتحقق من صحة مفاتيح API ورموز JWT:
api_service : التحقق من صحة مفاتيح API عن طريق التحقق من وجود معلمات الاستعلام المطلوبة والرؤوس في الطلب. يستدعي طريقة Validator.api_key .non_api_service : يتحقق من صحة الرموز JWT من خلال التحقق من وجود ملف تعريف ارتباط "التفويض" أو "التفويض" في الطلب. يستدعي طريقة Validator.jwt لفك تشفير الرمز المميز JWT والتحقق منه. تحتوي فئة Validator على طريقتين ثابتتين للتحقق من صحة مفاتيح API ورموز JWT:
api_key : يتحقق من مفتاح الوصول إلى API ، سر شرق ، وخط الزمن. إرجاع كائن UserToken إذا كان التحقق من الصحة ناجحًا.jwt : فك تشفير الرمز المميز ويتحقق من رمز JWT. إرجاع كائن UserToken إذا كان التحقق من الصحة ناجحًا. وظيفة access_control هي دالة غير متزامنة تتولى الطلب وتدفق الاستجابة للبرامج الوسيطة. يقوم بتهيئة حالة الطلب باستخدام فئة StateManager ، ويحدد نوع المصادقة المطلوبة لعنوان URL المطلوب (مفتاح API أو JWT الرمز المميز) ، ويؤدي إلى صحة المصادقة باستخدام فئة AccessControl . في حالة حدوث خطأ أثناء عملية التحقق من الصحة ، يتم رفع استثناء HTTP المناسب.
يتم تعريف المرافق الرمزية في ملف token.py . أنه يحتوي على وظيفتين:
create_access_token : إنشاء رمز JWT مع البيانات المحددة ووقت انتهاء الصلاحية.token_decode : فك تشفير الرموز والتحقق من رمز JWT. يثير استثناء إذا انتهت صلاحية الرمز المميز أو لا يمكن فك تشفيره. يحتوي ملف params_utils.py
hash_params : يأخذ معلمات الاستعلام والمفتاح السري كمدخلات وإرجاع سلسلة تجزئة مشفرة BASE64. يحتوي ملف date_utils.py على فئة UTC مع وظائف الأداة المساعدة للعمل مع التواريخ والجداول الزمنية:
now : إرجاع DateTime UTC الحالي مع اختلاف ساعة اختيارية.timestamp : إرجاع الطابع الزمني للموسيقى UTC الحالية مع اختلاف ساعة اختيارية.timestamp_to_datetime : يحول الطابع الزمني إلى كائن DateTime مع اختلاف ساعة اختيارية. يحتوي ملف logger.py على فئة ApiLogger ، التي تقوم بتسجيل معلومات طلب API ومعلومات الاستجابة ، بما في ذلك عنوان URL ، الطريقة ، رمز الحالة ، معلومات العميل ، وقت المعالجة ، وتفاصيل الخطأ (إن أمكن). يتم استدعاء وظيفة المسجل في نهاية وظيفة access_control لتسجيل الطلب والاستجابة المعالجة.
لاستخدام البرامج الوسيطة token_validator في تطبيق Fastapi الخاص بك ، ما عليك سوى استيراد وظيفة access_control وأضفها كبرنامج وسيط إلى مثيل Fastapi الخاص بك:
from app . middlewares . token_validator import access_control
app = FastAPI ()
app . add_middleware ( dispatch = access_control , middleware_class = BaseHTTPMiddleware )تأكد أيضًا من إضافة CORS و Middlewares المضيف الموثوق به للتحكم الكامل في الوصول:
app . add_middleware (
CORSMiddleware ,
allow_origins = config . allowed_sites ,
allow_credentials = True ,
allow_methods = [ "*" ],
allow_headers = [ "*" ],
)
app . add_middleware (
TrustedHostMiddleware ,
allowed_hosts = config . trusted_hosts ,
except_path = [ "/health" ],
) الآن ، ستتم معالجة أي طلبات واردة لتطبيق Fastapi الخاص بك من خلال البرامج الوسيطة token_validator وغيرها من الأدوات المتوسطة ، مما يضمن معالجة الطلبات المصرح بها والمصادقة فقط.
يوفر app.database.connection واجهة سهلة الاستخدام لإدارة اتصالات قاعدة البيانات وتنفيذ استعلامات SQL باستخدام SQLAlchemy و Redis. وهو يدعم MySQL ، ويمكن دمجه بسهولة مع هذا المشروع.
أولاً ، استيراد الفئات المطلوبة من الوحدة:
from app . database . connection import MySQL , SQLAlchemy , CacheFactory بعد ذلك ، قم بإنشاء مثيل لفئة SQLAlchemy وتكوينه باستخدام إعدادات قاعدة البيانات الخاصة بك:
from app . common . config import Config
config : Config = Config . get ()
db = SQLAlchemy ()
db . start ( config ) يمكنك الآن استخدام مثيل db لتنفيذ استعلامات SQL وإدارة الجلسات:
# Execute a raw SQL query
result = await db . execute ( "SELECT * FROM users" )
# Use the run_in_session decorator to manage sessions
@ db . run_in_session
async def create_user ( session , username , password ):
await session . execute ( "INSERT INTO users (username, password) VALUES (:username, :password)" , { "username" : username , "password" : password })
await create_user ( "JohnDoe" , "password123" ) لاستخدام التخزين المؤقت لـ redis ، قم بإنشاء مثيل لفئة CacheFactory وتكوينه مع إعدادات redis الخاصة بك:
cache = CacheFactory ()
cache . start ( config ) يمكنك الآن استخدام مثيل ذاكرة cache للتفاعل مع Redis:
# Set a key in Redis
await cache . redis . set ( "my_key" , "my_value" )
# Get a key from Redis
value = await cache . redis . get ( "my_key" ) في الواقع ، في هذا المشروع ، تقوم فئة MySQL بالإعداد الأولي عند بدء تشغيل التطبيق ، ويتم إجراء جميع اتصالات قاعدة البيانات مع متغيرات db و cache فقط الموجودة في نهاية الوحدة النمطية. ؟
سيتم إجراء جميع إعدادات DB في create_app() في app.common.app_settings . على سبيل المثال ، ستبدو دالة create_app() في app.common.app_settings هكذا:
def create_app ( config : Config ) -> FastAPI :
# Initialize app & db & js
new_app = FastAPI (
title = config . app_title ,
description = config . app_description ,
version = config . app_version ,
)
db . start ( config = config )
cache . start ( config = config )
js_url_initializer ( js_location = "app/web/main.dart.js" )
# Register routers
# ...
return new_app يستخدم هذا المشروع طريقة بسيطة وفعالة للتعامل مع عمليات قاعدة البيانات (إنشاء ، قراءة ، تحديث ، حذف) باستخدام SQLalChemy واثنين من الوحدة النمطية والمسار: app.database.models.schema و app.database.crud .
وحدة schema.py هي المسؤولة عن تحديد نماذج قاعدة البيانات وعلاقاتها باستخدام SqlalChemy. ويشمل مجموعة من الفئات التي ترث من Base ، مثيل من declarative_base() . يمثل كل فئة جدولًا في قاعدة البيانات ، وتمثل سماته أعمدة في الجدول. ترث هذه الفئات أيضًا من فئة Mixin ، والتي توفر بعض الأساليب والسمات الشائعة لجميع النماذج.
توفر فئة Mixin بعض السمات والأساليب الشائعة لجميع الفئات التي ترث منها. تشمل بعض السمات:
id : INTEGER المفتاح الأساسي للجدول.created_at : DateTime لوقت إنشاء السجل.updated_at : DateTime لوقت آخر تحديث السجل.ip_address : عنوان IP للعميل الذي أنشأ السجل أو تحديثه.كما يوفر العديد من طرق الفصل التي تؤدي عمليات CRUD باستخدام SQLAlchemy ، مثل:
add_all() : يضيف سجلات متعددة إلى قاعدة البيانات.add_one() : يضيف سجلًا واحدًا إلى قاعدة البيانات.update_where() : تحديث السجلات في قاعدة البيانات بناءً على مرشح.fetchall_filtered_by() : يجلب جميع السجلات من قاعدة البيانات التي تتطابق مع المرشح المقدم.one_filtered_by() : يجلب سجلًا واحدًا من قاعدة البيانات التي تتطابق مع المرشح المقدم.first_filtered_by() : يجلب السجل الأول من قاعدة البيانات التي تتطابق مع المرشح المقدم.one_or_none_filtered_by() : يجلب سجلًا واحدًا أو يعيد None إذا لم تتطابق سجلات مع المرشح المقدم. تحتوي وحدة users.py و api_keys.py على مجموعة من الوظائف التي تؤدي عمليات CRUD باستخدام الفئات المحددة في schema.py . تستخدم هذه الوظائف طرق الفئة التي توفرها فئة Mixin للتفاعل مع قاعدة البيانات.
تشمل بعض الوظائف في هذه الوحدة:
create_api_key() : إنشاء مفتاح API جديد للمستخدم.get_api_keys() : يسترجع جميع مفاتيح API للمستخدم.get_api_key_owner() : يسترجع مالك مفتاح API.get_api_key_and_owner() : يسترجع مفتاح API ومالكه.update_api_key() : تحديث مفتاح API.delete_api_key() : حذف مفتاح API.is_email_exist() : يتحقق إذا كان هناك بريد إلكتروني في قاعدة البيانات.get_me() : يسترجع معلومات المستخدم بناءً على معرف المستخدم.is_valid_api_key() : يتحقق إذا كان مفتاح API صالحًا.register_new_user() : يسجل مستخدمًا جديدًا في قاعدة البيانات.find_matched_user() : يجد مستخدمًا مع بريد إلكتروني مطابق في قاعدة البيانات. لاستخدام عمليات CRUD المقدمة ، استيراد الوظائف ذات الصلة من وحدة crud.py واتصل بها بالمعلمات المطلوبة. على سبيل المثال:
import asyncio
from app . database . crud . users import register_new_user , get_me , is_email_exist
from app . database . crud . api_keys import create_api_key , get_api_keys , update_api_key , delete_api_key
async def main ():
# `user_id` is an integer index in the MySQL database, and `email` is user's actual name
# the email will be used as `user_id` in chat. Don't confuse with `user_id` in MySQL
# Register a new user
new_user = await register_new_user ( email = "[email protected]" , hashed_password = "..." )
# Get user information
user = await get_me ( user_id = 1 )
# Check if an email exists in the database
email_exists = await is_email_exist ( email = "[email protected]" )
# Create a new API key for user with ID 1
new_api_key = await create_api_key ( user_id = 1 , additional_key_info = { "user_memo" : "Test API Key" })
# Get all API keys for user with ID 1
api_keys = await get_api_keys ( user_id = 1 )
# Update the first API key in the list
updated_api_key = await update_api_key ( updated_key_info = { "user_memo" : "Updated Test API Key" }, access_key_id = api_keys [ 0 ]. id , user_id = 1 )
# Delete the first API key in the list
await delete_api_key ( access_key_id = api_keys [ 0 ]. id , access_key = api_keys [ 0 ]. access_key , user_id = 1 )
if __name__ == "__main__" :
asyncio . run ( main ())
