orchestrator

package
v0.0.0-...-ca5b39a Latest Latest
Warning

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

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

README

Orchestrator Package

The orchestrator package is the core component of llm-proxy's bidirectional data transformation proxy. It implements the request pipeline that routes client requests through inbound transformers, unified request routing, outbound transformers, and provider communication.

Architecture Overview

The request pipeline follows this flow:

Client → Inbound Transformer → Unified Request Router → Outbound Transformer → Provider

This architecture provides:

  • Zero learning curve for OpenAI SDK users
  • Auto failover and load balancing across channels
  • Real-time tracing and per-project usage logs
  • Support for multiple API formats (OpenAI, Anthropic, Gemini, and custom variants)
  • Model-aware circuit breaking and auto-failover
  • Dynamic request body and header overrides with template support
  • Quota enforcement and prompt injection
  • Model access control via API key profiles
  • Channel selection with tag-based filtering
  • Native tool support for Anthropic and Google APIs

File Structure

Core Components
  • orchestrator.go - Main orchestrator implementation that coordinates the entire request pipeline
  • inbound.go - Handles inbound request processing and persistent stream wrapping
  • outbound.go - Manages outbound request processing and persistent stream wrapping
  • transformer.go - Persistent transformer factory with state management
  • request.go - Request persistence middleware
  • request_execution.go - Request execution coordination
  • retry.go - Retry logic and error handling utilities
  • state.go - Orchestrator state management (PersistenceState)
  • performance.go - Performance monitoring and metrics
  • prompt.go - Prompt injection logic for projects and models
  • quota.go - API key quota enforcement middleware
  • override.go - Request body and header override middleware with template support
  • model_circuit_breaker.go - Circuit breaker tracker for specific models on channels
  • tester.go - Testing utilities
Load Balancing
  • load_balancer.go - Core load balancing logic and LoadBalancer struct with partial sorting
  • load_balancer_debug.go - Debug utilities for load balancing decisions
  • lb_strategy_rr.go - Round-robin strategy with inactivity decay and request count capping
  • lb_strategy_bp.go - Error-aware strategy that penalizes channels with recent failures
  • lb_strategy_composite.go - Composite strategy combining multiple approaches with weights
  • lb_strategy_weight.go - Weight-based strategy using channel ordering weight
  • lb_strategy_model_aware_circuit_breaker.go - Strategy that considers model health on specific channels
  • lb_strategy_random.go - Simple random strategy for tie-breaking
Candidate Selection
  • candidates.go - Main candidate selection logic for channels/models with association cache
  • candidates_anthropic.go - Anthropic-specific candidate logic (native tools support)
  • candidates_google.go - Google/Gemini-specific candidate logic (native tools support)
  • candidates_stream_policy.go - Filters candidates based on request stream requirement and channel policy
  • select_candidates.go - Candidate selection middleware with API key profile filtering
Connection Tracking
  • connection_tracker.go - Connection tracking utilities (DefaultConnectionTracker)
  • connection_tracking.go - Connection state management
Model Management
  • model_mapper.go - Model mapping and compatibility
  • model_access.go - Model access control middleware for API key profiles
  • transform_options.go - Transform options application (ForceArrayInstructions, ForceArrayInputs)
Documentation
  • load-balancing.md - Detailed load balancing documentation
  • README.md - This file

Candidate Selection Stage

Before any score-based load balancing runs, an optional sticky stage can pin a candidate:

  • Trace Sticky Candidate Selection - Implemented by LoadBalancedSelector (a CandidateSelector decorator), not a LoadBalanceStrategy. When enabled via TraceStickyMode, it places the cached previous trace or thread channel at the front of the candidate list before score-based sorting. Because it does not implement Score(), it is not one of the combinable strategies below and cannot be passed to NewLoadBalancer.

Load Balancing Strategies

After the candidate selection stage, the orchestrator supports multiple score-based load balancing strategies that can be combined:

  1. Error Aware - Penalizes channels with recent failures, consecutive errors, and low success rates
  2. Round Robin - Distributes requests evenly across channels using historical request count with inactivity decay
  3. Connection Aware - Considers active connection count per channel
  4. Weight - Uses channel ordering weight for prioritization
  5. Model Aware Circuit Breaker - Dynamically penalizes channels where the requested model is currently failing
  6. Random - Adds a small random factor to break ties between channels with identical scores

The load balancer uses partial sorting for efficient top-k candidate selection based on retry policy configuration.

Key Interfaces

  • LoadBalanceStrategy - Interface for load balancing strategies with Score() and ScoreWithDebug() methods
  • ChannelMetricsProvider - Provides channel performance metrics (AggregatedMetrics)
  • RetryPolicyProvider - Supplies retry policy configuration
  • PreviousChannelProvider - Provides the cached previous channel for trace and thread sticky selection
  • CandidateSelector - Interface for selecting channel model candidates
  • ConnectionTracker - Interface for tracking active connections per channel
  • PromptProvider - Supplies enabled prompts for injection
  • ModelCircuitBreakerProvider - Provides model-level circuit breaker statistics and weights

Pipeline Architecture

The orchestrator uses a pipeline-based architecture with middleware support:

  1. Inbound Pipeline:

    • API key authentication
    • Quota enforcement
    • Model access control
    • Model mapping
    • Candidate selection (with API key profile and stream policy filtering)
    • Prompt injection
    • Request persistence

  2. Outbound Pipeline:

    • Channel selection
    • Request body and header overrides
    • Transform options application
    • Performance tracking
    • Request execution persistence
    • Connection tracking
    • Model circuit breaking (optional)
    • Provider communication
    • Response transformation
    • Usage logging

State Management

The PersistenceState struct maintains shared state across the request pipeline:

  • API key and user information
  • Request and execution tracking
  • Channel model candidates
  • Performance metrics
  • Load balancer and retry policy references

Testing

Comprehensive test coverage includes:

  • Unit tests for individual components
  • Integration tests for end-to-end flows
  • Load balancing strategy tests
  • Candidate selection tests (including cache, tags, decorator tests)
  • Performance and stress tests
  • Connection tracking tests

Run tests with: go test ./internal/server/orchestrator/...

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrChannelQueueFull is returned when the wait queue is full at acquire time.
	ErrChannelQueueFull = errors.New("channel concurrency queue full")
	// ErrChannelQueueTimeout is returned when the per-channel queue timeout elapses
	// while a request is still waiting for a slot.
	ErrChannelQueueTimeout = errors.New("channel concurrency queue wait timeout")
)

Sentinel errors returned by ChannelLimiter.Acquire.

View Source 
var ErrLocalRPMExhausted = errors.New("local channel rpm exhausted")
View Source 
var Module = fx.Module("orchestrator",
	fx.Provide(NewDefaultSelector),
	fx.Provide(NewAdapterCandidateSelector),
	fx.Provide(NewCandidateSelectorDiagnostics),
	fx.Provide(NewChannelLimiterManager),
	fx.Provide(func(svc *biz.ProviderQuotaService) ProviderQuotaStatusProvider { return svc }),
)

Functions

func EnableDebugMode 

func EnableDebugMode(ctx context.Context, opts *DebugOptions) context.Context

EnableDebugMode enables debug mode for the load balancer in the given context.

func ExtractErrorCode 

func ExtractErrorCode(err error) int

ExtractErrorCode extracts HTTP error code from error.

func ExtractErrorInfo 

func ExtractErrorInfo(err error) *biz.ExecutionErrorInfo

ExtractErrorInfo extracts HTTP status code and sanitized response body from error.

func ExtractErrorMessage 

func ExtractErrorMessage(err error) string

ExtractErrorMessage extracts HTTP error message from error.

func ExtractStatusCodeFromError 

func ExtractStatusCodeFromError(err error) int

ExtractStatusCodeFromError attempts to extract HTTP status code from various error types.

func IsDebugEnabled 

func IsDebugEnabled(ctx context.Context) bool

IsDebugEnabled checks if debug mode is enabled in the context.

func NewPersistentTransformers 

func NewPersistentTransformers(state *PersistenceState, wrapped transformer.Inbound) (*PersistentInboundTransformer, *PersistentOutboundTransformer)

NewPersistentTransformers creates enhanced persistent transformers with pre-constructed state.

func NewQuotaExhaustedError 

func NewQuotaExhaustedError(modelName string) error

func SelectAPIFormat 

func SelectAPIFormat(endpoints []objects.ChannelEndpoint, req *llm.Request) string

