Method Stream

pkg/agent/model_fallback.go:165–223

Stream 执行流式请求，支持自动降级

(
	ctx context.Context,
	messages []types.Message,
	opts *provider.StreamOptions,
)

// Stream 执行流式请求，支持自动降级
func (m *ModelFallbackManager) Stream(
	ctx context.Context,
	messages []types.Message,
	opts *provider.StreamOptions,
) (<-chan provider.StreamChunk, error) {
	m.stats.TotalRequests++

	var lastErr error

	// 遍历所有启用的模型
	for i, fb := range m.fallbacks {
		if !fb.Enabled {
			continue
		}

		modelKey := fmt.Sprintf("%s/%s", fb.Config.Provider, fb.Config.Model)

		// 尝试执行，支持重试
		for retry := 0; retry <= fb.MaxRetries; retry++ {
			if retry > 0 {
				fallbackLog.Debug(ctx, "retrying model (stream)", map[string]any{"retry": retry, "max_retries": fb.MaxRetries, "model": modelKey})

				// 重试前等待一小段时间
				backoff := time.Duration(retry) * 500 * time.Millisecond
				select {
				case <-ctx.Done():
					return nil, ctx.Err()
				case <-time.After(backoff):
				}
			}

			// 执行流式请求
			stream, err := fb.provider.Stream(ctx, messages, opts)
			if err == nil {
				// 成功
				m.stats.SuccessRequests++
				m.stats.ModelUsageCount[modelKey]++
				m.currentIndex = i

				fallbackLog.Debug(ctx, "success with model (stream)", map[string]any{"model": modelKey, "retry": retry})
				return stream, nil
			}

			lastErr = err
			fallbackLog.Warn(ctx, "error with model (stream)", map[string]any{"model": modelKey, "retry": retry, "max_retries": fb.MaxRetries, "error": err})
		}

		// 所有重试都失败，尝试下一个模型
		if i < len(m.fallbacks)-1 {
			m.stats.FallbackCount++
			m.stats.LastFallbackTime = time.Now()
			fallbackLog.Info(ctx, "falling back to next model (stream)", map[string]any{"from_model": modelKey})
		}
	}

	// 所有模型都失败
	m.stats.FailedRequests++
	return nil, fmt.Errorf("all models failed (stream), last error: %w", lastErr)