Python API¶

这一页穷尽 flower 顶层 __all__ 的 62 个公开符号:签名、参数、默认值、语义、 公开属性和方法。读完不必再打开源码找参数。

组织方式按关心的事情分,不按模块文件分 —— 想知道"怎么拦住协调者 自己动手",去 hook 层;想知道"上一步的结果怎么传给下一步",去流程。 术语一律按术语表。

版本 0.1.0,依赖 claude-agent-sdk>=0.2.152。所有签名逐字对应源码。

from flower import Runtime, Workflow, Step, coordinator, worker   # 顶层一次导入

这页有什么¶

六个会咬人的默认值¶

这六条不是边角料,是最常见的六次翻车。每条在对应小节里有完整说明。

运行时¶

源码:flower/core/runtime.py

Runtime 是执行核心。它持有工作区、会话存储、工作台、 韧性策略和换代策略,对外只有一个动词:run 一步。 重试、被打断后续跑、上下文满了换代,全在这一个调用里面完成。

Runtime¶

Runtime(
    *,
    workspace: str | Path,
    run_dir: str | Path = "runs",
    portable: bool = True,
    trim: TrimPolicy | bool = False,
    ephemeral: EphemeralPolicy | bool = True,
    keep_denials: int = 1,
    workbench: Workbench | bool = False,
    spill_threshold: int | None = 4000,
    resilience: Resilience | bool = True,
    handoff: HandoffPolicy | bool = True,
)

构造参数全部是 keyword-only(* 在最前),workspace 必填。

会话存储是写死的:永远是 PruningSessionStore(run_dir/"sessions.db", workspace=..., policy=<TrimPolicy>, ephemeral=<EphemeralPolicy>, prune=PrunePolicy(keep_denials=...))。 构造参数不提供换后端的入口 —— 要换,自己构造 AgentSpec + build_options(session_store=...), 或者构造完覆盖 rt.store。

构造的最后两步是 load_dotenv() 和 check_credentials(),后者有错就 raise RuntimeError。 没凭证时是构造阶段就炸,不是等到 run()。

公开属性¶

类常量:INTERRUPTED = "interrupted-by-human"、HANDOFF_DUE = "context-full-handoff"、 INTERRUPT_NOTE(打断续跑时附在人的话后面的一段,说明"当时在飞的工具调用返回 interrupted 是打断的正常副作用,不是环境故障")。

公开方法¶

Runtime.run(...)¶

async def run(
    self,
    spec: AgentSpec,
    prompt: str,
    *,
    step_name: str | None = None,
    resume: str | None = None,
    fork: bool = False,
    resume_at: str | None = None,
    on_event: Callable[[Event], None] | None = None,
) -> StepResult

每步开头把上下文水位归零(self._ctx, self._warned = 0, False)。之后是一个循环,四条出口:

1. 成功 → 跳出。
2. 人打断(result.error == INTERRUPTED)→ 不受 max_attempts 约束,不等网络。 带着人的话 resume 同一条会话,attempt -= 1(打断不算失败尝试),prompt = 人的话 + INTERRUPT_NOTE。 没拿到 session_id 就只能停下。
3. 上下文满(result.error == HANDOFF_DUE,或 handoff.enabled 且拿到了 session_id 且 is_overflow(...) 命中)→ 也不受 max_attempts 约束。先查 len(result.retired) >= handoff.max_generations, 超了就把 error 换成一句诊断并跳出;否则写交接书 → resume=None, fork=False (全新会话)→ prompt 换成 h.prompt_block() → 水位归零 → attempt -= 1。
4. 可重试故障 → not resilience.enabled or attempt >= max_attempts 就跳出; classify(error) 判定不该重试也跳出;否则发 Event("retry")、wait_online() 挂着等网络、 sleep(delay_for(attempt));拿到过 session_id 就 resume 续跑(prompt 换成 resilience.resume_prompt),并把 result.resumed 置 True。

收尾:写 ended_at、追加进 self.results、写 manifest.json。

manifest.json 是追加语义:每次写都重读磁盘,按 run 字段去重(自己的行换新,别人的行留着), 所以同一个 run_dir 下并行跑两个 flower 是安全的 —— 前提是 run_id 不撞。

换代的三个观测点(都是 Event("handoff"),payload["phase"] 区分): near(逼近 warn_at,每代只发一次)、writing(正在写交接,要十几秒)、 done(payload 带 degraded / path / sections)。写交接那一轮用 replace(spec, max_budget_usd=None) 跑 —— 交接必须写得出来,不能卡在预算上; 且 on_event=None,这一轮不往 UI 刷。

交接落盘在 <workbench.notes>/交接-<步骤名>.md;没有工作台就不落盘,文书照样通过 prompt 交给接手者,只是事后翻不到。旧交接被移到 notes/archive/交接/<名>-<时间戳>.md。

StepResult¶

@dataclass
class StepResult:
    step: str
    session_id: str | None = None
    ok: bool = False
    cost_usd: float = 0.0
    num_turns: int = 0
    text: str = ""
    error: str | None = None
    started_at: float = 0.0
    ended_at: float = 0.0
    attempts: int = 1
    errors: list[str] = field(default_factory=list)
    resumed: bool = False
    retired: list[str] = field(default_factory=list)
    context: int = 0

一步跑完的全部账。

流程¶

源码:flower/workflow/