SelectAPIFormat selects the most appropriate APIFormat from a channel's resolved endpoints based on the request type and inbound API format. Prefers an endpoint whose API format matches the inbound request format so that pass-through can be enabled when identical formats are used. Falls back to the first capable endpoint, then the first endpoint.

func WithDebugInfo 

func WithDebugInfo(ctx context.Context, info *Debug) context.Context

WithDebugInfo stores debug information in the context.

Types

type AdapterCandidateSelector 

type AdapterCandidateSelector struct {
	AdapterService *biz.AdapterService
	ChannelService *biz.ChannelService
}

AdapterCandidateSelector 只在当前适配器快照的目标池内生成候选。

func NewAdapterCandidateSelector 

func NewAdapterCandidateSelector(adapterService *biz.AdapterService, channelService *biz.ChannelService) *AdapterCandidateSelector

func (*AdapterCandidateSelector) Select 

func (s *AdapterCandidateSelector) Select(ctx context.Context, req *llm.Request) ([]*ChannelModelsCandidate, error)

Select 将逻辑模型解析为目标模型，并严格使用目标声明的出站协议。 缺少适配器上下文时直接报内部配置错误，绝不回退到全渠道选择。

type AnthropicNativeToolsSelector 

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

AnthropicNativeToolsSelector is a decorator that prioritizes candidates supporting Anthropic native tools. When a request contains Anthropic native tools (web_search -> web_search_20250305), this selector filters out candidates whose channels don't support these tools (e.g., deepseek_anthropic). If no compatible candidates are found, it falls back to all candidates (allowing downstream fallback logic).

func WithAnthropicNativeToolsSelector 

func WithAnthropicNativeToolsSelector(wrapped CandidateSelector) *AnthropicNativeToolsSelector

WithAnthropicNativeToolsSelector creates a selector that prioritizes Anthropic native tool compatible candidates.

func (*AnthropicNativeToolsSelector) Select 

func (s *AnthropicNativeToolsSelector) Select(ctx context.Context, req *llm.Request) ([]*ChannelModelsCandidate, error)

type AssociationCacheEntrySnapshot 

type AssociationCacheEntrySnapshot struct {
	ModelID                 string                      `json:"modelId"`
	Associations            []*objects.ModelAssociation `json:"associations"`
	CandidateCount          int                         `json:"candidateCount"`
	ChannelCount            int                         `json:"channelCount"`
	LatestChannelUpdateTime time.Time                   `json:"latestChannelUpdateTime"`
	LatestModelUpdateTime   time.Time                   `json:"latestModelUpdateTime"`
	ChannelCacheVersion     int64                       `json:"channelCacheVersion"`
	CachedAt                time.Time                   `json:"cachedAt"`
}

type CandidateSelector 

type CandidateSelector interface {
	Select(ctx context.Context, req *llm.Request) ([]*ChannelModelsCandidate, error)
}

CandidateSelector defines the interface for selecting channel model candidates.

type CandidateSelectorDiagnostics 

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

func NewCandidateSelectorDiagnostics 

func NewCandidateSelectorDiagnostics(defaultSelector *DefaultSelector) *CandidateSelectorDiagnostics

func (*CandidateSelectorDiagnostics) ReadAssociationCache 

func (d *CandidateSelectorDiagnostics) ReadAssociationCache() []AssociationCacheEntrySnapshot

type ChannelDebug 

type ChannelDebug struct {
	// ChannelID is the channel identifier
	ChannelID int
	// ChannelName is the channel name
	ChannelName string
	// TotalScore is the sum of all strategy scores
	TotalScore float64
	// StrategyScores contains detailed scores from each strategy
	StrategyScores []StrategyDebug
	// Rank is the final ranking (1 = highest priority)
	Rank int
}

ChannelDebug holds debug information for a single channel.

type ChannelDecision 

type ChannelDecision struct {
	// Channel is the channel object
	Channel *biz.Channel
	// OriginalIndex is the candidate's position before sorting.
	OriginalIndex int
	// TotalScore is the sum of all strategy scores
	TotalScore float64
	// StrategyScores contains scores from each strategy
	StrategyScores []StrategyScore
	// FinalRank is the final ranking (1 = highest priority)
	FinalRank int
}

ChannelDecision holds detailed scoring information for a single channel.

type ChannelLimiter 

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

ChannelLimiter provides per-channel admission control as a hard blocking semaphore: at most `capacity` requests are in flight at once, and excess requests wait in a FIFO queue. The in-flight counter never exceeds capacity, so the cap is strict regardless of queue mode.

  • Bounded queue (queueSize > 0): up to queueSize requests wait; further arrivals get ErrChannelQueueFull immediately.
  • Unbounded queue (queueSize <= 0): excess requests always wait and are never rejected. This is the default when no QueueSize is configured.

Every waiter additionally honors an optional per-channel timeout (ErrChannelQueueTimeout) and the caller's context.

Caller must ensure capacity > 0.

func NewChannelLimiter 

func NewChannelLimiter(capacity, queueSize int, timeoutMs int64) *ChannelLimiter

NewChannelLimiter creates a limiter. timeoutMs == 0 means "no per-channel timeout" and the caller's context becomes the only deadline.

func (*ChannelLimiter) Acquire 

func (l *ChannelLimiter) Acquire(ctx context.Context) error

Acquire requests a slot. Returning nil means a slot was granted and Release MUST be called exactly once to give it back. Any non-nil error means no slot was acquired and Release must NOT be called.

Possible non-nil errors:

  • ErrChannelQueueFull when a bounded queue has no remaining capacity at entry time
  • ErrChannelQueueTimeout when the per-channel timeout elapses while waiting
  • ctx.Err() when the caller's context is cancelled (overrides the timeout)

func (*ChannelLimiter) Release 

func (l *ChannelLimiter) Release()

Release returns one slot. It transfers the slot directly to the head waiter (FIFO fairness) when one is queued; otherwise it decrements inFlight. Guards against decrementing below zero so an unmatched Release is a no-op rather than a panic.

func (*ChannelLimiter) Stats 

func (l *ChannelLimiter) Stats() (inFlight, waiting int)

Stats returns the current in-flight and waiting counts. Used by the load balancer scoring layer.

type ChannelLimiterManager 

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

ChannelLimiterManager owns one ChannelLimiter per channel, recreating an entry whenever the channel's rate-limit configuration changes.

The manager only allocates a limiter when MaxConcurrent > 0. Channels without a concurrency limit return nil from GetOrCreate so callers can fast-path past admission control entirely.

func NewChannelLimiterManager 

func NewChannelLimiterManager() *ChannelLimiterManager

NewChannelLimiterManager returns an empty manager.

func (*ChannelLimiterManager) Forget 

func (m *ChannelLimiterManager) Forget(channelID int)

Forget drops the limiter for a channel. Called from Channel CRUD paths after Update or Delete so the next GetOrCreate sees fresh configuration.

Forgetting a limiter while requests are still holding slots is safe — those requests Release into a now-orphaned limiter, GC'd once the last reference drops.

func (*ChannelLimiterManager) GetOrCreate 

func (m *ChannelLimiterManager) GetOrCreate(ch *biz.Channel) *ChannelLimiter

GetOrCreate returns the limiter for the given channel, rebuilding the entry transparently when the channel's rate-limit configuration has changed.

Returns nil when the channel has no concurrency limit configured. Callers should treat nil as "no admission control" and proceed without Acquire/Release.

When concurrency limiting is disabled but a stale entry exists (i.e. the user just cleared MaxConcurrent), the entry is dropped so Stats/Snapshot stop reporting it and downstream scoring code can't dereference now-nil rate-limit pointers. In-flight requests that still hold slots Release into the orphaned limiter; the limiter is GC'd once the last reference drops.

func (*ChannelLimiterManager) Snapshot 

func (m *ChannelLimiterManager) Snapshot() []ChannelLimiterSnapshot

Snapshot returns a copy of the current limiter state across all known channels. Safe to call from observable-gauge callbacks; the returned slice does not share any pointers with the manager's internal state.

func (*ChannelLimiterManager) Stats 

func (m *ChannelLimiterManager) Stats(channelID int) (inFlight, waiting int, ok bool)

Stats returns the current load for a channel. ok=false means the channel has no limiter (i.e. no concurrency limit configured), which the load balancer should treat as "unlimited capacity" rather than a hard zero.

type ChannelLimiterMetrics 

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

ChannelLimiterMetrics is the per-channel observability surface for the admission control system.

