Multi-Agent / agents.py
Danielfonseca1212's picture
Create agents.py
871145a verified
# agents.py — Três agentes com personas distintas
"""
ATLAS — Arguer PRO (defende a posição)
NEXUS — Arguer CON (ataca a posição)
ORACLE — Judge (avalia e decide)
"""
from openai import OpenAI
from typing import Generator, Dict, Any
import time
MODEL = "gpt-4o-mini"
# ── PERSONAS ──────────────────────────────────────────────────
ATLAS_SYSTEM = """Você é ATLAS, um debatedor brilhante e assertivo.
Sua missão: defender a posição PRO com argumentos técnicos precisos e exemplos concretos.
Estilo:
- Direto, confiante, sem hesitação
- Use dados, benchmarks e casos reais quando possível
- Antecipe objeções e as neutralize preventivamente
- Máximo 3 parágrafos por turno
- Comece sempre com seu argumento mais forte
- Nunca concorde com o oponente, mas reconheça pontos válidos antes de refutá-los
Você é ATLAS. Responda em português."""
NEXUS_SYSTEM = """Você é NEXUS, um crítico implacável e analítico.
Sua missão: atacar a posição com contra-argumentos devastadores e expor fragilidades.
Estilo:
- Cético, preciso, cirúrgico nas críticas
- Desmonte os argumentos do oponente ponto a ponto
- Traga evidências contraditórias e edge cases
- Máximo 3 parágrafos por turno
- Comece sempre atacando o argumento mais fraco do oponente
- Use perguntas retóricas para expor contradições
Você é NEXUS. Responda em português."""
ORACLE_SYSTEM = """Você é ORACLE, o juiz supremo do debate.
Sua missão: avaliar os argumentos com imparcialidade absoluta e emitir um veredicto fundamentado.
Analise o debate completo e responda EXCLUSIVAMENTE com JSON válido:
{
"vencedor": "ATLAS" ou "NEXUS" ou "EMPATE",
"placar_atlas": <0-10>,
"placar_nexus": <0-10>,
"criterios": {
"qualidade_argumentos": {"atlas": <0-10>, "nexus": <0-10>},
"uso_de_evidencias": {"atlas": <0-10>, "nexus": <0-10>},
"coerencia_logica": {"atlas": <0-10>, "nexus": <0-10>},
"poder_de_refutacao": {"atlas": <0-10>, "nexus": <0-10>}
},
"ponto_decisivo": "O argumento/momento que decidiu o debate em 1-2 frases",
"analise_atlas": "Análise do desempenho de ATLAS em 2-3 frases",
"analise_nexus": "Análise do desempenho de NEXUS em 2-3 frases",
"licao": "Principal insight técnico que emerge do debate em 1 frase"
}"""
# ── DEBATE ENGINE ─────────────────────────────────────────────
class DebateEngine:
def __init__(self, openai_api_key: str):
self.client = OpenAI(api_key=openai_api_key)
self.model = MODEL
self.history: list = [] # histórico completo do debate
def _call(self, system: str, messages: list,
temperature: float = 0.8, max_tokens: int = 500) -> str:
resp = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "system", "content": system}] + messages,
temperature=temperature,
max_tokens=max_tokens,
)
return resp.choices[0].message.content.strip()
def _event(self, agent: str, role: str, content: str, turn: int, elapsed_ms: int) -> Dict[str, Any]:
return {
"agent": agent,
"role": role,
"content": content,
"turn": turn,
"elapsed_ms": elapsed_ms,
}
def run_debate(self, topic: str, n_rounds: int = 3) -> Generator[Dict[str, Any], None, None]:
"""
Executa o debate e faz yield de cada evento em tempo real.
Evento: {agent, role, content, turn, elapsed_ms}
"""
t0 = time.time()
ms = lambda: int((time.time() - t0) * 1000)
self.history = []
# ── Abertura: ATLAS abre ──────────────────────────────
yield {"type": "status", "msg": f"ATLAS abre o debate sobre: {topic}"}
opening_prompt = (
f"O tema do debate é: **{topic}**\n\n"
f"Você é PRO. Faça seu argumento de abertura. Seja impactante."
)
atlas_msg = [{"role": "user", "content": opening_prompt}]
atlas_response = self._call(ATLAS_SYSTEM, atlas_msg, temperature=0.85)
self.history.append({"role": "atlas", "content": atlas_response, "turn": 0})
yield self._event("ATLAS", "pro", atlas_response, 0, ms())
# ── Rounds ────────────────────────────────────────────
for round_n in range(1, n_rounds + 1):
yield {"type": "status", "msg": f"Round {round_n}/{n_rounds}"}
# NEXUS responde
nexus_context = self._build_context_for("NEXUS", topic)
nexus_response = self._call(NEXUS_SYSTEM, nexus_context, temperature=0.85)
self.history.append({"role": "nexus", "content": nexus_response, "turn": round_n})
yield self._event("NEXUS", "con", nexus_response, round_n, ms())
# ATLAS contra-argumenta
atlas_context = self._build_context_for("ATLAS", topic)
atlas_response = self._call(ATLAS_SYSTEM, atlas_context, temperature=0.85)
self.history.append({"role": "atlas", "content": atlas_response, "turn": round_n})
yield self._event("ATLAS", "pro", atlas_response, round_n, ms())
# ── Closing statements ────────────────────────────────
yield {"type": "status", "msg": "Argumentos finais..."}
for agent, system, role in [
("NEXUS", NEXUS_SYSTEM, "con"),
("ATLAS", ATLAS_SYSTEM, "pro"),
]:
ctx = self._build_context_for(agent, topic)
ctx[-1]["content"] += "\n\nFaça seu argumento FINAL. Seja conclusivo e poderoso."
response = self._call(system, ctx, temperature=0.9, max_tokens=400)
self.history.append({"role": agent.lower(), "content": response, "turn": n_rounds + 1})
yield self._event(agent, role, response, n_rounds + 1, ms())
# ── ORACLE julga ──────────────────────────────────────
yield {"type": "status", "msg": "ORACLE deliberando..."}
debate_transcript = self._build_transcript(topic)
import json, re
raw = self._call(ORACLE_SYSTEM, [{"role": "user", "content": debate_transcript}],
temperature=0.1, max_tokens=600)
raw = re.sub(r'```json|```', '', raw).strip()
try:
verdict = json.loads(raw)
except Exception:
verdict = {
"vencedor": "EMPATE", "placar_atlas": 7, "placar_nexus": 7,
"criterios": {
"qualidade_argumentos": {"atlas": 7, "nexus": 7},
"uso_de_evidencias": {"atlas": 7, "nexus": 7},
"coerencia_logica": {"atlas": 7, "nexus": 7},
"poder_de_refutacao": {"atlas": 7, "nexus": 7},
},
"ponto_decisivo": "Debate equilibrado.",
"analise_atlas": "Boa performance.",
"analise_nexus": "Boa performance.",
"licao": raw[:200],
}
yield {"type": "verdict", "data": verdict, "elapsed_ms": ms()}
def _build_context_for(self, agent: str, topic: str) -> list:
"""Constrói o histórico de mensagens no formato correto para cada agente."""
messages = [{"role": "user", "content": f"Tema do debate: {topic}\n\nHistórico:"}]
for h in self.history:
label = "ATLAS (PRO)" if h["role"] == "atlas" else "NEXUS (CON)"
messages[0]["content"] += f"\n\n{label}:\n{h['content']}"
messages.append({"role": "user", "content": "Sua vez. Responda ao último argumento."})
return messages
def _build_transcript(self, topic: str) -> str:
lines = [f"TEMA: {topic}\n"]
for h in self.history:
label = "ATLAS (PRO)" if h["role"] == "atlas" else "NEXUS (CON)"
lines.append(f"{label} [turno {h['turn']}]:\n{h['content']}\n")
lines.append("\nAvalie este debate completo e emita seu veredicto em JSON.")
return "\n".join(lines)