API de Python¶

Esta página agota los 62 símbolos públicos del __all__ de nivel superior de flower: firmas, parámetros, valores por defecto, semántica, atributos y métodos públicos. Al terminarla ya no hace falta abrir el código fuente para buscar un parámetro.

La organización es por aquello que te importa, no por archivo de módulo — si quieres saber "cómo impedir que el coordinador haga el trabajo por su cuenta", ve a la capa de hooks; si quieres saber "cómo se pasa el resultado de un paso al siguiente", ve a workflow. La terminología sigue en todo momento el glosario.

Versión 0.1.0, depende de claude-agent-sdk>=0.2.152. Todas las firmas se corresponden literalmente con el código fuente.

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

Qué hay en esta página¶

Seis valores por defecto que muerden¶

Estas seis líneas no son detalles marginales: son los seis vuelcos más frecuentes. Cada una está explicada por completo en su sección correspondiente.

Runtime¶

Código fuente: flower/core/runtime.py

Runtime es el núcleo de ejecución. Mantiene el espacio de trabajo, el session store, el workbench, la política de resiliencia y la política de handoff, y hacia fuera expone un único verbo: run de un paso. Reintentos, continuación tras una interrupción, handoff cuando el contexto se llena: todo ocurre dentro de esa única llamada.

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,
)

Los parámetros del constructor son todos keyword-only (* al principio), y workspace es obligatorio.

El session store está fijado en el código: siempre es PruningSessionStore(run_dir/"sessions.db", workspace=..., policy=<TrimPolicy>, ephemeral=<EphemeralPolicy>, prune=PrunePolicy(keep_denials=...)). Los parámetros del constructor no ofrecen ninguna vía para cambiar el backend — si lo necesitas, construye tú mismo AgentSpec + build_options(session_store=...), o sobrescribe rt.store después de construir.

Los dos últimos pasos de la construcción son load_dotenv() y check_credentials(), y si el segundo falla, lanza RuntimeError. Sin credenciales revienta en la fase de construcción, no al llegar a run().

Atributos públicos¶

Constantes de clase: INTERRUPTED = "interrupted-by-human", HANDOFF_DUE = "context-full-handoff", INTERRUPT_NOTE (un fragmento que se añade tras las palabras de la persona al continuar después de una interrupción, explicando que "las llamadas a herramienta en vuelo que devuelven interrupted son un efecto secundario normal de la interrupción, no un fallo del entorno").

Métodos públicos¶

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

Al comienzo de cada paso se pone a cero el nivel de contexto (self._ctx, self._warned = 0, False). Luego viene un bucle con cuatro salidas:

1. Éxito → se sale.
2. Interrupción humana (result.error == INTERRUPTED) → no está sujeta a max_attempts, no espera a la red. Con las palabras de la persona hace resume de la misma sesión, attempt -= 1 (una interrupción no cuenta como intento fallido), y el prompt = palabras de la persona + INTERRUPT_NOTE. Si no se obtuvo session_id, no queda más remedio que parar.
3. Contexto lleno (result.error == HANDOFF_DUE, o bien handoff.enabled y se obtuvo session_id y se cumple is_overflow(...)) → tampoco está sujeta a max_attempts. Primero comprueba len(result.retired) >= handoff.max_generations; si se excede, sustituye el error por un diagnóstico y sale; si no, escribe el documento de handoff → resume=None, fork=False (sesión completamente nueva) → el prompt pasa a ser h.prompt_block() → nivel de contexto a cero → attempt -= 1.
4. Fallo reintentable → si not resilience.enabled or attempt >= max_attempts, sale; si classify(error) determina que no debe reintentarse, también sale; si no, emite Event("retry"), se queda esperando la red con wait_online(), y sleep(delay_for(attempt)); si en algún momento hubo session_id, hace resume (el prompt pasa a ser resilience.resume_prompt) y pone result.resumed a True.

Cierre: escribe ended_at, lo añade a self.results, y escribe manifest.json.

manifest.json tiene semántica de append: cada escritura relee el disco y deduplica por el campo run (renueva sus propias filas, conserva las de los demás), así que ejecutar dos flower en paralelo bajo el mismo run_dir es seguro — siempre que los run_id no colisionen.

Los tres puntos de observación del handoff (todos Event("handoff"), distinguidos por payload["phase"]): near (se acerca a warn_at, se emite una sola vez por generación), writing (está escribiendo el handoff, tarda una decena de segundos), done (el payload trae degraded / path / sections). El turno que escribe el handoff se ejecuta con replace(spec, max_budget_usd=None) — el handoff tiene que poder escribirse, no puede quedarse atascado en el presupuesto; y con on_event=None, ese turno no se refresca en la UI.

El handoff se escribe en <workbench.notes>/交接-<步骤名>.md; sin workbench no hay escritura a disco, el documento se entrega igualmente a quien releva por medio del prompt, solo que después no se puede consultar. Los handoffs antiguos se mueven a 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

