Changer de couche d'interaction¶

Le cœur de flower ignore l'existence de toute UI. Chaque chose qui se produit dans une exécution — le modèle parle, appelle un outil, le contexte est presque plein, il faut poser une question à un humain — est aplatie vers une seule et même structure de données : Event. La couche d'interaction ne connaît que Event et n'importe aucun type du SDK. C'est la frontière qui permet de changer d'UI sans toucher au cœur : terminal, Web, service HTTP, entièrement automatique et sans surveillance — ce qui change, c'est le consommateur d'Event, rien d'autre, pas une ligne.

Quel problème cela résout¶

Le flux de messages du SDK expose des types internes : AssistantMessage, ToolUseBlock, ToolResultBlock, ResultMessage, SystemMessage… Les consommer directement dans l'UI a deux conséquences : dès que le SDK monte de version, le front doit suivre ; et comme chaque forme de message est différente, chaque UI doit réécrire la logique « est-ce du corps de texte ou un appel d'outil ».

normalize(message) transforme un message du SDK en 0 à N Event (core/events.py). Le coût est une conversion ; ce qu'on obtient, c'est l'absence de dépendance de type entre la couche d'interaction et le SDK.

Cette frontière règle au passage quatre choses moins évidentes, toutes les quatre dans normalize() :

1. Les prises de parole des subagents sont marquées (payload["subagent"]). Sans cela, le brief de tâche envoyé et les paroles intermédiaires du subagent se mélangeraient au corps de texte du thread principal, et pollueraient ensuite, le long du workflow, le prompt de l'étape suivante.
2. Les messages d'erreur synthétiques produits lors d'une coupure réseau sont aiguillés vers kind="error". Lors d'une coupure, le SDK écrit API Error: … dans le transcript comme s'il s'agissait d'un message assistant ; ça ressemble à une parole du modèle (model vaut "<synthetic>"). Si on ne l'intercepte pas ici, ça finit dans StepResult.text, puis est transmis à l'étape suivante.
3. Les frontières de compact sont signalées explicitement (kind="reset"). Après la frontière, le modèle ne « se souvient » que du résumé, et le cache de prompt est rompu à cet endroit — une exécution à long horizon doit pouvoir le voir.
4. Le niveau de contexte sort avec chaque message (payload["context"] = input_tokens + cache_read_input_tokens + cache_creation_input_tokens). C'est la seule source du critère de handoff.

Comment s'en servir (code minimal)¶

Une couche d'interaction doit brancher trois choses : la sortie d'événements (où rendre), le canal de questions (qui répond), l'interruption (comment crier stop). Le bloc ci-dessous branche tout, il tourne tel quel :

import asyncio

from flower import Event, HumanChannel, Runtime, starter_flow


def sink(ev: Event) -> None:
    """Rend l'Event dans votre propre UI — c'est la seule chose à remplacer."""
    if ev.kind == "step":
        print(f"\n=== {ev.text} ({ev.payload['index']}/{ev.payload['total']}) ===")
    elif ev.kind == "text" and not ev.payload.get("subagent"):
        print(ev.text)
    elif ev.kind == "tool_call":
        print(f"  [{ev.tool}] {ev.text}")
    elif ev.kind == "handoff":
        print(f"  ~ 换代/{ev.payload.get('phase')}: {ev.text}")
    elif ev.kind == "retry":
        print(f"  ~ 重试: {ev.text}")
    elif ev.kind == "ask" and ev.payload.get("kind") == "mail":
        print(f"  ~ 人主动说:{ev.text}")
    # kind == "ask" qui n'est pas du mail : traité plus bas par answerer (mode pull)


async def answerer(ch: HumanChannel) -> None:
    """Récupération des questions en mode pull. Pour du Web / HTTP, c'est l'autre endroit à changer."""
    while True:
        ask = await ch.next_ask()           # sans timeout, on attend indéfiniment
        if ask is None:
            continue
        print(f"\n?? {ask.question} 选项={ask.options}")
        ch.answer(ask.id, "按你的判断来")     # ou ch.decline(ask.id, "先跳过")


async def main() -> None:
    wf = starter_flow("帮我做一个 X", workspace=".", run_dir="runs", timeout_s=60)
    # Runtime utilise le workbench déjà créé par le workflow — n'en fabriquez pas un autre
    rt = Runtime(workspace=".", run_dir="runs", workbench=wf.workbench)
    task = asyncio.create_task(answerer(wf.channel))
    try:
        ctx = await wf.run(rt, on_event=sink)
    finally:
        task.cancel()
        rt.close()                          # ferme la connexion SQLite
    print(rt.total_cost(), ctx.get("_failed_at"))


asyncio.run(main())

Deux finitions faciles à oublier : rt.close() doit impérativement être dans finally ; si ctx["_failed_at"] a une valeur, c'est que l'exécution s'est arrêtée en route (on_fail="stop") — ne la prenez pas pour un succès.

Trois sorties d'événements¶

await rt.run(spec, "…", on_event=sink)                  # 1. un seul agent
await wf.run(rt, on_event=sink, on_step=progress)       # 2. tout le workflow, transmis à chaque étape
wf = Workflow(steps=[...], channel=ch)                  # 3. canal de questions, branché sur la même sortie

Le branchement du troisième cas se fait dans Workflow.run : il n'est automatique que si on_event est non None et que channel.on_event est encore None. Si vous l'avez branché vous-même, il n'est pas écrasé :

ch = HumanChannel(on_event=my_own_sink)     # branché à la main, Workflow n'y touche pas

on_step(step, result) est un autre callback, appelé une fois à la fin de chaque étape (y compris en cas d'échec), et il reçoit le StepResult complet. Barre de progression, écriture sur disque, alertes : accrochez-les ici, n'essayez pas de les reconstituer depuis le flux d'Event — le corps de texte est découpé en plusieurs morceaux par les handoffs et les retries.

Terminal : celui par défaut¶

Il y en a un même sans écrire de code. flower "帮我做一个 X" passe par flower/cli.py, qui est une implémentation de référence de la couche d'interaction, pas une partie du framework ; elle peut être remplacée intégralement. Les options sont dans la référence CLI. Disons la taille telle qu'elle est. cli.py en entier fait 1264 lignes, 57KB — mais ce n'est pas l'ensemble qu'il faut remplacer. Le vrai point de remplacement, c'est la class Render à l'intérieur (cli.py:489-687, 197 lignes), dont la docstring dit justement : « Event → terminal. Changer d'UI, c'est changer cette seule classe. » Les mille et quelques lignes restantes sont l'interruption, l'oracle, les accusés de réception de la boîte de réception, le sauvetage sur signal — de l'outillage spécifique au terminal, qu'il n'y a de toute façon pas lieu de recopier pour du Web ou du HTTP.

Donc l'affirmation « environ 200 lignes remplaçables en bloc » tient — à condition qu'elle désigne Render, pas cli.py.

Si vous écrivez votre propre UI terminal, le point clé est le thread qui lit l'entrée standard :

import select
import sys
import threading


def start_input(ch: HumanChannel) -> threading.Event:
    """Lit l'entrée standard en continu : s'il y a une question en attente c'est une réponse, sinon ça va dans la boîte de réception. Renvoie le drapeau d'arrêt."""
    stop = threading.Event()

    def loop() -> None:
        while not stop.is_set():
            if not select.select([sys.stdin], [], [], 0.2)[0]:
                continue                        # polling, seul moyen de réagir au drapeau d'arrêt
            line = sys.stdin.readline()
            if not line:                        # EOF
                return
            raw = line.strip()
            if not raw:
                continue
            pend = ch.pending()
            if pend:
                ch.answer(pend[0].id, raw)      # sûr entre threads
            else:
                ch.send(raw)                    # va dans la boîte de réception, n'interrompt pas le travail en vol

    threading.Thread(target=loop, daemon=True, name="stdin").start()
    return stop

Les trois points ont été appris à la dure :

• Utiliser un thread daemon, pas asyncio.to_thread(input, ...). input() bloqué ne peut pas être annulé, et asyncio.run doit joindre les threads de l'exécuteur par défaut avant de sortir — résultat : le travail est fini, mais il faut encore appuyer une fois sur Entrée pour quitter.
• Faire du polling avec select, pas un input() direct dans la boucle. Même problème d'annulation : un thread bloqué sur input() ne sera jamais réveillé par stop.set().
• Lire en continu, pas seulement quand il y a une question. Si on ne lit que quand il y a une question, tout ce qui a été tapé pendant les heures de travail reste dans le tampon du terminal et sera avalé comme réponse à la question suivante — l'humain n'a même pas encore vu la question qu'elle est déjà « répondue ».

Web : file d'attente + WebSocket¶

events: asyncio.Queue[dict] = asyncio.Queue()


def sink(ev: Event) -> None:            # synchrone, dans le thread de la boucle d'événements, ne doit pas bloquer
    try:
        events.put_nowait({"kind": ev.kind, "text": ev.text,
                           "tool": ev.tool, "payload": ev.payload})
    except Exception:                   # une erreur du front ne doit pas emporter trois heures de travail
        pass


async def pump(ws) -> None:
    while True:
        await ws.send_json(await events.get())


@app.post("/answer")                    # thread de traitement des requêtes — un autre thread, c'est la norme
def answer(ask_id: str, text: str) -> dict:
    return {"ok": ch.answer(ask_id, text)}

ev.raw est l'objet SDK d'origine (un Ask dans les événements ask) : non sérialisable en JSON, et à ne pas transmettre au front — utiliser raw, c'est réattacher le front aux types du SDK, et toute cette couche n'aura servi à rien. Les quatre champs kind / text / tool / payload suffisent.

HTTP : numéro de séquence + polling¶

Sans connexion longue, numérotez les événements pour que le client les tire :

import itertools
from collections import deque

seq = itertools.count(1)
log: deque[dict] = deque(maxlen=2000)   # on ne garde que les plus récents, la mémoire ne croît pas avec la durée de l'exécution


def sink(ev: Event) -> None:
    log.append({"seq": next(seq), "kind": ev.kind, "text": ev.text,
                "tool": ev.tool, "payload": ev.payload})


@app.get("/events")                     # GET /events?after=128
def events(after: int = 0) -> list[dict]:
    return [e for e in log if e["seq"] > after]


@app.get("/asks")                       # ce qui est en attente de réponse en ce moment
def asks() -> list[dict]:
    return [{"id": a.id, "question": a.question, "options": a.options,
             "waited_s": a.waited_s} for a in ch.pending()]


@app.post("/answer")
def answer(ask_id: str, text: str) -> dict:
    return {"ok": ch.answer(ask_id, text)}      # False = cette question n'est plus en attente

Deux limites à reconnaître : quand maxlen est plein, les plus anciens sont perdus, et un client qui revient avec un after très vieux n'obtiendra pas tout — l'intervalle de polling doit être cohérent avec cette longueur ; et il faut absolument donner une valeur finie à timeout_s — quand personne ne fait de polling, une question ne se termine pas d'elle-même, et timeout_s=None laisse toute l'exécution suspendue pour toujours. La valeur par défaut de 1800.0 secondes est appropriée.

Entièrement automatique, sans surveillance : personne¶

wf = starter_flow("帮我做一个 X", workspace=".", run_dir="runs", timeout_s=0)
rt = Runtime(workspace=".", run_dir="runs", workbench=wf.workbench)
ctx = await wf.run(rt, on_event=None)       # tous les événements sont jetés

L'équivalent en ligne de commande est flower "帮我做一个 X" --timeout 0.

timeout_s=0 (idem pour les négatifs) est le mode entièrement automatique : la question n'entre pas dans la file d'attente et n'émet pas d'événement asked, elle est immédiatement soldée en state="timeout", et l'outil renvoie ce texte fixe —

无人应答。按你自己的判断继续,并把这个问题和你采用的假设写进「未知与假设」那一段。不要重复提问,也不要停在这里。

— et l'exécution continue sans encombre. Les questions-réponses restent ajoutées à HumanChannel(log_path=...) (starter_flow branche par défaut <工作台>/notes/问答记录.md), ce qui permet de voir après coup ce qui a été demandé et quelles hypothèses ont été prises.

Si vous ne voulez pas qu'il ouvre la bouche, utilisez max_asks=0 : la question est directement refusée (state="over_budget"), sans bloquer non plus. Attention, ce n'est pas « retirer l'outil » — allowed_tools n'est pas exclusif : dès qu'un channel est branché sur le coordinateur, les deux outils mcp__human__ask et mcp__human__inbox sont donnés ensemble, et restent appelables qu'ils soient listés ou non. Seuls le quota et le timeout peuvent empêcher une question.

Ce qu'il fait réellement¶

La forme d'Event¶

@dataclass
class Event:
    kind: EventKind                     # 15 valeurs, voir le tableau ci-dessous
    text: str = ""
    tool: str = ""                      # renseigné uniquement pour tool_call
    payload: dict[str, Any] = field(default_factory=dict)
    raw: Any = None                     # objet SDK d'origine / Ask, y toucher c'est se réattacher au SDK

str(ev) : pour tool_call c'est [nom d'outil] résumé, sinon c'est text ; si text est vide, c'est <kind>.