Five instruments:

  • axonhub_channel_inflight (observable gauge) — current in-flight count
  • axonhub_channel_queue_waiting (observable gauge) — current queue depth
  • axonhub_channel_queue_full_total (counter) — cumulative queue-full rejections
  • axonhub_channel_queue_timeout_total (counter) — cumulative wait-timeout exits
  • axonhub_channel_queue_wait_seconds (histogram) — wait time on successful acquires

The gauges read live state from ChannelLimiterManager.Snapshot() at scrape time; counters/histogram are pushed by middleware on each event.

func NewChannelLimiterMetrics 

func NewChannelLimiterMetrics(meter metric.Meter, manager *ChannelLimiterManager) (*ChannelLimiterMetrics, error)

NewChannelLimiterMetrics registers the channel-limiter metric instruments and wires the gauges' callback against manager. Pass a nil meter to obtain a no-op metrics struct (useful for tests that do not initialize OTel).

func (*ChannelLimiterMetrics) IncQueueFull 

func (m *ChannelLimiterMetrics) IncQueueFull(ctx context.Context, ch *biz.Channel)

IncQueueFull increments the queue-full rejection counter.

func (*ChannelLimiterMetrics) IncQueueTimeout 

func (m *ChannelLimiterMetrics) IncQueueTimeout(ctx context.Context, ch *biz.Channel)

IncQueueTimeout increments the queue-timeout exit counter.

func (*ChannelLimiterMetrics) ObserveQueueWait 

func (m *ChannelLimiterMetrics) ObserveQueueWait(ctx context.Context, ch *biz.Channel, dur time.Duration)

ObserveQueueWait records the wait duration for a successful Acquire.

type ChannelLimiterSnapshot 

type ChannelLimiterSnapshot struct {
	ChannelID   int
	ChannelName string
	InFlight    int
	Waiting     int
}

ChannelLimiterSnapshot is a point-in-time view of one channel's limiter for observability callbacks.

type ChannelMetricsProvider 

type ChannelMetricsProvider interface {
	GetChannelMetrics(ctx context.Context, channelID int) (*biz.AggregatedMetrics, error)
}

ChannelMetricsProvider provides channel performance metrics.

type ChannelModelsCandidate 

type ChannelModelsCandidate struct {
	Channel      *biz.Channel
	Priority     int
	Models       []biz.ChannelModelEntry
	APIFormat    string                   // selected endpoint API format for this candidate
	TraceSticky  bool                     // selected from the last successful trace or thread channel
	StreamPolicy objects.CapabilityPolicy // 目标级流式策略；空值表示跟随渠道策略
}

ChannelModelsCandidate represents a resolved channel and its matched model entries.

type ChannelQueueError 

type ChannelQueueError struct {
	ChannelID   int
	ChannelName string
	Reason      string
	Cause       error
	// contains filtered or unexported fields
}

ChannelQueueError represents a channel-level admission failure raised by ChannelLimiter inside the channelLimiter middleware.

It joins the typed cause (ErrChannelQueueFull / ErrChannelQueueTimeout) with a synthetic *httpclient.Error so the inbound TransformError path can pluck out a 429 response shape via errors.AsType[*httpclient.Error].

No Retry-After is set: the chat handler drops headers from synthetic errors, and its absence keeps HasRetryAfterHeader off so this local rejection cannot be misread as an upstream 429 by the cooldown middleware.

func (*ChannelQueueError) Error 

func (e *ChannelQueueError) Error() string

func (*ChannelQueueError) Unwrap 

func (e *ChannelQueueError) Unwrap() []error

Unwrap exposes both the typed sentinel and the synthetic transport error so errors.Is matches ErrChannelQueueFull / ErrChannelQueueTimeout while errors.AsType[*httpclient.Error] finds the 429 response shape.

type ChannelRequestTracker 

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

ChannelRequestTracker tracks per-channel request and token counts within fixed natural-minute buckets for rate limiting. It also manages cooldown periods for channels that received 429 errors.

func NewChannelRequestTracker 

func NewChannelRequestTracker() *ChannelRequestTracker

NewChannelRequestTracker creates a new rate limit tracker.

func (*ChannelRequestTracker) AddTokens 

func (t *ChannelRequestTracker) AddTokens(channelID int, tokens int64)

AddTokens adds token count for a channel.

func (*ChannelRequestTracker) GetCooldownUntil 

func (t *ChannelRequestTracker) GetCooldownUntil(channelID int) (time.Time, bool)

GetCooldownUntil returns the cooldown expiration time for a channel. Returns false if the channel is not in cooldown or the cooldown has expired.

func (*ChannelRequestTracker) GetRequestCount 

func (t *ChannelRequestTracker) GetRequestCount(channelID int) int64

GetRequestCount returns the current request count for a channel in the current window.

func (*ChannelRequestTracker) GetTokenCount 

func (t *ChannelRequestTracker) GetTokenCount(channelID int) int64

GetTokenCount returns the current token count for a channel in the current window.

func (*ChannelRequestTracker) IsCoolingDown 

func (t *ChannelRequestTracker) IsCoolingDown(channelID int) bool

IsCoolingDown checks if a channel is currently in a cooldown period. It also performs lazy cleanup by removing expired cooldown entries.

func (*ChannelRequestTracker) SetCooldown 

func (t *ChannelRequestTracker) SetCooldown(channelID int, until time.Time)

SetCooldown sets a cooldown period for a channel until the specified time. It only extends the cooldown; a shorter value will not overwrite an existing longer one.

func (*ChannelRequestTracker) TryAcquireRequest 

func (t *ChannelRequestTracker) TryAcquireRequest(channelID int, limit int64) bool

TryAcquireRequest atomically checks and consumes one request slot for a channel in the current fixed minute bucket. A non-positive limit means the caller has no RPM limit configured, so no slot is consumed.

type ChannelSelectionTracker 

type ChannelSelectionTracker interface {
	IncrementChannelSelection(channelID int)
}

ChannelSelectionTracker tracks channel selections for load balancing. This is used to increment request count at selection time rather than completion time, ensuring concurrent/burst requests don't all select the same channel.

type ChatCompletionOrchestrator 

type ChatCompletionOrchestrator struct {
	Inbound            transformer.Inbound
	RequestService     *biz.RequestService
	ChannelService     *biz.ChannelService
	SystemService      *biz.SystemService
	UsageLogService    *biz.UsageLogService
	QuotaService       *biz.QuotaService
	LiveStreamRegistry *biz.LiveStreamRegistry
	PromptProvider     PromptProvider
	PromptProtecter    PromptProtecter
	Middlewares        []pipeline.Middleware
	PipelineFactory    *pipeline.Factory
	ModelMapper        *ModelMapper
	// contains filtered or unexported fields
}

func NewChatCompletionOrchestrator 

func NewChatCompletionOrchestrator(
	channelService *biz.ChannelService,
	defaultSelector *DefaultSelector,
	requestService *biz.RequestService,
	httpClient *httpclient.HttpClient,
	inbound transformer.Inbound,
	systemService *biz.SystemService,
	usageLogService *biz.UsageLogService,
	promptService *biz.PromptService,
	quotaService *biz.QuotaService,
	promptProtectionRuleService *biz.PromptProtectionRuleService,
	liveStreamRegistry *biz.LiveStreamRegistry,
	channelLimiterManager *ChannelLimiterManager,
	quotaProvider ProviderQuotaStatusProvider,
) *ChatCompletionOrchestrator

func (*ChatCompletionOrchestrator) Process 

func (processor *ChatCompletionOrchestrator) Process(ctx context.Context, request *httpclient.Request) (ChatCompletionResult, error)

func (*ChatCompletionOrchestrator) WithAllowedChannels 

func (processor *ChatCompletionOrchestrator) WithAllowedChannels(allowedChannelIDs []int) *ChatCompletionOrchestrator

func (*ChatCompletionOrchestrator) WithChannelSelector 

func (processor *ChatCompletionOrchestrator) WithChannelSelector(selector CandidateSelector) *ChatCompletionOrchestrator

func (*ChatCompletionOrchestrator) WithProxy 

func (processor *ChatCompletionOrchestrator) WithProxy(proxy *httpclient.ProxyConfig) *ChatCompletionOrchestrator

type ChatCompletionResult 

type ChatCompletionResult struct {
	ChatCompletion       *httpclient.Response
	ChatCompletionStream streams.Stream[*httpclient.StreamEvent]
}

type CompositeStrategy 

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

CompositeStrategy combines multiple strategies with configurable weights.

func NewCompositeStrategy 