Toda la contabilidad de un paso terminado.

Flujo de trabajo¶

Código fuente: flower/workflow/

Un flujo de trabajo es un conjunto de pasos encadenados en orden, más cómo se pasa el estado entre pasos y cuándo se sale antes de tiempo. El framework no trae flujos hechos: el flujo lo escribes tú —— starter_flow no es más que una plantilla que funciona.

Alias de tipo Ctx = dict[str, Any] (flower.workflow.base.Ctx, está en flower.workflow.__all__, no en el __all__ de nivel superior).

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

La declaración de un paso. Step no es una función —— lo que de verdad ejecuta es Runtime.run(step.spec, prompt, ...). Los tres primeros campos son posicionales: Step("取词", terse, "读 seed.txt …") es una escritura válida.

Tres formas de enganchar sesiones (dentro de una misma ejecución):

on_reject cambia la semántica del reintento:

• Si no lo das → el siguiente intento se ejecuta desde cero (mismo prompt, mismo resume_from).
• Si lo das → el siguiente intento continúa la sesión que acaba de ser rechazada, con el prompt sustituido por su valor de retorno y fork forzado a False.
• Si devuelve cadena vacía → no hay devolución, degrada a ejecutar desde cero.
• Si result.session_id es None → también degrada a ejecutar desde cero.

Los tres valores de on_fail:

Pase o no pase, siempre se escribe ctx["_results"][name] = result; si result.session_id no está vacío, además se escribe en ctx["_sessions"] y se llama a 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

Ejecuta una serie de Step en orden y devuelve el ctx final. steps es posicional, Workflow([...]) es válido.

Claves privadas que run() escribe en el ctx (todas empiezan por _, así no chocan con nombres de paso):

Payload de Event("step"): {"index": i, "total": len(steps), "resumed": bool, "woke": int}.

Etiquetas de reintento: el intento 0 usa step.name; después, si hay on_reject usa f"{name}#round{attempt+1}", y si no f"{name}#retry{attempt}". En el manifiesto se ve de un vistazo cómo terminó ese paso. Los nombres con sufijo no entran en el linaje entre procesos —— Lineage.remember usa el nombre original.

runtime.on_session solo cubre la línea de runtime.run; un try/finally garantiza que se desmonta siempre antes del gate. prompt_cur / resume_cur / fork_cur son variables locales y no se escriben de vuelta en step —— el mismo objeto Step puede ejecutarse una segunda vez.

StepAbort¶

class StepAbort(Exception): ...

Lanzarlo desde gate = parar ya, no reintentar más. La diferencia con devolver False: False es «esta vez no, otra ronda»; StepAbort es «otra ronda tampoco sirve».

Al lanzarlo: ctx["_aborted"] = str(exc), passed = False, se sale del bucle de reintentos (sin consumir los retries restantes), y después sigue el camino de fallo normal según on_fail (por defecto "stop").

with_goal lo lanza en dos sitios: cuando no consigue ctx["_runtime"], y cuando el veredicto es unreachable y nadie responde.

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

Produce un Step que hace la clarificación previa: preguntar hasta dejar claro el requisito → parsearlo a Brief → si las cuatro secciones están completas, congelarlo y volcarlo a disco.

Los campos del Step resultante quedan así:

• resume_prompt = CLARIFY_RESUME.
• when: con always_ask=True → siempre True; si no, si Brief.load(brief_path) está completo lo inyecta en el ctx y luego devuelve False (salta el paso) —— hay que inyectarlo aunque se salte, si no aguas abajo no hay requisito.
• gate: Brief.parse(result.text); si está incompleto, escribe ctx[MISSING_KEY] y devuelve False; si está completo, congela con b.write(brief_path), lo inyecta en el ctx y devuelve True.
• reduce: devuelve ctx[BRIEF_KEY].prompt_block(), no el texto original del modelo —— en el original puede haber colado cosas de más.
• resume_from se queda en None por defecto: el paso siguiente es sesión nueva, recibe solo el brief, no ese intercambio de preguntas y respuestas. Las preguntas y respuestas de la clarificación previa nunca entraron en el contexto del coordinador; no es que entrasen y luego se recortaran.

Los tres sitios donde se inyecta en el 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

Produce un Step que fija el objetivo: el juez lee el brief y escribe el objetivo + la lista de comprobaciones, que se parsea a Goal y se congela en disco. Tiene la misma forma que clarify_step.

No hay parámetro can_run —— si quieres que el juez que fija el objetivo pueda ejecutar comandos, hay que pasar can_run=True vía **spec_kw. Si no lo pasas, no tendrá Bash y la regla de JUDGE_RULES que dice «primero mira bien en qué entorno estás» no se puede cumplir.

Además de parsear y congelar, el gate hace una cosa más: si el objetivo tiene entradas [此环境无法验证:…], en ese mismo momento emite un aviso vía ctx["_on_event"] con Event("task", payload={"unverifiable", "total", "path"}) —— el destino de esas entradas queda sellado justo al fijar el objetivo; cuando llega el veredicto ya te has gastado el dinero de una ronda entera de trabajo.

No se define resume_prompt —— fijar el objetivo debe reenviar el brief completo de todas formas.

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

Envuelve un Step existente con el guardián de objetivos: al final de cada ronda, un juez independiente dicta veredicto, y si no se ha logrado el objetivo lo devuelve para seguir trabajando.

El resultado es replace(step, retries=max(0, rounds - 1), gate=<nuevo gate>, on_reject=<nuevo on_reject>) —— con dataclasses.replace en vez de reconstruir campo a campo; una vez se reconstruyó y se olvidó resume_prompt, y no dio ningún error: simplemente volvía a mandar el brief entero al continuar.

El gate es async, y su recorrido es:

1. Falta ctx["_runtime"] → lanza StepAbort («no hay Runtime, no se puede dictaminar el objetivo»). No finjas que ha pasado.
2. ctx[ROUND_KEY] += 1.
3. Obtener el objetivo: primero un Goal completo en ctx[GOAL_KEY]; si no, Goal.load(goal_path); si no, un Goal() vacío.
4. await rt.run(judger, VERIFY_PROMPT..., step_name=f"{label}#{轮次}", on_event=...). El juez es un Runtime.run independiente, con resume siempre None —— siempre sesión nueva; el step_name lleva el número de ronda, así que no entra en el linaje entre procesos.
5. Verdict.parse(vr.text) se escribe en ctx[VERDICT_KEY].
6. v.achieved → devuelve True.
7. Si no es unreachable (incluidos los casos ambiguos con v.ok=False) → en caso ambiguo añade una razón por defecto y devuelve False. Toda ambigüedad cuenta como no logrado —— un «parece que está bien» no puede dar el trabajo por cerrado.
8. unreachable → await channel.ask(...) pregunta a la persona, con tres opciones:
9. Nadie responde (a.state != "answered") → lanza StepAbort. Seguir girando en vacío es la opción más cara.
10. «Acepta este resultado y sigue adelante» → devuelve True.
11. «Modificar el objetivo» → vuelve a preguntar por el objetivo nuevo, g.amend(...).write(goal_path), actualiza ctx[GOAL_KEY] y devuelve False.
12. El resto (incluida una respuesta libre escrita por la persona) → se toma como «te has equivocado en el veredicto», se anota lo que dijo en v.reason y devuelve False.

El on_reject es síncrono: devuelve ctx[VERDICT_KEY].feedback(), y si no hay Verdict devuelve "" (degrada a ejecutar desde cero).

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

Monta un flujo de tres pasos que funciona de fábrica: clarificar requisitos → fijar objetivo → trabajar (con guardián de objetivos). Es lo que usa el comando flower.

Montaje fijo:

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)

Ramas de comportamiento:

• isolate=True y el workspace no es un repositorio git → lanza ValueError, en vez de esperar a que la herramienta Agent dé error (para entonces ya te has gastado el dinero).
• Detección de despertar: si Brief.load(brief_path) existe y está complete(), se cuenta como despertar. Si no es un despertar y ask está vacío → lanza ValueError("要给一句诉求,例如 flower '帮我做一个 X'").
• Al despertar, esa frase aterriza en tres sitios a la vez, y si falta uno el efecto se pierde en silencio: se anexa al brief (ch.amend(said, label="唤醒时追加"), si ya está en el fichero no se reescribe), hace que goal_step(always_set=True) rehaga la lista (si no se rehace, el juez sigue leyendo el objetivo viejo) y se le entrega directamente al coordinador (en su contexto está el objetivo viejo: si no se la das, trabaja con el criterio viejo y luego lo juzgan con el nuevo).

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

Sondeo de solo lectura antes de arrancar: no escribe ni un byte. Sirve para decirle a la persona, antes de empezar de verdad, si esto continúa lo anterior o empieza de cero.

El dict devuelto:

La ubicación del workbench se define una sola vez, aquí y en starter_flow: isolate=True → <ws>.parent/.flower-<ws.name> (fuera del repositorio); si no, <ws>/.flower. Cuando el driver quiere saber dónde está el brief también pasa por esta función —— si construyes la ruta a mano y te equivocas no salta ningún error, simplemente deja de funcionar en silencio.

Fábrica de roles¶

Código fuente: flower/core/roles.py

Los cinco roles son funciones fábrica. Cada rol = un bloque de reglas inyectado + un conjunto de herramientas + un conjunto de hooks. worker() produce un AgentDefinition del SDK (para despachar a un subagent); los otros cuatro producen un AgentSpec (abren su propia sesión).

Los roles en sí no montan hooks: interceptar herramientas lo hace Runtime._attempt, que los instala automáticamente según spec.delegate_only, ver capa de hooks.

