_utils/resumo_completo_cursor.py

import os
from langchain_core.messages import HumanMessage
from typing import cast
from _utils.LLMs.LLM_class import LLM
from _utils.bubble_integrations.enviar_resposta_final import enviar_resposta_final
from _utils.custom_exception_handler import custom_exception_handler_wihout_api_handler
from _utils.gerar_relatorio_modelo_usuario.prompts import (
    prompt_gerar_query_dinamicamente,
)
from _utils.gerar_relatorio_modelo_usuario.GerarDocumento import (
    GerarDocumento,
)
from _utils.gerar_relatorio_modelo_usuario.contextual_retriever import (
    ContextualRetriever,
)
from _utils.gerar_relatorio_modelo_usuario.utils import (
    generate_document_title,
    gerar_resposta_compilada,
    get_full_text_and_all_PDFs_chunks,
    get_response_from_auxiliar_contextual_prompt,
)
from _utils.models.gerar_relatorio import (
    RetrievalConfig,
)
import markdown

from _utils.prompts.Prompt_class import Prompt
from _utils.utils import convert_markdown_to_HTML


def reciprocal_rank_fusion(result_lists, weights=None):
    """Combine multiple ranked lists using reciprocal rank fusion"""
    fused_scores = {}
    num_lists = len(result_lists)
    if weights is None:
        weights = [1.0] * num_lists

    for i in range(num_lists):
        for doc_id, score in result_lists[i]:
            if doc_id not in fused_scores:
                fused_scores[doc_id] = 0
            fused_scores[doc_id] += weights[i] * score

    # Sort by score in descending order
    sorted_results = sorted(fused_scores.items(), key=lambda x: x[1], reverse=True)

    return sorted_results


os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com"
os.environ.get("LANGCHAIN_API_KEY")
os.environ["LANGCHAIN_PROJECT"] = "VELLA"


async def get_llm_summary_answer_by_cursor_complete(
    serializer, listaPDFs, isBubble=False
):
    """Parâmetro "contexto" só deve ser passado quando quiser utilizar o teste com ragas, e assim, não quiser passar PDFs"""
    try:
        # Configuration
        config = RetrievalConfig(
            num_chunks=serializer["num_chunks_retrieval"],
            embedding_weight=serializer["embedding_weight"],
            bm25_weight=serializer["bm25_weight"],
            context_window=serializer["context_window"],
            chunk_overlap=serializer["chunk_overlap"],
        )

        contextual_retriever = ContextualRetriever(
            config, serializer["claude_context_model"]
        )

        # Initialize enhanced summarizer
        summarizer = GerarDocumento(
            config=config,
            embedding_model=serializer["hf_embedding"],
            chunk_overlap=serializer["chunk_overlap"],
            chunk_size=serializer["chunk_size"],
            num_k_rerank=serializer["num_k_rerank"],
            model_cohere_rerank=serializer["model_cohere_rerank"],
            # prompt_auxiliar=serializer["prompt_auxiliar"],
            gpt_model=serializer["model"],
            gpt_temperature=serializer["gpt_temperature"],
            prompt_gerar_documento=serializer["prompt_gerar_documento"],
            reciprocal_rank_fusion=reciprocal_rank_fusion,
        )

        all_PDFs_chunks, full_text_as_array, full_text_as_string = (
            await get_full_text_and_all_PDFs_chunks(
                listaPDFs,
                summarizer.splitter,
                serializer["should_use_llama_parse"],
                isBubble,
            )
        )

        is_contextualized_chunk = serializer["should_have_contextual_chunks"]

        if is_contextualized_chunk:
            response_auxiliar_summary = (
                await get_response_from_auxiliar_contextual_prompt(
                    full_text_as_array, full_text_as_string
                )
            )

            print("\nCOMEÇANDO A FAZER AS REQUISIÇÕES DO CONTEXTUAL")
            contextualized_chunks = await contextual_retriever.contextualize_all_chunks(
                all_PDFs_chunks, response_auxiliar_summary
            )
            print("\nTERMINOU DE FAZER TODAS AS REQUISIÇÕES DO CONTEXTUAL")
            chunks_processados = contextualized_chunks
        else:
            chunks_processados = all_PDFs_chunks

        llm = LLM()
        prompt_para_gerar_query_dinamico = prompt_gerar_query_dinamicamente(
            cast(str, response_auxiliar_summary)
        )

        print("\nCOMEÇANDO REQUISIÇÃO PARA GERAR O QUERY DINAMICAMENTE DO VECTOR STORE")
        query_gerado_dinamicamente_para_o_vector_store = await llm.google_gemini(
            "gemini-2.5-pro-exp-03-25"
        ).ainvoke([HumanMessage(content=prompt_para_gerar_query_dinamico)])

        print(
            "\nquery_gerado_dinamicamente_para_o_vector_store: ",
            query_gerado_dinamicamente_para_o_vector_store.content,
        )

        # Create enhanced vector store and BM25 index
        vector_store, bm25, chunk_ids = (
            summarizer.vector_store.create_enhanced_vector_store(
                chunks_processados, is_contextualized_chunk
            )
        )

        llm_ultimas_requests = serializer["llm_ultimas_requests"]
        print("\nCOMEÇANDO A FAZER ÚLTIMA REQUISIÇÃO")
        structured_summaries = await summarizer.gerar_documento_final(
            vector_store,
            bm25,
            chunk_ids,
            llm_ultimas_requests,
            cast(
                str, query_gerado_dinamicamente_para_o_vector_store.content
            ),  # prompt_auxiliar_SEM_CONTEXT,
        )
        print("\nTERMINOU DE FAZER A ÚLTIMA REQUISIÇÃO")

        if not isinstance(structured_summaries, list):
            from rest_framework.response import Response

            return Response({"erro": structured_summaries})

        texto_completo = summarizer.resumo_gerado + "\n\n"

        for x in structured_summaries:
            texto_completo = texto_completo + x["content"] + "\n"
            x["source"]["text"] = x["source"]["text"][0:200]
            x["source"]["context"] = x["source"]["context"][0:200]

        texto_completo_como_html = convert_markdown_to_HTML(texto_completo)
        print("\ntexto_completo_como_html", texto_completo_como_html)

        if is_contextualized_chunk:
            prompt_titulo_do_documento = response_auxiliar_summary
        else:
            prompt_titulo_do_documento = texto_completo_como_html
        titulo_do_documento = await generate_document_title(
            cast(str, prompt_titulo_do_documento)
        )

        if isBubble:
            print("COMEÇANDO A REQUISIÇÃO FINAL PARA O BUBBLE")
            enviar_resposta_final(
                serializer["doc_id"],
                serializer["form_response_id"],
                serializer["version"],
                texto_completo_como_html,
                False,
                cast(str, titulo_do_documento),
            )
            print("TERMINOU A REQUISIÇÃO FINAL PARA O BUBBLE")

        return {
            "texto_completo": texto_completo_como_html,
            "titulo_do_documento": titulo_do_documento,
            "resultado": structured_summaries,
            "parametros-utilizados": gerar_resposta_compilada(serializer),
        }
    except Exception as e:
        custom_exception_handler_wihout_api_handler(e, serializer)
        raise