Add server metrics monitor and DPO client by Yunnglin · Pull Request #132 · modelscope/twinkle

Yunnglin · 2026-03-30T06:32:52Z

PR type

Bug Fix
New Feature
Document Updates
More Models or Datasets Support

PR information

Write the detail information belongs to this PR.

Experiment results

Paste your experiment result here(if needed).

gemini-code-assist

Code Review

This pull request implements a centralized metrics system using Ray's utility metrics to monitor HTTP requests, task queues, and server resources across the gateway, model, processor, and sampler components. Feedback focuses on preventing metric cardinality explosion in the middleware by using route templates, adhering to PEP 8 by moving inline imports to the top level, and refining type hints and function signatures for better maintainability and type safety.

src/twinkle/server/utils/metrics.py

src/twinkle/server/gateway/server.py

src/twinkle/server/gateway/twinkle_gateway_handlers.py

src/twinkle/server/model/app.py

src/twinkle/server/processor/app.py

src/twinkle/server/sampler/app.py

src/twinkle/server/utils/metrics.py

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

Copilot

Pull request overview

This PR adds Ray Serve–compatible observability (HTTP/task-queue/resource metrics) and extends Tinker/Twinkle training flows to support DPO-style reference logprobs, alongside several operational/cookbook updates.

Changes:

Introduce a centralized ray.util.metrics module and wire it into Gateway/Model/Sampler/Processor apps plus task queue + rate limiter instrumentation.
Extend Tinker forward/forward_backward plumbing to support DPO (ref_logps extraction + ref_outputs propagation) and adjust backend collect behavior.
Update Ray launcher initialization behavior and add/refresh cookbook configs & scripts (including DPO client examples).

Reviewed changes

Copilot reviewed 26 out of 26 changed files in this pull request and generated 6 comments.

Show a summary per file

File	Description
src/twinkle/server/utils/task_queue.py	Adds task-queue metrics (queue depth/wait, execution time, task status counts) and passes gauges into the rate limiter.
src/twinkle/server/utils/state/server_state.py	Adds a background loop to periodically publish resource gauges (sessions/models/futures).
src/twinkle/server/utils/rate_limiter.py	Adds optional metrics gauge updates for active tokens tracked by the limiter.
src/twinkle/server/utils/metrics.py	New central metrics definitions + FastAPI middleware for request counters/latency histograms.
src/twinkle/server/sampler/app.py	Registers HTTP metrics middleware; sets task queue deployment label to `Sampler`.
src/twinkle/server/processor/app.py	Registers HTTP metrics middleware for `Processor`.
src/twinkle/server/model/app.py	Registers HTTP metrics middleware; sets task queue deployment label to `Model`.
src/twinkle/server/model/tinker_handlers.py	Adjusts template selection for Qwen3.5 and changes forward path to use updated backend return shape.
src/twinkle/server/model/backends/transformers_model.py	Refactors Tinker forward paths and updates Twinkle-native collect behavior for forward outputs.
src/twinkle/server/model/backends/megatron_model.py	Refactors Tinker forward paths and updates Twinkle-native collect behavior for forward outputs.
src/twinkle/server/model/backends/common.py	Adds shared helpers for Tinker loss setup/output building and ref_logps → ref_outputs conversion.
src/twinkle/server/common/datum.py	Extracts `ref_logps` from Datum loss inputs for DPO.
src/twinkle/model/megatron/multi_lora_megatron.py	Binds adapter_name into LoRA save converter via `functools.partial`.
src/twinkle/metric/dpo.py	Accepts non-tensor `logps` by converting to a tensor before alignment.
src/twinkle/loss/dpo.py	Accepts non-tensor `ref_logps` by converting to a tensor before alignment.
src/twinkle_client/utils/patch_tinker.py	Extends typing imports and introduces a new patch-state flag variable.
src/twinkle_client/common/serialize.py	Adds BaseModel serialization handling for client HTTP parameter serialization.
src/twinkle/server/launcher.py	Changes Ray initialization to attempt connecting to an existing cluster via `address='auto'`.
src/twinkle/server/gateway/server.py	Registers HTTP metrics middleware for `Gateway`.
src/twinkle/server/gateway/twinkle_gateway_handlers.py	Adds a `/twinkle/status` endpoint returning cleanup/resource counts.
pyproject.toml	Removes the upper bound on the `datasets` dependency.
cookbook/client/twinkle/self_host/dpo.py	Adds a Twinkle-native self-host DPO training example script.
cookbook/client/tinker/self_host/dpo.py	Adds a Tinker-compatible self-host DPO training example script.
cookbook/client/server/megatron/server_config.yaml	Minor YAML formatting tweak.
cookbook/client/server/megatron/server_config_4b.yaml	Updates sample deployment ports and various sizing/limit parameters.
cookbook/client/server/megatron/run.sh	Replaces the minimal launcher with a parameterized Ray+Prometheus+server startup script.

Copilot · 2026-04-07T12:12:32Z

src/twinkle/server/launcher.py

+            # Connect to existing cluster if available, otherwise start local instance
+            ray.init(
+                address='auto',
+                namespace=namespace,
+                runtime_env=runtime_env,
+            )
            logger.info(f'Ray initialized with namespace={namespace}')


ray.init(address='auto', ...) does not "start local instance" when no cluster is running; it raises an error if it can't connect. If this launcher is used in single-node mode, initialization will fail. Consider trying address='auto' first and falling back to a plain ray.init(namespace=..., runtime_env=...) (or making the address configurable) so local startup still works.

Suggested change

# Connect to existing cluster if available, otherwise start local instance

ray.init(

address='auto',

namespace=namespace,

runtime_env=runtime_env,

)

logger.info(f'Ray initialized with namespace={namespace}')

# Try to connect to an existing cluster first, and fall back to

# starting a local Ray instance when no cluster is available.

try:

ray.init(

address='auto',

namespace=namespace,

runtime_env=runtime_env,

)

logger.info(f'Connected to existing Ray cluster with namespace={namespace}')

except Exception as exc:

logger.info(

'Failed to connect to an existing Ray cluster with '

f"address='auto'; starting a local Ray instance instead: {exc}")

ray.init(

namespace=namespace,

runtime_env=runtime_env,

)

logger.info(f'Started local Ray instance with namespace={namespace}')

Copilot · 2026-04-07T12:12:32Z

src/twinkle/server/model/backends/common.py

+    @staticmethod
+    def _normalize_tensor_output(value):
+        """Normalize various output formats (tensor, list of tensors, nested lists, floats) to a single tensor.
+
+        Handles:
+        - torch.Tensor: detach and move to cpu
+        - list of torch.Tensor: cat along dim=0
+        - nested lists: recursively flatten and cat
+        - list of floats/int: convert to tensor
+        """
+        if value is None:
+            return None
+
+        if isinstance(value, torch.Tensor):
+            return value.detach().cpu()
+
+        if isinstance(value, list):
+            return torch.as_tensor(value, dtype=torch.float32).detach().cpu()
+


_normalize_tensor_output() currently routes any list through torch.as_tensor(...). This will break when the model returns a list of tensors (common for Megatron/TP/PP outputs) because torch.as_tensor([tensor, ...]) fails or produces an object tensor. The previous logic handled list-of-tensors by concatenating; please restore support for list/tuple of tensors (and optionally nested lists) to avoid runtime failures in _tinker_build_output().

Copilot · 2026-04-07T12:12:32Z

src/twinkle/server/model/backends/common.py

+        elif loss_fn == 'importance_sampling':
+            has_ref_logps = any('ref_logps' in d.loss_fn_inputs for d in inputs)
+            if has_ref_logps:
+                beta = kwargs.pop('dpo_beta', 0.1)
+                loss_type = kwargs.pop('dpo_loss_type', 'sigmoid')
+                sft_weight = kwargs.pop('dpo_sft_weight', 0.0)
+                self.set_loss(
+                    'DPOLoss', adapter_name=adapter_name, beta=beta, loss_type=loss_type, sft_weight=sft_weight)
+                self.add_metric('DPOMetric', adapter_name=adapter_name, beta=beta)
+            else:


_tinker_setup_loss() calls add_metric('DPOMetric', ...) every time loss_fn=='importance_sampling' with ref_logps. add_metric() appends to train_status.metrics, so this will accumulate duplicate DPOMetric instances over steps and can skew metric reporting / leak memory. Consider adding the metric only once per adapter (e.g., check existing metric types before appending) or making add_metric idempotent for this case.

Copilot · 2026-04-07T12:12:33Z

src/twinkle/server/model/tinker_handlers.py

+                output, loss = self.model.tinker_forward_only(
+                    inputs=datum_list, adapter_name=adapter_name, **loss_fn_config)
                return types.ForwardBackwardOutput(
                    loss_fn_output_type='CrossEntropyLossReturn',
                    loss_fn_outputs=output,
-                    metrics={'loss:sum': loss},
+                    metrics={'loss:avg': loss},
                )


The /tinker/forward handler reports metrics={'loss:avg': loss}, but tinker_forward_only() now returns a hard-coded 0.0 loss in both Transformers and Megatron backends. This makes the returned loss metric misleading and can break clients that rely on it. Either compute the forward loss as before (via calculate_loss) or omit/rename the loss metric for forward-only requests.

Copilot · 2026-04-07T12:12:33Z

src/twinkle_client/common/serialize.py

+        # Pydantic models: convert to dict for JSON serialization by requests
+        return obj.model_dump(mode='json')


serialize_object() is annotated to return str but returns a dict for Pydantic BaseModel. This function is also used by _serialize_params() for both JSON bodies and query params; returning a dict can lead to incorrect URL encoding / unexpected server-side parsing. Consider consistently returning JSON strings (e.g., json.dumps(model_dump(...))) or updating _serialize_params() to handle BaseModel differently for params vs JSON.

Suggested change

# Pydantic models: convert to dict for JSON serialization by requests

return obj.model_dump(mode='json')

# Pydantic models: serialize to a JSON string to keep behavior

# consistent across JSON bodies and query params.

return json.dumps(obj.model_dump(mode='json'), ensure_ascii=False)

Copilot · 2026-04-07T12:12:33Z

cookbook/client/server/megatron/run.sh

+            echo "  ./run.sh --cpu-workers 4 --head ''            # 纯 CPU 模式"
+            exit 0
+            ;;
+        *)
+            print_error "未知参数: $1"
+            echo "使用 --help 查看帮助"
+            exit 1
+            ;;


print_error is called in the argument-parsing loop before the function is defined. In bash, functions must be defined before they are invoked at runtime, so passing an unknown flag will error with print_error: command not found instead of printing the intended message. Define helper functions before the while [[ $# -gt 0 ]] loop (or inline the error printing) to make the error path work.

Yunnglin added 5 commits March 26, 2026 23:29

feat(metrics): add monitoring layer with status endpoint and ray metrics

cd383b2

feat(server_state): 添加资源指标更新功能

b546807

fix: resolve conflicts and add monitoring system

783648e

update

3e86e4f

update

88a94fd

gemini-code-assist bot reviewed Mar 30, 2026

View reviewed changes

Yunnglin and others added 16 commits March 30, 2026 15:37

Merge remote-tracking branch 'origin' into add_monitor

3dbd995

update

d61482c

update

a3cd140

Update src/twinkle/server/utils/metrics.py

85cc08f

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

update

74c6c19

update run shell

52f21b3

update

4a50c43

update

4fe4551

update twinkle dpo

1c42798

Merge remote-tracking branch 'origin' into add_monitor

f45fc13

update twinkle dpo

21a7c17

Merge remote-tracking branch 'origin' into add_monitor

7e46338

update transformers dpo

b396ade

update megatron dpo

df1f9a0

update megatron dpo

cec02d2

update megatron dpo

ea6df52

Yunnglin changed the title ~~Add server metrics monitor~~ Add server metrics monitor and DPO client Apr 7, 2026

Yunnglin marked this pull request as ready for review April 7, 2026 12:05

Copilot AI review requested due to automatic review settings April 7, 2026 12:05

Copilot started reviewing on behalf of Yunnglin April 7, 2026 12:06 View session

Copilot AI reviewed Apr 7, 2026

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add server metrics monitor and DPO client#132

Add server metrics monitor and DPO client#132
Yunnglin wants to merge 21 commits intomainfrom
add_monitor

Yunnglin commented Mar 30, 2026

Uh oh!

gemini-code-assist bot left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI Apr 7, 2026

Uh oh!

Copilot AI Apr 7, 2026

Uh oh!

Copilot AI Apr 7, 2026

Uh oh!

Copilot AI Apr 7, 2026

Uh oh!

Copilot AI Apr 7, 2026

Uh oh!

Copilot AI Apr 7, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

		# Pydantic models: convert to dict for JSON serialization by requests
		return obj.model_dump(mode='json')

-        # Pydantic models: convert to dict for JSON serialization by requests
-        return obj.model_dump(mode='json')
+        # Pydantic models: serialize to a JSON string to keep behavior
+        # consistent across JSON bodies and query params.
+        return json.dumps(obj.model_dump(mode='json'), ensure_ascii=False)

Conversation

Yunnglin commented Mar 30, 2026

PR type

PR information

Experiment results

Uh oh!

gemini-code-assist bot left a comment

Choose a reason for hiding this comment

Code Review

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Apr 7, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 7, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 7, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 7, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 7, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 7, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants