feat: implement decision-based routing with plugin architecture

Signed-off-by: bitliu <[email protected]>

Author: Xunzhuo

deploy/kubernetes/crds/examples/intelligentroute-example.yaml

@@ -80,6 +80,18 @@ spec:
           configuration:
             enabled: true
             threshold: 0.5
+        - type: "header_mutation"
+          configuration:
+            add:
+              - name: "X-Decision-Name"
+                value: "urgent_technical"
+              - name: "X-Priority"
+                value: "high"
+            update:
+              - name: "User-Agent"
+                value: "semantic-router/urgent"
+            delete:
+              - "X-Debug-Info"
 
     - name: "math_problems"
       priority: 80

deploy/kubernetes/crds/intelligentroute-crd.yaml

@@ -220,6 +220,7 @@ spec:
                                 - jailbreak
                                 - pii
                                 - system_prompt
+                                - header_mutation
                               description: Plugin type
                             configuration:
                               type: object

src/semantic-router/examples/decision-based-routing.yaml

@@ -73,6 +73,16 @@ decisions:
         use_reasoning: true
         reasoning_effort: "high"
     plugins:
+      - type: "header_mutation"
+        configuration:
+          add:
+            - name: "X-Decision-Name"
+              value: "complex-coding"
+            - name: "X-Reasoning-Enabled"
+              value: "true"
+          update:
+            - name: "User-Agent"
+              value: "semantic-router/coding"
       - type: "semantic-cache"
         configuration:
           enabled: true

src/semantic-router/pkg/apis/vllm.ai/v1alpha1/types_route.go