流程是按顺序串起来的一组步骤,外加步骤之间怎么传状态、 什么时候提前退出。框架不提供现成流程,流程是你写的 —— starter_flow 只是一个能跑的样板。

类型别名 Ctx = dict[str, Any](flower.workflow.base.Ctx,在 flower.workflow.__all__ 里, 不在顶层 __all__)。

Step¶

@dataclass
class Step:
    name: str
    spec: AgentSpec
    prompt: str | Callable[[Ctx], str]

    resume_from: str | None = None
    fork: bool = False

    retries: int = 0
    gate: Callable[[StepResult, Ctx], bool] | None = None
    on_fail: str = "stop"
    when: Callable[[Ctx], bool] | None = None
    on_reject: Callable[[StepResult, Ctx], str] | None = None
    resume_prompt: str | Callable[[Ctx], str] | None = None
    reduce: Callable[[StepResult, Ctx], str] | None = None

一个步骤的声明。Step 本身不是函数 —— 真正执行的是 Runtime.run(step.spec, prompt, ...)。 前三个字段是位置参数,Step("取词", terse, "读 seed.txt …") 是合法写法。

三种会话接法(同一次运行内):

on_reject 改变重试语义:

• 不给 → 下次尝试重头跑(同样的 prompt、同样的 resume_from)。
• 给了 → 下次尝试续跑刚被否掉的那条会话,prompt 换成它的返回值,fork 强制为 False。
• 返回空字符串 → 不打回,退化成重头跑。
• result.session_id 为 None → 也退化成重头跑。

on_fail 三种取值:

无论过没过,ctx["_results"][name] = result 都会写;result.session_id 非空时还会写进 ctx["_sessions"] 并 lineage.remember(...)。

Workflow¶

@dataclass
class Workflow:
    steps: list[Step]
    name: str = "workflow"
    context: Ctx = field(default_factory=dict)
    channel: Any = None
    workbench: Any = None
    continuous: bool = True

    async def run(
        self,
        runtime: Runtime,
        *,
        on_event: Callable[[Event], None] | None = None,
        on_step: Callable[[Step, StepResult], None] | None = None,
    ) -> Ctx

按顺序跑一串 Step,返回最终的 ctx。steps 是位置参数,Workflow([...]) 合法。

run() 写进 ctx 的私有键(全部以 _ 开头,不会和步骤名撞):

Event("step") 的 payload:{"index": i, "total": len(steps), "resumed": bool, "woke": int}。

重试标签:第 0 次用 step.name;之后有 on_reject 用 f"{name}#round{attempt+1}", 没有则 f"{name}#retry{attempt}"。manifest 里一眼能看出这一步是怎么走完的。 带后缀的名字不进跨进程血缘 —— Lineage.remember 用的是原名。

runtime.on_session 只罩 runtime.run 这一句,用 try/finally 保证 gate 之前一定摘掉。 prompt_cur / resume_cur / fork_cur 是局部变量,不写回 step —— 同一个 Step 对象可能被跑第二次。

StepAbort¶

class StepAbort(Exception): ...

由 gate 抛出 = 立刻停,不要再重试。和"返回 False"的区别:False 是"这次不行,再来一轮"; StepAbort 是"再来也没用"。

抛出后:ctx["_aborted"] = str(exc)、passed = False、跳出重试循环(不消耗剩余的 retries), 然后按普通失败走 on_fail(默认 "stop")。

with_goal 在两处抛它:拿不到 ctx["_runtime"] 时,以及判定结果是 unreachable 而没人应答时。

clarify_step()¶

def clarify_step(
    channel: HumanChannel,
    *,
    brief_path: str | Path,
    prompt: str | Callable[[Ctx], str],
    name: str = "确认需求",
    spec: AgentSpec | None = None,
    instructions: str = "",
    always_ask: bool = False,
    on_fail: str = "stop",
    retries: int = 0,
    **spec_kw,
) -> Step

产出一个做前置确认的 Step:问清需求 → 解析成 Brief → 四段齐全就冻结落盘。

产出的 Step 里各字段是这样填的:

• resume_prompt = CLARIFY_RESUME。
• when:always_ask=True → 恒 True;否则 Brief.load(brief_path) 齐全就把它灌进 ctx 然后返回 False(跳过) —— 跳过也要灌,否则下游拿不到需求。
• gate:Brief.parse(result.text),不全 → 写 ctx[MISSING_KEY] 并返回 False; 齐全 → b.write(brief_path) 冻结,灌进 ctx,返回 True。
• reduce:返回 ctx[BRIEF_KEY].prompt_block(),不是模型原文 —— 原文里可能夹着它多写的东西。
• resume_from 保持默认 None:下一步是新会话,只拿到确认书,拿不到那段问答。 前置确认的问答从来没进过协调者的上下文,不是进去之后被剪掉的。

灌进 ctx 的三处:ctx[BRIEF_KEY] = b、ctx[name] = b.prompt_block()、ctx.pop(MISSING_KEY, None)。

goal_step()¶

def goal_step(
    channel: HumanChannel,
    *,
    goal_path: str | Path,
    brief_key: str = "确认需求",
    name: str = "设定目标",
    spec: AgentSpec | None = None,
    instructions: str = "",
    always_set: bool = False,
    on_fail: str = "stop",
    retries: int = 0,
    **spec_kw: Any,
) -> Step

产出一个设定目标的 Step:让判定者读确认书,写出目标 + 判定清单, 解析成 Goal 后冻结落盘。和 clarify_step 是同一个形状。

没有 can_run 形参 —— 想让设目标的判定者能跑命令,只能通过 **spec_kw 传 can_run=True。 不传的话它拿不到 Bash,JUDGE_RULES 里"先看清楚你在什么环境"那条执行不了。

gate 除了解析和冻结,还额外做一件事:目标里有 [此环境无法验证:…] 条目时,当场通过 ctx["_on_event"] 发一个 Event("task", payload={"unverifiable", "total", "path"}) 提醒 —— 这些条目的命运在设目标这一刻就定了,等到判定时已经花掉整轮干活的钱。

没有设 resume_prompt —— 设定目标本来就该重发全文确认书。

with_goal()¶

def with_goal(
    step: Step,
    channel: HumanChannel,
    *,
    goal_path: str | Path,
    spec: AgentSpec | None = None,
    rounds: int = 3,
    instructions: str = "",
    can_run: bool = False,
    name: str | None = None,
    **spec_kw: Any,
) -> Step

给一个已有的 Step 套上目标看守:每轮结束后由判定者独立判定, 没达成就打回去接着做。

产出是 replace(step, retries=max(0, rounds - 1), gate=<新 gate>, on_reject=<新 on_reject>) —— 用 dataclasses.replace 而不是逐字段重建,重建过一次就漏了 resume_prompt,而且不报错, 只是接续时把整份确认书又发了一遍。

gate 是 async 的,流程:

1. ctx["_runtime"] 缺失 → 抛 StepAbort("拿不到 Runtime,无法判定目标")。别假装通过。
2. ctx[ROUND_KEY] += 1。
3. 取目标:优先 ctx[GOAL_KEY] 里齐全的 Goal,否则 Goal.load(goal_path),否则空 Goal()。
4. await rt.run(judger, VERIFY_PROMPT..., step_name=f"{label}#{轮次}", on_event=...)。 判定者是独立的一次 Runtime.run,resume 恒为 None —— 永远是新会话;step_name 带轮次, 所以不进跨进程血缘。
5. Verdict.parse(vr.text) 写进 ctx[VERDICT_KEY]。
6. v.achieved → 返回 True。
7. 不是 unreachable(含 v.ok=False 的含糊情况)→ 含糊时补一句默认 reason,返回 False。 含糊一律按未达成 —— 不能让一句"看起来可以"把活收掉。
8. unreachable → await channel.ask(...) 问人,三个选项:
9. 没人应答(a.state != "answered")→ 抛 StepAbort。继续空转是最贵的选择。
10. 「接受这个结果,就这样往下走」→ 返回 True。
11. 「修改目标」→ 再问一次新目标,g.amend(...).write(goal_path)、更新 ctx[GOAL_KEY],返回 False。
12. 其余(含人自己打的自由回答)→ 当"你判断错了",把人的说法记进 v.reason,返回 False。

on_reject 是同步的:返回 ctx[VERDICT_KEY].feedback(),没有 Verdict 就返回 "" (退化成重头跑)。

starter_flow()¶

def starter_flow(
    ask: str,
    *,
    workspace: str | Path = ".",
    run_dir: str | Path = "runs",
    new: bool = False,
    isolate: bool = False,
    clarify_only: bool = False,
    goal: bool = True,
    rounds: int = 3,
    judge_can_run: bool = False,
    max_asks: int | None = None,
    timeout_s: float | None = 1800.0,
    instructions: str = "",
    worker_prompt: str = "你负责实现。每改一处就跑一次验证,别攒到最后。",
    brief_name: str = "需求.md",
    goal_name: str = "目标.md",
    log_name: str = "问答记录.md",
) -> Workflow

装配一个开箱能跑的三步流程:确认需求 → 设定目标 → 干活(带目标看守)。命令行 flower 用的就是它。

固定装配:

Workflow(name="starter", channel=ch, workbench=wb, steps=[...])
# ch = HumanChannel(log_path=<notes>/问答记录.md, amend_path=<brief_path>,
#                   max_asks=max_asks, timeout_s=timeout_s)
# 协调者 = coordinator("协调者", "", {"coder": worker(..., isolate=isolate)}, channel=ch)

行为分支:

• isolate=True 且 workspace 不是 git 仓库 → 抛 ValueError,不等到 Agent 工具报错才发现 (那时钱已经花了)。
• 唤醒判定:Brief.load(brief_path) 存在且 complete() 就算唤醒。不是唤醒且 ask 为空 → 抛 ValueError("要给一句诉求,例如 flower '帮我做一个 X'")。
• 唤醒时那句话同时落到三个地方,少一个都会静默失效:追进确认书 (ch.amend(said, label="唤醒时追加"),已在文件里就不重复写)、 让 goal_step(always_set=True) 重推清单(不重推的话判定者读的还是老目标)、 直接送到协调者面前(它的上下文里是旧目标,不给会按旧标准干然后被按新标准判)。

wake_state()¶

def wake_state(
    workspace: str | Path = ".",
    *,
    run_dir: str | Path = "runs",
    isolate: bool = False,
    brief_name: str = "需求.md",
    goal_name: str = "目标.md",
) -> dict

起跑之前的只读探测,一个字节都不写。 用来在真正开跑前告诉人"这是接着上次,还是从头开始"。

返回的 dict:

