# Gerado por IA
import sys
import os
import json
import urllib.request
import asyncio
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from config import settings
from database import init_db
from ai_agent import AIAgent
from orchestrator import Orchestrator
from telegram_bot import agent_registry
from orchestrator_shared import _orchestrator_ref
def _orchestrator_ref_set(orch):
global _orchestrator_ref
_orchestrator_ref = orch
# ---------------------------------------------------------------------------
# Helpers HTTP (urllib — sem dependências extra)
# ---------------------------------------------------------------------------
def _http_post(url: str, payload: dict, headers: dict | None = None, timeout: int = 60) -> dict:
data = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(url, data=data, headers=headers or {"Content-Type": "application/json"})
with urllib.request.urlopen(req, timeout=timeout) as resp:
return json.loads(resp.read())
def _http_get(url: str, headers: dict | None = None, timeout: int = 30) -> dict:
req = urllib.request.Request(url, headers=headers or {})
with urllib.request.urlopen(req, timeout=timeout) as resp:
return json.loads(resp.read())
# ---------------------------------------------------------------------------
# Teste de modelo
# ---------------------------------------------------------------------------
def _extract_reply(data: dict) -> str | None:
"""Extrai o texto de resposta de um chat completion, verificando content e reasoning."""
choices = data.get("choices", [])
if not choices:
return None
msg = choices[0].get("message") or {}
# content direto
content = msg.get("content")
if content and str(content).strip() and str(content).strip().lower() != "none":
return str(content).strip()
# reasoning (modelos que usam este campo para resposta)
reasoning = msg.get("reasoning")
if reasoning and str(reasoning).strip():
return str(reasoning).strip()
return None
def test_model(model: str, retries: int = 2) -> tuple[bool, str]:
"""Retorna (ok, resposta_ou_erro). Tenta ate 'retries' vezes em caso de 429."""
import time
for attempt in range(retries + 1):
try:
data = _http_post(
f"{settings.OPENROUTER_BASE_URL}/chat/completions",
{
"model": model,
"messages": [{"role": "user", "content": "Responda apenas: OK"}],
"max_tokens": 100,
},
headers={
"Authorization": f"Bearer {settings.OPENROUTER_API_KEY}",
"Content-Type": "application/json",
"HTTP-Referer": "https://multiagentsai.local",
"X-Title": "MultiAgentsAI",
},
timeout=65,
)
reply = _extract_reply(data)
if reply:
return True, reply[:80]
fr = (data.get("choices", [{}])[0] or {}).get("finish_reason", "?")
return False, f"Resposta vazia (finish_reason={fr})"
except urllib.error.HTTPError as exc:
if exc.code == 429 and attempt < retries:
wait = 5 * (attempt + 1)
print(f" [Rate-limit] Aguardando {wait}s antes de retry...")
time.sleep(wait)
continue
body = exc.read().decode("utf-8", errors="replace")[:200]
return False, f"HTTP {exc.code}: {body}"
except Exception as exc:
if attempt < retries:
time.sleep(3)
continue
return False, str(exc)
return False, "Maximo de tentativas excedido"
# ---------------------------------------------------------------------------
# Busca de modelo substituto
# ---------------------------------------------------------------------------
def find_replacement_model(provider_hint: str = "") -> str | None:
"""Busca um modelo de chat gratuito alternativo no OpenRouter."""
try:
data = _http_get(
f"{settings.OPENROUTER_BASE_URL}/models",
headers={
"Authorization": f"Bearer {settings.OPENROUTER_API_KEY}",
},
timeout=30,
)
free_models = [m for m in data.get("data", []) if ":free" in m.get("id", "")]
except Exception as exc:
print(f" [Aviso] Nao foi possivel buscar modelos: {exc}")
return None
# Ordena por tamanho de contexto (maior primeiro)
free_models.sort(key=lambda m: m.get("context_length", 0), reverse=True)
already_used = {a["model"] for a in settings.AGENTS}
for m in free_models:
mid = m["id"]
if mid in already_used:
continue
# Pula modelos de embedding
if "embedding" in mid.lower():
continue
print(f" [Teste] Modelo alternativo: {mid}")
ok, info = test_model(mid)
if ok:
print(f" [OK] {mid} -> {info[:60]}")
return mid
else:
print(f" [FALHOU] {mid} -> {info[:80]}")
return None
# ---------------------------------------------------------------------------
# Validacao e substituicao de agentes
# ---------------------------------------------------------------------------
def validate_and_fix_agents() -> list[dict]:
"""Testa todos os agentes; substitui modelos falhos automaticamente."""
print("=" * 55)
print(" Validando modelos de IA...")
print("=" * 55)
agents = list(settings.AGENTS)
used_models: set[str] = set()
for i, agent in enumerate(agents):
model = agent["model"]
name = agent["name"]
if model in used_models:
# Evita duplicatas
print(f"\n[{name}] Modelo {model} ja esta em uso por outro agente. Procurando alternativa...")
replacement = find_replacement_model()
if replacement:
model = replacement
agents[i]["model"] = replacement
used_models.add(replacement)
print(f"[{name}] Substituido por: {replacement}")
else:
print(f"[{name}] ERRO: Nenhum modelo alternativo encontrado!")
continue
print(f"\n[{name}] Testando modelo: {model}")
ok, info = test_model(model)
if ok:
print(f" [OK] {name} -> {info[:80]}")
used_models.add(model)
else:
print(f" [FALHOU] {name} -> {info[:100]}")
print(f" [Busca] Procurando modelo alternativo...")
replacement = find_replacement_model()
if replacement:
agents[i]["model"] = replacement
used_models.add(replacement)
print(f" [Substituido] {name}: {model} -> {replacement}")
else:
print(f" [ERRO] Nenhum modelo alternativo encontrado para {name}!")
print("\n" + "=" * 55)
print(" Validacao concluida.")
print("=" * 55 + "\n")
return agents
# ---------------------------------------------------------------------------
# Envio /start
# ---------------------------------------------------------------------------
def send_start_command(validated_agents: list[dict]) -> None:
"""Envia o comando /start para o Telegram apos a inicializacao."""
url = f"https://api.telegram.org/bot{settings.TELEGRAM_BOT_TOKEN}/sendMessage"
agents_info = "\n".join(
f" - {a['name']} ({a['model']})"
for a in validated_agents
)
msg = (
"MultiAgentsAI iniciado!\n"
f"Agentes ativos: {len(validated_agents)}\n"
f"{agents_info}\n\n"
"Os agentes estao conversando automaticamente.\n"
"Comandos:\n"
"/status - Status do sistema\n"
"/historico - Ultimas mensagens de hoje\n"
"/agentes - Lista de agentes\n"
"/participar [mensagem] - Enviar mensagem como usuario\n"
"/pausar /continuar - Pausar/retomar agentes"
)
payload = json.dumps({
"chat_id": settings.TELEGRAM_CHAT_ID,
"text": msg,
"parse_mode": "HTML",
}).encode("utf-8")
req = urllib.request.Request(url, data=payload, headers={"Content-Type": "application/json"})
try:
with urllib.request.urlopen(req, timeout=15) as resp:
resp.read()
except Exception as exc:
print(f"[Start] Erro ao enviar /start: {exc}")
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def main():
init_db()
# Valida e corrige modelos antes de iniciar
validated_agents = validate_and_fix_agents()
if not validated_agents:
print("ERRO: Nenhum agente com modelo valido!")
sys.exit(1)
agents = [AIAgent(a["name"], a["model"], a["index"]) for a in validated_agents]
for agent in agents:
agent_registry[agent.name] = agent
orchestrator = Orchestrator(agents)
# Envia /start automatico para o Telegram
print("[Start] Enviando comando /start para o Telegram...")
send_start_command(validated_agents)
print("[Start] Comando /start enviado.\n")
try:
asyncio.run(orchestrator.run())
except KeyboardInterrupt:
print("\n\n[MultiAgentsAI] Encerrado pelo usuario (Ctrl+C).")
if __name__ == "__main__":
main()