loom

README

English | 中文

Loom is a small Go library for agent workflows that need to pause for a human, resume later, and survive a crash.

An agent in Loom is an explicit graph of steps over an inspectable State. The engine executes the steps in graph order, checkpoints after each one through a pluggable Store, and can freeze mid-run (yield) and continue later (Resume) — on another day, in another process. What it does not guarantee: your steps are your functions, so a step that calls an LLM or reads a clock is as reproducible as you make it. Loom holds the topology, the execution order, and the recovery; the rest is yours.

Status: v0.8.0, pre-1.0. API changes are additive; breaking changes ship with a minor bump and migration notes. Requires Go 1.24+. MIT licensed.

Quickstart

go get github.com/jinyitao123/loom   # inside your Go module

package main

import (
    "context"
    "fmt"
    "log"

    "github.com/jinyitao123/loom"
)

func main() {
    greet := func(_ context.Context, s loom.State) (loom.State, error) {
        return loom.State{"output": "Hello, " + s["name"].(string) + "!"}, nil
    }

    g := loom.NewGraph("greeter", "greet")
    g.AddStep("greet", greet, loom.End())

    result, err := g.Run(context.Background(), loom.State{"name": "World"}, nil)
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println(result.State["output"]) // Hello, World!
}

API reference: pkg.go.dev/github.com/jinyitao123/loom

What this buys you

Pause for human approval. A step sets __yield: true; the graph freezes with its state checkpointed. Days later, after the human decides, Resume picks up exactly where it stopped:

result, err := g.Run(ctx, input, store)
// result.StopReason == "yielded"

result, err = g.Resume(ctx, result.RunID, loom.State{"approved": true}, store)

Survive a crash. Every step auto-checkpoints. Process dies after step C, restart, resume — it continues at step D with C's state intact:

result, err = g.Resume(ctx, runID, loom.State{}, pgStore)

Newly written checkpoints carry schema_version: 1; newer binaries read legacy checkpoints, older binaries fail closed on newer ones. The same rule holds for Resume, ResumeAt, and history reads.

Facts

• Five primitives — State, Step, Router, Graph, Store. No Agent class, no Chain abstraction, no Memory base type; everything else is composed, not inherited.
• The whole root package is ~1,650 lines across nine files (wc -l, comments included). Stdlib ~3,100; contract interfaces ~200.
• 306 tests and 4 fuzz targets; CI race-checks the CLI on Linux, macOS, and Windows with an 85% coverage gate.
• The core library has one runtime dependency (google/uuid). pgx is pulled only if you import pgstore.
• Benchmarks live in tests/ — step overhead is in the microsecond range with the in-memory store. Run them yourself: go test ./tests/ -run '^$' -bench=.

What the kernel deliberately doesn't know

The kernel doesn't know	So you can
What an LLM is	Use OpenAI, Claude, DeepSeek, local models — swap freely inside a Step
What MCP is	Plug in any tool protocol — MCP / A2A / custom RPC
How to store memory	RAG, graph DB, full-text search — what goes in State is your call
How to serve HTTP	Gin, Echo, net/http — Loom is a library, not a service

The kernel executes steps in graph order, checkpoints along the way, and pauses on yield. Everything else — prompts, tools, memory, transport — is your domain, assembled from stdlib building blocks (tool loop, permissions, budgets, sessions, sub-graphs) or your own code.

When not to use Loom

• You want batteries included. Built-in RAG, vector memory, a visual flow designer, a hosted platform — Loom has none of these, on purpose. LangGraph or the OpenAI Agents SDK will get you there faster.
• Your stack is Python or TypeScript. Loom is Go only.
• You need a control plane. Multi-tenant auth, an approval UI, audit trails, addressable agents — that's Weave (repo going public soon), the platform layer built on top of Loom. Loom itself stays a library plus a single-process CLI.

Note the layers differ: the kernel knows nothing about LLMs or memory; stdlib ships session and prompt utilities; the loom CLI adds MCP tools, session resume, and semantic memory on top.

Comparison

	Loom	LangGraph	OpenAI Agents SDK
Language	Go	Python	Python
Core library size¹	~1.6K LOC (root package, 9 files)	~27.9K LOC (libs/langgraph/langgraph)	—
Persistence	Auto checkpoint per step	Checkpointer (opt-in)	None built in
LLM coupling	Zero	LangChain ecosystem	OpenAI-first
Tool protocol	Any	LangChain tools	function calling
Sub-graph nesting	Native	Native	Handoffs
Human-in-the-loop	yield / resume, state checkpointed	interrupt	Limited

