core

package
v0.5.11 Latest Latest
Warning

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

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

Documentation

Overview

Package core 提供 Hexagon 框架的核心接口和类型

本文件实现异步 API 变体：

  • AsyncRunnable: 异步执行接口
  • Future: 异步结果封装
  • 并行执行: 多任务并行
  • 超时控制: 异步超时处理

设计借鉴：

  • Java CompletableFuture
  • Rust async/await
  • Python asyncio

compose.go 提供 Runnable 组合适配器和自动流转换

核心功能：

  • InvokeToStream: 将 Invoke-only 的 Runnable 适配为支持流输出
  • StreamToInvoke: 将 Stream-only 的 Runnable 适配为支持同步返回
  • Compose: 组合两个 Runnable，自动处理类型和流转换
  • ComposeStream: 组合两个 Runnable，输出始终为流

在编排层自动应用 Concat（流→值）与 Box（值→流）。

使用示例： 

// 组合两个 Runnable，类型自动匹配
combined := Compose[string, int, string](parser, formatter)
result, err := combined.Invoke(ctx, "42")

// 输出始终为流
streaming := ComposeStream[string, int, string](parser, formatter)
sr, err := streaming.Stream(ctx, "42")

Package core 提供 Hexagon 框架的核心接口和类型

本文件实现 WithFallback 机制：

  • Fallback: 降级处理

  • Retry: 重试机制

  • CircuitBreaker: 熔断器

  • RunnableWithFallback: 带降级的 Runnable

  • Resilience4j: 弹性模式

  • Polly: 弹性和瞬态故障处理

Package core 提供 Hexagon 框架的核心接口和类型

本包定义了框架的基础抽象：

  • Runnable[I, O]: 六范式统一执行接口
  • Component[I, O]: 向后兼容的组件接口
  • Option: 执行选项
  • Schema: JSON Schema 类型定义

六范式统一执行模型：

  • Invoke: 普通输入 → 普通输出
  • Stream: 普通输入 → 流输出
  • Batch: 多输入 → 多输出（并发）
  • Collect: 流输入 → 普通输出
  • Transform: 流输入 → 流输出
  • BatchStream: 多输入 → 流输出

设计借鉴：

  • LangChain: Runnable 接口 (invoke/stream/batch)
  • Eino: 四范式 (Invoke/Stream/Collect/Transform)
  • Spring AI: Fluent API 风格

使用示例： 

// 实现 Runnable 接口
type MyComponent struct{}

func (c *MyComponent) Invoke(ctx context.Context, input string, opts ...Option) (string, error) {
    return "processed: " + input, nil
}

// 其他方法由 BaseRunnable 自动实现

Index

Constants

View Source 
const (
	// CircuitClosed 关闭状态（正常）
	CircuitClosed = circuit.StateClosed
	// CircuitOpen 打开状态（熔断）
	CircuitOpen = circuit.StateOpen
	// CircuitHalfOpen 半开状态（尝试恢复）
	CircuitHalfOpen = circuit.StateHalfOpen
)

Variables

View Source 
var (
	// ErrAllFallbacksFailed 所有降级都失败
	ErrAllFallbacksFailed = errors.New("all fallbacks failed")

	// ErrCircuitOpen 熔断器打开
	ErrCircuitOpen = circuit.ErrCircuitOpen

	// ErrMaxRetriesExceeded 超过最大重试次数
	ErrMaxRetriesExceeded = errors.New("max retries exceeded")
)

Functions

func InvokeWithCallback 

func InvokeWithCallback[I, O any](
	ctx context.Context,
	runnable Runnable[I, O],
	input I,
	callback Callback[O],
	opts ...Option,
)

InvokeWithCallback 带回调的调用

func NewSliceStream 

func NewSliceStream[T any](items []T) *stream.StreamReader[T]

NewSliceStream 从切片创建流（向后兼容） Deprecated: 请使用 stream.FromSlice

Types

type AsyncRunnable 

