chain

package
v0.5.13 Latest Latest
Warning

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

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

Documentation

Overview

Package chain 提供 Hexagon AI Agent 框架的链式编排

Chain 是一种简化的编排模式，将多个组件按顺序串联执行。 相比 Graph，Chain 更加简单直观，适合简单的线性流程。

基本用法： 

chain := NewChain[Input, Output]("my-chain").
    Pipe(step1).
    Pipe(step2).
    Pipe(step3).
    Build()

result, err := chain.Invoke(ctx, input)

Package chain 提供 Hexagon AI Agent 框架的链式编排

本文件提供链的构建期 I/O 类型校验：把相邻节点的输出/输入反射类型记录下来， 在 Compile 时逐对校验，类型不匹配立即报错——而不是等到运行时数据流到中途 才因 %T 断言失败而中断。

三条兼容规则（上游输出 out 能否流入下游输入 in）：

  • 相等：out == in（含 out 可直接赋值给 in）
  • 下游接口被实现：in 是接口且 out 实现它
  • 上游接口留运行期：out 是接口（具体值运行期才确定）→ 不在构建期否决，留运行期断言

typed.go 提供编译时类型安全的管道组合

Go 泛型限制无法支持可变长度类型参数，因此提供 Pipe2/Pipe3/Pipe4 三个固定长度版本。 每个 Pipe 函数将多个类型安全的函数串联，编译时即可检查中间类型匹配。

使用示例： 

// 类型安全的三步管道：string → int → bool
step := Pipe3[string, int, bool, string]("pipeline",
    func(ctx context.Context, s string) (int, error) { return len(s), nil },
    func(ctx context.Context, n int) (bool, error) { return n > 5, nil },
    func(ctx context.Context, b bool) (string, error) { return fmt.Sprintf("%v", b), nil },
)

chain := NewTypedChain("my-chain", step)
result, err := chain.Invoke(ctx, "hello world")

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Chain 

type Chain[I, O any] struct {
	// contains filtered or unexported fields
}

Chain 链式组件

func (*Chain[I, O]) Batch 

func (c *Chain[I, O]) Batch(ctx context.Context, inputs []I, opts ...core.Option) ([]O, error)

Batch 批量执行链

func (*Chain[I, O]) BatchStream 

func (c *Chain[I, O]) BatchStream(ctx context.Context, inputs []I, opts ...core.Option) (*stream.StreamReader[O], error)

BatchStream 批量流式执行

func (*Chain[I, O]) Collect 

func (c *Chain[I, O]) Collect(ctx context.Context, input *stream.StreamReader[I], opts ...core.Option) (O, error)

Collect 收集流式输入并执行

func (*Chain[I, O]) Description 

func (c *Chain[I, O]) Description() string

Description 返回链描述

func (*Chain[I, O]) InputSchema 

func (c *Chain[I, O]) InputSchema() *core.Schema

InputSchema 返回输入 Schema

func (*Chain[I, O]) Invoke 

func (c *Chain[I, O]) Invoke(ctx context.Context, input I, opts ...core.Option) (O, error)

Invoke 执行链

func (*Chain[I, O]) Name 

func (c *Chain[I, O]) Name() string

Name 返回链名称

func (*Chain[I, O]) OutputSchema 

func (c *Chain[I, O]) OutputSchema() *core.Schema

OutputSchema 返回输出 Schema

func (*Chain[I, O]) Stream 

func (c *Chain[I, O]) Stream(ctx context.Context, input I, opts ...core.Option) (*stream.StreamReader[O], error)

Stream 流式执行链

func (*Chain[I, O]) Transform 

func (c *Chain[I, O]) Transform(ctx context.Context, input *stream.StreamReader[I], opts ...core.Option) (*stream.StreamReader[O], error)

Transform 转换流

type ChainBuilder 

type ChainBuilder[I, O any] struct {
	// contains filtered or unexported fields
}

ChainBuilder 链构建器

func NewChain 

func NewChain[I, O any](name string) *ChainBuilder[I, O]

NewChain 创建链构建器

func (*ChainBuilder[I, O]) Build 

func (b *ChainBuilder[I, O]) Build() (*Chain[I, O], error)

Build 构建链

func (*ChainBuilder[I, O]) MustBuild 

func (b *ChainBuilder[I, O]) MustBuild() *Chain[I, O]

MustBuild 构建链，失败时 panic

⚠️ 警告：构建失败时会 panic。 仅在初始化时使用，不要在运行时调用。 推荐使用 Build() 方法并正确处理错误。

使用场景：

  • 程序启动时的全局初始化
  • 测试代码中

func (*ChainBuilder[I, O]) Pipe 

func (b *ChainBuilder[I, O]) Pipe(r core.Runnable[any, any]) *ChainBuilder[I, O]

Pipe 添加 Runnable 到链中

func (*ChainBuilder[I, O]) PipeFunc 

func (b *ChainBuilder[I, O]) PipeFunc(name string, fn func(ctx context.Context, input any) (any, error)) *ChainBuilder[I, O]

PipeFunc 添加函数到链中

func (*ChainBuilder[I, O]) Use 

func (b *ChainBuilder[I, O]) Use(middleware ...Middleware) *ChainBuilder[I, O]

Use 添加中间件

func (*ChainBuilder[I, O]) WithDescription 

func (b *ChainBuilder[I, O]) WithDescription(desc string) *ChainBuilder[I, O]

WithDescription 设置描述

type CompileError 

type CompileError struct {
	FromNode string
	ToNode   string
	OutType  reflect.Type
	InType   reflect.Type
}

CompileError 描述 Compile 期发现的相邻节点 I/O 类型不匹配。

func (*CompileError) Error 

func (e *CompileError) Error() string

type CompiledChain 

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

CompiledChain 是通过 I/O 类型校验后的可执行链。

func Compile 

func Compile(nodes ...TypedNode) (*CompiledChain, error)

Compile 校验相邻节点的 I/O 类型（三规则），任一对不兼容即在构建期返回 *CompileError， 不必等运行时数据流到中途才暴露类型错误。校验通过返回可执行的 CompiledChain。

func (*CompiledChain) Len 

func (c *CompiledChain) Len() int

Len 返回链中节点数量。

func (*CompiledChain) Run 

func (c *CompiledChain) Run(ctx context.Context, input any) (any, error)

Run 顺序执行编译后的链：每个节点的输出作为下一个节点的输入。

type Middleware 

type Middleware func(next StepFunc) StepFunc

Middleware 中间件

func LoggingMiddleware 

func LoggingMiddleware(logger func(name string, input, output any, err error)) Middleware

LoggingMiddleware 日志中间件

func RecoverMiddleware 

func RecoverMiddleware() Middleware

RecoverMiddleware panic 恢复中间件

func RetryMiddleware 

func RetryMiddleware(maxRetries int, shouldRetry func(error) bool) Middleware

RetryMiddleware 重试中间件

type Parallel 

type Parallel[I, O any] struct {
	// contains filtered or unexported fields
}

Parallel 并行执行多个组件

func NewParallel 

func NewParallel[I, O any](name string, merge func(results []O) O) *Parallel[I, O]

NewParallel 创建并行执行器

func (*Parallel[I, O]) Add 

func (p *Parallel[I, O]) Add(handler func(ctx context.Context, input I) (O, error)) *Parallel[I, O]

Add 添加并行执行的处理函数

func (*Parallel[I, O]) Invoke 

func (p *Parallel[I, O]) Invoke(ctx context.Context, input I, opts ...core.Option) (O, error)

Invoke 执行并行处理

type StepFunc 

type StepFunc func(ctx context.Context, input any) (any, error)

StepFunc 步骤函数

type TypeCompat 

type TypeCompat int

TypeCompat 描述相邻节点输出→输入的类型兼容判定结果。 

const (
	// TypeIncompatible 不兼容：Compile 期报错
	TypeIncompatible TypeCompat = iota
	// TypeEqual 相等或可直接赋值
	TypeEqual
	// TypeImplements 下游输入是接口且上游输出实现它
	TypeImplements
	// TypeRuntimeAssert 上游输出是接口（或类型未知）：编译期无法确定，留运行期断言
	TypeRuntimeAssert
)

type TypedChain 

type TypedChain[I, O any] struct {
	// contains filtered or unexported fields
}

TypedChain 类型安全的链 将 TypedStep 包装为完整的 Runnable[I, O] 实现

与普通 Chain 的区别：

  • Chain 内部步骤使用 any，运行时才发现类型错误
  • TypedChain 所有步骤的类型在编译时即被检查

线程安全：TypedChain 是不可变的，可安全并发使用

func NewTypedChain 

func NewTypedChain[I, O any](name string, step *TypedStep[I, O]) *TypedChain[I, O]

NewTypedChain 创建类型安全的链

参数：

  • name: 链名称
  • step: 通过 Pipe2/Pipe3/Pipe4/Then 组合的类型安全步骤

func (*TypedChain[I, O]) Batch 

func (c *TypedChain[I, O]) Batch(ctx context.Context, inputs []I, opts ...core.Option) ([]O, error)

Batch 批量执行链

func (*TypedChain[I, O]) BatchStream 

func (c *TypedChain[I, O]) BatchStream(ctx context.Context, inputs []I, opts ...core.Option) (*stream.StreamReader[O], error)

