跳到导航
Knowledge BaseModular ComponentsRouterDevelop the Router

Router Examples

Worked examples for the KvRouter Python API, Kubernetes deployments, and custom routing patterns.
以 Markdown 格式查看

For quick start instructions, see the Router README. This document provides further examples for using the Dynamo Router, including Python API usage, Kubernetes deployments, and custom routing patterns.

Using KvRouter Python API

Instead of launching the KV Router via command line, you can create a KvRouter object directly in Python. Your application can then query worker selection, inspect worker loads, and route or track requests itself.

Multiple Routers from the Same Runtime: Do not create multiple independently managed KvRouter instances from the same DistributedRuntime. Routers created from endpoints owned by the same runtime share that runtime’s primary cancellation token, so dropping one router can cancel background work used by the others. For one in-process frontend, use a single KvRouter; for independent router lifetimes, use separate frontend processes or create each router from a separate DistributedRuntime.

With the event loop available as loop, independent in-process router lifetimes require separate runtimes:

router_a = KvRouter(DistributedRuntime(loop, "etcd", "tcp").endpoint("dynamo.backend.generate"), 16, KvRouterConfig())
router_b = KvRouter(DistributedRuntime(loop, "etcd", "tcp").endpoint("dynamo.backend.generate"), 16, KvRouterConfig())

Methods

The KvRouter provides the following methods:

  • generate(token_ids, model, ...): Route and execute a request, returning an async stream of responses. Automatically handles worker selection, state tracking, and lifecycle management.

  • best_worker(token_ids, router_config_override=None, request_id=None, update_indexer=False): Query which worker would be selected for given tokens. Returns (worker_id, dp_rank, overlap_blocks).

    • Without request_id: Query-only, doesn’t update router state
    • With request_id: Updates router lifecycle state to track the request. Note: If used with request_id, you must call mark_prefill_complete() and free() at the appropriate lifecycle points to maintain accurate load tracking
    • With update_indexer=True: Records the selected worker in the approximate indexer for future overlap predictions. This is only meaningful when use_kv_events=False

  • get_potential_loads(token_ids): Get detailed load information for all workers, including potential prefill tokens, potential decode blocks, and active requests. Returns a list of load dictionaries.

  • get_overlap_scores(token_ids, ...): Get per-worker KV overlap by storage tier, including shared-cache overlap when configured.

  • mark_prefill_complete(request_id): Signal that a request has completed its prefill phase. Only used for manual lifecycle management when using best_worker() for manual routing instead of generate().

  • free(request_id): Signal that a request has completed and its resources should be released. Only used for manual lifecycle management when using best_worker() for manual routing instead of generate().

  • dump_events(): Dump all KV cache events from the router’s indexer as a JSON string. Useful for debugging and analysis.

Setup

First, launch at least two backend engines so the router has a meaningful selection to make. Each worker needs a unique system port and KV event endpoint:

export PYTHONHASHSEED=0
DYN_SYSTEM_PORT=8081 CUDA_VISIBLE_DEVICES=0 python -m dynamo.vllm \
  --model Qwen/Qwen3-0.6B \
  --block-size 64 \
  --kv-events-config '{"publisher":"zmq","topic":"kv-events","endpoint":"tcp://*:20080","enable_kv_cache_events":true}' &
DYN_SYSTEM_PORT=8082 CUDA_VISIBLE_DEVICES=1 python -m dynamo.vllm \
  --model Qwen/Qwen3-0.6B \
  --block-size 64 \
  --kv-events-config '{"publisher":"zmq","topic":"kv-events","endpoint":"tcp://*:20081","enable_kv_cache_events":true}' &

Example Script

import asyncio
from dynamo.runtime import DistributedRuntime
from dynamo.llm import KvRouter, KvRouterConfig
async def main():
    # Get runtime and create endpoint
    loop = asyncio.get_running_loop()
    runtime = DistributedRuntime(loop, "etcd", "tcp")
    endpoint = runtime.endpoint("dynamo.backend.generate")
    # Create KV router
    kv_router_config = KvRouterConfig()
    router = KvRouter(
        endpoint=endpoint,
        block_size=64,
        kv_router_config=kv_router_config
    )
    # Optional startup gate shared with the frontend and standalone indexer:
    # os.environ["DYN_ROUTER_MIN_INITIAL_WORKERS"] = "2"
    # Your input tokens
    token_ids = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
    # Route the request and stream the response
    stream = await router.generate(
        token_ids=token_ids,
        model="Qwen/Qwen3-0.6B",
        stop_conditions={
            "max_tokens": 20,        # Generate exactly 20 tokens
            "ignore_eos": True,      # Don't stop at EOS token
        },
        sampling_options={
            "temperature": 0.7,
            "top_p": 0.9,
        },
    )
    # Collect generated tokens
    generated_tokens = []
    async for response in stream:
        if isinstance(response, dict) and "token_ids" in response:
            generated_tokens.extend(response["token_ids"])
    print(f"Generated {len(generated_tokens)} tokens: {generated_tokens}")
if __name__ == "__main__":
    asyncio.run(main())

Per-Request Overrides

generate() and best_worker() accept a router_config_override dictionary. With the default worker-selection policy, only its load-tracking fields take effect:

FieldEffect
track_prefill_tokensWhether the request’s prompt counts as prefill load, both when scoring workers and in the selected worker’s tracked load. Set it to False when another worker already ran the prefill.
assume_kv_reuseWhether active-block tracking counts the request’s prefix blocks once when the same blocks are already active on the worker. Set it to False when the worker does not deduplicate those blocks, such as KV blocks transferred from a prefill worker.

For example, to select a decode worker for a request whose prompt was prefilled elsewhere:

worker_id, dp_rank, overlap = await router.best_worker(
    token_ids,
    request_id="req-456",
    router_config_override={
        "track_prefill_tokens": False,  # Don't charge prompt work to this worker
        "assume_kv_reuse": False,       # Count transferred blocks as new on this worker
    },
)

Because this call passes request_id, manage the request lifecycle as described in Manual State Management.

The default policy ignores the scoring fields overlap_score_credit, prefill_load_scale, shared_cache_multiplier, and router_temperature in router_config_override, so they do not change which worker it selects. Set scoring weights and temperature as default-policy parameters in the YAML file passed to --router-policy-config, or to KvRouterConfig(router_policy_config=...) from Python; see Configure the Default Policy. A custom worker-selection policy can read a per-request temperature through router_temperature_override().

K8s Examples

For basic Kubernetes deployment with the KV Router, see Dynamo Frontend Routing.

Complete K8s Examples

For A/B Testing and Advanced K8s Setup: See the comprehensive KV Router A/B Benchmarking Guide for step-by-step instructions on deploying, configuring, and benchmarking the KV router in Kubernetes.

Example with Advanced Configuration

apiVersion: nvidia.com/v1alpha1
kind: DynamoGraphDeployment
metadata:
  name: my-deployment
spec:
  services:
    Frontend:
      componentType: frontend
      replicas: 1
      envs:
        - name: DYN_ROUTER_MODE
          value: kv
        - name: DYN_ROUTER_TEMPERATURE
          value: "0.5"  # Add some randomness to prevent worker saturation
        - name: DYN_ROUTER_KV_OVERLAP_SCORE_CREDIT
          value: "1.0"  # Prefer device-local KV cache reuse
        - name: DYN_ROUTER_PREFILL_LOAD_SCALE
          value: "1.5"  # Prioritize TTFT over ITL
        - name: DYN_KV_CACHE_BLOCK_SIZE
          value: "16"
      extraPodSpec:
        mainContainer:
          image: nvcr.io/nvidia/ai-dynamo/vllm-runtime:1.2.1

Alternative: Using Command Args in K8s

You can also pass CLI arguments directly in the container command:

extraPodSpec:
  mainContainer:
    image: nvcr.io/nvidia/ai-dynamo/vllm-runtime:1.2.1
    command:
      - /bin/sh
      - -c
    args:
      - "python3 -m dynamo.frontend --router-mode kv --router-temperature 0.5 --http-port 8000"

Recommendation: Use environment variables for easier configuration management and consistency with Dynamo’s K8s patterns.

Routing Patterns

The KvRouter supports multiple usage patterns depending on your control requirements:

Call generate() directly and let the router handle everything:

stream = await router.generate(token_ids=tokens, model="model-name")
  • Best for: Most use cases
  • Router automatically: Selects best worker, updates state, routes request, tracks lifecycle

2. Manual State Management (Advanced)

Use best_worker(request_id=...) to select and track, then manage the request yourself:

worker_id, dp_rank, overlap = await router.best_worker(
    tokens,
    request_id="req-123",
)
client = await endpoint.client()
request = {
    "model": "Qwen/Qwen3-0.6B",
    "token_ids": tokens,
    "stop_conditions": {"max_tokens": 20},
    "sampling_options": {},
    "routing": {"dp_rank": dp_rank},
}
stream = await client.direct(request, worker_id)
try:
    first = await anext(stream)
    if first.is_error():
        raise RuntimeError(f"Worker returned an error: {first.comments()}")
    print(first.data())
    await router.mark_prefill_complete("req-123")
    async for response in stream:
        if response.is_error():
            raise RuntimeError(f"Worker returned an error: {response.comments()}")
        print(response.data())
finally:
    await router.free("req-123")
  • Best for: Custom request handling with router state tracking
  • Requires: Calling mark_prefill_complete() and free() at correct lifecycle points
  • Approximate mode: Pass update_indexer=True to best_worker() when use_kv_events=False so the router learns from manual worker selections
  • Direct dispatch: Client.direct() targets the returned worker instance; include the returned DP rank in the preprocessed request’s routing field
  • Caution: Incorrect lifecycle management degrades load balancing accuracy

3. Hierarchical Router Probing

Query without state updates, then route through a chosen router:

# Probe multiple routers without updating state
worker_id_1, dp_rank, overlap_1 = await router_1.best_worker(tokens)  # No request_id
worker_id_2, dp_rank, overlap_2 = await router_2.best_worker(tokens)
# Pick the best router and corresponding worker based on results
if overlap_1 > overlap_2:
    chosen_router, chosen_worker = router_1, worker_id_1
else:
    chosen_router, chosen_worker = router_2, worker_id_2
stream = await chosen_router.generate(tokens, model="model-name", worker_id=chosen_worker)
  • Best for: Multi-tier deployments (e.g., Envoy Gateway routing to multiple router groups)
  • Advantage: Query multiple routers before committing to one

4. Custom Load-Based Routing

Use get_potential_loads() to implement custom routing logic:

loads = await router.get_potential_loads(tokens)
# Apply custom logic (e.g., weighted scoring, constraints)
best_worker = min(loads, key=lambda x: custom_cost_fn(x))
stream = await router.generate(tokens, model="model-name", worker_id=best_worker['worker_id'])
  • Best for: Custom optimization strategies beyond the built-in cost function
  • Advantage: Full control over worker selection logic
  • See also: Detailed example below in “Custom Routing Example: Minimizing TTFT”

In every pattern, generate() and best_worker() accept router_config_override to adjust a request’s load tracking without recreating the router. Under the default policy, the override does not change scoring weights or temperature; see Per-Request Overrides.

Custom Routing Example: Minimizing TTFT

Here’s an example of using get_potential_loads() to implement custom routing that minimizes Time To First Token (TTFT) by selecting the worker with the least prefill work:

import asyncio
from dynamo.runtime import DistributedRuntime
from dynamo.llm import KvRouter, KvRouterConfig
async def minimize_ttft_routing():
    # Setup router
    loop = asyncio.get_running_loop()
    runtime = DistributedRuntime(loop, "etcd", "tcp")
    endpoint = runtime.endpoint("dynamo.backend.generate")
    router = KvRouter(
        endpoint=endpoint,
        block_size=64,
        kv_router_config=KvRouterConfig()
    )
    # Your input tokens
    token_ids = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
    # Get potential loads for all workers
    potential_loads = await router.get_potential_loads(token_ids)
    # Find worker with minimum prefill tokens (best for TTFT)
    best_worker = min(potential_loads, key=lambda x: x['potential_prefill_tokens'])
    print(f"Worker loads: {potential_loads}")
    print(f"Selected worker {best_worker['worker_id']} with {best_worker['potential_prefill_tokens']} prefill tokens")
    # Route directly to the selected worker
    stream = await router.generate(
        token_ids=token_ids,
        model="Qwen/Qwen3-0.6B",
        worker_id=best_worker['worker_id'],  # Force routing to optimal worker
        stop_conditions={"max_tokens": 20}
    )
    # Process response
    async for response in stream:
        if isinstance(response, dict) and "token_ids" in response:
            print(f"Generated tokens: {response['token_ids']}")
if __name__ == "__main__":
    asyncio.run(minimize_ttft_routing())

This approach gives you complete control over routing decisions, allowing you to optimize for different metrics based on your specific requirements. As some examples:

  • Minimize TTFT: Select worker with lowest potential_prefill_tokens
  • Maximize cache reuse: Use best_worker() which considers both prefill and decode loads
  • Balance load: Consider both potential_prefill_tokens and potential_decode_blocks together
  • Balance active batches: Include active_requests when decode step latency depends materially on request count

See Router Design for architecture details and the cost function algorithm.

KV Event Publishing for Custom Engines

For full documentation on implementing KV event publishing for custom inference engines, see the dedicated KV Event Publishing for Custom Engines guide. It covers:

  • Direct publishing: Call publish_stored() / publish_removed() to push events over the Dynamo event plane
  • ZMQ relay: For engines that emit raw KV events over ZMQ (like SGLang and vLLM), the same KvEventPublisher subscribes to the ZMQ socket and relays events automatically
  • API reference, event structure, ZMQ wire format, and best practices

Advertising a separate KV-state endpoint

By default, consumers assume a worker’s KV state is described by the same endpoint it serves requests on. Pass kv_state_endpoint to WorkerConfig when KV-state ownership lives somewhere other than the serving endpoint, so the two can be discovered independently:

config = WorkerConfig(
    namespace="dynamo",
    component="backend",
    endpoint="generate",
    kv_state_endpoint="dynamo.kvstate.events",
    model_name=model_name,
)

Leave it unset for the common case. Existing deployments stay wire-compatible, since an unset value resolves to the serving endpoint.

Global Router (Hierarchical Routing)

For deployments with multiple worker pools, the Global Router enables hierarchical routing by sitting between the frontend and local routers. It selects the appropriate pool for each request based on configurable policies, supporting disaggregated topologies where pools are tuned for different workload characteristics.

See Also