type AsyncRunnable[I, O any] interface {
	// InvokeAsync 异步调用
	InvokeAsync(ctx context.Context, input I, opts ...Option) *Future[O]

	// BatchAsync 异步批量调用
	BatchAsync(ctx context.Context, inputs []I, opts ...Option) *Future[[]O]
}

AsyncRunnable 异步执行接口

type AsyncWrapper 

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

AsyncWrapper 将同步 Runnable 包装为异步

func WrapAsync 

func WrapAsync[I, O any](r Runnable[I, O]) *AsyncWrapper[I, O]

WrapAsync 包装为异步

func (*AsyncWrapper[I, O]) BatchAsync 

func (w *AsyncWrapper[I, O]) BatchAsync(ctx context.Context, inputs []I, opts ...Option) *Future[[]O]

BatchAsync 异步批量调用

func (*AsyncWrapper[I, O]) InvokeAsync 

func (w *AsyncWrapper[I, O]) InvokeAsync(ctx context.Context, input I, opts ...Option) *Future[O]

InvokeAsync 异步调用

type BaseRunnable 

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

BaseRunnable 提供 Runnable 接口的基础实现 只需实现 Invoke 方法，其他方法自动推导

func NewRunnable 

func NewRunnable[I, O any](name, description string, invokeFn func(context.Context, I, ...Option) (O, error)) *BaseRunnable[I, O]

NewRunnable 创建基础 Runnable

func (*BaseRunnable[I, O]) Batch 

func (r *BaseRunnable[I, O]) Batch(ctx context.Context, inputs []I, opts ...Option) ([]O, error)

Batch 批量执行组件 使用 toolkit/poolx 协程池进行并发控制，避免 goroutine 爆炸

func (*BaseRunnable[I, O]) BatchStream 

func (r *BaseRunnable[I, O]) BatchStream(ctx context.Context, inputs []I, opts ...Option) (*StreamReader[O], error)

BatchStream 批量流式 使用 toolkit/poolx 协程池进行并发控制

func (*BaseRunnable[I, O]) Collect 

func (r *BaseRunnable[I, O]) Collect(ctx context.Context, input *StreamReader[I], opts ...Option) (O, error)

Collect 流收集

func (*BaseRunnable[I, O]) Description 

func (r *BaseRunnable[I, O]) Description() string

Description 返回组件描述

func (*BaseRunnable[I, O]) InputSchema 

func (r *BaseRunnable[I, O]) InputSchema() *Schema

InputSchema 返回输入 Schema

func (*BaseRunnable[I, O]) Invoke 

func (r *BaseRunnable[I, O]) Invoke(ctx context.Context, input I, opts ...Option) (O, error)

Invoke 执行组件

func (*BaseRunnable[I, O]) Name 

func (r *BaseRunnable[I, O]) Name() string

Name 返回组件名称

func (*BaseRunnable[I, O]) OutputSchema 

func (r *BaseRunnable[I, O]) OutputSchema() *Schema

OutputSchema 返回输出 Schema

func (*BaseRunnable[I, O]) Stream 

func (r *BaseRunnable[I, O]) Stream(ctx context.Context, input I, opts ...Option) (*StreamReader[O], error)

Stream 流式执行组件

func (*BaseRunnable[I, O]) Transform 

func (r *BaseRunnable[I, O]) Transform(ctx context.Context, input *StreamReader[I], opts ...Option) (*StreamReader[O], error)

Transform 流转换

注意：从 Invoke 推导时，goroutine 会正确处理 context 取消和 EOF， 确保不会发生 goroutine 泄漏。

func (*BaseRunnable[I, O]) WithBatch 

func (r *BaseRunnable[I, O]) WithBatch(fn func(context.Context, []I, ...Option) ([]O, error)) *BaseRunnable[I, O]

WithBatch 设置 Batch 实现

func (*BaseRunnable[I, O]) WithBatchStream 

func (r *BaseRunnable[I, O]) WithBatchStream(fn func(context.Context, []I, ...Option) (*StreamReader[O], error)) *BaseRunnable[I, O]

WithBatchStream 设置 BatchStream 实现

func (*BaseRunnable[I, O]) WithCollect 

