Update app.py

app.py

@@ -1,58 +1,59 @@
-# app.py — Isaac (Gemma 2 9B + Intent Gateway + RAG Filosófico)
+# app.py — Isaac (CPU, Gemma 2 2B GGUF via llama-cpp-python)
 import os
-import re
 import json
-import torch
 import gradio as gr
-from typing import List, Dict, Any, Tuple
+from typing import List, Dict, Any
 
-from transformers import (
-    AutoModelForCausalLM,
-    AutoTokenizer,
-    BitsAndBytesConfig,
-)
+from sklearn.feature_extraction.text import TfidfVectorizer
+from sklearn.metrics.pairwise import cosine_similarity
 
-# -------- Config do Modelo --------
-MODEL_ID = os.environ.get("MODEL_ID", "google/gemma-2-9b-it")  # requer aceitar termos no Hub
-USE_4BIT = os.environ.get("USE_4BIT", "1") == "1"
+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")
 
-# -------- Fonte filosófica --------
-EPC_MD_PATH = os.environ.get("EPC_MD_PATH", "epct0.md")  # seu arquivo já existente
+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."
+)
 
-# -------- Recuperação simples (TF-IDF) --------
-from sklearn.feature_extraction.text import TfidfVectorizer
-from sklearn.metrics.pairwise import cosine_similarity
+# -------- 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]]:
-    """
-    Lê uma tabela Markdown com colunas: Id | Resumo | Porquês | Tags
-    e retorna uma lista de itens com esses campos.
-    """
-    with open(md_path, "r", encoding="utf-8") as f:
-        text = f.read()
-
     rows = []
-    for line in text.splitlines():
-        line = line.strip()
-        # linhas de dados começam com '|' e têm pelo menos 4 colunas
-        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]
-                })
+    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
-]
+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)
 
@@ -65,37 +66,11 @@ def retrieve_guidelines(query: str, k: int = 4) -> List[Dict[str, Any]]:
     return [DOCS[i] for i in idxs]
 
 def format_guidelines(items: List[Dict[str, Any]]) -> str:
-    lines = []
-    for d in items:
-        lines.append(f"- [{d['id']}] {d['resumo']} | Porquês: {d['porques']} | Tags: {d['tags']}")
-    return "\n".join(lines) if lines else "- (nenhuma diretriz selecionada)"
-
-# -------- Carregamento do Modelo --------
-bnb_config = None
-if USE_4BIT:
-    bnb_config = BitsAndBytesConfig(
-        load_in_4bit=True,
-        bnb_4bit_use_double_quant=True,
-        bnb_4bit_quant_type="nf4",
-        bnb_4bit_compute_dtype=torch.float16,
-    )
-
-tokenizer = AutoTokenizer.from_pretrained(MODEL_ID, use_fast=True)
-model = AutoModelForCausalLM.from_pretrained(
-    MODEL_ID,
-    device_map="auto",
-    torch_dtype=torch.float16 if not USE_4BIT else None,
-    quantization_config=bnb_config,
-)
-
-# -------- Prompting / Template --------
-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) Seja específico e explique a relação entre as diretrizes escolhidas e o plano proposto; "
-    "4) Mantenha tom claro e conciso."
-)
+    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 = (
@@ -111,44 +86,28 @@ def make_messages(initial_request: str, intent: str, selected_items: List[Dict[s
         {"role": "user", "content": user_compound},
     ]
 
-def generate_with_template(messages: List[Dict[str, str]]) -> str:
-    input_ids = tokenizer.apply_chat_template(
-        messages,
-        add_generation_prompt=True,
-        return_tensors="pt"
-    ).to(model.device)
-
-    outputs = model.generate(
-        input_ids=input_ids,
-        max_new_tokens=MAX_NEW_TOKENS,
-        do_sample=True,
+def generate(messages: List[Dict[str, str]]) -> str:
+    out = llm.create_chat_completion(
+        messages=messages,
         temperature=TEMPERATURE,
         top_p=TOP_P,
-        eos_token_id=tokenizer.eos_token_id,
-        pad_token_id=tokenizer.eos_token_id,
+        max_tokens=MAX_NEW_TOKENS,
     )
-    text = tokenizer.decode(outputs[0][input_ids.shape[-1]:], skip_special_tokens=True)
-    return text.strip()
+    return out["choices"][0]["message"]["content"].strip()
 
-# -------- Lógica de Conversa (Gateway de Intenção) --------
-# Estados por sessão: {"stage": "await_intent" | "ready", "initial_request": str, "intent": str, "items": [...]}
+# -------- 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":
-        # Primeiro turno: tratar user_input como pedido inicial e perguntar intenção
         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?"
-        )
-        # permanecer aguardando intenção do usuário
+        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":
-        # Recebe o 'porquê/para quê', seleciona diretrizes e mostra scaffold
         state["intent"] = user_input
         items = retrieve_guidelines(user_input, k=4)
         state["items"] = items
@@ -168,29 +127,20 @@ def chat_turn(user_input: str, state: Dict[str, Any]):
         return "\n".join(scaffold), state
 
     if state["stage"] == "ready":
-        # Gera resposta final usando o modelo, com as diretrizes no contexto
         if user_input.strip().lower() not in {"prossiga", "ok", "pode seguir", "pode prosseguir"}:
-            # exigir confirmação simples para manter o fluxo
             return "Confirme com 'prossiga' para gerar a resposta final baseada no preparo acima.", state
 
-        messages = make_messages(
-            state["initial_request"],
-            state["intent"],
-            state["items"]
-        )
-        out = generate_with_template(messages)
-
-        # reset ciclo
+        messages = make_messages(state["initial_request"], state["intent"], state["items"])
+        text = generate(messages)
         state = {"stage": "await_intent", "initial_request": "", "intent": "", "items": []}
-        return out, state
+        return text, state
 
-    # fallback defensivo
     state = {"stage": "await_intent", "initial_request": "", "intent": "", "items": []}
     return "Vamos recomeçar: descreva seu pedido inicial.", state
 
 # -------- UI --------
-with gr.Blocks(title="Isaac — Gemma 2 9B") as demo:
-    gr.Markdown("# Isaac — Gemma 2 9B\nFluxo: Intenção → Preparação → Resposta.", elem_id="title")
+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...")
@@ -204,6 +154,4 @@ with gr.Blocks(title="Isaac — Gemma 2 9B") as demo:
     inp.submit(ui_handle, [inp, state], [chat, state, inp])
 
 if __name__ == "__main__":
-    # gradio server
     demo.launch(server_name="0.0.0.0", server_port=int(os.environ.get("PORT", 7860)))
-