Замена слоя взаимодействия¶

Ядро flower не знает, что UI существует. Каждое событие внутри run — модель говорит, вызывает инструмент, контекст почти заполнен, надо спросить человека — сплющивается в одну и ту же структуру данных Event. Слой взаимодействия знает только Event и не импортирует ни одного типа SDK. Это и есть та граница, которая позволяет менять UI, не трогая ядро: терминал, Web, HTTP-сервис, полностью автоматический режим без человека — меняется потребитель Event, всё остальное не меняется ни на строку.

Какую задачу это решает¶

Поток сообщений SDK — это внутренние типы: AssistantMessage, ToolUseBlock, ToolResultBlock, ResultMessage, SystemMessage… Если UI потребляет их напрямую, получаются два следствия: SDK обновился — фронтенд обязан меняться следом; и у каждого типа сообщения своя форма, поэтому в каждом UI приходится заново писать разбор «это основной текст или вызов инструмента».

normalize(message) превращает одно сообщение SDK в от 0 до N Event (core/events.py). Цена — одно преобразование, выигрыш — отсутствие типовой зависимости между слоем взаимодействия и SDK.

Заодно эта граница решает ещё четыре не столь очевидные вещи, и все четыре — внутри normalize():

1. Реплики subagent помечаются (payload["subagent"]). Иначе задание на раздачу работы и промежуточные реплики subagent смешаются с основным текстом главного потока, а дальше по workflow загрязнят prompt следующего шага.
2. Синтетические сообщения об ошибке при обрыве связи отводятся в kind="error". При обрыве SDK записывает API Error: … в transcript как assistant-сообщение, и выглядит оно как речь модели (model равен "<synthetic>"). Если не перехватить его здесь, оно попадёт в StepResult.text и уедет в следующий шаг.
3. Границы compact сообщаются явно (kind="reset"). После границы модель «помнит» только резюме, и кэш prompt тоже рвётся именно здесь — long-horizon run обязан это видеть.
4. Уровень заполнения контекста идёт с каждым сообщением (payload["context"] = input_tokens + cache_read_input_tokens + cache_creation_input_tokens). Это единственный источник для критерия handoff.

Как этим пользоваться (минимальный код)¶

Слою взаимодействия нужно подключить три вещи: выход событий (куда рендерить), канал вопросов (кто отвечает) и прерывание (как остановить). Ниже подключено всё, код запускается как есть:

import asyncio

from flower import Event, HumanChannel, Runtime, starter_flow


def sink(ev: Event) -> None:
    """Отрисовать Event в вашем UI — это единственное, что нужно менять."""
    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", который не mail, обрабатывает answerer ниже (pull-модель)


async def answerer(ch: HumanChannel) -> None:
    """Pull-режим получения вопросов. При переходе на Web / HTTP менять надо и эту корутину."""
    while True:
        ask = await ch.next_ask()           # без timeout ждёт бесконечно
        if ask is None:
            continue
        print(f"\n?? {ask.question} 选项={ask.options}")
        ch.answer(ask.id, "按你的判断来")     # или ch.decline(ask.id, "先跳过")


async def main() -> None:
    wf = starter_flow("帮我做一个 X", workspace=".", run_dir="runs", timeout_s=60)
    # Runtime использует workbench, созданный самим workflow — не собирайте второй
    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()                          # закрыть соединение SQLite
    print(rt.total_cost(), ctx.get("_failed_at"))


asyncio.run(main())

Две вещи в завершении, которые легко упустить: rt.close() обязательно в finally; если в ctx["_failed_at"] есть значение — run остановился на полпути (on_fail="stop"), не считайте это успехом.

Три выхода событий¶

await rt.run(spec, "…", on_event=sink)                  # 1. отдельный agent
await wf.run(rt, on_event=sink, on_step=progress)       # 2. весь workflow, передаётся каждому шагу
wf = Workflow(steps=[...], channel=ch)                  # 3. канал вопросов, на тот же выход

Третий вариант подключается внутри Workflow.run: автоподключение происходит, только если on_event не None и channel.on_event всё ещё None. Если вы подключили выход сами, он не будет перезаписан:

ch = HumanChannel(on_event=my_own_sink)     # подключено вами, Workflow это не трогает

on_step(step, result) — второй callback, вызывается один раз после каждого шага (включая упавшие) и получает полный StepResult. Прогресс-бар, запись на диск, алерты вешайте сюда, не собирайте это из потока Event — основной текст будет разорван handoff и повторами на несколько кусков.