@@ -244,9 +244,9 @@ type ModelRef struct {
 
 // DecisionPlugin defines a plugin configuration for a decision
 type DecisionPlugin struct {
-	// Type is the plugin type (semantic-cache, jailbreak, pii, system_prompt)
+	// Type is the plugin type (semantic-cache, jailbreak, pii, system_prompt, header_mutation)
 	// +kubebuilder:validation:Required
-	// +kubebuilder:validation:Enum=semantic-cache;jailbreak;pii;system_prompt
+	// +kubebuilder:validation:Enum=semantic-cache;jailbreak;pii;system_prompt;header_mutation
 	Type string `json:"type" yaml:"type"`
 
 	// Configuration is the plugin-specific configuration as a raw JSON object

src/semantic-router/pkg/config/config.go

@@ -546,6 +546,19 @@ type SystemPromptPluginConfig struct {
 	Mode         string `json:"mode,omitempty" yaml:"mode,omitempty"` // "replace" or "insert"
 }
 
+// HeaderMutationPluginConfig represents configuration for header_mutation plugin
+type HeaderMutationPluginConfig struct {
+	Add    []HeaderPair `json:"add,omitempty" yaml:"add,omitempty"`
+	Update []HeaderPair `json:"update,omitempty" yaml:"update,omitempty"`
+	Delete []string     `json:"delete,omitempty" yaml:"delete,omitempty"`
+}
+
+// HeaderPair represents a header name-value pair
+type HeaderPair struct {
+	Name  string `json:"name" yaml:"name"`
+	Value string `json:"value" yaml:"value"`
+}
+
 // Helper methods for Decision to access plugin configurations
 
 // GetPluginConfig returns the configuration for a specific plugin type
@@ -633,6 +646,49 @@ func (d *Decision) GetSystemPromptConfig() *SystemPromptPluginConfig {
 	return result
 }
 
+// GetHeaderMutationConfig returns the header_mutation plugin configuration
+func (d *Decision) GetHeaderMutationConfig() *HeaderMutationPluginConfig {
+	config := d.GetPluginConfig("header_mutation")
+	if config == nil {
+		return nil
+	}
+
+	result := &HeaderMutationPluginConfig{}
+
+	// Parse add headers
+	if add, ok := config["add"].([]interface{}); ok {
+		for _, item := range add {
+			if headerMap, ok := item.(map[string]interface{}); ok {
+				name, _ := headerMap["name"].(string)
+				value, _ := headerMap["value"].(string)
+				result.Add = append(result.Add, HeaderPair{Name: name, Value: value})
+			}
+		}
+	}
+
+	// Parse update headers
+	if update, ok := config["update"].([]interface{}); ok {
+		for _, item := range update {
+			if headerMap, ok := item.(map[string]interface{}); ok {
+				name, _ := headerMap["name"].(string)
+				value, _ := headerMap["value"].(string)
+				result.Update = append(result.Update, HeaderPair{Name: name, Value: value})
+			}
+		}
+	}
+
+	// Parse delete headers
+	if del, ok := config["delete"].([]interface{}); ok {
+		for _, item := range del {
+			if headerName, ok := item.(string); ok {
+				result.Delete = append(result.Delete, headerName)
+			}
+		}
+	}
+
+	return result
+}
+
 // RuleCombination defines how to combine multiple rule conditions with AND/OR operators
 type RuleCombination struct {
 	// Operator specifies how to combine conditions: "AND" or "OR"

src/semantic-router/pkg/extproc/processor_req_body.go

@@ -64,8 +64,9 @@ func (r *OpenAIRouter) handleRequestBody(v *ext_proc.ProcessingRequest_RequestBo
 	// Get content from messages
 	userContent, nonUserMessages := extractUserAndNonUserContent(openAIRequest)
 
-	// Perform classification and model selection once at the beginning
-	categoryName, classificationConfidence, reasoningDecision, selectedModel := r.performClassificationAndModelSelection(originalModel, userContent, nonUserMessages)
+	// Perform decision evaluation and model selection once at the beginning
+	// Use decision-based routing if decisions are configured, otherwise fall back to category-based
+	categoryName, classificationConfidence, reasoningDecision, selectedModel := r.performDecisionEvaluationAndModelSelection(originalModel, userContent, nonUserMessages, ctx)
 
 	// Perform security checks with category-specific settings
 	if response, shouldReturn := r.performSecurityChecks(ctx, userContent, nonUserMessages, categoryName); shouldReturn {
@@ -159,7 +160,7 @@ func (r *OpenAIRouter) handleAutoModelRouting(openAIRequest *openai.ChatCompleti
 	}
 
 	// Create response with mutations
-	response = r.createRoutingResponse(matchedModel, selectedEndpoint, modifiedBody)
+	response = r.createRoutingResponse(matchedModel, selectedEndpoint, modifiedBody, ctx)
 
 	// Log routing decision
 	r.logRoutingDecision(ctx, "auto_routing", originalModel, matchedModel, categoryName, reasoningDecision.UseReasoning, selectedEndpoint)
@@ -274,14 +275,17 @@ func (r *OpenAIRouter) modifyRequestBodyForAutoRouting(openAIRequest *openai.Cha
 }
 
 // createRoutingResponse creates a routing response with mutations
-func (r *OpenAIRouter) createRoutingResponse(model string, endpoint string, modifiedBody []byte) *ext_proc.ProcessingResponse {
+func (r *OpenAIRouter) createRoutingResponse(model string, endpoint string, modifiedBody []byte, ctx *RequestContext) *ext_proc.ProcessingResponse {
 	bodyMutation := &ext_proc.BodyMutation{
 		Mutation: &ext_proc.BodyMutation_Body{
 			Body: modifiedBody,
 		},
 	}
 
 	setHeaders := []*core.HeaderValueOption{}
+	removeHeaders := []string{"content-length"}
+
+	// Add standard routing headers
 	if endpoint != "" {
 		setHeaders = append(setHeaders, &core.HeaderValueOption{
 			Header: &core.HeaderValue{
@@ -299,8 +303,21 @@ func (r *OpenAIRouter) createRoutingResponse(model string, endpoint string, modi
 		})
 	}
 
+	// Apply header mutations from decision's header_mutation plugin
+	if ctx.VSRSelectedDecision != nil {
+		pluginSetHeaders, pluginRemoveHeaders := r.buildHeaderMutations(ctx.VSRSelectedDecision)
+		if len(pluginSetHeaders) > 0 {
+			setHeaders = append(setHeaders, pluginSetHeaders...)
+			logging.Infof("Applied %d header mutations from decision %s", len(pluginSetHeaders), ctx.VSRSelectedDecision.Name)
+		}
+		if len(pluginRemoveHeaders) > 0 {
+			removeHeaders = append(removeHeaders, pluginRemoveHeaders...)
+			logging.Infof("Applied %d header deletions from decision %s", len(pluginRemoveHeaders), ctx.VSRSelectedDecision.Name)
+		}
+	}
+
 	headerMutation := &ext_proc.HeaderMutation{
-		RemoveHeaders: []string{"content-length"},
+		RemoveHeaders: removeHeaders,
 		SetHeaders:    setHeaders,
 	}
 

src/semantic-router/pkg/extproc/processor_req_header.go

@@ -9,6 +9,7 @@ import (
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/trace"
 
+	"github.com/vllm-project/semantic-router/src/semantic-router/pkg/config"
 	"github.com/vllm-project/semantic-router/src/semantic-router/pkg/headers"
 	"github.com/vllm-project/semantic-router/src/semantic-router/pkg/observability/logging"
 	"github.com/vllm-project/semantic-router/src/semantic-router/pkg/observability/tracing"
@@ -33,11 +34,12 @@ type RequestContext struct {
 	TTFTSeconds  float64
 
 	// VSR decision tracking
-	VSRSelectedCategory     string // The category selected by VSR
-	VSRReasoningMode        string // "on" or "off" - whether reasoning mode was determined to be used
-	VSRSelectedModel        string // The model selected by VSR
-	VSRCacheHit             bool   // Whether this request hit the cache
-	VSRInjectedSystemPrompt bool   // Whether a system prompt was injected into the request
+	VSRSelectedCategory     string           // The category selected by VSR
+	VSRReasoningMode        string           // "on" or "off" - whether reasoning mode was determined to be used
+	VSRSelectedModel        string           // The model selected by VSR
+	VSRCacheHit             bool             // Whether this request hit the cache
+	VSRInjectedSystemPrompt bool             // Whether a system prompt was injected into the request
+	VSRSelectedDecision     *config.Decision // The decision selected by DecisionEngine
 
 	// Tracing context
 	TraceContext context.Context // OpenTelemetry trace context for span propagation

src/semantic-router/pkg/extproc/req_filter_classification.go

@@ -67,7 +67,7 @@ func (r *OpenAIRouter) performClassificationAndModelSelection(originalModel stri
 // performDecisionEvaluationAndModelSelection performs decision evaluation using DecisionEngine
 // Returns (decisionName, confidence, reasoningDecision, selectedModel)
 // This is the new approach that uses Decision-based routing with AND/OR rule combinations
-func (r *OpenAIRouter) performDecisionEvaluationAndModelSelection(originalModel string, userContent string, nonUserMessages []string) (string, float64, entropy.ReasoningDecision, string) {
+func (r *OpenAIRouter) performDecisionEvaluationAndModelSelection(originalModel string, userContent string, nonUserMessages []string, ctx *RequestContext) (string, float64, entropy.ReasoningDecision, string) {
 	var decisionName string
 	var evaluationConfidence float64
 	var reasoningDecision entropy.ReasoningDecision
@@ -110,6 +110,9 @@ func (r *OpenAIRouter) performDecisionEvaluationAndModelSelection(originalModel
 		return "", 0.0, entropy.ReasoningDecision{}, r.Config.DefaultModel
 	}
 
+	// Store the selected decision in context for later use (e.g., header mutations)
+	ctx.VSRSelectedDecision = result.Decision
+
 	decisionName = result.Decision.Name
 	evaluationConfidence = result.Confidence
 	logging.Infof("Decision Evaluation Result: decision=%s, confidence=%.3f, matched_rules=%v",

src/semantic-router/pkg/extproc/req_filter_header_mutation.go

@@ -0,0 +1,57 @@
+package extproc
+
+import (
+	corev3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
+	"github.com/vllm-project/semantic-router/src/semantic-router/pkg/config"
+	"github.com/vllm-project/semantic-router/src/semantic-router/pkg/observability/logging"
+)
+
+// buildHeaderMutations builds header mutations based on the decision's header_mutation plugin configuration
+// Returns (setHeaders, removeHeaders) to be applied to the request
+func (r *OpenAIRouter) buildHeaderMutations(decision *config.Decision) ([]*corev3.HeaderValueOption, []string) {
+	if decision == nil {
+		return nil, nil
+	}
+
+	// Get header mutation configuration
+	headerConfig := decision.GetHeaderMutationConfig()
+	if headerConfig == nil {
+		return nil, nil
+	}
+
+	logging.Debugf("Building header mutations for decision %s: add=%d, update=%d, delete=%d",
+		decision.Name, len(headerConfig.Add), len(headerConfig.Update), len(headerConfig.Delete))
+
+	var setHeaders []*corev3.HeaderValueOption
+	var removeHeaders []string
+
+	// Apply additions (add new headers)
+	for _, headerPair := range headerConfig.Add {
+		setHeaders = append(setHeaders, &corev3.HeaderValueOption{
+			Header: &corev3.HeaderValue{
+				Key:      headerPair.Name,
+				RawValue: []byte(headerPair.Value),
+			},
+		})
+		logging.Debugf("Adding header: %s=%s", headerPair.Name, headerPair.Value)
+	}
+
+	// Apply updates (modify existing headers - in Envoy this is the same as set)
+	for _, headerPair := range headerConfig.Update {
+		setHeaders = append(setHeaders, &corev3.HeaderValueOption{
+			Header: &corev3.HeaderValue{
+				Key:      headerPair.Name,
+				RawValue: []byte(headerPair.Value),
+			},
+		})
+		logging.Debugf("Updating header: %s=%s", headerPair.Name, headerPair.Value)
+	}
+
+	// Apply deletions
+	for _, headerName := range headerConfig.Delete {
+		removeHeaders = append(removeHeaders, headerName)
+		logging.Debugf("Deleting header: %s", headerName)
+	}
+
+	return setHeaders, removeHeaders
+}