Delete app.py

app.py

@@ -1,368 +0,0 @@
-# -*- coding: utf-8 -*-
-"""
-Pipeline v10 - VERSÃO FINAL COM LÓGICA DE RETOMADA REATIVADA.
-
-Este script corrige a desativação acidental da funcionalidade de resumir a
-pipeline após uma pausa. Agora, o sistema está totalmente interativo.
-
-PRINCIPAIS FUNCIONALIDADES:
--   LÓGICA DE RETOMADA (RESUMIR) REATIVADA: O orquestrador agora chama
-    corretamente a função `resumir_pipeline` quando o estado está 'paused',
-    garantindo que a resposta do usuário seja usada para continuar o raciocínio.
--   SANITIZAÇÃO DE TEXTO: Mantida para robustez.
--   ROBUSTEZ COM TRY/EXCEPT: Mantida para tratamento de erros.
--   BYPASS INTELIGENTE (FAST PATH): Mantido para respostas rápidas.
--   PIPELINE GENERALIZADA: Mantida para versatilidade.
--   LOGS DE API: Mantidos para depuração.
-"""
-
-# ============================================================================
-# 1. IMPORTAÇÕES E CONFIGURAÇÃO INICIAL
-# ============================================================================
-import json
-import os
-import re
-import warnings
-from datetime import datetime
-from typing import Dict, List, Tuple, Any
-
-import gradio as gr
-import google.generativeai as genai
-
-warnings.filterwarnings("ignore", category=FutureWarning, module="google.api_core")
-
-# --- Configuração da API ---
-API_KEY = os.getenv("GOOGLE_API_KEY")
-if not API_KEY:
-    raise ValueError("A variável de ambiente GOOGLE_API_KEY não foi configurada.")
-
-genai.configure(api_key=API_KEY)
-
-# --- Definição dos Modelos ---
-COUNSELOR_MODEL = genai.GenerativeModel("gemini-2.5-flash")
-SUPERVISOR_MODEL = genai.GenerativeModel("gemini-2.5-pro")
-
-# --- Título da Interface ---
-TITLE = "# 🚀 Pipeline v10 | Raciocínio Adaptativo e Interativo\n**Com 'Fast Path', Pausa/Retomada e tratamento de erros.**"
-
-# ============================================================================
-# 2. PROMPTS CENTRALIZADOS
-# ============================================================================
-# (Os prompts permanecem os mesmos da versão anterior, sem alterações)
-PROMPTS = {
-    "P1_TRIAGEM": """
-METACOGNIÇÃO - TRIAGEM INICIAL.
-PERGUNTA: {pergunta}
----
-Analise a pergunta do usuário e classifique-a de acordo com o tipo e a sua confiança para respondê-la de forma direta.
-
-Use as seguintes categorias:
-- tipo: 'factual' (pede um fato concreto), 'objetiva_tecnica' (pede uma resposta técnica com uma única resposta correta), 'subjetiva_complexa' (envolve opinião, análise, múltiplos fatores).
-- confianca: 'extrema' (certeza absoluta, como "Qual a capital do Brasil?"), 'alta', 'media', 'baixa', 'insuficiente'.
-
-RETORNE JSON:
-{{
-  "tipo": "...",
-  "confianca": "...",
-  "decisao": "Com base na classificação, decida se deve 'responder_direto' (para factual/extrema ou objetiva_tecnica com confiança alta/extrema) ou 'analisar_profundamente' (para todas as outras)."
-}}
-""",
-    "GERAR_RESPOSTA_DIRETA": """
-TAREFA: Resposta Direta (Bypass)
-A pergunta do usuário foi classificada como factual com extrema certeza ou objetiva-técnica.
-PERGUNTA: "{pergunta}"
----
-Forneça uma resposta direta, precisa e concisa.
-
-RETORNE JSON:
-{{"resposta_direta": "Sua resposta concisa aqui."}}
-""",
-    "JUSTIFICAR_BYPASS": """
-METACOGNIÇÃO - JUSTIFICATIVA DE BYPASS.
-ANÁLISE DA TRIAGEM (P1): {p1}
----
-Você está pulando a pipeline de raciocínio profundo. Justifique esta decisão.
-
-RETORNE JSON:
-{{
-  "justificativa_bypass": {{
-    "motivo": "A pergunta foi identificada como [tipo da pergunta] para a qual a confiança de uma resposta direta é [nível de confiança].",
-    "acao_tomada": "A pipeline de análise profunda foi pulada para fornecer uma resposta mais rápida e eficiente.",
-    "proximo_passo": "A resposta direta será submetida a uma verificação final de qualidade."
-  }}
-}}
-""",
-    "P2_CENARIOS": """
-METACOGNIÇÃO - GERAÇÃO DE CENÁRIOS.
-ANÁLISE (P1): {p1}, PERGUNTA: {pergunta}
----
-Gere o cenário de interpretação mais provável e um cenário alternativo (improvável, mas plausível).
-RETORNE JSON:
-{{
-  "cenarios": {{
-    "provaveis": [{{"id": "C1_PROVAVEL", "desc": "Descrição do cenário mais provável."}}],
-    "improvaveis": [{{"id": "C2_IMPROVAVEL", "desc": "Descrição do cenário alternativo."}}]
-  }}, "decisao": "prosseguir"
-}}
-""",
-    "P4_CRUZAR_VALIDACOES": """
-METACOGNIÇÃO - ABSTRAÇÃO DE CONHECcimento.
-ANÁLISES ANTERIORES: {p1}, {p2}, {p3}
----
-Identifique o princípio fundamental que rege a discussão.
-RETORNE JSON: {{"principio_central": "Descrição do princípio identificado."}}
-""",
-    "P5_LACUNAS_FINAIS": """
-METACOGNIÇÃO - ANÁLISE DE INCERTEZA.
-PRINCÍPIO CENTRAL (P4): {p4}
-PERGUNTA ORIGINAL: {pergunta}
----
-Com base no princípio central, identifique a principal lacuna de informação. Formule uma pergunta clara para o usuário.
-
-RETORNE JSON:
-{{
-  "pontos_de_incerteza": ["A principal dúvida que ainda resta."],
-  "decisao_interna": "questionar",
-  "pergunta_chave_para_usuario": "A pergunta clara e contextual que você fará ao usuário."
-}}
-""",
-    "P7_SINTETIZAR": """
-SINTETIZADOR.
-DADOS DO JULGAMENTO (P6): {p6}
----
-Converta a análise técnica em uma resposta final coesa.
-RETORNE JSON: {{"resposta": "O texto da sua resposta final aqui."}}
-""",
-    "P8_VERIFICAR": """
-VERIFICADOR FINAL.
-RESPOSTA A SER VERIFICADA: {resposta_a_verificar}
----
-Realize uma verificação tripla (factual, lógica, clareza). Se houver problemas, corrija-os.
-RETORNE JSON:
-{{
-  "todas_aprovadas": true|false,
-  "problemas_identificados": ["..."],
-  "resposta_corrigida": "O texto da resposta corrigida, se necessário. Senão, null."
-}}
-"""
-}
-# ============================================================================
-# 3. CLASSES E FUNÇÕES HELPERS
-# ============================================================================
-
-class Logger:
-    # ... (código do Logger sem alterações) ...
-    def __init__(self, verbose: bool = True): self.verbose = verbose
-    def log(self, msg: str, level: str = "INFO"):
-        log_msg = f"[{datetime.now().strftime('%H:%M:%S')}] [{level.upper()}] {msg}"; print(log_msg)
-        if level.upper() in ["TASK", "START", "SUCCESS", "ERROR", "WARN"]: print("=" * 70)
-
-logger = Logger(verbose=True)
-
-def sanitizar_texto(texto: str) -> str:
-    """Limpa e normaliza uma string para garantir um formato seguro e consistente."""
-    if not isinstance(texto, str): return ""
-    texto_limpo = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f]', '', texto)
-    texto_limpo = re.sub(r'\s{2,}', ' ', texto_limpo).replace('\\n', '\n')
-    return texto_limpo.strip()
-
-def chamar_gemini_json(modelo: genai.GenerativeModel, prompt: str, temperatura: float = 0.4, max_tokens: int = 2048) -> Dict:
-    prompt_completo = f"{prompt}\n\n---\n\n**INSTRUÇÃO OBRIGATÓRIA: Sua resposta DEVE ser um único e válido objeto JSON.**"
-    print(f"\n{'='*25} 💬 API INPUT PARA [{modelo.model_name}] {'='*25}\n{prompt_completo}\n{'='*78}\n")
-    try:
-        response = modelo.generate_content(prompt_completo, generation_config=genai.types.GenerationConfig(temperature=temperatura, max_output_tokens=max_tokens))
-        resposta_bruta = response.text or ""
-        print(f"\n{'='*25} 📥 API RAW OUTPUT DE [{modelo.model_name}] {'='*25}\n{resposta_bruta}\n{'='*78}\n")
-        
-        if not resposta_bruta.strip():
-            logger.log("A API Gemini retornou uma resposta vazia. Causa provável: Filtros de segurança.", "WARN")
-            return {"erro": "API_EMPTY_RESPONSE", "causa_provavel": "Filtro de segurança do modelo."}
-        
-        match = re.search(r'\{.*\}', resposta_bruta, re.DOTALL)
-        if not match:
-            return {"erro": "JSON_NOT_FOUND", "detalhes": "Nenhum objeto JSON encontrado na resposta da API."}
-        
-        return json.loads(match.group(0))
-    except Exception as e:
-        logger.log(f"Falha na chamada da API ou no parse do JSON: {e}", "ERROR")
-        return {"erro": "API_CALL_OR_PARSE_FAILED", "detalhes": str(e)}
-
-def criar_dna() -> Dict:
-    return { "historico_chat": [], "meta": {"total_turnos": 0}, "pipeline_state": { "status": "completed", "paused_at_step": None, "saved_data": {} } }
-
-# ============================================================================
-# 4. PASSOS DA PIPELINE
-# ============================================================================
-# (Funções de passo permanecem as mesmas)
-def passo_1_triagem(pergunta: str) -> Dict:
-    logger.log("📊 PASSO 1: TRIAGEM INICIAL", "TASK")
-    return chamar_gemini_json(COUNSELOR_MODEL, PROMPTS["P1_TRIAGEM"].format(pergunta=pergunta))
-def passo_gerar_resposta_direta(pergunta: str) -> Dict:
-    logger.log("⚡ FAST PATH: GERANDO RESPOSTA DIRETA", "TASK")
-    return chamar_gemini_json(COUNSELOR_MODEL, PROMPTS["GERAR_RESPOSTA_DIRETA"].format(pergunta=pergunta))
-def passo_justificar_bypass(p1: Dict) -> Dict:
-    logger.log("⚡ FAST PATH: GERANDO JUSTIFICATIVA", "TASK")
-    return chamar_gemini_json(COUNSELOR_MODEL, PROMPTS["JUSTIFICAR_BYPASS"].format(p1=json.dumps(p1)))
-def passo_2_cenarios(pergunta: str, p1: Dict) -> Dict:
-    logger.log("🧠 PASSO 2: GERAÇÃO DE CENÁRIOS", "TASK")
-    return chamar_gemini_json(COUNSELOR_MODEL, PROMPTS["P2_CENARIOS"].format(p1=json.dumps(p1), pergunta=pergunta))
-def passo_4_cruzar_validacoes(p1: Dict, p2: Dict, p3: Dict) -> Dict:
-    logger.log("🧠 PASSO 4: IDENTIFICAÇÃO DO PRINCÍPIO CENTRAL", "TASK")
-    return chamar_gemini_json(COUNSELOR_MODEL, PROMPTS["P4_CRUZAR_VALIDACOES"].format(p1=json.dumps(p1), p2=json.dumps(p2), p3=json.dumps(p3)))
-def passo_5_lacunas_finais(pergunta: str, p4: Dict) -> Dict:
-    logger.log("🧠 PASSO 5: ANÁLISE DE INCERTEZA", "TASK")
-    return chamar_gemini_json(COUNSELOR_MODEL, PROMPTS["P5_LACUNAS_FINAIS"].format(p4=json.dumps(p4), pergunta=pergunta))
-def passo_8_verificar(resposta_a_verificar: str) -> Dict:
-    logger.log("✅ PASSO 8: VERIFICAÇÃO FINAL", "TASK")
-    return chamar_gemini_json(SUPERVISOR_MODEL, PROMPTS["P8_VERIFICAR"].format(resposta_a_verificar=resposta_a_verificar))
-def passo_3_isolar_cenarios(p2: Dict) -> Dict: return {"simulado": True}
-def passo_6_ponderar(p2: Dict, p4: Dict, p5: Dict) -> Dict: return {"simulado": True}
-def passo_7_sintetizar(p6: Dict) -> Dict: return {"resposta": "Resposta vinda da pipeline completa."}
-
-# ============================================================================
-# 5. ORQUESTRADOR PRINCIPAL
-# ============================================================================
-
-def iniciar_nova_pipeline(pergunta_original: str, historico: List[Dict], anexo: Any, dna: Dict) -> Tuple[str, List, Dict]:
-    pergunta = sanitizar_texto(pergunta_original)
-    logger.log(f"INICIANDO NOVA PIPELINE (Input Sanitizado): '{pergunta[:70]}...'", "START")
-    
-    p1 = passo_1_triagem(pergunta)
-    if "erro" in p1: return f"Erro na Triagem: {p1.get('detalhes', 'Não foi possível classificar a pergunta.')}", historico, dna
-
-    if p1.get("decisao") == "responder_direto":
-        logger.log("DECISÃO: Tomar o Caminho Rápido (Bypass).", "INFO")
-        resposta_direta_data = passo_gerar_resposta_direta(pergunta)
-        justificativa_data = passo_justificar_bypass(p1)
-
-        if "erro" in resposta_direta_data or "erro" in justificativa_data:
-            return "Erro ao gerar a resposta direta.", historico, dna
-            
-        resposta_direta = resposta_direta_data.get("resposta_direta", "Não foi possível gerar a resposta.")
-        justificativa = justificativa_data.get("justificativa_bypass", {})
-        
-        verificacao = passo_8_verificar(resposta_direta)
-        resposta_final = verificacao.get("resposta_corrigida") or resposta_direta
-        
-        justificativa_texto = f"**JUSTIFICATIVA DE RESPOSTA DIRETA:**\n- **Motivo:** {justificativa.get('motivo', 'N/A')}\n- **Ação:** {justificativa.get('acao_tomada', 'N/A')}\n\n---\n"
-        resposta_formatada = justificativa_texto + resposta_final
-        
-        novo_historico = historico + [{"role": "user", "content": pergunta_original}, {"role": "assistant", "content": resposta_formatada}]
-        logger.log("PIPELINE (FAST PATH) CONCLUÍDA", "SUCCESS")
-        return "PIPELINE_COMPLETED", novo_historico, dna
-
-    else:
-        logger.log("DECISÃO: Tomar o Caminho Completo de Análise Profunda.", "INFO")
-        p2 = passo_2_cenarios(pergunta, p1)
-        p3 = passo_3_isolar_cenarios(p2)
-        p4 = passo_4_cruzar_validacoes(p1, p2, p3)
-        p5 = passo_5_lacunas_finais(pergunta, p4)
-
-        if p5.get("decisao_interna") == "questionar":
-            logger.log("INTERRUPÇÃO no P5. Salvando estado no DNA.", "WARN")
-            dna['pipeline_state'] = {"status": "paused", "paused_at_step": "P5", "saved_data": {'p1':p1, 'p2':p2, 'p3':p3, 'p4':p4, 'pergunta_original': pergunta, 'historico_original': historico}}
-            pergunta_do_bot = p5.get('pergunta_chave_para_usuario', 'Preciso de mais informações.')
-            historico_atualizado = historico + [{"role": "user", "content": pergunta_original}, {"role": "assistant", "content": pergunta_do_bot}]
-            return "PIPELINE_PAUSED", historico_atualizado, dna
-
-        p6 = passo_6_ponderar(p2, p4, p5)
-        p7 = passo_7_sintetizar(p6)
-        p8 = passo_8_verificar(p7.get("resposta", ""))
-        resposta_final = p8.get("resposta_corrigida") or p7.get("resposta", "Não foi possível gerar a resposta.")
-        novo_historico = historico + [{"role": "user", "content": pergunta_original}, {"role": "assistant", "content": resposta_final}]
-        logger.log("PIPELINE (COMPLETA) CONCLUÍDA", "SUCCESS")
-        return "PIPELINE_COMPLETED", novo_historico, dna
-
-def resumir_pipeline(esclarecimento_usuario_original: str, dna: Dict) -> Tuple[str, List, Dict]:
-    esclarecimento_usuario = sanitizar_texto(esclarecimento_usuario_original)
-    logger.log(f"RESUMINDO PIPELINE com esclarecimento: '{esclarecimento_usuario[:70]}...'", "START")
-    
-    estado_salvo = dna['pipeline_state']['saved_data']
-    ponto_parada = dna['pipeline_state']['paused_at_step']
-    historico = estado_salvo['historico_original']
-    pergunta_original = estado_salvo['pergunta_original']
-    pergunta_atualizada = f"{pergunta_original}\n\n---ESCLARECIMENTO ADICIONAL DO USUÁRIO---\n{esclarecimento_usuario}"
-    
-    dna['pipeline_state'] = criar_dna()['pipeline_state'] # Limpa o estado de pausa
-
-    # Continua do Ponto de Parada
-    if ponto_parada == 'P5':
-        logger.log("Retomando a partir do passo P6...", "INFO")
-        p1, p2, p3, p4 = estado_salvo['p1'], estado_salvo['p2'], estado_salvo['p3'], estado_salvo['p4']
-        
-        # Reavalia as lacunas com a nova informação, mas agora com o objetivo de responder
-        p5_novo = passo_5_lacunas_finais(pergunta_atualizada, p4)
-        p5_novo['decisao_interna'] = 'responder' # Força a continuação
-        
-        p6 = passo_6_ponderar(p2, p4, p5_novo)
-    else:
-        # Fallback para outros pontos de parada (se implementados)
-        raise ValueError(f"Ponto de retomada não implementado: {ponto_parada}")
-    
-    p7 = passo_7_sintetizar(p6)
-    p8 = passo_8_verificar(p7.get("resposta", ""))
-    resposta_final = p8.get("resposta_corrigida") or p7.get("resposta", "Não foi possível gerar uma resposta.")
-    
-    # Monta o histórico final para exibição
-    historico_da_pausa = historico + [{"role": "user", "content": pergunta_original}, {"role": "assistant", "content": estado_salvo.get('pergunta_chave_para_usuario', '...')}]
-    novo_historico = historico_da_pausa + [{"role": "user", "content": esclarecimento_usuario_original}, {"role": "assistant", "content": resposta_final}]
-    
-    logger.log("PIPELINE (RESUMIDA) CONCLUÍDA", "SUCCESS")
-    return "PIPELINE_COMPLETED", novo_historico, dna
-
-def executar_pipeline(pergunta: str, historico: List[Dict], anexo: Any, dna: Dict) -> Tuple[str, List, Dict]:
-    try:
-        if 'pipeline_state' not in dna: dna.update(criar_dna())
-        
-        # CORREÇÃO: A chamada para 'resumir_pipeline' foi reativada.
-        if dna['pipeline_state']['status'] == 'paused':
-            return resumir_pipeline(pergunta, dna)
-        else:
-            return iniciar_nova_pipeline(pergunta, historico, anexo, dna)
-            
-    except Exception as e:
-        logger.log(f"Erro catastrófico no orquestrador: {e}", "ERROR")
-        resposta_erro = f"Ocorreu um erro inesperado na pipeline: {e}"
-        return "PIPELINE_ERROR", historico + [{"role": "user", "content": pergunta}, {"role": "assistant", "content": resposta_erro}], dna
-
-# ============================================================================
-# 6. INTERFACE COM GRADIO
-# ============================================================================
-def converter_historico_para_gradio(historico_interno: List[Dict]) -> List[List[str]]:
-    historico_gradio = []
-    for i in range(0, len(historico_interno), 2):
-        if i + 1 < len(historico_interno):
-            historico_gradio.append([historico_interno[i]['content'], historico_interno[i+1]['content']])
-    return historico_gradio
-
-def chat_interface(pergunta: str, historico_gradio: List[List[str]], anexo: Any, dna_json_str: str) -> Tuple[List, str, str, None]:
-    try:
-        dna = json.loads(dna_json_str) if dna_json_str and dna_json_str.strip() else criar_dna()
-    except:
-        dna = criar_dna()
-        
-    historico_interno = [item for turno in historico_gradio for item in ({"role": "user", "content": turno[0]}, {"role": "assistant", "content": turno[1]}) if turno and item['content']]
-
-    _ , novo_historico_para_exibir, dna_atualizado = executar_pipeline(pergunta, historico_interno, anexo, dna)
-    
-    novo_historico_gradio = converter_historico_para_gradio(novo_historico_para_exibir)
-
-    return novo_historico_gradio, "", json.dumps(dna_atualizado, indent=2, ensure_ascii=False), None
-
-if __name__ == "__main__":
-    with gr.Blocks(title="Pipeline v10 - Raciocínio Adaptativo", theme=gr.themes.Soft()) as demo:
-        gr.Markdown(TITLE)
-        with gr.Row():
-            with gr.Column(scale=3):
-                chatbot = gr.Chatbot(label="Chat", height=600, bubble_full_width=False)
-                input_textbox = gr.Textbox(label="Digite sua pergunta...", lines=3)
-                with gr.Row():
-                    submit_button = gr.Button("🚀 Enviar", variant="primary", scale=1)
-                    file_upload = gr.File(label="Anexar Arquivo", scale=1)
-            with gr.Column(scale=2):
-                dna_view = gr.Code(label="DNA (Estado da Conversa)", language="json", interactive=False, value=json.dumps(criar_dna(), indent=2, ensure_ascii=False))
-        dna_json_hidden = gr.Textbox(value=json.dumps(criar_dna()), visible=False)
-        submit_button.click(fn=chat_interface, inputs=[input_textbox, chatbot, file_upload, dna_json_hidden], outputs=[chatbot, input_textbox, dna_json_hidden, file_upload])
-        input_textbox.submit(fn=chat_interface, inputs=[input_textbox, chatbot, file_upload, dna_json_hidden], outputs=[chatbot, input_textbox, dna_json_hidden, file_upload])
-        dna_json_hidden.change(fn=lambda x: x, inputs=[dna_json_hidden], outputs=[dna_view])
-    demo.launch(server_name="0.0.0.0", server_port=7860, share=False)