Терминал: тот, что по умолчанию¶

Он есть и без вашего кода. flower "帮我做一个 X" идёт через flower/cli.py — это референсная реализация слоя взаимодействия, а не часть фреймворка, её можно заменить целиком; ключи см. в справочнике CLI. Скажем честно о масштабе. Весь cli.py — это 1264 строки, 57KB, — но менять надо не весь файл. Настоящая точка замены — класс Render внутри него (cli.py:489-687, 197 строк), в его docstring прямо написано: «Event → терминал. Смена UI — это смена одного этого класса.» Остальная тысяча с лишним строк — прерывание, oracle, квитанции inbox, спасение по сигналам, то есть специфичная для терминала обвязка, которую при переходе на Web или HTTP переносить и не нужно.

Так что формулировка «около 200 строк заменяются целиком» верна — при условии, что речь о Render, а не о cli.py.

Если пишете терминальный UI сами, ключевой момент — поток, читающий стандартный ввод:

import select
import sys
import threading


def start_input(ch: HumanChannel) -> threading.Event:
    """Постоянно читать stdin: есть висящий вопрос — это ответ, иначе в inbox. Возвращает флаг остановки."""
    stop = threading.Event()

    def loop() -> None:
        while not stop.is_set():
            if not select.select([sys.stdin], [], [], 0.2)[0]:
                continue                        # опрос, чтобы реагировать на флаг остановки
            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)      # безопасно между потоками
            else:
                ch.send(raw)                    # в inbox, не прерывая работу в полёте

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

Все три пункта набиты на практике:

• Daemon-поток, а не asyncio.to_thread(input, ...). Заблокированный input() нельзя отменить, а asyncio.run перед выходом делает join потоков дефолтного executor — в итоге работа закончена, а чтобы выйти, нужно ещё раз нажать Enter.
• Опрос через select, а не input() прямо в цикле. Та же невозможность отмены: поток, застрявший в input(), уже не разбудить никаким stop.set().
• Читать постоянно, а не только когда есть вопрос. Если читать только при наличии вопроса, то всё, что человек набрал за те несколько часов работы, останется в буфере терминала и будет съедено как ответ на следующий вопрос — человек ещё не увидел вопроса, а на него уже «ответили».

Web: очередь + WebSocket¶

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


def sink(ev: Event) -> None:            # синхронно, в потоке event loop, блокировать нельзя
    try:
        events.put_nowait({"kind": ev.kind, "text": ev.text,
                           "tool": ev.tool, "payload": ev.payload})
    except Exception:                   # ошибка фронтенда не должна уносить трёхчасовую работу
        pass


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


@app.post("/answer")                    # поток обработки запроса — другой поток, и это норма
def answer(ask_id: str, text: str) -> dict:
    return {"ok": ch.answer(ask_id, text)}

ev.raw — это исходный объект SDK (в событии ask — Ask), он не сериализуется в JSON и не должен уезжать на фронтенд: воспользовались raw — значит, привязали фронтенд обратно к типам SDK, и весь этот слой сделан зря. Четырёх полей kind / text / tool / payload достаточно.

HTTP: порядковый номер + опрос¶

Когда постоянного соединения нет, нумеруйте события, чтобы клиент их вычитывал:

import itertools
from collections import deque

seq = itertools.count(1)
log: deque[dict] = deque(maxlen=2000)   # храним только последние, память не растёт вместе с длительностью run


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")                       # что сейчас висит в ожидании ответа
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 = этот вопрос уже никто не ждёт

Здесь две границы, которые надо признать: при заполнении maxlen теряются самые старые записи, и клиент, вернувшийся с очень старым after, получит неполную выборку — интервал опроса должен соответствовать этой длине; и timeout_s обязан иметь конечное значение — когда никто не опрашивает, вопрос сам не завершится, а timeout_s=None подвесит весь run навсегда. Значение по умолчанию 1800.0 секунд подходит.

Полностью автоматический режим без человека¶

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)       # все события отбрасываются

Эквивалент в командной строке — flower "帮我做一个 X" --timeout 0.

timeout_s=0 (и любое отрицательное значение) — это полностью автоматический режим: вопрос не попадает в очередь ожидания и не порождает событие asked, он сразу закрывается как state="timeout", и инструмент возвращает такой фиксированный текст —

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