工作台位置只在这里和 starter_flow 里定义一次:isolate=True → <ws>.parent/.flower-<ws.name> (仓库外);否则 <ws>/.flower。驱动程序想知道确认书在哪也走这个函数 —— 自己拼路径拼错了不报错, 只是静默失效。

角色工厂¶

源码:flower/core/roles.py

五个角色都是工厂函数。每个角色 = 一段注入的规则文本 + 一组工具 + 一组 hook。 worker() 产出的是 SDK 的 AgentDefinition(派给 subagent 用),另外四个产出 AgentSpec (自己起一条会话)。

角色本身不挂 hook —— 拦工具的活是 Runtime._attempt 按 spec.delegate_only 自动装的, 见 hook 层。

内部工具组常量(未导出,但决定了默认值):

COORDINATOR_TOOLS = ["Agent", "TodoWrite", "Read"]
WEB_TOOLS         = ["WebFetch", "WebSearch"]
WORKER_TOOLS      = ["Read", "Write", "Edit", "Bash", "Glob", "Grep", "WebFetch", "WebSearch"]

coordinator()¶

def coordinator(
    name: str,
    instructions: str,
    workers: dict[str, AgentDefinition],
    *,
    channel: Any = None,
    can_read: bool = True,
    glance: bool = True,
    model: str | None = None,
    effort: str | None = None,
    max_turns: int | None = None,
    max_budget_usd: float | None = None,
    permission_mode: str = "acceptEdits",
    compact: Any = None,
    hooks: dict[str, Any] | None = None,
    env: dict[str, str] | None = None,
) -> AgentSpec

造主线程上那个协调者:拆解任务、派活、读报告、做决策, 但不动手。前三个参数是位置参数。

产出的 AgentSpec 里固定的三项:delegate_only=True、agents=workers、 workbench 保持 AgentSpec 的默认 True。

workers 的 web 工具会被并进来¶

造完清单之后,coordinator() 会遍历每个 AgentDefinition.tools,凡是落在 WEB_TOOLS(WebFetch、WebSearch,roles.py:33)里的,就往协调者自己的 allowed_tools 里也加一份(roles.py:523-526)。

理由:allowed_tools 和 disallowed_tools 一样是会话级的。 这是全文关于这一点最硬的 一条证据 —— 它不只影响主线程。不在这份会话级清单里的工具,subagent 调用时也要走权限审批; 无人值守时没人批,harness 回一句 Claude requested permissions to use X, but you haven't granted it yet (toolDenialKind=user-rejected),而模型会反复重试同一个调用。实测栽过:给执行者加了 WebFetch/WebSearch,却只写进 AgentDefinition.tools,那次 novel 运行二十多次 user-rejected、一个字都没写出来(roles.py:513-518)。

两个字段的会话级性质相同,症状不同:disallowed_tools 是当场报错, allowed_tools 是静默重试到死。后者更难查,因为屏幕上什么都不像出错。

只并只读、无副作用的那些。 Write/Edit/Bash 故意不并:主线程一旦对它们免审批, delegate_guard 那道"协调者不动手"的墙就没意义了;而 subagent 的 Bash/Write 本来就走得通(实测 462 次放行,roles.py:520-522)。

源码里明确写着不要用 disallowed_tools 实现"只协调不动手" —— 那是会话级的, 会把 subagent 的 Bash/Write 一起禁掉,见 AgentSpec 那条警告。 正确做法就是这里的 delegate_only=True + allowed_tools 不给, 再由 delegate_guard 按 agent_id 只拦主线程。

channel 给了就是两个工具一起来,不是可选的:MCP server 一挂就是两个都在,而 allowed_tools 不排他,列不列都调得动。无人值守时每次 ask 都会卡满 timeout_s —— 那种场合用 HumanChannel(timeout_s=0)。

worker()¶

def worker(
    description: str,
    prompt: str,
    *,
    tools: list[str] | None = None,
    model: str = "inherit",
    effort: str | int | None = None,
    max_turns: int | None = None,
    permission_mode: str | None = None,
    skills: list[str] | None = None,
    discipline: bool = True,
    isolate: bool = False,
) -> AgentDefinition

造真正干活的 subagent 定义。前两个参数是位置参数。 返回的是 SDK 的 AgentDefinition,直接塞进 coordinator(workers={...})。

isolate=True 要求 workspace 是 git 仓库,否则 Agent 工具直接报 "not in a git repository", 不会静默退化。而且标记是 Python 属性 —— 对 AgentDefinition 做 dataclasses.replace() 会丢掉它,隔离静默失效。

clarify()¶

def clarify(
    name: str,
    channel: Any,
    *,
    instructions: str = "",
    can_read: bool = True,
    model: str | None = None,
    effort: str | None = None,
    max_turns: int | None = None,
    max_budget_usd: float | None = None,
) -> AgentSpec

造确认者:动手之前把需求问清楚,不做事、只提问,最后输出恰好四段。

产出的 AgentSpec:allowed_tools = [channel.tool_name] + (可读时那五个)、 mcp_servers = channel.mcp_servers()、workbench=False(它没有写工具,索引对它没意义)、 permission_mode 继承 AgentSpec 的默认 "default"。 没有 Write / Edit / Bash / Agent,也没有 inbox(和协调者不同)。

judge()¶