¹ wc -l on the core package, comments included, measured 2026-07-30: loom@9f4c974, langgraph@4134145. Same tool, same rules on both sides.

The loom CLI

The repo also ships loom, a standalone agent engine built on the library — prompt JSON on stdin, one agent turn, NDJSON events on stdout, with MCP tool servers, session resume, semantic memory, and sub-agent orchestration compiled from an agent spec.

go install github.com/jinyitao123/loom/cmd/loom@latest

# or from a GitHub Release tarball (linux / macOS):
curl -fsSL https://raw.githubusercontent.com/jinyitao123/loom/main/install.sh | sh

See cmd/loom/README.md for the event wire format, and docs/host-integration.md for driving it from any host process.

Project structure

loom/
├── graph.go          Execution engine: State × Step × Router → Run / Resume
├── state.go          Typed map with registrable merge policies
├── step.go           type Step func(ctx, State) (State, error)
├── router.go         Control flow: Always / Branch / Condition
├── store.go          5-method persistence interface
├── lifecycle.go      Host observation seams (allocation / terminal / checkpoint)
├── options.go        GraphOption: merge / checkpoint / budget
├── memstore.go       In-memory Store (for testing)
│
├── contract/         Pure interfaces: LLM / ToolDispatcher / Embedder
├── stdlib/           Pre-built Steps & Hooks (tool loop, permissions, budget, session, …)
├── pgstore/          PostgreSQL Store
├── provider/         LLM providers (OpenAI-compatible / DeepSeek)
├── cmd/loom/         The `loom` CLI
├── tests/            Black-box test suite (public API only)
└── docs/             Host-integration contract & orchestration design

Contributing

See CONTRIBUTING.md. The short version: five primitives, layered imports, no business vocabulary in the kernel — and the CI guards to prove it.

License

MIT

Documentation

Index

• Constants
• Variables
• type CheckpointEvent
• type CheckpointInfo
• type CheckpointObservationStage
• type CheckpointObserver
• type CheckpointPolicy
• type Edge
• type Graph
  • func NewGraph(name string, entry string, opts ...GraphOption) *Graph
  • func (g *Graph) AddStep(name string, step Step, after Router)
  • func (g *Graph) Entry() string
  • func (g *Graph) History(ctx context.Context, store Store, runID string) ([]CheckpointInfo, error)
  • func (g *Graph) Resume(ctx context.Context, runID string, input State, store Store) (*RunResult, error)
  • func (g *Graph) ResumeAt(ctx context.Context, runID string, seq int64, input State, store Store) (*RunResult, error)
  • func (g *Graph) ResumeAtWithLifecycle(ctx context.Context, sourceRunID string, seq int64, allocatedRunID string, ...) (*RunResult, error)
  • func (g *Graph) ResumeWithLifecycle(ctx context.Context, runID string, input State, store Store, ...) (*RunResult, error)
  • func (g *Graph) Run(ctx context.Context, input State, store Store) (*RunResult, error)
  • func (g *Graph) RunWithLifecycle(ctx context.Context, input State, store Store, hooks LifecycleHooks) (*RunResult, error)
  • func (g *Graph) SetHooks(h HookPoints)
  • func (g *Graph) SetTopology(topo []StepInfo)
  • func (g *Graph) StepNames() []string
  • func (g *Graph) Topology() []StepInfo
• type GraphOption
  • func WithCheckpointHistory(keep int) GraphOption
  • func WithCheckpointPolicy(p CheckpointPolicy) GraphOption
  • func WithMaxIterations(n int) GraphOption
  • func WithMergeConfig(cfg *MergeConfig) GraphOption
  • func WithStepBudget(n int64) GraphOption
• type HookPoints
• type LifecycleHooks
• type MemStore
  • func NewMemStore() *MemStore
  • func (s *MemStore) Delete(_ context.Context, ns, key string) error
  • func (s *MemStore) Get(_ context.Context, ns, key string) ([]byte, error)
  • func (s *MemStore) List(_ context.Context, ns, prefix string) ([]string, error)
  • func (s *MemStore) Put(_ context.Context, ns, key string, value []byte) error
  • func (s *MemStore) Tx(_ context.Context, fn func(Store) error) error
• type MergeConfig
  • func DefaultMergeConfig() *MergeConfig
  • func NewMergeConfig() *MergeConfig
  • func (mc *MergeConfig) Register(key string, policy MergePolicy)
• type MergePolicy
• type Router
  • func Always(next string) Router
  • func Branch(key string, routes map[string]string, fallback string) Router
  • func BranchFunc(extract func(State) string, routes map[string]string, fallback string) Router
  • func Condition(pred func(State) bool, ifTrue, ifFalse string) Router
  • func End() Router
• type RunAllocated
• type RunAllocationEvent
• type RunExecutionContext
• type RunResult
• type State
  • func UnmarshalState(b []byte) (State, error)
  • func (s State) Marshal() ([]byte, error)
  • func (s State) Merge(update State, cfg *MergeConfig) State
• type Step
• type StepHook
• type StepInfo
• type StopReason
• type Store
• type TerminalizationEvent
• type Terminalizer

Constants

const CurrentCheckpointSchema = 1

CurrentCheckpointSchema is the newest checkpoint format this binary can read.

Variables

var (
	ErrMaxIterations      = errors.New("loom: max iterations exceeded")           // 单图迭代熔断：本图步数超过 maxIter 上限（防路由环路死循环）
	ErrBudgetExhausted    = errors.New("loom: global step budget exhausted")      // 全局预算耗尽：父图与所有子图共享的步数预算被扣完
	ErrStepNotFound       = errors.New("loom: step not found")                    // 拓扑错误：路由到了图中未注册的步骤名
	ErrCheckpointNotFound = errors.New("loom: checkpoint not found")              // 恢复失败：按 runID 找不到检查点（从未运行或已被清理）
	ErrCorruptCheckpoint  = errors.New("loom: corrupt checkpoint")                // 恢复失败：检查点存在但 JSON 反序列化失败（数据损坏/版本不兼容）
	ErrNestedTx           = errors.New("loom: nested transactions not supported") // 协议约束：Store.Tx 内再开事务必须返回此错误，禁止静默的伪嵌套
)

包级哨兵错误：内核所有可预期的失败模式收敛为这几种固定值， 供调用方（含 Store 实现方）做程序化分支处理。

Types

type CheckpointEvent

type CheckpointEvent struct {
	Stage     CheckpointObservationStage
	RunID     string
	GraphName string
	Seq       int64
}

CheckpointEvent describes one latest-checkpoint persistence boundary.

type CheckpointInfo

type CheckpointInfo struct {
	Seq        int64     `json:"seq"`                   // 步序号：History 按此对应的零填充键升序返回
	LastStep   string    `json:"last_step"`             // 该快照完成（或暂停于）的步骤名
	YieldPhase string    `json:"yield_phase,omitempty"` // 暂停阶段；普通步骤完成快照为空
	SavedAt    time.Time `json:"saved_at"`              // 检查点成功写入 latest 前生成的时间戳
}

CheckpointInfo is the public metadata view of one historical checkpoint. CheckpointInfo 是单条历史检查点的公开元数据视图：不暴露完整 State，供列表与审计使用。

type CheckpointObservationStage

type CheckpointObservationStage string

CheckpointObservationStage identifies the latest-checkpoint boundary being observed. History writes are intentionally outside this lifecycle seam.

const (
	CheckpointLatestPutBefore CheckpointObservationStage = "latest_put_before"
	CheckpointLatestPutAfter  CheckpointObservationStage = "latest_put_after"
)

type CheckpointObserver

type CheckpointObserver interface {
	ObserveCheckpoint(context.Context, CheckpointEvent) error
}

CheckpointObserver synchronously observes latest-checkpoint persistence.

type CheckpointPolicy

type CheckpointPolicy int

CheckpointPolicy controls how checkpoint failures are handled. CheckpointPolicy（检查点策略）决定检查点落盘失败时引擎的取舍： 是“尽力而为、失败只告警继续跑”，还是“必须落盘、失败即中止”。

const (
	// CheckpointBestEffort logs checkpoint failures but does not abort.
	// 尽力而为策略（默认）：检查点写失败仅记 Warn 日志、执行继续。
	// 取舍：优先保证运行可用性，代价是失败后若进程崩溃将无法从最新进度恢复。
	CheckpointBestEffort CheckpointPolicy = iota
	// CheckpointRequired aborts the graph if checkpoint fails.
	// 强制落盘策略：检查点写失败立即中止整个图。
	// 取舍：优先保证可恢复性（每一步都有可靠快照），适合 HITL 等必须能恢复的场景。
	CheckpointRequired
)

type Edge

type Edge struct {
	// To：目标步骤名；空串 "" 表示终点，与路由器“返回空串即停机”的协议一致。
	To string `json:"to"` // target step name ("" = end)
	// Label：该边的条件标签（例如分支取值），仅供可视化展示。
	Label string `json:"label,omitempty"` // condition label
}

Edge describes a possible transition from one step to another (for topology export). Edge 描述一条可能的转移边，仅用于拓扑导出/可视化，不参与实际路由决策。

type Graph

type Graph struct {
	Name string // 图名：用作检查点命名空间（"checkpoint:"+Name）及日志/错误标识
	// contains filtered or unexported fields
}

Graph is an executable composition of Steps and Routers. Graph（图）是步骤与路由器的可执行组合体，是内核对外的执行入口。 一切上层能力（agent、子图、HITL）都由这一个结构承载；字段在构造期定型，运行期只读， 因此同一个 Graph 可被多个 goroutine 并发 Run（各自的状态互相独立）。

func NewGraph

func NewGraph(name string, entry string, opts ...GraphOption) *Graph

NewGraph creates a new graph with the given entry step. NewGraph 构造一个图：设定图名与入口步骤，应用函数式选项，最后冻结合并配置。

func (*Graph) AddStep

func (g *Graph) AddStep(name string, step Step, after Router)

AddStep registers a step with an optional router for the "after" transition. AddStep 注册一个步骤及其“执行后”路由器；after 传 nil 表示该步骤是终端步骤，执行完即整图停机。

func (*Graph) Entry

func (g *Graph) Entry() string

Entry returns the graph's configured entry step.

func (*Graph) History

func (g *Graph) History(ctx context.Context, store Store, runID string) ([]CheckpointInfo, error)

History lists valid historical checkpoint metadata for a run in ascending step order. History 列出某次运行的有效历史元数据：依赖 Store.List 的字典序契约与零填充键，天然按步升序返回。 单条键读取失败或 JSON 损坏只记 Warn 并跳过，不能让局部坏档阻断其余历史的审计与恢复。

func (*Graph) Resume

func (g *Graph) Resume(ctx context.Context, runID string, input State, store Store) (*RunResult, error)

Resume restarts a yielded graph from its checkpoint. Resume 从检查点恢复一次已暂停（HITL）的运行： 读检查点 → 把人工提供的 input 增量合并进快照状态 → 按 yield_phase 决定重入方式 （mid_step=重跑暂停的那一步；after_step=跳过该步直接走它的路由）。 兼容差异：Resume 对空 phase 按 mid_step 处理，因为 v1.0 latest 表示“暂停中”的恢复点； ResumeAt 则对空 phase 按 after_step 处理，因为历史条目是步骤完成后的快照，重跑会造成双重作用。

func (*Graph) ResumeAt

func (g *Graph) ResumeAt(ctx context.Context, runID string, seq int64, input State, store Store) (*RunResult, error)

ResumeAt forks a new run from an immutable historical checkpoint. ResumeAt 从任意历史检查点分叉出全新运行：读取源快照但只向新 UUID 写检查点，源存档树绝不改写。 与 Resume 的 v1.0 兼容语义不同，空 yield phase 在这里按 after_step 处理；历史条目记录的是步骤完成 后的状态，若按 mid_step 重跑原步骤，外部副作用与状态增量都可能重复发生。fork 默认继承源快照 在分叉点的 __budget_remaining；宿主可在 input 中显式提供同名键，借由下方 Merge 覆盖该余额。

func (*Graph) ResumeAtWithLifecycle

func (g *Graph) ResumeAtWithLifecycle(
	ctx context.Context,
	sourceRunID string,
	seq int64,
	allocatedRunID string,
	input State,
	store Store,
	hooks LifecycleHooks,
) (*RunResult, error)

ResumeAtWithLifecycle forks a historical checkpoint into the exact caller-allocated run identity and applies per-invocation lifecycle hooks.

func (*Graph) ResumeWithLifecycle

func (g *Graph) ResumeWithLifecycle(
	ctx context.Context,
	runID string,
	input State,
	store Store,
	hooks LifecycleHooks,
) (*RunResult, error)

ResumeWithLifecycle resumes an existing run with per-invocation lifecycle hooks. Resume does not allocate a new run identity.

func (*Graph) Run

func (g *Graph) Run(ctx context.Context, input State, store Store) (*RunResult, error)

Run executes the graph from the entry step to completion or yield. Run 从入口步骤开始执行图，直到正常完成、HITL 暂停或出错。 主循环各阶段固定顺序：单图熔断 → 全局预算 → Before 钩子 → 执行步骤 → 合并增量 → After 钩子 → 检查点 → yield（暂停）检查 → 路由决定下一跳。

func (*Graph) RunWithLifecycle

func (g *Graph) RunWithLifecycle(
	ctx context.Context,
	input State,
	store Store,
	hooks LifecycleHooks,
) (*RunResult, error)

RunWithLifecycle executes a caller-allocated run with per-invocation lifecycle hooks. input must carry the authoritative non-empty __run_id.

func (*Graph) SetHooks

func (g *Graph) SetHooks(h HookPoints)

SetHooks attaches before/after hooks. SetHooks 挂载前置/后置钩子集合（整体替换而非追加）。

func (*Graph) SetTopology

func (g *Graph) SetTopology(topo []StepInfo)

SetTopology declares the graph's topology for visualization. Called by graph builders (CompileAgent, newMirrorGraph, etc.) after constructing the graph. SetTopology 声明图的拓扑（仅供可视化）；由图构建器建图完成后调用，对执行路径无任何影响。

func (*Graph) StepNames

func (g *Graph) StepNames() []string

StepNames returns the registered step names in stable lexical order.

func (*Graph) Topology

func (g *Graph) Topology() []StepInfo

Topology returns the declared topology, or nil if not set. Topology 返回已声明的拓扑；未声明时返回 nil，调用方需自行判空。

type GraphOption

type GraphOption func(*Graph)

GraphOption configures a Graph at construction time. GraphOption 是构造期配置函数（函数式选项模式）：只在 NewGraph 时生效，图建成后配置即定型。

func WithCheckpointHistory

func WithCheckpointHistory(keep int) GraphOption

WithCheckpointHistory configures optional per-step checkpoint history retention. WithCheckpointHistory(keep) 配置可选的按步检查点历史：0=关闭（默认，保持 latest 覆盖写）； -1=全部保留；n>0=只保留最近 n 条。历史写入始终是尽力而为，不改变 checkpointPolicy。

func WithCheckpointPolicy

func WithCheckpointPolicy(p CheckpointPolicy) GraphOption

WithCheckpointPolicy 设置检查点失败策略（默认 CheckpointBestEffort）。

func WithMaxIterations

func WithMaxIterations(n int) GraphOption

WithMaxIterations 设置单图迭代上限（per-graph 熔断，默认 100），防止路由环路导致死循环。

func WithMergeConfig

func WithMergeConfig(cfg *MergeConfig) GraphOption

WithMergeConfig 指定图的合并配置；注意 NewGraph 会随即将其冻结（frozen），运行期不可再改策略。

func WithStepBudget

func WithStepBudget(n int64) GraphOption

WithStepBudget sets a global step limit shared across parent and all sub-graphs. WithStepBudget 设置全局步数预算：与 maxIter 不同，它在根图 Run 时装入 atomic 计数器、 经 context 传递给所有子图共享，是第二层熔断——即使每个子图各自不超 maxIter， 全体累计步数也不能超出此预算。只应在根图上设置。

type HookPoints

type HookPoints struct {
	Before []StepHook // 前置钩子：在步骤执行前运行，任一报错则该步骤不会被执行
	After  []StepHook // 后置钩子：在步骤增量合并进状态之后运行，可观察到合并后的完整状态
}

HookPoints holds before/after hooks for graph execution. HookPoints 汇集图执行的前置/后置钩子，各自按注册顺序依次执行。

type LifecycleHooks

type LifecycleHooks struct {
	RunAllocated       RunAllocated
	ExecutionContext   RunExecutionContext
	Terminalizer       Terminalizer
	CheckpointObserver CheckpointObserver
}

LifecycleHooks configures per-execution lifecycle callbacks. Hooks are passed to an invocation instead of stored on Graph so one Graph remains safe for concurrent callers with different lifecycle policies.

type MemStore

type MemStore struct {
	// contains filtered or unexported fields
}

MemStore is an in-memory Store implementation for testing. MemStore 是内存版 Store 实现，仅供测试使用：不落盘、进程退出即失。 并发安全手段：一把 sync.RWMutex 保护两级 map——读操作持读锁可并行，写操作持写锁互斥。

func NewMemStore

func NewMemStore() *MemStore

NewMemStore creates a new empty in-memory store. NewMemStore 创建一个空的内存存储。

func (*MemStore) Delete

func (s *MemStore) Delete(_ context.Context, ns, key string) error

Delete 并发安全删除：取写锁后委托给无锁内核。

func (*MemStore) Get

func (s *MemStore) Get(_ context.Context, ns, key string) ([]byte, error)

Get 并发安全读取：取读锁后委托给无锁内核。

func (*MemStore) List

func (s *MemStore) List(_ context.Context, ns, prefix string) ([]string, error)

List 并发安全列举：取读锁后委托给无锁内核。

func (*MemStore) Put

func (s *MemStore) Put(_ context.Context, ns, key string, value []byte) error

Put 并发安全写入：取写锁后委托给无锁内核。

func (*MemStore) Tx

func (s *MemStore) Tx(_ context.Context, fn func(Store) error) error

Tx runs a function inside a simulated transaction using copy-on-write. Tx 用“整库深拷贝 + 整体替换”模拟事务：持写锁期间在快照副本上执行 fn， fn 成功则把副本整体换入（提交），失败则丢弃副本（回滚），原数据全程不被触碰。 代价是 O(全库) 的拷贝且事务期间阻塞所有其他操作——仅测试场景可接受。

type MergeConfig

type MergeConfig struct {
	// contains filtered or unexported fields
}

MergeConfig holds per-key merge policies. It is mutable during construction and frozen once passed to a Graph. MergeConfig（合并配置）保存“键 → 合并策略”的映射及兜底策略。 生命周期协议：构造期可随意 Register；一旦挂到 Graph 上即被冻结—— 原因是配置可能被多个图共享，且运行中改策略会让同一图内不同时刻的合并语义不一致， 破坏检查点的可重放性。

func DefaultMergeConfig

func DefaultMergeConfig() *MergeConfig

DefaultMergeConfig returns stdlib's recommended defaults. DefaultMergeConfig 返回 stdlib 推荐默认值：messages 键用 AppendSlice 累积会话历史，其余键覆盖。

func NewMergeConfig

func NewMergeConfig() *MergeConfig

NewMergeConfig 创建一份空白配置：无按键策略、兜底为 Overwrite。

func (*MergeConfig) Register

func (mc *MergeConfig) Register(key string, policy MergePolicy)

Register 为指定键注册合并策略；只允许在配置尚未挂到 Graph（未冻结）时调用。

type MergePolicy

type MergePolicy func(existing, incoming any) any

MergePolicy defines how a specific key is merged. MergePolicy（合并策略）定义单个键的合并规则：输入旧值与新值，返回合并结果。 策略只关心“怎么合”，“何时合”由引擎在每个步骤之后统一驱动。

var (
	// Overwrite：新值直接覆盖旧值，是最朴素也是默认的合并语义。
	Overwrite MergePolicy = func(_, incoming any) any { return incoming }

	// AppendSlice：把新切片追加到旧切片之后，典型用于 messages 这类只增不减的会话历史。
	AppendSlice MergePolicy = func(existing, incoming any) any {
		e, _ := existing.([]any)
		i, _ := incoming.([]any)
		return append(e, i...)
	}

	// SumInt：整型累加，适合进程内计数器类键（如重试次数）。
	SumInt MergePolicy = func(existing, incoming any) any {
		e, _ := existing.(int)
		i, _ := incoming.(int)
		return e + i
	}

	// SumFloat：浮点累加，适合成本、token 用量等度量；JSON 反序列化的数字默认就是 float64，可安全跨检查点累加。
	SumFloat MergePolicy = func(existing, incoming any) any {
		e, _ := existing.(float64)
		i, _ := incoming.(float64)
		return e + i
	}
)

Built-in policies. 内置合并策略：覆盖最常见的四种合并语义。 各策略均使用宽容的类型断言（断言失败取零值），保证 JSON 往返后的类型漂移不会引发 panic。

type Router

type Router func(ctx context.Context, state State) (string, error)

Router determines the next step based on current state. Returns the name of the next step, or "" to halt. Router（路由器）在每个步骤之后决定下一跳：输入当前状态，返回下一个步骤名。 核心协议：返回空字符串 "" 表示停机（图正常结束）。 设计立场：路由是确定性代码而非模型决策——LLM 只产出状态内容，走哪条边由代码裁决， 因此控制流可测试、可回放、与模型输出解耦。

func Always

func Always(next string) Router

Always returns the same next step. Always 构造无条件路由器：无论状态如何都固定跳到 next，用于线性流水线的顺序衔接。

func Branch

func Branch(key string, routes map[string]string, fallback string) Router

Branch routes based on a state key's string value. Branch 按状态中 key 对应值的字符串形式查表路由：命中 routes 走对应步骤，未命中走 fallback。

func BranchFunc

func BranchFunc(extract func(State) string, routes map[string]string, fallback string) Router

BranchFunc routes based on a user-supplied key extractor. BranchFunc 是 Branch 的泛化版本：路由键由调用方提供的提取函数计算， 适合键值需要组合多个状态字段或额外加工的场景。

func Condition

func Condition(pred func(State) bool, ifTrue, ifFalse string) Router

Condition routes based on a predicate. Condition 构造二元条件路由器：谓词为真走 ifTrue、为假走 ifFalse，即最简单的 if/else 分支。

func End

func End() Router

End always halts the graph. End 构造终止路由器：恒定返回空字符串，按协议表示停机（图正常完成）。

type RunAllocated

type RunAllocated interface {
	RunAllocated(context.Context, RunAllocationEvent) error
}

RunAllocated synchronously admits one run identity before execution starts.

type RunAllocationEvent

type RunAllocationEvent struct {
	RunID       string
	GraphName   string
	ParentRunID string
	ParentSeq   int64
	State       State
}

RunAllocationEvent describes one run identity before any graph step or router is evaluated.

type RunExecutionContext

type RunExecutionContext interface {
	ExecutionContext(context.Context, RunAllocationEvent) (context.Context, error)
}

RunExecutionContext optionally derives the context used by routers and steps after durable allocation. The returned context must inherit the input context so checkpoint observation and cancellation remain invocation-local.

type RunResult

type RunResult struct {
	State      State      // 停机时的完整状态快照（可能含 __error 等引擎协议键）
	LastStep   string     // 最后执行（或试图执行）的步骤名，Resume 与排障都依赖它定位
	Yielded    bool       // 是否因 HITL 暂停而停：true 时调用方应保留 RunID 以便后续 Resume
	Steps      int        // 本次调用实际执行的步骤数（不含 Resume 之前的历史步数）
	RunID      string     // 运行标识（__run_id）：同一次业务运行跨 Run/Resume 保持不变，是检查点的主键
	StopReason StopReason // 停机原因分类
}

RunResult contains the final state and metadata of a graph execution. RunResult 是一次图执行的结果：停机时的完整状态 + 停机元数据。

type State

type State map[string]any

State is the execution context shared across all steps. Keys are strings. Values are any JSON-serializable type. State（状态）是贯穿所有步骤的共享执行上下文：键为字符串、值为任意可 JSON 序列化类型的 map。 约束：值必须能过 JSON 往返（检查点依赖 Marshal），塞入不可序列化的值会在落盘时报错； 双下划线前缀键（__run_id、__yield、__yield_phase、__error、__failed_step）为引擎协议键，业务侧不应占用。

func UnmarshalState

func UnmarshalState(b []byte) (State, error)

UnmarshalState 从 JSON 字节串还原状态；注意 JSON 往返后数值统一变为 float64、数组变为 []any。

func (State) Marshal

func (s State) Marshal() ([]byte, error)

Marshal 把状态序列化为 JSON 字节串，是检查点落盘所用的编码格式。

func (State) Merge

func (s State) Merge(update State, cfg *MergeConfig) State

Merge applies the update to the current state using registered policies. Merge 把步骤返回的增量 update 按策略并入当前状态，并且每次都返回一个全新 map、绝不原地修改。 这是内核的核心不变量：步骤只产出增量、引擎负责合并，旧状态保持不可变—— 由此每个检查点都是自洽快照，恢复重放与并发读取都不会被后续修改污染。

type Step

type Step func(ctx context.Context, state State) (State, error)

Step is the atomic unit of execution. It receives the current state and returns the delta to merge. Returning a non-nil error aborts the graph. Step（步骤）是引擎的最小执行单元：输入当前状态（State），返回“状态增量”而非完整状态。 核心协议：步骤只产出增量，合并由引擎统一调用 State.Merge 完成——这保证每个检查点都是 合并后的自洽快照，也让恢复（resume）时重跑步骤是安全的。 返回非 nil error 会中止整个图：引擎会把错误现场写入 __error / __failed_step 协议键， 并尽力落一次检查点保存现场。

type StepHook

type StepHook func(ctx context.Context, stepName string, state State) error

StepHook runs before or after each step execution. StepHook（步骤钩子）在每个步骤执行前/后被调用，承载日志、鉴权、指标等横切逻辑； 返回非 nil error 会以 StopHookAbort 中止整个图——钩子对执行拥有否决权。

type StepInfo

type StepInfo struct {
	Name string `json:"name"` // 步骤名
	// Detail：人类可读的补充说明（如该步骤使用的模型或工具）。
	Detail string `json:"detail,omitempty"` // human-readable annotation
	Edges  []Edge `json:"edges"`            // 出边列表：该步骤可能转移到的目标
}

StepInfo describes a step and its outgoing edges for topology export. StepInfo 描述一个步骤及其全部出边，仅用于拓扑导出/可视化。

type StopReason

type StopReason string

StopReason classifies why a graph execution ended. StopReason（停机原因）对图执行的终止方式做枚举分类，供调用方在结果上做程序化分支。

const (
	// StopCompleted：正常结束——路由器返回空字符串表示停机。
	StopCompleted StopReason = "completed" // normal termination (router returned "")
	// StopYielded：HITL 暂停——步骤置 __yield 主动让出控制权，等待人工介入后 Resume。
	StopYielded StopReason = "yielded" // HITL pause (__yield)
	// StopMaxIter：单图迭代熔断——本图步数超过 maxIter 上限。
	StopMaxIter StopReason = "max_iter" // per-graph circuit breaker
	// StopBudget：全局预算耗尽——父图与所有子图共享的步数预算被扣完。
	StopBudget StopReason = "budget" // global step budget exhausted
	// StopError：步骤返回非 nil error（错误现场已写入 __error / __failed_step）。
	StopError StopReason = "error" // Step returned non-nil error
	// StopHookAbort：前置/后置钩子返回 error，行使否决权中止执行。
	StopHookAbort StopReason = "hook_abort" // Before/After hook returned error
)

type Store

type Store interface {
	Get(ctx context.Context, ns string, key string) ([]byte, error)       // 读取 ns 下 key 的值；协议约定键不存在必须返回错误（而非 nil, nil）
	Put(ctx context.Context, ns string, key string, value []byte) error   // 写入或覆盖 ns 下 key 的值；实现方应保证写入的持久性
	Delete(ctx context.Context, ns string, key string) error              // 删除 ns 下的 key；删除不存在的键视为成功（幂等语义）
	List(ctx context.Context, ns string, prefix string) ([]string, error) // 列出 ns 下以 prefix 开头的全部键，约定按字典序返回（结果确定可复现）

	// Tx runs a function inside a database transaction.
	// The Store passed to fn is bound to the transaction.
	// If fn returns nil, the transaction commits. If fn returns an error,
	// the transaction rolls back.
	// Nested Tx calls are NOT supported and must return an error.
	// Tx 在数据库事务内执行 fn：传给 fn 的 Store 已绑定到该事务，
	// fn 返回 nil 则整体提交、返回 error 则整体回滚——调用方以此获得多次读写的原子性。
	// 协议约束：不支持嵌套事务，在事务内再调 Tx 必须返回 ErrNestedTx，实现方不得静默忽略。
	Tx(ctx context.Context, fn func(Store) error) error
}

Store provides durable key-value storage with namespace isolation. Store（存储）是引擎的持久化抽象：带命名空间（ns）隔离的 KV 接口。 引擎用它落检查点（命名空间固定为 "checkpoint:<图名>"），stdlib 用它存记忆等长期数据； 命名空间隔离保证不同用途的数据互不污染。实现方（SQLite/Postgres/内存）只需满足此接口。

type TerminalizationEvent

type TerminalizationEvent struct {
	RunID                     string
	GraphName                 string
	StopReason                StopReason
	Yielded                   bool
	LastStep                  string
	State                     State
	LatestCheckpointSeq       int64
	LatestCheckpointPersisted bool
}

TerminalizationEvent describes the final observable state of one graph invocation.

type Terminalizer

type Terminalizer interface {
	Terminalize(context.Context, TerminalizationEvent) error
}

Terminalizer synchronously observes one non-nil RunResult before it is returned to the caller.