financebot-ifrj / app.py
IgorValpassos's picture
Update app.py
abaf2f8 verified
# app.py — FinanceBot IFRJ (PT-BR, assertivo e rápido | Gradio + ctransformers)
# 100% grátis (modelo GGUF local), com streaming e "skills" determinísticas
# Tópicos: Juros (simples/compostos), Price x SAC, Equivalência de taxas,
# Aposentadoria (PMT/FV e renda-alvo), PGBL x VGBL, Renda Fixa x Variável, Armadilhas comuns
import os, re, json, math, time
import gradio as gr
# ==========================
# Identidade e comportamento
# ==========================
SYSTEM_PROMPT = (
"Você é o FinanceBot IFRJ 💚 — tutor claro, paciente e assertivo para estudantes do ensino médio. "
"SEMPRE responda em português do Brasil, mesmo se a pergunta vier em outro idioma. "
"Foque em educação financeira e matemática financeira do cotidiano (orçamento, juros simples e compostos, "
"Price x SAC, investimentos básicos de renda fixa e variável, PGBL x VGBL, aposentadoria). "
"Seja objetivo, explique termos rapidamente, mostre passos curtos e exemplos numéricos quando útil. "
"Evite recomendações personalizadas de investimento. Em dúvidas tributárias, fale em termos gerais. "
"Sempre que possível, apresente fórmulas em LaTeX entre $$...$$ e resultados com separador brasileiro."
)
OFFTOPIC_MSG = (
"Vamos focar em **educação financeira** e **matemática financeira** do dia a dia. "
"Posso ajudar com: orçamento pessoal, **juros** (simples/compostos), **Price x SAC**, "
"**equivalência de taxas**, **renda fixa x variável**, **PGBL x VGBL**, **aposentadoria** etc."
)
# ==========================
# LaTeX e UI
# ==========================
latex_delimiter_set = [
{"left": "\\begin{equation}", "right": "\\end{equation}", "display": True},
{"left": "\\begin{align}", "right": "\\end{align}", "display": True},
{"left": "\\begin{alignat}", "right": "\\end{alignat}", "display": True},
{"left": "\\begin{gather}", "right": "\\end{gather}", "display": True},
{"left": "\\begin{CD}", "right": "\\end{CD}", "display": True},
{"left": "$$", "right": "$$", "display": True},
{"left": "\\[", "right": "\\]", "display": True},
]
# Tema (IFRJ green) — sem .set(...) com tokens não suportados
theme = gr.themes.Soft(primary_hue="green", secondary_hue="emerald")
# (Opcional) Analytics / header extra + garantir tema claro no load
html_header = """
<script defer src="https://cloud.umami.is/script.js" data-website-id="28b59ada-b311-4b77-85ba-9243a591bac2"></script>
<script>
document.addEventListener('DOMContentLoaded', function () {
try { localStorage.setItem('theme', 'light'); } catch(e) {}
document.body.classList.remove('dark');
document.body.classList.add('light');
});
</script>
"""
# CSS compatível com versões antigas do Gradio
custom_css = """
:root { --ifrj-green:#006837; }
.gradio-container { max-width: 1060px !important; margin: 0 auto; }
header, footer { display:none !important; }
#app-title { color: var(--ifrj-green); font-weight: 800; }
.card { background:#fff; border:1px solid #e6f4ea; border-radius:14px; padding:16px; box-shadow:0 10px 30px rgba(0,0,0,0.08); }
button, .gr-button, .gr-button-primary, button.primary { background: var(--ifrj-green) !important; border-color: var(--ifrj-green) !important; color: #fff !important; }
button:hover, .gr-button:hover, .gr-button-primary:hover, button.primary:hover { filter: brightness(0.95); }
@media (max-width: 640px){ .gradio-container { padding: 8px !important; } }
"""
# Forçar sempre tema claro (sem alternância)
js_force_light = """
function () {
const b = document.body;
b.classList.remove('dark');
b.classList.add('light');
try { localStorage.setItem('theme', 'light'); } catch(e) {}
}
"""
# ==========================
# Modelo local (gratuito)
# ==========================
PRIMARY_REPO = os.getenv("MODEL_REPO", "bartowski/Qwen2.5-0.5B-Instruct-GGUF")
PRIMARY_FILE = os.getenv("MODEL_FILE", "Qwen2.5-0.5B-Instruct-Q4_K_M.gguf")
PRIMARY_TYPE = os.getenv("MODEL_TYPE", "qwen2")
FALLBACK_REPO = "TheBloke/TinyLlama-1.1B-Chat-v1.0-GGUF"
FALLBACK_FILE = "tinyllama-1.1b-chat-v1.0.Q3_K_M.gguf"
FALLBACK_TYPE = "llama"
MAX_NEW_TOKENS = int(os.getenv("MAX_NEW_TOKENS", "180"))
TEMPERATURE = float(os.getenv("TEMPERATURE", "0.5"))
TOP_P = float(os.getenv("TOP_P", "0.9"))
TOP_K = int(os.getenv("TOP_K", "40"))
REPEAT_PENALTY = float(os.getenv("REPEAT_PENALTY", "1.05"))
N_THREADS = int(os.getenv("N_THREADS", str(min(4, (os.cpu_count() or 2)))))
CTX_LEN = int(os.getenv("CTX_LEN", "2048"))
MAX_MESSAGES = int(os.getenv("MAX_MESSAGES", "8"))
MAX_PROMPT_CH = int(os.getenv("MAX_PROMPT_CH", "4800"))
STOP_SEQ = ["Usuário:", "Sistemas:", "Sistema:", "Assistant:", "Assistente:"]
_llm = None
def _load_llm():
"""Carrega modelo local tentando Qwen 0.5B e caindo para TinyLlama se necessário."""
global _llm
if _llm is not None:
return _llm
from ctransformers import AutoModelForCausalLM
candidates = [
(PRIMARY_REPO, PRIMARY_FILE, PRIMARY_TYPE),
("Qwen/Qwen2.5-0.5B-Instruct-GGUF", "qwen2.5-0.5b-instruct-q4_k_m.gguf", "qwen2"),
(FALLBACK_REPO, FALLBACK_FILE, FALLBACK_TYPE),
]
last_err = None
for repo, fname, mtype in candidates:
try:
_llm = AutoModelForCausalLM.from_pretrained(
repo, model_file=fname, model_type=mtype, gpu_layers=0, context_length=CTX_LEN
)
print(f"[FinanceBot] Modelo OK: {repo}/{fname} ({mtype})")
return _llm
except Exception as e:
last_err = e
print(f"[FinanceBot] Falhou {repo}/{fname} -> {e}")
raise RuntimeError(f"Não foi possível carregar modelo local. Último erro: {last_err}")
# ==========================
# Utilitários (parse e prompt)
# ==========================
def _truncate_messages(messages):
msgs = (messages or [])[-MAX_MESSAGES:]
while len(json.dumps(msgs, ensure_ascii=False)) > MAX_PROMPT_CH and len(msgs) > 4:
msgs = msgs[2:]
return msgs
def _to_prompt(messages):
parts = [f"Sistema: {SYSTEM_PROMPT.strip()}"]
for m in messages:
role = (m.get("role") or "").lower()
content = (m.get("content") or "").strip()
if not content: continue
if role == "assistant": parts.append(f"Assistente: {content}")
elif role == "system": parts.append(f"Sistema: {content}")
else: parts.append(f"Usuário: {content}")
parts.append("Assistente:")
prompt = "\n".join(parts)
if len(prompt) > MAX_PROMPT_CH:
prompt = prompt[-MAX_PROMPT_CH:]
if not prompt.endswith("Assistente:"):
prompt = prompt.rsplit("Assistente:", 1)[0] + "Assistente:"
return prompt
def _pct(text):
m = re.search(r"(\d+(?:[.,]\d+)?)\s*%+", text)
return float(m.group(1).replace(",", "."))/100 if m else None
def _money(text):
m = re.search(r"(?:R\$\s*)?(\d{1,3}(?:\.\d{3})*(?:,\d+)?|\d+(?:[.,]\d+)?)", text)
if not m: return None
s = m.group(1).replace(".", "").replace(",", ".")
try: return float(s)
except: return None
def _int(text):
m = re.search(r"(\d+)", text)
return int(m.group(1)) if m else None
def _n_periodos(texto):
m = re.search(r"por\s+(\d+)\s+(mes|mês|meses|ano|anos)", texto.lower())
if m:
n = int(m.group(1)); unidade = m.group(2)
if "ano" in unidade: return n*12
return n
return _int(texto)
def _br_number(x):
s = f"{x:,.2f}"
return s.replace(",", "X").replace(".", ",").replace("X", ".")
# ==========================
# SKILLS determinísticas (precisas, PT-BR)
# ==========================
def skill_juros_simples(msg):
if "juros simples" not in msg.lower(): return None
i = _pct(msg); P = _money(msg); n = _n_periodos(msg)
if not (i and n and P):
return ("Para **juros simples** use as fórmulas:\n"
"- $$J = P\\cdot i\\cdot n$$\n"
"- $$M = P + J$$\n"
"Informe **capital (R$)**, **taxa** (% por período) e **tempo** (meses/anos).")
J = P * i * n
M = P + J
return (f"**Juros simples**\n"
f"- P (capital): R$ {_br_number(P)}\n"
f"- i (taxa): {i*100:.4g}%/período\n"
f"- n (períodos): {n}\n"
f"- $$J = P\\cdot i\\cdot n = \\text{{R$}}\\ { _br_number(J) }$$\n"
f"- $$M = P + J = \\text{{R$}}\\ { _br_number(M) }$$")
def skill_juros_compostos(msg):
if "juros compostos" not in msg.lower(): return None
i = _pct(msg); P = _money(msg); n = _n_periodos(msg)
if not (i and n and P):
return ("Para **juros compostos** use as fórmulas:\n"
"- $$M = P\\cdot (1+i)^n$$\n"
"- $$J = M - P$$\n"
"Informe **capital (R$)**, **taxa** (% por período) e **tempo** (meses/anos).")
M = P * ((1 + i) ** n)
J = M - P
return (f"**Juros compostos**\n"
f"- P: R$ {_br_number(P)}\n"
f"- i: {i*100:.4g}%/período\n"
f"- n: {n}\n"
f"- $$M = P\\cdot (1+i)^n = \\text{{R$}}\\ { _br_number(M) }$$\n"
f"- $$J = M - P = \\text{{R$}}\\ { _br_number(J) }$$")
def skill_equivalencia_taxas(msg):
lo = msg.lower()
if "equivalên" not in lo and "equivale" not in lo: return None
i = _pct(msg)
if i is None:
return ("**Equivalência de taxas**: $$1+i_{dest}=(1+i_{orig})^k$$.\n"
"Ex.: '2% ao mês equivale a quanto ao ano?'")
if "mês" in lo or "mes" in lo:
i_aa = (1 + i)**12 - 1
return (f"**Equivalência** (mensal → anual)\n"
f"- i_mensal: {i*100:.4g}% a.m.\n"
f"- i_anual equivalente: {(i_aa*100):.4g}% a.a.")
if "ano" in lo or "anual" in lo:
i_am = (1 + i)**(1/12) - 1
return (f"**Equivalência** (anual → mensal)\n"
f"- i_anual: {i*100:.4g}% a.a.\n"
f"- i_mensal equivalente: {(i_am*100):.4g}% a.m.")
return ("Diga se a taxa é **mensal** ou **anual** para eu converter.")
def skill_price_sac(msg):
lo = msg.lower()
if all(k not in lo for k in ["price", "sac", "financiamento", "parcela"]): return None
P = _money(msg); i = _pct(msg); n = _n_periodos(msg)
if not (P and i is not None and n):
return ("Para comparar **Price x SAC** informe **capital (R$)**, **taxa** (% a.m., por ex.) e **prazo** (meses).\n"
"Ex.: 'financiamento 120.000, 1% a.m., 120 meses'.")
if i == 0:
parcela_price = P / n
else:
parcela_price = P * (i / (1 - (1 + i) ** (-n)))
amort = P / n
juros_1 = P * i
parcela_sac_1 = amort + juros_1
return (f"**Price x SAC** (estimativa)\n"
f"- Valor (P): R$ {_br_number(P)} | Taxa: {i*100:.3g}% a.m. | Prazo: {n} meses\n"
f"- Price — parcela fixa ≈ R$ {_br_number(parcela_price)}\n"
f"- SAC — 1ª parcela ≈ R$ {_br_number(parcela_sac_1)} (depois reduz)\n"
f"Dica: *Price = parcelas iguais*; *SAC = amortização constante*. "
f"Sempre verifique o **CET** e seguros embutidos.")
def skill_aposentadoria(msg):
lo = msg.lower()
if "aposent" not in lo and "renda passiva" not in lo and "viver de renda" not in lo:
return None
renda = _money(msg)
i = _pct(msg) or 0.0
n = _n_periodos(msg)
if any(k in lo for k in ["por mês","por mes","/mês","/mes","mensal"]):
if i <= 0 or not renda:
return ("Para **renda mensal alvo**, informe a renda desejada e uma **taxa mensal líquida** (ex.: 0,6% a.m.).")
capital = renda / i
return (f"**Aposentadoria — renda-alvo**\n"
f"- Renda desejada: R$ {_br_number(renda)}/mês\n"
f"- Taxa líquida estimada: {i*100:.3g}% a.m.\n"
f"- Patrimônio necessário ≈ R$ {_br_number(capital)}\n"
f"Obs.: considere impostos, inflação e risco; use margem de segurança.")
if n and any(k in lo for k in ["juntar", "acumular", "meta", "ter"]):
FV = _money(msg)
if FV and i > 0:
PMT = FV * i / ((1 + i)**n - 1)
return (f"**Aposentadoria — meta de patrimônio**\n"
f"- Meta (FV): R$ {_br_number(FV)}\n"
f"- Prazo: {n} meses | Taxa: {i*100:.3g}% a.m.\n"
f"- Aporte mensal (PMT) ≈ R$ {_br_number(PMT)}\n"
f"Dica: reavalie taxa/inflação anualmente e mantenha reserva de emergência.")
return ("Posso calcular:\n"
"- **Renda-alvo**: diga renda mensal e taxa líquida (ex.: 0,6% a.m.).\n"
"- **Meta de patrimônio**: informe valor final, prazo (meses/anos) e taxa mensal.")
def skill_pgbl_vgbl(msg):
lo = msg.lower()
if "pgbl" not in lo and "vgbl" not in lo:
return None
return (
"**PGBL x VGBL (visão geral)**\n"
"- **PGBL**: permite deduzir contribuições da base do IR (até 12% da renda tributável) — "
"tende a fazer sentido para quem declara no modelo completo e tem IR a pagar. Na retirada, "
"o IR incide sobre o **valor total**.\n"
"- **VGBL**: não deduz na declaração, mas na retirada o IR incide **apenas sobre os rendimentos**. "
"Em geral, indicado para quem declara no **modelo simplificado** ou já atingiu o limite do PGBL.\n"
"- **Tributação**: regime **progressivo** (tabela) ou **regressivo** (alíquotas decrescentes no tempo). "
"Isto é informativo — **não é recomendação**."
)
def skill_rf_rv(msg):
lo = msg.lower()
gatilhos = ["renda fixa", "renda variável", "renda variavel", "tesouro", "cdi", "cdb", "ações", "acoes", "fundos", "etf", "bolsa"]
if not any(g in lo for g in gatilhos): return None
return (
"**Renda Fixa x Renda Variável (resumo)**\n"
"- **Renda Fixa**: previsibilidade de regras (Tesouro, CDB, LCI/LCA). Pode ser prefixada, pós (CDI/IPCA) ou híbrida. "
"Atenção a prazos, liquidez, IOF (até 30 dias) e IR (salvo isenções como LCI/LCA).\n"
"- **Renda Variável**: oscila (ações, ETFs, FIIs). Maior potencial de retorno e risco; horizonte de longo prazo.\n"
"- **Princípios**: tenha **reserva de emergência**, entenda custos/impostos, defina objetivo e horizonte."
)
_FINANCE_KEYWORDS = (
"juros", "price", "sac", "financiamento", "parcela", "equivalên", "renda fixa",
"renda variável", "renda variavel", "tesouro", "cdb", "cdi", "poupança", "poupanca",
"investimento", "investimentos", "etf", "ações", "acoes", "bolsa", "pgbl", "vgbl",
"aposent", "reserva", "orcamento", "orçamento", "ipca", "selic", "custo", "despesa", "receita"
)
def is_offtopic(text):
lo = (text or "").lower()
return not any(k in lo for k in _FINANCE_KEYWORDS)
def try_skills(user_msg):
for fn in (
skill_juros_simples, skill_juros_compostos, skill_equivalencia_taxas,
skill_price_sac, skill_aposentadoria, skill_pgbl_vgbl, skill_rf_rv,
):
ans = fn(user_msg)
if ans: return ans
# armadilhas por último
return skill_armadilhas(user_msg)
def skill_armadilhas(msg):
lo = msg.lower()
if "armadil" not in lo and "pegad" not in lo and "golpe" not in lo: return None
return (
"**Armadilhas comuns**\n"
"- Confundir taxa **ao mês** com **ao ano**; converter errado (equivalência).\n"
"- Comparar só a **parcela** do financiamento e ignorar **CET**, seguros, taxas.\n"
"- Resgatar antes de **30 dias** (IOF) ou antes do vencimento (marcação a mercado) sem entender impactos.\n"
"- Misturar **reserva de emergência** com investimentos de risco.\n"
"- Promessas de 'ganhos garantidos' acima do mercado — **alerta**."
)
# ==========================
# Resposta com streaming
# ==========================
def _stream_local(prompt):
llm = _load_llm()
return llm(
prompt,
max_new_tokens=MAX_NEW_TOKENS,
temperature=TEMPERATURE,
top_p=TOP_P,
top_k=TOP_K,
repetition_penalty=REPEAT_PENALTY,
threads=N_THREADS,
stop=STOP_SEQ,
stream=True,
)
def respond_stream(user_input, history_messages):
user_input = (user_input or "").strip()
if not user_input:
yield gr.update(), history_messages
return
if is_offtopic(user_input):
new_hist = (history_messages or []) + [
{"role":"user","content":user_input},
{"role":"assistant","content":OFFTOPIC_MSG},
]
yield gr.update(value=""), new_hist
return
skill = try_skills(user_input)
if skill:
new_hist = (history_messages or []) + [
{"role":"user","content":user_input},
{"role":"assistant","content":skill},
]
yield gr.update(value=""), new_hist
return
msgs = _truncate_messages(history_messages or []) + [{"role":"user","content":user_input}]
current = msgs + [{"role":"assistant","content":""}]
yield gr.update(value=""), current
prompt = _to_prompt(msgs)
try:
acc = []
last_yield = time.time()
for chunk in _stream_local(prompt):
acc.append(chunk)
now = time.time()
if now - last_yield >= 0.04:
current[-1]["content"] = "".join(acc).strip()
yield gr.update(value=""), current
last_yield = now
if not "".join(acc).strip():
current[-1]["content"] = "Não consegui gerar uma resposta agora. Tente novamente."
else:
current[-1]["content"] = "".join(acc).strip()
yield gr.update(value=""), current
except Exception as e:
current[-1]["content"] = f"Falha na geração: {e}"
yield gr.update(value=""), current
def clear_chat():
return []
# ==========================
# Ações auxiliares (retry, histórico, comuns)
# ==========================
_last_user_msg = {"text": ""}
def save_user_message(message, history):
_last_user_msg["text"] = (message or "").strip()
return history, message
def retry_last(history):
if not history:
return history, ""
if history[-1].get("role") == "assistant":
history = history[:-1]
msg = _last_user_msg.get("text", "")
return history, msg
def download_history(history):
lines = []
for m in history or []:
role = m.get("role","")
content = (m.get("content") or "").replace("\r","")
if role == "user":
lines.append(f"User: {content}\n")
elif role == "assistant":
lines.append(f"Assistant: {content}\n")
path = "chat_history.txt"
with open(path, "w", encoding="utf-8") as f:
f.writelines(lines)
return path # DownloadButton usa o nome do arquivo retornado
def upload_history(file):
hist = []
if not file: return hist
with open(file, "r", encoding="utf-8") as f:
lines = f.readlines()
buf = []; who = None
for line in lines:
if line.startswith("User: "):
if buf and who: hist.append({"role": who, "content": "".join(buf).strip()})
who = "user"; buf = [line.replace("User: ","",1)]
elif line.startswith("Assistant: "):
if buf and who: hist.append({"role": who, "content": "".join(buf).strip()})
who = "assistant"; buf = [line.replace("Assistant: ","",1)]
else:
buf.append(line)
if buf and who: hist.append({"role": who, "content": "".join(buf).strip()})
return hist
def common_questions(_ignored, history):
exemplos = (
"Quais são as diferenças entre **juros simples** e **juros compostos**?\n"
"Como comparar **Price x SAC** em um financiamento?\n"
"O que é **equivalência de taxas** e como converter de a.m. para a.a.?\n"
"Como montar uma **reserva de emergência**?\n"
"Qual a diferença entre **PGBL** e **VGBL**?\n"
"O que considerar em **renda fixa x renda variável**?"
)
new_hist = (history or []) + [
{"role":"user","content":"Perguntas comuns sobre educação financeira"},
{"role":"assistant","content":exemplos},
]
return "", new_hist
# ==========================
# UI (tema IFRJ)
# ==========================
USER_AVATAR = "logo-ifrj.png" if os.path.exists("logo-ifrj.png") else None
BOT_AVATAR = "financebot.png" if os.path.exists("financebot.png") else None
with gr.Blocks(
title="FinanceBot IFRJ",
theme=theme,
css=custom_css,
head=html_header,
) as demo:
gr.Markdown("<h1 id='app-title'>FinanceBot IFRJ 💚</h1>")
gr.Markdown(
"Tutor do IFRJ para **educação financeira** e **matemática financeira** do dia a dia. "
"Faça sua pergunta (ex.: *juros compostos 1,5% por 12 meses em 1000*, *comparar Price x SAC: 120 mil, 1% a.m., 120 meses*).",
elem_classes=["card"]
)
chatbot = gr.Chatbot(
label="Conversa",
type="messages",
render_markdown=True,
latex_delimiters=latex_delimiter_set,
avatar_images=(USER_AVATAR, BOT_AVATAR),
height=520,
)
with gr.Row():
user_box = gr.Textbox(
label="Sua pergunta",
placeholder="Ex.: juros compostos 1,5% por 12 meses em 1000",
autofocus=True,
scale=7,
)
send_btn = gr.Button("Enviar", variant="primary", scale=1)
with gr.Row():
retry_btn = gr.Button("↻ Tentar de novo (última pergunta)")
clear_btn = gr.Button("Limpar conversa")
commons_btn = gr.Button("Perguntas comuns")
toggle_btn = gr.Button("Tema claro")
with gr.Accordion("Histórico", open=False):
with gr.Row():
dl_btn = gr.DownloadButton("Baixar histórico (.txt)")
up_btn = gr.UploadButton("Enviar histórico (.txt)", file_types=["text"])
# Fluxo
send_btn.click(save_user_message, [user_box, chatbot], [chatbot, user_box]) \
.then(respond_stream, [user_box, chatbot], [user_box, chatbot])
user_box.submit(save_user_message, [user_box, chatbot], [chatbot, user_box]) \
.then(respond_stream, [user_box, chatbot], [user_box, chatbot])
retry_btn.click(lambda h: retry_last(h), inputs=chatbot, outputs=[chatbot, user_box]) \
.then(respond_stream, [user_box, chatbot], [user_box, chatbot])
clear_btn.click(fn=clear_chat, inputs=None, outputs=chatbot)
commons_btn.click(common_questions, [user_box, chatbot], [user_box, chatbot])
dl_btn.click(download_history, inputs=chatbot, outputs=dl_btn)
up_btn.upload(lambda f, h: ("", (h or []) + upload_history(f)), [up_btn, chatbot], [user_box, chatbot])
toggle_btn.click(lambda: None, [], [], js=js_force_light)
# ==========================
# Lançamento (público por padrão) — retrocompatível
# ==========================
USER_NAME = os.environ.get("USER_NAME", "")
PASSWORD = os.environ.get("PASSWORD", "")
if __name__ == "__main__":
port = int(os.getenv("PORT", "7860"))
app = demo
try:
app = demo.queue() # se a versão suportar
except Exception:
app = demo # fallback sem fila
if USER_NAME and PASSWORD:
app.launch(server_name="0.0.0.0", server_port=port, auth=(USER_NAME, PASSWORD), show_api=False)
else:
app.launch(server_name="0.0.0.0", server_port=port, show_api=False)