def judge(
    name: str,
    channel: Any,
    *,
    instructions: str = "",
    can_run: bool = False,
    model: str | None = None,
    effort: str | None = None,
    max_turns: int | None = None,
    max_budget_usd: float | None = None,
) -> AgentSpec

造判定者:要么开跑前设定目标,要么每轮结束后判定这一轮。

产出的 AgentSpec:allowed_tools = [channel.tool_name, "Read", "Glob", "Grep"] + (can_run 时)["Bash"]、 workbench=False、其余同 clarify()。没有 Write / Edit / Agent,也没有 inbox。

取舍:can_run=True 判定更硬(能真跑验收命令),代价是判定者因此能改动工作区 —— Bash 本身就能写文件。要绝对中立的判定就别开。

oracle()¶

def oracle(
    name: str = "旁路问答",
    *,
    instructions: str = "",
    model: str | None = None,
    effort: str | None = None,
    max_turns: int | None = 12,
    max_budget_usd: float | None = 0.5,
) -> AgentSpec

造旁路顾问:运行还在跑的时候问它"现在到哪了",它看一眼最近事件和工作台 再回答。它说的话不会进入那次运行的上下文。

产出的 AgentSpec:allowed_tools = ["Read", "Glob", "Grep"](没有 channel —— 它不提问, 只回答)、workbench=True(五个角色里唯一一个非协调者却开工作台的 —— 它就是要去读那些产出和笔记)。

五份规则文本¶

五个常量都在 __all__ 里,可以直接 import 出来读、拼、改。

agent 定义¶

源码:flower/core/agent.py

AgentSpec 是一个专项 agent 的完整声明,build_options 把它编译成 SDK 的 ClaudeAgentOptions。 角色工厂产出的就是 AgentSpec —— 需要工厂之外的组合时,直接构造它。

AgentSpec¶

@dataclass
class AgentSpec:
    name: str
    instructions: str
    allowed_tools: list[str] = field(default_factory=lambda: ["Read", "Glob", "Grep"])
    disallowed_tools: list[str] = field(default_factory=list)
    model: str | None = None
    effort: str | None = None
    max_turns: int | None = None
    max_budget_usd: float | None = None
    permission_mode: str = "default"
    agents: dict[str, Any] | None = None
    mcp_servers: dict[str, Any] = field(default_factory=dict)
    hooks: dict[str, Any] | None = None
    compact: CompactPolicy | None = None
    env: dict[str, str] = field(default_factory=dict)
    glance: bool = False
    workbench: bool = True
    delegate_only: bool = False

build_options()¶

def build_options(
    spec: AgentSpec,
    *,
    cwd: str | Path | None = None,
    session_store: SessionStore | None = None,
    resume: str | None = None,
    fork: bool = False,
    resume_at: str | None = None,
    use_plugin: bool = True,
    portable: bool = True,
    add_dirs: list[str] | None = None,
    flush: str = "eager",
    prelude: str = "",
) -> ClaudeAgentOptions

把 AgentSpec 编译成 SDK 的 ClaudeAgentOptions。Runtime._attempt 内部调的就是它; 自己驱动 SDK(不用 Runtime)时也从这里进。

映射关系:

PLUGIN_DIR 是仓库根的 plugin/(相对 flower/core/agent.py 往上三层)。pip 装完这个目录未必存在, 代码用 is_dir() 判过。

CompactPolicy¶

@dataclass
class CompactPolicy:
    mode: str = "auto"
    window: int | None = None

    def env(self) -> dict[str, str]: ...

auto-压缩的开关面板,产物是一组要注入子进程的环境变量。 压缩算法本身在 harness 二进制里改不了,能改的只有"触不触发"。

HandoffPolicy¶

@dataclass
class HandoffPolicy:
    enabled: bool = True
    window: int = field(default_factory=default_window)
    headroom: int = 50_000
    max_generations: int = 8

    @property
    def at(self) -> int: ...        # max(10_000, window - headroom)
    @property
    def warn_at(self) -> int: ...   # max(1_000, at - 20_000)

上下文快满时"写交接书换新会话"而不是压缩的策略对象。

default_window()¶

def default_window() -> int

按环境变量 ANTHROPIC_MODEL 或 ANTHROPIC_DEFAULT_OPUS_MODEL 的模型名字符串猜上下文窗口:

默认取积极值。判大了不是硬错:API 会以 prompt is too long 退回,Runtime 认这个信号 (内部的 is_overflow)并当场换代 —— 但那一代的交接是降级版本。

文书¶

源码:brief.py · handoff.py · goal.py

四个 dataclass,都是"把模型的一段回话解析成固定几段,再落盘"。共同的形状: parse() 解析、missing() / complete() 查齐不齐、to_markdown() 给人看、 prompt_block() 给下游模型看、write() / load() 落盘与读回。

Brief¶

@dataclass
class Brief:
    goal: str = ""
    accept: str = ""
    bounds: str = ""
    unknowns: str = ""
    path: Path | None = field(default=None, compare=False)

需求确认书,恰好四段,固定顺序 goal → accept → bounds → unknowns,中文段名分别是「目标」「验收标准」「边界」「未知与假设」。

解析规则(易错点集中处):

