feat: adicionando mais refatorações e a nova rota para chamadas simples a LLMs

_utils/gerar_documento_utils/GerarDocumento.py

@@ -42,6 +42,7 @@ import time
 from setup.tokens import openai_api_key, cohere_api_key
 from setup.logging import Axiom
 import tiktoken
+from setup.environment import default_model
 
 
 def reciprocal_rank_fusion(result_lists, weights=None):
@@ -65,6 +66,10 @@ def reciprocal_rank_fusion(result_lists, weights=None):
 
 @dataclass
 class GerarDocumentoUtils:
+    axiom_instance: Axiom
+    temperature = 0.0
+    model = default_model
+
     def criar_output_estruturado(self, summaries: List[str | Any], sources: Any):
         structured_output = []
         for idx, summary in enumerate(summaries):
@@ -109,6 +114,65 @@ class GerarDocumentoUtils:
             chunk_overlap=serializer.chunk_overlap,
         )
 
+    async def checar_se_resposta_vazia_do_documento_final(
+        self, llm_ultimas_requests: str, prompt: str
+    ):
+        llm = self.select_model_for_last_requests(llm_ultimas_requests)  # type: ignore
+        documento_gerado = ""
+        tentativas = 0
+
+        while tentativas < 5 and not documento_gerado:
+            tentativas += 1
+            try:
+                resposta = llm.invoke(prompt)
+                if hasattr(resposta, "content") and resposta.content.strip():  # type: ignore
+                    if isinstance(resposta.content, list):
+                        resposta.content = "\n".join(resposta.content)  # type: ignore
+
+                    documento_gerado = resposta.content.strip()  # type: ignore
+                else:
+                    print(f"Tentativa {tentativas}: resposta vazia ou inexistente.")
+            except Exception as e:
+                llm = self.select_model_for_last_requests("gemini-2.0-flash")
+                print(f"Tentativa {tentativas}: erro ao invocar o modelo: {e}")
+            time.sleep(5)
+
+        if not documento_gerado:
+            try:
+                self.axiom_instance.send_axiom(
+                    "TENTANDO GERAR DOCUMENTO FINAL COM GPT 4o-mini COMO ÚLTIMA TENTATIVA"
+                )
+                documento_gerado = self.ultima_tentativa_requisicao(prompt)
+            except Exception as e:
+                raise Exception(
+                    "Falha ao gerar o documento final na última tentativa."
+                ) from e
+
+        return documento_gerado
+
+    def select_model_for_last_requests(
+        self,
+        llm_ultimas_requests: Literal[
+            "gpt-4o-mini", "deepseek-chat", "gemini-2.0-flash", "gemini-2.5-pro"
+        ],
+    ):
+        llm_instance = LLM()
+        if llm_ultimas_requests == "gpt-4o-mini":
+            llm = ChatOpenAI(
+                temperature=self.temperature,
+                model=self.model,
+                api_key=SecretStr(openai_api_key),
+            )
+        elif llm_ultimas_requests == "deepseek-chat":
+            llm = llm_instance.deepseek()
+        elif llm_ultimas_requests == "gemini-2.0-flash":
+            llm = llm_instance.google_gemini("gemini-2.0-flash")
+        elif llm_ultimas_requests == "gemini-2.5-pro":
+            llm = llm_instance.google_gemini("gemini-2.5-pro-preview-05-06")
+        elif llm_ultimas_requests == "gemini-2.5-flash":
+            llm = llm_instance.google_gemini("gemini-2.5-flash-preview-04-17")
+        return llm
+
 
 class GerarDocumento:
     lista_pdfs: List[str]
@@ -118,7 +182,7 @@ class GerarDocumento:
     isBubble: bool
     chunks_processados: List[ContextualizedChunk] | List[DocumentChunk]
     resumo_auxiliar: str
-    gerar_documento_utils = GerarDocumentoUtils()
+    gerar_documento_utils: GerarDocumentoUtils
     utils = UtilsClass()
     llm = LLM()
     enhanced_vector_store: tuple[Chroma, BM25Okapi, List[str]]
