Handoff Designer

Agent Handoff Designer

Quand utiliser ce skill

Un handoff est nécessaire dans quatre situations :

| Situation | Signal | Agent cible | |---|---|---| | Capability boundary | Outil ou connaissance manquant | Spécialiste équipé | | Spécialisation | Sous-tâche mieux traitée ailleurs | Agent optimisé | | Escalation | Risque ou complexité dépassant le seuil | Agent senior / humain | | Load balancing | Queue saturée | Instance parallèle |

Décision : si l'agent courant peut compléter la tâche avec une qualité ≥ 80 % sans outil supplémentaire, ne pas handoff — le coût de transition n'est pas justifié.

---

Workflow en étapes

1. Évaluer le déclenchement

from enum import Enum

class HandoffReason(Enum):
    CAPABILITY_BOUNDARY = "capability_boundary"
    SPECIALIZATION      = "specialization"
    ESCALATION          = "escalation"
    LOAD_BALANCING      = "load_balancing"

def should_handoff(task: dict, agent_caps: list[str]) -> tuple[bool, HandoffReason | None]:
    missing = [c for c in task.get("required_capabilities", []) if c not in agent_caps]
    if missing:
        return True, HandoffReason.CAPABILITY_BOUNDARY
    if task.get("risk_level", 0) > 7:
        return True, HandoffReason.ESCALATION
    return False, None

Critères de seuil recommandés : risk_level > 7, complexity_score > 8, `sentimentscore HandoffContext: summary = summarizeconversation(history) # appel LLM interne facts = extractkeyfacts(history) ref = storecontextsnapshot(history) # Redis / DynamoDB / mémoire partagée return HandoffContext( correlationid=str(uuid.uuid4()), sourceagent=source, targetagent=target, handoffreason=reason, conversationsummary=summary, userintent=detectintent(history), keyfacts=facts, taskprogress={}, userpreferences={}, contextref=ref, handoffchain=[source], )

**Règle de taille** : résumé + faits = max 2 000 tokens. Le transcript complet va dans le state store, pas dans le package.

---

### 3. Exécuter le protocole (avec ACK obligatoire)

```python
import asyncio

MAX_HOPS     = 5
TIMEOUT_S    = 5.0
RETRY_DELAY  = 2.0

async def execute_handoff(
    source_agent,
    target_agent,
    ctx: HandoffContext,
) -> dict:
    # Garde anti-boucle
    if ctx.hop_count >= MAX_HOPS:
        return {"success": False, "reason": "max_hops_exceeded"}
    if target_agent.id in ctx.handoff_chain:
        return {"success": False, "reason": "cycle_detected"}

    ctx.hop_count += 1
    ctx.handoff_chain.append(target_agent.id)

    for attempt in range(2):            # 1 essai + 1 retry
        try:
            ack = await asyncio.wait_for(
                target_agent.notify_handoff(ctx), timeout=TIMEOUT_S
            )
            if ack.get("ready"):
                source_agent.set_state("standby")
                return {"success": True, "target": target_agent.id}
        except asyncio.TimeoutError:
            if attempt == 0:
                await asyncio.sleep(RETRY_DELAY)

    # Fallback hiérarchique
    alt = find_alternative_agent(target_agent.id)
    if alt:
        return await execute_handoff(source_agent, alt, ctx)

    source_agent.set_state("active")    # return-to-sender
    return {"success": False, "reason": "target_unavailable"}

Séquence obligatoire : notify → ACK ready → switch source → confirm orchestrateur. Ne jamais passer en standby sans ACK.

---

4. Routing conditionnel

Définis les règles dans une table pure (testable unitairement) :

ROUTING_RULES = [
    {"cond": lambda c: c.get("intent") == "billing",        "target": "billing_agent"},
    {"cond": lambda c: c.get("sentiment_score", 1.0)  8,    "target": "senior_agent"},
    {"cond": lambda c: c.get("language") == "ar",           "target": "arabic_specialist"},
]

def route_handoff(context: dict) -> str:
    for rule in ROUTING_RULES:
        if rule["cond"](context):
            return rule["target"]
    return "default_agent"

LangGraph (2026) — utilise Command(goto="node", update={...}) dans le nœud source :

from langgraph.types import Command

def billing_router(state):
    if state["intent"] == "billing":
        return Command(goto="billing_node", update={"ctx": package_context(...)})
    return Command(goto="default_node")

OpenAI Assistants — partage le thread_id, annule le run courant, démarre un run sur l'assistant cible :

client.beta.threads.runs.cancel(thread_id=tid, run_id=rid)
new_run = client.beta.threads.runs.create(thread_id=tid, assistant_id=TARGET_ASST_ID)

CrewAI — Agent.delegate(task, agent=target_agent) + inject le contexte dans la description de la tâche.

Google A2A (2026) — expose un AgentCard avec capabilities, envoie un Task JSON via POST /tasks/send avec metadata.handoff_context.

---

5. UX transparente côté utilisateur

L'agent récepteur doit toujours :

1. S'introduire brièvement : "Je prends en charge votre demande depuis [Agent A]."
2. Démontrer le contexte : "Je vois que vous recherchez une facture pour le mois de mars…"
3. Ne pas reposer une question déjà répondue — vérifier key_facts avant toute question.
4. Si une info manque, expliquer pourquoi : "J'ai besoin de votre numéro client car il n'est pas encore dans le dossier."

Message de transition UI (optionnel) :

{ "type": "system_event", "event": "handoff_started",
  "message": "Transfert vers le spécialiste Facturation en cours…",
  "target_agent": "billing_agent" }

---

6. Chaînes multi-hop

À chaque hop :

• Incrémenter hop_count, bloquer si >= MAX_HOPS.
• Accumuler les résumés (pas re-résumer le résumé — perte de signal garanti).
• Conserver le correlation_id original pour le tracing distribué.
• Stocker un snapshot immutable dans le state store (ctx_snap_{hop}.json).

def accumulate_summary(existing: str, new_segment: str) -> str:
    # Concatène avec séparateur de hop — ne pas re-résumer
    return f"{existing}\n---hop---\n{new_segment}"

---

7. Métriques à suivre

| Métrique | Calcul | Seuil alerte | |---|---|---| | Taux de succès | handoffs réussis / total | 3 s | | Fréquence de retry | retries / total handoffs | > 10 % | | Task completion post-handoff | tâches complétées après handoff | 2 000 tokens** — Surcharger l'agent récepteur ralentit son inference et noie les faits clés. Fix : le transcript complet va dans le state store, le package ne contient que le résumé structuré.

---

Règles non-négociables

1. HandoffContext obligatoire — Aucun handoff sans résumé + intent + key_facts, même minimal.
2. ACK avant switch — Source reste active jusqu'à réception du ready: true.
3. MAX_HOPS = 5 — Au-delà : escalade superviseur ou humain, jamais de boucle supplémentaire.
4. correlation_id inchangé — Permet le tracing end-to-end sur toute la chaîne.
5. Documente le trade-off — Contexte minimal = latence réduite, risque context loss élevé. Contexte riche = fiabilité élevée, latence +. Choix explicite selon le SLA.

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: khalilbenaz
• Source: khalilbenaz/claude-skills-collection
• License: MIT
• Homepage: https://khalilbenaz.github.io/claude-skills-collection/

Install and usage instructions live in the source repository linked above.