feat: support Intern-S2-Preview TS forecaster

docs/en/multi_modal/multimodal_inputs.md

@@ -348,12 +348,13 @@ ______________________________________________________________________
 
 ## Time Series
 
-> **Note:** Time series input is currently supported for the **InternS1-Pro** model only.
+> **Note:** Time series understanding is supported for **InternS1-Pro** and **Intern-S2-Preview** models. Time series
+> forecasting is supported for **Intern-S2-Preview** models with a time-series forecaster only.
 
-The `time_series_url` content item requires a `sampling_rate` field (in Hz) alongside the URL.
+The `time_series_url` content item requires a URL. Include `sampling_rate` in Hz when it is known.
 
 <details>
-<summary>Complete example</summary>
+<summary>Time series understanding example</summary>
 
 ```python
 from openai import OpenAI
@@ -389,6 +390,110 @@ print(response.choices[0].message.content)
 
 </details>
 
+<details>
+<summary>Intern-S2-Preview time series forecast example</summary>
+
+```python
+from openai import OpenAI
+
+client = OpenAI(api_key='EMPTY', base_url='http://localhost:23333/v1')
+model_name = client.models.list().data[0].id
+
+response = client.chat.completions.create(
+    model=model_name,
+    messages=[{
+        'role': 'user',
+        'content': [
+            {
+                'type': 'time_series_url',
+                'time_series_url': {
+                    'url': 'file:///path/to/history.npy',
+                    # Optional when the Intern-S2-Preview processor can infer a default.
+                    'sampling_rate': 100,
+                },
+            },
+            {
+                'type': 'text',
+                'text': 'Forecast the next 48 time points for this time series.',
+            },
+        ],
+    }],
+    temperature=0.0,
+    max_completion_tokens=32,
+    extra_body={
+        # Intern-S2-Preview forecast checkpoints route supported TS requests to forecast output.
+        'enable_forecasting': True,
+        # Optional output-length override. Horizon prediction still runs when supported.
+        'forecast_horizon': 48,
+    },
+)
+
+message = response.model_dump()['choices'][0]['message']
+forecast = message['ts_forecast']
+print('predicted_horizon:', forecast['predicted_horizon'])
+print('point_forecast:', forecast['point_forecast'])
+print('quantile_forecast:', forecast['quantile_forecast'])
+```
+
+The forecast payload contains:
+
+- `predicted_horizon`: the horizon predicted by Intern-S2-Preview. It is computed when the checkpoint has a horizon head,
+  including when `forecast_horizon` is set.
+- `forecast_horizon`: optional request field that overrides the forecast array length only. It does not suppress horizon
+  prediction.
+- `point_forecast`: the point forecast array.
+- `quantile_forecast`: quantile forecasts in `[median, 0.1, 0.2, ..., 0.9]` order.
+
+</details>
+
+<details>
+<summary>Intern-S2-Preview streaming forecast example</summary>
+
+```python
+from openai import OpenAI
+
+client = OpenAI(api_key='EMPTY', base_url='http://localhost:23333/v1')
+model_name = client.models.list().data[0].id
+
+stream = client.chat.completions.create(
+    model=model_name,
+    messages=[{
+        'role': 'user',
+        'content': [
+            {
+                'type': 'time_series_url',
+                'time_series_url': {
+                    'url': 'file:///path/to/history.npy',
+                    'sampling_rate': 100,
+                },
+            },
+            {'type': 'text', 'text': 'Forecast the next 48 time points.'},
+        ],
+    }],
+    temperature=0.0,
+    max_completion_tokens=32,
+    stream=True,
+    extra_body={
+        'enable_forecasting': True,
+        'forecast_horizon': 48,
+    },
+)
+
+forecast = None
+for chunk in stream:
+    delta = chunk.model_dump()['choices'][0].get('delta', {})
+    if delta.get('content'):
+        print(delta['content'], end='', flush=True)
+    if delta.get('ts_forecast') is not None:
+        forecast = delta['ts_forecast']
+
+print('\nforecast:', forecast)
+```
+
+For streaming responses, `ts_forecast` is attached to the final assistant delta.
+
+</details>
+
 ______________________________________________________________________
 
 ## Local Files and Base64

docs/zh_cn/multi_modal/multimodal_inputs.md

@@ -348,12 +348,13 @@ ______________________________________________________________________
 
 ## 时序数据
 
-> **注意：** 时序数据输入目前仅支持 **InternS1-Pro** 模型。
+> **注意：** 时序理解目前支持 **InternS1-Pro** 和 **Intern-S2-Preview** 模型。时序预测仅支持带有时序预测器的
+> **Intern-S2-Preview** 模型。
 
-`time_series_url` 内容项需要在 URL 之外额外提供 `sampling_rate` 字段（单位：Hz）。
+`time_series_url` 内容项需要提供 URL。已知采样率时，可以通过 `sampling_rate` 字段传入，单位为 Hz。
 
 <details>
-<summary>完整示例</summary>
+<summary>时序理解示例</summary>
 
 ```python
 from openai import OpenAI
@@ -388,6 +389,109 @@ print(response.choices[0].message.content)
 
 </details>
 
+<details>
+<summary>Intern-S2-Preview 时序预测示例</summary>
+
+```python
+from openai import OpenAI
+
+client = OpenAI(api_key='EMPTY', base_url='http://localhost:23333/v1')
+model_name = client.models.list().data[0].id
+
+response = client.chat.completions.create(
+    model=model_name,
+    messages=[{
+        'role': 'user',
+        'content': [
+            {
+                'type': 'time_series_url',
+                'time_series_url': {
+                    'url': 'file:///path/to/history.npy',
+                    # 当 Intern-S2-Preview processor 可以推断默认值时，该字段可选。
+                    'sampling_rate': 100,
+                },
+            },
+            {
+                'type': 'text',
+                'text': '请预测该时序接下来的 48 个时间点。',
+            },
+        ],
+    }],
+    temperature=0.0,
+    max_completion_tokens=32,
+    extra_body={
+        # Intern-S2-Preview 预测 checkpoint 会将支持的时序请求路由到预测输出。
+        'enable_forecasting': True,
+        # 可选输出长度覆盖项。支持时仍会执行 horizon 预测。
+        'forecast_horizon': 48,
+    },
+)
+
+message = response.model_dump()['choices'][0]['message']
+forecast = message['ts_forecast']
+print('predicted_horizon:', forecast['predicted_horizon'])
+print('point_forecast:', forecast['point_forecast'])
+print('quantile_forecast:', forecast['quantile_forecast'])
+```
+
+预测结果包含：
+
+- `predicted_horizon`: Intern-S2-Preview 预测出的 horizon。当 checkpoint 带有 horizon head 时会计算该字段，
+  即使请求中设置了 `forecast_horizon`。
+- `forecast_horizon`: 可选请求字段，仅用于覆盖预测数组长度。它不会抑制 horizon 预测。
+- `point_forecast`: 点预测数组。
+- `quantile_forecast`: 分位数预测，顺序为 `[median, 0.1, 0.2, ..., 0.9]`。
+
+</details>
+
+<details>
+<summary>Intern-S2-Preview 流式预测示例</summary>
+
+```python
+from openai import OpenAI
+
+client = OpenAI(api_key='EMPTY', base_url='http://localhost:23333/v1')
+model_name = client.models.list().data[0].id
+
+stream = client.chat.completions.create(
+    model=model_name,
+    messages=[{
+        'role': 'user',
+        'content': [
+            {
+                'type': 'time_series_url',
+                'time_series_url': {
+                    'url': 'file:///path/to/history.npy',
+                    'sampling_rate': 100,
+                },
+            },
+            {'type': 'text', 'text': '请预测接下来的 48 个时间点。'},
+        ],
+    }],
+    temperature=0.0,
+    max_completion_tokens=32,
+    stream=True,
+    extra_body={
+        'enable_forecasting': True,
+        'forecast_horizon': 48,
+    },
+)
+
+forecast = None
+for chunk in stream:
+    delta = chunk.model_dump()['choices'][0].get('delta', {})
+    if delta.get('content'):
+        print(delta['content'], end='', flush=True)
+    if delta.get('ts_forecast') is not None:
+        forecast = delta['ts_forecast']
+
+print('\nforecast:', forecast)
+```
+
+对于流式响应，`ts_forecast` 会附加在最后一个 assistant delta 上。
+
+</details>
+
 ______________________________________________________________________
 
 ## 本地文件与 Base64

lmdeploy/messages.py

@@ -110,6 +110,9 @@ class GenerationConfig:
             Must be non-negative; values below 0 are treated as 0.
         repetition_ngram_threshold: The number of times an n-gram must be repeated to trigger early stop.
             Must be non-negative; values below 0 are treated as 0.
+        forecast_horizon: Optional horizon used by time series forecast outputs.
+            Forecast routing is controlled by ``enable_forecasting``.
+        enable_forecasting: Whether to route supported time-series requests to forecast output.
     """
 
     n: int = 1
@@ -151,6 +154,10 @@ class GenerationConfig:
     repetition_ngram_size: int = 0
     repetition_ngram_threshold: int = 0
 
+    # time series forecasting
+    forecast_horizon: int | None = None
+    enable_forecasting: bool | None = None
+
     def convert_stop_bad_words_to_ids(self, tokenizer: Tokenizer):
         """Convert stop_words/bad_sords to ids and append the ids to
         stop_token_ids/bad_token_ids."""
@@ -200,6 +207,9 @@ def __post_init__(self):
         assert self.temperature >= 0 and self.temperature <= 2  # [0,2]
         assert 0 <= self.min_p <= 1, \
             f'min_p should be in range [0, 1], but found {self.min_p}'
+        assert (self.forecast_horizon is None
+                or (type(self.forecast_horizon) is int and self.forecast_horizon > 0)), \
+            'forecast_horizon must be a positive integer'
         if self.repetition_ngram_size <= 0 or self.repetition_ngram_threshold <= 0:
             self.repetition_ngram_size = 0
             self.repetition_ngram_threshold = 0
@@ -597,6 +607,7 @@ class Response:
     last_hidden_state: torch.Tensor = None
     index: int = 0
     routed_experts: Any = None
+    multimodal_outputs: dict[str, Any] | None = None
     cached_tokens: int = 0
 
     def __str__(self):
@@ -627,6 +638,7 @@ def _format_tensor(name: str, tensor: torch.Tensor | None) -> list[str]:
         fields.extend(_format_tensor('logits', self.logits))
         fields.extend(_format_tensor('last_hidden_state', self.last_hidden_state))
         fields.extend(_format_tensor('routed_experts', self.routed_experts))
+        fields.append(f'multimodal_outputs={self.multimodal_outputs}')
         return '\n'.join(fields)
 
     def extend(self, other: 'Response') -> 'Response':
@@ -653,6 +665,8 @@ def extend(self, other: 'Response') -> 'Response':
             self.logprobs = self.logprobs or []
             self.logprobs += other.logprobs
         self.routed_experts = other.routed_experts
+        self.multimodal_outputs = other.multimodal_outputs
+        self.cached_tokens = other.cached_tokens
         return self
 
 
@@ -739,6 +753,7 @@ class EngineOutput:
     cache_block_ids: list[int] | None = None
     req_metrics: RequestMetrics | None = None
     routed_experts: torch.Tensor = None
+    multimodal_outputs: dict[str, Any] | None = None
     ce_loss: float = None
 
 

lmdeploy/metrics/stats.py

@@ -240,6 +240,14 @@ def update_from_output(self, outputs: EngineOutput, req_stats: RequestStats):
 
         new_generation_tokens = len(outputs.token_ids)
         if new_generation_tokens == 0:
+            if outputs.status != ResponseType.SUCCESS:
+                # time series forecast-only responses can finish without emitting text tokens.
+                req_stats.finish_reason = outputs.status
+                req_stats.finish_time = self.iteration_timestamp
+                if req_stats.first_token_time == 0:
+                    req_stats.first_token_time = self.iteration_timestamp
+                    req_stats.lastest_token_time = self.iteration_timestamp
+                    self.prompt_tokens = req_stats.prompt_tokens
             return
 
         self.new_generation_tokens = new_generation_tokens

lmdeploy/pytorch/engine/engine.py

@@ -55,12 +55,15 @@ class InferOutput:
     # when Prefill Engine is Done.
     cache_block_ids: list[int] = None
 
-    # for logging
+    # logging
     req_metrics: RequestMetrics = None
 
     # expert ids
     routed_experts: torch.Tensor = None
 
+    # multimodal outputs, e.g. time series predictions
+    multimodal_outputs: dict[str, Any] | None = None
+
     # summed, unnormalized cross-entropy (NLL) of the input prompt
     ce_loss: float = None
 

lmdeploy/pytorch/engine/engine_instance.py

@@ -228,6 +228,7 @@ async def async_stream_infer(self,
 
                 cache_block_ids = resp.data.get('cache_block_ids', None) if resp.data else None
                 req_metrics = resp.data.get('req_metrics', None) if resp.data else None
+                multimodal_outputs = resp.data.get('multimodal_outputs', None) if resp.data else None
                 logprobs = resp.data.pop('logprobs', None) if resp.data else None
 
                 if resp.type == ResponseType.SUCCESS:
@@ -238,6 +239,7 @@ async def async_stream_infer(self,
                                        token_ids[output_offset:].tolist(),
                                        cache_block_ids=cache_block_ids,
                                        req_metrics=req_metrics,
+                                       multimodal_outputs=multimodal_outputs,
                                        logprobs=logprobs)
                     output_offset = len(token_ids)
                 elif resp.type in (ResponseType.FINISH, ResponseType.CANCEL):
@@ -265,6 +267,7 @@ async def async_stream_infer(self,
                                        cache_block_ids=cache_block_ids,
                                        req_metrics=req_metrics,
                                        routed_experts=routed_experts,
+                                       multimodal_outputs=multimodal_outputs,
                                        logprobs=logprobs,
                                        ce_loss=ce_loss)
                     break