interrupt

package
v0.5.11 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Aug 12, 2026 License: Apache-2.0 Imports: 8 Imported by: 0

Documentation

Overview

Package interrupt 提供 Hexagon 框架的中断恢复能力

本包实现 Human-in-the-Loop 模式,支持在任意节点中断执行并等待人工输入。

核心功能:

  • Interrupt: 在节点中触发中断
  • Resume: 恢复执行
  • Checkpoint: 检查点持久化

设计借鉴:

  • LangGraph: interrupt() 函数
  • LangGraph: Command 恢复机制
  • LangGraph: Checkpointer 持久化

使用示例:

func reviewNode(ctx context.Context, state *State) error {
    // 中断等待人工审核
    result, err := interrupt.Interrupt(ctx, ReviewRequest{
        Content: state.Content,
    })
    if err != nil {
        return err
    }
    state.Approved = result.Approved
    return nil
}

// 恢复执行
graph.Resume(ctx, threadID, interrupt.Command{
    Resume: ApprovalResult{Approved: true},
})

Index

Constants

This section is empty.

Variables

View Source
var (
	// ErrInterrupted 表示执行被中断
	ErrInterrupted = errors.New("execution interrupted")

	// ErrNoCheckpoint 表示没有找到检查点
	ErrNoCheckpoint = errors.New("no checkpoint found")

	// ErrInvalidResume 表示恢复数据无效
	ErrInvalidResume = errors.New("invalid resume data")

	// ErrTimeout 表示等待超时
	ErrTimeout = errors.New("interrupt timeout")

	// ErrCanceled 表示被取消
	ErrCanceled = errors.New("interrupt canceled")
)

Functions

func AppendAddressSegment

func AppendAddressSegment(ctx context.Context, segType AddressSegmentType, id, subID string) context.Context

AppendAddressSegment 在 context 中追加一个地址段

同时检查 globalResumeInfo:

  • 如果新地址匹配某个中断点地址 → 注入 InterruptState 到 context
  • 如果新地址匹配某个恢复目标 → 注入 resumeData 和 isResumeTarget=true
  • 如果新地址的后代是恢复目标 → 标记 isResumeTarget=true

func BatchResumeWithData

func BatchResumeWithData(ctx context.Context, resumeData map[string]any) context.Context

BatchResumeWithData 批量恢复多个中断点

适用于 CompositeInterrupt 场景:多个子中断需要同时恢复。 resumeData: map[interruptID]data

func CompositeInterrupt

func CompositeInterrupt(ctx context.Context, info any, state any, subErrors ...error) error

CompositeInterrupt 组合中断 — 聚合多个子中断

用于多个子组件(如多工具调用、多子图)同时触发中断的场景。 从 subErrors 中提取 InterruptSignal,构建树状结构。 非 InterruptSignal 的 error 会被忽略。

用法:

func toolsNode(ctx context.Context, state *State) (*State, error) {
    var errs []error
    for _, call := range state.ToolCalls {
        toolCtx := interrupt.AppendAddressSegment(ctx, interrupt.SegmentTool, call.Name, call.ID)
        _, err := executeTool(toolCtx, call)
        if err != nil {
            if _, ok := interrupt.IsInterruptSignal(err); ok {
                errs = append(errs, err)
                continue
            }
            return state, err
        }
    }
    if len(errs) > 0 {
        return state, interrupt.CompositeInterrupt(ctx, "多个工具需要确认", nil, errs...)
    }
    return state, nil
}

func ContextWithInterruptHandler

func ContextWithInterruptHandler(ctx context.Context, handler *Handler) context.Context

ContextWithInterruptHandler 添加中断处理器到 context

func ContextWithNodeID

func ContextWithNodeID(ctx context.Context, nodeID string) context.Context

ContextWithNodeID 添加节点 ID 到 context

func ContextWithThreadID

func ContextWithThreadID(ctx context.Context, threadID string) context.Context

ContextWithThreadID 添加线程 ID 到 context

func GetInterruptState

func GetInterruptState[T any](ctx context.Context) (wasInterrupted bool, hasState bool, state T)

GetInterruptState 获取中断时保存的组件状态

用于 StatefulInterrupt 场景:组件在中断前保存了内部状态(如处理进度), 恢复时通过此函数获取,跳过已完成的工作。

返回值:

  • wasInterrupted: 当前组件是否曾参与中断(地址匹配到了中断记录)
  • hasState: 是否有保存的状态且类型匹配
  • state: 类型安全的状态对象

