Notes pratiques : Votre graphe d’agents ne doit pas être en Python : Compilation d’un

Les notes suivantes reconstituent une approche pratique pour aborder le sujet « Votre graphe d’agents ne doit pas être écrit en Python : compilation d’un flux de travail multi-agents à partir d’un seul fichier YAML ». L’accent est mis sur les contrats, les vérifications et les placeholders de code à insérer, plutôt que sur une présentation motivante. Lors de la phase d’aperçu, notez d’abord le contrat : les entrées requises, le signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Documentez à la fois le parcours normal et le parcours de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non livrés font partie intégrante du produit, et non d’une mise en forme ultérieure.

Le problème auquel personne ne vous avertit

Le problème que personne ne met en garde concerne le fait que cette étape fonctionne le mieux lorsqu’elle est traitée comme une surface mesurable. Capturez un exemple réussi, un cas d’échec et la note de réversion avant d’élargir le périmètre. Préférez des unités petites et testables aux scripts complexes. Lorsqu’une étape échoue, l’erreur doit indiquer une seule responsabilité plutôt qu’un processus embrouillé. Fixez l’interpréteur ainsi que le fichier de verrouillage des dépendances avant d’enseigner la boucle. Les différences entre l’ordinateur portable et les environnements CI sont la cause la plus fréquente d’échecs silencieux dans les démos API.

À quoi ressemble le flux de travail sous forme de données

Cette étape du flux de travail fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un transcript idéal, un cas d’échec et la note de réversion avant d’élargir le périmètre. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Nommez les artefacts, définites des vérifications de succès et refusez toute mise en œuvre partielle silencieuse. Fixez l’interpréteur ainsi que le fichier de verrouillage des dépendances avant d’enseigner la boucle. Les écarts entre l’ordinateur portable et les outils CI sont la cause la plus fréquente de dysfonctionnements silencieux dans les démos API.

entry: entry_agent
exit: exit

guardrails:
  - Reject queries that are outside the application's domain.
  - Reject queries about the system, agents, design, or internal workings.state_schema:
  query:
    type: str
    description: "User query or current message."
  chat_history:
    type: list
    annotated_with: add_messages
    description: "Conversation history between user and system."
  result:
    type: dict
    description: "Result from the processing agent."agents:
  - name: agent_one
    kind: function
    impl: your_package.agents.agent_one.agent_one_fn  - name: agent_two
    kind: function
    impl: your_package.agents.agent_two.agent_two_fnworkflow:
  nodes:
    - id: agent_one
      agent: agent_one
      writes: [query, result]
      next: decision_router    - id: decision_router
      kind: router
      router:
        impl: your_package.agents.routers.route_after_agent_one
        reads: [result]
        edges:
          agent_two: agent_two
          human_agent: human_agent

Truc n°1 : Générer votre classe d’état en temps de exécution à partir d’un schéma

Le premier astuce : la création de votre environnement de test fonctionne le mieux lorsqu’il est considéré comme une surface mesurable. Capturez un exemple idéal de fonctionnement, un cas d’échec et une note de réversion avant d’élargir le périmètre. Enregistrez les temps d’exécution ainsi que le coût des tokens ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les factures inattendues lorsque le processus passe d’une démonstration à des environnements partagés. Fixez l’interpréteur et le fichier de verrouillage des dépendances avant d’enseigner la boucle de traitement. Les différences entre l’ordinateur portable et les environnements CI constituent la cause la plus fréquente de dysfonctionnement silencieux dans les démonstrations API. Le premier astuce : la création de votre environnement de test fonctionne le mieux lorsqu’il est considéré comme une surface mesurable. Capturez un exemple idéal de fonctionnement, un cas d’échec et une note de réversion avant d’élargir le périmètre. Documentez en même temps le parcours optimal et les procédures de récupération. Les tentatives de réexécution, les contrôles manuels et le traitement des messages non livrés font partie intégrante du produit, et non d’améliorations apportées ultérieurement.

# your_package/orchestrator/schema.py
annotations = {}
for key, value in state_schema.items():
    type_str = value.get("type", "str")
    # Convert YAML string to Python type
    py_type = eval(type_str)
    if value.get("annotated_with") == "add_messages":
        py_type = Annotated[list, {}]
    annotations[key] = py_type

spec = Spec(
    ...
    state=TypedDict("State", annotations),   # <- dynamic class, born at boot
    ...
)

Truc n°2 : Agents référencés par une chaîne ponctuée, résolus via importlib

Pour l’étape des agents référencés du Truc n°2, définissez les entrées, le responsable de l’étape et les critères de fin avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Préférez des unités petites et testables à des scripts complexes. Lorsqu’une étape échoue, l’échec doit pointer vers une seule responsabilité plutôt que vers un pipeline embrouillé. Séparez la construction du client du cycle de messages afin que les fournisseurs puissent être remplacés sans avoir à réécrire la machine d’état de la conversation.