func (r *BaseRunnable[I, O]) WithCollect(fn func(context.Context, *StreamReader[I], ...Option) (O, error)) *BaseRunnable[I, O]

WithCollect 设置 Collect 实现

func (*BaseRunnable[I, O]) WithStream 

func (r *BaseRunnable[I, O]) WithStream(fn func(context.Context, I, ...Option) (*StreamReader[O], error)) *BaseRunnable[I, O]

WithStream 设置 Stream 实现

func (*BaseRunnable[I, O]) WithTransform 

func (r *BaseRunnable[I, O]) WithTransform(fn func(context.Context, *StreamReader[I], ...Option) (*StreamReader[O], error)) *BaseRunnable[I, O]

WithTransform 设置 Transform 实现

type Callback 

type Callback[T any] func(result T, err error)

Callback 回调类型

type CircuitBreaker 

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

CircuitBreaker 是 toolkit 熔断器的领域配置适配器，不复制状态机。

func NewCircuitBreaker 

func NewCircuitBreaker(config ...*CircuitBreakerConfig) (*CircuitBreaker, error)

NewCircuitBreaker 创建熔断器

func (*CircuitBreaker) Acquire 

func (cb *CircuitBreaker) Acquire() (*circuit.Permit, error)

Acquire 获取与一次执行严格绑定的许可。

func (*CircuitBreaker) Close 

func (cb *CircuitBreaker) Close()

Close 关闭熔断器并释放生命周期资源。

func (*CircuitBreaker) State 

func (cb *CircuitBreaker) State() CircuitState

State 获取当前状态

type CircuitBreakerConfig 

type CircuitBreakerConfig struct {
	// FailureThreshold 失败阈值
	FailureThreshold int

	// SuccessThreshold 成功阈值（半开状态）
	SuccessThreshold int

	// HalfOpenMaxRequests 半开状态允许的最大并发探测数；为零时使用默认值 3。
	HalfOpenMaxRequests int

	// Timeout 熔断超时
	Timeout time.Duration

	// OnStateChange 状态变化回调
	OnStateChange func(from, to CircuitState)
}

CircuitBreakerConfig 熔断器配置

func DefaultCircuitBreakerConfig 

func DefaultCircuitBreakerConfig() *CircuitBreakerConfig

DefaultCircuitBreakerConfig 默认熔断器配置

type CircuitState 

type CircuitState = circuit.State

CircuitState 熔断器状态，直接复用 toolkit 的唯一状态定义。

type Component 

type Component[I, O any] interface {
	Runnable[I, O]
}

Component 是 Runnable 的向后兼容别名 保持与旧版本 API 的兼容性

type FallbackConfig 

type FallbackConfig struct {
	// ExceptionsToHandle 要处理的异常类型
	ExceptionsToHandle []error

	// OnFallback 降级回调
	OnFallback func(err error, fallbackIndex int)
}

FallbackConfig Fallback 配置

type FallbackOption 

type FallbackOption func(*FallbackConfig)

FallbackOption Fallback 选项

func WithExceptions 

func WithExceptions(errs ...error) FallbackOption

WithExceptions 设置要处理的异常类型

func WithFallbackCallback 

func WithFallbackCallback(fn func(err error, fallbackIndex int)) FallbackOption

WithFallbackCallback 设置降级回调

type Future 

type Future[T any] struct {
	// contains filtered or unexported fields
}

Future 异步结果

线程安全：Future 使用 sync.Once 确保只完成一次， 并使用 sync.RWMutex 保护结果的读写，确保并发安全。

func Any 

func Any[T any](futures ...*Future[T]) *Future[T]

Any 任意成功（返回第一个成功的）

线程安全：使用互斥锁保护错误计数和错误列表。

func Delay 

func Delay[T any](duration time.Duration, fn func() (T, error)) *Future[T]

Delay 延迟执行

func NewFuture 

func NewFuture[T any]() *Future[T]

NewFuture 创建 Future

func Parallel 

func Parallel[T any](futures ...*Future[T]) *Future[[]T]

Parallel 并行执行多个 Future

