Eino 实战经验与最佳实践

your-eino-project/
├── cmd/                    # 应用入口
│   └── server/
│       └── main.go
├── internal/               # 内部包
│   ├── config/            # 配置管理
│   ├── handlers/          # 业务处理器
│   ├── middleware/        # 中间件
│   └── models/           # 数据模型
├── pkg/                   # 可复用包
│   ├── chains/           # 预定义链
│   ├── components/       # 自定义组件
│   └── utils/           # 工具函数
├── configs/              # 配置文件
├── deployments/          # 部署配置
├── docs/                # 文档
├── examples/            # 示例代码
├── scripts/             # 脚本
├── tests/               # 测试
├── go.mod
├── go.sum
├── Dockerfile
├── docker-compose.yml
└── README.md

// go.mod 示例
module your-company/your-eino-app

go 1.21

require (
    github.com/cloudwego/eino v0.1.0
    github.com/cloudwego/eino-ext v0.1.0
    
    // 基础依赖
    github.com/gin-gonic/gin v1.9.1
    github.com/spf13/viper v1.16.0
    github.com/sirupsen/logrus v1.9.3
    
    // 数据库
    gorm.io/gorm v1.25.4
    gorm.io/driver/postgres v1.5.2
    
    // 缓存
    github.com/redis/go-redis/v9 v9.1.0
    
    // 监控
    github.com/prometheus/client_golang v1.16.0
    go.opentelemetry.io/otel v1.16.0
)

graph TB
    subgraph "表现层 (Presentation Layer)"
        API[REST API]
        WS[WebSocket]
        CLI[CLI Interface]
    end
    
    subgraph "业务层 (Business Layer)"
        SVC[Service Layer]
        CHAIN[Chain Orchestration]
        WORKFLOW[Workflow Management]
    end
    
    subgraph "组件层 (Component Layer)"
        MODEL[Model Components]
        TOOL[Tool Components]
        RETRIEVER[Retriever Components]
    end
    
    subgraph "基础设施层 (Infrastructure Layer)"
        DB[Database]
        CACHE[Cache]
        MQ[Message Queue]
        STORAGE[File Storage]
    end
    
    API --> SVC
    WS --> SVC
    CLI --> SVC
    
    SVC --> CHAIN
    SVC --> WORKFLOW
    
    CHAIN --> MODEL
    CHAIN --> TOOL
    CHAIN --> RETRIEVER
    
    MODEL --> DB
    TOOL --> CACHE
    RETRIEVER --> STORAGE
    
    style API fill:#e8f5e8
    style SVC fill:#fff3e0
    style MODEL fill:#f3e5f5
    style DB fill:#e3f2fd

// service/chat_service.go
package service

import (
    "context"
    "fmt"
    "time"
    
    "github.com/cloudwego/eino/compose"
    "github.com/cloudwego/eino/schema"
)

// ChatService 聊天服务接口
type ChatService interface {
    // Chat 处理单轮对话
    Chat(ctx context.Context, req *ChatRequest) (*ChatResponse, error)
    
    // StreamChat 处理流式对话
    StreamChat(ctx context.Context, req *ChatRequest) (<-chan *ChatChunk, error)
    
    // MultiTurnChat 处理多轮对话
    MultiTurnChat(ctx context.Context, req *MultiTurnChatRequest) (*ChatResponse, error)
}

// ChatRequest 聊天请求
type ChatRequest struct {
    UserID    string            `json:"user_id"`
    SessionID string            `json:"session_id"`
    Message   string            `json:"message"`
    Context   map[string]any    `json:"context,omitempty"`
    Options   *ChatOptions      `json:"options,omitempty"`
}

// ChatResponse 聊天响应
type ChatResponse struct {
    MessageID   string            `json:"message_id"`
    Content     string            `json:"content"`
    TokenUsage  *TokenUsage       `json:"token_usage,omitempty"`
    Metadata    map[string]any    `json:"metadata,omitempty"`
    ProcessTime time.Duration     `json:"process_time"`
}

// chatServiceImpl 聊天服务实现
type chatServiceImpl struct {
    chatChain    compose.Runnable[*ChatInput, *schema.Message]
    ragChain     compose.Runnable[*RAGInput, *schema.Message]
    toolChain    compose.Runnable[*ToolInput, *schema.Message]
    
    sessionRepo  SessionRepository
    messageRepo  MessageRepository
    
    logger       Logger
    metrics      Metrics
}

func NewChatService(
    chatChain compose.Runnable[*ChatInput, *schema.Message],
    ragChain compose.Runnable[*RAGInput, *schema.Message],
    toolChain compose.Runnable[*ToolInput, *schema.Message],
    sessionRepo SessionRepository,
    messageRepo MessageRepository,
    logger Logger,
    metrics Metrics,
) ChatService {
    return &chatServiceImpl{
        chatChain:   chatChain,
        ragChain:    ragChain,
        toolChain:   toolChain,
        sessionRepo: sessionRepo,
        messageRepo: messageRepo,
        logger:      logger,
        metrics:     metrics,
    }
}

func (s *chatServiceImpl) Chat(ctx context.Context, req *ChatRequest) (*ChatResponse, error) {
    startTime := time.Now()
    
    // 记录请求指标
    s.metrics.IncChatRequests(req.UserID)
    
    // 构建输入
    input := &ChatInput{
        UserID:    req.UserID,
        SessionID: req.SessionID,
        Message:   req.Message,
        Context:   req.Context,
    }
    
    // 选择合适的链
    chain, err := s.selectChain(ctx, req)
    if err != nil {
        s.metrics.IncChatErrors("chain_selection_error")
        return nil, fmt.Errorf("failed to select chain: %w", err)
    }
    
    // 执行链
    result, err := chain.Invoke(ctx, input)
    if err != nil {
        s.metrics.IncChatErrors("chain_execution_error")
        s.logger.Error("Chain execution failed", "error", err, "user_id", req.UserID)
        return nil, fmt.Errorf("failed to execute chain: %w", err)
    }
    
    // 保存消息历史
    if err := s.saveMessageHistory(ctx, req, result); err != nil {
        s.logger.Warn("Failed to save message history", "error", err)
    }
    
    processTime := time.Since(startTime)
    s.metrics.ObserveChatLatency(processTime)
    
    return &ChatResponse{
        MessageID:   generateMessageID(),
        Content:     result.Content,
        TokenUsage:  extractTokenUsage(result),
        Metadata:    extractMetadata(result),
        ProcessTime: processTime,
    }, nil
}

func (s *chatServiceImpl) selectChain(ctx context.Context, req *ChatRequest) (compose.Runnable[any, *schema.Message], error) {
    // 根据请求特征选择合适的链
    if s.needsRAG(req) {
        return s.ragChain, nil
    }
    
    if s.needsTools(req) {
        return s.toolChain, nil
    }
    
    return s.chatChain, nil
}

// config/config.go
package config

import (
    "fmt"
    "time"
    
    "github.com/spf13/viper"
)

// Config 应用配置
type Config struct {
    Server   ServerConfig   `mapstructure:"server"`
    Database DatabaseConfig `mapstructure:"database"`
    Redis    RedisConfig    `mapstructure:"redis"`
    LLM      LLMConfig      `mapstructure:"llm"`
    RAG      RAGConfig      `mapstructure:"rag"`
    Logging  LoggingConfig  `mapstructure:"logging"`
    Metrics  MetricsConfig  `mapstructure:"metrics"`
}

// ServerConfig 服务器配置
type ServerConfig struct {
    Host         string        `mapstructure:"host"`
    Port         int           `mapstructure:"port"`
    ReadTimeout  time.Duration `mapstructure:"read_timeout"`
    WriteTimeout time.Duration `mapstructure:"write_timeout"`
    IdleTimeout  time.Duration `mapstructure:"idle_timeout"`
}

// LLMConfig 大语言模型配置
type LLMConfig struct {
    Provider    string            `mapstructure:"provider"`
    Model       string            `mapstructure:"model"`
    APIKey      string            `mapstructure:"api_key"`
    BaseURL     string            `mapstructure:"base_url"`
    MaxTokens   int               `mapstructure:"max_tokens"`
    Temperature float64           `mapstructure:"temperature"`
    Timeout     time.Duration     `mapstructure:"timeout"`
    RetryCount  int               `mapstructure:"retry_count"`
    Extra       map[string]any    `mapstructure:"extra"`
}