— и run продолжается без остановки. Вопросы и ответы всё равно дописываются в HumanChannel(log_path=...) (starter_flow по умолчанию подключает <工作台>/notes/问答记录.md), так что потом видно, что он спрашивал и какие допущения принял.

Если вообще не хотите, чтобы он открывал рот, используйте max_asks=0: вопрос сразу отклоняется (state="over_budget") и тоже не блокирует. Обратите внимание: это не «убрать инструмент» — allowed_tools не является исключающим списком, и как только к координатору подключён channel, ему выдаются оба инструмента, mcp__human__ask и mcp__human__inbox, и он может вызвать их независимо от того, перечислены они или нет. Остановить вопрос могут только квота и таймаут.

Что оно делает на самом деле¶

Форма Event¶

@dataclass
class Event:
    kind: EventKind                     # 15 значений, см. таблицу ниже
    text: str = ""
    tool: str = ""                      # заполнено только у tool_call
    payload: dict[str, Any] = field(default_factory=dict)
    raw: Any = None                     # исходный объект SDK / Ask, тронул — привязался к SDK

str(ev): для tool_call это [имя инструмента] краткое описание, для остальных — text; если text пуст — <kind>.

15 значений EventKind¶

| ask | HumanChannel | Нужен ответ человека, у вопроса появился исход, или человек сам что-то сказал | Вопрос / реплика человека | Две роли, см. ниже | | retry | Runtime | Идёт повтор / ожидание сети | Одна поясняющая строка | step, attempt | | step | Workflow.run | Граница шага | Имя шага | index, total, resumed, woke | | handoff | Runtime | Handoff: приближение / запись / завершение | Пояснение с уровнем заполнения | phase, step, context, window + см. ниже |

Четыре kind порождаются не в normalize(): ask приходит из HumanChannel, retry и handoff — из Runtime, step — из Workflow.run. Помещать их в один и тот же EventKind — сознательное решение: UI знает один набор Event и не должен заводить отдельный путь для «нужен ответ человека» или «граница шага».

Когда пишете UI, оставляйте ветку else. В EventKind будут добавляться новые значения, и старый UI не должен из-за этого падать.

Три phase у handoff¶

Сам механизм описан в handoff.

Две роли ask¶

Event("ask") несёт одновременно и «вопрос», и «реплику, которую человек сказал сам», поэтому UI обязан сначала посмотреть payload["kind"]:

Один вопрос порождает не менее двух событий: при постановке (state="asked") и при появлении исхода (answered / timeout / declined / over_budget / invalid). UI достаточно обновлять одну и ту же запись по payload["id"].

Спросить человека: Ask и 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"                                # пять исходов, см. выше
    answer: str = ""

    @property
    def waited_s(self) -> float: ...                    # сколько секунд ждал, один знак после запятой
    def event(self, remaining: int = 0) -> Event: ...

HumanChannel — это внутрипроцессный MCP-сервер плюс набор методов для UI. Со стороны модели видны только два инструмента: mcp__human__ask (задать вопрос, вызов подвисает в ожидании) и mcp__human__inbox (проверить inbox, не блокирует, при пустом ящике сразу возвращает поясняющую строку). Полный конструктор:

HumanChannel(
    *,                                  # всё keyword-only
    on_event=None,                      # push-выход. Workflow подключает автоматически, только если здесь None
    max_asks=None,                      # None = без ограничения; 0 = спрашивать нельзя. Сверх квоты — отказ, без блокировки
    timeout_s=1800.0,                   # None = ждать вечно; <= 0 = сразу пустой ответ
    log_path=None,                      # вопросы и ответы дописываются в этот файл, контекст не занимают
    amend_path=None,                    # реплики человека по ходу run дописываются в этот файл, обычно это brief
    over_budget_text=OVER_BUDGET,       # три фиксированных ответа, можно заменить своими
    timeout_text=TIMEOUT,
    declined_text=DECLINED,
)

amend_path пропускают чаще всего, а именно он определяет, переживёт ли границу шага изменённое человеком по ходу дела требование. Каждый шаг — это новая сессия и read-only срез: сказанное во время run попало только в контекст того agent, который тогда работал, а следующий шаг (например, вердикт) — совершенно новая сессия, читающая 需求.md и 目标.md, и она не видит вашей реплики, поэтому судит по старым границам и признаёт исправленное выходом за рамки. amend_path дописывает каждое сообщение в brief — именно дописывает, а не перезаписывает: прежнее требование остаётся историей, и видеть, что именно изменилось, лучше, чем не видеть. starter_flow по умолчанию подключает <工作台>/notes/需求.md.