func GetResumeContext

func GetResumeContext[T any](ctx context.Context) (isResumeTarget bool, hasData bool, data T)

GetResumeContext 获取恢复上下文

用于组件内部判断当前是否为恢复目标,以及获取恢复时携带的数据。

返回值:

  • isResumeTarget: 当前组件(或其后代)是否为恢复目标
  • hasData: 是否携带了恢复数据且类型匹配
  • data: 类型安全的恢复数据

func Interrupt

func Interrupt[T any](ctx context.Context, payload any) (T, error)

Interrupt 在当前节点触发中断,等待人工输入

调用此函数会: 1. 保存当前状态到检查点 2. 返回 InterruptError 3. 等待 Resume 恢复

参数:

  • ctx: 上下文,必须包含 Handler
  • payload: 中断时携带的数据,会传递给恢复方

返回:

  • T: 恢复时传入的数据
  • error: 错误

func InterruptForApproval

func InterruptForApproval(ctx context.Context, content string) (bool, error)

InterruptForApproval 等待审批

func InterruptForChoice

func InterruptForChoice(ctx context.Context, question string, options []string) (string, error)

InterruptForChoice 等待用户选择

func InterruptForInput

func InterruptForInput(ctx context.Context, prompt string) (string, error)

InterruptForInput 等待用户输入

func InterruptSignalFunc

func InterruptSignalFunc(ctx context.Context, info any) error

InterruptSignalFunc 基础中断 — 不保存组件状态

用于简单的中断场景,如等待审批、请求用户输入等。 中断信号通过 error 返回值传播到调用方。

用法:

func reviewNode(ctx context.Context, state *State) (*State, error) {
    return state, interrupt.InterruptSignalFunc(ctx, "需要审核此内容")
}

func InterruptWithOptions

func InterruptWithOptions[T any](ctx context.Context, payload any, opts ...InterruptOption) (T, error)

InterruptWithOptions 带选项的中断

func MustInterrupt

func MustInterrupt[T any](ctx context.Context, payload any) T

MustInterrupt 简化版中断,panic on error

⚠️ 警告:中断失败时会 panic。 仅在确定中断一定成功时使用。 推荐使用 Interrupt[T]() 方法并正确处理错误。

使用场景:

  • 测试代码中
  • 确定上下文已正确设置的场景

func NodeIDFromContext

func NodeIDFromContext(ctx context.Context) string

NodeIDFromContext 从 context 获取节点 ID

func PopulateResumeInfo

func PopulateResumeInfo(ctx context.Context,
	id2Addr map[string]Address,
	id2State map[string]any,
) context.Context

PopulateResumeInfo 从持久化数据恢复全局恢复信息到 context

在恢复执行前调用,将之前保存的中断点信息注入 context。 之后再调用 Resume/ResumeWithData 标记要恢复的中断点。

用法:

// 从存储加载中断信息
ctx = interrupt.PopulateResumeInfo(ctx, savedAddrs, savedStates)
// 恢复指定中断点
ctx = interrupt.ResumeWithData(ctx, interruptID, approvalData)
// 重新执行图
state, err = graph.Run(ctx, restoredState)

func Resume

func Resume(ctx context.Context, interruptIDs ...string) context.Context

Resume 标记中断点为已恢复(不携带数据)

对指定的中断 ID,将恢复数据设为 nil 并标记为恢复目标。 执行时 AppendAddressSegment 会检测到恢复标记, 使 GetResumeContext 返回 isResumeTarget=true。

func ResumeWithData

func ResumeWithData(ctx context.Context, interruptID string, data any) context.Context

ResumeWithData 恢复单个中断点并携带数据

data 会通过 GetResumeContext[T] 传递给中断组件。 组件可以利用此数据执行不同的恢复逻辑。

func SignalToPersistenceMaps

func SignalToPersistenceMaps(signal *InterruptSignal) (
	id2Addr map[string]Address,
	id2State map[string]any,
)

SignalToPersistenceMaps 将信号树扁平化为两个 map

遍历信号树的所有节点,提取每个中断点的地址和状态。 返回的 map 可以直接 JSON/gob 序列化保存。

参数:

  • signal: 中断信号树的根节点

返回:

  • id2Addr: 中断 ID → 层级地址
  • id2State: 中断 ID → 组件内部状态(仅包含有 State 的节点)

func StatefulInterrupt