func NewCompositeStrategy(strategies ...LoadBalanceStrategy) *CompositeStrategy

NewCompositeStrategy creates a new composite strategy.

func (*CompositeStrategy) Name 

func (c *CompositeStrategy) Name() string

Name returns the strategy name.

func (*CompositeStrategy) Score 

func (c *CompositeStrategy) Score(ctx context.Context, channel *biz.Channel) float64

Score combines all strategy scores with their weights. Production path without debug logging.

func (*CompositeStrategy) ScoreWithDebug 

func (c *CompositeStrategy) ScoreWithDebug(ctx context.Context, channel *biz.Channel) (float64, StrategyScore)

ScoreWithDebug combines all strategy scores with detailed debug information. Debug path with comprehensive logging.

func (*CompositeStrategy) WithWeights 

func (c *CompositeStrategy) WithWeights(weights ...float64) *CompositeStrategy

WithWeights sets custom weights for the strategies. weights slice should match the order of strategies.

type Debug 

type Debug struct {
	// RequestID is the unique identifier for the request
	RequestID string
	// Timestamp when the decision was made
	Timestamp time.Time
	// Model being requested
	Model string
	// InputChannels are channels before sorting
	InputChannels []ChannelDebug
	// OutputChannels are channels after sorting
	OutputChannels []ChannelDebug
	// TotalDuration is the time spent on load balancing
	TotalDuration time.Duration
}

Debug holds detailed debug information about a load balancing decision.

func GetDebugInfo 

func GetDebugInfo(ctx context.Context) *Debug

GetDebugInfo retrieves debug information from the context if available.

type DebugOptions 

type DebugOptions struct {
	// Enabled indicates whether debug mode is active
	Enabled bool

	// RecordDecisionDetails records detailed decision information
	RecordDecisionDetails bool

	// RecordStrategyDetails records detailed strategy calculation info
	RecordStrategyDetails bool

	// MaxRecordsPerMinute limits debug records to prevent log flooding
	MaxRecordsPerMinute int
}

DebugOptions holds the configuration for debug mode.

func DefaultDebugOptions 

func DefaultDebugOptions() *DebugOptions

DefaultDebugOptions returns default debug options.

func GetDebugOptions 

func GetDebugOptions(ctx context.Context) *DebugOptions

GetDebugOptions retrieves debug options from the context.

type DecisionLog 

type DecisionLog struct {
	// Timestamp when the decision was made
	Timestamp time.Time
	// ChannelCount is the number of channels considered
	ChannelCount int
	// TotalDuration is the time spent on load balancing
	TotalDuration time.Duration
	// Channels contains detailed information for each channel
	Channels []ChannelDecision
}

DecisionLog represents a complete load balancing decision.

type DefaultSelector 

type DefaultSelector struct {
	ChannelService *biz.ChannelService
	ModelService   *biz.ModelService // Optional: for llm-proxy Model resolution
	SystemService  *biz.SystemService
	// contains filtered or unexported fields
}

DefaultSelector directly selects enabled channels supporting the requested model.

func NewDefaultSelector 

func NewDefaultSelector(channelService *biz.ChannelService, modelService *biz.ModelService, systemService *biz.SystemService) *DefaultSelector

func (*DefaultSelector) DiscoverTestModelTargets 

func (s *DefaultSelector) DiscoverTestModelTargets(ctx context.Context, modelID string, protocol string) ([]*TestModelTarget, error)

DiscoverTestModelTargets 返回指定协议池中有完整匹配 endpoint 的具体目标。 复用 DefaultSelector 的模型设置、开发者继承、条件关联和 runtime channel cache；因此查询结果与随后执行测试使用的是同一套服务端裁决逻辑。

func (*DefaultSelector) Select 

func (s *DefaultSelector) Select(ctx context.Context, req *llm.Request) ([]*ChannelModelsCandidate, error)

type ErrorAwareStrategy 

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

ErrorAwareStrategy deprioritizes channels based on their recent error history. It calculates a health score by applying time-decayed penalties for failures.

This strategy only applies PENALTIES for errors and does not provide boosts for successful requests. This design prevents the "Matthew effect" where high-performing channels dominate the distribution at the expense of others.

Historical success rate is intentionally excluded to allow channels to recover quickly after a period of instability, as long as they are currently working.

Penalties applied:

  • Consecutive failures: -30 per failure, decaying linearly over the cooldown period.
  • Recent failure: A base penalty of -40 that decays linearly over the cooldown period.

func NewErrorAwareStrategy 

func NewErrorAwareStrategy(metricsProvider ChannelMetricsProvider) *ErrorAwareStrategy

NewErrorAwareStrategy creates a new error-aware strategy.

func (*ErrorAwareStrategy) Name 

func (s *ErrorAwareStrategy) Name() string

Name returns the strategy name.

func (*ErrorAwareStrategy) Score 

func (s *ErrorAwareStrategy) Score(ctx context.Context, channel *biz.Channel) float64

Score returns a health score based on recent errors. Production path without debug logging.

func (*ErrorAwareStrategy) ScoreWithDebug 

func (s *ErrorAwareStrategy) ScoreWithDebug(ctx context.Context, channel *biz.Channel) (float64, StrategyScore)

ScoreWithDebug returns a health score with detailed debug information. Debug path with comprehensive logging.

type GoogleNativeToolsSelector 

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

GoogleNativeToolsSelector is a decorator that prioritizes candidates supporting Google native tools. When a request contains Google native tools (google_search, google_url_context, google_code_execution), this selector filters out candidates whose channels don't support these tools (e.g., gemini_openai). If no compatible candidates are found, it falls back to all candidates (allowing downstream fallback logic).

func WithGoogleNativeToolsSelector 

func WithGoogleNativeToolsSelector(wrapped CandidateSelector) *GoogleNativeToolsSelector

WithGoogleNativeToolsSelector creates a selector that prioritizes Google native tool compatible candidates.

func (*GoogleNativeToolsSelector) Select 

func (s *GoogleNativeToolsSelector) Select(ctx context.Context, req *llm.Request) ([]*ChannelModelsCandidate, error)

type InboundPersistentStream 

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

InboundPersistentStream wraps a stream and tracks all responses for final saving to database. It implements the streams.Stream interface and handles persistence in the Close method.

func NewInboundPersistentStream 

func NewInboundPersistentStream(
	ctx context.Context,
	stream streams.Stream[*httpclient.StreamEvent],
	request *ent.Request,
	requestExec *ent.RequestExecution,
	requestService *biz.RequestService,
	transformer transformer.Inbound,
	perf *biz.PerformanceRecord,
	state *PersistenceState,
) *InboundPersistentStream

func (*InboundPersistentStream) Close 

func (ts *InboundPersistentStream) Close() error

func (*InboundPersistentStream) Current 

func (ts *InboundPersistentStream) Current() *httpclient.StreamEvent

func (*InboundPersistentStream) Err 

func (ts *InboundPersistentStream) Err() error

func (*InboundPersistentStream) Next 

func (ts *InboundPersistentStream) Next() bool

type LatencyAwareStrategy 

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

LatencyAwareStrategy prioritizes channels using request-type-specific UX signals. Streaming requests are scored by first-token latency plus output throughput. Non-streaming requests are scored by end-to-end latency.

func NewLatencyAwareStrategy 

func NewLatencyAwareStrategy(metricsProvider ChannelMetricsProvider) *LatencyAwareStrategy

func (*LatencyAwareStrategy) Name 

func (s *LatencyAwareStrategy) Name() string

func (*LatencyAwareStrategy) Score 

func (s *LatencyAwareStrategy) Score(ctx context.Context, channel *biz.Channel) float64

func (*LatencyAwareStrategy) ScoreWithDebug 

func (s *LatencyAwareStrategy) ScoreWithDebug(ctx context.Context, channel *biz.Channel) (float64, StrategyScore)

type LoadBalanceStrategy 

type LoadBalanceStrategy interface {
	// Score calculates a score for a channel. Higher scores indicate higher priority.
	// Returns a score between 0 and 1000.
	// This is the production path with minimal overhead.
	Score(ctx context.Context, channel *biz.Channel) float64

	// ScoreWithDebug calculates a score with detailed debug information.
	// Returns the score and a StrategyScore with debug details.
	// This should have identical logic to Score() except for debug logging.
	ScoreWithDebug(ctx context.Context, channel *biz.Channel) (float64, StrategyScore)

	// Name returns the strategy name for debugging and logging.
	Name() string
}

LoadBalanceStrategy defines the interface for load balancing strategies. Each strategy can score and sort channels based on different criteria.

type LoadBalancedSelector 

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

LoadBalancedSelector is a decorator that sorts candidates using load balancing strategies.

func WithLoadBalancedSelector 

func WithLoadBalancedSelector(wrapped CandidateSelector, loadBalancer *LoadBalancer, policy RetryPolicyProvider) *LoadBalancedSelector

WithLoadBalancedSelector creates a selector that applies load balancing to sort candidates. The policy is used to determine the retry policy for early stopping.

func WithTraceStickyLoadBalancedSelector 

func WithTraceStickyLoadBalancedSelector(
	wrapped CandidateSelector,
	loadBalancer *LoadBalancer,
	policy RetryPolicyProvider,
	previousChannelProvider PreviousChannelProvider,
) *LoadBalancedSelector

WithTraceStickyLoadBalancedSelector creates a load-balanced selector that can prioritize the last successful trace or thread channel before normal load balancing.

func (*LoadBalancedSelector) Select 

func (s *LoadBalancedSelector) Select(ctx context.Context, req *llm.Request) ([]*ChannelModelsCandidate, error)

type LoadBalancer 

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

LoadBalancer applies multiple strategies to sort channels by priority.

func NewLoadBalancer 

func NewLoadBalancer(systemService RetryPolicyProvider, selectionTracker ChannelSelectionTracker, strategies ...LoadBalanceStrategy) *LoadBalancer

NewLoadBalancer creates a new load balancer with the given strategies. All strategy scores are summed with equal weight; registration order does not affect scoring.

func (*LoadBalancer) Sort 

func (lb *LoadBalancer) Sort(ctx context.Context, candidates []*ChannelModelsCandidate, model string, stream bool) []*ChannelModelsCandidate

Sort sorts candidates according to the configured strategies. Returns a new slice with top k candidates sorted by descending priority. The top k value is calculated internally based on the retry policy.

func (*LoadBalancer) SortWithoutTracking 

func (lb *LoadBalancer) SortWithoutTracking(ctx context.Context, candidates []*ChannelModelsCandidate, model string, stream bool) []*ChannelModelsCandidate

SortWithoutTracking sorts candidates without recording a channel selection. It is used when a sticky candidate is already selected and the remaining candidates are only fallbacks.

func (*LoadBalancer) TrackSelection 

func (lb *LoadBalancer) TrackSelection(candidate *ChannelModelsCandidate)

TrackSelection records a selected channel.

func (*LoadBalancer) WithRoundRobinHealthFilter 

func (lb *LoadBalancer) WithRoundRobinHealthFilter(filter *RoundRobinHealthStrategy) *LoadBalancer

WithRoundRobinHealthFilter moves recently failing channels behind healthy round-robin candidates after the round-robin order is calculated.

func (*LoadBalancer) WithoutWeightTieBreaker 

func (lb *LoadBalancer) WithoutWeightTieBreaker() *LoadBalancer

WithoutWeightTieBreaker disables OrderingWeight as a tie-breaker and uses input order when scores are equal.

type LocalRPMExhaustedError 

type LocalRPMExhaustedError struct {
	ChannelID   int
	ChannelName string
	Limit       int64
	// contains filtered or unexported fields
}

LocalRPMExhaustedError represents a local per-channel RPM admission failure.

The embedded synthetic 429 lets existing response transformation produce a rate-limit-shaped client error, while the typed wrapper lets retry and cooldown code distinguish it from provider 429 responses.

func (*LocalRPMExhaustedError) Error 

func (e *LocalRPMExhaustedError) Error() string

func (*LocalRPMExhaustedError) Unwrap 

func (e *LocalRPMExhaustedError) Unwrap() []error

type ModelAwareCircuitBreakerStrategy 

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

ModelAwareCircuitBreakerStrategy implements a load balancing strategy that considers model circuit breaker status. It adjusts channel scores based on the state of the requested model on each channel.

func NewModelAwareCircuitBreakerStrategy 

func NewModelAwareCircuitBreakerStrategy(cbProvider ModelCircuitBreakerProvider) *ModelAwareCircuitBreakerStrategy

NewModelAwareCircuitBreakerStrategy creates a new circuit breaker load balancing strategy.

func (*ModelAwareCircuitBreakerStrategy) Name 

func (s *ModelAwareCircuitBreakerStrategy) Name() string

Name returns the strategy name.

func (*ModelAwareCircuitBreakerStrategy) Score 

func (s *ModelAwareCircuitBreakerStrategy) Score(ctx context.Context, channel *biz.Channel) float64

Score calculates the score based on model circuit breaker status. This is the production path with minimal overhead.

func (*ModelAwareCircuitBreakerStrategy) ScoreWithDebug 

func (s *ModelAwareCircuitBreakerStrategy) ScoreWithDebug(ctx context.Context, channel *biz.Channel) (float64, StrategyScore)

ScoreWithDebug calculates the score with detailed debug information.

type ModelCircuitBreakerProvider 

type ModelCircuitBreakerProvider interface {
	GetEffectiveWeight(ctx context.Context, channelID int, modelID string, baseWeight float64) float64
	GetModelCircuitBreakerStats(ctx context.Context, channelID int, modelID string) *biz.ModelCircuitBreakerStats
}

ModelCircuitBreakerProvider provides model circuit breaker information.

type ModelMapper 

type ModelMapper struct{}

ModelMapper handles model mapping based on API key profiles.

func NewModelMapper 

func NewModelMapper() *ModelMapper

NewModelMapper creates a new ModelMapper instance.

func (*ModelMapper) MapModel 

func (m *ModelMapper) MapModel(ctx context.Context, apiKey *ent.APIKey, originalModel string) string

MapModel applies model mapping from API key profiles if an active profile exists Returns the mapped model name or the original model if no mapping is found.

func (*ModelMapper) ReplaceResponseModel 

func (m *ModelMapper) ReplaceResponseModel(response *llm.Response, requestModel string)

ReplaceResponseModel replaces the model field in llm.Response with the original client request model.

type NoopPerformanceRecording 

type NoopPerformanceRecording struct {
	pipeline.DummyMiddleware
}

func (*NoopPerformanceRecording) Name 

func (m *NoopPerformanceRecording) Name() string

type OutboundPersistentStream 

type OutboundPersistentStream struct {
	RequestService  *biz.RequestService
	UsageLogService *biz.UsageLogService
	// contains filtered or unexported fields
}

OutboundPersistentStream wraps a stream and tracks all responses for final saving to database. It implements the streams.Stream interface and handles persistence in the Close method.

func NewOutboundPersistentStream 

func NewOutboundPersistentStream(
	ctx context.Context,
	stream streams.Stream[*httpclient.StreamEvent],
	request *ent.Request,
	requestExec *ent.RequestExecution,
	requestService *biz.RequestService,
	usageLogService *biz.UsageLogService,
	outboundTransformer transformer.Outbound,
	perf *biz.PerformanceRecord,
	state *PersistenceState,
) *OutboundPersistentStream

func (*OutboundPersistentStream) Close 

func (ts *OutboundPersistentStream) Close() error

func (*OutboundPersistentStream) Current 

func (ts *OutboundPersistentStream) Current() *httpclient.StreamEvent

func (*OutboundPersistentStream) Err 

func (ts *OutboundPersistentStream) Err() error

func (*OutboundPersistentStream) Next 

func (ts *OutboundPersistentStream) Next() bool

type PersistenceState 

type PersistenceState struct {
	APIKey *ent.APIKey

	RequestService      *biz.RequestService
	UsageLogService     *biz.UsageLogService
	ChannelService      *biz.ChannelService
	PromptProvider      PromptProvider
	PromptProtecter     PromptProtecter
	RetryPolicyProvider RetryPolicyProvider
	CandidateSelector   CandidateSelector
	LoadBalancer        *LoadBalancer

	// Adapter 运行时元数据来自当前选中的不可变快照。
	Adapter *objects.RuntimeAdapter

	// Request state
	ModelMapper *ModelMapper
	// Proxy config, will be used to override channel's default proxy config.
	Proxy *httpclient.ProxyConfig

	// OriginalModel is the model after API key profile mapping, used for channel selection
	OriginalModel string
	RawRequest    *httpclient.Request
	LlmRequest    *llm.Request

	// OriginalRequestStream stores the client's original stream intent before any
	// candidate-specific forcing to provider-side streaming happens.
	OriginalRequestStream *bool

	// Persistence state
	Request     *ent.Request
	RequestExec *ent.RequestExecution

	// ChannelModelsCandidates is the primary state for channel selection
	ChannelModelsCandidates []*ChannelModelsCandidate
	// Candidate state - current candidate index of ChannelModelsCandidates
	CurrentCandidateIndex int
	// CurrentCandidate is the currently selected candidate of ChannelModelsCandidates
	CurrentCandidate *ChannelModelsCandidate
	// CurrentModelIndex is the current model index in CurrentCandidate.Models
	CurrentModelIndex int

	// Perf is the performance record for the current request.
	Perf *biz.PerformanceRecord

	// StreamCompleted tracks whether the stream has response successfully completed.
	// This is used to distinguish between a stream that was canceled mid-way
	// versus a stream that completed successfully but the client disconnected
	// immediately after receiving the last chunk.
	StreamCompleted bool

	// RawProviderResponse stores the raw provider response for non-stream response pass-through.
	RawProviderResponse *httpclient.Response

	// RawProviderRequest stores the actual outbound provider request for pass-through checks.
	RawProviderRequest *httpclient.Request

	// RawStreamCh receives raw provider stream events for stream response pass-through.
	RawStreamCh chan *httpclient.StreamEvent

	// RawStreamErrRef points to the current attempt's local error variable used by the
	// captureRawProviderStream fan-out goroutine. Using a per-attempt pointer (instead of
	// a single shared field) prevents data races when retries spawn a new goroutine before
	// the previous one has exited.
	RawStreamErrRef *error

	// RawStreamCancel cancels the current attempt's fan-out goroutine started by
	// captureRawProviderStream. Must be called in PrepareForRetry and NextChannel so the
	// abandoned goroutine exits promptly and releases its upstream HTTP connection.
	RawStreamCancel context.CancelFunc

	// PassThroughApplied records whether the inbound request body was substituted during pass-through.
	PassThroughApplied bool
}

PersistenceState holds shared state with channel management and retry capabilities. TODO: move the dependencies out of the state to make it a real state.

type PersistentInboundTransformer 

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

PersistentInboundTransformer wraps an inbound transformer with enhanced capabilities.

func (*PersistentInboundTransformer) AggregateStreamChunks 

func (p *PersistentInboundTransformer) AggregateStreamChunks(ctx context.Context, chunks []*httpclient.StreamEvent) ([]byte, llm.ResponseMeta, error)

func (*PersistentInboundTransformer) TransformError 

func (p *PersistentInboundTransformer) TransformError(ctx context.Context, rawErr error) *httpclient.Error

func (*PersistentInboundTransformer) TransformRequest 

func (p *PersistentInboundTransformer) TransformRequest(ctx context.Context, request *httpclient.Request) (*llm.Request, error)

func (*PersistentInboundTransformer) TransformResponse 

func (p *PersistentInboundTransformer) TransformResponse(ctx context.Context, response *llm.Response) (*httpclient.Response, error)

func (*PersistentInboundTransformer) TransformStream 

func (p *PersistentInboundTransformer) TransformStream(ctx context.Context, stream streams.Stream[*llm.Response]) (streams.Stream[*httpclient.StreamEvent], error)

type PersistentOutboundTransformer 

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

PersistentOutboundTransformer wraps an outbound transformer with shared persistence state.

func (*PersistentOutboundTransformer) APIFormat 

func (p *PersistentOutboundTransformer) APIFormat() llm.APIFormat

APIFormat returns the API format of the transformer.

func (*PersistentOutboundTransformer) AggregateStreamChunks 

func (p *PersistentOutboundTransformer) AggregateStreamChunks(
	ctx context.Context, req *httpclient.Request,
	chunks []*httpclient.StreamEvent,
) ([]byte, llm.ResponseMeta, error)

func (*PersistentOutboundTransformer) CanRetry 

func (p *PersistentOutboundTransformer) CanRetry(err error) bool

CanRetry returns true if the current channel can be retried. It implements the pipeline.ChannelRetryable interface, it just check the error is retryable, the pipeline will ensure the maxSameChannelRetries is not exceeded.

func (*PersistentOutboundTransformer) CustomizeExecutor 

func (p *PersistentOutboundTransformer) CustomizeExecutor(executor pipeline.Executor) pipeline.Executor

CustomizeExecutor customizes the executor for the current channel. If the current channel has an executor, it will be used. Otherwise, the default executor will be used.

The customized executor will be used to execute the request. e.g. the aws bedrock process need a custom executor to handle the request. It implements the pipeline.ChannelCustomizedExecutor interface.

func (*PersistentOutboundTransformer) GetCurrentChannel 

func (p *PersistentOutboundTransformer) GetCurrentChannel() *biz.Channel

GetCurrentChannel returns the current channel.

func (*PersistentOutboundTransformer) GetCurrentModelID 

func (p *PersistentOutboundTransformer) GetCurrentModelID() string

GetCurrentModelID returns the current model ID for logging purposes.

func (*PersistentOutboundTransformer) GetRequest 

func (p *PersistentOutboundTransformer) GetRequest() *ent.Request

GetRequest returns the current request.

func (*PersistentOutboundTransformer) GetRequestExecution 

func (p *PersistentOutboundTransformer) GetRequestExecution() *ent.RequestExecution

GetRequestExecution returns the current request execution.

func (*PersistentOutboundTransformer) GetRequestedModel 

func (p *PersistentOutboundTransformer) GetRequestedModel() string

GetRequestedModel returns the originally requested model ID.

func (*PersistentOutboundTransformer) HasMoreChannels 

func (p *PersistentOutboundTransformer) HasMoreChannels() bool

HasMoreChannels returns true if there are more candidates available for retry. It implements the pipeline.Retryable interface.

func (*PersistentOutboundTransformer) NextChannel 

func (p *PersistentOutboundTransformer) NextChannel(ctx context.Context) error

NextChannel moves to the next available candidate for retry. It implements the pipeline.Retryable interface.

func (*PersistentOutboundTransformer) PrepareForRetry 

func (p *PersistentOutboundTransformer) PrepareForRetry(ctx context.Context) error

PrepareForRetry implements the pipeline.ChannelRetryable interface. This will reset the request execution for the same channel, so that the same request can be retried. It will try the next model in the same channel if available.

func (*PersistentOutboundTransformer) TransformError 

func (p *PersistentOutboundTransformer) TransformError(ctx context.Context, rawErr *httpclient.Error) *llm.ResponseError

func (*PersistentOutboundTransformer) TransformRequest 

func (p *PersistentOutboundTransformer) TransformRequest(ctx context.Context, llmRequest *llm.Request) (*httpclient.Request, error)

func (*PersistentOutboundTransformer) TransformResponse 

func (p *PersistentOutboundTransformer) TransformResponse(ctx context.Context, response *httpclient.Response) (*llm.Response, error)

func (*PersistentOutboundTransformer) TransformStream 

func (p *PersistentOutboundTransformer) TransformStream(ctx context.Context, req *httpclient.Request, stream streams.Stream[*httpclient.StreamEvent]) (streams.Stream[*llm.Response], error)

type PreviousChannelProvider 

type PreviousChannelProvider interface {
	GetPreviousChannelID(ctx context.Context, traceID int) (int, error)
	GetPreviousChannelIDByThread(ctx context.Context, threadID int) (int, error)
}

PreviousChannelProvider provides the most recently selected channel for trace and thread routing scopes.

type PromptProtecter 

type PromptProtecter interface {
	Protect(ctx context.Context, req *llm.Request) (*llm.Request, error)
}

type PromptProvider 

type PromptProvider interface {
	GetEnabledPrompts(ctx context.Context, projectID int) ([]*ent.Prompt, error)
}

type ProviderQuotaSelector 

type ProviderQuotaSelector struct {

	// FilteredCount holds the number of candidates removed by the last Select() call.
	// It is only populated in ExhaustedOnly mode; DePrioritize mode returns early
	// without setting it. Read after Select() to distinguish "no candidates due to
	// quota exhaustion" from "no candidates at all".
	FilteredCount int
	// contains filtered or unexported fields
}

func WithProviderQuotaSelector 

func WithProviderQuotaSelector(wrapped CandidateSelector, provider ProviderQuotaStatusProvider, systemService QuotaEnforcementSettingsProvider) *ProviderQuotaSelector

func (*ProviderQuotaSelector) Select 

func (s *ProviderQuotaSelector) Select(ctx context.Context, req *llm.Request) ([]*ChannelModelsCandidate, error)