impl: your_package.agents.agent_one.agent_one_fn

# your_package/orchestrator/schema.py
def _import_from_path(dotted: str) -> Callable[..., Any]:
    """Import a callable from a dotted path like 'package.module.function'."""
    if not dotted or "." not in dotted:
        raise ImportError(f"Invalid impl path: {dotted!r}")
    mod_path, attr = dotted.rsplit(".", 1)
    mod = importlib.import_module(mod_path)
    fn = getattr(mod, attr)
    if not callable(fn):
        raise TypeError(f"Imported object is not callable: {dotted}")
    return fn

def agent_impl_map(spec: Spec) -> Dict[str, Optional[Callable]]:
    """Map agent name -> callable (or None if impl missing)."""
    return {a.name: _import_from_path(a.impl) if a.impl else None
            for a in spec.agents}

Truc n°3 : Le compilateur — les nœuds YAML deviennent des nœuds de graphe

Pour la étape 3, celle du compilateur, il convient de définir les entrées, le responsable de l’étape et les critères de fin avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Donnez des noms aux artefacts, définites des vérifications de succès et refusez les terminaisons partielles silencieuses. Séparez la construction du client du cycle de messages afin que les fournisseurs puissent être remplacés sans avoir à réécrire la machine d’états de la conversation.

# your_package/orchestrator/runner.py
def build(self):
    graph = StateGraph(state_schema=self.spec.state)   # our generated TypedDict

    def _add_task_node(node):
        async def _node(state: Dict[str, Any]) -> Dict[str, Any]:
            res = await self._call_agent(node.agent, state, node.id)
            if getattr(node, "writes", None):
                if isinstance(res, dict):
                    # Only let the node write the keys it declared in YAML
                    filtered = {k: v for k, v in res.items() if k in node.writes}
                    return filtered or res
                key = node.writes[0]
                return {key: res}
            return res
        graph.add_node(node.id, _node)    # Build every node
    for node in self.spec.workflow.nodes:
        if getattr(node, "router", None):
            _add_router_node(node)
        else:
            _add_task_node(node)    graph.set_entry_point(entry)    # Inline "next:" edges from YAML become static edges
    for node in self.spec.workflow.nodes:
        if getattr(node, "next", None):
            graph.add_edge(node.id, node.next)    # Terminal nodes wire to END
    for node in self.spec.workflow.nodes:
        if getattr(node, "terminal", False):
            graph.add_edge(node.id, END)    self._runnable = graph.compile(checkpointer=self.checkpoint)
    return self

Routeurs : embranchements conditionnels en tant que table de recherche

Pour les branches conditionnelles des routeurs en tant qu’étape, définissez les entrées, le responsable de l’étape et les critères de sortie avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Enregistrez les temps d’exécution ainsi que le coût des jetons ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les factures inattendues lorsque le parcours passe d’un environnement de démonstration à des environnements partagés. Séparez la construction du client de la boucle de messages afin que les fournisseurs puissent être remplacés sans avoir à réécrire la machine d’état de la conversation. Pour les branches conditionnelles des routeurs en tant qu’étape, définissez les entrées, le responsable de l’étape et les critères de sortie avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Documentez ensemble le parcours optimal et le parcours de récupération. Les tentatives de réessai, les contrôles humains et la gestion des messages non livrés font partie du produit, et non d’un retard.

Polonais.

# your_package/orchestrator/runner.py
def _add_router_node(node):
    router = self.router_fns[node.id]

    def _router_fn():
        def _f(state):
            out = router(state)
            # Routers may return either a label, or (state_updates, label)
            if isinstance(out, tuple):
                updates, label = out
                if isinstance(updates, dict):
                    for k, v in updates.items():
                        state[k] = v
            else:
                label = out
            return label
        return _f    graph.add_node(node.id, lambda s: {})
    graph.add_conditional_edges(node.id, _router_fn(), node.router.edges)

Truc n°4 : L’appel adaptatif — les agents peuvent écrire la signature qu’ils souhaitent

Lors de l’étape adaptative du Truc n°4, notez d’abord le contrat : les entrées requises, le signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Préférez des unités petites et testables plutôt que des scripts volumineux. Lorsqu’une étape échoue, l’échec doit pointer vers une seule responsabilité et non vers un processus complexe. Enregistrez l’ID de la demande, l’ID du modèle et le temps de latence à chaque appel. Sans cette trace, les erreurs intermittentes du fournisseur ressemblent à des bugs de l’application.