• 剥围栏时**遇到未闭合的 `` 或_~就从它开始整段丢弃** —— 实测确认者会把整份代码贴进回话。 模型输出被截断时,后面的段全部解析不出来,于是complete()为False`,gate 打回重来。
• 标题正则容忍 ## 目标 / **目标** / 目标: / 3. 边界,也容忍标题后直接跟正文。
• 别名表按长度倒序编译,否则"未知"会先吃掉"未知与假设"。
• 同名段重复出现时取第一个有内容的。
• 手工编辑确认书时若照抄了 to_markdown() 的 "(未填)" 占位文字,那一段仍算缺失。

Handoff¶

@dataclass
class Handoff:
    doing: str = ""
    decided: str = ""
    deadends: str = ""
    next: str = ""
    scene: str = ""
    step: str = ""
    path: Path | None = field(default=None, compare=False)

换代时写的交接书,五段。

必填只有 doing 和 next 两段 —— 硬性要求"走不通的路"非空会逼模型编造。

同模块里三个未导出但语义关键的成员:is_overflow(*texts) 匹配 prompt is too long、 context length exceeded、maximum context length、too many total text bytes、 input length and max_tokens exceed 等,把"硬错"变成"当场换代";HANDOFF_PROMPT 是让 当前会话自己写交接的那段提示词(含 {used} {window} 两个占位符,不是一个新角色 —— 只有它自己有那段上下文);degraded(step, prompt, *, why="") 在交接写不出来时机械拼一份, scene 里塞原始任务的前 1200 字符。

Goal¶

@dataclass
class Goal:
    statement: str = ""
    checks: list[str] = field(default_factory=list)
    path: Path | None = None

目标看守里的目标 + 判定清单。

Verdict¶

@dataclass
class Verdict:
    state: str = ""
    reason: str = ""
    failed: list[str] = field(default_factory=list)

判定者一轮判定的结果,恰好三段:结论 / 理由 / 未通过。

parse 的识别顺序:

1. 先按标题段取「结论」/「判定」。
2. 没有标题段时,整段 strip 后 fullmatch(r"1|true") → 达成;fullmatch(r"0|false") → 没到。
3. 否则在结论文本里按状态词表(长词在前)找第一个命中词。 「无法验证 / 没法验证 / 验证不了 / 无法判定 / unverifiable」全部归到 unreachable —— 实测栽过:目标平台 macOS、跑在 Linux 容器,判定者看源码分支就判了通过。
4. 仍无 → 找孤立的 \b1\b → 达成,\b0\b → 没到。
5. 全都不中 → state=""、ok=False。

unreachable 和 not_yet 是两个不同的结论:前者会走"停下来问人"那条路,不是"再来一轮"。

hook 层¶

源码:flower/core/guard.py

这一层是 flower 的执行边界:哪些工具主线程不许碰、超长结果怎么剪、哪个角色去独立 worktree, 全部由 SDK hook 强制,不靠提示词。理由很直接 —— 提示词是建议,模型可以不听:已实测即便系统提示 明确写着"不要用 worktree",isolate_guard 的注入照样生效(模型传的是 None,落地的是 'worktree')。

九个导出:五个返回 HookMatcher 的 guard 工厂(whitelist_guard 可能返回 None)、一个组装器、一个合并器、两个隔离标记函数。 它们不需要手工挂 —— Runtime 会按 AgentSpec 自动装配。手工挂只在自己驱动 SDK (不经过 Runtime)时才需要。

主线程判定统一走一个函数:_is_main_thread(data) = not data.get("agent_id") —— subagent 的 tool-lifecycle hook 数据里带 agent_id,主线程不带。 所有"只拦主线程"的 guard 都靠这一条。

工具组常量(模块级,未导出,但决定了默认 matcher):

HANDS_ON   = "Bash|Write|Edit|NotebookEdit"
WRITE_ONLY = "Write|Edit|NotebookEdit"
BULKY      = "Bash|Read|Grep|Glob|WebFetch|WebSearch"

速查表:哪个 guard 挂在哪个 SDK 事件上¶

从这张表能读出的关键推论:Runtime(workbench=False) 时 workbench_hooks 整个不装; 而对 delegate_only=True 的协调者,whitelist_guard 也被跳过 —— 主线程一道墙都没有。 详见 Runtime 那条警告。

whitelist_guard()¶

def whitelist_guard(allowed: list[str] | None, *, role: str = "这个角色") -> HookMatcher | None

让 allowed_tools 对动手的那四个工具真正排他。

• 挂在 PreToolUse,matcher 是 "|".join(banned),banned = Bash Write Edit NotebookEdit 中不在 allowed 里的那些。
• 命中即 permissionDecision: "deny",文案大意:"XX 没有 YY。这是有意的,不是配置漏了。 把结论写进你的回话正文里,框架会从那里取 —— 不要试别的写法绕过去。"
• 只拦本会话主线程,subagent 放行 —— subagent 的工具由 AgentDefinition.tools 决定。
• 没有需要拦的工具时返回 None(比如 worker() 那种全套工具的角色),调用方据此决定装不装。

它为什么必须存在:allowed_tools 是免审批清单,不是排他白名单。实测两处证据 —— 设目标的判定者跑了 11 次 Bash;$0.1 的探针里 allowed_tools=["Read"] 的 agent 照样调得动 Write/Bash。 所以 clarify() / judge() 的"没有写工具"靠的是这道 hook,不是白名单本身。

好处是它从 allowed_tools 派生,所以 judge(can_run=True) 自动保留 Bash、仍然拦 Write/Edit —— 不需要额外的开关。

delegate_guard()¶

def delegate_guard(*, tools: str = HANDS_ON, allow_glance: bool = False) -> HookMatcher

主线程自己动手 → 拒绝,并指路。

• 挂在 PreToolUse,matcher 就是 tools。
• 主线程调这四个工具 → deny,理由里给出下一步怎么做:用 Agent 工具派一个 subagent 去做, 任务里写清目标与验收标准,并要求它把长产出写进 .flower/artifacts/、回话只给路径与结论。
• subagent 一律放行。

和 whitelist_guard 的区别是措辞:两者拦同一批工具,但这一道说的是"去派人",更对路。 所以 delegate_only=True 的角色只装这一道,重复装会让模型收到两条矛盾指引。

allow_glance=True 的放行判据和"结果会不会被剪掉"是同一个函数(is_ephemeral)—— 放行集合必须等于过期集合,改一边就必须改另一边。

spill_guard()¶

def spill_guard(
    workbench: Workbench,
    *,
    threshold: int = 4000,
    tools: str = BULKY,
    main_only: bool = False,
) -> HookMatcher

超过阈值的工具结果当场落盘,上下文里只留一行指针 —— 不是等上下文满了 再回头压缩。

• 挂在 PostToolUse,返回 {"hookSpecificOutput": {"hookEventName": "PostToolUse", "updatedToolOutput": <剪过的>}}。
• 落盘文件名是内容的 sha256 前 16 位 + .txt,上下文里换成一行指针 + 开头 400 字符。
• updatedToolOutput 必须保持原工具的输出结构,所以只替换 dict 里过长的字符串字段; list 一律不碰(里面可能是图片块)。结构不对会被拒(原文照旧,不报错)。
• 读落盘文件本身时必须放行 —— 否则"用 Read 读它"是句空话:读回来的全文又被落盘,无限循环。 实测撞到过,模型连试五种写法绕。

index_guard()¶

def index_guard(workbench: Workbench) -> HookMatcher

往工作台写了东西就刷新 INDEX.md,下一个 agent 开局就知道它存在。

挂在 PostToolUse,matcher "Write|Edit"。tool_input["file_path"] resolve 后落在 workbench.root 内就调 workbench.refresh()。永远返回 {} —— 它不改任何东西,只有副作用。

isolate_guard()¶

def isolate_guard(agents: dict[str, AgentDefinition], *, on_inject: Any = None) -> HookMatcher

按角色给 subagent 分配独立 git worktree,实现隔离。

挂在 PreToolUse,matcher "Agent"。三个条件同时成立才注入:tool_name == "Agent"、 tool_input 既没有 cwd 也没有 isolation、subagent_type 对应的角色被 isolated() 标记过。 成立就返回 permissionDecision: "allow" + updatedInput(把 isolation 设成 "worktree")。

isolation 与 cwd 在 Agent 工具里互斥 —— 模型自己指定了 cwd 就尊重它。 "要不要隔离"是角色的属性,不是全局开关,也不是每次派活都判断;不需要隔离的角色一个字节都不会被加上。

开隔离就要把工作台挪出仓库。 被隔离的 agent 写不进共享 checkout, 所以工作台必须用 home= 指到仓库外。starter_flow(isolate=True) 用 <ws>.parent/.flower-<ws.name>,Runtime(workbench=True) 用 <run_dir>/workbench —— 两个都在仓库外,但不是同一个目录,别混着用。

isolated() / wants_isolation()¶

def isolated(agent: AgentDefinition, flag: bool = True) -> AgentDefinition
def wants_isolation(agent: AgentDefinition | None) -> bool

给一个 subagent 定义打上"需要独立工作区"的标记,以及读回这个标记。

标记走 object.__setattr__ 打的 Python 侧属性 _flower_isolate,不是 dataclass 字段 —— SDK 用 asdict() 序列化,只认已声明字段,所以这个标记不会漏到 CLI 那边(已实测)。

代价:对 AgentDefinition 做 dataclasses.replace() 会丢掉这个标记,隔离静默失效。

worker(isolate=True) 内部就是走 isolated()。

workbench_hooks()¶

def workbench_hooks(
    workbench: Workbench,
    *,
    delegate_only: bool = True,
    spill_threshold: int | None = 4000,
    agents: dict[str, AgentDefinition] | None = None,
    allow_glance: bool = False,
) -> dict[str, list[HookMatcher]]

一次装好工作台需要的那几个 hook。Runtime._attempt 调的就是它。

产出:

• PreToolUse:delegate_only=True → [delegate_guard(allow_glance=allow_glance)]; 有被标记的角色 → 追加 isolate_guard(agents)。
• PostToolUse:总是 [index_guard(workbench)];spill_threshold 为真 → 追加 spill_guard(workbench, threshold=spill_threshold)。
• 空列表的事件键会被剔掉,不返回空 list。

merge_hooks()¶

def merge_hooks(*groups: dict[str, list[Any]] | None) -> dict[str, list[Any]]

按事件名把多组 hook 配置拼接。

用 extend,不去重 —— 同一个 guard 传两次就会装两次。Runtime 用它把 spec.hooks、 workbench_hooks(...) 和 whitelist_guard 合到一起。

工作台¶

源码:flower/core/workbench.py

Workbench¶

@dataclass
class Workbench:
    workspace: Path
    dirname: str = ".flower"
    max_index_entries: int = 40
    home: Path | None = None

落盘的工作目录,三个子目录 + 一份索引。索引注入到 system prompt,所以 agent 每轮都知道 手上有什么。

脚本自述格式:首 8 行内的 # desc: 一句话(也认 // 和 -- 注释符), 退化到第一行非空注释或 docstring 首行(截 100 字)。

prompt_block() 注入的三条规矩:

1. 要跑第二次的脚本写进 scripts/,首行 # desc:。
2. 超过 2000 字符的产出写进 artifacts/,对话里只给路径和结论。
3. 关键决策写进 notes/,一个决策一个文件。

external=True 时 prompt_block() 会额外插一句"用绝对路径访问它"。

索引 subagent 继承不到。 它走的是会话级的 system_prompt.append,而 subagent 有自己的 system prompt(实测 $0.2461)。所以"长产出写 artifacts/、工作台在哪"这两条必须由 协调者在任务书里转述 —— 那是唯一通道,不是冗余。 WORKER_RULES 里故意没写这条:真实路径由 Workbench 生成,写死会错。

会话存储¶

源码:sqlite.py · trim.py · prune.py

三层继承:SqliteSessionStore ← TrimmingSessionStore ← PruningSessionStore。 Runtime 永远用最外层,三层的策略由构造参数控制。

三层各管一件事:落盘、按体积和价值裁剪、按"是不是错误"剪除。 裁剪和剪除都发生在 load()(即 resume 把历史喂回模型)那一刻,SQLite 里的原始记录一个字节不动。

SqliteSessionStore¶

class SqliteSessionStore(SessionStore):
    def __init__(self, path: str | Path) -> None

实现 SDK 的 SessionStore 协议,三张表 entries / meta / summaries。 store key 是 project_key/session_id[/subpath] —— 子 agent 的 transcript 用 subpath 区分。

内部的 _next_mtime 保证严格单调 —— list_sessions 和 summary sidecar 共用这个时钟, 否则 SDK 的 staleness 快路径会误判。

TrimPolicy¶

@dataclass
class TrimPolicy:
    keep_recent: int = 20
    min_chars: int = 2000
    spill_dirname: str = ".flower/spill"
    enabled: bool = True

两个 spill 目录不是同一个。 spill_guard 落在 <workbench.root>/spill/(可以在工作区外); TrimPolicy.spill_dirname 落在 <workspace>/.flower/spill/(必须在工作区内)。 两者分别对应"当场剪"和"resume 时剪",目录不同是有意的,别改成一个。

EphemeralPolicy¶

@dataclass
class EphemeralPolicy:
    enabled: bool = True
    keep_recent: int = 6
    max_chars: int = 2000
    text: str = "[{cmd} 的结果已过期(第 {age} 轮前),当前状态可能已变。需要请重新执行]"

一次性命令的结果过期策略。

只作用于 Bash 工具的结果,且命令要匹配一次性命令白名单。Read 不在此列 —— 文件内容不会因时间流逝而失真到误导的程度。过期内容不落盘,直接扔。

is_ephemeral()¶

def is_ephemeral(cmd: str) -> bool

判一条 Bash 命令是不是一次性命令。 delegate_guard 的放行判断和裁剪的过期判断共用这一个函数 —— 协调者能自己跑的命令集合 必须等于结果会被标记过期的集合。放行不裁剪,过期的 git status 会永久占上下文还会误导; 裁剪不放行,协调者为一条 ls 派个 subagent,4.3k 启动成本换几十字符。

判定顺序:

1. 空 / 全空白 → False。
2. 命令替换($(、反引号、<(、>()或"会改状态"的写法命中 → False。
3. 摘掉安全重定向(2>&1、&> /dev/null 之类)后仍含 > 或 < → False。
4. 摘掉 && / || / ; / | 后仍剩单个 &(后台执行)→ False。
5. 按 && / || / ; / | 拆开,每一段都必须命中白名单。

白名单动词大类:只读的 git 子命令(status diff log show branch rev-parse 等)、 目录与系统信息(ls pwd df du date whoami env 等)、进程与容器 (ps top lsof docker ps kubectl get 等)、看文件(cat head tail wc stat find tree)、 查路径(which whereis command -v type)、文本处理(grep rg sort uniq awk sed jq diff 等)。

即使动词在白名单,这些写法也会被拦:xargs、exec、eval、source、tee、 find -delete / -ok / -fprint、sed -i、sort -o、awk 里的 system( 和 print >、 git branch -D/-d/-m、git * --force/--hard/--prune。

第一版一刀切拒绝所有复合命令,实测让 glance 完全失效(协调者三次尝试全被拦), 所以改成了逐段判。

TrimmingSessionStore¶

class TrimmingSessionStore(SqliteSessionStore):
    def __init__(
        self,
        path: str | Path,
        workspace: str | Path,
        policy: TrimPolicy | None = None,
        ephemeral: EphemeralPolicy | None = None,
    ) -> None

公开属性:workspace、policy、ephemeral、last_report: dict[str, int]。

load() 的顺序:super().load() → 清空 last_report → ephemeral.enabled 则 expire() → policy.enabled 则 trim()。enabled=False 时该步整个跳过。

只裁纯文本:遇到 image / document 块原样留着。

两条结构性红线:tool_result 块本身必须在,只能换 content(少一个就是 "Missing Tool Result Block");isCompactSummary 条目不能动。