На практике ($0.6767) это сработало даже лучше ожидаемого: человек сказал «заодно сообщи суммарное число байт», координатор проверил inbox и ответил — «hand уже прочитал это из дописанного по ходу run раздела в .flower/notes/需求.md и посчитал, повторно отправлять не нужно». Subagent прочитал это из файла, а не с чьего-то пересказа.

Публичные члены:

answer, decline, send можно вызывать из любого потока. Поток обработки запросов Web-бэкенда и поток ввода TUI живут в других потоках — это норма, а не краевой случай. Внутри используется loop.call_soon_threadsafe, потому что asyncio.Future.set_result не потокобезопасен.

Push и pull — выбирайте что-то одно:

Три варианта «0 / None» означают разное, и если их перепутать — в автономном режиме будет либо зависание, либо ни одного вопроса:

Прерывание: остановить можно из любого потока¶

rt.interrupt("别改 Makefile,那两行直接改"); пустая строка — просто прервать, ничего не говоря. Три свойства:

• Продолжение той же сессии (resume), а не старт с нуля — сделанная работа и контекст на месте. Переиспользуется готовый путь повтора после обрыва сети, меняется только «причина сбоя» на «человек прервал» и resume_prompt на реплику человека.
• Не расходует max_attempts. Это квота для сбоев, а не для человека.
• Кооперативно: разрыв происходит на границе сообщения, задача не отменяется жёстко. Цена — задержка до следующего сообщения (если subagent сейчас работает, надо дождаться его возврата), выигрыш — состояние не рвётся на полпути.

Скажем и о цене честно: прерывание заставляет subagent в полёте потерять полуфабрикат (проверено на обрыве сети в HT001, см. issue #2). Референсная реализация в терминале пишет об этом прямо в подсказке, чтобы человек знал до нажатия; в своём UI стоит сделать так же.

Если прерывать не хочется, а надо лишь добавить требование, используйте inbox (ch.send(...)) — он ничего не прерывает, задержка равна ближайшей контрольной точке agent.

Oracle: спросить, не мешая run¶

Чтобы узнать, «где мы сейчас», не нужно прерывать, и спрашивать координатора тоже не следует: такой диалог навсегда займёт контекст главного потока (там лежат решения, а не протокол вопросов и ответов), и ему придётся отложить текущую работу. За десятичасовой run три мимоходом заданных вопроса оплачивают обе эти цены.

Oracle — это read-only обходной путь. У него только инструменты Read / Glob / Grep, workbench включён, ограничители по умолчанию: max_turns=12, max_budget_usd=0.5. В терминале он вызывается строкой, начинающейся с ?, и отвечает по двум источникам: окно последних событий (фиксированные 60 штук) и brief, цель, заметки и артефакты в workbench. Он использует отдельный Runtime (<run_dir>/aside), поэтому затраты и происхождение сессий не попадают в основной manifest — этот манифест фиксирует, «какие шаги выполнил данный run», а мимоходом заданный вопрос шагом не является.

По замерам два вопроса стоили в сумме $0.5190, и manifest основного run не вырос ни на байт.

Два жёстких правила¶

Когда этого делать не нужно¶

Чего не следует делать в слое взаимодействия¶

Когда менять вообще не надо¶

• Хочется лишь поменять цвета, добавить или убрать одну строку вывода — достаточно поправить функцию рендеринга. Прерывание, oracle, квитанции inbox, спасение по SIGHUP / SIGTERM, ожидание завершения обходного пути перед выходом в референсной терминальной реализации переписывать заново дорого.
• Нужно выполнить один шаг без интерактива — используйте flower once. Он не идёт по пути интерактивного драйвера, и там изначально нет ни прерывания по Ctrl+C, ни потока ответов на stdin, ни oracle, ни спасения по сигналам.
• На самом деле хочется поменять workflow, а не UI — см. проектирование workflow. Слой взаимодействия решает только, кто смотрит и кто отвечает; сколько шагов выполнять, как выносить вердикт и когда выходить досрочно, решает Workflow.
• Хочется поменять session store, модель или бюджет — ни одно из трёх не лежит на этой границе, см. справочник Python API.