type ProviderQuotaStatusProvider 

type ProviderQuotaStatusProvider interface {
	GetQuotaStatus(channelID int) *biz.QuotaChannelStatus
}

ProviderQuotaStatusProvider provides quota status information for channels.

type QuotaAwareStrategy 

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

func NewQuotaAwareStrategy 

func NewQuotaAwareStrategy(provider ProviderQuotaStatusProvider, systemService QuotaEnforcementSettingsProvider) *QuotaAwareStrategy

func (*QuotaAwareStrategy) Name 

func (s *QuotaAwareStrategy) Name() string

func (*QuotaAwareStrategy) Score 

func (s *QuotaAwareStrategy) Score(ctx context.Context, channel *biz.Channel) float64

func (*QuotaAwareStrategy) ScoreWithDebug 

func (s *QuotaAwareStrategy) ScoreWithDebug(ctx context.Context, channel *biz.Channel) (float64, StrategyScore)

type QuotaEnforcementSettingsProvider 

type QuotaEnforcementSettingsProvider interface {
	QuotaEnforcementSettingsOrDefault(ctx context.Context) *biz.QuotaEnforcementSettings
}

type QuotaExhaustedError 

type QuotaExhaustedError struct {
	ModelName string
}

QuotaExhaustedError is returned when all channels are quota exhausted for a model.

func (*QuotaExhaustedError) Error 

func (e *QuotaExhaustedError) Error() string

type RandomStrategy 

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

RandomStrategy adds a small random score to break ties between channels with identical scores from other strategies.

func NewRandomStrategy 

func NewRandomStrategy() *RandomStrategy

NewRandomStrategy creates a new random strategy.

func (*RandomStrategy) Name 

func (s *RandomStrategy) Name() string

Name returns the strategy name.

func (*RandomStrategy) Score 

func (s *RandomStrategy) Score(ctx context.Context, channel *biz.Channel) float64

Score returns a random score between min and max.

func (*RandomStrategy) ScoreWithDebug 

func (s *RandomStrategy) ScoreWithDebug(ctx context.Context, channel *biz.Channel) (float64, StrategyScore)

ScoreWithDebug returns a random score with detailed debug information.

type RateLimitAwareStrategy 

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

RateLimitAwareStrategy adjusts channel scores based on configured RPM/TPM rate limits and per-channel concurrency limits. It composes two sub-scores (RPM/TPM and concurrency), returning the stricter one. Channels at any hard limit are returned as rateLimitExhaustedScore so the load balancer drops them to last.

func NewRateLimitAwareStrategy 

func NewRateLimitAwareStrategy(tracker *ChannelRequestTracker, limiterManager *ChannelLimiterManager) *RateLimitAwareStrategy

NewRateLimitAwareStrategy creates a new rate limit aware load balancing strategy. limiterManager may be nil for tests that do not exercise the concurrency dimension; in that case the concurrency sub-score collapses to maxScore.

func (*RateLimitAwareStrategy) Name 

func (s *RateLimitAwareStrategy) Name() string

Name returns the strategy name.

func (*RateLimitAwareStrategy) Score 

func (s *RateLimitAwareStrategy) Score(ctx context.Context, channel *biz.Channel) float64

Score is the production-path scorer with minimal overhead.

func (*RateLimitAwareStrategy) ScoreWithDebug 

func (s *RateLimitAwareStrategy) ScoreWithDebug(ctx context.Context, channel *biz.Channel) (float64, StrategyScore)

ScoreWithDebug returns the same score as Score along with detailed diagnostic info.

type RenderContext 

type RenderContext struct {
	// RequestModel is the model used in the original request.
	RequestModel string `json:"request_model"`
	// Model is the model sent to the LLM service.
	Model string `json:"model"`
	// Metadata is the metadata used in the current request.
	Metadata map[string]string `json:"metadata"`
	// RequestHeader is the filtered request headers used in the current request.
	RequestHeader map[string]string `json:"request_header"`
	// PromptCacheKey is the prompt cache key provided by the original request.
	PromptCacheKey string `json:"prompt_cache_key"`
	// ReasoningEffort is the reasoning effort used in the current request.
	ReasoningEffort string `json:"reasoning_effort"`
}

RenderContext is the context used for rendering override templates.

type RetryPolicyProvider 

type RetryPolicyProvider interface {
	RetryPolicyOrDefault(ctx context.Context) *biz.RetryPolicy
}

RetryPolicyProvider interface defines the methods needed from RetryPolicyProvider.

type RoundRobinHealthStrategy 

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

RoundRobinHealthStrategy pushes repeatedly failing channels behind healthy round-robin candidates. It is intentionally only used by the round-robin top-level strategy so adaptive balancing can keep its softer ErrorAware scoring.

func NewRoundRobinHealthStrategy 

func NewRoundRobinHealthStrategy(metricsProvider ChannelMetricsProvider) *RoundRobinHealthStrategy

func (*RoundRobinHealthStrategy) IsUnhealthy 

func (s *RoundRobinHealthStrategy) IsUnhealthy(ctx context.Context, channel *biz.Channel) bool

func (*RoundRobinHealthStrategy) Name 

func (s *RoundRobinHealthStrategy) Name() string

func (*RoundRobinHealthStrategy) Score 

func (s *RoundRobinHealthStrategy) Score(ctx context.Context, channel *biz.Channel) float64

func (*RoundRobinHealthStrategy) ScoreWithDebug 

func (s *RoundRobinHealthStrategy) ScoreWithDebug(ctx context.Context, channel *biz.Channel) (float64, StrategyScore)

type RoundRobinStrategy 

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

RoundRobinStrategy prioritizes channels based on their request count history. Channels with fewer historical requests get higher priority to ensure even load distribution. This strategy is particularly effective when combined with other strategies in a composite approach.

func NewRoundRobinStrategy 

func NewRoundRobinStrategy(metricsProvider ChannelMetricsProvider) *RoundRobinStrategy

NewRoundRobinStrategy creates a new round-robin load balancing strategy. This strategy implements true round-robin by prioritizing channels with fewer historical requests.

func (*RoundRobinStrategy) Name 

func (s *RoundRobinStrategy) Name() string

Name returns the strategy name.

func (*RoundRobinStrategy) Score 

func (s *RoundRobinStrategy) Score(ctx context.Context, channel *biz.Channel) float64

Score returns a priority score based on the channel's historical request count. Production path without debug logging. Channels with fewer requests receive higher scores to promote even distribution.

func (*RoundRobinStrategy) ScoreWithDebug 

func (s *RoundRobinStrategy) ScoreWithDebug(ctx context.Context, channel *biz.Channel) (float64, StrategyScore)

ScoreWithDebug returns a priority score with detailed debug information. Debug path with comprehensive logging.

type SelectedChannelsSelector 

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

SelectedChannelsSelector is a decorator that filters candidates by allowed channel IDs.

func WithSelectedChannelsSelector 

func WithSelectedChannelsSelector(wrapped CandidateSelector, allowedChannelIDs []int) *SelectedChannelsSelector

WithSelectedChannelsSelector creates a selector that filters by allowed channel IDs. If allowedChannelIDs is nil or empty, all candidates from the wrapped selector are returned.

func (*SelectedChannelsSelector) Select 

func (s *SelectedChannelsSelector) Select(ctx context.Context, req *llm.Request) ([]*ChannelModelsCandidate, error)

type SpecifiedChannelSelector 

type SpecifiedChannelSelector struct {
	ChannelService *biz.ChannelService
	ChannelID      objects.GUID
	// SelectedAPIKey, if non-empty, forces the outbound to use this specific API key.
	// Used by the channel key test flow to test a single key.
	SelectedAPIKey string
}

SpecifiedChannelSelector allows selecting specific channels (including disabled ones) for testing.

func NewSpecifiedChannelSelector 

func NewSpecifiedChannelSelector(channelService *biz.ChannelService, channelID objects.GUID) *SpecifiedChannelSelector

func (*SpecifiedChannelSelector) Select 

func (s *SpecifiedChannelSelector) Select(ctx context.Context, req *llm.Request) ([]*ChannelModelsCandidate, error)

type StrategyDebug 

type StrategyDebug struct {
	// StrategyName is the name of the strategy
	StrategyName string
	// Score is the score calculated
	Score float64
	// Duration is the time spent scoring
	Duration time.Duration
	// Details contains strategy-specific information
	Details map[string]any
}

StrategyDebug holds debug information for a single strategy's score.

type StrategyScore 

