Make multimodal cache weights use optional token data

Use TokenizedPrompt as an optional data dependency so multimodal cache affinity can use tokenized placeholder lengths when token-producer is configured, while preserving the lightweight unit-weight fallback when it is absent.

Signed-off-by: Guy Girmonsky <guygir@gmail.com>

Author: guygir

deploy/config/epp-weighted-mm-embeddings-cache-config.yaml

@@ -0,0 +1,51 @@
+# Sample EPP configuration extending the default scoring baseline with
+# tokenized multimodal embeddings cache affinity.
+apiVersion: llm-d.ai/v1alpha1
+kind: EndpointPickerConfig
+plugins:
+  - type: token-producer
+    parameters:
+      modelName: hf-repo/model-name           # set to the model used in the vLLM deployment
+      vllm:
+        url: http://localhost:8000
+  - type: endpoint-notification-source
+  - type: metrics-data-source
+  - type: core-metrics-extractor
+  - type: mm-embeddings-cache-producer
+    parameters:
+      cacheSizeInMBPerServer: 2048
+  - type: mm-embeddings-cache-scorer
+  - type: precise-prefix-cache-scorer
+    parameters:
+      tokenProcessorConfig:
+        blockSize: 64
+      indexerConfig:
+        kvBlockIndexConfig:
+          enableMetrics: true
+  - type: queue-scorer
+  - type: kv-cache-utilization-scorer
+  - type: max-score-picker
+  - type: single-profile-handler
+  - type: decode-filter
+dataLayer:
+  sources:
+    - pluginRef: metrics-data-source
+      extractors:
+        - pluginRef: core-metrics-extractor
+    - pluginRef: endpoint-notification-source
+      extractors:
+        - pluginRef: mm-embeddings-cache-producer
+        - pluginRef: precise-prefix-cache-scorer
+schedulingProfiles:
+  - name: default
+    plugins:
+      - pluginRef: decode-filter
+      - pluginRef: queue-scorer
+        weight: 2
+      - pluginRef: kv-cache-utilization-scorer
+        weight: 2
+      - pluginRef: precise-prefix-cache-scorer
+        weight: 3
+      - pluginRef: mm-embeddings-cache-scorer
+        weight: 1
+      - pluginRef: max-score-picker

pkg/epp/datalayer/data_graph.go

@@ -222,21 +222,33 @@ func buildDAG(producers map[string]plugin.ProducerPlugin, consumers map[string]p
 				continue
 			}
 			dependencies := consumer.Consumes()
