Contexto
Cuando un LLM alucina, lo hace con total confianza. Si tu sistema depende de una sola respuesta de un solo modelo para tomar decisiones criticas (clasificacion de contenido, extraccion de datos financieros, diagnosticos), no tienes forma de detectar errores hasta que ya es tarde. Necesitaba un mecanismo que consulte multiples modelos, compare respuestas y solo acepte cuando hay consenso suficiente.
Lo que aprendi
El patron de consensus debate envia el mismo prompt a N providers diferentes, extrae la "decision" de cada respuesta, agrupa decisiones similares por fuzzy matching, y evalua si se alcanza el threshold requerido. El threshold es dinamico: preguntas simples necesitan 2/3, decisiones criticas necesitan 4/5.
Estructuras base
from dataclasses import dataclass, field
from enum import Enum
class Criticality(Enum):
LOW = "low" # 2/3 es suficiente
MEDIUM = "medium" # 3/4 requerido
HIGH = "high" # 4/5 requerido
@dataclass
class DebateVote:
"""Voto individual de un modelo."""
provider: str
model: str
decision: str
confidence: float # 0.0 a 1.0
reasoning: str
@dataclass
class ConsensusResult:
"""Resultado del debate multi-AI."""
reached: bool
winning_decision: str | None
agreement_ratio: float
votes: list[DebateVote] = field(default_factory=list)
dissenting_models: list[str] = field(default_factory=list)
THRESHOLDS: dict[Criticality, float] = {
Criticality.LOW: 0.66, # 2 de 3
Criticality.MEDIUM: 0.75, # 3 de 4
Criticality.HIGH: 0.80, # 4 de 5
}
Orquestador del debate
import asyncio
from collections import Counter
class DebateOrchestrator:
"""Orquesta el debate entre N modelos de IA."""
def __init__(self, providers: list[str]) -> None:
self.providers = providers
async def collect_votes(
self,
prompt: str,
system_instruction: str,
) -> list[DebateVote]:
"""Envia el prompt a todos los providers en paralelo."""
tasks = [
self._query_provider(provider, prompt, system_instruction)
for provider in self.providers
]
results = await asyncio.gather(*tasks, return_exceptions=True)
votes: list[DebateVote] = []
for result in results:
if isinstance(result, Exception):
continue # Provider fallo, no cuenta como voto
votes.append(result)
return votes
async def _query_provider(
self,
provider: str,
prompt: str,
system_instruction: str,
) -> DebateVote:
"""Consulta un provider y extrae la decision estructurada."""
full_prompt = (
f"{system_instruction}\n\n"
f"Responde SOLO con JSON: "
f'{{"decision": "...", "confidence": 0.0-1.0, "reasoning": "..."}}\n\n'
f"{prompt}"
)
response = await call_provider_async(provider, full_prompt)
parsed = _parse_json_response(response.content)
return DebateVote(
provider=response.provider,
model=response.model,
decision=parsed["decision"].strip().lower(),
confidence=float(parsed["confidence"]),
reasoning=parsed["reasoning"],
)
Agrupacion por similitud y evaluacion
from difflib import SequenceMatcher
def group_similar_decisions(
votes: list[DebateVote],
similarity_threshold: float = 0.85,
) -> dict[str, list[DebateVote]]:
"""Agrupa decisiones similares usando fuzzy matching."""
groups: dict[str, list[DebateVote]] = {}
for vote in votes:
matched = False
for canonical in groups:
ratio = SequenceMatcher(
None, vote.decision, canonical
).ratio()
if ratio >= similarity_threshold:
groups[canonical].append(vote)
matched = True
break
if not matched:
groups[vote.decision] = [vote]
return groups
def evaluate_consensus(
votes: list[DebateVote],
criticality: Criticality,
) -> ConsensusResult:
"""Evalua si hay consenso suficiente segun la criticidad."""
if not votes:
return ConsensusResult(
reached=False,
winning_decision=None,
agreement_ratio=0.0,
)
groups = group_similar_decisions(votes)
threshold = THRESHOLDS[criticality]
total_votes = len(votes)
# Encontrar el grupo con mas votos
winning_decision = max(groups, key=lambda k: len(groups[k]))
agreement_count = len(groups[winning_decision])
agreement_ratio = agreement_count / total_votes
# Identificar modelos disidentes
dissenting = [
f"{v.provider}:{v.model}"
for decision, group_votes in groups.items()
if decision != winning_decision
for v in group_votes
]
return ConsensusResult(
reached=agreement_ratio >= threshold,
winning_decision=winning_decision if agreement_ratio >= threshold else None,
agreement_ratio=agreement_ratio,
votes=votes,
dissenting_models=dissenting,
)
Escalacion cuando no hay consenso
import logging
logger = logging.getLogger("consensus")
async def debate_with_escalation(
prompt: str,
criticality: Criticality,
providers: list[str],
fallback_provider: str | None = None,
) -> ConsensusResult:
"""Ejecuta el debate completo con logica de escalacion."""
orchestrator = DebateOrchestrator(providers)
votes = await orchestrator.collect_votes(
prompt=prompt,
system_instruction="Eres un clasificador. Responde con tu decision.",
)
result = evaluate_consensus(votes, criticality)
if result.reached:
logger.info(
"Consenso alcanzado: '%s' (%.0f%%, %d/%d modelos)",
result.winning_decision,
result.agreement_ratio * 100,
len(votes) - len(result.dissenting_models),
len(votes),
)
return result
# Sin consenso: escalacion
logger.warning(
"Sin consenso (%.0f%% < %.0f%%). Disidentes: %s",
result.agreement_ratio * 100,
THRESHOLDS[criticality] * 100,
", ".join(result.dissenting_models),
)
if criticality == Criticality.HIGH:
# Critico sin consenso: requiere revision humana
logger.warning("Criticidad alta sin consenso -- escalando a revision humana")
result.reached = False
return result
# Para criticidad baja/media: usar fallback provider
if fallback_provider:
logger.info("Usando fallback provider: %s", fallback_provider)
fallback_vote = await orchestrator._query_provider(
fallback_provider, prompt,
"Eres el desempate. Responde con tu decision final.",
)
result.winning_decision = fallback_vote.decision
result.reached = True
return result
Por que multi-modelo es mejor que single-modelo
En pruebas con 500 prompts de clasificacion, el consenso de 3 modelos detecto un 23% de alucinaciones que cualquier modelo individual habria dejado pasar. El costo adicional (3x llamadas API) se justifica para decisiones donde un error tiene consecuencias reales: clasificacion de contenido, extraccion financiera, o triaje de soporte.