feat: client-side embedding training & multi-turn rollout support

docs/source_zh/使用指引/Embedding训练.md

@@ -106,6 +106,72 @@ for step, batch in enumerate(dataloader):
 
 ---
 
+## 通过 Twinkle Client 训练
+
+除了直接使用裸库 `TransformersModel`，你也可以通过 `twinkle_client` 的 `MultiLoraTransformersModel` 以 HTTP 方式驱动服务端完成 Embedding 训练。协议层（`/twinkle/*`）已经具备完成 Embedding 训练所需的全部能力，**无需引入任何新的高层封装函数**（例如 `setup_embedding_training`），只需按正确顺序调用现有方法即可。
+
+### 完整调用顺序
+
+client 化的 Embedding 训练遵循与裸库一致的调用顺序：
+
+```
+set_processor('InputProcessor')
+  -> set_loss('InfonceLoss', ...)
+  -> add_metric('EmbeddingMetric', is_training=True)
+  -> loop: forward_backward(inputs=mb, task='embedding') + clip_grad_and_step(...)
+  -> calculate_metric(is_training=True)
+```
+
+与裸库不同的是，client 侧传入的是**类名字符串**（如 `'InfonceLoss'`、`'EmbeddingMetric'`、`'InputProcessor'`），由服务端解析为对应的核心库类。
+
+### 示例
+
+```python
+from peft import LoraConfig
+from twinkle_client import init_twinkle_client
+from twinkle_client.model import MultiLoraTransformersModel
+
+# --- Connect to the running Twinkle server ---
+init_twinkle_client(base_url='http://127.0.0.1:8000', api_key='EMPTY_TOKEN')
+
+# --- Build the client model with a LoRA adapter for embedding ---
+model = MultiLoraTransformersModel(model_id='ms://Qwen/Qwen3.5-4B')
+model.add_adapter_to_model('emb_adapter', LoraConfig(target_modules='all-linear'))
+model.set_template('Qwen3_5Template')
+
+# --- Configure the embedding-training pipeline (order matters) ---
+# NOTE: pass class names as strings; the server resolves them to core-lib classes.
+model.set_processor('InputProcessor')
+model.set_loss('InfonceLoss', temperature=0.07, use_batch=True, hard_negatives=None)
+model.add_metric('EmbeddingMetric', is_training=True)
+
+# --- Training loop ---
+for step, mb in enumerate(minibatches):
+    # `task='embedding'` is forwarded through the /twinkle/* protocol as an extra
+    # kwarg and selects the embedding pooling + InfoNCE loss path on the server.
+    model.forward_backward(inputs=mb, task='embedding')
+    model.clip_grad_and_step(max_grad_norm=1.0)
+
+# --- Read back embedding metrics (pos_sim / neg_sim / loss) ---
+metric = model.calculate_metric(is_training=True)
+```
+
+### 关于 `TransformersEmbeddingPatch` 的自动应用
+
+Embedding 训练依赖 `TransformersEmbeddingPatch` 把 causal LM 的 `lm_head` 替换为 identity 输出，从而得到 per-token hidden states 用于池化。**你不需要显式调用 `apply_patch(TransformersEmbeddingPatch())`**：当你在 `forward_backward`（或 `forward_only`）中传入 `task='embedding'` 时，服务端的 `_resolve_task_context` 会在该次 `forward` 调用期间自动应用该 patch，并在调用结束后自动回滚。
+
+这意味着：
+
+- 同一个 adapter 在一次 `forward_backward(task='embedding')` 之后，紧接着调用不带 `task` 参数的 `forward_only` 仍会返回正常的词表维度 `logits`，不会残留上一次 embedding 任务的 identity hidden states。
+- 你在 client 侧真正需要显式调用的前置步骤只有三个：`set_processor('InputProcessor')`、`set_loss('InfonceLoss', ...)` 与 `add_metric('EmbeddingMetric', is_training=True)`。
+
+### 说明
+
+- 本节不引入任何新的高层封装函数，仅说明现有 `MultiLoraTransformersModel` 方法的正确调用顺序。
+- 各参数（`temperature`、`use_batch`、`hard_negatives` 等）的含义与取值建议参见上文「关键参数」；`calculate_metric` 返回的指标含义参见下文「监控指标」。
+
+---
+
 ## 监控指标
 
 `EmbeddingMetric` 报告关键训练信号：

docs/source_zh/使用指引/服务端和客户端/Tinker兼容客户端.md

@@ -195,3 +195,68 @@ rest_client = service_client.create_rest_client()
 rest_client.publish_checkpoint_from_tinker_path(save_result.path).result()
 print("Published checkpoint to ModelScope Hub")
 ```
+
+## 多轮工具调用（间接方案）
+
+Tinker 兼容客户端本身不提供多轮 agentic rollout 编排能力。如果你在用 Tinker Client 训练/管理 checkpoint，同时又需要多轮工具调用能力（**仅用于生成或评测**），可以在完全不改动 Tinker SDK 代码的前提下，额外使用一个标准的 OpenAI 兼容客户端指向 Gateway 的 `/chat/completions` 端点，并复用 `twinkle_agentic.rollout.APIMultiTurnRollout` 与 `ToolManager`。
+
+这是一条纯增量的使用路径：它复用的是 Twinkle Server 上一个独立于 Tinker 协议命名空间的、已经存在并跑通的 OpenAI 兼容端点，因此不涉及对 Tinker SDK 源码或 `patch_tinker.py` 中现有 monkeypatch 范围的任何修改。
+
+```python
+import os
+from twinkle.data_format.sampling import SamplingParams
+from twinkle_agentic.protocol.openai import OpenAI
+from twinkle_agentic.rollout import APIMultiTurnRollout
+from twinkle_agentic.tools.tool_manager import ToolManager
+# from your_project.tools import CalculatorTool, SearchTool
+
+# An OpenAI-compatible client pointing at the Gateway route,
+# NOT at tinker's base_url. This is a plain additive usage of an
+# already-existing endpoint; the tinker SDK is untouched.
+api = OpenAI(
+    model='Qwen/Qwen3.5-4B',
+    api_key=os.environ.get('MODELSCOPE_TOKEN', 'EMPTY_API_KEY'),
+    base_url='http://localhost:8000/api/v1',  # Gateway /chat/completions route
+)
+
+# Register your tools; ToolManager is reused directly from twinkle_agentic.
+tool_manager = ToolManager([
+    # CalculatorTool(),
+    # SearchTool(),
+])
+
+rollout = APIMultiTurnRollout(
+    api=api,
+    tool_manager=tool_manager,
+    sampling_params=SamplingParams(temperature=0.7, max_tokens=512),
+    max_turns=6,
+    concurrency=8,
+)
+
+trajectories_in = [
+    {'messages': [{'role': 'user', 'content': 'What is 12 * 34?'}]},
+]
+trajectories_out = rollout(trajectories_in)
+
+# trajectories_out[i]['messages'] contains the full conversation, including
+# assistant tool_calls and tool-response turns.
+# trajectories_out[i]['stop_reason'] is one of 'stop' | 'length' | 'max_turns' | 'api_error'.
+for traj in trajectories_out:
+    for message in traj['messages']:
+        print(message['role'], ':', message.get('content'))
+```
+
+### 能力边界：不可用于 RL 训练
+
+该间接方案返回的 Trajectory **不包含** `logprobs` 字段（端到端验证已确认走 Gateway `/chat/completions` 的 `APIMultiTurnRollout` 产出的 trajectory 中不存在 `logprobs`，也没有 `input_ids` 等训练所需字段）。因此该路径**仅适合生成数据、离线评测等场景，不能直接喂给依赖 token 级对齐信息的 RL 训练（如 GRPO）**。
+
+如果多轮 rollout 的产出需要直接用于 RL 训练（需要 token 级 `logprobs` 与 `labels` 对齐），请使用 Twinkle 客户端侧的 `ClientMultiTurnRollout`（通过 `/twinkle/sample` 返回 `new_input_feature`/`logprobs`），详见《Twinkle 客户端》文档。
+
+### 为什么 Tinker 客户端不支持 embedding 训练与可训练多轮 rollout
+
+这两项能力无法通过 Tinker 兼容客户端提供，根本原因在于 Tinker 协议的数据类型层面缺少所需结构：
+
+- **Embedding 训练**：`tinker.types.Datum` 不具备对比学习所需的样本分组结构（anchor/positive 成对分组），无法承载 embedding 训练的输入语义。
+- **可训练多轮 rollout**：`tinker.types.SampledSequence` 不具备 `new_input_feature` 字段，无法在多轮之间传递并累积 token 级对齐信息，因此产出无法用于训练。
+
+由于 Tinker 是第三方 SDK、不可修改其源码，即使想补齐这些字段也需要改动 SDK 本身（不允许）。因此上述能力仅在 Twinkle 客户端侧支持，Tinker 兼容客户端只提供上文所述的"仅生成/评测"间接多轮方案。

src/twinkle/server/sampler/backends/mock_sampler.py

@@ -14,7 +14,7 @@
 
 import numpy as np
 import time
-from typing import Any
+from typing import Any, Iterable
 
 from twinkle import remote_class, remote_function
 from twinkle.data_format import SampledSequence, SampleResponse, SamplingParams
@@ -23,16 +23,61 @@
 
 logger = get_logger()
 
+# Recognised ctor / per-call knobs that tune multi-turn control flow. Kept in a
+# set so the ctor can log only *genuinely* unknown kwargs (a real backend
+# signature drift) while still accepting these opt-in multi-turn knobs.
+_MULTI_TURN_KNOBS = ('stop_reason', 'tool_call_text', 'tool_call_turns')
+
 
 @remote_class()
 class MockSampler:
-    """Deterministic mock sampler for CPU-only testing."""
+    """Deterministic mock sampler for CPU-only testing.
+
+    Beyond the base sampler surface, this backend exposes a few opt-in knobs so
+    a local (CPU-only) multi-turn rollout e2e can be driven without a GPU:
+
+    - ``stop_reason`` (default ``'length'``): the ``SampledSequence.stop_reason``
+      emitted for every sampled sequence. ``'length'`` terminates a multi-turn
+      loop immediately; ``'stop'`` lets the loop proceed to tool-call parsing.
+    - ``tool_call_text`` (default ``None``): when set, this text is emitted as
+      ``SampledSequence.decoded`` on the configured turns so the rollout's
+      ``template.parse_tool_call`` can produce a tool call and exercise the
+      "sample -> tool -> bridge -> sample" control flow.
+    - ``tool_call_turns`` (default ``(1,)`` when ``tool_call_text`` is set):
+      the 1-based turn indices on which ``tool_call_text`` is injected. A turn
+      corresponds to one ``sample()`` invocation (one multi-turn round).
 
-    def __init__(self, model_id: str, *, seed: int = 0, vocab_size: int = 32, **kwargs: Any) -> None:
+    All three knobs may be supplied at construction time or overridden per call
+    via ``sample()`` keyword arguments or matching attributes on
+    ``sampling_params``. When none are configured the backend keeps its previous
+    behaviour exactly (``stop_reason='length'``, ``decoded=None``), so existing
+    callers are unaffected. Outputs stay deterministic and CPU-only.
+    """
+
+    def __init__(
+        self,
+        model_id: str,
+        *,
+        seed: int = 0,
+        vocab_size: int = 32,
+        stop_reason: str = 'length',
+        tool_call_text: str | None = None,
+        tool_call_turns: Iterable[int] | None = None,
+        **kwargs: Any,
+    ) -> None:
         self.model_id = model_id
         self._seed = int(seed)
         self._vocab_size = int(vocab_size)
         self._adapters: dict[str, Any] = {}
+        # Multi-turn control-flow knobs (see class docstring).
+        self._stop_reason = str(stop_reason)
+        self._tool_call_text = tool_call_text
+        self._tool_call_turns = self._normalize_turns(tool_call_turns, tool_call_text)
+        # Per-round counter: incremented once per ``sample()`` call so that
+        # ``tool_call_turns`` can address individual multi-turn rounds. Only
+        # consulted when tool-call injection is active, so the default path
+        # stays fully stateless and deterministic.
+        self._round = 0
         # Surface (rather than silently swallow) extra ctor kwargs: a real
         # backend signature drift then shows up as a visible DEBUG warning in
         # the mock e2e instead of being discarded without trace.
@@ -64,9 +109,19 @@ def sample(
             raise ValueError(f'max_tokens must be >= 1, got {max_tokens!r} '
                              '(set sampling_params.max_tokens to a positive integer)')
 
+        # Resolve per-call multi-turn knobs (precedence: explicit sample()
+        # kwargs > sampling_params attributes > ctor defaults) and advance the
+        # round counter that ``tool_call_turns`` addresses.
+        stop_reason = self._resolve_stop_reason(sampling_params, kwargs)
+        tool_call_text = self._resolve_tool_call_text(sampling_params, kwargs)
+        tool_call_turns = self._resolve_tool_call_turns(sampling_params, kwargs, tool_call_text)
+        self._round += 1
+        inject_tool_call = tool_call_text is not None and self._round in tool_call_turns
+        decoded = tool_call_text if inject_tool_call else None
+
         normalized = self._normalize_inputs(inputs)
         responses: list[SampleResponse] = []
-        for prompt_idx, _ in enumerate(normalized):
+        for prompt_idx, pif in enumerate(normalized):
             sequences: list[SampledSequence] = []
             for sample_idx in range(num_samples):
                 seed = stable_seed(self.model_id, adapter_name, self._seed, prompt_idx, sample_idx)
@@ -78,11 +133,14 @@ def sample(
                 # mock returns top-1 with the chosen token. The tinker handler
                 # flattens this to a single chosen-token logprob.
                 logprobs = [[(tok, float(lp))] for tok, lp in zip(tokens, logprobs_per_token)]
-                sequences.append(SampledSequence(
-                    stop_reason='length',
-                    tokens=tokens,
-                    logprobs=logprobs,
-                ))
+                sequences.append(
+                    SampledSequence(
+                        stop_reason=stop_reason,
+                        tokens=tokens,
+                        logprobs=logprobs,
+                        decoded=decoded,
+                        new_input_feature=self._build_new_input_feature(pif, tokens),
+                    ))
             responses.append(SampleResponse(sequences=sequences))
         return responses
 
@@ -150,3 +208,77 @@ def _resolve_max_tokens(params: SamplingParams | None) -> int | None:
         if params is None:
             return None
         return getattr(params, 'max_tokens', None)
+
+    @staticmethod
+    def _normalize_turns(turns: Iterable[int] | None, tool_call_text: str | None) -> frozenset[int]:
+        """Normalise the ``tool_call_turns`` config into a ``frozenset[int]``.
+
+        When ``tool_call_text`` is set but no turns are given, default to
+        ``{1}`` — inject the tool call exactly once, on the first round.
+        """
+        if turns is None:
+            return frozenset({1}) if tool_call_text is not None else frozenset()
+        return frozenset(int(t) for t in turns)
+
+    def _resolve_stop_reason(self, params: SamplingParams | None, kwargs: dict[str, Any]) -> str:
+        if 'stop_reason' in kwargs and kwargs['stop_reason'] is not None:
+            return str(kwargs['stop_reason'])
+        override = getattr(params, 'stop_reason', None)
+        if override is not None:
+            return str(override)
+        return self._stop_reason
+
+    def _resolve_tool_call_text(self, params: SamplingParams | None, kwargs: dict[str, Any]) -> str | None:
+        if 'tool_call_text' in kwargs:
+            return kwargs['tool_call_text']
+        override = getattr(params, 'tool_call_text', None)
+        if override is not None:
+            return override
+        return self._tool_call_text
+
+    def _resolve_tool_call_turns(
+        self,
+        params: SamplingParams | None,
+        kwargs: dict[str, Any],
+        tool_call_text: str | None,
+    ) -> frozenset[int]:
+        if 'tool_call_turns' in kwargs and kwargs['tool_call_turns'] is not None:
+            return self._normalize_turns(kwargs['tool_call_turns'], tool_call_text)
+        override = getattr(params, 'tool_call_turns', None)
+        if override is not None:
+            return self._normalize_turns(override, tool_call_text)
+        # Fall back to the ctor-configured turns, but keep the "default to {1}
+        # when text is injected via a per-call override" behaviour consistent.
+        if tool_call_text is not None and not self._tool_call_turns:
+            return frozenset({1})
+        return self._tool_call_turns
+
+    @staticmethod
+    def _build_new_input_feature(pif: Any, tokens: list[int]) -> dict[str, Any]:
+        """Deterministically append the freshly sampled ``tokens`` to ``pif``.
+
+        Produces a plain-dict ``InputFeature`` that carries the running context
+        for the next multi-turn round: ``input_ids`` is the prior prompt plus
+        this round's sampled tokens, and ``labels`` marks the sampled tokens as
+        trainable (their own ids) while prior/context positions stay ``-100``.
+        This mirrors the shape a real sampler's ``concat_input_feature`` yields,
+        which the multi-turn rollout relies on (it reads
+        ``new_input_feature.input_ids`` and counts trainable ``labels``).
+
+        The mock does not depend on a template, so the append is a simple,
+        deterministic list concatenation performed entirely on the CPU.
+        """
+        feat: dict[str, Any] = dict(pif) if isinstance(pif, dict) else {}
+        prev_ids = list(feat.get('input_ids') or [])
+        prev_labels = feat.get('labels')
+        if prev_labels is not None and len(prev_labels) == len(prev_ids):
+            labels = list(prev_labels)
+        else:
+            # No (or misaligned) prior labels: treat the entire prior context as
+            # non-trainable so only this round's sampled tokens count.
+            labels = [-100] * len(prev_ids)
+
+        feat['input_ids'] = prev_ids + list(tokens)
+        feat['labels'] = labels + list(tokens)
+        feat['length'] = len(feat['input_ids'])
+        return feat

src/twinkle_agentic/rollout/__init__.py

@@ -1,6 +1,7 @@
 # Copyright (c) ModelScope Contributors. All rights reserved.
 from .api_multi_turn import APIMultiTurnRollout
 from .base import Rollout
+from .bridge import extend_with_bridge
 from .multi_turn import MultiTurnRollout
 from .multi_turn_condense import MultiTurnCondenseRollout
 
@@ -9,4 +10,5 @@
     'MultiTurnCondenseRollout',
     'MultiTurnRollout',
     'Rollout',
+    'extend_with_bridge',
 ]