- 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
145 lines
4.5 KiB
Python
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")
|