@@ -143,6 +207,7 @@ class GerarDocumento:
         self.config = self.gerar_documento_utils.create_retrieval_config(serializer)
         self.logger = logging.getLogger(__name__)
         # self.prompt_auxiliar = prompt_auxiliar
+        self.gerar_documento_utils = GerarDocumentoUtils(axiom_instance)
         self.gpt_model = serializer.model
         self.gpt_temperature = serializer.gpt_temperature
         self.prompt_gerar_documento = serializer.prompt_gerar_documento
@@ -318,29 +383,6 @@ class GerarDocumento:
 
         return sources, contexts
 
-    def select_model_for_last_requests(
-        self,
-        llm_ultimas_requests: Literal[
-            "gpt-4o-mini", "deepseek-chat", "gemini-2.0-flash", "gemini-2.5-pro"
-        ],
-    ):
-        llm_instance = LLM()
-        if llm_ultimas_requests == "gpt-4o-mini":
-            llm = ChatOpenAI(
-                temperature=self.gpt_temperature,
-                model=self.gpt_model,
-                api_key=SecretStr(openai_api_key),
-            )
-        elif llm_ultimas_requests == "deepseek-chat":
-            llm = llm_instance.deepseek()
-        elif llm_ultimas_requests == "gemini-2.0-flash":
-            llm = llm_instance.google_gemini("gemini-2.0-flash")
-        elif llm_ultimas_requests == "gemini-2.5-pro":
-            llm = llm_instance.google_gemini("gemini-2.5-pro-preview-05-06")
-        elif llm_ultimas_requests == "gemini-2.5-flash":
-            llm = llm_instance.google_gemini("gemini-2.5-flash-preview-04-17")
-        return llm
-
     async def do_last_requests(
         self,
     ) -> List[Dict]:
@@ -361,14 +403,15 @@ class GerarDocumento:
             )
 
             llm_ultimas_requests = self.llm_ultimas_requests
-            llm = self.select_model_for_last_requests(llm_ultimas_requests)  # type: ignore
             prompt_instance = Prompt()
             context_do_prompt_primeira_etapa = "\n\n".join(contexts)
             prompt_primeira_etapa = prompt_gerar_documento.format(
                 context=context_do_prompt_primeira_etapa,
             )
 
-            documento_gerado = await self.checar_se_resposta_vazia_do_documento_final(
+            self.gerar_documento_utils.model = self.gpt_model
+            self.gerar_documento_utils.temperature = self.gpt_temperature
+            documento_gerado = await self.gerar_documento_utils.checar_se_resposta_vazia_do_documento_final(
                 llm_ultimas_requests, prompt_primeira_etapa
             )
 
@@ -386,7 +429,7 @@ class GerarDocumento:
                     dynamic_dict={"context": context_do_prompt_primeira_etapa},
                 )
                 # documento_gerado = llm.invoke(prompt_etapa_2).content
-                documento_gerado = self.checar_se_resposta_vazia_do_documento_final(
+                documento_gerado = self.gerar_documento_utils.checar_se_resposta_vazia_do_documento_final(
                     llm_ultimas_requests, prompt_etapa_2.to_string()
                 )
                 resposta_segunda_etapa = documento_gerado
@@ -404,7 +447,7 @@ class GerarDocumento:
                     },
                 )
                 # documento_gerado = llm.invoke(prompt_etapa_3).content