func ParallelWithLimit 

func ParallelWithLimit[T any](limit int, futures ...*Future[T]) *Future[[]T]

ParallelWithLimit 带并发限制的并行执行

func Race 

func Race[T any](futures ...*Future[T]) *Future[T]

Race 竞争执行（返回第一个完成的）

func Retry 

func Retry[T any](maxRetries int, delay time.Duration, fn func() (T, error)) *Future[T]

Retry 重试执行

在独立 goroutine 中以固定延迟重试 fn，结果经 Future 异步返回。 重试循环本身下沉至 toolkit/util/retry.Do，本函数保留 Future/goroutine 封装。

语义对齐说明（与下沉前手写循环逐项等价）：

  • 总尝试次数：手写循环为 i 0..maxRetries，即首次调用 + maxRetries 次重试， 共 maxRetries+1 次，故 toolkit 的 Attempts 取 maxRetries+1。
  • 固定延迟：手写循环每次重试前固定 Sleep(delay)，无指数退避。这里显式以 DelayType(FixedDelay) 锁定 Hexagon 的固定延迟合同；delay>0 时同时将 MaxDelay 设为 delay，避免默认 30s 上限在 delay>30s 时反向裁剪固定延迟（FixedDelay 返回值会被 MaxDelay 封顶）。delay=0 时保留 toolkit 的默认正上限， FixedDelay 仍返回 0。FixedDelay 路径不叠加抖动，与手写循环一致。
  • 重试条件：手写循环对任意非 nil 错误重试，等同 toolkit 默认条件。
  • 最终错误可解包：手写循环重试耗尽直接返回原始 lastErr。toolkit 当前 默认以双 %w 同时保留 ErrMaxAttemptsReached 与原始 lastErr，Hexagon 无需再注入兼容 Option。

说明：本函数签名不含 context.Context，仍使用 retry.Do（非 ctx 版）， 延迟期间不感知取消——这是签名约束，ctx 感知属另一议题，本次下沉不引入。

func RunAsync 

func RunAsync[T any](fn func() (T, error)) *Future[T]

RunAsync 异步执行函数

func RunAsyncWithContext 

func RunAsyncWithContext[T any](ctx context.Context, fn func(context.Context) (T, error)) *Future[T]

RunAsyncWithContext 带上下文的异步执行

func (*Future[T]) Catch 

func (f *Future[T]) Catch(fn func(error) (T, error)) *Future[T]

Catch 错误处理

func (*Future[T]) Complete 

func (f *Future[T]) Complete(result T, err error)

Complete 完成 Future

线程安全：可以被多个 goroutine 并发调用，但只有第一次调用会生效。

func (*Future[T]) Get 

func (f *Future[T]) Get() (T, error)

Get 获取结果（阻塞）

线程安全：等待 Future 完成后返回结果，多个 goroutine 可以并发调用。

func (*Future[T]) GetWithContext 

func (f *Future[T]) GetWithContext(ctx context.Context) (T, error)

GetWithContext 带上下文获取结果

func (*Future[T]) GetWithTimeout 

func (f *Future[T]) GetWithTimeout(timeout time.Duration) (T, error)

GetWithTimeout 带超时获取结果

func (*Future[T]) IsDone 

func (f *Future[T]) IsDone() bool

IsDone 是否完成

func (*Future[T]) Then 

func (f *Future[T]) Then(fn func(T) (T, error)) *Future[T]

Then 链式处理

type Option 

type Option interface {
	Apply(*Options)
}

Option 执行选项

func WithMaxRetries 

func WithMaxRetries(n int) Option

WithMaxRetries 设置最大重试次数

func WithMetadata 

func WithMetadata(meta map[string]any) Option

WithMetadata 设置元数据

func WithNodeID 

func WithNodeID(id string) Option

WithNodeID 设置目标节点ID

func WithNodeType 

func WithNodeType(typ string) Option

WithNodeType 设置目标节点类型

func WithStreamBuffer 

func WithStreamBuffer(size int) Option

WithStreamBuffer 设置流缓冲区大小