# your_package/orchestrator/runner.py
async def _adapt_and_call(self, fn, state, node_id):
    """
    Adaptively call agent functions so implementations receive what they expect:
    - def agent(**kwargs):        → pass **state (+ inject 'query' if missing)
    - def agent(query, **kwargs): → pass query=..., plus any **extra
    - def agent(state):           → pass state
    - def agent(query):           → pass query
    - def agent():                → call without args
    """
    sig = inspect.signature(fn)
    params = sig.parameters
    has_var_kw = any(p.kind == inspect.Parameter.VAR_KEYWORD
                     for p in params.values())

    kwargs = {}
    if has_var_kw:
        kwargs.update(state)
    if "state" in params:
        kwargs["state"] = state
    if "query" in params or has_var_kw:
        kwargs.setdefault("query", self._fallback_query(state))    # A lone positional 'query' → call it positionally
    if (len(params) == 1
            and next(iter(params.keys())) == "query"):
        return await _maybe_await(fn(self._fallback_query(state)))    res = fn(**kwargs)
    return await res if hasattr(res, "__await__") else res

Truc n°5 : Remplacement en temps réel d’un nœud par session (intervention humaine)

Lorsque vous travaillez sur la technique 5 consistant au remplacement en temps réel d’une étape, notez d’abord les conditions prévues : entrées requises, signal de succès, et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Donnez des noms aux éléments concernés, définez des vérifications de succès, et refusez tout achèvement partiel silencieux. Enregistrez l’ID de la demande, l’ID du modèle et le temps de réponse pour chaque appel. Sans ce suivi, les erreurs intermittentes du fournisseur ressemblent à des bugs de l’application.

# your_package/services/session_service.py (paraphrased)
if websocket is not None:
    session_handler = SessionHandler(websocket, user_id=user_id, session_id=session_id, ...)
    _runner.agent_fns["human_agent"] = _import_from_function(
        make_input_method(session_handler)
    )

Qu’est-ce que cette architecture vous apporte réellement

Lors de l’étape « Quelle est réellement cette architecture ? », notez d’abord le contrat : les entrées requises, le signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Enregistrez les temps d’exécution ainsi que le coût des tokens ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les factures inattendues lorsque le processus passe de l’environnement de démonstration à des environnements partagés. Enregistrez l’ID de la requête, l’ID du modèle et le temps de latence pour chaque appel. Sans ce suivi, les erreurs intermittentes du fournisseur sont perçues comme des bugs de l’application. Lors de l’étape « Quelle est réellement cette architecture ? », notez d’abord le contrat : les entrées requises, le signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Documentez ensemble le parcours normal et le parcours de récupération. Les tentatives de réessai, les contrôles humains et la gestion des messages non livrés font partie intégrante du produit, et non d’améliorations apportées ultérieurement.

Le point clé

La phase de récupération fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un exemple réussi, un cas d’échec et la note de réversion avant d’élargir le périmètre. Préférez des unités petites et testables aux scripts complexes. Lorsqu’une étape échoue, l’erreur doit indiquer une seule responsabilité plutôt qu’un processus embrouillé. Fixez l’interpréteur ainsi que le fichier de verrouillage des dépendances avant d’enseigner la boucle. Les différences entre l’ordinateur portable et l’environnement CI sont la cause la plus fréquente d’échecs silencieux lors des démonstrations API.

Liste de contrôle opérationnelle

Pour la phase de liste de contrôle opérationnelle, définissez les entrées, le responsable de l’étape et les critères de fin avant de modifier du code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché.

Gardez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans avoir à lire l’ensemble du système.

Séparez la construction du client du cycle de messages afin que les fournisseurs puissent être remplacés sans avoir à réécrire la machine à états de la conversation.

Faites un point d’état après les opérations coûteuses. La reprise ne doit pas facturer à nouveau la même appel de LLM lorsque l’opérateur réessaie un nœud ultérieur.

Fixez les versions des dépendances et enregistrez le digest de l’image ayant servi à exécuter la démonstration. La reproductibilité vaut mieux que les connaissances internes au groupe.

Considérez cette étape comme un contrat entre les entrées et les sorties validées. Nommez les artefacts, définez des vérifications de succès et refusez toute complétion partielle silencieuse.

Au préalable de promouvoir l’ensemble technique, figez les versions, conservez une transcription « or » pour le chemin critique, et vérifiez les étapes de rollback. Les environnements partagés nécessitent des limites de débit, des contrôles d’attribution, ainsi qu’un responsable clair pour la rotation des secrets. Préférez une fiabilité solide à des démonstrations brillantes mais ponctuelles.

Note de batch pour 2822ea5988ca : gardez les clés du fournisseur hors du répertoire, fixez un plafond pour les tokens par session, et stockez les transcriptions à côté des fichiers d’évaluation afin que les remplacements ultérieurs de modèles restent comparables.