type StrategyScore struct {
	// StrategyName is the name of the strategy
	StrategyName string
	// Score is the score calculated by this strategy
	Score float64
	// Details contains strategy-specific information
	Details map[string]any
	// Duration is the time spent on scoring
	Duration time.Duration
}

StrategyScore holds the detailed scoring information from a single strategy.

type StreamPolicySelector 

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

func WithStreamPolicySelector 

func WithStreamPolicySelector(wrapped CandidateSelector) *StreamPolicySelector

func (*StreamPolicySelector) Select 

func (s *StreamPolicySelector) Select(ctx context.Context, req *llm.Request) ([]*ChannelModelsCandidate, error)

type TagsFilterSelector 

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

TagsFilterSelector is a decorator that filters candidates by allowed channel tags.

func WithChannelTagsFilterSelector 

func WithChannelTagsFilterSelector(wrapped CandidateSelector, tags []string, matchMode objects.ChannelTagsMatchMode) *TagsFilterSelector

WithChannelTagsFilterSelector creates a selector that filters by tags and match mode. If tags is empty, all candidates from the wrapped selector are returned.

func (*TagsFilterSelector) Select 

func (s *TagsFilterSelector) Select(ctx context.Context, req *llm.Request) ([]*ChannelModelsCandidate, error)

type TestAPIKeyResult 

type TestAPIKeyResult struct {
	KeyPrefix string
	Success   bool
	Latency   float64
	Error     *string
	Disabled  bool
}

TestAPIKeyResult represents the result of testing a single API key.

type TestChannelAPIKeysResult 

type TestChannelAPIKeysResult struct {
	ChannelID    objects.GUID
	Total        int
	SuccessCount int
	FailedCount  int
	Results      []*TestAPIKeyResult
}

TestChannelAPIKeysResult represents the aggregated result of testing all API keys.

type TestChannelOrchestrator 

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

TestChannelOrchestrator handles channel testing functionality. It is stateless and can be reused across multiple test requests.

func NewTestChannelOrchestrator 

func NewTestChannelOrchestrator(
	channelService *biz.ChannelService,
	requestService *biz.RequestService,
	systemService *biz.SystemService,
	usageLogService *biz.UsageLogService,
	promptProtectionRuleService *biz.PromptProtectionRuleService,
	httpClient *httpclient.HttpClient,
) *TestChannelOrchestrator

NewTestChannelOrchestrator creates a new TestChannelOrchestrator.

func (*TestChannelOrchestrator) TestAdapter 

func (processor *TestChannelOrchestrator) TestAdapter(
	ctx context.Context,
	selector CandidateSelector,
	modelID string,
	apiFormat string,
	proxy *httpclient.ProxyConfig,
) (*TestChannelResult, error)

TestAdapter 执行 Adapter runtime snapshot 中的指定逻辑模型绑定。

func (*TestChannelOrchestrator) TestChannel 

func (processor *TestChannelOrchestrator) TestChannel(
	ctx context.Context,
	channelID objects.GUID,
	modelID *string,
	proxy *httpclient.ProxyConfig,
) (*TestChannelResult, error)

TestChannel tests a specific channel with the default OpenAI Chat protocol.

func (*TestChannelOrchestrator) TestChannelAPIKeys 

func (processor *TestChannelOrchestrator) TestChannelAPIKeys(
	ctx context.Context,
	channelID objects.GUID,
	modelID *string,
	proxy *httpclient.ProxyConfig,
) (*TestChannelAPIKeysResult, error)

TestChannelAPIKeys tests all API keys for a specific channel individually.

func (*TestChannelOrchestrator) TestChannelWithProtocol 

func (processor *TestChannelOrchestrator) TestChannelWithProtocol(
	ctx context.Context,
	channelID objects.GUID,
	modelID *string,
	protocol string,
	proxy *httpclient.ProxyConfig,
) (*TestChannelResult, error)

TestChannelWithProtocol tests a specific channel without allowing a different protocol endpoint to be selected as a fallback.

func (*TestChannelOrchestrator) TestModel 

func (processor *TestChannelOrchestrator) TestModel(
	ctx context.Context,
	selector CandidateSelector,
	modelID string,
	protocol string,
	channelID objects.GUID,
	physicalModelID string,
	proxy *httpclient.ProxyConfig,
) (*TestChannelResult, error)

TestModel 执行逻辑 Model 的协议池测试。selector 必须由服务端根据该 Model 的有效协议池构造，调用方不得把它替换成全渠道 selector。

func (*TestChannelOrchestrator) TestSingleAPIKey 

func (processor *TestChannelOrchestrator) TestSingleAPIKey(
	ctx context.Context,
	channelID objects.GUID,
	key string,
	modelID *string,
	proxy *httpclient.ProxyConfig,
) (*TestAPIKeyResult, error)

TestSingleAPIKey tests a single API key for a channel. It verifies that the provided key belongs to the channel before testing.

type TestChannelRequest 

type TestChannelRequest struct {
	ChannelID objects.GUID
	ModelID   *string
}

TestChannelRequest represents a channel test request.

type TestChannelResult 

type TestChannelResult struct {
	Latency   float64
	Success   bool
	Message   *string
	Error     *string
	RequestID *objects.GUID
}

TestChannelResult 表示一次受管测试的结果。 RequestID 是本次主 Request 的 Relay ID；只有 pipeline 到达 RequestService.CreateRequest 后才会赋值，前置拒绝不会创建记录。

type TestModelTarget 

type TestModelTarget struct {
	ChannelID       objects.GUID
	ChannelName     string
	PhysicalModelID string
	Protocol        string
	APIFormat       string
}

TestModelTarget 是逻辑 Model 某个协议池中由服务端认可的具体目标。

type WeightRoundRobinStrategy 

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

WeightRoundRobinStrategy implements weighted round-robin load balancing. It distributes requests proportionally based on channel weights.

The algorithm normalizes request counts by weight, so higher weight channels need proportionally more requests to get the same penalty.

Formula: 

normalizedCount = effectiveCount / (weight / 100.0)
score = maxScore * exp(-normalizedCount / scalingFactor)

This means:

  • weight=80, 80 requests → normalized=100 → score ~77
  • weight=20, 20 requests → normalized=100 → score ~77
  • weight=80, 0 requests → normalized=0 → score=150
  • weight=20, 0 requests → normalized=0 → score=150

All channels start equal, but higher weight channels can handle more requests before their score drops. This achieves proportional distribution: - weight=80 gets ~80/(80+50+20+10) = 50% of requests - weight=50 gets ~50/(80+50+20+10) = 31% of requests - weight=20 gets ~20/(80+50+20+10) = 12.5% of requests - weight=10 gets ~10/(80+50+20+10) = 6.25% of requests

Score range: 10-150.

func NewWeightRoundRobinStrategy 

func NewWeightRoundRobinStrategy(metricsProvider ChannelMetricsProvider) *WeightRoundRobinStrategy

NewWeightRoundRobinStrategy creates a new weighted round-robin strategy.

func (*WeightRoundRobinStrategy) Name 

func (s *WeightRoundRobinStrategy) Name() string

Name returns the strategy name.

func (*WeightRoundRobinStrategy) Score 

func (s *WeightRoundRobinStrategy) Score(ctx context.Context, channel *biz.Channel) float64

Score returns a weighted round-robin score. Production path without debug logging.

func (*WeightRoundRobinStrategy) ScoreWithDebug 

func (s *WeightRoundRobinStrategy) ScoreWithDebug(ctx context.Context, channel *biz.Channel) (float64, StrategyScore)

ScoreWithDebug returns a weighted round-robin score with detailed debug information. Debug path with comprehensive logging.

type WeightStrategy 

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

WeightStrategy prioritizes channels based on their ordering weight. Higher weight = higher priority.

func NewWeightStrategy 

func NewWeightStrategy() *WeightStrategy

NewWeightStrategy creates a new weight-based strategy.

func (*WeightStrategy) Name 

func (s *WeightStrategy) Name() string

Name returns the strategy name.

func (*WeightStrategy) Score 

func (s *WeightStrategy) Score(ctx context.Context, channel *biz.Channel) float64

Score returns a score based on the channel's ordering weight. Score is normalized to 0-maxScore range. Production path without debug logging.

func (*WeightStrategy) ScoreWithDebug 

func (s *WeightStrategy) ScoreWithDebug(ctx context.Context, channel *biz.Channel) (float64, StrategyScore)

ScoreWithDebug returns a score with detailed debug information. Debug path with comprehensive logging.

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.