Buckets:
tahamajs/Sysmem2_in_AI / ComputerAssignments /CA14_multi_agent_reasoning /integrations /rag_system.py
| #!/usr/bin/env python3 | |
| """ | |
| سیستم RAG کامل با Gemini API | |
| Complete RAG System with Gemini API | |
| """ | |
| import os | |
| import json | |
| import logging | |
| from typing import List, Dict, Any, Optional, Tuple | |
| from pathlib import Path | |
| import pickle | |
| # LangChain imports | |
| from langchain_google_genai import ChatGoogleGenerativeAI, GoogleGenerativeAIEmbeddings | |
| from langchain_community.document_loaders import ( | |
| TextLoader, | |
| PyPDFLoader, | |
| DirectoryLoader, | |
| CSVLoader, | |
| JSONLoader, | |
| WebBaseLoader, | |
| ) | |
| from langchain_community.vectorstores import Chroma, FAISS | |
| from langchain.text_splitter import RecursiveCharacterTextSplitter, TokenTextSplitter | |
| from langchain.chains import RetrievalQA, ConversationalRetrievalChain | |
| from langchain.memory import ConversationBufferWindowMemory, ConversationSummaryMemory | |
| from langchain_core.prompts import PromptTemplate | |
| from langchain_core.callbacks import StreamingStdOutCallbackHandler | |
| from langchain_core.runnables import RunnablePassthrough | |
| # Gemini configuration | |
| from gemini_config import get_gemini_config | |
| # تنظیم لاگ | |
| logging.basicConfig(level=logging.INFO) | |
| logger = logging.getLogger(__name__) | |
| class DocumentProcessor: | |
| """پردازشگر اسناد""" | |
| def __init__(self, chunk_size: int = 1000, chunk_overlap: int = 200): | |
| """مقداردهی اولیه""" | |
| self.chunk_size = chunk_size | |
| self.chunk_overlap = chunk_overlap | |
| # ایجاد text splitter | |
| self.text_splitter = RecursiveCharacterTextSplitter( | |
| chunk_size=chunk_size, | |
| chunk_overlap=chunk_overlap, | |
| length_function=len, | |
| separators=["\n\n", "\n", " ", ""], | |
| ) | |
| # ایجاد token splitter برای متن‌های طولانی | |
| self.token_splitter = TokenTextSplitter( | |
| chunk_size=chunk_size, chunk_overlap=chunk_overlap | |
| ) | |
| def load_documents(self, paths: List[str]) -> List[Any]: | |
| """بارگذاری اسناد از مسیرهای مختلف""" | |
| documents = [] | |
| for path in paths: | |
| try: | |
| if os.path.isfile(path): | |
| # بارگذاری فایل تکی | |
| doc = self._load_single_file(path) | |
| if doc: | |
| documents.extend(doc) | |
| elif os.path.isdir(path): | |
| # بارگذاری پوشه | |
| docs = self._load_directory(path) | |
| documents.extend(docs) | |
| else: | |
| logger.warning(f"مسیر یافت نشد: {path}") | |
| except Exception as e: | |
| logger.error(f"خطا در بارگذاری {path}: {e}") | |
| logger.info(f"📄 {len(documents)} سند بارگذاری شد") | |
| return documents | |
| def _load_single_file(self, path: str) -> Optional[List[Any]]: | |
| """بارگذاری فایل تکی""" | |
| try: | |
| if path.endswith(".pdf"): | |
| loader = PyPDFLoader(path) | |
| elif path.endswith(".csv"): | |
| loader = CSVLoader(path, encoding="utf-8") | |
| elif path.endswith(".json"): | |
| loader = JSONLoader(path, jq_schema=".", text_content=False) | |
| elif path.endswith(".txt"): | |
| loader = TextLoader(path, encoding="utf-8") | |
| else: | |
| logger.warning(f"فرمت فایل پشتیبانی نمی‌شود: {path}") | |
| return None | |
| return loader.load() | |
| except Exception as e: | |
| logger.error(f"خطا در بارگذاری فایل {path}: {e}") | |
| return None | |
| def _load_directory(self, directory: str) -> List[Any]: | |
| """بارگذاری پوشه""" | |
| try: | |
| loader = DirectoryLoader( | |
| directory, | |
| glob="**/*.txt", | |
| loader_cls=TextLoader, | |
| loader_kwargs={"encoding": "utf-8"}, | |
| ) | |
| return loader.load() | |
| except Exception as e: | |
| logger.error(f"خطا در بارگذاری پوشه {directory}: {e}") | |
| return [] | |
| def split_documents( | |
| self, documents: List[Any], use_token_splitter: bool = False | |
| ) -> List[Any]: | |
| """تقسیم اسناد به بخش‌های کوچک‌تر""" | |
| try: | |
| if use_token_splitter: | |
| splits = self.token_splitter.split_documents(documents) | |
| else: | |
| splits = self.text_splitter.split_documents(documents) | |
| logger.info(f"📝 {len(splits)} بخش ایجاد شد") | |
| return splits | |
| except Exception as e: | |
| logger.error(f"خطا در تقسیم اسناد: {e}") | |
| return documents | |
| class VectorStoreManager: | |
| """مدیر vector store""" | |
| def __init__(self, embeddings, persist_directory: str = "./vector_store"): | |
| """مقداردهی اولیه""" | |
| self.embeddings = embeddings | |
| self.persist_directory = persist_directory | |
| self.vector_store = None | |
| def create_vector_store( | |
| self, documents: List[Any], store_type: str = "chroma" | |
| ) -> Any: | |
| """ایجاد vector store""" | |
| try: | |
| if store_type.lower() == "chroma": | |
| self.vector_store = Chroma.from_documents( | |
| documents=documents, | |
| embedding=self.embeddings, | |
| persist_directory=self.persist_directory, | |
| ) | |
| elif store_type.lower() == "faiss": | |
| self.vector_store = FAISS.from_documents( | |
| documents=documents, embedding=self.embeddings | |
| ) | |
| # ذخیره FAISS | |
| self.vector_store.save_local(self.persist_directory) | |
| else: | |
| raise ValueError(f"نوع vector store پشتیبانی نمی‌شود: {store_type}") | |
| logger.info(f"✅ Vector store از نوع {store_type} ایجاد شد") | |
| return self.vector_store | |
| except Exception as e: | |
| logger.error(f"خطا در ایجاد vector store: {e}") | |
| raise | |
| def load_vector_store(self, store_type: str = "chroma") -> Any: | |
| """بارگذاری vector store موجود""" | |
| try: | |
| if store_type.lower() == "chroma": | |
| self.vector_store = Chroma( | |
| persist_directory=self.persist_directory, | |
| embedding_function=self.embeddings, | |
| ) | |
| elif store_type.lower() == "faiss": | |
| self.vector_store = FAISS.load_local( | |
| self.persist_directory, | |
| self.embeddings, | |
| allow_dangerous_deserialization=True, | |
| ) | |
| else: | |
| raise ValueError(f"نوع vector store پشتیبانی نمی‌شود: {store_type}") | |
| logger.info(f"✅ Vector store از نوع {store_type} بارگذاری شد") | |
| return self.vector_store | |
| except Exception as e: | |
| logger.error(f"خطا در بارگذاری vector store: {e}") | |
| raise | |
| def get_retriever( | |
| self, search_type: str = "similarity", k: int = 4, **kwargs | |
| ) -> Any: | |
| """دریافت retriever""" | |
| if self.vector_store is None: | |
| raise ValueError("Vector store ایجاد نشده است") | |
| return self.vector_store.as_retriever( | |
| search_type=search_type, search_kwargs={"k": k, **kwargs} | |
| ) | |
| class RAGSystem: | |
| """سیستم RAG کامل""" | |
| def __init__(self, model_name: str = "gemini-1.5-pro", temperature: float = 0.7): | |
| """مقداردهی اولیه""" | |
| self.model_name = model_name | |
| self.temperature = temperature | |
| self.config = get_gemini_config() | |
| # ایجاد مدل‌ها | |
| self.llm = ChatGoogleGenerativeAI( | |
| model=self.model_name, | |
| temperature=self.temperature, | |
| google_api_key=self.config.api_key, | |
| convert_system_message_to_human=True, | |
| streaming=True, | |
| ) | |
| self.embeddings = GoogleGenerativeAIEmbeddings( | |
| model="models/embedding-001", google_api_key=self.config.api_key | |
| ) | |
| # ایجاد پردازشگر اسناد | |
| self.doc_processor = DocumentProcessor() | |
| # ایجاد مدیر vector store | |
| self.vector_manager = VectorStoreManager(self.embeddings) | |
| # حافظه مکالمه | |
| self.memory = ConversationBufferWindowMemory( | |
| k=5, memory_key="chat_history", return_messages=True | |
| ) | |
| # chain های مختلف | |
| self.qa_chain = None | |
| self.conversational_chain = None | |
| logger.info("✅ سیستم RAG ایجاد شد") | |
| def setup_documents( | |
| self, | |
| document_paths: List[str], | |
| store_type: str = "chroma", | |
| recreate: bool = False, | |
| ) -> None: | |
| """تنظیم اسناد برای RAG""" | |
| try: | |
| # بررسی وجود vector store | |
| if not recreate and os.path.exists(self.vector_manager.persist_directory): | |
| logger.info("بارگذاری vector store موجود...") | |
| self.vector_manager.load_vector_store(store_type) | |
| else: | |
| logger.info("ایجاد vector store جدید...") | |
| # بارگذاری اسناد | |
| documents = self.doc_processor.load_documents(document_paths) | |
| # تقسیم اسناد | |
| splits = self.doc_processor.split_documents(documents) | |
| # ایجاد vector store | |
| self.vector_manager.create_vector_store(splits, store_type) | |
| # ایجاد retriever | |
| self.retriever = self.vector_manager.get_retriever( | |
| search_type="mmr", k=6, fetch_k=20, lambda_mult=0.7 | |
| ) | |
| logger.info("✅ اسناد برای RAG تنظیم شدند") | |
| except Exception as e: | |
| logger.error(f"خطا در تنظیم اسناد: {e}") | |
| raise | |
| def create_qa_chain(self, custom_prompt: Optional[str] = None) -> RetrievalQA: | |
| """ایجاد chain برای Q&A""" | |
| try: | |
| # Prompt پیش‌فرض | |
| if custom_prompt is None: | |
| custom_prompt = """ | |
| شما یک دستیار هوشمند هستید که بر اساس اسناد ارائه شده پاسخ می‌دهید. | |
| متن اسناد: | |
| {context} | |
| سوال: {question} | |
| دستورالعمل‌ها: | |
| 1. فقط از اطلاعات موجود در اسناد پاسخ دهید | |
| 2. اگر پاسخ در اسناد نیست، بگویید "اطلاعات کافی در اسناد موجود نیست" | |
| 3. پاسخ را به فارسی و واضح ارائه دهید | |
| 4. در صورت نیاز، منابع را ذکر کنید | |
| پاسخ: | |
| """ | |
| prompt = PromptTemplate( | |
| template=custom_prompt, input_variables=["context", "question"] | |
| ) | |
| # ایجاد chain | |
| self.qa_chain = RetrievalQA.from_chain_type( | |
| llm=self.llm, | |
| chain_type="stuff", | |
| retriever=self.retriever, | |
| chain_type_kwargs={"prompt": prompt}, | |
| return_source_documents=True, | |
| ) | |
| logger.info("✅ Chain Q&A ایجاد شد") | |
| return self.qa_chain | |
| except Exception as e: | |
| logger.error(f"خطا در ایجاد chain Q&A: {e}") | |
| raise | |
| def create_conversational_chain(self) -> ConversationalRetrievalChain: | |
| """ایجاد chain مکالمه‌ای""" | |
| try: | |
| self.conversational_chain = ConversationalRetrievalChain.from_llm( | |
| llm=self.llm, | |
| retriever=self.retriever, | |
| memory=self.memory, | |
| return_source_documents=True, | |
| verbose=True, | |
| max_tokens_limit=4000, | |
| ) | |
| logger.info("✅ Chain مکالمه‌ای ایجاد شد") | |
| return self.conversational_chain | |
| except Exception as e: | |
| logger.error(f"خطا در ایجاد chain مکالمه‌ای: {e}") | |
| raise | |
| def ask_question( | |
| self, question: str, use_conversation: bool = False | |
| ) -> Dict[str, Any]: | |
| """پرسیدن سوال از سیستم RAG""" | |
| try: | |
| if use_conversation and self.conversational_chain: | |
| result = self.conversational_chain.invoke({"question": question}) | |
| elif self.qa_chain: | |
| result = self.qa_chain.invoke({"query": question}) | |
| else: | |
| raise ValueError("هیچ chain فعال نیست") | |
| return { | |
| "question": question, | |
| "answer": result["result"], | |
| "sources": [ | |
| doc.page_content[:200] + "..." | |
| for doc in result.get("source_documents", []) | |
| ], | |
| "metadata": [ | |
| doc.metadata for doc in result.get("source_documents", []) | |
| ], | |
| } | |
| except Exception as e: | |
| logger.error(f"خطا در پاسخ‌دهی: {e}") | |
| return { | |
| "question": question, | |
| "answer": f"خطا: {str(e)}", | |
| "sources": [], | |
| "metadata": [], | |
| } | |
| def batch_questions( | |
| self, questions: List[str], use_conversation: bool = False | |
| ) -> List[Dict[str, Any]]: | |
| """پرسیدن چندین سوال""" | |
| results = [] | |
| for question in questions: | |
| logger.info(f"پردازش سوال: {question}") | |
| result = self.ask_question(question, use_conversation) | |
| results.append(result) | |
| return results | |
| def evaluate_system( | |
| self, test_questions: List[str], expected_answers: List[str] = None | |
| ) -> Dict[str, Any]: | |
| """ارزیابی سیستم RAG""" | |
| try: | |
| results = self.batch_questions(test_questions) | |
| evaluation = { | |
| "total_questions": len(test_questions), | |
| "successful_answers": len( | |
| [r for r in results if "خطا" not in r["answer"]] | |
| ), | |
| "failed_answers": len([r for r in results if "خطا" in r["answer"]]), | |
| "avg_sources": sum(len(r["sources"]) for r in results) / len(results), | |
| "results": results, | |
| } | |
| evaluation["success_rate"] = ( | |
| evaluation["successful_answers"] / evaluation["total_questions"] | |
| ) | |
| logger.info(f"📊 ارزیابی سیستم: {evaluation['success_rate']:.2%} موفقیت") | |
| return evaluation | |
| except Exception as e: | |
| logger.error(f"خطا در ارزیابی سیستم: {e}") | |
| return {"error": str(e)} | |
| def save_system_state(self, filepath: str) -> None: | |
| """ذخیره وضعیت سیستم""" | |
| try: | |
| state = { | |
| "model_name": self.model_name, | |
| "temperature": self.temperature, | |
| "persist_directory": self.vector_manager.persist_directory, | |
| "memory_state": self.memory.save_context({"input": ""}, {"output": ""}), | |
| } | |
| with open(filepath, "w", encoding="utf-8") as f: | |
| json.dump(state, f, ensure_ascii=False, indent=2) | |
| logger.info(f"✅ وضعیت سیستم در {filepath} ذخیره شد") | |
| except Exception as e: | |
| logger.error(f"خطا در ذخیره وضعیت: {e}") | |
| def load_system_state(self, filepath: str) -> None: | |
| """بارگذاری وضعیت سیستم""" | |
| try: | |
| with open(filepath, "r", encoding="utf-8") as f: | |
| state = json.load(f) | |
| self.model_name = state["model_name"] | |
| self.temperature = state["temperature"] | |
| self.vector_manager.persist_directory = state["persist_directory"] | |
| logger.info(f"✅ وضعیت سیستم از {filepath} بارگذاری شد") | |
| except Exception as e: | |
| logger.error(f"خطا در بارگذاری وضعیت: {e}") | |
| # توابع کمکی | |
| def create_sample_documents() -> List[str]: | |
| """ایجاد اسناد نمونه""" | |
| sample_docs = [ | |
| """ | |
| هوش مصنوعی و یادگیری ماشین | |
| هوش مصنوعی (AI) شاخه‌ای از علوم کامپیوتر است که به ایجاد سیستم‌های هوشمند می‌پردازد. | |
| یادگیری ماشین (ML) زیرمجموعه‌ای از هوش مصنوعی است که به ماشین‌ها امکان یادگیری از داده‌ها را می‌دهد. | |
| انواع یادگیری ماشین: | |
| 1. یادگیری نظارت شده (Supervised Learning) | |
| 2. یادگیری بدون نظارت (Unsupervised Learning) | |
| 3. یادگیری تقویتی (Reinforcement Learning) | |
| کاربردهای هوش مصنوعی: | |
| - تشخیص تصویر و پردازش زبان طبیعی | |
| - سیستم‌های توصیه‌گر | |
| - خودروهای خودران | |
| - پزشکی و تشخیص بیماری‌ها | |
| """, | |
| """ | |
| پردازش زبان طبیعی (NLP) | |
| پردازش زبان طبیعی شاخه‌ای از هوش مصنوعی است که با تعامل بین کامپیوتر و زبان انسان سروکار دارد. | |
| وظایف اصلی NLP: | |
| 1. تحلیل احساسات (Sentiment Analysis) | |
| 2. ترجمه ماشینی (Machine Translation) | |
| 3. خلاصه‌سازی متن (Text Summarization) | |
| 4. تولید متن (Text Generation) | |
| 5. پاسخ‌دهی به سوالات (Question Answering) | |
| مدل‌های زبانی بزرگ (LLMs) مانند GPT، BERT، و Gemini انقلابی در حوزه NLP ایجاد کرده‌اند. | |
| این مدل‌ها قادر به درک و تولید متن با کیفیت بالا هستند. | |
| """, | |
| """ | |
| سیستم‌های RAG (Retrieval-Augmented Generation) | |
| RAG ترکیبی از بازیابی اطلاعات و تولید متن است که دقت و صحت پاسخ‌های مدل‌های زبانی را بهبود می‌بخشد. | |
| اجزای اصلی RAG: | |
| 1. بازیاب (Retriever): یافتن اطلاعات مرتبط از پایگاه دانش | |
| 2. تولیدکننده (Generator): تولید پاسخ بر اساس اطلاعات بازیابی شده | |
| 3. پایگاه دانش (Knowledge Base): مجموعه اسناد و اطلاعات | |
| مزایای RAG: | |
| - کاهش توهمات مدل | |
| - دسترسی به اطلاعات به‌روز | |
| - قابلیت استناد به منابع | |
| - بهبود دقت پاسخ‌ها | |
| کاربردهای RAG در سیستم‌های پشتیبانی، تحقیق، و تحلیل اسناد است. | |
| """, | |
| ] | |
| # ذخیره اسناد نمونه | |
| docs_dir = Path("sample_documents") | |
| docs_dir.mkdir(exist_ok=True) | |
| for i, doc in enumerate(sample_docs, 1): | |
| with open(docs_dir / f"document_{i}.txt", "w", encoding="utf-8") as f: | |
| f.write(doc.strip()) | |
| return [str(docs_dir / f"document_{i}.txt") for i in range(1, len(sample_docs) + 1)] | |
| def demo_rag_system(): | |
| """نمایش سیستم RAG""" | |
| try: | |
| # ایجاد اسناد نمونه | |
| sample_docs = create_sample_documents() | |
| # ایجاد سیستم RAG | |
| rag = RAGSystem() | |
| # تنظیم اسناد | |
| rag.setup_documents(sample_docs, recreate=True) | |
| # ایجاد chain ها | |
| rag.create_qa_chain() | |
| rag.create_conversational_chain() | |
| # سوالات تست | |
| test_questions = [ | |
| "هوش مصنوعی چیست؟", | |
| "انواع یادگیری ماشین کدامند؟", | |
| "RAG چه مزایایی دارد؟", | |
| "کاربردهای NLP چیست؟", | |
| ] | |
| print("🤖 سیستم RAG آماده است!") | |
| print("=" * 50) | |
| # تست سوالات | |
| for question in test_questions: | |
| print(f"\n❓ سوال: {question}") | |
| result = rag.ask_question(question) | |
| print(f"✅ پاسخ: {result['answer']}") | |
| print(f"📚 منابع: {len(result['sources'])} منبع") | |
| # ارزیابی سیستم | |
| print("\n📊 ارزیابی سیستم:") | |
| evaluation = rag.evaluate_system(test_questions) | |
| print(f"نرخ موفقیت: {evaluation['success_rate']:.2%}") | |
| print(f"میانگین منابع: {evaluation['avg_sources']:.1f}") | |
| logger.info("✅ نمایش سیستم RAG با موفقیت انجام شد") | |
| except Exception as e: | |
| logger.error(f"❌ خطا در نمایش سیستم RAG: {e}") | |
| if __name__ == "__main__": | |
| demo_rag_system() | |
Xet Storage Details
- Size:
- 21.5 kB
- Xet hash:
- 81c4713f281d02986258b6e79ac13d581e4b3155ca1737c58f90546b35e7dfa5
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.