func WithTimeout 

func WithTimeout(ms int64) Option

WithTimeout 设置超时

type OptionFunc 

type OptionFunc func(*Options)

OptionFunc 函数式选项

func (OptionFunc) Apply 

func (f OptionFunc) Apply(opts *Options)

type Options 

type Options struct {
	// 基础选项
	Timeout    int64          // 超时时间（毫秒）
	MaxRetries int            // 最大重试次数
	Metadata   map[string]any // 元数据

	// 流式选项
	StreamBufferSize int   // 流缓冲区大小
	StreamTimeout    int64 // 流操作超时（毫秒）

	// 节点选项（用于图编排）
	NodeID   string // 目标节点ID
	NodeType string // 目标节点类型

	// 扩展选项
	Extra map[string]any
}

Options 选项集合

func ApplyOptions 

func ApplyOptions(opts ...Option) *Options

ApplyOptions 应用选项

type Promise 

type Promise[T any] struct {
	// contains filtered or unexported fields
}

Promise 承诺模式

func NewPromise 

func NewPromise[T any]() *Promise[T]

NewPromise 创建 Promise

func (*Promise[T]) Future 

func (p *Promise[T]) Future() *Future[T]

Future 获取 Future

func (*Promise[T]) Reject 

func (p *Promise[T]) Reject(err error)

Reject 失败

func (*Promise[T]) Resolve 

func (p *Promise[T]) Resolve(value T)

Resolve 成功完成

type Result 

type Result[T any] struct {
	Value T
	Err   error
}

Result 结果

type ResultChannel 

type ResultChannel[T any] struct {
	// contains filtered or unexported fields
}

ResultChannel 结果通道

func NewResultChannel 

func NewResultChannel[T any](buffer int) *ResultChannel[T]

NewResultChannel 创建结果通道

func (*ResultChannel[T]) Channel 

func (rc *ResultChannel[T]) Channel() <-chan Result[T]

Channel 获取底层通道

func (*ResultChannel[T]) Close 

func (rc *ResultChannel[T]) Close()

Close 关闭通道

func (*ResultChannel[T]) Receive 

func (rc *ResultChannel[T]) Receive() (T, error)

Receive 接收结果

func (*ResultChannel[T]) ReceiveWithContext 

func (rc *ResultChannel[T]) ReceiveWithContext(ctx context.Context) (T, error)

ReceiveWithContext 带上下文接收

func (*ResultChannel[T]) Send 

func (rc *ResultChannel[T]) Send(value T, err error)

Send 发送结果

type RetryConfig 

type RetryConfig struct {
	// MaxRetries 最大重试次数，必须非负
	MaxRetries int

	// InitialDelay 初始延迟，必须非负；零值表示立即重试
	InitialDelay time.Duration

	// MaxDelay 最大延迟，必须非负；零值表示不等待
	MaxDelay time.Duration

	// Multiplier 延迟倍数；零值表示固定使用 InitialDelay
	Multiplier float64

	// Jitter 抖动比例 (0-1)
	Jitter float64

	// RetryOn 判断是否重试
	RetryOn func(error) bool

	// OnRetry 重试回调
	OnRetry func(attempt int, err error)
}

RetryConfig 重试配置

func DefaultRetryConfig 

func DefaultRetryConfig() *RetryConfig

DefaultRetryConfig 默认重试配置

type Runnable 