Les 15 EventKind¶

| ask | HumanChannel | Il faut une réponse humaine, une question a trouvé son issue, ou un humain a parlé de lui-même | La question / la parole de l'humain | Deux identités, voir plus bas | | retry | Runtime | Retry en cours / en attente du réseau | Une phrase d'explication | step, attempt | | step | Workflow.run | Frontière d'étape | Nom de l'étape | index, total, resumed, woke | | handoff | Runtime | Handoff : approche / écriture en cours / terminé | Une phrase avec le niveau de contexte | phase, step, context, window + voir plus bas |

Quatre kinds ne sont pas produits par normalize() : ask vient de HumanChannel, retry et handoff viennent de Runtime, step vient de Workflow.run. Les mettre dans le même EventKind est délibéré — l'UI ne connaît qu'un seul jeu d'Event, sans avoir à ouvrir une voie séparée pour « il faut une réponse humaine » ou « frontière d'étape ».

En écrivant une UI, laissez une branche else. EventKind gagnera de nouveaux membres, et une vieille UI ne devrait pas planter pour autant.

Les trois phases de handoff¶

Pour le mécanisme lui-même, voir handoff.

Les deux identités d'ask¶

Event("ask") porte à la fois « une question » et « une parole spontanée de l'humain ». L'UI doit d'abord regarder payload["kind"] :