Constantes internas de grupos de herramientas (no exportadas, pero determinan los valores por defecto):

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

Construye el coordinador que vive en el hilo principal: descompone la tarea, reparte trabajo, lee informes, decide, pero no toca nada. Los tres primeros parámetros son posicionales.

Tres cosas quedan fijas en el AgentSpec producido: delegate_only=True, agents=workers y workbench mantiene el valor por defecto True de AgentSpec.

Las herramientas web de workers se fusionan¶

Tras armar la lista, coordinator() recorre cada AgentDefinition.tools y, por cada herramienta que caiga dentro de WEB_TOOLS (WebFetch, WebSearch, roles.py:33), añade también una copia al allowed_tools del propio coordinador (roles.py:523-526).

Motivo: allowed_tools, igual que disallowed_tools, es de ámbito de sesión. Esta es la evidencia más dura de todo el documento sobre este punto: no afecta solo al hilo principal. Una herramienta que no esté en esa lista de ámbito de sesión también pasa por aprobación de permisos cuando la invoca un subagent; sin nadie vigilando no hay quien apruebe, y el harness responde Claude requested permissions to use X, but you haven't granted it yet (toolDenialKind=user-rejected), con el modelo reintentando la misma llamada una y otra vez. Comprobado a base de fallar: se le añadió WebFetch/WebSearch al ejecutor pero solo se escribió en AgentDefinition.tools, y aquella ejecución de novel acumuló más de veinte user-rejected sin escribir una sola palabra (roles.py:513-518).

Los dos campos tienen la misma naturaleza de ámbito de sesión, pero los síntomas son distintos: disallowed_tools falla en el acto, allowed_tools reintenta en silencio hasta morir. El segundo es más difícil de diagnosticar, porque en pantalla nada parece un error.

Solo se fusionan las de solo lectura, sin efectos secundarios. Write/Edit/Bash se dejan fuera a propósito: en cuanto el hilo principal no necesite aprobación para ellas, el muro de delegate_guard («el coordinador no toca nada») deja de tener sentido; y el Bash/Write de un subagent ya funciona de todos modos (462 permisos concedidos medidos, roles.py:520-522).

En el código fuente está escrito explícitamente que no se use disallowed_tools para implementar «solo coordinar, no tocar»: es de ámbito de sesión y prohibiría también el Bash/Write de los subagents, ver la advertencia en AgentSpec. La forma correcta es la de aquí: delegate_only=True + no conceder allowed_tools, y dejar que delegate_guard bloquee solo el hilo principal según agent_id.

Si se da channel, vienen las dos herramientas juntas, no es opcional: al montar el MCP server están ambas, y allowed_tools no es exclusivo, así que se pueden invocar estén listadas o no. Sin supervisión humana, cada ask se queda bloqueado hasta agotar timeout_s: para esos casos, usar 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

Construye la definición del subagent que realmente trabaja. Los dos primeros parámetros son posicionales. Devuelve un AgentDefinition del SDK, que se mete directamente en coordinator(workers={...}).

isolate=True exige que el workspace sea un repositorio git; si no, la herramienta Agent falla directamente con "not in a git repository", no degrada en silencio. Y la marca es un atributo de Python: hacer dataclasses.replace() sobre el AgentDefinition la pierde, y el aislamiento deja de funcionar sin avisar.

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

Construye el clarificador: antes de tocar nada, deja claros los requisitos; no hace trabajo, solo pregunta, y al final produce exactamente cuatro secciones.

El AgentSpec producido: allowed_tools = [channel.tool_name] + (esas cinco si puede leer), mcp_servers = channel.mcp_servers(), workbench=False (no tiene herramientas de escritura, el índice no le sirve de nada), y permission_mode hereda el "default" por defecto de AgentSpec. No tiene Write / Edit / Bash / Agent, ni tampoco inbox (a diferencia del coordinador).

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

Construye el juez: o bien fija el objetivo antes de arrancar, o bien emite el veredicto de cada ronda al terminarla.

El AgentSpec producido: allowed_tools = [channel.tool_name, "Read", "Glob", "Grep"] + (si can_run) ["Bash"], workbench=False, el resto igual que clarify(). No tiene Write / Edit / Agent, ni tampoco inbox.

Compromiso: con can_run=True el veredicto es más sólido (puede ejecutar de verdad los comandos de aceptación), a cambio de que el juez pueda modificar el área de trabajo: Bash por sí solo ya permite escribir ficheros. Si quieres un veredicto absolutamente neutral, no lo actives.

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

Construye el oráculo: con la ejecución todavía en marcha, le preguntas «¿por dónde va esto?» y responde tras echar un vistazo a los eventos recientes y al banco de trabajo. Lo que dice no entra en el contexto de esa ejecución.

El AgentSpec producido: allowed_tools = ["Read", "Glob", "Grep"] (sin canal: no pregunta, solo responde), workbench=True (el único de los cinco roles que no es coordinador y aun así abre el banco de trabajo: precisamente necesita leer esos artefactos y notas).

Los cinco bloques de reglas¶

Las cinco constantes están en __all__, se pueden importar directamente para leerlas, concatenarlas o modificarlas.

Definición de agente¶

Código fuente: flower/core/agent.py

AgentSpec es la declaración completa de un agente especializado, y build_options la compila al ClaudeAgentOptions del SDK. Lo que produce la fábrica de roles es precisamente un AgentSpec: cuando necesites una combinación fuera de la fábrica, constrúyelo directamente.

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

Compila un AgentSpec al ClaudeAgentOptions del SDK. Es lo que llama internamente Runtime._attempt; si conduces el SDK por tu cuenta (sin Runtime), también entras por aquí.

Correspondencias:

PLUGIN_DIR es el directorio plugin/ de la raíz del repositorio (tres niveles por encima de flower/core/agent.py). Tras instalar con pip ese directorio puede no existir, y el código lo comprueba con is_dir().

CompactPolicy¶

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

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

El panel de interruptores del auto-compact; lo que produce es un conjunto de variables de entorno a inyectar en el subproceso. El algoritmo de compact en sí vive en el binario del harness y no se puede cambiar; lo único que se puede cambiar es «si se dispara o no».

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)

El objeto de política para, cuando el contexto está a punto de llenarse, «escribir el documento de relevo y cambiar de sesión» en lugar de hacer compact.

default_window()¶

def default_window() -> int

Adivina la ventana de contexto a partir de la cadena del nombre del modelo en las variables de entorno ANTHROPIC_MODEL o ANTHROPIC_DEFAULT_OPUS_MODEL:

Por defecto tira por lo optimista. Pasarse de grande no es un error duro: la API rechaza con prompt is too long, y Runtime reconoce esa señal (el is_overflow interno) y hace relevo en el acto, pero el relevo de esa generación es una versión degradada.

Documentos¶

Código fuente: brief.py · handoff.py · goal.py

Cuatro dataclasses, todas dedicadas a «parsear una respuesta del modelo en unas secciones fijas y volcarla a disco». Forma común: parse() parsea, missing() / complete() comprueban si está completo, to_markdown() es para humanos, prompt_block() es para el modelo de aguas abajo, write() / load() vuelcan y releen.

Brief¶

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

El brief, exactamente cuatro secciones, en orden fijo goal → accept → bounds → unknowns; los nombres de sección en chino son 「目标」「验收标准」「边界」「未知与假设」.

Reglas de parseo (donde se concentran los fallos):

• Al quitar las cercas, **si encuentra un `` o_~sin cerrar, descarta desde ahí todo el resto**: se ha comprobado que el clarificador pega código entero en la respuesta. Cuando la salida del modelo se trunca, ninguna de las secciones posteriores se parsea, con lo quecomplete()esFalse` y el gate lo devuelve para repetir.
• La regex de títulos tolera ## 目标 / **目标** / 目标: / 3. 边界, y también que el cuerpo venga inmediatamente tras el título.
• La tabla de alias se compila en orden inverso de longitud; si no, «未知» se comería antes a «未知与假设».
• Si una sección con el mismo nombre aparece repetida, se toma la primera con contenido.
• Si al editar el brief a mano se copia literalmente el marcador "(未填)" de to_markdown(), esa sección sigue contando como ausente.

Handoff¶

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

El documento de relevo que se escribe al hacer relevo, cinco secciones.

Solo son obligatorias doing y next: exigir por narices que «caminos que no llevan a nada» no esté vacío obligaría al modelo a inventar.

En el mismo módulo hay tres miembros no exportados pero semánticamente clave: is_overflow(*texts) casa con prompt is too long, context length exceeded, maximum context length, too many total text bytes, input length and max_tokens exceed, etc., y convierte un «error duro» en «relevo en el acto»; HANDOFF_PROMPT es el prompt que hace que la propia sesión actual escriba su relevo (contiene los marcadores {used} y {window}; no es un rol nuevo: solo ella tiene ese contexto); degraded(step, prompt, *, why="") monta mecánicamente un relevo cuando no se ha logrado escribir, metiendo en scene los primeros 1200 caracteres de la tarea original.

Goal¶

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

El objetivo y la lista de comprobación del guardián de objetivo.

Verdict¶

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

El resultado de una ronda de veredicto del juez, exactamente tres secciones: conclusión / motivo / no superados.

Orden de reconocimiento de parse:

1. Primero busca por sección con título 「结论」/「判定」.
2. Si no hay sección con título, tras hacer strip al bloque entero: fullmatch(r"1|true") → alcanzado; fullmatch(r"0|false") → aún no.
3. Si no, busca en el texto de la conclusión la primera coincidencia de la tabla de palabras de estado (las palabras largas primero). 「无法验证 / 没法验证 / 验证不了 / 无法判定 / unverifiable」 caen todas en unreachable: comprobado a base de fallar: la plataforma objetivo era macOS, se ejecutaba en un contenedor Linux, y el juez miró la rama del código fuente y dio el visto bueno.
4. Si aún no hay nada → busca un \b1\b aislado → alcanzado; \b0\b → aún no.
5. Si nada casa → state="", ok=False.

unreachable y not_yet son dos conclusiones distintas: la primera lleva por el camino de «parar y preguntar a una persona», no por el de «otra ronda».

Capa de hooks¶

Código fuente: flower/core/guard.py

Esta capa es el límite de ejecución de flower: qué herramientas no puede tocar el hilo principal, cómo se recortan los resultados demasiado largos, qué rol va a un worktree independiente — todo lo impone el hook del SDK, no el prompt. La razón es directa: el prompt es una sugerencia, el modelo puede ignorarlo; se ha medido que aunque el system prompt diga explícitamente «no uses worktree», la inyección de isolate_guard surte efecto igual (el modelo pasa None, lo que aterriza es 'worktree').

Nueve exportaciones: cinco fábricas de guard que devuelven HookMatcher (whitelist_guard puede devolver None), un ensamblador, un combinador y dos funciones de marcado de aislamiento. No hace falta engancharlos a mano — Runtime los monta automáticamente según el AgentSpec. Engancharlos a mano solo hace falta si conduces el SDK tú mismo (sin pasar por Runtime).

La decisión de si es el hilo principal pasa siempre por una sola función: _is_main_thread(data) = not data.get("agent_id") — los datos del hook de tool-lifecycle de un subagent llevan agent_id, los del hilo principal no. Todos los guards que «solo bloquean el hilo principal» se apoyan en esta línea.

Constantes de grupos de herramientas (a nivel de módulo, no exportadas, pero determinan los matchers por defecto):

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

Tabla rápida: qué guard va en qué evento del SDK¶

La conclusión clave que se lee en esta tabla: con Runtime(workbench=False) no se monta nada de workbench_hooks; y para un coordinador con delegate_only=True también se salta whitelist_guard — el hilo principal no tiene ni un muro. Ver el aviso en Runtime.

whitelist_guard()¶

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

Hace que allowed_tools sea realmente excluyente para esas cuatro herramientas que actúan.

• Va en PreToolUse, el matcher es "|".join(banned), donde banned = las de Bash Write Edit NotebookEdit que no estén en allowed.
• Si acierta, permissionDecision: "deny", con un texto que viene a decir: «XX no tiene YY. Es intencionado, no es que falte configuración. Escribe la conclusión en el cuerpo de tu respuesta, el framework la recogerá de ahí — no intentes rodearlo con otra forma de escribirlo.»
• Solo bloquea el hilo principal de esta sesión, los subagents pasan — las herramientas de un subagent las decide AgentDefinition.tools.
• Si no hay nada que bloquear devuelve None (por ejemplo un rol como worker() con el juego completo de herramientas), y quien lo llama decide si lo monta o no.

Por qué tiene que existir: allowed_tools es una lista de exención de aprobación, no una whitelist excluyente. Dos pruebas medidas: el juez que fijaba objetivos ejecutó Bash 11 veces; en la sonda de $0.1, un agent con allowed_tools=["Read"] seguía pudiendo llamar a Write/Bash. Por eso el «no tiene herramientas de escritura» de clarify() / judge() se apoya en este hook, no en la whitelist en sí.

La ventaja es que se deriva de allowed_tools, así que judge(can_run=True) conserva Bash automáticamente y sigue bloqueando Write/Edit — sin necesidad de un interruptor adicional.

delegate_guard()¶

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

El hilo principal actúa por su cuenta → se rechaza, y se le indica el camino.

• Va en PreToolUse, el matcher es directamente tools.
• El hilo principal llama a una de esas cuatro herramientas → deny, y el motivo indica el siguiente paso: usa la herramienta Agent para despachar un subagent, escribe en el encargo el objetivo y los criterios de aceptación, y exígele que escriba los productos largos en .flower/artifacts/ y que en la respuesta solo dé la ruta y la conclusión.
• Los subagents pasan siempre.

La diferencia con whitelist_guard está en la formulación: ambos bloquean el mismo conjunto de herramientas, pero este dice «despacha a alguien», que es lo pertinente. Por eso un rol con delegate_only=True monta solo este; montar los dos haría que el modelo recibiera dos indicaciones contradictorias.

El criterio de paso de allow_glance=True y el de «¿se va a recortar este resultado?» son la misma función (is_ephemeral): el conjunto que se deja pasar debe ser igual al conjunto que caduca; si cambias uno, tienes que cambiar el otro.

spill_guard()¶

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

Los resultados de herramienta que superen el umbral se vuelcan a disco en el acto, dejando en el contexto solo una línea de puntero — no se espera a que el contexto esté lleno para comprimir a posteriori.

• Va en PostToolUse, devuelve {"hookSpecificOutput": {"hookEventName": "PostToolUse", "updatedToolOutput": <lo recortado>}}.
• El nombre del fichero volcado son los primeros 16 dígitos del sha256 del contenido + .txt, y en el contexto se sustituye por una línea de puntero + los primeros 400 caracteres.
• updatedToolOutput debe conservar la estructura de salida de la herramienta original, así que solo se sustituyen los campos string demasiado largos del dict; las list no se tocan nunca (pueden contener bloques de imagen). Si la estructura no cuadra, se rechaza (el original queda tal cual, sin error).
• Al leer el propio fichero volcado hay que dejar pasar — si no, «léelo con Read» es una frase vacía: el texto completo que se lee vuelve a volcarse, bucle infinito. Se topó con esto en pruebas reales: el modelo intentó cinco formas seguidas de rodearlo.

index_guard()¶

def index_guard(workbench: Workbench) -> HookMatcher

Si se escribe algo en el workbench, refresca INDEX.md, y el siguiente agent sabe desde el arranque que eso existe.

Va en PostToolUse, matcher "Write|Edit". Si tool_input["file_path"] tras resolverse cae dentro de workbench.root, llama a workbench.refresh(). Siempre devuelve {} — no modifica nada, solo tiene efectos secundarios.

isolate_guard()¶

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

Asigna un git worktree independiente al subagent según su rol, implementando el aislamiento.

Va en PreToolUse, matcher "Agent". Solo se inyecta si se cumplen tres condiciones a la vez: tool_name == "Agent", tool_input no tiene ni cwd ni isolation, y el rol correspondiente al subagent_type está marcado con isolated(). Si se cumplen, devuelve permissionDecision: "allow" + updatedInput (poniendo isolation a "worktree").

isolation y cwd son mutuamente excluyentes en la herramienta Agent — si el modelo ha especificado cwd por su cuenta, se respeta. «Aislar o no» es una propiedad del rol, no un interruptor global ni algo que se decida en cada despacho; a un rol que no necesita aislamiento no se le añade ni un byte.

Si activas el aislamiento tienes que sacar el workbench del repositorio. Un agent aislado no puede escribir en el checkout compartido, así que el workbench debe apuntar fuera del repositorio con home=. starter_flow(isolate=True) usa <ws>.parent/.flower-<ws.name>, y Runtime(workbench=True) usa <run_dir>/workbench — ambos están fuera del repositorio, pero no son el mismo directorio, no los mezcles.

isolated() / wants_isolation()¶

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

Marca una definición de subagent como «necesita espacio de trabajo propio», y lee esa marca de vuelta.

La marca es un atributo del lado Python, _flower_isolate, puesto con object.__setattr__, no un campo del dataclass — el SDK serializa con asdict() y solo reconoce los campos declarados, así que esta marca no se filtra al CLI (comprobado).

El precio: hacer dataclasses.replace() sobre un AgentDefinition pierde esta marca, y el aislamiento deja de funcionar en silencio.

worker(isolate=True) internamente pasa por 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]]

Monta de una vez los hooks que necesita el workbench. Es lo que llama Runtime._attempt.

Resultado:

• PreToolUse: delegate_only=True → [delegate_guard(allow_glance=allow_glance)]; si hay roles marcados → se añade isolate_guard(agents).
• PostToolUse: siempre [index_guard(workbench)]; si spill_threshold es verdadero → se añade spill_guard(workbench, threshold=spill_threshold).
• Las claves de evento con lista vacía se eliminan, no se devuelven listas vacías.

merge_hooks()¶

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

Concatena varios grupos de configuración de hooks por nombre de evento.

Usa extend, no deduplica — pasar el mismo guard dos veces lo monta dos veces. Runtime lo usa para combinar spec.hooks, workbench_hooks(...) y whitelist_guard.

Workbench¶

Código fuente: flower/core/workbench.py

Workbench¶

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

El directorio de trabajo donde se vuelca todo: tres subdirectorios + un índice. El índice se inyecta en el system prompt, así que el agent sabe en cada turno qué tiene a mano.

Formato de autodescripción de los scripts: # desc: 一句话 dentro de las primeras 8 líneas (también acepta // y -- como marcadores de comentario), degradando al primer comentario no vacío o a la primera línea del docstring (truncada a 100 caracteres).

Las tres reglas que inyecta prompt_block():

1. Los scripts que se vayan a ejecutar más de una vez se escriben en scripts/, con # desc: en la primera línea.
2. Los productos de más de 2000 caracteres se escriben en artifacts/, y en la conversación solo se da la ruta y la conclusión.
3. Las decisiones clave se escriben en notes/, un fichero por decisión.

Con external=True, prompt_block() añade una frase extra: «accede a ello con ruta absoluta».

El índice no lo heredan los subagents. Va por el system_prompt.append a nivel de sesión, y un subagent tiene su propio system prompt (medido: $0.2461). Por eso esas dos cosas —«los productos largos van a artifacts/» y «dónde está el workbench»— tiene que transmitirlas el coordinador en el brief de tarea: ese es el único canal, no es redundancia. En WORKER_RULES no está escrito a propósito: la ruta real la genera Workbench, dejarla fija sería un error.

Almacén de sesiones¶

Código fuente: sqlite.py · trim.py · prune.py

Tres niveles de herencia: SqliteSessionStore ← TrimmingSessionStore ← PruningSessionStore. Runtime usa siempre el más externo; las políticas de los tres niveles se controlan por parámetros del constructor.

Cada nivel se ocupa de una cosa: volcar a disco, recortar por volumen y valor, y podar según «esto es un error o no». Tanto el recorte como la poda ocurren en el momento de load() (es decir, cuando el resume devuelve el historial al modelo); el registro original en SQLite no se toca ni un byte.

SqliteSessionStore¶

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

Implementa el protocolo SessionStore del SDK, con tres tablas: entries / meta / summaries. La clave del store es project_key/session_id[/subpath] — los transcripts de los subagents se distinguen por el subpath.

El _next_mtime interno garantiza monotonía estricta — list_sessions y el sidecar de summary comparten este reloj; si no, la ruta rápida de staleness del SDK se equivocaría.

TrimPolicy¶

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

Los dos directorios de spill no son el mismo. spill_guard vuelca en <workbench.root>/spill/ (puede estar fuera del espacio de trabajo); TrimPolicy.spill_dirname vuelca en <workspace>/.flower/spill/ (tiene que estar dentro del espacio de trabajo). Corresponden respectivamente a «recortar en el acto» y «recortar al hacer resume»; que sean directorios distintos es intencionado, no los unifiques.

EphemeralPolicy¶

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

Política de caducidad de los resultados de los comandos efímeros.

Solo actúa sobre los resultados de la herramienta Bash, y el comando debe coincidir con la whitelist de comandos efímeros. Read no entra aquí: el contenido de un fichero no se desvirtúa con el paso del tiempo hasta el punto de inducir a error. El contenido caducado no se vuelca, se tira directamente.

is_ephemeral()¶

def is_ephemeral(cmd: str) -> bool

Decide si un comando Bash es un comando efímero. La decisión de paso de delegate_guard y la de caducidad del recorte comparten esta misma función: el conjunto de comandos que el coordinador puede ejecutar por sí mismo tiene que ser igual al conjunto de resultados que se marcan como caducados. Dejar pasar sin recortar hace que un git status caducado ocupe contexto para siempre y además induzca a error; recortar sin dejar pasar hace que el coordinador despache un subagent para un ls, cambiando 4.3k de coste de arranque por unas decenas de caracteres.

Orden de decisión:

1. Vacío / todo espacios → False.
2. Sustitución de comandos ($(, comillas invertidas, <(, >() o cualquier forma «que cambia estado» → False.
3. Si tras quitar las redirecciones seguras (2>&1, &> /dev/null y similares) sigue habiendo > o < → False.
4. Si tras quitar && / || / ; / | queda un & suelto (ejecución en segundo plano) → False.
5. Se parte por && / || / ; / |, y cada tramo tiene que estar en la whitelist.

Grandes categorías de verbos en la whitelist: subcomandos de git de solo lectura (status, diff, log, show, branch, rev-parse, etc.), información de directorios y sistema (ls, pwd, df, du, date, whoami, env, etc.), procesos y contenedores (ps, top, lsof, docker ps, kubectl get, etc.), ver ficheros (cat, head, tail, wc, stat, find, tree), buscar rutas (which, whereis, command -v, type), procesamiento de texto (grep, rg, sort, uniq, awk, sed, jq, diff, etc.).

Aunque el verbo esté en la whitelist, estas formas se bloquean igual: xargs, exec, eval, source, tee, find -delete / -ok / -fprint, sed -i, sort -o, system( y print > dentro de awk, git branch -D/-d/-m, git * --force/--hard/--prune.

La primera versión rechazaba de un plumazo todos los comandos compuestos, y en pruebas reales dejó el glance completamente inútil (los tres intentos del coordinador fueron bloqueados), así que se cambió a evaluar tramo por tramo.

TrimmingSessionStore¶

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

Atributos públicos: workspace, policy, ephemeral, last_report: dict[str, int].

Orden de load(): super().load() → vaciar last_report → si ephemeral.enabled, expire() → si policy.enabled, trim(). Con enabled=False ese paso se salta entero.

Solo recorta texto plano: los bloques image / document se dejan tal cual.

Dos líneas rojas estructurales: el bloque tool_result tiene que seguir ahí, solo se puede cambiar el content (si falta uno, es «Missing Tool Result Block»); las entradas isCompactSummary no se tocan.

trim_report()¶