换交互层¶

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 处理(拉式)


async def answerer(ch: HumanChannel) -> None:
    """拉式取提问。换成 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 用 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())

await rt.run(spec, "…", on_event=sink)                  # 1. 单个 agent
await wf.run(rt, on_event=sink, on_step=progress)       # 2. 整个流程,转交给每一步
wf = Workflow(steps=[...], channel=ch)                  # 3. 提问通道,接到同一个出口

ch = HumanChannel(on_event=my_own_sink)     # 自己接,Workflow 不动它

import select
import sys
import threading


def start_input(ch: HumanChannel) -> threading.Event:
    """一直读标准输入:有待答提问就是答案,否则进收件箱。返回停止位。"""
    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)                    # 进收件箱,不打断在飞的活

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

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


def sink(ev: Event) -> None:            # 同步、在事件循环所在的线程里、不能阻塞
    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)}

import itertools
from collections import deque

seq = itertools.count(1)
log: deque[dict] = deque(maxlen=2000)   # 只留最近的,内存不跟着运行时长涨


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 = 这个提问已经不在等了

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)       # 事件全丢弃

@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

@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(
    *,                                  # 全部 keyword-only
    on_event=None,                      # 推式出口。Workflow 只在它是 None 时才自动接
    max_asks=None,                      # None = 不限;0 = 不许问。超额直接回绝,不阻塞
    timeout_s=1800.0,                   # None = 永远等;<= 0 = 立刻落空
    log_path=None,                      # 问答追加进这个文件,不占上下文
    amend_path=None,                    # 人在运行途中说的话追加进这个文件,通常是需求确认书
    over_budget_text=OVER_BUDGET,       # 三句固定回复,可以换成自己的
    timeout_text=TIMEOUT,
    declined_text=DECLINED,
)