# app.py — Isaac (CPU, Gemma 2 2B GGUF via llama-cpp-python) import os import json import gradio as gr from typing import List, Dict, Any from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.metrics.pairwise import cosine_similarity from llama_cpp import Llama # -------- Config -------- MODEL_PATH = os.environ.get("MODEL_PATH", "gemma-2-2b-it-Q4_K_M.gguf") # aponte para o GGUF local N_CTX = int(os.environ.get("N_CTX", "4096")) N_THREADS = int(os.environ.get("N_THREADS", "8")) N_BATCH = int(os.environ.get("N_BATCH", "128")) MAX_NEW_TOKENS = int(os.environ.get("MAX_NEW_TOKENS", "512")) TEMPERATURE = float(os.environ.get("TEMPERATURE", "0.7")) TOP_P = float(os.environ.get("TOP_P", "0.9")) EPC_MD_PATH = os.environ.get("EPC_MD_PATH", "epct0-3.md") SYSTEM_RULES = ( "Regras: 1) Sempre pergunte objetivo e contexto do pedido antes de responder; " "2) Após entender o 'porquê/para quê', aplique um checklist filosófico (controle, impermanência, obstáculos, julgamentos, prioridades, papéis) e só então responda; " "3) Explique a relação entre as diretrizes escolhidas e o plano; 4) Seja claro e conciso." ) # -------- Carregar modelo (CPU) -------- llm = Llama( model_path=MODEL_PATH, chat_format="gemma", # usa o formato nativo de chat do Gemma n_ctx=N_CTX, n_threads=N_THREADS, n_batch=N_BATCH, verbose=False, ) # -------- Parser + RAG leve -------- def parse_epct_table(md_path: str) -> List[Dict[str, Any]]: rows = [] with open(md_path, "r", encoding="utf-8") as f: for line in f: line = line.strip() if line.startswith("|") and not set(line).issubset(set("|:- ")): parts = [c.strip() for c in line.strip("|").split("|")] if len(parts) >= 4 and parts[0] != "Id": rows.append({ "id": parts[0], "resumo": parts[1], "porques": parts[2], "tags": parts[3] }) return rows DOCS = parse_epct_table(EPC_MD_PATH) DOC_TEXTS = [f"{d['resumo']} {d['porques']} {d['tags']}" for d in DOCS] VECTORIZER = TfidfVectorizer(stop_words="portuguese", max_features=4096) DOC_MATRIX = VECTORIZER.fit_transform(DOC_TEXTS) def retrieve_guidelines(query: str, k: int = 4) -> List[Dict[str, Any]]: if not query.strip(): return [] qv = VECTORIZER.transform([query]) sims = cosine_similarity(qv, DOC_MATRIX)[0] idxs = sims.argsort()[::-1][:k] return [DOCS[i] for i in idxs] def format_guidelines(items: List[Dict[str, Any]]) -> str: if not items: return "- (nenhuma diretriz selecionada)" return "\n".join([f"- [{d['id']}] {d['resumo']} | Porquês: {d['porques']} | Tags: {d['tags']}" for d in items]) # -------- Mensagens e geração -------- def make_messages(initial_request: str, intent: str, selected_items: List[Dict[str, Any]]) -> List[Dict[str, str]]: guidelines_block = format_guidelines(selected_items) user_compound = ( f"Pedido original:\n{initial_request}\n\n" f"Objetivo/porquê declarado:\n{intent}\n\n" f"Diretrizes selecionadas (do corpus):\n{guidelines_block}\n\n" "Tarefa:\n1) Produza uma 'Preparação' breve (itens do checklist aplicados ao caso).\n" "2) Em seguida, produza 'Resposta' com passos claros e proporcionais ao objetivo.\n" "3) Mostre a conexão entre as diretrizes e a recomendação.\n" ) return [ {"role": "system", "content": SYSTEM_RULES}, {"role": "user", "content": user_compound}, ] def generate(messages: List[Dict[str, str]]) -> str: out = llm.create_chat_completion( messages=messages, temperature=TEMPERATURE, top_p=TOP_P, max_tokens=MAX_NEW_TOKENS, ) return out["choices"][0]["message"]["content"].strip() # -------- Máquina de estados -------- def chat_turn(user_input: str, state: Dict[str, Any]): if state is None or "stage" not in state: state = {"stage": "await_intent", "initial_request": "", "intent": "", "items": []} if state["stage"] == "await_intent": state["initial_request"] = user_input reply = ("Antes de responder, poderia explicar seu objetivo e contexto? " "O que você quer alcançar e por quê isso é importante agora?") state["stage"] = "collect_intent" return reply, state if state["stage"] == "collect_intent": state["intent"] = user_input items = retrieve_guidelines(user_input, k=4) state["items"] = items scaffold = [ "Preparação (rascunho):", "- Controle: separe o que depende de você e foque no controlável.", "- Perspectiva: identifique julgamentos que amplificam o problema.", "- Obstáculos: antecipe contratempos realistas e pré-compromissos.", "- Prioridades/Papéis: alinhe ação ao essencial e ao seu papel atual.", "", "Diretrizes candidatas:", format_guidelines(items), "", "Se estiver de acordo, diga 'prossiga' para receber a resposta estruturada." ] state["stage"] = "ready" return "\n".join(scaffold), state if state["stage"] == "ready": if user_input.strip().lower() not in {"prossiga", "ok", "pode seguir", "pode prosseguir"}: return "Confirme com 'prossiga' para gerar a resposta final baseada no preparo acima.", state messages = make_messages(state["initial_request"], state["intent"], state["items"]) text = generate(messages) state = {"stage": "await_intent", "initial_request": "", "intent": "", "items": []} return text, state state = {"stage": "await_intent", "initial_request": "", "intent": "", "items": []} return "Vamos recomeçar: descreva seu pedido inicial.", state # -------- UI -------- with gr.Blocks(title="Isaac — CPU (Gemma 2 2B GGUF)") as demo: gr.Markdown("# Isaac — CPU (Gemma 2 2B)\nFluxo: Intenção → Preparação → Resposta.") state = gr.State({"stage": "await_intent", "initial_request": "", "intent": "", "items": []}) chat = gr.Chatbot(height=480) inp = gr.Textbox(label="Mensagem", placeholder="Descreva seu pedido...") send = gr.Button("Enviar") def ui_handle(user_msg, s): reply, s2 = chat_turn(user_msg or "", s) return chat + [[user_msg, reply]], s2, "" send.click(ui_handle, [inp, state], [chat, state, inp]) inp.submit(ui_handle, [inp, state], [chat, state, inp]) if __name__ == "__main__": demo.launch(server_name="0.0.0.0", server_port=int(os.environ.get("PORT", 7860)))