result-aggregator

Agent Result Aggregator

Quand l'utiliser

Situation	Pattern recommandé
Chaque agent couvre une partie distincte	Fusion complémentaire
Plusieurs agents traitent la même question	Déduplication + ranking
Les agents aboutissent à des conclusions différentes	Résolution de conflits
Un agent a échoué / timeout	Fallback sur résultats partiels

---

Workflow en 7 étapes

1. Définir le schéma de sortie avant tout

Spécifier le contrat du résultat final avant de lancer les sous-agents.

from pydantic import BaseModel
from typing import Any

class AggregatedResult(BaseModel):
    summary: str
    details: dict[str, Any]
    sources: list[dict]      # {agent_id, confidence, status}
    confidence: float        # min des confidences individuelles, pas la moyenne
    conflicts_resolved: list[dict]
    metadata: dict           # nb_agents, success_rate, duration_ms

Critère de décision : si le schéma change après l'agrégation, l'étape est trop tardive.

---

2. Collecter en parallèle avec timeout strict

import asyncio

async def collect(agents: list, timeout: int = 60) -> list:
    tasks = [asyncio.wait_for(a.get_result(), timeout=timeout) for a in agents]
    raw = await asyncio.gather(*tasks, return_exceptions=True)
    return [
        {"agent_id": agents[i].id, "status": "failed", "data": None, "confidence": 0.0}
        if isinstance(r, Exception)
        else {"agent_id": agents[i].id, "status": "complete", "data": r, "confidence": r.confidence}
        for i, r in enumerate(raw)
    ]

Règle : ne jamais bloquer sur un agent lent. Timeout = SLA de l'agent le plus lent × 1,5.

---

3. Valider et scorer chaque résultat

Catégories : valid · partial · invalid · empty

def validate(result: dict, required_fields: list[str]) -> dict:
    if result["status"] == "failed":
        return {"category": "empty", "score": 0.0}
    data = result.get("data") or {}
    missing = [f for f in required_fields if f not in data]
    if missing:
        return {"category": "partial", "score": 0.4, "missing": missing}
    confidence_bonus = result.get("confidence", 0) * 0.3
    return {"category": "valid", "score": 0.7 + confidence_bonus}