feat: adicionando e melhorando utilitários do langchain

_utils/langchain_utils/Chain_class.py

@@ -1,11 +1,23 @@
+from setup.easy_imports import RunnablePassthrough, create_retrieval_chain
+
 class Chain:
-    def __init__(self, prompt, model):
-        self.prompt = prompt
-        self.model = model
 
-    def create_prompt_model_chain(self):
-        return self.prompt | self.model
+    def create_prompt_model_chain(self, prompt, model):
+        return prompt | model
+
+    def create_prompt_model_retriever_chain(self, prompt, model, retriever):
+        chain = prompt | model
+        return create_retrieval_chain(retriever, chain)
+
+    def invoke_retrieval_chain(self, chain, busca_no_vetor):
+        resposta = chain.invoke({"input": busca_no_vetor})
+
+        class Resposta:
+            def __init__(self, resposta):
+                self.final_answer = resposta["answer"].content
+                self.complete_obj = resposta
+
+        return Resposta(resposta)
+
 
-    def invoke_prompt_model_chain(self, invoke_params):
-        chain = self.create_prompt_model_chain()
-        return chain.invoke(invoke_params)
+chain = Chain()

_utils/langchain_utils/Document_class.py

@@ -0,0 +1,19 @@
+from setup.easy_imports import PyPDFLoader
+
+
+class Document_Class:
+    def load_pdf(self, pdf, ocr=False):
+        return PyPDFLoader(pdf, extract_images=ocr).load()
+
+    def load_and_split_pdf(self, pdf, ocr=False):
+        return PyPDFLoader(pdf, extract_images=ocr).load_and_split()
+
+    def get_pdf_text(self, pdf, ocr=False):
+        document = self.load_pdf(pdf, ocr)
+        texto = ""
+        for x in document:
+            texto += x.page_content
+        return texto
+
+
+document = Document_Class()

_utils/langchain_utils/LLM_class.py

@@ -6,14 +6,15 @@ import os
 
 deepseek_api_key = cast(str, os.environ.get("DEEPSEEKK_API_KEY"))
 google_api_key = cast(str, os.environ.get("GOOGLE_API_KEY_PEIXE"))
+open_ai_token = cast(str, os.environ.get("OPENAI_API_KEY"))
 
 
 class LLM:
     def __init__(self):
         pass
 
-    # def create_GPT_model(self, model=default_model):
-    #     return ChatOpen()
+    def open_ai(self, model="gpt-4o-mini"):
+        return ChatOpenAI(api_key=SecretStr(open_ai_token), model=model)
 
     def deepseek(self, model="deepseek-chat"):
         return ChatOpenAI(

_utils/langchain_utils/Prompt_class.py

@@ -11,4 +11,10 @@ class Prompt:
         )
         return prompt_template
 
+    def create_and_invoke_prompt(self, user_prompt, system_prompt="", dynamic_dict={}):
+        return ChatPromptTemplate.from_messages(
+            [("system", system_prompt), ("user", user_prompt)]
+        ).invoke(dynamic_dict)
+
+
 prompt = Prompt()

_utils/langchain_utils/Splitter_class.py

@@ -170,3 +170,31 @@ class Splitter:
         char_count += len(text)
 
         return chunks
+
+
+class Splitter_Simple:
+    def __init__(
+        self,
+        chunk_size=1000,
+        chunk_overlap=400,
+    ):
+        self.text_splitter = RecursiveCharacterTextSplitter(
+            chunk_size=chunk_size, chunk_overlap=chunk_overlap
+        )
+
+    async def load_and_split_document(self, pdf_path: str):
+        """Load PDF and split into chunks with metadata"""
+        print("\nCOMEÇANDO LEITURA DO PDF")
+        pages = PyPDFLoader(pdf_path).load_and_split(self.text_splitter)
+        print("\nTERMINADO LEITURA DO PDF")
+
+        return pages
+
+    def load_and_split_text(self, text: str) -> List[Document]:
+        documents: List[Document] = []
+        chunks = self.text_splitter.split_text(text)
+
+        for chunk in chunks:
+            documents.append(Document(page_content=chunk))
+
+        return documents

_utils/langchain_utils/embeddings.py

@@ -0,0 +1,13 @@
+import os
+
+from pydantic import Secret
+from setup.easy_imports import OpenAIEmbeddings
+from setup.tokens import openai_api_key
+
+
+class EmbeddingClass:
+    def open_ai(self):
+        return OpenAIEmbeddings(api_key=Secret(openai_api_key))  # type: ignore
+
+
+embedding = EmbeddingClass()

_utils/langchain_utils/retriever.py

@@ -0,0 +1,15 @@
+from _utils.langchain_utils.vector_stores import vector_store
+
+
+class Retriever:
+    def chroma_retriever(
+        self, lista_de_documents, search_type="similarity", search_kwargs={"k": 1}
+    ):
+        retriever = vector_store.chroma(lista_de_documents).as_retriever(
+            search_type=search_type,
+            search_kwargs=search_kwargs,
+        )
+        return retriever
+
+
+retriever = Retriever()

_utils/langchain_utils/vector_stores.py

@@ -0,0 +1,10 @@
+from setup.easy_imports import Chroma
+from _utils.langchain_utils.embeddings import embedding
+
+
+class VectorStoreClass:
+    def chroma(self, lista_de_documents):
+        return Chroma.from_documents(lista_de_documents, embedding.open_ai())
+
+
+vector_store = VectorStoreClass()

setup/easy_imports.py

@@ -11,14 +11,17 @@ from langchain_huggingface import HuggingFaceEmbeddings
 
 # from langchain_community.embeddings import HuggingFaceEmbeddings
 from langchain.prompts import PromptTemplate
+from langchain_core.runnables import RunnablePassthrough
 from langchain_core.prompts import ChatPromptTemplate
 from langchain_community.document_loaders import PyPDFLoader, Docx2txtLoader
 from langchain_community.vectorstores import Chroma
 from langchain_google_genai import ChatGoogleGenerativeAI
 
 # from langchain_community.chat_models import ChatOpenAI
-from langchain_openai import ChatOpenAI
+from langchain_openai import ChatOpenAI, OpenAIEmbeddings
 from langchain.schema import Document
 from langchain.chains import create_extraction_chain
+from langchain.chains.retrieval import create_retrieval_chain
+from langchain.chains.combine_documents import create_stuff_documents_chain
 
 from rank_bm25 import BM25Okapi

setup/tokens.py

@@ -0,0 +1,11 @@
+import os
+from typing import cast
+
+openai_api_key = cast(str, os.environ.get("OPENAI_API_KEY", ""))
+claude_api_key = cast(str, os.environ.get("CLAUDE_API_KEY"))
+langchain_api_key = cast(str, os.environ.get("LANGCHAIN_API_KEY"))
+hugging_face_api_key = cast(str, os.environ.get("HUGGINGFACEHUB_API_TOKEN"))
+bubble_token = cast(str, os.environ.get("BUBBLE_TOKEN"))
+cohere_api_key = cast(str, os.environ.get("COHERE_API_KEY", ""))
+deepseek_api_key = cast(str, os.environ.get("DEEPSEEKK_API_KEY"))
+google_api_key = cast(str, os.environ.get("GOOGLE_API_KEY_PEIXE"))