final touches for it working with dp

examples/vllm_v1/components/worker.py

@@ -25,7 +25,7 @@
 from utils.args import parse_vllm_args
 from utils.protocol import PreprocessedRequest
 from vllm.config import VllmConfig
-from vllm.distributed.kv_events import ZmqEventPublisher
+from vllm.distributed.kv_events import KVEventsConfig, ZmqEventPublisher
 from vllm.inputs import TokensPrompt
 from vllm.sampling_params import SamplingParams
 from vllm.usage.usage_lib import UsageContext
@@ -103,10 +103,15 @@ class VllmBaseWorker:
     def __init__(self):
         class_name = self.__class__.__name__
         self.engine_args = parse_vllm_args(class_name, "")
+        self.engine_args.kv_events_config = KVEventsConfig(
+            enable_kv_cache_events=True, publisher="zmq"
+        )
         if not self.engine_args.block_size:
             logger.info(f"block_size not set, default to {BLOCK_SIZE}")
             self.engine_args.block_size = BLOCK_SIZE
 
+        os.environ["VLLM_NO_USAGE_STATS"] = "1"  # Avoid internal HTTP requests
+
         model_config = self.engine_args.create_model_config()
         self.default_sampling_params = model_config.get_diff_sampling_param()
 

examples/vllm_v1/configs/agg.yaml

@@ -15,8 +15,7 @@
 Common:
   model: Qwen/Qwen3-0.6B
   data-parallel-size: 2
-  router: kv
-  block-size: 64
+  block-size: 16
   max-model-len: 16384
   served_model_name: Qwen/Qwen3-0.6B
 
@@ -29,7 +28,7 @@ VllmDecodeWorker:
   max-num-batched-tokens: 16384
   enable-prefix-caching: true
   ServiceArgs:
-    workers: 2  # 2 workers
+    workers: 1  # 2 workers
     resources:
       gpu: 2  # 2 dp ranks
   common-configs: [model, served_model_name, block-size, data-parallel-size, max-model-len]

examples/vllm_v1/utils/args.py

@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+
 # TODO: rename to avoid ambiguity with vllm package
 from vllm.engine.arg_utils import AsyncEngineArgs
 from vllm.utils import FlexibleArgumentParser
@@ -23,6 +24,7 @@
 def parse_vllm_args(service_name, prefix) -> AsyncEngineArgs:
     config = ServiceConfig.get_instance()
     vllm_args = config.as_args(service_name, prefix=prefix)
+
     parser = FlexibleArgumentParser()
     parser.add_argument(
         "--enable-disagg", action="store_true", help="Enable disaggregation"
@@ -31,4 +33,5 @@ def parse_vllm_args(service_name, prefix) -> AsyncEngineArgs:
     args = parser.parse_args(vllm_args)
     engine_args = AsyncEngineArgs.from_cli_args(args)
     engine_args.enable_disagg = args.enable_disagg
+
     return engine_args