type Runnable[I, O any] interface {

	// Invoke 同步调用：普通输入 → 普通输出
	// 这是最基本的执行方式
	Invoke(ctx context.Context, input I, opts ...Option) (O, error)

	// Stream 流式输出：普通输入 → 流输出
	// 用于需要流式返回的场景，如 LLM 对话
	Stream(ctx context.Context, input I, opts ...Option) (*StreamReader[O], error)

	// Batch 批量调用：多输入 → 多输出（并发执行）
	// 自动并发执行，结果保持输入顺序
	Batch(ctx context.Context, inputs []I, opts ...Option) ([]O, error)

	// Collect 流收集：流输入 → 普通输出
	// 消费整个流，返回单一结果
	Collect(ctx context.Context, input *StreamReader[I], opts ...Option) (O, error)

	// Transform 流转换：流输入 → 流输出
	// 流到流的转换，支持背压
	Transform(ctx context.Context, input *StreamReader[I], opts ...Option) (*StreamReader[O], error)

	// BatchStream 批量流式：多输入 → 流输出（合并流）
	// 并发执行多个输入，将结果合并为单一流
	BatchStream(ctx context.Context, inputs []I, opts ...Option) (*StreamReader[O], error)

	// Name 返回组件名称
	Name() string

	// Description 返回组件描述
	Description() string

	// InputSchema 返回输入参数的 Schema
	InputSchema() *Schema

	// OutputSchema 返回输出参数的 Schema
	OutputSchema() *Schema
}

Runnable 是所有可执行组件的统一接口 六范式统一执行模型，超越 Eino 的四范式

类型参数：

  • I: 输入类型
  • O: 输出类型

func Compose 

func Compose[I, M, O any](r1 Runnable[I, M], r2 Runnable[M, O]) Runnable[I, O]

Compose 组合两个 Runnable，自动处理类型匹配

执行路径（优先级从高到低）：

  • Invoke 路径: r1.Invoke → r2.Invoke
  • Stream 路径: r1.Invoke → r2.Stream（或 r1.Stream → Map(r2.Invoke)）

类型参数：

  • I: 输入类型（第一个 Runnable 的输入）
  • M: 中间类型（r1 的输出 = r2 的输入）
  • O: 输出类型（第二个 Runnable 的输出）

func ComposeStream 

func ComposeStream[I, M, O any](r1 Runnable[I, M], r2 Runnable[M, O]) Runnable[I, O]

ComposeStream 组合两个 Runnable，输出始终为流

与 Compose 的区别：Stream 方法会尝试使用 r1.Stream + r2.Transform 以获得真正的端到端流式体验。

执行路径：

  • Invoke 路径: r1.Invoke → r2.Invoke（同 Compose）
  • Stream 路径: r1.Stream → 对每个中间元素调用 r2.Stream → Merge

func InvokeToStream 

func InvokeToStream[I, O any](r Runnable[I, O]) Runnable[I, O]

InvokeToStream 将 Invoke-only 的 Runnable 适配为返回流

对于只实现了 Invoke 的组件，自动将结果包装为 FromValue 单元素流。 适用于将同步组件接入流式管道的场景。

返回的 Runnable 特性：

  • Invoke: 直接委托给原始 Runnable
  • Stream: 调用 Invoke 后通过 FromValue 包装为流
  • 其他方法通过 BaseRunnable 自动推导

func RunnableFunc 

func RunnableFunc[I, O any](name string, fn func(context.Context, I) (O, error)) Runnable[I, O]

RunnableFunc 从函数创建 Runnable

func RunnableLambda 

func RunnableLambda[I, O any](fn func(I) O) Runnable[I, O]

RunnableLambda 从 lambda 创建 Runnable（简化版）

func StreamToInvoke 

func StreamToInvoke[I, O any](r Runnable[I, O]) Runnable[I, O]

StreamToInvoke 将 Stream-only 的 Runnable 适配为返回值

对于只实现了 Stream 的组件，自动将流结果通过 Concat 合并为单值。 适用于将流式组件接入同步管道的场景。

返回的 Runnable 特性：

  • Stream: 直接委托给原始 Runnable
  • Invoke: 调用 Stream 后通过 Concat 合并为单值
  • 其他方法通过 BaseRunnable 自动推导

type RunnableWithCircuitBreaker 

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

RunnableWithCircuitBreaker 带熔断器的 Runnable

func WithCircuitBreaker 

func WithCircuitBreaker[I, O any](runnable Runnable[I, O], config ...*CircuitBreakerConfig) (*RunnableWithCircuitBreaker[I, O], error)

WithCircuitBreaker 创建带熔断器的 Runnable

func (*RunnableWithCircuitBreaker[I, O]) Batch 

