Files
nickpons666 8398e988b0 Feat: Agregar agente Groq con integración RAG
- Nuevo módulo groq_agent.py para consultas a la API de Groq
- Panel de administración en /groq para configurar API key, modelo y prompt
- Comando /rag en Discord y Telegram para consultar el RAG
- Sistema de prompt personalizable guardado en base de datos
- Soporte para variables de entorno en Docker
- Fix: starlette version para evitar bug con Jinja2
2026-03-26 21:23:19 -06:00

145 lines
4.5 KiB
Python

import yaml
import os
from dotenv import load_dotenv
load_dotenv()
_config = None
def load_config(config_path: str = None) -> dict:
global _config
if _config is None:
config = {}
defaults = {
"discord": {
"token": ""
},
"telegram": {
"token": ""
},
"libretranslate": {
"url": ""
},
"web": {
"host": "0.0.0.0",
"port": 8000,
"admin_username": "",
"admin_password": ""
},
"database": {
"type": "sqlite",
"path": "bots_config.db",
"host": "",
"port": 3306,
"user": "",
"password": "",
"name": ""
},
"languages": {
"enabled": [
{"code": "en", "name": "English"},
{"code": "es", "name": "Español"},
{"code": "fr", "name": "Français"},
{"code": "de", "name": "Deutsch"},
{"code": "it", "name": "Italiano"},
{"code": "pt", "name": "Português"}
]
},
"groq": {
"api_key": "",
"model": "llama-3.3-70b-versatile",
"rag_url": "http://localhost:8004"
}
}
config = defaults.copy()
if config_path is None:
config_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), "config.yaml")
if os.path.exists(config_path):
with open(config_path, "r") as f:
yaml_config = yaml.safe_load(f)
if yaml_config:
_merge_config(config, yaml_config)
env_mappings = {
"DISCORD_TOKEN": ("discord", "token"),
"TELEGRAM_TOKEN": ("telegram", "token"),
"LIBRETRANSLATE_URL": ("libretranslate", "url"),
"WEB_HOST": ("web", "host"),
"WEB_PORT": ("web", "port", int),
"ADMIN_USERNAME": ("web", "admin_username"),
"ADMIN_PASSWORD": ("web", "admin_password"),
"DATABASE_PATH": ("database", "path"),
"DB_TYPE": ("database", "type"),
"DB_HOST": ("database", "host"),
"DB_PORT": ("database", "port", int),
"DB_USER": ("database", "user"),
"DB_PASSWORD": ("database", "password"),
"DB_NAME": ("database", "name"),
"GROQ_API_KEY": ("groq", "api_key"),
"GROQ_MODEL": ("groq", "model"),
"RAG_API_URL": ("groq", "rag_url"),
}
for env_key, (section, key, *transform) in env_mappings.items():
env_val = os.getenv(env_key)
if env_val:
if transform:
env_val = transform[0](env_val)
config[section][key] = env_val
_config = config
return _config
def _merge_config(config: dict, yaml_config: dict):
for key, value in yaml_config.items():
if key in config and isinstance(config[key], dict) and isinstance(value, dict):
config[key].update({k: v for k, v in value.items() if v})
elif value:
config[key] = value
def get_config() -> dict:
if _config is None:
load_config()
return _config
def get_discord_token() -> str:
return get_config().get("discord", {}).get("token", "")
def get_telegram_token() -> str:
return get_config().get("telegram", {}).get("token", "")
def get_libretranslate_url() -> str:
return get_config().get("libretranslate", {}).get("url", "")
def get_languages() -> list:
return get_config().get("languages", {}).get("enabled", [])
def get_db_path() -> str:
return get_config().get("database", {}).get("path", "bots_config.db")
def get_db_config() -> dict:
return get_config().get("database", {})
def get_db_type() -> str:
return get_config().get("database", {}).get("type", "sqlite")
def get_web_config() -> dict:
return get_config().get("web", {})
def get_groq_config() -> dict:
return get_config().get("groq", {})
def get_groq_api_key() -> str:
return get_config().get("groq", {}).get("api_key", "")
def get_groq_model() -> str:
return get_config().get("groq", {}).get("model", "llama-3.3-70b-versatile")
def get_groq_rag_url() -> str:
return get_config().get("groq", {}).get("rag_url", "http://localhost:8004")