state-synchronizer

Agent State Synchronizer

Quand utiliser ce skill

Utilise ce skill quand plusieurs agents accèdent et modifient un état commun en parallèle : collecte distribuée, workflows coordinés, récupération après panne. Sans synchronisation explicite : race conditions, écrasements silencieux, résultats non déterministes.

---

Workflow en 8 étapes

1. Définir le shared state (schéma minimal)

Ne partager que ce qui est nécessaire à la coordination — pas l'état interne de chaque agent.

from pydantic import BaseModel
from typing import Any
from datetime import datetime

class SharedState(BaseModel):
    version: int = 0
    schema_version: str = "1.0"
    task_assignments: dict[str, str] = {}   # task_id → agent_id
    task_results: dict[str, Any] = {}        # task_id → result
    agent_status: dict[str, str] = {}        # agent_id → "idle"|"working"|"done"|"error"
    global_context: dict[str, Any] = {}      # lecture seule pour tous
    last_updated: datetime = datetime.utcnow()
    last_updated_by: str = ""

Critères de design :

• READ_ALL / WRITE_OWN — chaque agent n'écrit que ses propres champs → moins de conflits.
• Fine-grained (un verrou par champ) vs coarse-grained (un seul verrou global) : préférer fine-grained sauf si les transactions multi-champs sont fréquentes.
• Versionner le schéma (schema_version) dès le départ pour faciliter les migrations.

---

2. Choisir le state store

Store	Cas d'usage	Avantages	Limites
Dict Python	Mono-process, tests	Ultra-rapide, zéro infra	Pas de persistance, un seul process
Redis	Multi-process, dev/prod	Atomic ops, pub/sub, TTL natif	Consistance éventuelle par défaut
PostgreSQL	Persistance forte requise	ACID, SELECT FOR UPDATE	Plus lent, surcharge opérationnelle
Event log	Auditabilité, replay	Immuable, debuggable	Reconstruction de l'état coûteuse

Recommandation : Redis pour la majorité des systèmes multi-agents en 2026.