func (r *RunnableWithCircuitBreaker[I, O]) Batch(ctx context.Context, inputs []I, opts ...Option) ([]O, error)

Batch 批量执行

func (*RunnableWithCircuitBreaker[I, O]) BatchStream 

func (r *RunnableWithCircuitBreaker[I, O]) BatchStream(ctx context.Context, inputs []I, opts ...Option) (*StreamReader[O], error)

BatchStream 批量流式

func (*RunnableWithCircuitBreaker[I, O]) Collect 

func (r *RunnableWithCircuitBreaker[I, O]) Collect(ctx context.Context, input *StreamReader[I], opts ...Option) (O, error)

Collect 流收集

func (*RunnableWithCircuitBreaker[I, O]) Description 

func (r *RunnableWithCircuitBreaker[I, O]) Description() string

Description 返回描述

func (*RunnableWithCircuitBreaker[I, O]) InputSchema 

func (r *RunnableWithCircuitBreaker[I, O]) InputSchema() *Schema

InputSchema 返回输入 Schema

func (*RunnableWithCircuitBreaker[I, O]) Invoke 

func (r *RunnableWithCircuitBreaker[I, O]) Invoke(ctx context.Context, input I, opts ...Option) (O, error)

Invoke 执行（带熔断）

func (*RunnableWithCircuitBreaker[I, O]) Name 

func (r *RunnableWithCircuitBreaker[I, O]) Name() string

Name 返回名称

func (*RunnableWithCircuitBreaker[I, O]) OutputSchema 

func (r *RunnableWithCircuitBreaker[I, O]) OutputSchema() *Schema

OutputSchema 返回输出 Schema

func (*RunnableWithCircuitBreaker[I, O]) Stream 

func (r *RunnableWithCircuitBreaker[I, O]) Stream(ctx context.Context, input I, opts ...Option) (*StreamReader[O], error)

Stream 流式执行（带熔断）

func (*RunnableWithCircuitBreaker[I, O]) Transform 

func (r *RunnableWithCircuitBreaker[I, O]) Transform(ctx context.Context, input *StreamReader[I], opts ...Option) (*StreamReader[O], error)

Transform 流转换

type RunnableWithFallback 

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

RunnableWithFallback 带降级的 Runnable

func WithFallback 

func WithFallback[I, O any](primary Runnable[I, O], fallbacks ...Runnable[I, O]) *RunnableWithFallback[I, O]

WithFallback 创建带降级的 Runnable

示例: 

runnable := core.WithFallback(
    primaryRunnable,
    fallbackRunnable1,
    fallbackRunnable2,
)
result, err := runnable.Invoke(ctx, input)

func (*RunnableWithFallback[I, O]) Batch 

func (r *RunnableWithFallback[I, O]) Batch(ctx context.Context, inputs []I, opts ...Option) ([]O, error)

Batch 批量执行（带降级）

func (*RunnableWithFallback[I, O]) BatchStream 

func (r *RunnableWithFallback[I, O]) BatchStream(ctx context.Context, inputs []I, opts ...Option) (*StreamReader[O], error)

BatchStream 批量流式（带降级）

func (*RunnableWithFallback[I, O]) Collect 

func (r *RunnableWithFallback[I, O]) Collect(ctx context.Context, input *StreamReader[I], opts ...Option) (O, error)

Collect 流收集（带降级）

func (*RunnableWithFallback[I, O]) Description 

func (r *RunnableWithFallback[I, O]) Description() string

Description 返回描述

func (*RunnableWithFallback[I, O]) InputSchema 

func (r *RunnableWithFallback[I, O]) InputSchema() *Schema

InputSchema 返回输入 Schema

func (*RunnableWithFallback[I, O]) Invoke 

func (r *RunnableWithFallback[I, O]) Invoke(ctx context.Context, input I, opts ...Option) (O, error)

Invoke 执行（带降级）

func (*RunnableWithFallback[I, O]) Name 

func (r *RunnableWithFallback[I, O]) Name() string

Name 返回名称

func (*RunnableWithFallback[I, O]) OutputSchema 