func StatefulInterrupt(ctx context.Context, info any, state any) error

StatefulInterrupt 有状态中断 — 保存组件内部状态

用于需要在恢复时跳过已完成工作的场景。 state 参数保存组件的处理进度,恢复后通过 GetInterruptState[T] 获取。

用法:

func batchNode(ctx context.Context, state *State) (*State, error) {
    // 恢复检查
    _, hasState, progress := interrupt.GetInterruptState[*Progress](ctx)
    start := 0
    if hasState { start = progress.LastIndex + 1 }

    for i := start; i < len(items); i++ {
        if needsReview(items[i]) {
            return state, interrupt.StatefulInterrupt(ctx,
                ReviewRequest{Item: items[i]},
                &Progress{LastIndex: i},
            )
        }
    }
    return state, nil
}

func ThreadIDFromContext

func ThreadIDFromContext(ctx context.Context) string

ThreadIDFromContext 从 context 获取线程 ID

Types

type Address

type Address []AddressSegment

Address 层级地址,由多个段组成 例如: [node:step1, tool:search, tool:search:call_1] 表示在 step1 节点中,search 工具的 call_1 调用

func GetCurrentAddress

func GetCurrentAddress(ctx context.Context) Address

GetCurrentAddress 从 context 获取当前地址

func (Address) Append

func (a Address) Append(seg AddressSegment) Address

Append 追加一个段,返回新地址(不修改原地址)

func (Address) Equals

func (a Address) Equals(other Address) bool

Equals 判断两个地址是否完全相等

func (Address) IsDescendantOf

func (a Address) IsDescendantOf(ancestor Address) bool

IsDescendantOf 判断当前地址是否是 ancestor 的后代 即 ancestor 是当前地址的前缀

func (Address) String

func (a Address) String() string

String 返回地址的字符串表示 格式: "node:step1;tool:search:call_1"

type AddressSegment

type AddressSegment struct {
	Type  AddressSegmentType
	ID    string
	SubID string
}

AddressSegment 地址中的一个段,代表层级结构中的一级

每个段包含:

  • Type: 段类型(node/tool/subgraph/agent)
  • ID: 主标识符(节点名、工具名等)
  • SubID: 辅助标识符(区分同名组件的不同调用实例,如并行工具调用)

func (AddressSegment) Equals

func (s AddressSegment) Equals(other AddressSegment) bool

Equals 判断两个段是否相等

func (AddressSegment) String

func (s AddressSegment) String() string

String 返回段的字符串表示 格式: "type:id" 或 "type:id:subID"(当 SubID 非空时)

type AddressSegmentType

type AddressSegmentType string

AddressSegmentType 地址段类型,用于区分不同层级的组件

const (
	// SegmentNode 图节点
	SegmentNode AddressSegmentType = "node"

	// SegmentTool 工具调用
	SegmentTool AddressSegmentType = "tool"

	// SegmentSubgraph 子图
	SegmentSubgraph AddressSegmentType = "subgraph"

	// SegmentAgent Agent
	SegmentAgent AddressSegmentType = "agent"
)

type Checkpoint

type Checkpoint struct {
	ThreadID   string         `json:"thread_id"`
	NodeID     string         `json:"node_id"`
	Payload    any            `json:"payload,omitempty"`
	Status     Status         `json:"status"`
	State      any            `json:"state,omitempty"`
	ResumeData any            `json:"resume_data,omitempty"`
	Metadata   map[string]any `json:"metadata,omitempty"`
	Timestamp  time.Time      `json:"timestamp"`
	Version    int            `json:"version"`
}

Checkpoint 检查点数据

type Command

type Command struct {
	// Resume 恢复值,会作为 Interrupt 的返回值
	Resume any `json:"resume,omitempty"`

	// Goto 跳转到指定节点(可选)
	Goto string `json:"goto,omitempty"`

	// Update 更新状态(可选)
	Update map[string]any `json:"update,omitempty"`
}

Command 恢复命令

type Handler

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

Handler 中断处理器

func HandlerFromContext

func HandlerFromContext(ctx context.Context) *Handler

HandlerFromContext 从 context 获取中断处理器

func NewHandler

func NewHandler(checkpointer checkpoint.Checkpointer) *Handler

NewHandler 创建中断处理器

func (*Handler) Cancel

func (h *Handler) Cancel(threadID string, reason string) error

Cancel 取消中断

func (*Handler) GetPending

