Retracted
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 ¶
- Variables
- func AppendAddressSegment(ctx context.Context, segType AddressSegmentType, id, subID string) context.Context
- func BatchResumeWithData(ctx context.Context, resumeData map[string]any) context.Context
- func CompositeInterrupt(ctx context.Context, info any, state any, subErrors ...error) error
- func ContextWithInterruptHandler(ctx context.Context, handler *Handler) context.Context
- func ContextWithNodeID(ctx context.Context, nodeID string) context.Context
- func ContextWithThreadID(ctx context.Context, threadID string) context.Context
- func GetInterruptState[T any](ctx context.Context) (wasInterrupted bool, hasState bool, state T)
- func GetResumeContext[T any](ctx context.Context) (isResumeTarget bool, hasData bool, data T)
- func Interrupt[T any](ctx context.Context, payload any) (T, error)
- func InterruptForApproval(ctx context.Context, content string) (bool, error)
- func InterruptForChoice(ctx context.Context, question string, options []string) (string, error)
- func InterruptForInput(ctx context.Context, prompt string) (string, error)
- func InterruptSignalFunc(ctx context.Context, info any) error
- func InterruptWithOptions[T any](ctx context.Context, payload any, opts ...InterruptOption) (T, error)
- func MustInterrupt[T any](ctx context.Context, payload any) T
- func NodeIDFromContext(ctx context.Context) string
- func PopulateResumeInfo(ctx context.Context, id2Addr map[string]Address, id2State map[string]any) context.Context
- func Resume(ctx context.Context, interruptIDs ...string) context.Context
- func ResumeWithData(ctx context.Context, interruptID string, data any) context.Context
- func SignalToPersistenceMaps(signal *InterruptSignal) (id2Addr map[string]Address, id2State map[string]any)
- func StatefulInterrupt(ctx context.Context, info any, state any) error
- func ThreadIDFromContext(ctx context.Context) string
- type Address
- type AddressSegment
- type AddressSegmentType
- type Checkpoint
- type Command
- type Handler
- type InterruptContext
- type InterruptError
- type InterruptOption
- type InterruptSignal
- type PendingInfo
- type Status
Constants ¶
This section is empty.
Variables ¶
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 ¶
BatchResumeWithData 批量恢复多个中断点
适用于 CompositeInterrupt 场景：多个子中断需要同时恢复。 resumeData: map[interruptID]data
func CompositeInterrupt ¶
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 ¶
ContextWithInterruptHandler 添加中断处理器到 context
func ContextWithNodeID ¶
ContextWithNodeID 添加节点 ID 到 context
func ContextWithThreadID ¶
ContextWithThreadID 添加线程 ID 到 context
func GetInterruptState ¶
GetInterruptState 获取中断时保存的组件状态
用于 StatefulInterrupt 场景：组件在中断前保存了内部状态（如处理进度）， 恢复时通过此函数获取，跳过已完成的工作。
返回值：
- wasInterrupted: 当前组件是否曾参与中断（地址匹配到了中断记录）
- hasState: 是否有保存的状态且类型匹配
- state: 类型安全的状态对象
func GetResumeContext ¶
GetResumeContext 获取恢复上下文
用于组件内部判断当前是否为恢复目标，以及获取恢复时携带的数据。
返回值：
- isResumeTarget: 当前组件（或其后代）是否为恢复目标
- hasData: 是否携带了恢复数据且类型匹配
- data: 类型安全的恢复数据
func Interrupt ¶
Interrupt 在当前节点触发中断，等待人工输入
调用此函数会： 1. 保存当前状态到检查点 2. 返回 InterruptError 3. 等待 Resume 恢复
参数：
- ctx: 上下文，必须包含 Handler
- payload: 中断时携带的数据，会传递给恢复方
返回：
- T: 恢复时传入的数据
- error: 错误
func InterruptForApproval ¶
InterruptForApproval 等待审批
func InterruptForChoice ¶
InterruptForChoice 等待用户选择
func InterruptForInput ¶
InterruptForInput 等待用户输入
func InterruptSignalFunc ¶
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 ¶
MustInterrupt 简化版中断，panic on error
⚠️ 警告：中断失败时会 panic。 仅在确定中断一定成功时使用。 推荐使用 Interrupt[T]() 方法并正确处理错误。
使用场景：
- 测试代码中
- 确定上下文已正确设置的场景
func NodeIDFromContext ¶
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 ¶
Resume 标记中断点为已恢复（不携带数据）
对指定的中断 ID，将恢复数据设为 nil 并标记为恢复目标。 执行时 AppendAddressSegment 会检测到恢复标记， 使 GetResumeContext 返回 isResumeTarget=true。
func ResumeWithData ¶
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 ¶
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 ¶
ThreadIDFromContext 从 context 获取线程 ID
Types ¶
type Address ¶
type Address []AddressSegment
Address 层级地址，由多个段组成 例如: [node:step1, tool:search, tool:search:call_1] 表示在 step1 节点中，search 工具的 call_1 调用
func GetCurrentAddress ¶
GetCurrentAddress 从 context 获取当前地址
func (Address) Append ¶
func (a Address) Append(seg AddressSegment) Address
Append 追加一个段，返回新地址（不修改原地址）
func (Address) IsDescendantOf ¶
IsDescendantOf 判断当前地址是否是 ancestor 的后代 即 ancestor 是当前地址的前缀
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 ¶
HandlerFromContext 从 context 获取中断处理器
func (*Handler) GetPending ¶
func (h *Handler) GetPending(threadID string) *PendingInfo
GetPending 获取待处理的中断
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 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) Unwrap ¶
func (s *InterruptSignal) Unwrap() error
Unwrap 支持 errors.Is/As 解包 如果有子中断信号，返回第一个子信号
type PendingInfo ¶
PendingInfo 待处理中断信息
Source Files