项目文件夹

文件
T
Asim Aslam 9fdcc24cce Implement durable execution and scoped state management for flows (#2972)
* docs: design note for flow steps + Checkpoint durable execution

* docs: fold in durable-execution decisions (State struct, single Step, run retention, retry)

* docs: rename State.Payload to State.Data

* feat(flow): ordered steps + Checkpoint durable execution

A flow can now be an ordered list of steps (a task with stages) instead
of a single LLM turn. State carries typed Data plus a Stage marker; each
step is checkpointed before and after via a pluggable Checkpoint
(store-backed by default), so a run survives a crash and resumes where it
stopped without re-running completed steps. Flow-level Retry with a
per-step override; runs retained for audit unless DeleteOnSuccess.

Step actions: Call (RPC), LLM (augmented turn), Dispatch (to an agent),
or any StepFunc. Single-step and agent-dispatch flows are unchanged.

* feat(flow): top-level re-exports + durable flow example

Expose the step/checkpoint API from the micro package (FlowSteps,
FlowStep, FlowState, FlowRetry, FlowWithCheckpoint, FlowCall/LLM/Dispatch,
Checkpoint, StoreCheckpoint) and add a runnable, key-free example
demonstrating crash + resume.

* docs: document durable flow steps (guide, README, CLI help)

* docs: blog post + changelog for durable workflows

* fix(flow): scope checkpoint keys by flow name (flow/{name}/runs/{id})

Run keys were flow/runs/{id} — a single global keyspace shared by every
flow on the default store. Namespace them by flow name so each flow's
state is kept apart. StoreCheckpoint now takes a scope argument (the flow
passes its name by default).

* feat(store): Scope handle; scope agent and flow state by name

Add store.Scope(s, database, table) — a store handle that confines every
operation to a database/table without mutating the shared store, so
co-located components don't clobber each other's table (the failure mode
of the global Init(Table(...)) approach).

Use it to keep each agent's memory and plan in its own table
(agent/{name}) and each flow's runs in its own (flow/{name}), instead of
one global table partitioned only by key prefix. Services already scope
by service name.

* feat: consistent state model — service store scoping, flow registry, list/history CLI

- service: scope store via store.Scope (database service / table name),
  retiring the Init(store.Table(name)) global-mutation hack; bridge the
  default store so handlers using store.DefaultStore stay isolated.
- flow: register in the registry as type=flow while running (with trigger
  and step count), deregister on Stop. Live discovery, like agents.
- cli: micro flow list (registry), micro flow runs <name> (durable store),
  micro agent history <name> (durable store). list = running, runs/history
  = durable, mirroring the service model.

* test: mini-universe end-to-end harness + scheduled GitHub Action

internal/harness/universe boots a small but real go-micro world — four
services, a durable checkout flow that crashes at payment and resumes,
and a guardrailed agent with a tool wrapper reached over RPC — drives the
scenario, asserts the end state (10 checks), and shuts down. Everything
is real except the LLM (mocked), so it's deterministic and needs no key;
-provider anthropic runs it live. Exits non-zero on failure, so it's an
end-to-end test, not just a demo.

Adds .github/workflows/universe.yml (push/PR/daily/dispatch) running the
universe + existing harnesses on the mock provider, plus an opt-in job
that runs live when ANTHROPIC_API_KEY is set. 'make harness' runs them
locally.

* ci: run the live universe job against AtlasCloud (ATLASCLOUD_API_KEY)

* ci: run the live universe job only on schedule or manual dispatch

The deterministic mock job still runs on push/PR/daily; the live
(AtlasCloud) job runs daily and on manual workflow_dispatch only, so
changes don't burn API credits on every PR but can still be checked
against a real model on demand.

---------

Co-authored-by: Claude <noreply@anthropic.com>
2026-06-17 12:02:47 +01:00

123 行
4.1 KiB
Go

package flow
// Options configures a Flow.
type Options struct {
// TriggerTopic is the broker topic that triggers this flow.
TriggerTopic string
// Prompt is a Go template string. {{.Data}} is the event payload.
Prompt string
// SystemPrompt is the system instruction for the LLM.
SystemPrompt string
// Provider is the AI provider name (e.g. "anthropic", "openai").
Provider string
// APIKey for the AI provider.
APIKey string
// Model overrides the provider's default model.
Model string
// BaseURL overrides the provider's default base URL.
BaseURL string
// HistoryLimit is the max messages per flow execution.
HistoryLimit int
// OnResult is called after each execution with the result.
OnResult func(Result)
// Agent, if set, names a registered agent the flow hands each event
// to (over RPC). The flow triggers; the agent reasons. When empty,
// the flow runs a single augmented-LLM step itself.
Agent string
// Steps, if set, makes the flow run an ordered list of steps per
// event instead of a single LLM step — the deterministic-workflow
// path. Checkpointed between steps when a Checkpoint is set.
Steps []Step
// Retry is the flow-level retry count applied to each step (0 = no
// retry). A Step's own Retry field overrides this.
Retry int
// Checkpoint is the durability backend for stepped runs. Nil with
// steps present means a store-backed default; set it to swap backends.
Checkpoint Checkpoint
// DeleteOnSuccess removes a run's checkpoint when it completes
// successfully. Failed runs are always retained. Default: retain all.
DeleteOnSuccess bool
}
// Option applies a configuration to Options.
type Option func(*Options)
// Trigger sets the broker topic that triggers this flow.
func Trigger(topic string) Option {
return func(o *Options) { o.TriggerTopic = topic }
}
// Prompt sets the prompt template. Use {{.Data}} for the event payload.
func Prompt(p string) Option {
return func(o *Options) { o.Prompt = p }
}
// SystemPrompt sets the system instruction for the LLM.
func SystemPrompt(p string) Option {
return func(o *Options) { o.SystemPrompt = p }
}
// Provider sets the AI provider name.
func Provider(name string) Option {
return func(o *Options) { o.Provider = name }
}
// APIKey sets the API key for the AI provider.
func APIKey(key string) Option {
return func(o *Options) { o.APIKey = key }
}
// Model sets the model name.
func Model(name string) Option {
return func(o *Options) { o.Model = name }
}
// BaseURL sets the provider base URL.
func BaseURL(url string) Option {
return func(o *Options) { o.BaseURL = url }
}
// HistoryLimit sets the max messages per execution.
func HistoryLimit(n int) Option {
return func(o *Options) { o.HistoryLimit = n }
}
// OnResult sets a callback for each execution result.
func OnResult(fn func(Result)) Option {
return func(o *Options) { o.OnResult = fn }
}
// Agent makes the flow hand each event to a named registered agent over
// RPC instead of running its own LLM step. The flow triggers; the agent
// reasons (with its plan, delegate, memory, and guardrails).
func Agent(name string) Option {
return func(o *Options) { o.Agent = name }
}
// Steps sets the ordered steps of the flow. A flow with steps runs them
// in order per event, checkpointing between each, instead of the
// single-step prompt/agent behavior. Step names must be unique.
func Steps(steps ...Step) Option {
return func(o *Options) { o.Steps = steps }
}
// Retry sets the flow-level retry count applied to each step (0 = no
// retry). A Step's own Retry field overrides this.
func Retry(n int) Option {
return func(o *Options) { o.Retry = n }
}
// WithCheckpoint sets the durability backend. With a checkpoint, a run is
// persisted before and after each step and can be resumed after a crash.
// Stepped flows default to a store-backed checkpoint; use this to swap it.
func WithCheckpoint(c Checkpoint) Option {
return func(o *Options) { o.Checkpoint = c }
}
// DeleteOnSuccess removes a run's checkpoint when it completes
// successfully. Failed runs are always retained. Default: retain all.
func DeleteOnSuccess() Option {
return func(o *Options) { o.DeleteOnSuccess = true }
}