Documentation
¶
Overview ¶
Package graph 提供 Hexagon AI Agent 框架的图编排引擎
barrier.go 实现延迟/屏障节点 (Barrier/Join Node):
- BarrierNode: 等待所有指定的上游并行分支完成后再继续
- MapReduceNode: 将数据分片并行处理后聚合结果
- FanOutFanIn: 扇出扇入模式,自动并行执行后汇聚
对标 LangGraph 的 map-reduce 和 barrier 模式。
使用示例:
// 方式 1: Barrier 等待多个分支
graph := NewGraph[MyState]("pipeline").
AddNode("step_a", handlerA).
AddNode("step_b", handlerB).
AddBarrier("join", mergeFunc, "step_a", "step_b").
Build()
// 方式 2: MapReduce 模式
graph := NewGraph[MyState]("mr").
AddMapReduce("process", splitFunc, mapFunc, reduceFunc).
Build()
Package graph 提供 Hexagon AI Agent 框架的图编排引擎 ¶
cache.go 实现节点级缓存功能,参考 LangGraph 的 Node-level Caching:
- 基于节点输入的哈希值缓存执行结果
- 避免重复执行相同输入的节点(如重复的 LLM 调用)
- 支持内存缓存和自定义缓存后端
- 支持 TTL 过期和容量限制
使用示例:
graph := NewGraph[MyState]("my-graph").
AddNode("llm_call", handler).
WithNodeCache("llm_call", NewMemoryNodeCache(
WithCacheTTL(5 * time.Minute),
WithCacheCapacity(100),
)).
Build()
Package graph 提供图编排引擎 ¶
本文件实现增强的检查点系统,对标 LangGraph:
- 分支执行支持
- 版本管理
- 状态差异追踪
- 从任意检查点恢复
- 检查点配置
Package graph 提供图编排引擎 ¶
本文件实现检查点恢复执行器:
- 从任意检查点恢复执行
- 支持断点续跑
- 支持分支执行
Package graph 提供 Hexagon AI Agent 框架的图编排引擎 ¶
command.go 实现 Command API(状态更新+节点跳转合一):
- Command: 统一的命令对象,同时携带状态更新和路由指令
- CommandHandler: 返回 Command 的节点处理函数
- CommandNode: 使用 Command API 的节点
对标 LangGraph 的 Command API,让常见的"更新状态并跳转到下一节点"操作更简洁。
使用示例:
// 传统方式需要分别设置状态和条件路由
// Command API 方式:一步到位
graph := NewGraph[MyState]("flow").
AddCommandNode("classify", func(ctx context.Context, state MyState) (*Command[MyState], error) {
if state.IsUrgent {
return Goto[MyState]("urgent_handler").
WithUpdate(func(s MyState) MyState { s.Priority = "high"; return s }),
nil
}
return Goto[MyState]("normal_handler").
WithUpdate(func(s MyState) MyState { s.Priority = "normal"; return s }),
nil
}).
Build()
Package graph 提供 Hexagon AI Agent 框架的图编排引擎 ¶
distributed.go 实现分布式图执行支持:
- RemoteNodeExecutor: 远程节点执行器接口
- HTTPNodeExecutor: 基于 HTTP 的远程节点执行
- DistributedGraph: 支持节点分布在不同机器上执行
- NodePlacement: 节点放置策略
对标 LangGraph Cloud 的分布式调度能力。
使用示例:
// 注册远程执行器
registry := NewRemoteRegistry()
registry.Register("gpu-node", NewHTTPNodeExecutor("http://gpu-server:8080"))
// 配置节点分布
graph := NewGraph[MyState]("distributed-flow").
AddNode("preprocess", preprocessHandler).
AddNode("inference", inferenceHandler).
WithNodePlacement("inference", "gpu-node").
Build()
result, err := graph.RunDistributed(ctx, state, registry)
Package graph 提供 Hexagon AI Agent 框架的图编排引擎 ¶
functional.go 实现 Functional API(函数式工作流定义):
- Entrypoint: 标记工作流入口函数
- Task: 定义可被编排的异步任务
- Workflow: 通过函数注册自动构建图
对标 LangGraph 的 @entrypoint/@task 装饰器模式。 Go 没有装饰器语法,采用注册器+泛型实现同等能力。
使用示例:
wf := NewWorkflow[MyState]("my-flow")
// 定义任务
fetchTask := DefineTask(wf, "fetch", fetchFunc)
processTask := DefineTask(wf, "process", processFunc)
// 定义入口点:编排任务执行顺序
DefineEntrypoint(wf, func(ctx context.Context, state MyState) (MyState, error) {
state, err := fetchTask.Run(ctx, state)
if err != nil { return state, err }
return processTask.Run(ctx, state)
})
result, err := wf.Run(ctx, initialState)
Package graph 提供 Hexagon AI Agent 框架的图编排引擎 ¶
Graph 编排引擎允许将多个节点(Agent、Tool、函数)组织成有向图, 支持条件路由、并行执行、检查点恢复等高级特性。
基本用法:
graph := NewGraph[MyState]("my-graph").
AddNode("step1", step1Handler).
AddNode("step2", step2Handler).
AddEdge(START, "step1").
AddEdge("step1", "step2").
AddEdge("step2", END).
Build()
result, err := graph.Run(ctx, initialState)
Package graph 提供图编排引擎 ¶
本文件实现增强的 Human-in-the-Loop (HITL) 功能:
- 人工审批:关键操作前等待人工确认
- 人工输入:请求人工提供额外输入
- 人工校验:人工校验 LLM 输出
- 人工接管:在特定条件下切换为人工操作
设计借鉴:
- LangGraph: Human-in-the-loop
- Mastra: Human 节点
- CrewAI: 人工协作模式
Package graph 提供 Hexagon AI Agent 框架的图编排引擎 ¶
本文件实现循环图支持:
- WhileLoop: while 循环
- DoWhile: do-while 循环
- ForLoop: for 循环(有限次数)
- ForEach: 遍历集合
- Until: 条件退出循环
- LoopWithBreak: 带中断的循环
设计借鉴:
- LangGraph: 条件循环
- Mastra: 工作流循环
- BPMN: 循环网关
Package graph 提供 Hexagon AI Agent 框架的图编排引擎 ¶
Package graph 提供 Hexagon 框架的图编排能力 ¶
本文件实现状态机模式,对标 LangGraph 的状态机设计。
设计借鉴:
- LangGraph: StateGraph 状态机
- 传统状态机: FSM (Finite State Machine)
使用示例:
sm := graph.NewStateMachine[*MyState]().
State("init", initHandler).
State("process", processHandler).
State("review", reviewHandler).
Transition("init", "process", alwaysTrue).
Transition("process", "review", needsReview).
Transition("review", "done", approved).
Initial("init").
Final("done").
Build()
result, err := sm.Run(ctx, initialState)
Package graph 提供 Hexagon AI Agent 框架的图编排引擎 ¶
stream_mode.go 实现多种流式输出模式,参考 LangGraph 的设计:
- StreamModeValues: 每步输出完整状态快照
- StreamModeUpdates: 每步输出状态增量变化
- StreamModeMessages: 输出 LLM 生成的 token 流
- StreamModeCustom: 用户自定义事件流
- StreamModeDebug: 详细调试信息流
使用示例:
stream, err := graph.StreamRun(ctx, state,
WithStreamMode(StreamModeUpdates),
)
for event := range stream.Events() {
fmt.Println(event.Node, event.Data)
}
Package graph 提供图编排引擎 ¶
本文件实现时间旅行调试功能:
- 状态快照:保存每一步执行的状态
- 时间回溯:回到任意历史状态
- 状态重放:从历史状态继续执行
- 差异对比:对比不同时间点的状态
设计借鉴:
- LangGraph: Time Travel 功能
- Redux DevTools: 时间旅行调试
- Event Sourcing: 事件溯源模式
Package graph 提供 Hexagon AI Agent 框架的图编排引擎 ¶
visualize.go 实现图结构的可视化导出功能,支持:
- Mermaid: 适用于 Markdown 文档和在线渲染
- DOT (Graphviz): 适用于生成高质量图片
- ASCII: 适用于终端输出
Index ¶
- Constants
- Variables
- func Always[S any]() func(context.Context, S) bool
- func Break() error
- func ComputeCacheKey(nodeName string, state any) string
- func Conditional[S any](condition func(S) bool, ifTrue, ifFalse string) func(context.Context, S) (string, error)
- func ContextWithNodeError(ctx context.Context, err error) context.Context
- func Continue() error
- func DefineEntrypoint[S State](wf *Workflow[S], handler func(ctx context.Context, state S) (S, error))
- func EmitCustomEvent(ctx context.Context, name string, data any)
- func End[S any](nextState string) func(context.Context, S) (string, error)
- func Never[S any]() func(context.Context, S) bool
- func NodeErrorFromContext(ctx context.Context) error
- func PassThrough[S any]() func(context.Context, S) (string, error)
- func RunConditional[S State](ctx context.Context, state S, condition func(S) string, ...) (S, error)
- func RunParallel[S State](ctx context.Context, state S, merger func(original S, results []S) S, ...) (S, error)
- func SelectFirst[S State](state S, branches ...BranchConfig[S]) string
- func When[S any](predicate func(S) bool) func(context.Context, S) bool
- func WithStreamChannel(ctx context.Context, ch *StreamChannel) context.Context
- type AppendReducer
- type BarrierMerger
- type BranchConfig
- type BranchInfo
- type BulkheadConfig
- type CacheStats
- type Channel
- type ChannelConfig
- type ChannelHITLHandler
- type Checkpoint
- type CheckpointQuery
- type CheckpointRunner
- func (r *CheckpointRunner[S]) Fork(ctx context.Context, checkpointID string, branchName string, ...) (S, error)
- func (r *CheckpointRunner[S]) GetCurrentCheckpoint() *EnhancedCheckpoint
- func (r *CheckpointRunner[S]) GetHistory(ctx context.Context, limit int) ([]*EnhancedCheckpoint, error)
- func (r *CheckpointRunner[S]) Resume(ctx context.Context, checkpointID string) (S, error)
- func (r *CheckpointRunner[S]) ResumeFromLatest(ctx context.Context, threadID string) (S, error)
- func (r *CheckpointRunner[S]) Run(ctx context.Context, threadID string, initialState S) (S, error)
- type CheckpointRunnerConfig
- type CheckpointSaver
- type CheckpointStats
- type CheckpointStatus
- type CheckpointVersion
- type CircuitBreakerConfig
- type CleanupPolicy
- type Command
- func Goto[S State](target string) *Command[S]
- func GotoEnd[S State]() *Command[S]
- func GotoIf[S State](condition bool, ifTrue, ifFalse string) *Command[S]
- func GotoSwitch[S State](label string, routes map[string]string) *Command[S]
- func UpdateAndEnd[S State](update func(S) S) *Command[S]
- func UpdateAndGoto[S State](target string, update func(S) S) *Command[S]
- func (c *Command[S]) ApplyUpdates(state S) S
- func (c *Command[S]) Sends() []Send
- func (c *Command[S]) Target() string
- func (c *Command[S]) WithMetadata(key string, value any) *Command[S]
- func (c *Command[S]) WithSend(sends ...Send) *Command[S]
- func (c *Command[S]) WithState(state S) *Command[S]
- func (c *Command[S]) WithUpdate(update func(S) S) *Command[S]
- type CommandHandler
- type CompiledGraph
- type ConditionalHandler
- type DebugView
- type DefaultPregelMerger
- type DynamicGraph
- func (g *DynamicGraph[S]) AddEdgeDynamic(from, to string) error
- func (g *DynamicGraph[S]) AddNodeDynamic(name string, handler NodeHandler[S]) error
- func (g *DynamicGraph[S]) Build() (*DynamicGraph[S], error)
- func (g *DynamicGraph[S]) OnModified(callback func(g *DynamicGraph[S]))
- func (g *DynamicGraph[S]) RemoveEdgeDynamic(from, to string) error
- func (g *DynamicGraph[S]) RemoveNodeDynamic(name string) error
- func (g *DynamicGraph[S]) ReplaceNodeHandler(name string, handler NodeHandler[S]) error
- func (g *DynamicGraph[S]) Snapshot() *GraphSnapshot[S]
- func (g *DynamicGraph[S]) Version() int64
- type Edge
- type EdgeBuilder
- type EdgeSnapshot
- type EdgeType
- type EnhancedCheckpoint
- type EnhancedCheckpointSaver
- type ErrorHandler
- type EventType
- type Executable
- type ExecutionPlan
- type ExecutionResult
- type ExecutionStats
- type ExecutionTrace
- type ExportFormat
- type ExportOption
- type FieldValidation
- type FileCheckpointSaver
- func (s *FileCheckpointSaver) Delete(ctx context.Context, id string) error
- func (s *FileCheckpointSaver) DeleteThread(ctx context.Context, threadID string) error
- func (s *FileCheckpointSaver) List(ctx context.Context, threadID string) ([]*Checkpoint, error)
- func (s *FileCheckpointSaver) Load(ctx context.Context, threadID string) (*Checkpoint, error)
- func (s *FileCheckpointSaver) LoadByID(ctx context.Context, id string) (*Checkpoint, error)
- func (s *FileCheckpointSaver) Save(ctx context.Context, checkpoint *Checkpoint) error
- type Graph
- func (g *Graph[S]) Export(format ExportFormat, opts ...ExportOption) string
- func (g *Graph[S]) GetNodeCache(nodeName string) NodeCache
- func (g *Graph[S]) GetNodePlacements() []NodePlacement
- func (g *Graph[S]) Run(ctx context.Context, initialState S, opts ...RunOption) (S, error)
- func (g *Graph[S]) RunDistributed(ctx context.Context, initialState S, registry *RemoteRegistry, ...) (S, error)
- func (g *Graph[S]) RunPregelMode(ctx context.Context, initialState S, opts ...PregelOption) (S, int, error)
- func (g *Graph[S]) Stream(ctx context.Context, initialState S, opts ...RunOption) (<-chan StreamEvent[S], error)
- func (g *Graph[S]) StreamPregelMode(ctx context.Context, initialState S, opts ...PregelOption) (<-chan PregelEvent[S], error)
- func (g *Graph[S]) StreamRun(ctx context.Context, state S, opts ...StreamRunOption) (*StreamChannel, error)
- type GraphBuilder
- func (b *GraphBuilder[S]) AddBarrier(name string, merger BarrierMerger[S], waitFor ...string) *GraphBuilder[S]
- func (b *GraphBuilder[S]) AddCommandNode(name string, handler CommandHandler[S]) *GraphBuilder[S]
- func (b *GraphBuilder[S]) AddConditionalEdge(from string, router RouterFunc[S], edges map[string]string) *GraphBuilder[S]
- func (b *GraphBuilder[S]) AddDoWhileLoop(name string, condition func(S) bool, body NodeHandler[S], ...) *GraphBuilder[S]
- func (b *GraphBuilder[S]) AddEdge(from, to string) *GraphBuilder[S]
- func (b *GraphBuilder[S]) AddFanOutFanIn(name string, branches map[string]NodeHandler[S], merger BarrierMerger[S]) *GraphBuilder[S]
- func (b *GraphBuilder[S]) AddForLoop(name string, iterations int, ...) *GraphBuilder[S]
- func (b *GraphBuilder[S]) AddLoopBackEdge(from, to string, condition func(S) bool, maxIterations int) *GraphBuilder[S]
- func (b *GraphBuilder[S]) AddMapReduce(name string, split SplitFunc[S], mapFn MapFunc[S], reduce ReduceFunc[S], ...) *GraphBuilder[S]
- func (b *GraphBuilder[S]) AddNode(name string, handler NodeHandler[S]) *GraphBuilder[S]
- func (b *GraphBuilder[S]) AddNodeWithBuilder(node *Node[S]) *GraphBuilder[S]
- func (b *GraphBuilder[S]) AddRetryLoop(name string, body NodeHandler[S], config *RetryConfig) *GraphBuilder[S]
- func (b *GraphBuilder[S]) AddUntilLoop(name string, condition func(S) bool, body NodeHandler[S], ...) *GraphBuilder[S]
- func (b *GraphBuilder[S]) AddWhileLoop(name string, condition func(S) bool, body NodeHandler[S], ...) *GraphBuilder[S]
- func (b *GraphBuilder[S]) Build() (*Graph[S], error)
- func (b *GraphBuilder[S]) MustBuild() *Graph[S]
- func (b *GraphBuilder[S]) SetEntryPoint(node string) *GraphBuilder[S]
- func (b *GraphBuilder[S]) SetFinishPoint(nodes ...string) *GraphBuilder[S]
- func (b *GraphBuilder[S]) WithCheckpointer(saver CheckpointSaver) *GraphBuilder[S]
- func (b *GraphBuilder[S]) WithMetadata(key string, value any) *GraphBuilder[S]
- func (b *GraphBuilder[S]) WithNodeCache(nodeName string, cache NodeCache) *GraphBuilder[S]
- func (b *GraphBuilder[S]) WithNodePlacement(nodeName, executorName string) *GraphBuilder[S]
- func (b *GraphBuilder[S]) WithNodePlacementNoFallback(nodeName, executorName string) *GraphBuilder[S]
- type GraphComposer
- type GraphSnapshot
- type HITLCallback
- type HITLHandler
- type HITLManager
- func (m *HITLManager) Cancel(requestID string) error
- func (m *HITLManager) Export() ([]byte, error)
- func (m *HITLManager) GetActiveRequests() []*HITLRequest
- func (m *HITLManager) GetHistory(limit int) []*HITLRecord
- func (m *HITLManager) GetStats() HITLStats
- func (m *HITLManager) Submit(ctx context.Context, request *HITLRequest) (*HITLResponse, error)
- type HITLManagerConfig
- type HITLNode
- func NewApprovalNode[S State](id string, handler HITLHandler, opts ...HITLNodeOption[S]) *HITLNode[S]
- func NewInputNode[S State](id string, handler HITLHandler, schema map[string]any, ...) *HITLNode[S]
- func NewReviewNode[S State](id string, handler HITLHandler, opts ...HITLNodeOption[S]) *HITLNode[S]
- type HITLNodeOption
- func WithHITLCondition[S State](condition func(state S) bool) HITLNodeOption[S]
- func WithHITLDescription[S State](desc string) HITLNodeOption[S]
- func WithHITLPriority[S State](priority HITLPriority) HITLNodeOption[S]
- func WithHITLTimeout[S State](timeout time.Duration) HITLNodeOption[S]
- func WithHITLTitle[S State](title string) HITLNodeOption[S]
- type HITLOption
- type HITLPriority
- type HITLRecord
- type HITLRequest
- type HITLResponse
- type HITLStats
- type HITLType
- type HTTPExecutorOption
- type HTTPNodeExecutor
- type HumanInTheLoop
- func (h *HumanInTheLoop[S]) Resume(ctx context.Context, threadID string, response *InterruptResponse) (S, *Interrupt, error)
- func (h *HumanInTheLoop[S]) RunWithInterrupt(ctx context.Context, threadID string, initialState S) (S, *Interrupt, error)
- func (h *HumanInTheLoop[S]) WaitAndResume(ctx context.Context, threadID string, interruptID string) (S, error)
- type InputField
- type InputSchema
- type Interrupt
- type InterruptBuilder
- func (b *InterruptBuilder) Build() *Interrupt
- func (b *InterruptBuilder) WithData(key string, value any) *InterruptBuilder
- func (b *InterruptBuilder) WithInputSchema(schema *InputSchema) *InterruptBuilder
- func (b *InterruptBuilder) WithMessage(message string) *InterruptBuilder
- func (b *InterruptBuilder) WithOptions(options ...InterruptOption) *InterruptBuilder
- func (b *InterruptBuilder) WithTimeout(timeout time.Duration) *InterruptBuilder
- func (b *InterruptBuilder) WithTitle(title string) *InterruptBuilder
- func (b *InterruptBuilder) WithType(t InterruptType) *InterruptBuilder
- type InterruptConfig
- type InterruptError
- type InterruptHandler
- type InterruptOption
- type InterruptResponse
- type InterruptStatus
- type InterruptType
- type ListOptions
- type LoopConfig
- type LoopCounter
- type MapFunc
- type MapState
- type MemoryCacheOption
- type MemoryCheckpointSaver
- func (s *MemoryCheckpointSaver) Delete(ctx context.Context, id string) error
- func (s *MemoryCheckpointSaver) DeleteThread(ctx context.Context, threadID string) error
- func (s *MemoryCheckpointSaver) List(ctx context.Context, threadID string) ([]*Checkpoint, error)
- func (s *MemoryCheckpointSaver) Load(ctx context.Context, threadID string) (*Checkpoint, error)
- func (s *MemoryCheckpointSaver) LoadByID(ctx context.Context, id string) (*Checkpoint, error)
- func (s *MemoryCheckpointSaver) Save(ctx context.Context, checkpoint *Checkpoint) error
- type MemoryEnhancedCheckpointSaver
- func (s *MemoryEnhancedCheckpointSaver) Cleanup(ctx context.Context, policy *CleanupPolicy) (int, error)
- func (s *MemoryEnhancedCheckpointSaver) CreateBranch(ctx context.Context, checkpointID string, branchName string) (*EnhancedCheckpoint, error)
- func (s *MemoryEnhancedCheckpointSaver) GetBranches(ctx context.Context, threadID string) ([]*BranchInfo, error)
- func (s *MemoryEnhancedCheckpointSaver) GetHistory(ctx context.Context, checkpointID string, limit int) ([]*EnhancedCheckpoint, error)
- func (s *MemoryEnhancedCheckpointSaver) ListEnhanced(ctx context.Context, threadID string, opts *ListOptions) ([]*EnhancedCheckpoint, error)
- func (s *MemoryEnhancedCheckpointSaver) LoadEnhanced(ctx context.Context, threadID string) (*EnhancedCheckpoint, error)
- func (s *MemoryEnhancedCheckpointSaver) LoadEnhancedByID(ctx context.Context, id string) (*EnhancedCheckpoint, error)
- func (s *MemoryEnhancedCheckpointSaver) MergeBranch(ctx context.Context, sourceBranchID, targetBranchID string, ...) (*EnhancedCheckpoint, error)
- func (s *MemoryEnhancedCheckpointSaver) SaveEnhanced(ctx context.Context, checkpoint *EnhancedCheckpoint) error
- func (s *MemoryEnhancedCheckpointSaver) Search(ctx context.Context, query *CheckpointQuery) ([]*EnhancedCheckpoint, error)
- type MemoryInterruptHandler
- func (h *MemoryInterruptHandler) Cancel(ctx context.Context, id string) error
- func (h *MemoryInterruptHandler) Create(ctx context.Context, interrupt *Interrupt) error
- func (h *MemoryInterruptHandler) Get(ctx context.Context, id string) (*Interrupt, error)
- func (h *MemoryInterruptHandler) List(ctx context.Context, threadID string) ([]*Interrupt, error)
- func (h *MemoryInterruptHandler) ListPending(ctx context.Context) ([]*Interrupt, error)
- func (h *MemoryInterruptHandler) Resolve(ctx context.Context, id string, response *InterruptResponse, resolvedBy string) error
- func (h *MemoryInterruptHandler) Wait(ctx context.Context, id string) (*Interrupt, error)
- func (h *MemoryInterruptHandler) WaitWithTimeout(ctx context.Context, id string, timeout time.Duration) (*Interrupt, error)
- type MemoryNodeCache
- type MemorySnapshotStorage
- func (s *MemorySnapshotStorage) Clear(ctx context.Context) error
- func (s *MemorySnapshotStorage) Delete(ctx context.Context, index int) error
- func (s *MemorySnapshotStorage) Load(ctx context.Context, index int) (*StateSnapshot, error)
- func (s *MemorySnapshotStorage) LoadRange(ctx context.Context, start, end int) ([]*StateSnapshot, error)
- func (s *MemorySnapshotStorage) Save(ctx context.Context, snapshot *StateSnapshot) error
- type MergeStrategy
- type MultiRouter
- type MultiRouterFunc
- type Node
- func BarrierNode[S State](name string, merger BarrierMerger[S], waitFor ...string) *Node[S]
- func BranchNode[S State](name string, branches map[string]*Graph[S], selector func(S) string) *Node[S]
- func BulkheadNode[S State](name string, handler NodeHandler[S], config *BulkheadConfig) *Node[S]
- func CatchNode[S State](name string, handler ErrorHandler[S], errorTypes ...error) *Node[S]
- func CircuitBreakerNode[S State](name string, handler NodeHandler[S], config *CircuitBreakerConfig) *Node[S]
- func ConditionalNode[S State](name string, router ConditionalHandler[S]) *Node[S]
- func ConditionalSubgraph[S State](name string, selector func(S) int, subgraphs ...*Graph[S]) *Node[S]
- func DoWhileLoopNode[S State](name string, condition func(S) bool, body NodeHandler[S], ...) *Node[S]
- func EndNode[S State]() *Node[S]
- func FallbackNode[S State](name string, primaryHandler, fallbackHandler NodeHandler[S]) *Node[S]
- func FanOutFanInNode[S State](name string, branches map[string]NodeHandler[S], merger BarrierMerger[S]) *Node[S]
- func ForEachLoopNode[S State, T any](name string, getItems func(S) []T, ...) *Node[S]
- func ForLoopNode[S State](name string, iterations int, ...) *Node[S]
- func LoopSubgraph[S State](name string, subgraph *Graph[S], condition func(S, int) bool, ...) *Node[S]
- func MapReduceNode[S State](name string, split SplitFunc[S], mapFn MapFunc[S], reduce ReduceFunc[S], ...) *Node[S]
- func ParallelForEachLoopNode[S State, T any](name string, getItems func(S) []T, ...) *Node[S]
- func ParallelNode[S State](name string, handlers ...NodeHandler[S]) *Node[S]
- func ParallelNodeWithMerger[S State](name string, merger StateMerger[S], handlers ...NodeHandler[S]) *Node[S]
- func ParallelSubgraphs[S State](name string, subgraphs []*Graph[S], merger StateMerger[S]) *Node[S]
- func RetryLoopNode[S State](name string, body NodeHandler[S], config *RetryConfig) *Node[S]
- func RetryNode[S State](name string, handler NodeHandler[S], policy *RetryPolicy) *Node[S]
- func StartNode[S State]() *Node[S]
- func SubgraphNode[S State](name string, subgraph *Graph[S], stateMapper ...*SubgraphStateMapper[S]) *Node[S]
- func TimeoutNode[S State](name string, handler NodeHandler[S], timeoutMs int64) *Node[S]
- func UntilLoopNode[S State](name string, condition func(S) bool, body NodeHandler[S], ...) *Node[S]
- func WhileLoopNode[S State](name string, condition func(S) bool, body NodeHandler[S], ...) *Node[S]
- type NodeBuilder
- func (b *NodeBuilder[S]) Build() *Node[S]
- func (b *NodeBuilder[S]) WithMetadata(key string, value any) *NodeBuilder[S]
- func (b *NodeBuilder[S]) WithRetry(policy *RetryPolicy) *NodeBuilder[S]
- func (b *NodeBuilder[S]) WithTimeout(ms int64) *NodeBuilder[S]
- func (b *NodeBuilder[S]) WithType(t NodeType) *NodeBuilder[S]
- type NodeCache
- type NodeHandler
- type NodePlacement
- type NodeResult
- type NodeStats
- type NodeType
- type OverwriteReducer
- type PregelConfig
- type PregelEvent
- type PregelEventType
- type PregelExecutor
- type PregelExecutorOption
- type PregelOption
- type PregelStateMerger
- type RedisCheckpointOption
- type RedisCheckpointSaver
- func (s *RedisCheckpointSaver) Close() error
- func (s *RedisCheckpointSaver) Delete(ctx context.Context, id string) error
- func (s *RedisCheckpointSaver) DeleteThread(ctx context.Context, threadID string) error
- func (s *RedisCheckpointSaver) GetCheckpointCount(ctx context.Context, threadID string) (int64, error)
- func (s *RedisCheckpointSaver) List(ctx context.Context, threadID string) ([]*Checkpoint, error)
- func (s *RedisCheckpointSaver) ListThreads(ctx context.Context, pattern string, limit int64) ([]string, error)
- func (s *RedisCheckpointSaver) Load(ctx context.Context, threadID string) (*Checkpoint, error)
- func (s *RedisCheckpointSaver) LoadByID(ctx context.Context, id string) (*Checkpoint, error)
- func (s *RedisCheckpointSaver) LoadByThreadIDWithWarnings(ctx context.Context, threadID string) ([]*Checkpoint, []error, error)
- func (s *RedisCheckpointSaver) Prune(ctx context.Context, threadID string, keepCount int64) error
- func (s *RedisCheckpointSaver) Save(ctx context.Context, checkpoint *Checkpoint) error
- type ReduceFunc
- type Reducer
- type RemoteNodeExecutor
- type RemoteRegistry
- type RetryConfig
- type RetryPolicy
- type Router
- type RouterFunc
- type RunOption
- type Send
- type SnapshotStorage
- type SplitFunc
- type State
- type StateAnnotation
- type StateDiff
- type StateMachine
- func (sm *StateMachine[S]) AddFinal(names ...string) *StateMachine[S]
- func (sm *StateMachine[S]) AddState(name string, handler func(ctx context.Context, state S) (string, error)) *StateMachine[S]
- func (sm *StateMachine[S]) AddStateWithHooks(name string, handler func(ctx context.Context, state S) (string, error), ...) *StateMachine[S]
- func (sm *StateMachine[S]) AddTransition(from, to string, condition func(ctx context.Context, state S) bool) *StateMachine[S]
- func (sm *StateMachine[S]) AddTransitionWithPriority(from, to string, condition func(ctx context.Context, state S) bool, ...) *StateMachine[S]
- func (sm *StateMachine[S]) GetPending(threadID string) *interrupt.PendingInfo
- func (sm *StateMachine[S]) Resume(ctx context.Context, threadID string, cmd interrupt.Command) error
- func (sm *StateMachine[S]) Run(ctx context.Context, initialState S) (S, error)
- func (sm *StateMachine[S]) RunWithThreadID(ctx context.Context, threadID string, initialState S) (S, error)
- func (sm *StateMachine[S]) RunWithTrace(ctx context.Context, initialState S) (S, *ExecutionTrace, error)
- func (sm *StateMachine[S]) SetCheckpointer(cp checkpoint.Checkpointer) *StateMachine[S]
- func (sm *StateMachine[S]) SetInitial(name string) *StateMachine[S]
- func (sm *StateMachine[S]) SetMaxSteps(max int) *StateMachine[S]
- type StateMachineBuilder
- func (b *StateMachineBuilder[S]) Build() *StateMachine[S]
- func (b *StateMachineBuilder[S]) Checkpointer(cp checkpoint.Checkpointer) *StateMachineBuilder[S]
- func (b *StateMachineBuilder[S]) Final(names ...string) *StateMachineBuilder[S]
- func (b *StateMachineBuilder[S]) Initial(name string) *StateMachineBuilder[S]
- func (b *StateMachineBuilder[S]) MaxSteps(max int) *StateMachineBuilder[S]
- func (b *StateMachineBuilder[S]) State(name string, handler func(ctx context.Context, state S) (string, error)) *StateMachineBuilder[S]
- func (b *StateMachineBuilder[S]) Transition(from, to string, condition func(ctx context.Context, state S) bool) *StateMachineBuilder[S]
- type StateMerger
- type StateNode
- type StateSnapshot
- type StreamChannel
- type StreamEvent
- type StreamMode
- type StreamModeEvent
- type StreamModeEventType
- type StreamRunOption
- type SubgraphStateMapper
- type TaskDef
- type TaskOption
- type ThreadConfig
- type TimeTravelDebugger
- func (d *TimeTravelDebugger) Clear()
- func (d *TimeTravelDebugger) Compare(index1, index2 int) ([]StateDiff, error)
- func (d *TimeTravelDebugger) CurrentIndex() int
- func (d *TimeTravelDebugger) Export() ([]byte, error)
- func (d *TimeTravelDebugger) FindByNodeID(nodeID string) []*StateSnapshot
- func (d *TimeTravelDebugger) FindByTimeRange(start, end time.Time) []*StateSnapshot
- func (d *TimeTravelDebugger) FindErrors() []*StateSnapshot
- func (d *TimeTravelDebugger) GetBranchHistory(branchID string) []*StateSnapshot
- func (d *TimeTravelDebugger) GetBranches() []string
- func (d *TimeTravelDebugger) GetDebugView() *DebugView
- func (d *TimeTravelDebugger) GetHistory() []*StateSnapshot
- func (d *TimeTravelDebugger) GetSnapshot(index int) (*StateSnapshot, error)
- func (d *TimeTravelDebugger) GoBack() error
- func (d *TimeTravelDebugger) GoForward() error
- func (d *TimeTravelDebugger) GoTo(index int) error
- func (d *TimeTravelDebugger) Import(data []byte) error
- func (d *TimeTravelDebugger) Replay(ctx context.Context) (map[string]any, error)
- func (d *TimeTravelDebugger) ReplayFrom(ctx context.Context, index int) (map[string]any, error)
- func (d *TimeTravelDebugger) Run(ctx context.Context, initialState map[string]any) (map[string]any, error)
- func (d *TimeTravelDebugger) SearchSnapshots(predicate func(*StateSnapshot) bool) []*StateSnapshot
- type TimeTravelOption
- type TraceStep
- type Transition
- type TriggerMode
- type Workflow
Constants ¶
const ( // START 起始节点名称 START = "__start__" // END 结束节点名称 END = "__end__" )
特殊节点名称常量
Variables ¶
var ( // ErrLoopBreak 循环中断 ErrLoopBreak = errors.New("loop break") // ErrLoopContinue 循环继续 ErrLoopContinue = errors.New("loop continue") // ErrMaxIterationsReached 达到最大迭代次数 ErrMaxIterationsReached = errors.New("max iterations reached") // ErrLoopTimeout 循环超时 ErrLoopTimeout = errors.New("loop timeout") )
var ( // ErrNoInitialState 没有初始状态 ErrNoInitialState = errors.New("no initial state defined") // ErrStateNotFound 状态未找到 ErrStateNotFound = errors.New("state not found") // ErrNoTransition 没有可用的转换 ErrNoTransition = errors.New("no valid transition from current state") // ErrMaxStepsExceeded 超过最大步数 ErrMaxStepsExceeded = errors.New("max steps exceeded") )
Functions ¶
func ComputeCacheKey ¶
ComputeCacheKey 计算节点缓存 key 基于节点名称和输入状态的 SHA256 哈希
func Conditional ¶
func Conditional[S any](condition func(S) bool, ifTrue, ifFalse string) func(context.Context, S) (string, error)
Conditional 条件处理器
func ContextWithNodeError ¶
ContextWithNodeError 将节点错误添加到 context
func DefineEntrypoint ¶
func DefineEntrypoint[S State](wf *Workflow[S], handler func(ctx context.Context, state S) (S, error))
DefineEntrypoint 定义工作流入口点 entrypoint 函数编排各任务的执行顺序和逻辑
func EmitCustomEvent ¶
EmitCustomEvent 在节点处理函数中发射自定义事件 需要通过 context 传递 StreamChannel
func NodeErrorFromContext ¶
NodeErrorFromContext 从 context 获取节点错误
func PassThrough ¶
PassThrough 直通处理器(不做任何处理)
func RunConditional ¶
func RunConditional[S State](ctx context.Context, state S, condition func(S) string, routes map[string]*TaskDef[S]) (S, error)
RunConditional 条件执行任务 根据 condition 返回值选择执行哪个任务
func RunParallel ¶
func RunParallel[S State](ctx context.Context, state S, merger func(original S, results []S) S, tasks ...*TaskDef[S]) (S, error)
RunParallel 并行执行多个任务 所有任务共享同一初始状态的克隆,结果通过 merger 合并
func SelectFirst ¶
func SelectFirst[S State](state S, branches ...BranchConfig[S]) string
SelectFirst 选择第一个满足条件的分支
func WithStreamChannel ¶
func WithStreamChannel(ctx context.Context, ch *StreamChannel) context.Context
WithStreamChannel 将 StreamChannel 注入到 context 中
Types ¶
type AppendReducer ¶
AppendReducer 追加式合并器 将新值追加到切片中
func (AppendReducer[S, V]) Reduce ¶
func (r AppendReducer[S, V]) Reduce(state S, key string, value V) S
type BarrierMerger ¶
BarrierMerger 屏障合并函数 接收原始状态和所有上游分支的输出状态,返回合并后的状态
type BranchConfig ¶
type BranchConfig[S State] struct { // Condition 分支条件 Condition func(S) bool // Target 目标节点 Target string }
BranchConfig 分支配置
type BranchInfo ¶
type BranchInfo struct {
ID string `json:"id"`
Name string `json:"name"`
ThreadID string `json:"thread_id"`
BaseCheckpointID string `json:"base_checkpoint_id"`
LatestCheckpointID string `json:"latest_checkpoint_id"`
CheckpointCount int `json:"checkpoint_count"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
BranchInfo 分支信息
type BulkheadConfig ¶
type BulkheadConfig struct {
// MaxConcurrent 最大并发数
MaxConcurrent int
// MaxWait 最大等待时间(毫秒)
MaxWait int64
}
BulkheadConfig 舱壁配置(并发隔离)
type CacheStats ¶
type CacheStats struct {
// Hits 命中次数
Hits int64 `json:"hits"`
// Misses 未命中次数
Misses int64 `json:"misses"`
// Size 当前缓存条目数
Size int `json:"size"`
// Evictions 驱逐次数
Evictions int64 `json:"evictions"`
}
CacheStats 缓存统计信息
type Channel ¶
type Channel[V any] struct { // Name 通道名称(对应状态字段) Name string // Default 默认值 Default V // Reducer 合并函数 Reducer func(current V, new V) V }
Channel 通道定义 用于定义状态中的特定字段如何被更新
func MessageChannel ¶
MessageChannel 消息通道(追加模式)
func (*Channel[V]) WithReducer ¶
WithReducer 设置自定义合并函数
type ChannelConfig ¶
type ChannelConfig struct {
// Type 通道类型: "overwrite", "append", "custom"
Type string `json:"type"`
// Default 默认值
Default any `json:"default,omitempty"`
}
ChannelConfig 通道配置
type ChannelHITLHandler ¶
type ChannelHITLHandler struct {
// contains filtered or unexported fields
}
ChannelHITLHandler 基于 channel 的处理器
func NewChannelHITLHandler ¶
func NewChannelHITLHandler(bufferSize int) *ChannelHITLHandler
NewChannelHITLHandler 创建 channel 处理器
func (*ChannelHITLHandler) GetRequests ¶
func (h *ChannelHITLHandler) GetRequests() <-chan *HITLRequest
GetRequests 获取请求 channel(供 UI 消费)
func (*ChannelHITLHandler) Handle ¶
func (h *ChannelHITLHandler) Handle(ctx context.Context, request *HITLRequest) (*HITLResponse, error)
Handle 实现 HITLHandler 接口
func (*ChannelHITLHandler) SubmitResponse ¶
func (h *ChannelHITLHandler) SubmitResponse(response *HITLResponse) error
SubmitResponse 提交响应
type Checkpoint ¶
type Checkpoint struct {
// ID 检查点 ID
ID string `json:"id"`
// ThreadID 线程 ID(用于区分不同的执行实例)
ThreadID string `json:"thread_id"`
// GraphName 图名称
GraphName string `json:"graph_name"`
// CurrentNode 当前节点
CurrentNode string `json:"current_node"`
// State 状态快照
State json.RawMessage `json:"state"`
// PendingNodes 待执行的节点
PendingNodes []string `json:"pending_nodes"`
// CompletedNodes 已完成的节点
CompletedNodes []string `json:"completed_nodes"`
// Metadata 元数据
Metadata map[string]any `json:"metadata,omitempty"`
// InterruptAddrs 中断点地址映射(中断 ID -> 序列化的地址)
// 用于持久化 InterruptSignal 树中各中断点的层级地址
InterruptAddrs map[string]json.RawMessage `json:"interrupt_addrs,omitempty"`
// InterruptStates 中断点状态映射(中断 ID -> 序列化的组件状态)
// 用于持久化 StatefulInterrupt 保存的组件内部状态
InterruptStates map[string]json.RawMessage `json:"interrupt_states,omitempty"`
// CreatedAt 创建时间
CreatedAt time.Time `json:"created_at"`
// UpdatedAt 更新时间
UpdatedAt time.Time `json:"updated_at"`
// ParentID 父检查点 ID(用于构建历史链)
ParentID string `json:"parent_id,omitempty"`
}
Checkpoint 检查点 用于保存图执行的中间状态
type CheckpointQuery ¶
type CheckpointQuery struct {
ThreadID string `json:"thread_id,omitempty"`
GraphName string `json:"graph_name,omitempty"`
Status CheckpointStatus `json:"status,omitempty"`
BranchID string `json:"branch_id,omitempty"`
Tags []string `json:"tags,omitempty"`
StartTime *time.Time `json:"start_time,omitempty"`
EndTime *time.Time `json:"end_time,omitempty"`
Metadata map[string]any `json:"metadata,omitempty"`
Limit int `json:"limit,omitempty"`
Offset int `json:"offset,omitempty"`
}
CheckpointQuery 检查点查询
type CheckpointRunner ¶
type CheckpointRunner[S State] struct { // contains filtered or unexported fields }
CheckpointRunner 检查点恢复执行器 支持从任意检查点恢复图执行
func NewCheckpointRunner ¶
func NewCheckpointRunner[S State](g *Graph[S], saver EnhancedCheckpointSaver, config *CheckpointRunnerConfig) *CheckpointRunner[S]
NewCheckpointRunner 创建检查点执行器
func (*CheckpointRunner[S]) Fork ¶
func (r *CheckpointRunner[S]) Fork(ctx context.Context, checkpointID string, branchName string, modifyState func(S) S) (S, error)
Fork 从检查点创建分支并执行
func (*CheckpointRunner[S]) GetCurrentCheckpoint ¶
func (r *CheckpointRunner[S]) GetCurrentCheckpoint() *EnhancedCheckpoint
GetCurrentCheckpoint 获取当前检查点
func (*CheckpointRunner[S]) GetHistory ¶
func (r *CheckpointRunner[S]) GetHistory(ctx context.Context, limit int) ([]*EnhancedCheckpoint, error)
GetHistory 获取执行历史
func (*CheckpointRunner[S]) Resume ¶
func (r *CheckpointRunner[S]) Resume(ctx context.Context, checkpointID string) (S, error)
Resume 从检查点恢复执行
func (*CheckpointRunner[S]) ResumeFromLatest ¶
func (r *CheckpointRunner[S]) ResumeFromLatest(ctx context.Context, threadID string) (S, error)
ResumeFromLatest 从最新检查点恢复
type CheckpointRunnerConfig ¶
type CheckpointRunnerConfig struct {
// AutoSave 是否自动保存检查点
AutoSave bool
// SaveInterval 保存间隔(每执行多少个节点保存一次)
SaveInterval int
// SaveOnError 错误时是否保存检查点
SaveOnError bool
// SaveOnInterrupt 中断时是否保存检查点
SaveOnInterrupt bool
// MaxRetries 最大重试次数
MaxRetries int
// RetryDelay 重试延迟
RetryDelay time.Duration
}
CheckpointRunnerConfig 检查点执行器配置
func DefaultCheckpointRunnerConfig ¶
func DefaultCheckpointRunnerConfig() *CheckpointRunnerConfig
DefaultCheckpointRunnerConfig 默认配置
type CheckpointSaver ¶
type CheckpointSaver interface {
// Save 保存检查点
Save(ctx context.Context, checkpoint *Checkpoint) error
// Load 加载检查点
Load(ctx context.Context, threadID string) (*Checkpoint, error)
// LoadByID 根据 ID 加载检查点
LoadByID(ctx context.Context, id string) (*Checkpoint, error)
// List 列出线程的所有检查点
List(ctx context.Context, threadID string) ([]*Checkpoint, error)
// Delete 删除检查点
Delete(ctx context.Context, id string) error
// DeleteThread 删除线程的所有检查点
DeleteThread(ctx context.Context, threadID string) error
}
CheckpointSaver 检查点保存器接口
type CheckpointStats ¶
type CheckpointStats struct {
StepCount int `json:"step_count"` // 步骤计数
TotalDuration time.Duration `json:"total_duration"` // 总耗时
NodeDurations map[string]time.Duration `json:"node_durations,omitempty"` // 各节点耗时
LLMTokens int `json:"llm_tokens,omitempty"` // LLM Token 数
ToolCalls int `json:"tool_calls,omitempty"` // 工具调用次数
}
CheckpointStats 检查点统计
type CheckpointStatus ¶
type CheckpointStatus string
CheckpointStatus 检查点状态
const ( // CheckpointStatusPending 等待执行 CheckpointStatusPending CheckpointStatus = "pending" // CheckpointStatusRunning 执行中 CheckpointStatusRunning CheckpointStatus = "running" // CheckpointStatusCompleted 已完成 CheckpointStatusCompleted CheckpointStatus = "completed" // CheckpointStatusFailed 失败 CheckpointStatusFailed CheckpointStatus = "failed" // CheckpointStatusInterrupted 中断 CheckpointStatusInterrupted CheckpointStatus = "interrupted" )
type CheckpointVersion ¶
type CheckpointVersion struct {
Major int `json:"major"` // 主版本号(不兼容变更)
Minor int `json:"minor"` // 次版本号(向后兼容变更)
Patch int `json:"patch"` // 补丁版本号
}
CheckpointVersion 检查点版本
type CircuitBreakerConfig ¶
type CircuitBreakerConfig struct {
// FailureThreshold 失败阈值(连续失败多少次后熔断)
FailureThreshold int
// SuccessThreshold 成功阈值(连续成功多少次后恢复)
SuccessThreshold int
// Timeout 熔断超时时间(毫秒)
Timeout int64
// HalfOpenRequests 半开状态允许的请求数
HalfOpenRequests int
}
CircuitBreakerConfig 熔断器配置
func DefaultCircuitBreakerConfig ¶
func DefaultCircuitBreakerConfig() *CircuitBreakerConfig
DefaultCircuitBreakerConfig 默认熔断器配置
type CleanupPolicy ¶
type CleanupPolicy struct {
// MaxAge 最大保留时间
MaxAge time.Duration `json:"max_age,omitempty"`
// MaxCount 最大保留数量(每个线程)
MaxCount int `json:"max_count,omitempty"`
// KeepCompleted 是否保留已完成的检查点
KeepCompleted bool `json:"keep_completed,omitempty"`
// KeepBranchHeads 是否保留分支头
KeepBranchHeads bool `json:"keep_branch_heads,omitempty"`
// KeepTagged 是否保留有标签的检查点
KeepTagged bool `json:"keep_tagged,omitempty"`
}
CleanupPolicy 清理策略
type Command ¶
type Command[S State] struct { // contains filtered or unexported fields }
Command 命令对象 同时携带状态更新操作和路由决策
func GotoSwitch ¶
GotoSwitch 多路选择跳转
func UpdateAndEnd ¶
UpdateAndEnd 创建一个更新状态并结束的命令
func UpdateAndGoto ¶
UpdateAndGoto 创建一个更新状态并跳转的命令(最常见场景的快捷方式)
func (*Command[S]) ApplyUpdates ¶
func (c *Command[S]) ApplyUpdates(state S) S
ApplyUpdates 应用所有状态更新
func (*Command[S]) WithMetadata ¶
WithMetadata 附加元数据
func (*Command[S]) WithUpdate ¶
WithUpdate 添加状态更新函数 可链式调用多次,按顺序执行
type CommandHandler ¶
CommandHandler 命令处理函数类型 返回 Command 而不是直接返回状态和路由
type CompiledGraph ¶
type CompiledGraph[S State] struct { *Graph[S] // ExecutionPlan 执行计划 ExecutionPlan *ExecutionPlan // Stats 执行统计 Stats *ExecutionStats }
CompiledGraph 编译后的图 提供更高效的执行和更多运行时特性
func (*CompiledGraph[S]) Run ¶
func (cg *CompiledGraph[S]) Run(ctx context.Context, initialState S, opts ...RunOption) (S, error)
Run 执行编译后的图
func (*CompiledGraph[S]) RunWithStats ¶
func (cg *CompiledGraph[S]) RunWithStats(ctx context.Context, initialState S, opts ...RunOption) (S, *ExecutionResult, error)
RunWithStats 执行并返回详细统计
func (*CompiledGraph[S]) Visualize ¶
func (cg *CompiledGraph[S]) Visualize() string
Visualize 可视化图(返回 Mermaid 格式)
type ConditionalHandler ¶
ConditionalHandler 条件处理函数 返回下一个要执行的节点名称
type DebugView ¶
type DebugView struct {
// TotalSnapshots 总快照数
TotalSnapshots int `json:"total_snapshots"`
// CurrentIndex 当前索引
CurrentIndex int `json:"current_index"`
// Branches 分支列表
Branches []string `json:"branches"`
// Errors 错误数量
Errors int `json:"errors"`
// TotalDuration 总耗时
TotalDuration time.Duration `json:"total_duration"`
// NodeExecutions 节点执行统计
NodeExecutions map[string]int `json:"node_executions"`
}
DebugView 调试视图
type DefaultPregelMerger ¶
type DefaultPregelMerger[S State] struct{}
DefaultPregelMerger 默认状态合并器 使用最后一个状态(向后兼容)
func (*DefaultPregelMerger[S]) Merge ¶
func (m *DefaultPregelMerger[S]) Merge(base S, states []S) S
Merge 使用最后一个状态
type DynamicGraph ¶
DynamicGraph 动态图
支持运行时动态修改图结构,包括添加/删除节点和边。 线程安全,可在执行过程中修改。
func NewDynamicGraph ¶
func NewDynamicGraph[S State](name string) *DynamicGraph[S]
NewDynamicGraph 创建动态图
func (*DynamicGraph[S]) AddEdgeDynamic ¶
func (g *DynamicGraph[S]) AddEdgeDynamic(from, to string) error
AddEdgeDynamic 动态添加边
func (*DynamicGraph[S]) AddNodeDynamic ¶
func (g *DynamicGraph[S]) AddNodeDynamic(name string, handler NodeHandler[S]) error
AddNodeDynamic 动态添加节点
与普通 AddNode 不同,此方法可在图已编译后调用。
func (*DynamicGraph[S]) Build ¶
func (g *DynamicGraph[S]) Build() (*DynamicGraph[S], error)
Build 构建动态图
func (*DynamicGraph[S]) OnModified ¶
func (g *DynamicGraph[S]) OnModified(callback func(g *DynamicGraph[S]))
OnModified 注册修改回调
func (*DynamicGraph[S]) RemoveEdgeDynamic ¶
func (g *DynamicGraph[S]) RemoveEdgeDynamic(from, to string) error
RemoveEdgeDynamic 动态移除边
func (*DynamicGraph[S]) RemoveNodeDynamic ¶
func (g *DynamicGraph[S]) RemoveNodeDynamic(name string) error
RemoveNodeDynamic 动态移除节点
func (*DynamicGraph[S]) ReplaceNodeHandler ¶
func (g *DynamicGraph[S]) ReplaceNodeHandler(name string, handler NodeHandler[S]) error
ReplaceNodeHandler 替换节点处理函数
func (*DynamicGraph[S]) Snapshot ¶
func (g *DynamicGraph[S]) Snapshot() *GraphSnapshot[S]
Snapshot 创建图快照
type Edge ¶
type Edge struct {
// From 源节点名称
From string
// To 目标节点名称
To string
// Type 边类型
Type EdgeType
// Condition 条件(仅用于条件边)
// 返回 true 时边才有效
Condition func(state State) bool
// Label 边标签(用于条件路由的匹配)
Label string
// Priority 优先级(数值越小优先级越高)
Priority int
}
Edge 图的边
type EdgeBuilder ¶
type EdgeBuilder struct {
// contains filtered or unexported fields
}
EdgeBuilder 边构建器
func (*EdgeBuilder) WithCondition ¶
func (b *EdgeBuilder) WithCondition(cond func(state State) bool) *EdgeBuilder
WithCondition 设置条件
func (*EdgeBuilder) WithLabel ¶
func (b *EdgeBuilder) WithLabel(label string) *EdgeBuilder
WithLabel 设置标签
func (*EdgeBuilder) WithPriority ¶
func (b *EdgeBuilder) WithPriority(priority int) *EdgeBuilder
WithPriority 设置优先级
type EdgeSnapshot ¶
EdgeSnapshot 边快照
type EnhancedCheckpoint ¶
type EnhancedCheckpoint struct {
// 基本信息
ID string `json:"id"`
ThreadID string `json:"thread_id"`
GraphName string `json:"graph_name"`
Version CheckpointVersion `json:"version"`
Status CheckpointStatus `json:"status"`
// 执行状态
CurrentNode string `json:"current_node"`
PendingNodes []string `json:"pending_nodes"`
CompletedNodes []string `json:"completed_nodes"`
VisitedNodes []string `json:"visited_nodes"` // 所有访问过的节点(含重复)
// 状态数据
State json.RawMessage `json:"state"`
StateDiff json.RawMessage `json:"state_diff,omitempty"` // 与父检查点的差异
StateHash string `json:"state_hash,omitempty"` // 状态哈希,用于快速比较
// 分支支持
ParentID string `json:"parent_id,omitempty"` // 父检查点 ID
BranchID string `json:"branch_id,omitempty"` // 分支 ID
BranchName string `json:"branch_name,omitempty"` // 分支名称
ChildIDs []string `json:"child_ids,omitempty"` // 子检查点 ID 列表
// 元数据
Metadata map[string]any `json:"metadata,omitempty"`
Tags []string `json:"tags,omitempty"`
Description string `json:"description,omitempty"`
// 时间戳
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
// 执行统计
Stats *CheckpointStats `json:"stats,omitempty"`
}
EnhancedCheckpoint 增强检查点 支持分支、版本管理、状态差异等高级特性
type EnhancedCheckpointSaver ¶
type EnhancedCheckpointSaver interface {
CheckpointSaver // 继承基础接口
// SaveEnhanced 保存增强检查点
SaveEnhanced(ctx context.Context, checkpoint *EnhancedCheckpoint) error
// LoadEnhanced 加载最新的增强检查点
LoadEnhanced(ctx context.Context, threadID string) (*EnhancedCheckpoint, error)
// LoadEnhancedByID 根据 ID 加载增强检查点
LoadEnhancedByID(ctx context.Context, id string) (*EnhancedCheckpoint, error)
// ListEnhanced 列出线程的所有增强检查点
ListEnhanced(ctx context.Context, threadID string, opts *ListOptions) ([]*EnhancedCheckpoint, error)
// GetHistory 获取检查点历史链
GetHistory(ctx context.Context, checkpointID string, limit int) ([]*EnhancedCheckpoint, error)
// GetBranches 获取分支列表
GetBranches(ctx context.Context, threadID string) ([]*BranchInfo, error)
// CreateBranch 从检查点创建分支
CreateBranch(ctx context.Context, checkpointID string, branchName string) (*EnhancedCheckpoint, error)
// MergeBranch 合并分支
MergeBranch(ctx context.Context, sourceBranchID, targetBranchID string, strategy MergeStrategy) (*EnhancedCheckpoint, error)
// Search 搜索检查点
Search(ctx context.Context, query *CheckpointQuery) ([]*EnhancedCheckpoint, error)
// Cleanup 清理旧检查点
Cleanup(ctx context.Context, policy *CleanupPolicy) (int, error)
}
EnhancedCheckpointSaver 增强检查点保存器接口
type ErrorHandler ¶
ErrorHandler 错误处理函数类型 接收原始状态和错误,返回处理后的状态和是否已处理
type Executable ¶
type Executable interface {
// ExecuteNode 执行单个节点
ExecuteNode(ctx context.Context, nodeID string, state map[string]any) (output any, nextNode string, err error)
// GetEntryPoint 获取入口节点
GetEntryPoint() string
// GetNodeName 获取节点名称
GetNodeName(nodeID string) string
}
Executable 可执行接口,用于时间旅行调试
type ExecutionPlan ¶
type ExecutionPlan struct {
// TopologicalOrder 拓扑排序后的节点顺序
TopologicalOrder []string
// ParallelGroups 可并行执行的节点组
ParallelGroups [][]string
// CriticalPath 关键路径
CriticalPath []string
// Dependencies 节点依赖关系
Dependencies map[string][]string
}
ExecutionPlan 执行计划
type ExecutionResult ¶
type ExecutionResult struct {
// StartTime 开始时间
StartTime time.Time
// EndTime 结束时间
EndTime time.Time
// Duration 总耗时
Duration time.Duration
// NodeTiming 每个节点的耗时
NodeTiming map[string]time.Duration
// Error 错误
Error error
}
ExecutionResult 执行结果
type ExecutionStats ¶
type ExecutionStats struct {
// TotalExecutions 总执行次数
TotalExecutions int64
// TotalDuration 总执行时间
TotalDuration time.Duration
// NodeStats 节点统计
NodeStats map[string]*NodeStats
// LastExecution 最后一次执行时间
LastExecution time.Time
}
ExecutionStats 执行统计
type ExportFormat ¶
type ExportFormat int
ExportFormat 导出格式
const ( // FormatMermaid Mermaid 格式(适合嵌入 Markdown) FormatMermaid ExportFormat = iota // FormatDOT Graphviz DOT 格式(适合生成图片) FormatDOT // FormatASCII ASCII 文本格式(适合终端输出) FormatASCII )
type ExportOption ¶
type ExportOption func(*exportConfig)
ExportOption 导出选项
func WithHighlightNodes ¶
func WithHighlightNodes(nodes ...string) ExportOption
WithHighlightNodes 设置高亮节点
func WithShowConditions ¶
func WithShowConditions(show bool) ExportOption
WithShowConditions 是否显示条件边标签
type FieldValidation ¶
type FieldValidation struct {
// Min 最小值(用于 number)
Min *float64 `json:"min,omitempty"`
// Max 最大值(用于 number)
Max *float64 `json:"max,omitempty"`
// MinLength 最小长度(用于 text)
MinLength *int `json:"min_length,omitempty"`
// MaxLength 最大长度(用于 text)
MaxLength *int `json:"max_length,omitempty"`
// Pattern 正则表达式
Pattern string `json:"pattern,omitempty"`
}
FieldValidation 字段验证规则
type FileCheckpointSaver ¶
type FileCheckpointSaver struct {
// contains filtered or unexported fields
}
FileCheckpointSaver 基于文件系统的检查点保存器
将检查点以 JSON 文件形式存储在文件系统中。 目录结构:
baseDir/threads/{threadID}/ - 线程目录
baseDir/threads/{threadID}/index.json - 索引文件,记录检查点 ID 列表
baseDir/threads/{threadID}/{id}.json - 检查点 JSON 文件
线程安全:使用 sync.RWMutex 保护所有读写操作
func NewFileCheckpointSaver ¶
func NewFileCheckpointSaver(baseDir string) (*FileCheckpointSaver, error)
NewFileCheckpointSaver 创建基于文件系统的检查点保存器
参数:
- baseDir: 基础目录路径,如果不存在会自动创建
返回:
- *FileCheckpointSaver: 文件检查点保存器实例
- error: 创建目录失败时返回错误
func (*FileCheckpointSaver) Delete ¶
func (s *FileCheckpointSaver) Delete(ctx context.Context, id string) error
Delete 删除指定的检查点
删除检查点文件并从索引中移除。
参数:
- ctx: 上下文(预留,当前未使用)
- id: 要删除的检查点 ID
返回:
- error: 删除失败时返回错误
func (*FileCheckpointSaver) DeleteThread ¶
func (s *FileCheckpointSaver) DeleteThread(ctx context.Context, threadID string) error
DeleteThread 删除线程的所有检查点
删除整个线程目录及其中所有检查点文件。
参数:
- ctx: 上下文(预留,当前未使用)
- threadID: 要删除的线程 ID
返回:
- error: 删除失败时返回错误
func (*FileCheckpointSaver) List ¶
func (s *FileCheckpointSaver) List(ctx context.Context, threadID string) ([]*Checkpoint, error)
List 列出线程的所有检查点
按保存顺序返回线程中的所有检查点。
参数:
- ctx: 上下文(预留,当前未使用)
- threadID: 线程 ID
返回:
- []*Checkpoint: 检查点列表,如果线程不存在则返回 nil
- error: 读取错误时返回错误
func (*FileCheckpointSaver) Load ¶
func (s *FileCheckpointSaver) Load(ctx context.Context, threadID string) (*Checkpoint, error)
Load 加载线程的最新检查点
返回线程中最后保存的检查点。
参数:
- ctx: 上下文(预留,当前未使用)
- threadID: 线程 ID
返回:
- *Checkpoint: 最新的检查点
- error: 线程不存在或没有检查点时返回错误
func (*FileCheckpointSaver) LoadByID ¶
func (s *FileCheckpointSaver) LoadByID(ctx context.Context, id string) (*Checkpoint, error)
LoadByID 根据 ID 加载检查点
遍历所有线程目录查找指定 ID 的检查点。
参数:
- ctx: 上下文(预留,当前未使用)
- id: 检查点 ID
返回:
- *Checkpoint: 找到的检查点
- error: 检查点不存在时返回错误
func (*FileCheckpointSaver) Save ¶
func (s *FileCheckpointSaver) Save(ctx context.Context, checkpoint *Checkpoint) error
Save 保存检查点到文件系统
如果检查点 ID 为空,会自动生成。 自动设置 UpdatedAt 时间戳,如果 CreatedAt 为零值则同时设置。 检查点数据以 JSON 格式写入文件,同时更新索引文件。
参数:
- ctx: 上下文(预留,当前未使用)
- checkpoint: 要保存的检查点
返回:
- error: 保存失败时返回错误
type Graph ¶
type Graph[S State] struct { // Name 图名称 Name string // Nodes 节点映射 Nodes map[string]*Node[S] // Edges 边列表 Edges []*Edge // EntryPoint 入口点 EntryPoint string // Checkpointer 检查点保存器 Checkpointer CheckpointSaver // Metadata 元数据 Metadata map[string]any // contains filtered or unexported fields }
Graph 图定义
func (*Graph[S]) Export ¶
func (g *Graph[S]) Export(format ExportFormat, opts ...ExportOption) string
Export 将图导出为指定格式的字符串
支持 Mermaid、DOT、ASCII 三种格式:
mermaid := graph.Export(FormatMermaid)
dot := graph.Export(FormatDOT, WithExportTitle("我的工作流"))
func (*Graph[S]) GetNodeCache ¶
GetNodeCache 获取节点缓存
func (*Graph[S]) GetNodePlacements ¶
func (g *Graph[S]) GetNodePlacements() []NodePlacement
GetNodePlacements 获取所有节点放置配置
func (*Graph[S]) RunDistributed ¶
func (g *Graph[S]) RunDistributed(ctx context.Context, initialState S, registry *RemoteRegistry, opts ...RunOption) (S, error)
RunDistributed 分布式执行图 支持将节点分发到远程机器执行
实现采用 save/restore handler 模式:在执行前保存原始 handler, 替换为远程执行包装器,执行完毕后恢复原始 handler。 这样避免永久修改节点的 Handler,确保同一个 Graph 可以安全地 多次在本地和分布式模式间切换。
func (*Graph[S]) RunPregelMode ¶
func (g *Graph[S]) RunPregelMode(ctx context.Context, initialState S, opts ...PregelOption) (S, int, error)
RunPregelMode 以 Pregel 模式执行图 支持循环图和迭代执行
func (*Graph[S]) Stream ¶
func (g *Graph[S]) Stream(ctx context.Context, initialState S, opts ...RunOption) (<-chan StreamEvent[S], error)
Stream 流式执行图(返回每个节点的输出)
返回的 channel 会在以下情况关闭:
- 图执行完成
- 发生错误
- context 被取消
调用者必须消费返回的 channel,否则可能导致 goroutine 泄露。 如果不需要继续消费,应取消传入的 context。
func (*Graph[S]) StreamPregelMode ¶
func (g *Graph[S]) StreamPregelMode(ctx context.Context, initialState S, opts ...PregelOption) (<-chan PregelEvent[S], error)
StreamPregelMode 以 Pregel 模式流式执行图 每个超级步发送一个事件
func (*Graph[S]) StreamRun ¶
func (g *Graph[S]) StreamRun(ctx context.Context, state S, opts ...StreamRunOption) (*StreamChannel, error)
StreamRun 以流式模式执行图 返回 StreamChannel 用于读取执行过程中的事件
type GraphBuilder ¶
type GraphBuilder[S State] struct { // contains filtered or unexported fields }
GraphBuilder 图构建器
func (*GraphBuilder[S]) AddBarrier ¶
func (b *GraphBuilder[S]) AddBarrier(name string, merger BarrierMerger[S], waitFor ...string) *GraphBuilder[S]
AddBarrier 在图构建器中添加屏障节点 等待所有指定上游节点完成后执行合并
func (*GraphBuilder[S]) AddCommandNode ¶
func (b *GraphBuilder[S]) AddCommandNode(name string, handler CommandHandler[S]) *GraphBuilder[S]
AddCommandNode 在图构建器中添加使用 Command API 的节点 Command 节点的处理函数返回 Command,自动处理状态更新和路由
func (*GraphBuilder[S]) AddConditionalEdge ¶
func (b *GraphBuilder[S]) AddConditionalEdge(from string, router RouterFunc[S], edges map[string]string) *GraphBuilder[S]
AddConditionalEdge 添加条件边 router 函数返回目标节点的标签 edges 是标签到目标节点的映射
func (*GraphBuilder[S]) AddDoWhileLoop ¶
func (b *GraphBuilder[S]) AddDoWhileLoop(name string, condition func(S) bool, body NodeHandler[S], config ...*LoopConfig) *GraphBuilder[S]
AddDoWhileLoop 添加 do-while 循环
func (*GraphBuilder[S]) AddEdge ¶
func (b *GraphBuilder[S]) AddEdge(from, to string) *GraphBuilder[S]
AddEdge 添加边
func (*GraphBuilder[S]) AddFanOutFanIn ¶
func (b *GraphBuilder[S]) AddFanOutFanIn(name string, branches map[string]NodeHandler[S], merger BarrierMerger[S]) *GraphBuilder[S]
AddFanOutFanIn 在图构建器中添加扇出扇入节点
func (*GraphBuilder[S]) AddForLoop ¶
func (b *GraphBuilder[S]) AddForLoop(name string, iterations int, body func(ctx context.Context, state S, index int) (S, error), config ...*LoopConfig) *GraphBuilder[S]
AddForLoop 添加 for 循环
func (*GraphBuilder[S]) AddLoopBackEdge ¶
func (b *GraphBuilder[S]) AddLoopBackEdge(from, to string, condition func(S) bool, maxIterations int) *GraphBuilder[S]
AddLoopBackEdge 添加循环回边
创建从 from 到 to 的循环边,支持条件判断
并发安全说明:使用原子操作 IncrementAndGet 来避免 Get 和 Increment 之间的 TOCTOU 竞态条件
func (*GraphBuilder[S]) AddMapReduce ¶
func (b *GraphBuilder[S]) AddMapReduce(name string, split SplitFunc[S], mapFn MapFunc[S], reduce ReduceFunc[S], maxConcurrency int) *GraphBuilder[S]
AddMapReduce 在图构建器中添加 MapReduce 节点
func (*GraphBuilder[S]) AddNode ¶
func (b *GraphBuilder[S]) AddNode(name string, handler NodeHandler[S]) *GraphBuilder[S]
AddNode 添加节点
func (*GraphBuilder[S]) AddNodeWithBuilder ¶
func (b *GraphBuilder[S]) AddNodeWithBuilder(node *Node[S]) *GraphBuilder[S]
AddNodeWithBuilder 使用构建器添加节点
func (*GraphBuilder[S]) AddRetryLoop ¶
func (b *GraphBuilder[S]) AddRetryLoop(name string, body NodeHandler[S], config *RetryConfig) *GraphBuilder[S]
AddRetryLoop 添加重试循环
func (*GraphBuilder[S]) AddUntilLoop ¶
func (b *GraphBuilder[S]) AddUntilLoop(name string, condition func(S) bool, body NodeHandler[S], config ...*LoopConfig) *GraphBuilder[S]
AddUntilLoop 添加 until 循环
func (*GraphBuilder[S]) AddWhileLoop ¶
func (b *GraphBuilder[S]) AddWhileLoop(name string, condition func(S) bool, body NodeHandler[S], config ...*LoopConfig) *GraphBuilder[S]
AddWhileLoop 添加 while 循环
func (*GraphBuilder[S]) MustBuild ¶
func (b *GraphBuilder[S]) MustBuild() *Graph[S]
MustBuild 构建图,失败时 panic
⚠️ 警告:构建失败时会 panic。 仅在初始化时使用,不要在运行时调用。 推荐使用 Build() 方法并正确处理错误。
使用场景:
- 程序启动时的全局初始化
- 测试代码中
func (*GraphBuilder[S]) SetEntryPoint ¶
func (b *GraphBuilder[S]) SetEntryPoint(node string) *GraphBuilder[S]
SetEntryPoint 设置入口点
func (*GraphBuilder[S]) SetFinishPoint ¶
func (b *GraphBuilder[S]) SetFinishPoint(nodes ...string) *GraphBuilder[S]
SetFinishPoint 设置结束点(添加到 END 的边)
func (*GraphBuilder[S]) WithCheckpointer ¶
func (b *GraphBuilder[S]) WithCheckpointer(saver CheckpointSaver) *GraphBuilder[S]
WithCheckpointer 设置检查点保存器
func (*GraphBuilder[S]) WithMetadata ¶
func (b *GraphBuilder[S]) WithMetadata(key string, value any) *GraphBuilder[S]
WithMetadata 设置元数据
func (*GraphBuilder[S]) WithNodeCache ¶
func (b *GraphBuilder[S]) WithNodeCache(nodeName string, cache NodeCache) *GraphBuilder[S]
WithNodeCache 为图构建器添加节点缓存
func (*GraphBuilder[S]) WithNodePlacement ¶
func (b *GraphBuilder[S]) WithNodePlacement(nodeName, executorName string) *GraphBuilder[S]
WithNodePlacement 配置节点的远程执行位置
func (*GraphBuilder[S]) WithNodePlacementNoFallback ¶
func (b *GraphBuilder[S]) WithNodePlacementNoFallback(nodeName, executorName string) *GraphBuilder[S]
WithNodePlacementNoFallback 配置节点的远程执行位置(不降级)
type GraphComposer ¶
type GraphComposer[S State] struct { // contains filtered or unexported fields }
GraphComposer 图组合器
用于组合多个图成为一个更大的图
func NewGraphComposer ¶
func NewGraphComposer[S State](name string) *GraphComposer[S]
NewGraphComposer 创建图组合器
func (*GraphComposer[S]) AddGraph ¶
func (c *GraphComposer[S]) AddGraph(g *Graph[S]) int
AddGraph 添加子图
返回子图的索引,用于后续连接
func (*GraphComposer[S]) Compose ¶
func (c *GraphComposer[S]) Compose() (*Graph[S], error)
Compose 组合成新图
func (*GraphComposer[S]) Connect ¶
func (c *GraphComposer[S]) Connect(fromGraph int, fromNode string, toGraph int, toNode string) *GraphComposer[S]
Connect 连接两个子图
参数:
- fromGraph: 源图索引
- fromNode: 源节点(空字符串表示图的默认出口)
- toGraph: 目标图索引
- toNode: 目标节点(空字符串表示图的默认入口)
func (*GraphComposer[S]) Sequential ¶
func (c *GraphComposer[S]) Sequential() *GraphComposer[S]
Sequential 顺序连接所有子图
type GraphSnapshot ¶
type GraphSnapshot[S State] struct { // Name 图名称 Name string // NodeNames 节点名称列表 NodeNames []string // Edges 边列表 Edges []EdgeSnapshot // Version 版本号 Version int64 // Metadata 元数据 Metadata map[string]any }
GraphSnapshot 图快照
type HITLCallback ¶
type HITLCallback func(ctx context.Context, request *HITLRequest) (*HITLResponse, error)
HITLCallback 回调式处理器
func (HITLCallback) Handle ¶
func (c HITLCallback) Handle(ctx context.Context, request *HITLRequest) (*HITLResponse, error)
Handle 实现 HITLHandler 接口
type HITLHandler ¶
type HITLHandler interface {
// Handle 处理人工介入请求
// 阻塞直到收到响应或超时
Handle(ctx context.Context, request *HITLRequest) (*HITLResponse, error)
}
HITLHandler 人工介入处理器接口
type HITLManager ¶
type HITLManager struct {
// contains filtered or unexported fields
}
HITLManager 人工介入管理器
func NewHITLManager ¶
func NewHITLManager(handler HITLHandler, config ...HITLManagerConfig) *HITLManager
NewHITLManager 创建管理器
func (*HITLManager) GetActiveRequests ¶
func (m *HITLManager) GetActiveRequests() []*HITLRequest
GetActiveRequests 获取活跃请求
func (*HITLManager) GetHistory ¶
func (m *HITLManager) GetHistory(limit int) []*HITLRecord
GetHistory 获取历史记录
func (*HITLManager) Submit ¶
func (m *HITLManager) Submit(ctx context.Context, request *HITLRequest) (*HITLResponse, error)
Submit 提交请求
type HITLManagerConfig ¶
type HITLManagerConfig struct {
// MaxPendingRequests 最大待处理请求数
MaxPendingRequests int
// DefaultTimeout 默认超时
DefaultTimeout time.Duration
// HistoryLimit 历史记录上限
HistoryLimit int
}
HITLManagerConfig 管理器配置
type HITLNode ¶
type HITLNode[S State] struct { // ID 节点 ID ID string // Name 节点名称 Name string // Type 介入类型 Type HITLType // Handler 处理器 Handler HITLHandler // RequestBuilder 请求构建器 RequestBuilder func(state S) *HITLRequest // ResponseHandler 响应处理器 ResponseHandler func(state S, response *HITLResponse) S // Condition 触发条件(可选) Condition func(state S) bool }
HITLNode 人工介入节点
func NewApprovalNode ¶
func NewApprovalNode[S State](id string, handler HITLHandler, opts ...HITLNodeOption[S]) *HITLNode[S]
NewApprovalNode 创建审批节点
func NewInputNode ¶
func NewInputNode[S State](id string, handler HITLHandler, schema map[string]any, opts ...HITLNodeOption[S]) *HITLNode[S]
NewInputNode 创建输入节点
func NewReviewNode ¶
func NewReviewNode[S State](id string, handler HITLHandler, opts ...HITLNodeOption[S]) *HITLNode[S]
NewReviewNode 创建审查节点
type HITLNodeOption ¶
HITLNodeOption HITL 节点选项
func WithHITLCondition ¶
func WithHITLCondition[S State](condition func(state S) bool) HITLNodeOption[S]
WithHITLCondition 设置触发条件
func WithHITLDescription ¶
func WithHITLDescription[S State](desc string) HITLNodeOption[S]
WithHITLDescription 设置描述
func WithHITLPriority ¶
func WithHITLPriority[S State](priority HITLPriority) HITLNodeOption[S]
WithHITLPriority 设置优先级
func WithHITLTimeout ¶
func WithHITLTimeout[S State](timeout time.Duration) HITLNodeOption[S]
WithHITLTimeout 设置超时
type HITLOption ¶
type HITLOption struct {
// ID 选项 ID
ID string `json:"id"`
// Label 显示标签
Label string `json:"label"`
// Description 选项描述
Description string `json:"description,omitempty"`
// Recommended 是否推荐
Recommended bool `json:"recommended,omitempty"`
// Dangerous 是否危险操作
Dangerous bool `json:"dangerous,omitempty"`
}
HITLOption 选项
type HITLPriority ¶
type HITLPriority string
HITLPriority 优先级
const ( // PriorityLow 低优先级 PriorityLow HITLPriority = "low" // PriorityNormal 普通优先级 PriorityNormal HITLPriority = "normal" // PriorityHigh 高优先级 PriorityHigh HITLPriority = "high" // PriorityUrgent 紧急优先级 PriorityUrgent HITLPriority = "urgent" )
type HITLRecord ¶
type HITLRecord struct {
Request *HITLRequest `json:"request"`
Response *HITLResponse `json:"response,omitempty"`
Status string `json:"status"` // pending, completed, timeout, cancelled
StartedAt time.Time `json:"started_at"`
EndedAt *time.Time `json:"ended_at,omitempty"`
}
HITLRecord HITL 记录
type HITLRequest ¶
type HITLRequest struct {
// ID 请求唯一标识
ID string `json:"id"`
// Type 介入类型
Type HITLType `json:"type"`
// NodeID 触发节点
NodeID string `json:"node_id"`
// Title 请求标题
Title string `json:"title"`
// Description 请求描述
Description string `json:"description"`
// Context 上下文信息
Context map[string]any `json:"context,omitempty"`
// Options 可选项(用于审批)
Options []HITLOption `json:"options,omitempty"`
// InputSchema 输入 Schema(用于输入模式)
InputSchema map[string]any `json:"input_schema,omitempty"`
// CurrentOutput 当前输出(用于审查模式)
CurrentOutput any `json:"current_output,omitempty"`
// Timeout 超时时间
Timeout time.Duration `json:"timeout,omitempty"`
// Priority 优先级
Priority HITLPriority `json:"priority"`
// Metadata 元数据
Metadata map[string]any `json:"metadata,omitempty"`
// CreatedAt 创建时间
CreatedAt time.Time `json:"created_at"`
// ExpiresAt 过期时间
ExpiresAt *time.Time `json:"expires_at,omitempty"`
}
HITLRequest 人工介入请求
type HITLResponse ¶
type HITLResponse struct {
// RequestID 对应请求 ID
RequestID string `json:"request_id"`
// Approved 是否批准(用于审批模式)
Approved bool `json:"approved,omitempty"`
// SelectedOption 选择的选项 ID
SelectedOption string `json:"selected_option,omitempty"`
// Input 人工输入数据
Input map[string]any `json:"input,omitempty"`
// CorrectedOutput 纠正后的输出
CorrectedOutput any `json:"corrected_output,omitempty"`
// Feedback 反馈信息
Feedback string `json:"feedback,omitempty"`
// RespondedBy 响应者标识
RespondedBy string `json:"responded_by,omitempty"`
// RespondedAt 响应时间
RespondedAt time.Time `json:"responded_at"`
// Metadata 元数据
Metadata map[string]any `json:"metadata,omitempty"`
}
HITLResponse 人工响应
type HITLStats ¶
type HITLStats struct {
TotalRequests int64 `json:"total_requests"`
CompletedCount int64 `json:"completed_count"`
TimeoutCount int64 `json:"timeout_count"`
CancelledCount int64 `json:"cancelled_count"`
AverageWaitTime time.Duration `json:"average_wait_time"`
ApprovalRate float64 `json:"approval_rate"`
ByType map[HITLType]int64 `json:"by_type"`
ByPriority map[HITLPriority]int64 `json:"by_priority"`
}
HITLStats HITL 统计
type HITLType ¶
type HITLType string
HITLType 人工介入类型
const ( // HITLApproval 审批模式:需要人工批准才能继续 HITLApproval HITLType = "approval" // HITLInput 输入模式:需要人工提供额外输入 HITLInput HITLType = "input" // HITLReview 审查模式:人工审查 LLM 输出 HITLReview HITLType = "review" // HITLTakeover 接管模式:人工完全接管执行 HITLTakeover HITLType = "takeover" // HITLCorrection 纠正模式:人工纠正错误 HITLCorrection HITLType = "correction" )
type HTTPExecutorOption ¶
type HTTPExecutorOption func(*HTTPNodeExecutor)
HTTPExecutorOption HTTP 执行器选项
func WithHTTPHeader ¶
func WithHTTPHeader(key, value string) HTTPExecutorOption
WithHTTPHeader 添加自定义请求头
func WithHTTPTimeout ¶
func WithHTTPTimeout(timeout time.Duration) HTTPExecutorOption
WithHTTPTimeout 设置 HTTP 超时
type HTTPNodeExecutor ¶
type HTTPNodeExecutor struct {
// contains filtered or unexported fields
}
HTTPNodeExecutor 基于 HTTP 的远程节点执行器
func NewHTTPNodeExecutor ¶
func NewHTTPNodeExecutor(name, baseURL string, opts ...HTTPExecutorOption) *HTTPNodeExecutor
NewHTTPNodeExecutor 创建 HTTP 远程节点执行器
type HumanInTheLoop ¶
type HumanInTheLoop[S State] struct { // contains filtered or unexported fields }
HumanInTheLoop Human-in-the-loop 执行器 用于在图执行过程中处理人工干预
func NewHumanInTheLoop ¶
func NewHumanInTheLoop[S State](graph *Graph[S], handler InterruptHandler, saver CheckpointSaver) *HumanInTheLoop[S]
NewHumanInTheLoop 创建 Human-in-the-loop 执行器
func (*HumanInTheLoop[S]) Resume ¶
func (h *HumanInTheLoop[S]) Resume(ctx context.Context, threadID string, response *InterruptResponse) (S, *Interrupt, error)
Resume 恢复执行(在中断解决后)
func (*HumanInTheLoop[S]) RunWithInterrupt ¶
func (h *HumanInTheLoop[S]) RunWithInterrupt(ctx context.Context, threadID string, initialState S) (S, *Interrupt, error)
RunWithInterrupt 运行图,支持中断和恢复
func (*HumanInTheLoop[S]) WaitAndResume ¶
func (h *HumanInTheLoop[S]) WaitAndResume(ctx context.Context, threadID string, interruptID string) (S, error)
WaitAndResume 等待中断解决并自动恢复
type InputField ¶
type InputField struct {
// Name 字段名
Name string `json:"name"`
// Type 字段类型(text, number, boolean, select, textarea)
Type string `json:"type"`
// Label 显示标签
Label string `json:"label"`
// Description 描述
Description string `json:"description,omitempty"`
// Required 是否必填
Required bool `json:"required,omitempty"`
// Default 默认值
Default any `json:"default,omitempty"`
// Options 选项(用于 select 类型)
Options []string `json:"options,omitempty"`
// Validation 验证规则
Validation *FieldValidation `json:"validation,omitempty"`
}
InputField 输入字段
type InputSchema ¶
type InputSchema struct {
// Fields 输入字段
Fields []InputField `json:"fields"`
}
InputSchema 输入模式
type Interrupt ¶
type Interrupt struct {
// ID 唯一标识符
ID string `json:"id"`
// ThreadID 线程 ID
ThreadID string `json:"thread_id"`
// GraphName 图名称
GraphName string `json:"graph_name"`
// NodeName 触发中断的节点
NodeName string `json:"node_name"`
// Type 中断类型
Type InterruptType `json:"type"`
// Status 中断状态
Status InterruptStatus `json:"status"`
// Title 标题(用于显示)
Title string `json:"title"`
// Message 消息内容
Message string `json:"message"`
// Data 附加数据
Data map[string]any `json:"data,omitempty"`
// Options 选项(用于 Approval 类型)
Options []InterruptOption `json:"options,omitempty"`
// InputSchema 输入模式(用于 Input 类型)
InputSchema *InputSchema `json:"input_schema,omitempty"`
// Response 响应数据
Response *InterruptResponse `json:"response,omitempty"`
// Timeout 超时时间
Timeout time.Duration `json:"timeout,omitempty"`
// CreatedAt 创建时间
CreatedAt time.Time `json:"created_at"`
// UpdatedAt 更新时间
UpdatedAt time.Time `json:"updated_at"`
// ResolvedAt 解决时间
ResolvedAt *time.Time `json:"resolved_at,omitempty"`
// ResolvedBy 解决者
ResolvedBy string `json:"resolved_by,omitempty"`
}
Interrupt 中断定义
func ApprovalInterrupt ¶
ApprovalInterrupt 创建审批类型中断
func InputInterrupt ¶
func InputInterrupt(nodeID, title, message string, fields ...InputField) *Interrupt
InputInterrupt 创建输入类型中断
func IsInterruptError ¶
IsInterruptError 判断是否是中断错误
type InterruptBuilder ¶
type InterruptBuilder struct {
// contains filtered or unexported fields
}
InterruptBuilder 中断构建器
func (*InterruptBuilder) WithData ¶
func (b *InterruptBuilder) WithData(key string, value any) *InterruptBuilder
WithData 设置数据
func (*InterruptBuilder) WithInputSchema ¶
func (b *InterruptBuilder) WithInputSchema(schema *InputSchema) *InterruptBuilder
WithInputSchema 设置输入模式
func (*InterruptBuilder) WithMessage ¶
func (b *InterruptBuilder) WithMessage(message string) *InterruptBuilder
WithMessage 设置消息
func (*InterruptBuilder) WithOptions ¶
func (b *InterruptBuilder) WithOptions(options ...InterruptOption) *InterruptBuilder
WithOptions 设置选项
func (*InterruptBuilder) WithTimeout ¶
func (b *InterruptBuilder) WithTimeout(timeout time.Duration) *InterruptBuilder
WithTimeout 设置超时时间
func (*InterruptBuilder) WithTitle ¶
func (b *InterruptBuilder) WithTitle(title string) *InterruptBuilder
WithTitle 设置标题
func (*InterruptBuilder) WithType ¶
func (b *InterruptBuilder) WithType(t InterruptType) *InterruptBuilder
WithType 设置中断类型
type InterruptConfig ¶
type InterruptConfig struct {
// Handler 中断处理器
Handler InterruptHandler
// DefaultTimeout 默认超时时间
DefaultTimeout time.Duration
// AutoResume 自动恢复(当中断解决后自动继续执行)
AutoResume bool
}
InterruptConfig 中断配置
func NewInterruptConfig ¶
func NewInterruptConfig(handler InterruptHandler) *InterruptConfig
NewInterruptConfig 创建中断配置
func (*InterruptConfig) WithAutoResume ¶
func (c *InterruptConfig) WithAutoResume(autoResume bool) *InterruptConfig
WithAutoResume 设置自动恢复
func (*InterruptConfig) WithDefaultTimeout ¶
func (c *InterruptConfig) WithDefaultTimeout(timeout time.Duration) *InterruptConfig
WithDefaultTimeout 设置默认超时时间
type InterruptError ¶
type InterruptError struct {
Interrupt *Interrupt
}
InterruptError 中断错误
func (*InterruptError) Error ¶
func (e *InterruptError) Error() string
type InterruptHandler ¶
type InterruptHandler interface {
// Create 创建中断
Create(ctx context.Context, interrupt *Interrupt) error
// Get 获取中断
Get(ctx context.Context, id string) (*Interrupt, error)
// Resolve 解决中断
Resolve(ctx context.Context, id string, response *InterruptResponse, resolvedBy string) error
// Cancel 取消中断
Cancel(ctx context.Context, id string) error
// List 列出中断
List(ctx context.Context, threadID string) ([]*Interrupt, error)
// ListPending 列出待处理的中断
ListPending(ctx context.Context) ([]*Interrupt, error)
// Wait 等待中断解决
Wait(ctx context.Context, id string) (*Interrupt, error)
// WaitWithTimeout 带超时等待中断解决
WaitWithTimeout(ctx context.Context, id string, timeout time.Duration) (*Interrupt, error)
}
InterruptHandler 中断处理器接口
type InterruptOption ¶
type InterruptOption struct {
// Value 选项值
Value string `json:"value"`
// Label 显示标签
Label string `json:"label"`
// Description 描述
Description string `json:"description,omitempty"`
// Style 样式(primary, danger, default)
Style string `json:"style,omitempty"`
}
InterruptOption 中断选项
type InterruptResponse ¶
type InterruptResponse struct {
// Action 动作(approve, reject, submit, cancel)
Action string `json:"action"`
// Data 响应数据
Data map[string]any `json:"data,omitempty"`
// Comment 备注
Comment string `json:"comment,omitempty"`
}
InterruptResponse 中断响应
type InterruptStatus ¶
type InterruptStatus string
InterruptStatus 中断状态
const ( // InterruptStatusPending 等待处理 InterruptStatusPending InterruptStatus = "pending" // InterruptStatusApproved 已批准 InterruptStatusApproved InterruptStatus = "approved" // InterruptStatusRejected 已拒绝 InterruptStatusRejected InterruptStatus = "rejected" // InterruptStatusCompleted 已完成(带输入) InterruptStatusCompleted InterruptStatus = "completed" // InterruptStatusTimeout 超时 InterruptStatusTimeout InterruptStatus = "timeout" // InterruptStatusCancelled 已取消 InterruptStatusCancelled InterruptStatus = "cancelled" )
type InterruptType ¶
type InterruptType string
InterruptType 中断类型
const ( // InterruptTypeApproval 需要人工审批 InterruptTypeApproval InterruptType = "approval" // InterruptTypeInput 需要人工输入 InterruptTypeInput InterruptType = "input" // InterruptTypeReview 需要人工审核 InterruptTypeReview InterruptType = "review" // InterruptTypeCustom 自定义中断 InterruptTypeCustom InterruptType = "custom" )
type ListOptions ¶
type ListOptions struct {
Limit int // 限制数量
Offset int // 偏移量
Order string // 排序方式: "asc" 或 "desc"
Status CheckpointStatus // 状态过滤
BranchID string // 分支过滤
StartTime *time.Time // 开始时间
EndTime *time.Time // 结束时间
}
ListOptions 列表选项
type LoopConfig ¶
type LoopConfig struct {
// MaxIterations 最大迭代次数(0 表示无限制)
MaxIterations int
// Timeout 循环超时时间(0 表示无超时)
Timeout time.Duration
// OnIteration 每次迭代回调
OnIteration func(iteration int)
// OnBreak 中断回调
OnBreak func(iteration int, reason string)
// OnComplete 完成回调
OnComplete func(iterations int)
// BreakOnError 遇到错误时中断
BreakOnError bool
// ContinueOnError 遇到错误时继续
ContinueOnError bool
}
LoopConfig 循环配置
type LoopCounter ¶
type LoopCounter struct {
// contains filtered or unexported fields
}
LoopCounter 循环计数器
type MemoryCacheOption ¶
type MemoryCacheOption func(*MemoryNodeCache)
MemoryCacheOption 内存缓存选项
func WithCacheCapacity ¶
func WithCacheCapacity(capacity int) MemoryCacheOption
WithCacheCapacity 设置缓存容量
type MemoryCheckpointSaver ¶
type MemoryCheckpointSaver struct {
// contains filtered or unexported fields
}
MemoryCheckpointSaver 内存检查点保存器
func NewMemoryCheckpointSaver ¶
func NewMemoryCheckpointSaver() *MemoryCheckpointSaver
NewMemoryCheckpointSaver 创建内存检查点保存器
func (*MemoryCheckpointSaver) Delete ¶
func (s *MemoryCheckpointSaver) Delete(ctx context.Context, id string) error
Delete 删除检查点
func (*MemoryCheckpointSaver) DeleteThread ¶
func (s *MemoryCheckpointSaver) DeleteThread(ctx context.Context, threadID string) error
DeleteThread 删除线程的所有检查点
func (*MemoryCheckpointSaver) List ¶
func (s *MemoryCheckpointSaver) List(ctx context.Context, threadID string) ([]*Checkpoint, error)
List 列出线程的所有检查点
func (*MemoryCheckpointSaver) Load ¶
func (s *MemoryCheckpointSaver) Load(ctx context.Context, threadID string) (*Checkpoint, error)
Load 加载最新的检查点
func (*MemoryCheckpointSaver) LoadByID ¶
func (s *MemoryCheckpointSaver) LoadByID(ctx context.Context, id string) (*Checkpoint, error)
LoadByID 根据 ID 加载检查点
func (*MemoryCheckpointSaver) Save ¶
func (s *MemoryCheckpointSaver) Save(ctx context.Context, checkpoint *Checkpoint) error
Save 保存检查点
type MemoryEnhancedCheckpointSaver ¶
type MemoryEnhancedCheckpointSaver struct {
*MemoryCheckpointSaver
// contains filtered or unexported fields
}
MemoryEnhancedCheckpointSaver 内存增强检查点保存器
func NewMemoryEnhancedCheckpointSaver ¶
func NewMemoryEnhancedCheckpointSaver() *MemoryEnhancedCheckpointSaver
NewMemoryEnhancedCheckpointSaver 创建内存增强检查点保存器
func (*MemoryEnhancedCheckpointSaver) Cleanup ¶
func (s *MemoryEnhancedCheckpointSaver) Cleanup(ctx context.Context, policy *CleanupPolicy) (int, error)
Cleanup 清理旧检查点
func (*MemoryEnhancedCheckpointSaver) CreateBranch ¶
func (s *MemoryEnhancedCheckpointSaver) CreateBranch(ctx context.Context, checkpointID string, branchName string) (*EnhancedCheckpoint, error)
CreateBranch 从检查点创建分支
func (*MemoryEnhancedCheckpointSaver) GetBranches ¶
func (s *MemoryEnhancedCheckpointSaver) GetBranches(ctx context.Context, threadID string) ([]*BranchInfo, error)
GetBranches 获取分支列表
func (*MemoryEnhancedCheckpointSaver) GetHistory ¶
func (s *MemoryEnhancedCheckpointSaver) GetHistory(ctx context.Context, checkpointID string, limit int) ([]*EnhancedCheckpoint, error)
GetHistory 获取检查点历史链
func (*MemoryEnhancedCheckpointSaver) ListEnhanced ¶
func (s *MemoryEnhancedCheckpointSaver) ListEnhanced(ctx context.Context, threadID string, opts *ListOptions) ([]*EnhancedCheckpoint, error)
ListEnhanced 列出线程的所有增强检查点
func (*MemoryEnhancedCheckpointSaver) LoadEnhanced ¶
func (s *MemoryEnhancedCheckpointSaver) LoadEnhanced(ctx context.Context, threadID string) (*EnhancedCheckpoint, error)
LoadEnhanced 加载最新的增强检查点
func (*MemoryEnhancedCheckpointSaver) LoadEnhancedByID ¶
func (s *MemoryEnhancedCheckpointSaver) LoadEnhancedByID(ctx context.Context, id string) (*EnhancedCheckpoint, error)
LoadEnhancedByID 根据 ID 加载增强检查点
func (*MemoryEnhancedCheckpointSaver) MergeBranch ¶
func (s *MemoryEnhancedCheckpointSaver) MergeBranch(ctx context.Context, sourceBranchID, targetBranchID string, strategy MergeStrategy) (*EnhancedCheckpoint, error)
MergeBranch 合并分支
func (*MemoryEnhancedCheckpointSaver) SaveEnhanced ¶
func (s *MemoryEnhancedCheckpointSaver) SaveEnhanced(ctx context.Context, checkpoint *EnhancedCheckpoint) error
SaveEnhanced 保存增强检查点
func (*MemoryEnhancedCheckpointSaver) Search ¶
func (s *MemoryEnhancedCheckpointSaver) Search(ctx context.Context, query *CheckpointQuery) ([]*EnhancedCheckpoint, error)
Search 搜索检查点
type MemoryInterruptHandler ¶
type MemoryInterruptHandler struct {
// contains filtered or unexported fields
}
MemoryInterruptHandler 内存中断处理器
func NewMemoryInterruptHandler ¶
func NewMemoryInterruptHandler() *MemoryInterruptHandler
NewMemoryInterruptHandler 创建内存中断处理器
func (*MemoryInterruptHandler) Cancel ¶
func (h *MemoryInterruptHandler) Cancel(ctx context.Context, id string) error
Cancel 取消中断
func (*MemoryInterruptHandler) Create ¶
func (h *MemoryInterruptHandler) Create(ctx context.Context, interrupt *Interrupt) error
Create 创建中断
func (*MemoryInterruptHandler) ListPending ¶
func (h *MemoryInterruptHandler) ListPending(ctx context.Context) ([]*Interrupt, error)
ListPending 列出待处理的中断
func (*MemoryInterruptHandler) Resolve ¶
func (h *MemoryInterruptHandler) Resolve(ctx context.Context, id string, response *InterruptResponse, resolvedBy string) error
Resolve 解决中断
func (*MemoryInterruptHandler) WaitWithTimeout ¶
func (h *MemoryInterruptHandler) WaitWithTimeout(ctx context.Context, id string, timeout time.Duration) (*Interrupt, error)
WaitWithTimeout 带超时等待中断解决
type MemoryNodeCache ¶
type MemoryNodeCache struct {
// contains filtered or unexported fields
}
MemoryNodeCache 内存节点缓存 使用 LRU 策略,支持 TTL 过期
func NewMemoryNodeCache ¶
func NewMemoryNodeCache(opts ...MemoryCacheOption) *MemoryNodeCache
NewMemoryNodeCache 创建内存节点缓存
type MemorySnapshotStorage ¶
type MemorySnapshotStorage struct {
// contains filtered or unexported fields
}
MemorySnapshotStorage 内存快照存储
func NewMemorySnapshotStorage ¶
func NewMemorySnapshotStorage() *MemorySnapshotStorage
NewMemorySnapshotStorage 创建内存快照存储
func (*MemorySnapshotStorage) Clear ¶
func (s *MemorySnapshotStorage) Clear(ctx context.Context) error
Clear 清空所有快照
func (*MemorySnapshotStorage) Delete ¶
func (s *MemorySnapshotStorage) Delete(ctx context.Context, index int) error
Delete 删除快照
func (*MemorySnapshotStorage) Load ¶
func (s *MemorySnapshotStorage) Load(ctx context.Context, index int) (*StateSnapshot, error)
Load 加载快照
func (*MemorySnapshotStorage) LoadRange ¶
func (s *MemorySnapshotStorage) LoadRange(ctx context.Context, start, end int) ([]*StateSnapshot, error)
LoadRange 加载范围内的快照
func (*MemorySnapshotStorage) Save ¶
func (s *MemorySnapshotStorage) Save(ctx context.Context, snapshot *StateSnapshot) error
Save 保存快照
type MergeStrategy ¶
type MergeStrategy string
MergeStrategy 合并策略
const ( // MergeStrategyOverwrite 覆盖目标状态 MergeStrategyOverwrite MergeStrategy = "overwrite" // MergeStrategyMerge 合并状态 MergeStrategyMerge MergeStrategy = "merge" // MergeStrategyKeepBoth 保留两者 MergeStrategyKeepBoth MergeStrategy = "keep_both" )
type MultiRouter ¶
MultiRouter 多路由器 返回多个可能的下一节点
type MultiRouterFunc ¶
MultiRouterFunc 多路由函数类型
func (MultiRouterFunc[S]) RouteMulti ¶
func (f MultiRouterFunc[S]) RouteMulti(state S) []string
RouteMulti 实现 MultiRouter 接口
type Node ¶
type Node[S State] struct { // Name 节点名称 Name string // Type 节点类型 Type NodeType // Handler 节点处理函数 Handler NodeHandler[S] // Metadata 节点元数据 Metadata map[string]any // RetryPolicy 重试策略 RetryPolicy *RetryPolicy // Timeout 超时时间(毫秒) Timeout int64 }
Node 图节点
func BarrierNode ¶
func BarrierNode[S State](name string, merger BarrierMerger[S], waitFor ...string) *Node[S]
BarrierNode 创建屏障/延迟节点 等待所有指定的上游分支完成后,使用 merger 合并所有分支结果
参数:
- name: 节点名称
- merger: 状态合并函数
- waitFor: 需要等待的上游节点名称列表
func BranchNode ¶
func BranchNode[S State](name string, branches map[string]*Graph[S], selector func(S) string) *Node[S]
BranchNode 创建分支节点
根据状态决定下一步执行哪个分支。
参数:
- name: 节点名称
- branches: 分支映射(label -> 子图)
- selector: 分支选择器,返回分支的 label
func BulkheadNode ¶
func BulkheadNode[S State](name string, handler NodeHandler[S], config *BulkheadConfig) *Node[S]
BulkheadNode 创建舱壁节点 限制并发执行数量,实现资源隔离
func CatchNode ¶
func CatchNode[S State](name string, handler ErrorHandler[S], errorTypes ...error) *Node[S]
CatchNode 创建错误捕获节点 用于捕获上游节点的错误并进行处理
参数:
- name: 节点名称
- handler: 错误处理函数,返回 (新状态, 是否继续执行, 错误)
- errorTypes: 要捕获的错误类型(可选,nil 表示捕获所有错误)
func CircuitBreakerNode ¶
func CircuitBreakerNode[S State](name string, handler NodeHandler[S], config *CircuitBreakerConfig) *Node[S]
CircuitBreakerNode 创建熔断器节点 当失败次数达到阈值时,自动熔断,一段时间后进入半开状态尝试恢复
参数:
- name: 节点名称
- handler: 节点处理函数
- config: 熔断器配置
func ConditionalNode ¶
func ConditionalNode[S State](name string, router ConditionalHandler[S]) *Node[S]
ConditionalNode 创建条件节点
func ConditionalSubgraph ¶
func ConditionalSubgraph[S State](name string, selector func(S) int, subgraphs ...*Graph[S]) *Node[S]
ConditionalSubgraph 创建条件子图节点
根据条件选择执行不同的子图。
参数:
- name: 节点名称
- selector: 选择器函数,返回要执行的子图索引
- subgraphs: 可选的子图列表
func DoWhileLoopNode ¶
func DoWhileLoopNode[S State](name string, condition func(S) bool, body NodeHandler[S], config ...*LoopConfig) *Node[S]
DoWhileLoopNode 创建 do-while 循环节点
先执行 body,再检查 condition,至少执行一次
func FallbackNode ¶
func FallbackNode[S State](name string, primaryHandler, fallbackHandler NodeHandler[S]) *Node[S]
FallbackNode 创建降级节点 当主节点执行失败时,执行降级逻辑
参数:
- name: 节点名称
- primaryHandler: 主处理函数
- fallbackHandler: 降级处理函数
func FanOutFanInNode ¶
func FanOutFanInNode[S State](name string, branches map[string]NodeHandler[S], merger BarrierMerger[S]) *Node[S]
FanOutFanInNode 创建扇出扇入节点 将状态同时发送到多个处理函数并行执行,然后合并所有结果
与 ParallelNodeWithMerger 的区别:
- FanOutFanIn 返回每个分支的命名结果,便于后续处理
- 每个分支有自己的名称,可在 merger 中按名称访问
func ForEachLoopNode ¶
func ForEachLoopNode[S State, T any]( name string, getItems func(S) []T, body func(ctx context.Context, state S, item T, index int) (S, error), config ...*LoopConfig, ) *Node[S]
ForEachLoopNode 创建 forEach 循环节点
遍历集合执行循环体
func ForLoopNode ¶
func ForLoopNode[S State](name string, iterations int, body func(ctx context.Context, state S, index int) (S, error), config ...*LoopConfig) *Node[S]
ForLoopNode 创建 for 循环节点
执行固定次数的循环
func LoopSubgraph ¶
func LoopSubgraph[S State](name string, subgraph *Graph[S], condition func(S, int) bool, maxIterations int) *Node[S]
LoopSubgraph 创建循环子图节点
重复执行子图直到满足退出条件。
参数:
- name: 节点名称
- subgraph: 要循环执行的子图
- condition: 继续循环的条件(返回 true 继续,false 退出)
- maxIterations: 最大迭代次数(0 表示无限制)
func MapReduceNode ¶
func MapReduceNode[S State](name string, split SplitFunc[S], mapFn MapFunc[S], reduce ReduceFunc[S], maxConcurrency int) *Node[S]
MapReduceNode 创建 MapReduce 节点 实现分片-并行处理-聚合的完整模式
参数:
- name: 节点名称
- split: 将输入状态分割为多个子状态
- mapFn: 对每个子状态执行处理
- reduce: 将所有处理结果归约为最终状态
- maxConcurrency: 最大并行度(0 表示无限制)
func ParallelForEachLoopNode ¶
func ParallelForEachLoopNode[S State, T any]( name string, getItems func(S) []T, body func(ctx context.Context, item T, index int) error, merger func(state S, results []error) (S, error), maxConcurrency int, ) *Node[S]
ParallelForEachLoopNode 并行 forEach 循环
并行处理集合中的元素
func ParallelNode ¶
func ParallelNode[S State](name string, handlers ...NodeHandler[S]) *Node[S]
ParallelNode 创建并行执行节点 将多个节点的处理函数并行执行,并合并结果
警告:默认只保留最后一个结果。 如果需要正确合并多个 handler 的结果,请使用 ParallelNodeWithMerger。
func ParallelNodeWithMerger ¶
func ParallelNodeWithMerger[S State](name string, merger StateMerger[S], handlers ...NodeHandler[S]) *Node[S]
ParallelNodeWithMerger 创建带自定义状态合并器的并行执行节点
参数:
- name: 节点名称
- merger: 状态合并函数,用于合并所有 handler 的执行结果。 如果为 nil,默认返回最后一个结果(向后兼容)
- handlers: 要并行执行的处理函数列表
StateMerger 定义在 subgraph.go 中:func(original S, outputs []S) S
func ParallelSubgraphs ¶
func ParallelSubgraphs[S State](name string, subgraphs []*Graph[S], merger StateMerger[S]) *Node[S]
ParallelSubgraphs 创建并行子图执行节点
同时执行多个子图,并合并结果。
参数:
- name: 节点名称
- subgraphs: 要并行执行的子图列表
- merger: 状态合并函数
func RetryLoopNode ¶
func RetryLoopNode[S State](name string, body NodeHandler[S], config *RetryConfig) *Node[S]
RetryLoopNode 创建重试循环节点
func RetryNode ¶
func RetryNode[S State](name string, handler NodeHandler[S], policy *RetryPolicy) *Node[S]
RetryNode 创建带重试的节点 包装一个普通节点,添加重试能力
参数:
- name: 节点名称
- handler: 节点处理函数
- policy: 重试策略
func SubgraphNode ¶
func SubgraphNode[S State](name string, subgraph *Graph[S], stateMapper ...*SubgraphStateMapper[S]) *Node[S]
SubgraphNode 创建子图节点
将一个完整的图作为当前图的一个节点使用。 子图的输入是父节点的状态,输出合并回父状态。
参数:
- name: 节点名称
- subgraph: 子图
- stateMapper: 可选的状态映射函数
func TimeoutNode ¶
func TimeoutNode[S State](name string, handler NodeHandler[S], timeoutMs int64) *Node[S]
TimeoutNode 创建带超时的节点 包装一个普通节点,添加超时控制
参数:
- name: 节点名称
- handler: 节点处理函数
- timeoutMs: 超时时间(毫秒)
func UntilLoopNode ¶
func UntilLoopNode[S State](name string, condition func(S) bool, body NodeHandler[S], config ...*LoopConfig) *Node[S]
UntilLoopNode 创建 until 循环节点
循环执行直到条件满足
func WhileLoopNode ¶
func WhileLoopNode[S State](name string, condition func(S) bool, body NodeHandler[S], config ...*LoopConfig) *Node[S]
WhileLoopNode 创建 while 循环节点
循环执行 body 直到 condition 返回 false
示例:
loop := WhileLoopNode[MyState](
"retry-loop",
func(s MyState) bool { return s.RetryCount < 3 }, // 条件
func(ctx context.Context, s MyState) (MyState, error) { // 循环体
s.RetryCount++
return s, nil
},
)
type NodeBuilder ¶
type NodeBuilder[S State] struct { // contains filtered or unexported fields }
NodeBuilder 节点构建器
func NewNode ¶
func NewNode[S State](name string, handler NodeHandler[S]) *NodeBuilder[S]
NewNode 创建节点构建器
func (*NodeBuilder[S]) WithMetadata ¶
func (b *NodeBuilder[S]) WithMetadata(key string, value any) *NodeBuilder[S]
WithMetadata 设置元数据
func (*NodeBuilder[S]) WithRetry ¶
func (b *NodeBuilder[S]) WithRetry(policy *RetryPolicy) *NodeBuilder[S]
WithRetry 设置重试策略
func (*NodeBuilder[S]) WithTimeout ¶
func (b *NodeBuilder[S]) WithTimeout(ms int64) *NodeBuilder[S]
WithTimeout 设置超时时间(毫秒)
func (*NodeBuilder[S]) WithType ¶
func (b *NodeBuilder[S]) WithType(t NodeType) *NodeBuilder[S]
WithType 设置节点类型
type NodeCache ¶
type NodeCache interface {
// Get 获取缓存的状态
// key 是基于节点输入计算的哈希值
// 返回缓存的状态和是否命中
Get(key string) (any, bool)
// Set 设置缓存
Set(key string, value any)
// Delete 删除缓存
Delete(key string)
// Clear 清空所有缓存
Clear()
// Stats 返回缓存统计信息
Stats() CacheStats
}
NodeCache 节点缓存接口 用于缓存节点的执行结果,避免重复计算
type NodeHandler ¶
NodeHandler 节点处理函数类型
func CachedNodeHandler ¶
func CachedNodeHandler[S State](nodeName string, handler NodeHandler[S], cache NodeCache) NodeHandler[S]
CachedNodeHandler 创建带缓存的节点处理函数 包装原始 handler,自动检查和更新缓存
type NodePlacement ¶
type NodePlacement struct {
// NodeName 节点名称
NodeName string
// ExecutorName 远程执行器名称
ExecutorName string
// Fallback 是否在远程不可用时降级到本地执行
Fallback bool
}
NodePlacement 节点放置配置
type NodeResult ¶
type NodeResult[S State] struct { // State 更新后的状态 State S // NextNodes 下一个要执行的节点(用于条件路由) NextNodes []string // Error 执行错误 Error error // Metadata 执行元数据 Metadata map[string]any }
NodeResult 节点执行结果
type NodeStats ¶
type NodeStats struct {
// Executions 执行次数
Executions int64
// TotalDuration 总执行时间
TotalDuration time.Duration
// AverageDuration 平均执行时间
AverageDuration time.Duration
// Errors 错误次数
Errors int64
// LastError 最后一次错误
LastError error
// LastExecution 最后一次执行时间
LastExecution time.Time
}
NodeStats 节点统计
type NodeType ¶
type NodeType int
NodeType 节点类型
const ( // NodeTypeNormal 普通节点 NodeTypeNormal NodeType = iota // NodeTypeStart 起始节点 NodeTypeStart // NodeTypeEnd 结束节点 NodeTypeEnd // NodeTypeConditional 条件节点 NodeTypeConditional // NodeTypeParallel 并行节点 NodeTypeParallel // NodeTypeSubgraph 子图节点 NodeTypeSubgraph // NodeTypeCatch 错误捕获节点 NodeTypeCatch // NodeTypeRetry 重试节点 NodeTypeRetry // NodeTypeFallback 降级节点 NodeTypeFallback // NodeTypeLoop 循环节点 NodeTypeLoop )
const NodeTypeBarrier NodeType = 100
NodeTypeBarrier 屏障节点类型
type OverwriteReducer ¶
OverwriteReducer 覆盖式合并器 新值直接覆盖旧值
func (OverwriteReducer[S, V]) Reduce ¶
func (r OverwriteReducer[S, V]) Reduce(state S, key string, value V) S
type PregelConfig ¶
type PregelConfig struct {
// MaxSupersteps 最大超级步数量(防止无限循环)
// 默认 100
MaxSupersteps int
// TriggerMode 默认触发模式
// 默认 TriggerAnyPredecessor
TriggerMode TriggerMode
// ParallelExecution 是否并行执行同一超级步内的节点
// 默认 true
ParallelExecution bool
// TerminationCheck 终止检查函数
// 返回 true 表示应该终止迭代
// 如果为 nil,则只检查是否到达 END 节点
TerminationCheck func(step int, activeNodes []string) bool
// Debug 调试模式
Debug bool
}
PregelConfig Pregel 执行配置
func DefaultPregelConfig ¶
func DefaultPregelConfig() *PregelConfig
DefaultPregelConfig 返回默认的 Pregel 配置
type PregelEvent ¶
type PregelEvent[S State] struct { // Type 事件类型 Type PregelEventType // Superstep 当前超级步编号 Superstep int // ActiveNodes 活跃节点列表 ActiveNodes []string // State 当前状态 State S // Error 错误信息 Error error }
PregelEvent Pregel 执行事件
type PregelEventType ¶
type PregelEventType int
PregelEventType Pregel 事件类型
const ( // PregelEventSuperstepStart 超级步开始 PregelEventSuperstepStart PregelEventType = iota // PregelEventSuperstepEnd 超级步结束 PregelEventSuperstepEnd // PregelEventComplete 执行完成 PregelEventComplete // PregelEventError 错误 PregelEventError )
type PregelExecutor ¶
type PregelExecutor[S State] struct { // contains filtered or unexported fields }
PregelExecutor Pregel 图执行器 支持循环图的迭代执行
func NewPregelExecutor ¶
func NewPregelExecutor[S State](g *Graph[S], opts ...PregelOption) *PregelExecutor[S]
NewPregelExecutor 创建 Pregel 执行器
func NewPregelExecutorWithMerger ¶
func NewPregelExecutorWithMerger[S State](g *Graph[S], merger PregelStateMerger[S], opts ...PregelOption) *PregelExecutor[S]
NewPregelExecutorWithMerger 创建带状态合并器的 Pregel 执行器
type PregelExecutorOption ¶
type PregelExecutorOption[S State] func(*PregelExecutor[S])
PregelExecutorOption PregelExecutor 配置选项
func WithPregelMerger ¶
func WithPregelMerger[S State](merger PregelStateMerger[S]) PregelExecutorOption[S]
WithPregelMerger 设置状态合并器 用于在并行执行时正确合并多个节点的输出状态
type PregelOption ¶
type PregelOption func(*PregelConfig)
PregelOption Pregel 配置选项
func WithParallelExecution ¶
func WithParallelExecution(parallel bool) PregelOption
WithParallelExecution 设置是否并行执行
func WithPregelTriggerMode ¶
func WithPregelTriggerMode(mode TriggerMode) PregelOption
WithPregelTriggerMode 设置触发模式
func WithTerminationCheck ¶
func WithTerminationCheck(fn func(step int, activeNodes []string) bool) PregelOption
WithTerminationCheck 设置终止检查函数
type PregelStateMerger ¶
type PregelStateMerger[S State] interface { // Merge 合并多个状态为一个 // base 是执行前的基础状态 // states 是各节点执行后的状态列表 Merge(base S, states []S) S }
PregelStateMerger 状态合并器接口 用于在 Pregel 并行执行时合并多个节点的输出状态
type RedisCheckpointOption ¶
type RedisCheckpointOption func(*RedisCheckpointSaver)
RedisCheckpointOption 是 RedisCheckpointSaver 的配置选项
type RedisCheckpointSaver ¶
type RedisCheckpointSaver struct {
// contains filtered or unexported fields
}
RedisCheckpointSaver 基于 Redis 的检查点保存器
func NewRedisCheckpointSaver ¶
func NewRedisCheckpointSaver(client *redis.Client, opts ...RedisCheckpointOption) *RedisCheckpointSaver
NewRedisCheckpointSaver 创建基于 Redis 的检查点保存器
func NewRedisCheckpointSaverFromURL ¶
func NewRedisCheckpointSaverFromURL(redisURL string, opts ...RedisCheckpointOption) (*RedisCheckpointSaver, error)
NewRedisCheckpointSaverFromURL 从 URL 创建 Redis 检查点保存器
func (*RedisCheckpointSaver) Delete ¶
func (s *RedisCheckpointSaver) Delete(ctx context.Context, id string) error
Delete 删除检查点
func (*RedisCheckpointSaver) DeleteThread ¶
func (s *RedisCheckpointSaver) DeleteThread(ctx context.Context, threadID string) error
DeleteThread 删除线程的所有检查点
func (*RedisCheckpointSaver) GetCheckpointCount ¶
func (s *RedisCheckpointSaver) GetCheckpointCount(ctx context.Context, threadID string) (int64, error)
GetCheckpointCount 获取线程的检查点数量
func (*RedisCheckpointSaver) List ¶
func (s *RedisCheckpointSaver) List(ctx context.Context, threadID string) ([]*Checkpoint, error)
List 列出线程的所有检查点
func (*RedisCheckpointSaver) ListThreads ¶
func (s *RedisCheckpointSaver) ListThreads(ctx context.Context, pattern string, limit int64) ([]string, error)
ListThreads 列出所有线程 ID
func (*RedisCheckpointSaver) Load ¶
func (s *RedisCheckpointSaver) Load(ctx context.Context, threadID string) (*Checkpoint, error)
Load 加载最新的检查点
func (*RedisCheckpointSaver) LoadByID ¶
func (s *RedisCheckpointSaver) LoadByID(ctx context.Context, id string) (*Checkpoint, error)
LoadByID 根据 ID 加载检查点
func (*RedisCheckpointSaver) LoadByThreadIDWithWarnings ¶
func (s *RedisCheckpointSaver) LoadByThreadIDWithWarnings(ctx context.Context, threadID string) ([]*Checkpoint, []error, error)
LoadByThreadIDWithWarnings 加载线程的所有检查点,同时返回解析警告
func (*RedisCheckpointSaver) Save ¶
func (s *RedisCheckpointSaver) Save(ctx context.Context, checkpoint *Checkpoint) error
Save 保存检查点
type ReduceFunc ¶
type ReduceFunc[S State] func(original S, results []S) S
ReduceFunc 归约函数 将所有 map 结果归约为最终状态
type Reducer ¶
type Reducer[S State, V any] interface { // Reduce 将新值合并到状态中 Reduce(state S, key string, value V) S }
Reducer 状态合并器 用于合并多个节点的输出到状态中
type RemoteNodeExecutor ¶
type RemoteNodeExecutor interface {
// Execute 远程执行节点
// nodeName 节点名称
// stateData 序列化后的状态数据
// 返回序列化后的结果状态数据
Execute(ctx context.Context, nodeName string, stateData []byte) ([]byte, error)
// Ping 检查远程节点是否可用
Ping(ctx context.Context) error
// Name 返回执行器名称
Name() string
}
RemoteNodeExecutor 远程节点执行器接口 实现此接口以支持不同的远程执行方式(HTTP、gRPC、消息队列等)
type RemoteRegistry ¶
type RemoteRegistry struct {
// contains filtered or unexported fields
}
RemoteRegistry 远程执行器注册表
func (*RemoteRegistry) Get ¶
func (r *RemoteRegistry) Get(name string) (RemoteNodeExecutor, bool)
Get 获取远程执行器
func (*RemoteRegistry) HealthCheck ¶
func (r *RemoteRegistry) HealthCheck(ctx context.Context) map[string]error
HealthCheck 检查所有远程节点的健康状态
func (*RemoteRegistry) Register ¶
func (r *RemoteRegistry) Register(name string, executor RemoteNodeExecutor)
Register 注册远程执行器
type RetryConfig ¶
type RetryConfig struct {
// MaxRetries 最大重试次数
MaxRetries int
// Delay 基础延迟
Delay time.Duration
// MaxDelay 最大延迟
MaxDelay time.Duration
// Backoff 退避因子
Backoff float64
// Jitter 抖动比例 (0-1)
Jitter float64
// ShouldRetry 判断是否应该重试
ShouldRetry func(err error) bool
// OnRetry 重试回调
OnRetry func(attempt int, err error, delay time.Duration)
}
RetryConfig 重试配置
type RetryPolicy ¶
type RetryPolicy struct {
// MaxRetries 最大重试次数
MaxRetries int
// InitialDelay 初始延迟(毫秒)
InitialDelay int64
// MaxDelay 最大延迟(毫秒)
MaxDelay int64
// Multiplier 延迟倍数
Multiplier float64
// RetryOn 重试条件(返回 true 表示需要重试)
RetryOn func(error) bool
}
RetryPolicy 重试策略
type RouterFunc ¶
RouterFunc 路由函数类型
type SnapshotStorage ¶
type SnapshotStorage interface {
// Save 保存快照
Save(ctx context.Context, snapshot *StateSnapshot) error
// Load 加载快照
Load(ctx context.Context, index int) (*StateSnapshot, error)
// LoadRange 加载范围内的快照
LoadRange(ctx context.Context, start, end int) ([]*StateSnapshot, error)
// Delete 删除快照
Delete(ctx context.Context, index int) error
// Clear 清空所有快照
Clear(ctx context.Context) error
}
SnapshotStorage 快照存储接口
type SplitFunc ¶
type SplitFunc[S State] func(state S) []S
SplitFunc 数据分片函数 将输入状态分割为多个子状态(每个子状态由一个 map worker 处理)
type State ¶
type State interface {
// Clone 创建状态的深拷贝
Clone() State
}
State 是图状态的约束接口 所有状态类型都必须实现此接口
type StateAnnotation ¶
type StateAnnotation struct {
Channels map[string]ChannelConfig
}
StateAnnotation 状态注解 用于定义状态结构和通道
type StateDiff ¶
type StateDiff struct {
// Key 键名
Key string `json:"key"`
// OldValue 旧值
OldValue any `json:"old_value,omitempty"`
// NewValue 新值
NewValue any `json:"new_value,omitempty"`
// Type 差异类型:added, removed, changed
Type string `json:"type"`
}
StateDiff 状态差异
type StateMachine ¶
type StateMachine[S any] struct { // contains filtered or unexported fields }
StateMachine 状态机
func NewStateMachine ¶
func NewStateMachine[S any](name string) *StateMachine[S]
NewStateMachine 创建状态机
func (*StateMachine[S]) AddFinal ¶
func (sm *StateMachine[S]) AddFinal(names ...string) *StateMachine[S]
AddFinal 添加终态
func (*StateMachine[S]) AddState ¶
func (sm *StateMachine[S]) AddState(name string, handler func(ctx context.Context, state S) (string, error)) *StateMachine[S]
AddState 添加状态
func (*StateMachine[S]) AddStateWithHooks ¶
func (sm *StateMachine[S]) AddStateWithHooks( name string, handler func(ctx context.Context, state S) (string, error), onEnter func(ctx context.Context, state S) error, onExit func(ctx context.Context, state S) error, ) *StateMachine[S]
AddStateWithHooks 添加带钩子的状态
func (*StateMachine[S]) AddTransition ¶
func (sm *StateMachine[S]) AddTransition(from, to string, condition func(ctx context.Context, state S) bool) *StateMachine[S]
AddTransition 添加转换
func (*StateMachine[S]) AddTransitionWithPriority ¶
func (sm *StateMachine[S]) AddTransitionWithPriority(from, to string, condition func(ctx context.Context, state S) bool, priority int) *StateMachine[S]
AddTransitionWithPriority 添加带优先级的转换
func (*StateMachine[S]) GetPending ¶
func (sm *StateMachine[S]) GetPending(threadID string) *interrupt.PendingInfo
GetPending 获取待处理的中断
func (*StateMachine[S]) Resume ¶
func (sm *StateMachine[S]) Resume(ctx context.Context, threadID string, cmd interrupt.Command) error
Resume 恢复执行
func (*StateMachine[S]) Run ¶
func (sm *StateMachine[S]) Run(ctx context.Context, initialState S) (S, error)
Run 运行状态机
func (*StateMachine[S]) RunWithThreadID ¶
func (sm *StateMachine[S]) RunWithThreadID(ctx context.Context, threadID string, initialState S) (S, error)
RunWithThreadID 带线程 ID 运行状态机
func (*StateMachine[S]) RunWithTrace ¶
func (sm *StateMachine[S]) RunWithTrace(ctx context.Context, initialState S) (S, *ExecutionTrace, error)
RunWithTrace 带追踪运行
func (*StateMachine[S]) SetCheckpointer ¶
func (sm *StateMachine[S]) SetCheckpointer(cp checkpoint.Checkpointer) *StateMachine[S]
SetCheckpointer 设置检查点存储
func (*StateMachine[S]) SetInitial ¶
func (sm *StateMachine[S]) SetInitial(name string) *StateMachine[S]
SetInitial 设置初始状态
func (*StateMachine[S]) SetMaxSteps ¶
func (sm *StateMachine[S]) SetMaxSteps(max int) *StateMachine[S]
SetMaxSteps 设置最大步数
type StateMachineBuilder ¶
type StateMachineBuilder[S any] struct { // contains filtered or unexported fields }
StateMachineBuilder 状态机构建器(更流畅的 API)
func (*StateMachineBuilder[S]) Build ¶
func (b *StateMachineBuilder[S]) Build() *StateMachine[S]
Build 构建状态机
func (*StateMachineBuilder[S]) Checkpointer ¶
func (b *StateMachineBuilder[S]) Checkpointer(cp checkpoint.Checkpointer) *StateMachineBuilder[S]
Checkpointer 设置检查点存储
func (*StateMachineBuilder[S]) Final ¶
func (b *StateMachineBuilder[S]) Final(names ...string) *StateMachineBuilder[S]
Final 添加终态
func (*StateMachineBuilder[S]) Initial ¶
func (b *StateMachineBuilder[S]) Initial(name string) *StateMachineBuilder[S]
Initial 设置初始状态
func (*StateMachineBuilder[S]) MaxSteps ¶
func (b *StateMachineBuilder[S]) MaxSteps(max int) *StateMachineBuilder[S]
MaxSteps 设置最大步数
func (*StateMachineBuilder[S]) State ¶
func (b *StateMachineBuilder[S]) State(name string, handler func(ctx context.Context, state S) (string, error)) *StateMachineBuilder[S]
State 添加状态
func (*StateMachineBuilder[S]) Transition ¶
func (b *StateMachineBuilder[S]) Transition(from, to string, condition func(ctx context.Context, state S) bool) *StateMachineBuilder[S]
Transition 添加转换
type StateMerger ¶
type StateMerger[S State] func(original S, outputs []S) S
StateMerger 状态合并函数
将多个子图的输出状态合并为一个状态
type StateNode ¶
type StateNode[S any] struct { Name string OnEnter func(ctx context.Context, state S) error OnExit func(ctx context.Context, state S) error Handler func(ctx context.Context, state S) (string, error) // 返回下一状态名或空字符串使用转换 }
StateNode 状态节点定义
type StateSnapshot ¶
type StateSnapshot struct {
// Index 快照索引
Index int `json:"index"`
// Timestamp 时间戳
Timestamp time.Time `json:"timestamp"`
// NodeID 执行的节点 ID
NodeID string `json:"node_id"`
// NodeName 节点名称
NodeName string `json:"node_name"`
// State 状态数据(深拷贝)
State map[string]any `json:"state"`
// Input 节点输入
Input any `json:"input,omitempty"`
// Output 节点输出
Output any `json:"output,omitempty"`
// Error 错误信息
Error string `json:"error,omitempty"`
// Duration 执行耗时
Duration time.Duration `json:"duration"`
// Metadata 元数据
Metadata map[string]any `json:"metadata,omitempty"`
// ParentIndex 父快照索引(用于分支)
ParentIndex int `json:"parent_index,omitempty"`
// BranchID 分支 ID
BranchID string `json:"branch_id,omitempty"`
}
StateSnapshot 状态快照
type StreamChannel ¶
type StreamChannel struct {
// contains filtered or unexported fields
}
StreamChannel 流式事件通道 提供对图执行过程中产生的事件的读取能力
func NewStreamChannel ¶
func NewStreamChannel(bufferSize int) *StreamChannel
NewStreamChannel 创建流式事件通道
func (*StreamChannel) Emit ¶
func (sc *StreamChannel) Emit(event StreamModeEvent)
Emit 发射一个事件 如果通道已关闭或缓冲区已满,事件将被丢弃
func (*StreamChannel) Events ¶
func (sc *StreamChannel) Events() <-chan StreamModeEvent
Events 返回事件读取通道
type StreamEvent ¶
type StreamEvent[S State] struct { // Type 事件类型 Type EventType // NodeName 节点名称 NodeName string // State 当前状态 State S // Error 错误(仅用于 EventTypeError) Error error // Metadata 元数据 Metadata map[string]any }
StreamEvent 流事件
type StreamMode ¶
type StreamMode int
StreamMode 流式输出模式
const ( // StreamModeValues 完整状态快照模式 // 每个节点执行后输出完整的当前状态 StreamModeValues StreamMode = iota // StreamModeUpdates 增量更新模式 // 每个节点执行后只输出变化的部分 StreamModeUpdates // StreamModeMessages LLM 消息流模式 // 输出 LLM 生成的每个 token StreamModeMessages // StreamModeCustom 自定义事件模式 // 节点可以发射自定义事件 StreamModeCustom // StreamModeDebug 调试模式 // 输出详细的执行信息(节点开始/结束/耗时/错误) StreamModeDebug )
type StreamModeEvent ¶
type StreamModeEvent struct {
// Mode 事件对应的流模式
Mode StreamMode `json:"mode"`
// Node 产生事件的节点名称
Node string `json:"node"`
// Type 事件类型
Type StreamModeEventType `json:"type"`
// Data 事件数据
Data any `json:"data,omitempty"`
// Timestamp 事件时间戳
Timestamp time.Time `json:"timestamp"`
// Metadata 额外元数据
Metadata map[string]any `json:"metadata,omitempty"`
}
StreamModeEvent 流式事件
type StreamModeEventType ¶
type StreamModeEventType string
StreamModeEventType 流式事件类型
const ( // EventNodeStart 节点开始执行 EventNodeStart StreamModeEventType = "node_start" // EventNodeEnd 节点执行结束 EventNodeEnd StreamModeEventType = "node_end" // EventNodeError 节点执行错误 EventNodeError StreamModeEventType = "node_error" // EventStateUpdate 状态更新 EventStateUpdate StreamModeEventType = "state_update" // EventStateSnapshot 状态快照 EventStateSnapshot StreamModeEventType = "state_snapshot" // EventMessage LLM 消息 EventMessage StreamModeEventType = "message" // EventToken LLM Token EventToken StreamModeEventType = "token" // EventCustom 自定义事件 EventCustom StreamModeEventType = "custom" // EventGraphStart 图开始执行 EventGraphStart StreamModeEventType = "graph_start" // EventGraphEnd 图执行结束 EventGraphEnd StreamModeEventType = "graph_end" )
type StreamRunOption ¶
type StreamRunOption func(*streamRunConfig)
StreamRunOption 流式执行选项
func WithStreamBufferSize ¶
func WithStreamBufferSize(size int) StreamRunOption
WithStreamBufferSize 设置事件缓冲区大小
func WithStreamFilter ¶
func WithStreamFilter(filter func(StreamModeEvent) bool) StreamRunOption
WithStreamFilter 设置事件过滤器
func WithStreamMode ¶
func WithStreamMode(modes ...StreamMode) StreamRunOption
WithStreamMode 设置流式模式
type SubgraphStateMapper ¶
type SubgraphStateMapper[S State] struct { // Input 输入映射:将父状态转换为子图输入状态 Input func(parentState S) S // Output 输出映射:将子图输出合并到父状态 Output func(parentState, subgraphOutput S) S }
SubgraphStateMapper 子图状态映射器
用于在父图和子图之间转换状态
type TaskDef ¶
TaskDef 任务定义
func DefineTask ¶
func DefineTask[S State](wf *Workflow[S], name string, handler func(ctx context.Context, state S) (S, error), opts ...TaskOption[S]) *TaskDef[S]
DefineTask 定义一个任务 返回 TaskDef 可在 entrypoint 中调用 Run 执行
type TaskOption ¶
TaskOption 任务选项
func WithTaskCache ¶
func WithTaskCache[S State](cache NodeCache) TaskOption[S]
WithTaskCache 为任务添加缓存
func WithTaskCacheKey ¶
func WithTaskCacheKey[S State](keyFunc func(state S) string) TaskOption[S]
WithTaskCacheKey 自定义任务缓存 key 生成
type ThreadConfig ¶
type ThreadConfig struct {
// ThreadID 线程 ID
ThreadID string
// CheckpointSaver 检查点保存器
CheckpointSaver CheckpointSaver
// ResumeFromCheckpoint 是否从检查点恢复
ResumeFromCheckpoint bool
// CheckpointID 要恢复的检查点 ID(可选)
CheckpointID string
}
ThreadConfig 线程配置
func (*ThreadConfig) WithCheckpointSaver ¶
func (c *ThreadConfig) WithCheckpointSaver(saver CheckpointSaver) *ThreadConfig
WithCheckpointSaver 设置检查点保存器
func (*ThreadConfig) WithResume ¶
func (c *ThreadConfig) WithResume(checkpointID string) *ThreadConfig
WithResume 设置从检查点恢复
type TimeTravelDebugger ¶
type TimeTravelDebugger struct {
// contains filtered or unexported fields
}
TimeTravelDebugger 时间旅行调试器
功能:
- 记录每一步执行的状态快照
- 支持回溯到任意历史状态
- 支持从历史状态重新执行
- 支持状态差异对比
使用示例:
debugger := NewTimeTravelDebugger(graphAdapter) result, _ := debugger.Run(ctx, initialState) // 查看历史记录 history := debugger.GetHistory() // 回溯到某个时间点 debugger.GoTo(3) // 从该点重新执行 result2, _ := debugger.Replay(ctx)
func NewTimeTravelDebugger ¶
func NewTimeTravelDebugger(exec Executable, opts ...TimeTravelOption) *TimeTravelDebugger
NewTimeTravelDebugger 创建时间旅行调试器
func (*TimeTravelDebugger) Compare ¶
func (d *TimeTravelDebugger) Compare(index1, index2 int) ([]StateDiff, error)
Compare 比较两个快照的差异
func (*TimeTravelDebugger) CurrentIndex ¶
func (d *TimeTravelDebugger) CurrentIndex() int
CurrentIndex 获取当前索引
func (*TimeTravelDebugger) Export ¶
func (d *TimeTravelDebugger) Export() ([]byte, error)
Export 导出历史记录
func (*TimeTravelDebugger) FindByNodeID ¶
func (d *TimeTravelDebugger) FindByNodeID(nodeID string) []*StateSnapshot
FindByNodeID 按节点 ID 查找
func (*TimeTravelDebugger) FindByTimeRange ¶
func (d *TimeTravelDebugger) FindByTimeRange(start, end time.Time) []*StateSnapshot
FindByTimeRange 按时间范围查找
func (*TimeTravelDebugger) FindErrors ¶
func (d *TimeTravelDebugger) FindErrors() []*StateSnapshot
FindErrors 查找错误
func (*TimeTravelDebugger) GetBranchHistory ¶
func (d *TimeTravelDebugger) GetBranchHistory(branchID string) []*StateSnapshot
GetBranchHistory 获取指定分支的历史
func (*TimeTravelDebugger) GetBranches ¶
func (d *TimeTravelDebugger) GetBranches() []string
GetBranches 获取所有分支
func (*TimeTravelDebugger) GetDebugView ¶
func (d *TimeTravelDebugger) GetDebugView() *DebugView
GetDebugView 获取调试视图
func (*TimeTravelDebugger) GetHistory ¶
func (d *TimeTravelDebugger) GetHistory() []*StateSnapshot
GetHistory 获取历史记录
func (*TimeTravelDebugger) GetSnapshot ¶
func (d *TimeTravelDebugger) GetSnapshot(index int) (*StateSnapshot, error)
GetSnapshot 获取指定索引的快照
func (*TimeTravelDebugger) GoForward ¶
func (d *TimeTravelDebugger) GoForward() error
GoForward 前进一步
func (*TimeTravelDebugger) Import ¶
func (d *TimeTravelDebugger) Import(data []byte) error
Import 导入历史记录
func (*TimeTravelDebugger) ReplayFrom ¶
ReplayFrom 从指定快照重新执行
func (*TimeTravelDebugger) Run ¶
func (d *TimeTravelDebugger) Run(ctx context.Context, initialState map[string]any) (map[string]any, error)
Run 执行并记录历史
func (*TimeTravelDebugger) SearchSnapshots ¶
func (d *TimeTravelDebugger) SearchSnapshots(predicate func(*StateSnapshot) bool) []*StateSnapshot
SearchSnapshots 搜索快照
type TimeTravelOption ¶
type TimeTravelOption func(*TimeTravelDebugger)
TimeTravelOption 时间旅行调试器选项
func WithErrorHandler ¶
func WithErrorHandler(handler func(error)) TimeTravelOption
WithErrorHandler 设置错误处理函数 用于处理存储失败等非致命错误
func WithSnapshotStorage ¶
func WithSnapshotStorage(storage SnapshotStorage) TimeTravelOption
WithSnapshotStorage 设置快照存储
type Transition ¶
type Transition[S any] struct { From string To string Condition func(ctx context.Context, state S) bool Priority int // 优先级,数值越小越先检查 }
Transition 状态转换
type TriggerMode ¶
type TriggerMode int
TriggerMode 节点触发模式
const ( // TriggerAllPredecessors 所有前驱完成后触发(默认 DAG 模式) // 节点等待所有入边的前驱节点都执行完成后才开始执行 TriggerAllPredecessors TriggerMode = iota // TriggerAnyPredecessor 任意前驱完成后触发(Pregel 模式) // 节点在任意一个前驱节点完成后就可以开始执行 // 适用于循环图,允许节点在不同轮次多次执行 TriggerAnyPredecessor )