func (r *RunnableWithFallback[I, O]) OutputSchema() *Schema

OutputSchema 返回输出 Schema

func (*RunnableWithFallback[I, O]) Stream 

func (r *RunnableWithFallback[I, O]) Stream(ctx context.Context, input I, opts ...Option) (*StreamReader[O], error)

Stream 流式执行（带降级）

func (*RunnableWithFallback[I, O]) Transform 

func (r *RunnableWithFallback[I, O]) Transform(ctx context.Context, input *StreamReader[I], opts ...Option) (*StreamReader[O], error)

Transform 流转换（带降级）

func (*RunnableWithFallback[I, O]) WithOptions 

func (r *RunnableWithFallback[I, O]) WithOptions(opts ...FallbackOption) *RunnableWithFallback[I, O]

WithOptions 设置选项

type RunnableWithRetry 

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

RunnableWithRetry 带重试的 Runnable

func WithRetry 

func WithRetry[I, O any](runnable Runnable[I, O], config ...*RetryConfig) *RunnableWithRetry[I, O]

WithRetry 创建带重试的 Runnable

func (*RunnableWithRetry[I, O]) Batch 

func (r *RunnableWithRetry[I, O]) Batch(ctx context.Context, inputs []I, opts ...Option) ([]O, error)

Batch 批量执行（带重试）

func (*RunnableWithRetry[I, O]) BatchStream 

func (r *RunnableWithRetry[I, O]) BatchStream(ctx context.Context, inputs []I, opts ...Option) (*StreamReader[O], error)

BatchStream 批量流式

func (*RunnableWithRetry[I, O]) Collect 

func (r *RunnableWithRetry[I, O]) Collect(ctx context.Context, input *StreamReader[I], opts ...Option) (O, error)

Collect 流收集

func (*RunnableWithRetry[I, O]) Description 

func (r *RunnableWithRetry[I, O]) Description() string

Description 返回描述

func (*RunnableWithRetry[I, O]) InputSchema 

func (r *RunnableWithRetry[I, O]) InputSchema() *Schema

InputSchema 返回输入 Schema

func (*RunnableWithRetry[I, O]) Invoke 

func (r *RunnableWithRetry[I, O]) Invoke(ctx context.Context, input I, opts ...Option) (O, error)

Invoke 执行（带重试）

重试与退避逻辑下沉至 toolkit/util/retry.DoWithContext， 本方法仅负责承接 RunnableWithRetry 的输入/输出并捕获最近一次结果。

func (*RunnableWithRetry[I, O]) Name 

func (r *RunnableWithRetry[I, O]) Name() string

Name 返回名称

func (*RunnableWithRetry[I, O]) OutputSchema 

func (r *RunnableWithRetry[I, O]) OutputSchema() *Schema

OutputSchema 返回输出 Schema

func (*RunnableWithRetry[I, O]) Stream 

func (r *RunnableWithRetry[I, O]) Stream(ctx context.Context, input I, opts ...Option) (*StreamReader[O], error)

Stream 流式执行（带重试）

重试与退避逻辑下沉至 toolkit/util/retry.DoWithContext， 本方法仅负责承接流的获取并捕获最近一次成功的流读取器。

func (*RunnableWithRetry[I, O]) Transform 

func (r *RunnableWithRetry[I, O]) Transform(ctx context.Context, input *StreamReader[I], opts ...Option) (*StreamReader[O], error)

Transform 流转换

type Schema 

type Schema = llm.Schema

Schema 是 ai-core/llm.Schema 的别名

func SchemaOf 

func SchemaOf[T any]() *Schema

SchemaOf 从 Go 类型生成 Schema

type Stream 

type Stream[T any] = stream.StreamReader[T]

Stream 是 stream.StreamReader 的向后兼容别名 Deprecated: 请使用 StreamReader

type StreamReader 

type StreamReader[T any] = stream.StreamReader[T]

StreamReader 是 stream.StreamReader 的别名

type StreamWriter 

type StreamWriter[T any] = stream.StreamWriter[T]

StreamWriter 是 stream.StreamWriter 的别名

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.