-                documento_gerado = self.checar_se_resposta_vazia_do_documento_final(
+                documento_gerado = self.gerar_documento_utils.checar_se_resposta_vazia_do_documento_final(
                     llm_ultimas_requests, prompt_etapa_3.to_string()
                 )
                 texto_final_juntando_as_etapas += f"\n\n{documento_gerado}"
@@ -468,44 +511,6 @@ class GerarDocumento:
 
         self.axiom_instance.send_axiom("TERMINOU A REQUISIÇÃO FINAL PARA O BUBBLE")
 
-    async def checar_se_resposta_vazia_do_documento_final(
-        self, llm_ultimas_requests: str, prompt: str
-    ):
-        llm = self.select_model_for_last_requests(llm_ultimas_requests)  # type: ignore
-        documento_gerado = ""
-        tentativas = 0
-
-        while tentativas < 5 and not documento_gerado:
-            tentativas += 1
-            try:
-                resposta = llm.invoke(prompt)
-                if hasattr(resposta, "content") and resposta.content.strip():  # type: ignore
-                    if isinstance(resposta.content, list):
-                        resposta.content = "\n".join(resposta.content)  # type: ignore
-
-                    documento_gerado = resposta.content.strip()  # type: ignore
-                else:
-                    print(f"Tentativa {tentativas}: resposta vazia ou inexistente.")
-            except Exception as e:
-                llm = self.select_model_for_last_requests("gemini-2.0-flash")
-                print(f"Tentativa {tentativas}: erro ao invocar o modelo: {e}")
-            time.sleep(5)
-
-        if not documento_gerado:
-            try:
-                self.axiom_instance.send_axiom(
-                    "TENTANDO GERAR DOCUMENTO FINAL COM GPT 4o-mini COMO ÚLTIMA TENTATIVA"
-                )
-                documento_gerado = (
-                    self.gerar_documento_utils.ultima_tentativa_requisicao(prompt)
-                )
-            except Exception as e:
-                raise Exception(
-                    "Falha ao gerar o documento final na última tentativa."
-                ) from e
-
-        return documento_gerado
-
     async def gerar_ementa_final(
         self,
         llm_ultimas_requests: str,
@@ -516,7 +521,7 @@ class GerarDocumento:
         llm = self.select_model_for_last_requests(llm_ultimas_requests)  # type: ignore
         prompt_instance = Prompt()
 
-        documento_gerado = await self.checar_se_resposta_vazia_do_documento_final(
+        documento_gerado = await self.gerar_documento_utils.checar_se_resposta_vazia_do_documento_final(
             llm_ultimas_requests, prompt_primeira_etapa
         )
 

setup/installed_apps.py

@@ -16,4 +16,5 @@ INSTALLED_APPS = config_apps + [
     "modelos_usuarios",
     "ragas_api",
     "gerar_documento",
+    "simple_llm",
 ]

setup/urls.py

@@ -19,4 +19,5 @@ urlpatterns = config_urls + [
     path("", include("gerar_documento.urls")),
     path("", include("ragas_api.urls")),
     path("", include("modelos_usuarios.urls")),
+    path("", include("simple_llm.urls")),
 ]

simple_llm/admin.py

@@ -0,0 +1,3 @@
+from django.contrib import admin
+
+# Register your models here.

simple_llm/apps.py

@@ -0,0 +1,6 @@
+from django.apps import AppConfig
+
+
+class SimpleLlmConfig(AppConfig):
+    default_auto_field = 'django.db.models.BigAutoField'
+    name = 'simple_llm'

simple_llm/models.py

@@ -0,0 +1,3 @@
+from django.db import models
+
+# Create your models here.

simple_llm/serializer.py

@@ -0,0 +1,57 @@
+from dataclasses import dataclass, field
+from typing import List, Optional, Union
+from rest_framework import serializers
+from _utils.gerar_documento_utils.prompts import (
+    prompt_gerar_documento,
+)
+from gerar_documento.serializer import FileInfoSerializer, FileInfoSerializerData
+from gerar_documento.serializer_base import (
+    GerarDocumentoParametros,
+    GerarDocumentoParametrosData,
+)
+from setup.environment import default_model
+from django.core.files.uploadedfile import UploadedFile
+
+user_message = "What are the main points of this document?"
+
+
+class SimpleLLMInitialSerializer(serializers.Serializer):
+    files = serializers.ListField(child=serializers.FileField(), required=False)
+    user_text = serializers.CharField(required=False, default=user_message)
+    model = serializers.CharField(required=False, default=default_model)
+    prompt = serializers.CharField(required=False, default=prompt_gerar_documento)
+    llm_ultimas_requests = serializers.CharField(
+        required=False, default="gemini-2.0-flash"
+    )
+
+
+@dataclass
+class SimpleLLMInitialSerializerData:
+    files: List[dict] = field(default_factory=list)
+    user_text: str = ""
+    model: str = default_model
+    prompt: str = ""
+    llm_ultimas_requests: str = "gemini-2.0-flash"
+
+
+class SimpleLLMSerializer(SimpleLLMInitialSerializer):
+    files = serializers.ListField(child=FileInfoSerializer(), required=False)
+    bubble_editor_version = serializers.CharField(
+        required=False, default="version-test"
+    )  # Será o valor utilizado dentro da URL da requisição pro Bubble
+    doc_id = serializers.CharField(required=True)
+    form_response_id = serializers.CharField(required=True)
+    version = serializers.CharField(required=True)
+
+    def get_obj(self):
+        return SimpleSerializerData(**self.validated_data)  # type: ignore
+
+
+@dataclass
+class SimpleSerializerData(SimpleLLMInitialSerializerData):
+    files: List[FileInfoSerializerData] = field(default_factory=list)
+    bubble_editor_version: str = "version-test"
+
+    doc_id: str = ""
+    form_response_id: str = ""
+    version: str = ""

simple_llm/tests.py

@@ -0,0 +1,3 @@
+from django.test import TestCase
+
+# Create your tests here.

simple_llm/urls.py

@@ -0,0 +1,11 @@
+from django.urls import path
+
+from simple_llm.views import SimpleLLMView
+
+urlpatterns = [
+    path(
+        "llm",
+        SimpleLLMView.as_view(),
+        name="simple-llm",
+    ),
+]

simple_llm/views.py

@@ -0,0 +1,73 @@
+from datetime import datetime
+from _utils.custom_exception_handler import custom_exception_handler_without_api_handler
+from _utils.gerar_documento import gerar_documento
+from _utils.gerar_documento_utils.GerarDocumento import (
+    GerarDocumento,
+    GerarDocumentoUtils,
+)
+from _utils.langchain_utils.Prompt_class import Prompt
+from _utils.utils import convert_markdown_to_HTML
+from setup.logging import Axiom
+from setup.easy_imports import (
+    Response,
+    AsyncAPIView,
+    extend_schema,
+)
+from simple_llm.serializer import SimpleLLMSerializer
+
+
+class SimpleLLMView(AsyncAPIView):
+    # parser_classes = [MultiPartParser]
+    serializer = {}
+    axiom_instance = Axiom()
+
+    @extend_schema(
+        request=SimpleLLMSerializer,
+    )
+    async def post(self, request):
+        try:
+            self.axiom_instance.generate_new_uuid()
+            print(f"\n\nDATA E HORA DA REQUISIÇÃO: {datetime.now()}")
+            self.axiom_instance.send_axiom(
+                f"COMEÇOU NOVA REQUISIÇÃO - request.data: {request.data}"
+            )
+            serializer = SimpleLLMSerializer(data=request.data)
+            if serializer.is_valid(raise_exception=True):
+                obj = serializer.get_obj()  # type: ignore
+                if not serializer.validated_data:
+                    raise ValueError("Erro no validated_data")
+
+                self.serializer = obj
+
+                listaPDFs = [l.link_arquivo for l in obj.files]
+                self.axiom_instance.send_axiom(f"listaPDFs: {listaPDFs}")
+
+                summarizer = GerarDocumentoUtils(self.axiom_instance)
+
+                prompt_instance = Prompt()
+                prompt = prompt_instance.create_and_invoke_prompt(
+                    obj.prompt,
+                    dynamic_dict={"context": obj.user_text},
+                )
+
+                resposta_llm = (
+                    await summarizer.checar_se_resposta_vazia_do_documento_final(
+                        obj.llm_ultimas_requests, prompt.to_string()
+                    )
+                )
+                self.axiom_instance.send_axiom(f"resposta_llm: {resposta_llm}")
+
+                texto_completo_como_html = convert_markdown_to_HTML(
+                    resposta_llm
+                ).replace("resposta_segunda_etapa:", "<br><br>")
+
+                self.axiom_instance.send_axiom(
+                    f"texto_completo_como_html: {texto_completo_como_html}"
+                )
+
+                return Response({"resposta": texto_completo_como_html})
+        except Exception as e:
+            custom_exception_handler_without_api_handler(
+                e, serializer, self.axiom_instance
+            )
+            raise