-			if producer.Produces() != nil && dependencies.Required != nil {
-				for producedKey, producedData := range producer.Produces() {
-					if consumedData, ok := dependencies.Required[producedKey]; ok {
-						// Check types are same.
-						if reflect.TypeOf(producedData) != reflect.TypeOf(consumedData) {
-							return nil, errors.New("data type mismatch between produced and consumed data for key: " + producedKey.String())
-						}
-						if pluginToLayerExecutionOrder(producer) > pluginToLayerExecutionOrder(consumer) {
-							return nil, errors.New("invalid plugin layer execution order: producer " + pName + " needs to be executed before consumer " + cName)
-						}
-						// Consumer depends on producer, so add an edge from consumer to producer.
-						dag[cName] = append(dag[cName], pName)
-						break
+			if producer.Produces() == nil {
+				continue
+			}
+			for producedKey, producedData := range producer.Produces() {
+				consumedData, ok := dependencies.Required[producedKey]
+				optional := false
+				if !ok {
+					consumedData, ok = dependencies.Optional[producedKey]
+					optional = ok
+				}
+				if !ok {
+					continue
+				}
+				// Check types are same.
+				if reflect.TypeOf(producedData) != reflect.TypeOf(consumedData) {
+					if optional {
+						return nil, errors.New("data type mismatch between produced and optionally consumed data for key: " + producedKey.String())
 					}
+					return nil, errors.New("data type mismatch between produced and consumed data for key: " + producedKey.String())
+				}
+				if pluginToLayerExecutionOrder(producer) > pluginToLayerExecutionOrder(consumer) {
+					return nil, errors.New("invalid plugin layer execution order: producer " + pName + " needs to be executed before consumer " + cName)
 				}
+				// Consumer depends on producer; optional dependencies only add
+				// an edge when the producer is already configured.
+				dag[cName] = append(dag[cName], pName)
+				break
 			}
 		}
 	}

pkg/epp/datalayer/data_graph_test.go

@@ -37,9 +37,10 @@ import (
 const mockProducedDataKey = "mockProducedData"
 
 type mockDataProducerP struct {
-	name     string
-	produces map[fwkplugin.DataKey]any
-	consumes map[fwkplugin.DataKey]any
+	name             string
+	produces         map[fwkplugin.DataKey]any
+	consumes         map[fwkplugin.DataKey]any
+	optionalConsumes map[fwkplugin.DataKey]any
 }
 
 type mockProducedDataType struct {
@@ -59,7 +60,7 @@ func (m *mockDataProducerP) Produces() map[fwkplugin.DataKey]any {
 }
 
 func (m *mockDataProducerP) Consumes() fwkplugin.DataDependencies {
-	return fwkplugin.DataDependencies{Required: m.consumes}
+	return fwkplugin.DataDependencies{Required: m.consumes, Optional: m.optionalConsumes}
 }
 
 func (m *mockDataProducerP) Produce(ctx context.Context, request *fwksched.InferenceRequest, endpoints []fwksched.Endpoint) error {
@@ -165,6 +166,7 @@ func TestDAGAndTopologicalOrder(t *testing.T) {
 	pluginC := &mockDataProducerP{name: "C", consumes: map[fwkplugin.DataKey]any{dkB: nil}}
 	pluginD := &mockDataProducerP{name: "D", consumes: map[fwkplugin.DataKey]any{dkA: nil}}
 	pluginE := &mockDataProducerP{name: "E"} // No dependencies
+	pluginF := &mockDataProducerP{name: "F", optionalConsumes: map[fwkplugin.DataKey]any{dkA: nil}}
 
 	// Cycle plugins
 	pluginX := &mockDataProducerP{name: "X", produces: map[fwkplugin.DataKey]any{dkX: nil}, consumes: map[fwkplugin.DataKey]any{dkY: nil}}
@@ -220,6 +222,15 @@ func TestDAGAndTopologicalOrder(t *testing.T) {
 			},
 			expectedErr: "",
 		},
+		{
+			name:    "Optional dependency orders when producer is configured",
+			plugins: []fwkrc.DataProducer{pluginA, pluginF},
+			expectedDAG: map[string][]string{
+				"A/mock": {},
+				"F/mock": {"A/mock"},
+			},
+			expectedErr: "",
+		},
 		{
 			name:        "Graph with a cycle (X -> Y, Y -> X)",
 			plugins:     []fwkrc.DataProducer{pluginX, pluginY},
@@ -345,6 +356,15 @@ func TestCreateMissingDataProducers(t *testing.T) {
 			factoryRegistry: map[string]fwkplugin.FactoryFunc{producerTypeA: producerAFactory},
 			wantTypes:       nil,
 		},
+		{
+			name: "optional consumes does not create missing producer",
+			existingPlugins: []fwkplugin.Plugin{
+				&mockDataProducerP{name: "optional-consumer", optionalConsumes: map[fwkplugin.DataKey]any{keyA: nil}},
+			},
+			defaultProducerRegistry: map[string]string{keyA.String(): producerTypeA},
+			factoryRegistry:         map[string]fwkplugin.FactoryFunc{producerTypeA: producerAFactory},
+			wantTypes:               nil,
+		},
 		{
 			name: "producer already present by type - not duplicated",
 			existingPlugins: []fwkplugin.Plugin{
@@ -479,7 +499,7 @@ func (m *mockMayConsumerPlugin) Consumes() fwkplugin.DataDependencies {
 	return fwkplugin.DataDependencies{Optional: m.optionalConsumes}
 }
 
-// mockMixedConsumerPlugin is a plugin that has both required Consumes and optional OptionalConsumes.
+// mockMixedConsumerPlugin is a plugin that has both required and optional data dependencies.
 // This models a real plugin like prefix cache scorer — requires prefix-match data,
 // but optionally uses tokenized input and falls back to raw text if unavailable.
 type mockMixedConsumerPlugin struct {
@@ -511,7 +531,7 @@ func TestCreateMissingDataProducers_MayConsume(t *testing.T) {
 		wantErr         bool
 	}{
 		{
-			name: "OptionalConsumes key with no producer — warning only, no error",
+			name: "optional key with no producer — warning only, no error",
 			existingPlugins: []fwkplugin.Plugin{
 				&mockMayConsumerPlugin{
 					name:             "optional-consumer",
@@ -522,7 +542,7 @@ func TestCreateMissingDataProducers_MayConsume(t *testing.T) {
 			wantErr:         false, // must NOT error
 		},
 		{
-			name: "OptionalConsumes key with a producer present — no warning, no error",
+			name: "optional key with a producer present — no warning, no error",
 			existingPlugins: []fwkplugin.Plugin{
 				&mockDataProducerP{name: "producer", produces: map[fwkplugin.DataKey]any{keyA: nil}},
 				&mockMayConsumerPlugin{
@@ -535,10 +555,10 @@ func TestCreateMissingDataProducers_MayConsume(t *testing.T) {
 		},
 		{
 			// Models the real prefix cache scorer — it requires prefix-match data (Consumes)
-			// but optionally uses tokenized input (OptionalConsumes), falling back to raw text.
+			// but optionally uses tokenized input, falling back to raw text.
 			// The required key has a producer. The optional key does not.
 			// Result: no error. Warning logged for the missing optional key.
-			name: "plugin with both Consumes and OptionalConsumes — required key has producer, optional does not",
+			name: "plugin with both required and optional dependencies — required key has producer, optional does not",
 			existingPlugins: []fwkplugin.Plugin{
 				&mockDataProducerP{name: "required-producer", produces: map[fwkplugin.DataKey]any{keyA: nil}},
 				&mockMixedConsumerPlugin{

pkg/epp/framework/plugins/requestcontrol/dataproducer/multimodal/README.md

@@ -8,8 +8,10 @@ Produces multimodal embeddings cache match data for downstream scheduling plugin
 
 For each request, the producer extracts stable multimodal item hashes from:
 
-- `TokenizedPrompt.MultiModalFeatures`, when a `token-producer` is configured
+- `TokenizedPrompt.MultiModalFeatures`, when tokenized multimodal metadata is
+  available on the request
 - typed OpenAI chat-completions structured media blocks, as a lightweight fallback
+- `Generate.Features.MMHashes`, when present
 
 It keeps an in-memory LRU map from multimodal hash to the set of pods that recently
 handled that item. During scheduling, it attaches `EncoderCacheMatchInfo` to each
@@ -18,15 +20,24 @@ same image, video, or audio input.
 
 Repeated references to the same multimodal hash within one request count once.
 
-## Inputs Consumed
+## Item Weights
+
+Each matched multimodal item contributes to encoder-cache affinity. The scorer
+computes `matchedWeight / totalWeight`; this producer defines the per-item
+weight in that ratio.
 
-This plugin declares:
+- When tokenized multimodal metadata is available, each item weight is
+  `MultiModalFeature.Length` (falling back to `1` when length is zero).
+- Without tokenized multimodal metadata, each unique multimodal hash has item
+  weight `1`.
 
-- `TokenizedPrompt`
+## Inputs Consumed
 
-When `token-producer` is present, this orders tokenization before multimodal match
-data production. If tokenized prompt data is absent at runtime, the producer falls
-back to typed structured chat-completions media blocks.
+This producer declares `TokenizedPrompt` from `token-producer` as an optional
+dependency. If `token-producer` is configured, this producer runs after it and
+uses `TokenizedPrompt.MultiModalFeatures` placeholder lengths. If tokenized
+prompt data is absent at runtime, the producer falls back to typed structured
+chat-completions media blocks with item weight `1` per hash.
 
 ## Data Produced
 
@@ -41,7 +52,7 @@ The producer supports the following runtime parameters:
 - `cacheSizeInMBPerServer` (integer, default: `2048`, 2 GiB): per-endpoint memory budget in
   mebibytes (MiB) for the best-effort pod-affinity LRU.
 
-**Configuration Examples:**
+**Lightweight configuration example (tokenizer-free chat media):**
 
 ```yaml
 plugins:
@@ -60,13 +71,15 @@ schedulingProfiles:
         weight: 2
 ```
 
+**Tokenized multimodal weight example:**
+
 ```yaml
 plugins:
   - type: token-producer
     parameters:
       modelName: Qwen/Qwen2.5-1.5B-Instruct
       vllm:
-        http: http://localhost:8000
+        url: http://localhost:8000
   - type: mm-embeddings-cache-producer
     parameters:
       cacheSizeInMBPerServer: 2048
@@ -81,6 +94,9 @@ schedulingProfiles:
 ## Operational Notes
 
 - The cache is a best-effort routing signal, not a correctness dependency.
-- The producer remains tokenizer-free for request shapes where typed media blocks are
-  sufficient; `token-producer` is only required when relying on upstream multimodal
-  metadata.
+- `cacheSizeInMBPerServer` bounds the EPP routing LRU only; it is not a model-server
+  encoder cache capacity knob.
+- Structured chat media blocks are enough for lightweight cache affinity;
+  `token-producer` is not required.
+- Configure `token-producer` when multimodal placeholder lengths should
+  influence affinity scores.