Une question émet au moins deux événements : un au moment de la question (state="asked"), un autre à l'issue (answered / timeout / declined / over_budget / invalid). L'UI n'a qu'à mettre à jour la même entrée d'après payload["id"].

Demander à l'humain : Ask et HumanChannel¶

@dataclass
class Ask:
    id: str                                             # "q1", "q2"…
    question: str
    options: list[str] = field(default_factory=list)
    asked_at: float = field(default_factory=time.time)
    state: str = "asked"                                # voir les cinq issues ci-dessus
    answer: str = ""

    @property
    def waited_s(self) -> float: ...                    # secondes d'attente, une décimale
    def event(self, remaining: int = 0) -> Event: ...

HumanChannel est un serveur MCP in-process plus un jeu de méthodes destinées à l'UI. Côté modèle, seuls deux outils sont visibles : mcp__human__ask (poser une question, ça suspend et attend) et mcp__human__inbox (consulter la boîte de réception, sans blocage ; si elle est vide, ça renvoie immédiatement une phrase d'explication). Construction complète :

HumanChannel(
    *,                                  # tout en keyword-only
    on_event=None,                      # sortie en mode push. Workflow ne la branche automatiquement que si elle vaut None
    max_asks=None,                      # None = illimité ; 0 = interdit de demander. Au-delà du quota, refus direct, sans blocage
    timeout_s=1800.0,                   # None = attendre pour toujours ; <= 0 = échec immédiat
    log_path=None,                      # les questions-réponses sont ajoutées à ce fichier, sans occuper le contexte
    amend_path=None,                    # ce que l'humain dit en cours d'exécution est ajouté à ce fichier, en général le brief
    over_budget_text=OVER_BUDGET,       # trois réponses fixes, remplaçables par les vôtres
    timeout_text=TIMEOUT,
    declined_text=DECLINED,
)

amend_path est le paramètre le plus facile à oublier, et c'est lui qui détermine si « les exigences modifiées en cours de route par l'humain » survivent aux frontières d'étape. Chaque étape est une nouvelle session avec un instantané figé en lecture seule : ce qui est dit en cours d'exécution n'entre que dans le contexte de l'agent du moment ; l'étape suivante (par exemple le verdict) est une session toute neuve qui lit 需求.md et 目标.md, et ne voit pas ce que vous avez dit — elle juge donc encore selon les anciennes limites et déclare hors périmètre ce qui vient d'être corrigé. amend_path ajoute chaque message au brief — en ajout et non en écrasement : l'exigence initiale est de l'histoire, et voir ce qui a changé vaut mieux que ne pas le voir. starter_flow branche par défaut <工作台>/notes/需求.md.

Constaté en pratique ($0.6767), ce mécanisme marche encore mieux que prévu : l'humain dit « au passage, rapporte le nombre total d'octets » ; le coordinateur consulte la boîte de réception et rapporte — « hand l'a déjà lu dans le complément en cours d'exécution de .flower/notes/需求.md et l'a calculé, pas besoin de redéléguer ». Le subagent l'a lu dans le fichier, sans que personne le lui répète.

Membres publics :

answer, decline et send peuvent être appelés depuis n'importe quel thread. Le thread de traitement des requêtes d'un backend Web, le thread d'entrée d'une TUI sont dans d'autres threads — c'est la norme, pas un cas limite. En interne, ça passe par loop.call_soon_threadsafe, car asyncio.Future.set_result n'est pas thread-safe.

Push et pull sont deux modes de récupération : choisissez-en un :

Trois sémantiques « 0 / None » toutes différentes ; les confondre, c'est un blocage définitif sans surveillance ou pas la moindre question posée :

Interruption : n'importe quel thread peut crier stop¶

rt.interrupt("别改 Makefile,那两行直接改") ; une chaîne vide interrompt sans rien dire. Trois propriétés :

• Reprise de la même session (resume), pas un redémarrage à zéro — le travail déjà fait et le contexte sont conservés. On réutilise la voie existante du retry après coupure réseau, en remplaçant simplement la « cause de l'échec » par « l'humain a interrompu » et le resume_prompt par la parole de l'humain.
• Ne consomme pas max_attempts. C'est un quota pour les pannes, pas pour les humains.
• Coopératif : on coupe à une frontière de message, sans annuler brutalement la tâche. Le coût, c'est un délai jusqu'au message suivant (si un subagent est en cours, il faut attendre son retour) ; ce qu'on gagne, c'est de ne pas déchirer l'état en plein milieu.

Le coût, dit tel quel : une interruption fait perdre le travail à moitié fait d'un subagent en vol (constaté lors de la coupure réseau HT001, voir issue #2). L'implémentation de référence du terminal l'écrit explicitement dans son invite, pour qu'on le sache avant d'appuyer ; votre propre UI devrait faire pareil.

Si vous ne voulez pas interrompre mais simplement ajouter une exigence, utilisez la boîte de réception (ch.send(...)) — elle n'interrompt rien, et le délai est le prochain point de contrôle de l'agent.

Oracle : poser une question sans déranger l'exécution¶

Pour savoir « où on en est », inutile d'interrompre, et il ne faut pas demander au coordinateur : cet échange occuperait définitivement le contexte du thread principal (qui contient des décisions, pas un historique de questions-réponses), et il devrait interrompre ce qu'il est en train de faire. Sur une exécution de dix heures, trois questions posées en passant suffisent à payer ces deux coûts.

L'oracle est une voie latérale en lecture seule. Ses seuls outils sont Read / Glob / Grep, il a le workbench ouvert, et il vient avec des garde-fous par défaut : max_turns=12, max_budget_usd=0.5. Dans le terminal, une ligne commençant par ? le déclenche ; il répond à partir de deux choses : la fenêtre d'événements récents (60 entrées, fixe) et le brief, les objectifs, les notes et les livrables du workbench. Il utilise son propre Runtime (<run_dir>/aside), de sorte que le coût et le lignage de session ne se mélangent pas au manifeste principal — ce manifeste consigne « quelles étapes cette exécution a effectuées », et une question posée en passant n'est pas une étape.

Constaté en pratique : deux questions pour $0.5190 au total, sans un octet de plus dans le manifeste de l'exécution principale.

Deux règles dures¶

Quand ne pas s'en servir¶

Ce qu'il ne faut pas faire dans la couche d'interaction¶

Quand il n'y a tout simplement rien à remplacer¶

• Vous voulez seulement changer les couleurs, afficher une ligne de plus ou de moins — il suffit de modifier la fonction de rendu. Dans l'implémentation de référence du terminal, l'interruption, l'oracle, les accusés de réception de la boîte de réception, le sauvetage sur SIGHUP / SIGTERM, l'attente de la fin de la voie latérale avant de quitter : tout réécrire coûte cher.
• Vous voulez seulement exécuter une étape, sans interaction — utilisez flower once. Il ne passe pas par le pilote interactif : il n'a de toute façon ni interruption Ctrl+C, ni thread de réponse sur l'entrée standard, ni oracle, ni sauvetage sur signal.
• Ce que vous voulez changer, c'est en fait le workflow, pas l'UI — voir Concevoir un workflow. La couche d'interaction décide seulement qui regarde et qui répond ; c'est Workflow qui décide du nombre d'étapes, de la manière de juger et du moment de sortir par anticipation.
• Ce que vous voulez changer, c'est le stockage de session, le modèle ou le budget — aucun de ces trois éléments n'est sur cette frontière, voir la référence de l'API Python.