File size: 7,068 Bytes
7e493b0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 | """Aprendizado v1 — log local JSONL + recuperação de casos parecidos (few-shot).
Sem dependências novas, só stdlib. CPU-friendly, sem GPU.
- Cada troca bot<->cliente é anexada a `conversas.jsonl` (ver `DATA_DIR`).
- `POST /feedback` classifica uma resposta como "boa"/"ma".
- Respostas "ma" nunca voltam como exemplo; "boa" tem prioridade.
- Exemplos parecidos entram no prompt como few-shot (factos e tom).
- O ficheiro nunca deve ir para git (contém números de clientes): ver `.gitignore`.
"""
import json
import os
import re
import threading
import unicodedata
import uuid
from datetime import datetime, timezone
_STOPWORDS = {
"a", "o", "e", "de", "do", "da", "dos", "das", "em", "um", "uma",
"para", "com", "que", "quanto", "quantos", "qual", "quais", "como",
"quando", "onde", "os", "as", "uns", "umas", "no", "na", "nos",
"nas", "ao", "aos", "por", "mais", "muito", "isto", "isso", "este",
"esta", "esse", "essa", "meu", "minha", "seu", "sua", "voce", "você",
"tem", "têm", "e", "se", "qto", "qt", "q", "la", "lá", "ai", "aí",
"entao", "então", "pois", "mas", "ou", "tambem", "também",
}
def _tokens(texto: str) -> set:
sem_acento = unicodedata.normalize("NFKD", texto or "").encode("ascii", "ignore").decode()
palavras = re.findall(r"[a-z0-9]+", sem_acento.lower())
return {p for p in palavras if len(p) > 2 and p not in _STOPWORDS}
class LearningStore:
"""Guarda trocas em JSONL e recupera casos parecidos."""
def __init__(
self,
data_dir: str | None = None,
arquivo: str = "conversas.jsonl",
max_linhas: int = 2000,
top_k: int = 2,
min_score: float = 0.15,
):
self.data_dir = data_dir or os.getenv("DATA_DIR", "./data")
self.arquivo = arquivo
self.max_linhas = int(os.getenv("LEARNING_MAX_LINHAS", str(max_linhas)))
self.top_k = int(os.getenv("LEARNING_TOP_K", str(top_k)))
self.min_score = float(os.getenv("LEARNING_MIN_SCORE", str(min_score)))
self._lock = threading.Lock()
try:
os.makedirs(self.data_dir, exist_ok=True)
except Exception as exc:
print(f"[learning] não consegui criar {self.data_dir}: {exc}")
def _caminho(self) -> str:
return os.path.join(self.data_dir, self.arquivo)
def _ler_tudo(self) -> list:
registos = []
try:
with open(self._caminho(), "r", encoding="utf-8") as fh:
for linha in fh:
linha = linha.strip()
if not linha:
continue
try:
registos.append(json.loads(linha))
except json.JSONDecodeError:
continue
except FileNotFoundError:
pass
except Exception as exc:
print(f"[learning] erro a ler: {exc}")
return registos
def log_exchange(self, cliente: str, pergunta: str, resposta: str,
provedor: str = "", modelo: str = "",
mensagem_id: str | None = None) -> str:
mensagem_id = mensagem_id or uuid.uuid4().hex[:12]
registo = {
"id": mensagem_id,
"ts": datetime.now(timezone.utc).isoformat(),
"cliente": cliente or "unknown",
"pergunta": (pergunta or "")[:1000],
"resposta": (resposta or "")[:1000],
"provedor": provedor,
"modelo": modelo,
"avaliacao": None,
}
try:
with self._lock:
with open(self._caminho(), "a", encoding="utf-8") as fh:
fh.write(json.dumps(registo, ensure_ascii=False) + "\n")
self._podar()
except Exception as exc:
print(f"[learning] erro a registar: {exc}")
return mensagem_id
def _podar(self) -> None:
try:
with open(self._caminho(), "r", encoding="utf-8") as fh:
linhas = fh.readlines()
if len(linhas) > self.max_linhas + 500:
with open(self._caminho(), "w", encoding="utf-8") as fh:
fh.writelines(linhas[-self.max_linhas:])
except FileNotFoundError:
pass
except Exception as exc:
print(f"[learning] erro a podar: {exc}")
def rate(self, mensagem_id: str, nota: str, observacao: str = "") -> bool:
if nota not in ("boa", "ma"):
return False
try:
with self._lock:
registos = self._ler_tudo()
achou = False
for reg in registos:
if reg.get("id") == mensagem_id:
reg["avaliacao"] = nota
if observacao:
reg["observacao"] = observacao[:500]
achou = True
break
if not achou:
return False
with open(self._caminho(), "w", encoding="utf-8") as fh:
for reg in registos:
fh.write(json.dumps(reg, ensure_ascii=False) + "\n")
return True
except Exception as exc:
print(f"[learning] erro a avaliar: {exc}")
return False
def find_similar(self, pergunta: str, k: int | None = None) -> list:
base = _tokens(pergunta)
if not base:
return []
candidatos = []
try:
for reg in self._ler_tudo():
if reg.get("avaliacao") == "ma":
continue
pontos = _tokens(reg.get("pergunta", ""))
if not pontos:
continue
uniao = base | pontos
score = len(base & pontos) / len(uniao) if uniao else 0.0
if reg.get("avaliacao") == "boa":
score = min(1.0, score + 0.25)
if score >= self.min_score:
candidatos.append((score, reg))
except Exception as exc:
print(f"[learning] erro a procurar: {exc}")
return []
candidatos.sort(key=lambda item: item[0], reverse=True)
return [reg for _, reg in candidatos[: (k or self.top_k)]]
def format_examples(self, pergunta: str) -> str:
exemplos = self.find_similar(pergunta)
if not exemplos:
return ""
blocos = ["EXEMPLOS DE ATENDIMENTOS ANTERIORES (segue o mesmo padrão de factos e tom):"]
total = len(blocos[0])
for i, reg in enumerate(exemplos, 1):
perg = (reg.get("pergunta", "") or "")[:200]
resp = (reg.get("resposta", "") or "")[:300]
bloco = f"{i}) Cliente: {perg}\n Plenitude: {resp}"
if total + len(bloco) > 900:
break
blocos.append(bloco)
total += len(bloco)
if len(blocos) < 2:
return ""
return "\n".join(blocos)
def count(self) -> int:
return len(self._ler_tudo())
|