// RAGConfig RAG配置
type RAGConfig struct {
    VectorStore VectorStoreConfig `mapstructure:"vector_store"`
    Embedding   EmbeddingConfig   `mapstructure:"embedding"`
    Retrieval   RetrievalConfig   `mapstructure:"retrieval"`
}

// LoadConfig 加载配置
func LoadConfig(configPath string) (*Config, error) {
    viper.SetConfigFile(configPath)
    viper.SetConfigType("yaml")
    
    // 设置默认值
    setDefaults()
    
    // 读取环境变量
    viper.AutomaticEnv()
    
    // 读取配置文件
    if err := viper.ReadInConfig(); err != nil {
        return nil, fmt.Errorf("failed to read config file: %w", err)
    }
    
    var config Config
    if err := viper.Unmarshal(&config); err != nil {
        return nil, fmt.Errorf("failed to unmarshal config: %w", err)
    }
    
    // 验证配置
    if err := validateConfig(&config); err != nil {
        return nil, fmt.Errorf("invalid config: %w", err)
    }
    
    return &config, nil
}

func setDefaults() {
    // 服务器默认配置
    viper.SetDefault("server.host", "0.0.0.0")
    viper.SetDefault("server.port", 8080)
    viper.SetDefault("server.read_timeout", "30s")
    viper.SetDefault("server.write_timeout", "30s")
    viper.SetDefault("server.idle_timeout", "120s")
    
    // LLM默认配置
    viper.SetDefault("llm.max_tokens", 4096)
    viper.SetDefault("llm.temperature", 0.7)
    viper.SetDefault("llm.timeout", "60s")
    viper.SetDefault("llm.retry_count", 3)
    
    // RAG默认配置
    viper.SetDefault("rag.retrieval.top_k", 5)
    viper.SetDefault("rag.retrieval.score_threshold", 0.7)
}

func validateConfig(config *Config) error {
    if config.LLM.APIKey == "" {
        return fmt.Errorf("LLM API key is required")
    }
    
    if config.LLM.MaxTokens <= 0 {
        return fmt.Errorf("LLM max tokens must be positive")
    }
    
    if config.RAG.Retrieval.TopK <= 0 {
        return fmt.Errorf("RAG retrieval top_k must be positive")
    }
    
    return nil
}

// components/custom_retriever.go
package components

import (
    "context"
    "fmt"
    "sort"
    
    "github.com/cloudwego/eino/components/retriever"
    "github.com/cloudwego/eino/schema"
)

// HybridRetriever 混合检索器，结合向量检索和关键词检索
type HybridRetriever struct {
    vectorRetriever   retriever.Retriever
    keywordRetriever  retriever.Retriever
    
    vectorWeight      float64
    keywordWeight     float64
    
    logger           Logger
    metrics          Metrics
}

// HybridRetrieverConfig 混合检索器配置
type HybridRetrieverConfig struct {
    VectorRetriever   retriever.Retriever `validate:"required"`
    KeywordRetriever  retriever.Retriever `validate:"required"`
    VectorWeight      float64             `validate:"min=0,max=1"`
    KeywordWeight     float64             `validate:"min=0,max=1"`
    Logger           Logger
    Metrics          Metrics
}

func NewHybridRetriever(ctx context.Context, config *HybridRetrieverConfig) (*HybridRetriever, error) {
    if err := validateConfig(config); err != nil {
        return nil, fmt.Errorf("invalid config: %w", err)
    }
    
    return &HybridRetriever{
        vectorRetriever:  config.VectorRetriever,
        keywordRetriever: config.KeywordRetriever,
        vectorWeight:     config.VectorWeight,
        keywordWeight:    config.KeywordWeight,
        logger:          config.Logger,
        metrics:         config.Metrics,
    }, nil
}

