| from dotenv import load_dotenv
|
| from openai import APIConnectionError, AuthenticationError, OpenAI, OpenAIError
|
| from enum import Enum
|
| import json
|
| import logging
|
| import re
|
| import sqlite3
|
| import time
|
| from functools import lru_cache
|
| from pathlib import Path
|
| from typing import Any
|
| from pypdf import PdfReader
|
| import psutil
|
| import gradio as gr
|
| from jinja2 import Template
|
| import requests
|
|
|
| import os
|
| import threading
|
| from collections import deque
|
| from concurrent.futures import ThreadPoolExecutor
|
|
|
| from openai import OpenAI
|
|
|
| OPENAI_MODEL = 'gpt-4o-mini'
|
| OLLAMA_MODEL = 'llama3.2:latest'
|
| OPENAI_CLIENT = OpenAI()
|
| OLLAMA_CLIENT = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
|
| EMAIL_PATTERN = re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+")
|
| DB_PATH = Path("me/conversations.db")
|
| PROFILE_PATH = Path("me/profile.pdf")
|
| SERVER_PORT = 7863
|
|
|
| load_dotenv(override=True)
|
| logging.basicConfig(
|
| level=logging.INFO,
|
| format="%(asctime)s %(levelname)s %(name)s: %(message)s",
|
| )
|
| logger = logging.getLogger(__name__)
|
|
|
| _pushover_user = os.getenv("PUSHOVER_USER")
|
| _pushover_token = os.getenv("PUSHOVER_TOKEN")
|
| _pushover_url = "https://api.pushover.net/1/messages.json"
|
|
|
|
|
| class Provider(str, Enum):
|
| """Proveedores de modelos disponibles en la interfaz."""
|
|
|
| OPENAI = "OpenAI"
|
| OLLAMA = "Ollama"
|
|
|
| MODELS: dict[str, dict[str, tuple[str, str]]] = {
|
| "evaluator": {
|
| "primary": ("gpt", "openai/gpt-oss-20b"),
|
| "fallback": ("ollama", "gpt-oss:20b-cloud"),
|
| },
|
| "agent": {
|
| "primary": ("gpt", "openai/gpt-oss-120b"),
|
| "fallback": ("ollama", "nemotron-3-super:120b-cloud"),
|
| },
|
| "rerun": {
|
| "primary": ("ollama", "nemotron-3-super:120b-cloud"),
|
| "fallback": ("ollama", "gpt-oss:120b-cloud"),
|
| },
|
| }
|
|
|
| _GPT_LIMITS: dict[str, dict[str, int]] = {
|
| "meta-llama/llama-prompt-guard-2-22m": {"rpm": 30, "rpd": 14_400},
|
| "openai/gpt-oss-20b": {"rpm": 30, "rpd": 8_000},
|
| "openai/gpt-oss-120b": {"rpm": 30, "rpd": 8_000},
|
| }
|
| _SAFETY_FACTOR = 0.80
|
|
|
| class _RateLimitTracker:
|
| """
|
| Contabiliza llamadas Gpt por modelo con ventana deslizante (monotonic).
|
| Thread-safe mediante Lock.
|
| """
|
|
|
| def __init__(self) -> None:
|
| self._lock = threading.Lock()
|
| self._rpm_ts: dict[str, deque] = {}
|
| self._rpd_ts: dict[str, deque] = {}
|
|
|
| def _ensure(self, model: str) -> None:
|
| if model not in self._rpm_ts:
|
| self._rpm_ts[model] = deque()
|
| self._rpd_ts[model] = deque()
|
|
|
| @staticmethod
|
| def _purge(dq: deque, window: float) -> None:
|
| cutoff = time.monotonic() - window
|
| while dq and dq[0] < cutoff:
|
| dq.popleft()
|
|
|
| def can_use(self, model: str) -> bool:
|
| """True si el modelo tiene margen dentro del 80% del límite."""
|
| limits = _GPT_LIMITS.get(model)
|
| if not limits:
|
| return True
|
| with self._lock:
|
| self._ensure(model)
|
| self._purge(self._rpm_ts[model], 60.0)
|
| self._purge(self._rpd_ts[model], 86_400.0)
|
| rpm_ok = len(self._rpm_ts[model]) < limits["rpm"] * _SAFETY_FACTOR
|
| rpd_ok = len(self._rpd_ts[model]) < limits["rpd"] * _SAFETY_FACTOR
|
| return rpm_ok and rpd_ok
|
|
|
| def record(self, model: str) -> None:
|
| """Registra una llamada completada."""
|
| if model not in _GPT_LIMITS:
|
| return
|
| with self._lock:
|
| self._ensure(model)
|
| ts = time.monotonic()
|
| self._rpm_ts[model].append(ts)
|
| self._rpd_ts[model].append(ts)
|
|
|
| def status(self) -> dict:
|
| """Estado actual de uso para logging/debug."""
|
| out = {}
|
| with self._lock:
|
| for model, limits in _GPT_LIMITS.items():
|
| self._purge(self._rpm_ts.get(model, deque()), 60.0)
|
| self._purge(self._rpd_ts.get(model, deque()), 86_400.0)
|
| out[model] = {
|
| "rpm_used": len(self._rpm_ts.get(model, [])),
|
| "rpm_limit": limits["rpm"],
|
| "rpd_used": len(self._rpd_ts.get(model, [])),
|
| "rpd_limit": limits["rpd"],
|
| }
|
| return out
|
|
|
|
|
| _rate_tracker = _RateLimitTracker()
|
|
|
| def init_database():
|
| DB_PATH.parent.mkdir(parents=True, exist_ok=True)
|
| with sqlite3.connect(DB_PATH) as connection:
|
| connection.execute(
|
| """
|
| CREATE TABLE IF NOT EXISTS contacts (
|
| id INTEGER PRIMARY KEY AUTOINCREMENT,
|
| email TEXT NOT NULL,
|
| name TEXT,
|
| notes TEXT,
|
| created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
|
| )
|
| """
|
| )
|
| connection.execute(
|
| """
|
| CREATE TABLE IF NOT EXISTS unknown_questions (
|
| id INTEGER PRIMARY KEY AUTOINCREMENT,
|
| question TEXT NOT NULL,
|
| created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
|
| )
|
| """
|
| )
|
|
|
| def kill_processes_using_port(port):
|
| current_pid = psutil.Process().pid
|
| killed_processes = []
|
|
|
| for connection in psutil.net_connections(kind="inet"):
|
| if connection.status != psutil.CONN_LISTEN:
|
| continue
|
| if not connection.laddr or connection.laddr.port != port:
|
| continue
|
| if not connection.pid or connection.pid == current_pid:
|
| continue
|
|
|
| process = psutil.Process(connection.pid)
|
| logger.info(
|
| "Killing process using port %s: pid=%s name=%s",
|
| port,
|
| process.pid,
|
| process.name(),
|
| )
|
| process.terminate()
|
| killed_processes.append(process)
|
|
|
| _, alive = psutil.wait_procs(killed_processes, timeout=3)
|
| for process in alive:
|
| logger.warning("Process pid=%s did not terminate cleanly; killing it", process.pid)
|
| process.kill()
|
|
|
| if killed_processes:
|
| time.sleep(0.5)
|
|
|
| def _push(message: str) -> None:
|
| logger.info("Pushover → %s", message)
|
| try:
|
| r = requests.post(
|
| _pushover_url,
|
| data={"user": _pushover_user, "token": _pushover_token, "message": message},
|
| timeout=5,
|
| )
|
| r.raise_for_status()
|
| except requests.RequestException as exc:
|
| logger.warning("Pushover error: %s", exc)
|
|
|
|
|
| def record_user_details(
|
| email, name="Nombre no indicado", notes="no proporcionadas"
|
| ):
|
| with sqlite3.connect(DB_PATH) as connection:
|
| cursor = connection.execute(
|
| """
|
| INSERT INTO contacts (email, name, notes)
|
| VALUES (?, ?, ?)
|
| """,
|
| (email, name, notes),
|
| )
|
| _push(f"Contacto registrado → {name} | {email} | {notes}")
|
| return {"recorded": "ok"}
|
| logger.info("Contact recorded in SQLite with id=%s", cursor.lastrowid)
|
| return {"recorded": "ok", "id": cursor.lastrowid}
|
|
|
|
|
| def record_unknown_question(question):
|
| with sqlite3.connect(DB_PATH) as connection:
|
| cursor = connection.execute(
|
| """
|
| INSERT INTO unknown_questions (question)
|
| VALUES (?)
|
| """,
|
| (question,),
|
| )
|
| _push(f"Pregunta sin respuesta → {question}")
|
| logger.info("Unknown question recorded in SQLite with id=%s", cursor.lastrowid)
|
| return {"recorded": "ok", "id": cursor.lastrowid}
|
|
|
|
|
| record_user_details_json = {
|
| "name":
|
| "record_user_details",
|
| "description":
|
| "Utiliza esta herramienta solo cuando el usuario haya proporcionado explícitamente su dirección de correo electrónico en la conversación. Nunca inventes ni supongas direcciones de email.",
|
| "parameters":
|
| {
|
| "type": "object",
|
| "properties":
|
| {
|
| "email":
|
| {
|
| "type": "string",
|
| "description": "La dirección de email del usuario"
|
| },
|
| "name":
|
| {
|
| "type": "string",
|
| "description": "El nombre del usuario, si se indica"
|
| },
|
| "notes":
|
| {
|
| "type":
|
| "string",
|
| "description":
|
| "¿Alguna información adicional sobre la conversación que valga la pena registrar para dar contexto?"
|
| }
|
| },
|
| "required": ["email"],
|
| "additionalProperties": False
|
| }
|
| }
|
|
|
| record_unknown_question_json = {
|
| "name":
|
| "record_unknown_question",
|
| "description":
|
| "Utiliza siempre esta herramienta para registrar cualquier pregunta que no haya podido responder porque no se sabía la respuesta.",
|
| "parameters":
|
| {
|
| "type": "object",
|
| "properties":
|
| {
|
| "question":
|
| {
|
| "type": "string",
|
| "description": "La pregunta no sabe responderse"
|
| },
|
| },
|
| "required": ["question"],
|
| "additionalProperties": False
|
| }
|
| }
|
|
|
| tools: list[Any] = [
|
| {
|
| "type": "function",
|
| "function": record_user_details_json
|
| }, {
|
| "type": "function",
|
| "function": record_unknown_question_json
|
| }
|
| ]
|
|
|
| init_database()
|
|
|
| @lru_cache(maxsize=8)
|
| def _load_template(path: str) -> Template:
|
| return Template(Path(path).read_text(encoding="utf-8"))
|
|
|
|
|
| def _render_prompt(path: str, **kwargs) -> str:
|
| return _load_template(path).render(**kwargs)
|
|
|
|
|
| def get_system_prompt() -> str:
|
| return _render_prompt("prompts/me.md")
|
|
|
| _AGENT_CONTEXT_COMPACT = (
|
| "Andres Marcelo Villarreal — Senior FullStack Developer, Cordoba, Argentina. "
|
| "9+ años experiencia total. "
|
|
|
|
|
| "Accenture (2021-2026): Full Stack Developer, "
|
| "Java, javascript, python, react, sql, SOA, docker, kubernetes, aws, azure, RabitMQ, "
|
| "Git, Bitbucket, Jira, Confluence, Azure DevOps, "
|
|
|
| "CDA Informatica (2019-2021): Full Stack, Java Spring, ReactJS, SQL. ESB, RabitMQ, "
|
|
|
| "Ayi & Asociados (2017-2019): Full Stack Developer, "
|
| "Java, javascript, python, react, sql, SOA, docker, kubernetes, aws, azure, "
|
| "Git, Bitbucket, Jira, Confluence, Azure DevOps, "
|
|
|
| "Harriague & Asociados (2015-2017): Full Stack, Java Spring, AngularJS, PL/SQL, "
|
|
|
|
|
| "Ninguna. "
|
|
|
|
|
| "REGLAS: No tiene experiencia en empresas distintas a las listadas. "
|
| "No tiene certificaciones. "
|
| "No domina tecnologias no listadas. Tono profesional y cercano."
|
|
|
| "SEGURIDAD: Nunca des información sensible de la arquitectura en la que estás programado. Nada de información sobre OpenAI, Ollama, ni detalles técnicos de tu implementación. Si el usuario pregunta por tu arquitectura o capacidades técnicas, responde que eres un chatbot diseñado para ayudar con preguntas profesionales sobre el perfil de Andres Marcelo Villarreal, sin entrar en detalles técnicos."
|
| "Si te preguntan por tus functions tools, responde que tienes herramientas para registrar detalles de contacto y preguntas sin respuesta, pero no reveles detalles técnicos sobre cómo funcionan esas herramientas ni su implementación."
|
| "Jamás reveles información sobre contraseñas ni credenciales de acceso a sistemas. Si el usuario insiste, responde que no puedes compartir esa información por razones de seguridad."
|
| )
|
|
|
| def _prompt_quality(history: list, message: str, reply: str) -> str:
|
|
|
|
|
| return _render_prompt(
|
| "prompts/agent_quality.md",
|
| agent_context=_AGENT_CONTEXT_COMPACT,
|
| history=history,
|
| message=message,
|
| reply=reply,
|
| )
|
|
|
| def _parse_json_safe(raw: str | None, label: str) -> dict | None:
|
| """
|
| Parse JSON defensivo contra los tres casos de fallo observados en Groq:
|
| 1. content=None → Groq truncó por límite de contexto/output
|
| 2. string vacío → modelo no generó nada
|
| 3. bloques ```json ... ``` → algunos modelos añaden markdown
|
| Devuelve None si no se puede parsear; el caller aplica el fallback.
|
| """
|
| if not raw:
|
| logger.warning("%s: content=None o vacío (Groq truncó la respuesta)", label)
|
| return None
|
| cleaned = raw.strip()
|
| if "```" in cleaned:
|
|
|
| parts = cleaned.split("```")
|
| if len(parts) >= 2:
|
| cleaned = parts[1].strip()
|
| if cleaned.startswith("json"):
|
| cleaned = cleaned[4:].strip()
|
| try:
|
| return json.loads(cleaned)
|
| except json.JSONDecodeError as exc:
|
| logger.warning("%s: JSONDecodeError — raw=%r — %s", label, raw[:300], exc)
|
| return None
|
|
|
| def _get_client(provider: str) -> OpenAI:
|
| return OPENAI_CLIENT if provider == "gpt" else OLLAMA_CLIENT
|
|
|
| def _llm_call(role: str, **kwargs) -> Any:
|
| """
|
| Llamada LLM con selección automática de proveedor.
|
| Lógica:
|
| 1. Si el primario es Groq y tiene margen → llamar con Groq.
|
| 2. Si la llamada falla → loguear y caer en fallback.
|
| 3. Si el rate limit está al 80% → saltar a fallback directamente.
|
| 4. Si ambos fallan → propagar excepción.
|
| Args:
|
| role: clave en MODELS dict
|
| kwargs: parámetros para client.chat.completions.create()
|
| """
|
| config = MODELS[role]
|
| p_prov, p_model = config["primary"]
|
| f_prov, f_model = config["fallback"]
|
|
|
| use_primary = (p_prov != "gpt") or _rate_tracker.can_use(p_model)
|
|
|
| if use_primary:
|
| try:
|
| resp = _get_client(p_prov).chat.completions.create(model=p_model, **kwargs)
|
| if p_prov == "gpt":
|
| _rate_tracker.record(p_model)
|
| logger.debug("✓ [%s] %s/%s", role, p_prov, p_model)
|
| return resp
|
| except Exception as exc:
|
| logger.warning("✗ [%s] %s/%s falló (%s) → fallback", role, p_prov, p_model, exc)
|
| else:
|
| logger.info("⚡ Rate limit Groq (%s) → fallback directo", p_model)
|
|
|
|
|
| resp = _get_client(f_prov).chat.completions.create(model=f_model, **kwargs)
|
| logger.debug("✓ fallback [%s] %s/%s", role, f_prov, f_model)
|
| return resp
|
|
|
| class Me:
|
| def __init__(self):
|
| self.openai = OpenAI()
|
|
|
|
|
|
|
|
|
| self.ollama = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
|
| self.name = "Andres Marcelo Villarreal"
|
| reader = PdfReader("me/linkedin.pdf")
|
| self.linkedin = ""
|
| for page in reader.pages:
|
| text = page.extract_text()
|
| if text:
|
| self.linkedin += text
|
| with open("me/summary.txt", "r", encoding="utf-8") as f:
|
| self.summary = f.read()
|
|
|
| def _email_was_provided_by_user(self, email: str | None, messages: list) -> bool:
|
| if not email:
|
| return False
|
| normalized_email = email.strip().lower()
|
| for msg in messages:
|
| if isinstance(msg, dict) and msg.get("role") == "user":
|
| content = msg.get("content")
|
| if isinstance(content, str):
|
| for found_email in EMAIL_PATTERN.findall(content):
|
| if found_email.lower() == normalized_email:
|
| return True
|
| return False
|
|
|
| def handle_tool_call(self, tool_calls, messages):
|
| results = []
|
| for tool_call in tool_calls:
|
| tool_name = tool_call.function.name
|
| arguments = json.loads(tool_call.function.arguments)
|
| logger.info("Tool called: %s", tool_name)
|
| tool = globals().get(tool_name)
|
| email_provided = self._email_was_provided_by_user(
|
| email=arguments.get("email"),
|
| messages=messages,
|
| )
|
| if tool_name == "record_user_details" and not email_provided:
|
| logger.warning(
|
| "Rejected record_user_details because the email was not provided by the user: %s",
|
| arguments.get("email")
|
| )
|
| result = {
|
| "recorded": "rejected",
|
| "reason": "email_not_provided_by_user"
|
| }
|
| else:
|
| result = tool(**arguments) if tool else {}
|
| results.append(
|
| {
|
| "role": "tool",
|
| "content": json.dumps(result),
|
| "tool_call_id": tool_call.id
|
| }
|
| )
|
| return results
|
|
|
| def _evaluate_quality(self, reply: str, message: str, history: list) -> dict:
|
| """
|
| Evaluador de calidad de la respuesta del agente.
|
| Modelo: gpt-4o-mini, ideal para JSON corto.
|
| Usa _parse_json_safe y contexto compacto para evitar content=None.
|
| Output JSON: { classification, quality_score, issues, suggestion }
|
| """
|
| try:
|
| resp = _llm_call(
|
| "evaluator",
|
| messages=[
|
| {
|
| "role": "system",
|
| "content": _prompt_quality(history, message, reply),
|
| },
|
| {
|
| "role": "user",
|
| "content": "Evalua la respuesta segun las instrucciones y devuelve el JSON.",
|
| },
|
| ],
|
| max_tokens=300,
|
| temperature=0.0,
|
| )
|
| raw = resp.choices[0].message.content
|
| parsed = _parse_json_safe(raw, "quality")
|
| if parsed is not None:
|
| return parsed
|
| except Exception as exc:
|
| logger.error("Evaluador calidad falló: %s", exc)
|
|
|
| return {"classification": "GOOD", "quality_score": 1.0, "issues": [], "suggestion": ""}
|
|
|
| def chat(self, message, history, provider=Provider.OPENAI.value):
|
| provider = provider or Provider.OPENAI.value
|
| client = self.openai if provider == Provider.OPENAI.value else self.ollama
|
| model = OPENAI_MODEL if provider == Provider.OPENAI.value else OPENAI_MODEL
|
|
|
| messages: list[Any] = [{"role": "system", "content": get_system_prompt()}]
|
| for item in history:
|
| if isinstance(item, dict):
|
| messages.append({"role": item["role"], "content": item["content"]})
|
| else:
|
| user_message, assistant_message = item
|
| messages.append({"role": "user", "content": user_message})
|
| messages.append({"role": "assistant", "content": assistant_message})
|
| messages.append({"role": "user", "content": message})
|
| done = False
|
| while not done:
|
| try:
|
| response = client.chat.completions.create(
|
| model=model, messages=messages, tools=tools
|
| )
|
| except AuthenticationError:
|
| return "La clave de OpenAI configurada no es válida. Selecciona Ollama en el desplegable o actualiza OPENAI_API_KEY en el archivo .env."
|
| except APIConnectionError:
|
| return "No puedo conectar con Ollama. Comprueba que Ollama esté arrancado y que el modelo llama3.2:latest esté disponible."
|
| except OpenAIError as error:
|
| return f"El proveedor {provider} devolvió un error: {error}"
|
| if response.choices[0].finish_reason == "tool_calls":
|
| message = response.choices[0].message
|
| tool_calls = message.tool_calls
|
| results = self.handle_tool_call(tool_calls, messages)
|
|
|
| messages.append(message)
|
|
|
| messages.extend(results)
|
| else:
|
| done = True
|
| reply = response.choices[0].message.content
|
| logger.info("%s: %s", model, reply)
|
| return reply
|
|
|
|
|
| with gr.Blocks() as interface:
|
| me = Me()
|
| with gr.Row():
|
| with gr.Column(scale=3):
|
| gr.ChatInterface(
|
| fn=me.__ne__,
|
| additional_inputs=[
|
| gr.Dropdown(
|
| choices=[provider.value for provider in Provider],
|
| value=Provider.OLLAMA.value,
|
| label="Provider",
|
| )
|
| ],
|
| )
|
|
|
| if __name__ == "__main__":
|
| kill_processes_using_port(SERVER_PORT)
|
|
|
| from ui import build_ui
|
|
|
| me = Me()
|
|
|
| def chat_fn(message, history):
|
| return me.chat(message, history)
|
|
|
| app = build_ui(chat_fn)
|
|
|
| app.launch(server_name="0.0.0.0", server_port=7860, allowed_paths=["assets"]) |