BatchStream 批量流式执行

func (*TypedChain[I, O]) Collect 

func (c *TypedChain[I, O]) Collect(ctx context.Context, input *stream.StreamReader[I], opts ...core.Option) (O, error)

Collect 收集流式输入并执行

func (*TypedChain[I, O]) Description 

func (c *TypedChain[I, O]) Description() string

Description 返回链描述

func (*TypedChain[I, O]) InputSchema 

func (c *TypedChain[I, O]) InputSchema() *core.Schema

InputSchema 返回输入 Schema

func (*TypedChain[I, O]) Invoke 

func (c *TypedChain[I, O]) Invoke(ctx context.Context, input I, opts ...core.Option) (O, error)

Invoke 同步执行链

func (*TypedChain[I, O]) Name 

func (c *TypedChain[I, O]) Name() string

Name 返回链名称

func (*TypedChain[I, O]) OutputSchema 

func (c *TypedChain[I, O]) OutputSchema() *core.Schema

OutputSchema 返回输出 Schema

func (*TypedChain[I, O]) Stream 

func (c *TypedChain[I, O]) Stream(ctx context.Context, input I, opts ...core.Option) (*stream.StreamReader[O], error)

Stream 流式执行链 将 Invoke 结果包装为单元素流

func (*TypedChain[I, O]) Transform 

func (c *TypedChain[I, O]) Transform(ctx context.Context, input *stream.StreamReader[I], opts ...core.Option) (*stream.StreamReader[O], error)

Transform 流转换：对每个输入元素应用链

func (*TypedChain[I, O]) WithDescription 

func (c *TypedChain[I, O]) WithDescription(desc string) *TypedChain[I, O]

WithDescription 设置链描述

type TypedNode 

type TypedNode struct {
	Name    string
	InType  reflect.Type
	OutType reflect.Type
	Handler func(ctx context.Context, input any) (any, error)
}

TypedNode 是带输入/输出反射类型的链节点，供 Compile 做相邻节点 I/O 校验。

func NewTypedNode 

func NewTypedNode[I, O any](name string, fn func(ctx context.Context, input I) (O, error)) TypedNode

NewTypedNode 从类型安全的处理函数创建带反射类型的节点。

节点的输入/输出类型由泛型参数 I/O 捕获，使 Compile 能在构建期校验相邻节点衔接。

type TypedStep 

type TypedStep[I, O any] struct {
	// contains filtered or unexported fields
}

TypedStep 类型安全的步骤

func NewTypedStep 

func NewTypedStep[I, O any](name string, handler func(ctx context.Context, input I) (O, error)) *TypedStep[I, O]

NewTypedStep 创建类型安全的步骤

func Pipe2 

func Pipe2[I, M, O any](
	name string,
	fn1 func(ctx context.Context, input I) (M, error),
	fn2 func(ctx context.Context, input M) (O, error),
) *TypedStep[I, O]

Pipe2 连接两个类型安全的函数，编译时检查类型匹配

类型参数：

  • I: 管道输入类型
  • M: 中间类型（fn1 输出 = fn2 输入）
  • O: 管道输出类型

返回的 TypedStep 可直接用于 NewTypedChain 或继续通过 Then 组合

func Pipe3 

func Pipe3[I, M1, M2, O any](
	name string,
	fn1 func(ctx context.Context, input I) (M1, error),
	fn2 func(ctx context.Context, input M1) (M2, error),
	fn3 func(ctx context.Context, input M2) (O, error),
) *TypedStep[I, O]

Pipe3 连接三个类型安全的函数

类型参数：

  • I: 管道输入类型
  • M1: 第一个中间类型（fn1 输出 = fn2 输入）
  • M2: 第二个中间类型（fn2 输出 = fn3 输入）
  • O: 管道输出类型

func Pipe4 

func Pipe4[I, M1, M2, M3, O any](
	name string,
	fn1 func(ctx context.Context, input I) (M1, error),
	fn2 func(ctx context.Context, input M1) (M2, error),
	fn3 func(ctx context.Context, input M2) (M3, error),
	fn4 func(ctx context.Context, input M3) (O, error),
) *TypedStep[I, O]

Pipe4 连接四个类型安全的函数

类型参数：

  • I: 管道输入类型
  • M1: 第一个中间类型
  • M2: 第二个中间类型
  • M3: 第三个中间类型
  • O: 管道输出类型

func Then 

func Then[I, M, O any](first *TypedStep[I, M], second *TypedStep[M, O]) *TypedStep[I, O]

Then 连接另一个类型安全的步骤

func (*TypedStep[I, O]) ToStep 

func (s *TypedStep[I, O]) ToStep() step

ToStep 转换为通用步骤

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.