// Retrieve 实现 retriever.Retriever 接口
func (hr *HybridRetriever) Retrieve(ctx context.Context, query string, opts ...retriever.Option) ([]*schema.Document, error) {
    startTime := time.Now()
    defer func() {
        hr.metrics.ObserveRetrievalLatency(time.Since(startTime))
    }()
    
    // 解析选项
    options := retriever.GetOptions(opts...)
    
    // 并行执行两种检索
    vectorCh := make(chan retrievalResult, 1)
    keywordCh := make(chan retrievalResult, 1)
    
    // 向量检索
    go func() {
        docs, err := hr.vectorRetriever.Retrieve(ctx, query, opts...)
        vectorCh <- retrievalResult{docs: docs, err: err}
    }()
    
    // 关键词检索
    go func() {
        docs, err := hr.keywordRetriever.Retrieve(ctx, query, opts...)
        keywordCh <- retrievalResult{docs: docs, err: err}
    }()
    
    // 收集结果
    vectorResult := <-vectorCh
    keywordResult := <-keywordCh
    
    // 检查错误
    if vectorResult.err != nil {
        hr.logger.Error("Vector retrieval failed", "error", vectorResult.err)
        hr.metrics.IncRetrievalErrors("vector")
    }
    
    if keywordResult.err != nil {
        hr.logger.Error("Keyword retrieval failed", "error", keywordResult.err)
        hr.metrics.IncRetrievalErrors("keyword")
    }
    
    // 如果两个都失败，返回错误
    if vectorResult.err != nil && keywordResult.err != nil {
        return nil, fmt.Errorf("both retrievals failed: vector=%v, keyword=%v", 
            vectorResult.err, keywordResult.err)
    }
    
    // 合并和重排序结果
    mergedDocs := hr.mergeAndRerank(vectorResult.docs, keywordResult.docs)
    
    // 应用 top_k 限制
    if options.TopK > 0 && len(mergedDocs) > options.TopK {
        mergedDocs = mergedDocs[:options.TopK]
    }
    
    hr.metrics.ObserveRetrievalCount(len(mergedDocs))
    
    return mergedDocs, nil
}

type retrievalResult struct {
    docs []*schema.Document
    err  error
}

// mergeAndRerank 合并和重排序文档
func (hr *HybridRetriever) mergeAndRerank(vectorDocs, keywordDocs []*schema.Document) []*schema.Document {
    // 创建文档映射，避免重复
    docMap := make(map[string]*schema.Document)
    
    // 处理向量检索结果
    for i, doc := range vectorDocs {
        if doc == nil {
            continue
        }
        
        // 计算向量检索分数（基于排名）
        vectorScore := 1.0 - float64(i)/float64(len(vectorDocs))
        
        if existingDoc, exists := docMap[doc.ID]; exists {
            // 文档已存在，更新分数
            existingScore := existingDoc.Score()
            newScore := existingScore + hr.vectorWeight*vectorScore
            existingDoc.WithScore(newScore)
        } else {
            // 新文档
            newDoc := *doc
            newDoc.WithScore(hr.vectorWeight * vectorScore)
            docMap[doc.ID] = &newDoc
        }
    }
    
    // 处理关键词检索结果
    for i, doc := range keywordDocs {
        if doc == nil {
            continue
        }
        
        // 计算关键词检索分数（基于排名）
        keywordScore := 1.0 - float64(i)/float64(len(keywordDocs))
        
        if existingDoc, exists := docMap[doc.ID]; exists {
            // 文档已存在，更新分数
            existingScore := existingDoc.Score()
            newScore := existingScore + hr.keywordWeight*keywordScore
            existingDoc.WithScore(newScore)
        } else {
            // 新文档
            newDoc := *doc
            newDoc.WithScore(hr.keywordWeight * keywordScore)
            docMap[doc.ID] = &newDoc
        }
    }
    
    // 转换为切片并按分数排序
    result := make([]*schema.Document, 0, len(docMap))
    for _, doc := range docMap {
        result = append(result, doc)
    }
    
    sort.Slice(result, func(i, j int) bool {
        return result[i].Score() > result[j].Score()
    })
    
    return result
}

// GetType 返回组件类型
func (hr *HybridRetriever) GetType() string {
    return "HybridRetriever"
}

// chains/processors.go
package chains

import (
    "context"
    "fmt"
    "strings"
    "time"
    
    "github.com/cloudwego/eino/compose"
    "github.com/cloudwego/eino/schema"
)

// CreateQueryProcessor 创建查询预处理器
func CreateQueryProcessor(config *ProcessorConfig) *compose.Lambda {
    return compose.InvokableLambda(func(ctx context.Context, input string) (string, error) {
        startTime := time.Now()
        defer func() {
            config.Metrics.ObserveProcessingLatency("query_processor", time.Since(startTime))
        }()
        
        // 查询清理
        cleaned := strings.TrimSpace(input)
        if cleaned == "" {
            return "", fmt.Errorf("empty query")
        }
        
        // 查询增强
        enhanced := enhanceQuery(cleaned, config)
        
        config.Logger.Debug("Query processed", 
            "original", input, 
            "enhanced", enhanced)
        
        return enhanced, nil
    }, compose.WithLambdaType("QueryProcessor"))
}

// CreateResponseFormatter 创建响应格式化器
func CreateResponseFormatter(config *ProcessorConfig) *compose.Lambda {
    return compose.InvokableLambda(func(ctx context.Context, input *schema.Message) (*schema.Message, error) {
        startTime := time.Now()
        defer func() {
            config.Metrics.ObserveProcessingLatency("response_formatter", time.Since(startTime))
        }()
        
        // 格式化响应内容
        formatted := formatResponse(input.Content, config)
        
        // 创建新的消息
        result := &schema.Message{
            Role:    input.Role,
            Content: formatted,
            Extra: map[string]any{
                "formatted_at": time.Now(),
                "formatter":    "ResponseFormatter",
            },
        }
        
        // 复制其他字段
        if input.ResponseMeta != nil {
            result.ResponseMeta = input.ResponseMeta
        }
        
        return result, nil
    }, compose.WithLambdaType("ResponseFormatter"))
}

// CreateStreamingProcessor 创建流式处理器
func CreateStreamingProcessor(config *ProcessorConfig) *compose.Lambda {
    return compose.StreamableLambda(func(ctx context.Context, input string) (*schema.StreamReader[string], error) {
        // 创建流式输出
        sr, sw := schema.Pipe[string](10)
        
        go func() {
            defer sw.Close()
            
            // 模拟流式处理
            words := strings.Fields(input)
            for i, word := range words {
                select {
                case <-ctx.Done():
                    return
                default:
                }
                
                // 处理单词
                processed := processWord(word, config)
                
                if sw.Send(processed, nil) {
                    return // 流已关闭
                }
                
                // 模拟处理延迟
                if i < len(words)-1 {
                    time.Sleep(10 * time.Millisecond)
                }
            }
        }()
        
        return sr, nil
    }, compose.WithLambdaType("StreamingProcessor"))
}

// CreateBatchProcessor 创建批处理器
func CreateBatchProcessor(config *ProcessorConfig) *compose.Lambda {
    return compose.CollectableLambda(func(ctx context.Context, input *schema.StreamReader[string]) (string, error) {
        var items []string
        
        // 收集所有流数据
        for {
            item, err := input.Recv()
            if err == io.EOF {
                break
            }
            if err != nil {
                return "", fmt.Errorf("failed to receive stream item: %w", err)
            }
            
            items = append(items, item)
        }
        
        // 批量处理
        result := processBatch(items, config)
        
        return result, nil
    }, compose.WithLambdaType("BatchProcessor"))
}

// 辅助函数
func enhanceQuery(query string, config *ProcessorConfig) string {
    // 实现查询增强逻辑
    enhanced := query
    
    // 添加上下文信息
    if config.AddContext {
        enhanced = fmt.Sprintf("Context: %s\nQuery: %s", config.ContextInfo, enhanced)
    }
    
    // 添加指令
    if config.AddInstructions {
        enhanced = fmt.Sprintf("%s\nInstructions: %s", enhanced, config.Instructions)
    }
    
    return enhanced
}

func formatResponse(content string, config *ProcessorConfig) string {
    // 实现响应格式化逻辑
    formatted := content
    
    // 添加格式化标记
    if config.AddMarkdown {
        formatted = addMarkdownFormatting(formatted)
    }
    
    // 添加元信息
    if config.AddMetaInfo {
        formatted = fmt.Sprintf("%s\n\n---\n*Generated at: %s*", 
            formatted, time.Now().Format("2006-01-02 15:04:05"))
    }
    
    return formatted
}

// errors/errors.go
package errors

import (
    "fmt"
    "net/http"
)

// 定义错误类型
type ErrorType string

const (
    ErrorTypeValidation   ErrorType = "validation"
    ErrorTypeNotFound     ErrorType = "not_found"
    ErrorTypeUnauthorized ErrorType = "unauthorized"
    ErrorTypeRateLimit    ErrorType = "rate_limit"
    ErrorTypeInternal     ErrorType = "internal"
    ErrorTypeExternal     ErrorType = "external"
)

// AppError 应用错误
type AppError struct {
    Type       ErrorType         `json:"type"`
    Code       string            `json:"code"`
    Message    string            `json:"message"`
    Details    string            `json:"details,omitempty"`
    Metadata   map[string]any    `json:"metadata,omitempty"`
    Cause      error             `json:"-"`
    HTTPStatus int               `json:"-"`
}

func (e *AppError) Error() string {
    if e.Cause != nil {
        return fmt.Sprintf("%s: %s (caused by: %v)", e.Code, e.Message, e.Cause)
    }
    return fmt.Sprintf("%s: %s", e.Code, e.Message)
}

func (e *AppError) Unwrap() error {
    return e.Cause
}

// 错误构造函数
func NewValidationError(code, message string) *AppError {
    return &AppError{
        Type:       ErrorTypeValidation,
        Code:       code,
        Message:    message,
        HTTPStatus: http.StatusBadRequest,
    }
}

func NewNotFoundError(code, message string) *AppError {
    return &AppError{
        Type:       ErrorTypeNotFound,
        Code:       code,
        Message:    message,
        HTTPStatus: http.StatusNotFound,
    }
}

func NewInternalError(code, message string, cause error) *AppError {
    return &AppError{
        Type:       ErrorTypeInternal,
        Code:       code,
        Message:    message,
        Cause:      cause,
        HTTPStatus: http.StatusInternalServerError,
    }
}

func NewExternalError(code, message string, cause error) *AppError {
    return &AppError{
        Type:       ErrorTypeExternal,
        Code:       code,
        Message:    message,
        Cause:      cause,
        HTTPStatus: http.StatusBadGateway,
    }
}

// 错误处理中间件
func ErrorHandlerMiddleware() gin.HandlerFunc {
    return func(c *gin.Context) {
        c.Next()
        
        if len(c.Errors) > 0 {
            err := c.Errors.Last().Err
            
            var appErr *AppError
            if errors.As(err, &appErr) {
                c.JSON(appErr.HTTPStatus, gin.H{
                    "error": appErr,
                })
            } else {
                // 未知错误
                c.JSON(http.StatusInternalServerError, gin.H{
                    "error": NewInternalError("UNKNOWN_ERROR", "An unexpected error occurred", err),
                })
            }
        }
    }
}

// optimization/memory.go
package optimization

import (
    "context"
    "sync"
    "time"
    
    "github.com/cloudwego/eino/schema"
)

// ObjectPool 对象池
type ObjectPool[T any] struct {
    pool sync.Pool
    new  func() T
}

func NewObjectPool[T any](newFunc func() T) *ObjectPool[T] {
    return &ObjectPool[T]{
        pool: sync.Pool{
            New: func() any {
                return newFunc()
            },
        },
        new: newFunc,
    }
}

func (p *ObjectPool[T]) Get() T {
    return p.pool.Get().(T)
}

func (p *ObjectPool[T]) Put(obj T) {
    p.pool.Put(obj)
}

// 预定义对象池
var (
    MessagePool = NewObjectPool(func() *schema.Message {
        return &schema.Message{}
    })
    
    DocumentPool = NewObjectPool(func() *schema.Document {
        return &schema.Document{
            MetaData: make(map[string]any),
        }
    })
    
    StringBuilderPool = NewObjectPool(func() *strings.Builder {
        return &strings.Builder{}
    })
)

// 使用示例
func ProcessMessages(messages []*schema.Message) []*schema.Message {
    result := make([]*schema.Message, 0, len(messages))
    
    for _, msg := range messages {
        // 从池中获取对象
        processed := MessagePool.Get()
        defer MessagePool.Put(processed) // 使用完毕后归还
        
        // 重置对象状态
        *processed = schema.Message{
            Role:    msg.Role,
            Content: processContent(msg.Content),
        }
        
        result = append(result, processed)
    }
    
    return result
}

// StreamBuffer 流缓冲区管理
type StreamBuffer[T any] struct {
    buffer   []T
    capacity int
    mu       sync.RWMutex
}

func NewStreamBuffer[T any](capacity int) *StreamBuffer[T] {
    return &StreamBuffer[T]{
        buffer:   make([]T, 0, capacity),
        capacity: capacity,
    }
}

func (sb *StreamBuffer[T]) Add(item T) bool {
    sb.mu.Lock()
    defer sb.mu.Unlock()
    
    if len(sb.buffer) >= sb.capacity {
        return false // 缓冲区已满
    }
    
    sb.buffer = append(sb.buffer, item)
    return true
}

func (sb *StreamBuffer[T]) Flush() []T {
    sb.mu.Lock()
    defer sb.mu.Unlock()
    
    if len(sb.buffer) == 0 {
        return nil
    }
    
    result := make([]T, len(sb.buffer))
    copy(result, sb.buffer)
    sb.buffer = sb.buffer[:0] // 清空但保留容量
    
    return result
}

// optimization/concurrency.go
package optimization

import (
    "context"
    "sync"
    "time"
)

// WorkerPool 工作池
type WorkerPool[T any, R any] struct {
    workers    int
    jobCh      chan Job[T, R]
    resultCh   chan Result[R]
    wg         sync.WaitGroup
    ctx        context.Context
    cancel     context.CancelFunc
}

type Job[T any, R any] struct {
    ID   string
    Data T
    Fn   func(context.Context, T) (R, error)
}

type Result[R any] struct {
    ID     string
    Data   R
    Error  error
}

func NewWorkerPool[T any, R any](ctx context.Context, workers int) *WorkerPool[T, R] {
    ctx, cancel := context.WithCancel(ctx)
    
    pool := &WorkerPool[T, R]{
        workers:  workers,
        jobCh:    make(chan Job[T, R], workers*2),
        resultCh: make(chan Result[R], workers*2),
        ctx:      ctx,
        cancel:   cancel,
    }
    
    // 启动工作协程
    for i := 0; i < workers; i++ {
        pool.wg.Add(1)
        go pool.worker()
    }
    
    return pool
}

func (wp *WorkerPool[T, R]) worker() {
    defer wp.wg.Done()
    
    for {
        select {
        case <-wp.ctx.Done():
            return
        case job := <-wp.jobCh:
            result := Result[R]{ID: job.ID}
            result.Data, result.Error = job.Fn(wp.ctx, job.Data)
            
            select {
            case wp.resultCh <- result:
            case <-wp.ctx.Done():
                return
            }
        }
    }
}

func (wp *WorkerPool[T, R]) Submit(job Job[T, R]) bool {
    select {
    case wp.jobCh <- job:
        return true
    case <-wp.ctx.Done():
        return false
    default:
        return false // 队列已满
    }
}

func (wp *WorkerPool[T, R]) Results() <-chan Result[R] {
    return wp.resultCh
}

func (wp *WorkerPool[T, R]) Close() {
    wp.cancel()
    close(wp.jobCh)
    wp.wg.Wait()
    close(wp.resultCh)
}

// RateLimiter 速率限制器
type RateLimiter struct {
    tokens chan struct{}
    ticker *time.Ticker
    done   chan struct{}
}

func NewRateLimiter(rate int, burst int) *RateLimiter {
    rl := &RateLimiter{
        tokens: make(chan struct{}, burst),
        ticker: time.NewTicker(time.Second / time.Duration(rate)),
        done:   make(chan struct{}),
    }
    
    // 初始填充令牌
    for i := 0; i < burst; i++ {
        rl.tokens <- struct{}{}
    }
    
    // 定期添加令牌
    go func() {
        for {
            select {
            case <-rl.ticker.C:
                select {
                case rl.tokens <- struct{}{}:
                default:
                    // 令牌桶已满
                }
            case <-rl.done:
                return
            }
        }
    }()
    
    return rl
}

func (rl *RateLimiter) Wait(ctx context.Context) error {
    select {
    case <-rl.tokens:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

func (rl *RateLimiter) Close() {
    rl.ticker.Stop()
    close(rl.done)
}