func (h *Handler) GetPending(threadID string) *PendingInfo

GetPending 获取待处理的中断

func (*Handler) ListPending

func (h *Handler) ListPending() []*PendingInfo

ListPending 列出所有待处理的中断

func (*Handler) Resume

func (h *Handler) Resume(ctx context.Context, threadID string, cmd Command) error

Resume 恢复执行

type InterruptContext

type InterruptContext struct {
	// ID 中断点唯一 ID(对应 InterruptSignal.ID)
	ID string

	// Address 层级地址
	Address Address

	// Info 中断信息
	Info any

	// IsRoot 是否为根因(叶子节点)
	IsRoot bool

	// Parent 父中断(CompositeInterrupt 的父节点)
	Parent *InterruptContext
}

InterruptContext 用户面向的中断上下文

从信号树提取的平面视图,便于用户列举和处理所有中断点。 每个 InterruptContext 对应信号树中的一个节点。

func ToInterruptContexts

func ToInterruptContexts(signal *InterruptSignal, filterTypes ...AddressSegmentType) []*InterruptContext

ToInterruptContexts 将信号树转换为用户面向的平面列表

递归遍历信号树,将每个节点转换为 InterruptContext。 可通过 filterTypes 只保留指定地址段类型的节点, 例如只看 agent+tool 层,隐藏 node/subgraph 的实现细节。

参数:

  • signal: 中断信号树
  • filterTypes: 地址段类型过滤器(为空则保留所有节点)

返回所有匹配的 InterruptContext 列表

type InterruptError

type InterruptError struct {
	ThreadID  string
	NodeID    string
	Payload   any
	Timestamp time.Time

	// Signal 关联的中断信号(新系统)
	// 当通过新的 InterruptSignalFunc/StatefulInterrupt/CompositeInterrupt 触发中断时,
	// 此字段会被填充,用于桥接新旧两套中断系统
	Signal *InterruptSignal
}

InterruptError 中断错误,携带中断信息

func (*InterruptError) Error

func (e *InterruptError) Error() string

func (*InterruptError) Is

func (e *InterruptError) Is(target error) bool

type InterruptOption

type InterruptOption func(*interruptConfig)

InterruptOption 中断选项

func WithDefault

func WithDefault[T any](v T) InterruptOption

WithDefault 设置默认值(超时时使用)

func WithTimeout

func WithTimeout(d time.Duration) InterruptOption

WithTimeout 设置等待超时

func WithValidator

func WithValidator[T any](fn func(T) error) InterruptOption

WithValidator 设置输入验证器

type InterruptSignal

type InterruptSignal struct {
	// ID 中断信号唯一标识
	ID string

	// Address 中断发生的层级地址
	Address Address

	// Info 面向用户的中断信息(如审核请求、确认提示等)
	Info any

	// State 组件内部状态(StatefulInterrupt 保存的进度/上下文)
	State any

	// Subs 子中断信号列表(CompositeInterrupt 的子节点)
	Subs []*InterruptSignal

	// IsRoot 是否为根因(叶子节点,即实际触发中断的点)
	IsRoot bool
}

InterruptSignal 中断信号,实现 error 接口以零侵入传播

中断信号是一棵树状结构:

  • 叶子节点(IsRoot=true)是实际触发中断的点
  • 非叶子节点聚合了多个子中断(CompositeInterrupt)

通过实现 error 接口,信号可以通过 Go 标准错误传播机制透传, 无需修改任何组件的接口签名

func IsInterruptSignal

func IsInterruptSignal(err error) (*InterruptSignal, bool)

IsInterruptSignal 从 error 中提取 InterruptSignal 使用 errors.As 进行解包,支持嵌套错误

func (*InterruptSignal) Error

func (s *InterruptSignal) Error() string

Error 实现 error 接口

func (*InterruptSignal) Unwrap

func (s *InterruptSignal) Unwrap() error

Unwrap 支持 errors.Is/As 解包 如果有子中断信号,返回第一个子信号

type PendingInfo

type PendingInfo struct {
	ThreadID  string
	NodeID    string
	Payload   any
	Timestamp time.Time
}

PendingInfo 待处理中断信息

type Status

type Status string

Status 检查点状态

const (
	StatusRunning     Status = "running"
	StatusInterrupted Status = "interrupted"
	StatusResumed     Status = "resumed"
	StatusCompleted   Status = "completed"
	StatusFailed      Status = "failed"
)

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL