[2026春季][T2-1-2] CearX

python/infinilm/base_config.py

@@ -112,6 +112,7 @@ def __init__(self):
         self.port = self.args.port
         self.endpoint = self.args.endpoint
         self.ignore_eos = self.args.ignore_eos
+        self.stream_interval = self.args.stream_interval
         # PD separation (KV transfer)
         self.kv_transfer_config = self.args.kv_transfer_config
 
@@ -269,6 +270,12 @@ def _add_common_args(self):
             default=8,
             help="maximum batch size for server",
         )
+        self.parser.add_argument(
+            "--stream-interval",
+            type=int,
+            default=1,
+            help="emit first token immediately, then coalesce decode steps per SSE chunk",
+        )
         self.parser.add_argument(
             "--input-len", type=parse_list, default=10, help="input sequence length"
         )

python/infinilm/config/engine_config.py

@@ -47,6 +47,7 @@ class EngineConfig:
     max_tokens: int = 4096
     num_blocks: int = 512
     block_size: int = 256
+    kv_cache_dtype: Optional[str] = None
     max_cache_len: int = 4096
     temperature: float = 1.0
     top_p: float = 0.8

python/infinilm/llm/llm.py

@@ -24,6 +24,7 @@
     FinishReason,
     InferenceRequest,
     RequestOutput,
+    RequestStatus,
     TokenOutput,
 )
 from infinilm.llm.sampling_params import SamplingParams
@@ -177,6 +178,7 @@ def _update_requests(
                 case _:
                     raise ValueError(f"Unsupported cache_type: {self.cache_type}")
         pending = []
+        step_complete_requests = []
         for req, token_ids in zip(requests, sampled_tokens):
             if req.is_aborted():
                 logger.info(
@@ -186,15 +188,52 @@ def _update_requests(
                 # (status still RUNNING).
                 if not req.is_finished():
                     req.mark_canceled()
+                step_complete_requests.append(req)
                 continue
 
+            if is_prefill and not getattr(req, "_prefill_chunk_final", True):
+                # Intermediate chunk: KV has been written, but the sampled token
+                # is not a model output because the prompt is not complete yet.
+                req.num_computed_tokens = req._prefill_chunk_end
+                req.num_local_cached_tokens = req._prefill_chunk_end
+                req.status = RequestStatus.WAITING
+                self.scheduler.waiting_queue.sync_q.put(req)
+                continue
+
+            step_complete_requests.append(req)
+
+            if is_prefill:
+                req.num_computed_tokens = req.get_prompt_length()
+                req.num_local_cached_tokens = req.get_prompt_length()
+
             if not isinstance(token_ids, list):
                 token_ids = [token_ids]
 
             for token_id in token_ids:
                 if req.is_finished():
                     break
                 req.generated_token_ids.append(token_id)
+
+                # Non-streaming benchmark/API path: avoid per-token queue traffic and
+                # incremental detokenization. With ignore_eos=true and no stop strings,
+                # the only normal finish condition is max_tokens, so final text can be
+                # decoded once when the request completes.
+                fast_non_stream = (
+                    req._output_queue is None
+                    and req.sampling_params.ignore_eos
+                    and not req.sampling_params.stop
+                )
+                if fast_non_stream:
+                    is_finished = self._check_request_finished(req, token_id)
+                    if is_finished:
+                        req.generated_text = self.tokenizer.decode(
+                            req.generated_token_ids
+                        )
+                        req._token_decode_offset = len(req.generated_token_ids)
+                        req._text_output_offset = len(req.generated_text)
+                        req.mark_finished(req.finish_reason)
+                    continue
+
                 pending_tokens = req.generated_token_ids[req._token_decode_offset :]
                 delta = self.tokenizer.decode(pending_tokens)
                 holds_back = bool(delta) and delta.endswith("\ufffd")
@@ -214,12 +253,22 @@ def _update_requests(
                         req.mark_finished(req.finish_reason)
 
                 else:
+                    stream_interval = max(
+                        1, int((req.request_data or {}).get("stream_interval", 1))
+                    )
+                    should_emit = (
+                        is_finished
+                        or req.get_num_generated_tokens() == 1
+                        or req.get_num_generated_tokens() % stream_interval == 0
+                    )
+                    if not should_emit:
+                        continue
+
                     if holds_back and not is_finished:
                         token_text = ""
                     else:
                         if is_finished and req.finish_reason in (
                             FinishReason.EOS_TOKEN,
-                            FinishReason.LENGTH,
                             FinishReason.STOP_STRING,
                         ):
                             token_text = ""
@@ -246,7 +295,7 @@ def _update_requests(
                         continue
                     pending.append((req.output_queue.async_q, output))
 
-        self.scheduler.complete_requests(requests)
+        self.scheduler.complete_requests(step_complete_requests)
         return pending
 
     def _check_request_finished(self, req: InferenceRequest, token_id: int) -> bool:
@@ -323,6 +372,7 @@ def __init__(
         max_tokens: int = 4096,
         num_blocks: int = 512,
         block_size: int = 256,
+        kv_cache_dtype: Optional[str] = None,
         max_cache_len: int = 4096,
         temperature: float = 1.0,
         top_p: float = 0.8,
@@ -369,6 +419,7 @@ def __init__(
             max_tokens=max_tokens,
             num_blocks=num_blocks,
             block_size=block_size,
+            kv_cache_dtype=kv_cache_dtype,
             max_cache_len=max_cache_len,
             temperature=temperature,
             top_p=top_p,
@@ -530,6 +581,7 @@ def __init__(
         max_tokens: int = 512,
         num_blocks: int = 512,
         block_size: int = 256,
+        kv_cache_dtype: Optional[str] = None,
         max_cache_len: int = 4096,
         temperature: float = 1.0,
         top_p: float = 0.8,
@@ -579,6 +631,7 @@ def __init__(
             max_tokens=max_tokens,
             num_blocks=num_blocks,
             block_size=block_size,
+            kv_cache_dtype=kv_cache_dtype,
             max_cache_len=max_cache_len,
             temperature=temperature,
             top_p=top_p,
@@ -721,6 +774,7 @@ def add_request(
         request_id: Optional[str] = None,
         # For server use
         request_data: Optional[dict] = None,
+        init_output_queue: bool = True,
     ) -> InferenceRequest:
         """Add a request to the engine.
 
@@ -816,8 +870,10 @@ def add_request(
             kv_params = request_data["kv_transfer_params"]
             request.kv_transfer_params = kv_params
 
-        # Initialize output queue for streaming
-        _ = request.output_queue
+        # Initialize output queue only when the caller needs token streaming.
+        # Non-streaming HTTP responses can wait for request completion and decode once.
+        if init_output_queue:
+            _ = request.output_queue
 
         self.engine.add_request(request)
         return request
@@ -829,6 +885,7 @@ def add_chat_request(
         request_id: Optional[str] = None,
         request_data: Optional[dict] = None,
         add_generation_prompt: bool = True,
+        init_output_queue: bool = True,
         **kwargs,
     ) -> InferenceRequest:
         """Add a chat request to the engine.
@@ -850,6 +907,7 @@ def add_chat_request(
             sampling_params=sampling_params,
             request_id=request_id,
             request_data=request_data,
+            init_output_queue=init_output_queue,
         )
 
     async def stream_request(

python/infinilm/llm/model_runner/model_runner.py

@@ -77,6 +77,7 @@ def __init__(self, config: EngineConfig):
             cache_config=cache_config,
             enable_graph_compiling=config.enable_graph,
             attention_backend=config.attn_backend,
+            kv_cache_dtype=config.kv_cache_dtype,
             use_mla=config.use_mla,
             weight_load_mode=config.weight_load_mode,
             skip_legacy_moe=config.skip_legacy_moe,

python/infinilm/llm/request.py

@@ -181,6 +181,8 @@ def __init__(
         self._aborted: bool = False
         self._text_output_offset: int = 0
         self._token_decode_offset: int = 0
+        self._completion_loop: Optional[asyncio.AbstractEventLoop] = None
+        self._completion_event: Optional[asyncio.Event] = None
 
     @property
     def output_queue(self) -> janus.Queue:
@@ -222,6 +224,43 @@ def is_finished(self) -> bool:
             RequestStatus.TIMEOUT,
         ]
 
+    def bind_completion_event(self, loop: asyncio.AbstractEventLoop):
+        """Bind an asyncio event used by non-streaming HTTP waiters."""
+        self._completion_loop = loop
+        self._completion_event = asyncio.Event()
+        if self.is_finished():
+            self._completion_event.set()
+
+    async def wait_finished(self, timeout: Optional[float] = None):
+        """Wait until the request reaches a terminal state."""
+        if self.is_finished():
+            return True
+        if self._completion_event is None:
+            start = time.time()
+            while not self.is_finished():
+                if timeout is not None and time.time() - start > timeout:
+                    return False
+                await asyncio.sleep(0.001)
+            return True
+        try:
+            if timeout is None:
+                await self._completion_event.wait()
+            else:
+                await asyncio.wait_for(self._completion_event.wait(), timeout=timeout)
+            return True
+        except asyncio.TimeoutError:
+            return False
+
+    def _notify_finished(self):
+        event = self._completion_event
+        loop = self._completion_loop
+        if event is None:
+            return
+        if loop is not None and loop.is_running():
+            loop.call_soon_threadsafe(event.set)
+        else:
+            event.set()
+
     def abort(self):
         """Signal that the request has been aborted and should stop generation."""
         self._aborted = True
@@ -235,27 +274,31 @@ def mark_finished(self, reason: FinishReason):
         self.status = RequestStatus.FINISHED
         self.finish_reason = reason
         self.finished_time = time.time()
+        self._notify_finished()
 
     def mark_failed(self, reason: FinishReason = FinishReason.ERROR):
         """Mark the request as failed."""
         self.abort()
         self.status = RequestStatus.FAILED
         self.finish_reason = reason
         self.finished_time = time.time()
+        self._notify_finished()
 
     def mark_canceled(self):
         """Mark the request as canceled."""
         self.abort()
         self.status = RequestStatus.CANCELED
         self.finish_reason = FinishReason.CANCELED
         self.finished_time = time.time()
+        self._notify_finished()
 
     def mark_timeout(self):
         """Mark the request as timed out."""
         self.abort()
         self.status = RequestStatus.TIMEOUT
         self.finish_reason = FinishReason.TIMEOUT
         self.finished_time = time.time()
+        self._notify_finished()
 
     async def close(self):
         """Close the output queue and clean up resources."""