Add serving progress and adjust deepseek weight load#63
Conversation
superxf
commented
Jul 9, 2026
superxf
commented
- 为权重加载过程添加进度的日志打印
- 将权重加载的时机改到模型创建时，而不是在发出请求时再加载
Integrate the DeepSeekV4 Flash W8A8 NPU serving path with TP=8 support, model loading, tokenizer handling, worker startup, and OpenAI-compatible request flow. Add packed prefill/decode runner glue, host-side LM head projection, DeepSeek-specific cache slot management, scheduler-only KV block accounting, and kernel contract validation for the pypto-lib submodule. Document the DeepSeekV4 serving command and cover the integration with focused unit tests.
📝 WalkthroughWalkthroughThis PR adds end-to-end DeepSeekV4 W8A8 NPU serving support: a new weight loader and NPU executor under examples, CLI model-family detection and topology validation, a DeepSeekV4 directory loader, preflight hooks, configurable worker timeouts, multi-token-prediction decode context, tokenizer fallback loading, a pypto-lib bump, and extensive tests. ChangesDeepSeekV4 W8A8 serving support
Estimated code review effort: 5 (Critical) | ~120 minutes Possibly related issues
Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces serving integration for the DeepSeek V4 model on NPU platforms, including a dedicated executor, lazy weight loading, model topology validation, and support for multi-token prediction (MTP) by tracking previous token IDs and embeddings. The feedback highlights several critical issues and code quality improvements: a duplicate register_model method definition in pypto_executor.py with incorrect return type annotations, a logical bug in serving_worker.py when determining the previous token ID for empty output histories, redundant checks for device embedding support, unused variables/imports in weight_loader.py, silent skipping of missing constants in the kernel contract validation, and a missing UTF-8 encoding specification when reading the tokenizer configuration.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| def register_model(self, model_id: str, record: ModelRecord) -> None: | ||
| """Compile kernels for ``record`` and attach a runner to ``model_id``. | ||
|
|
||
| Returns the number of KV cache pages allocated on the device so the | ||
| caller can synchronise host-side block metadata. | ||
| """ | ||
| print("[register_model] compiling kernels …", flush=True) | ||
| def register_model(self, model_id: str, record: ModelRecord) -> None: | ||
| """Compile kernels for ``record`` and attach a runner to ``model_id``.""" |
There was a problem hiding this comment.
在 python/core/pypto_executor.py 中，存在两个连续的 register_model 方法定义。第一个定义（第 56-62 行）会被第二个定义（第 63 行起）完全覆盖，导致其中的 print 语句永远不会被执行。此外，第二个定义的返回类型标注为 -> None，但该方法实际上在第 88 行返回了 num_pages（一个整数），这与类型标注不符，且调用方（如 serving_worker.py）也期望获取返回的页数。建议删除冗余的第一个定义，并将保留的 register_model 方法的返回类型标注修正为 -> int。
| def register_model(self, model_id: str, record: ModelRecord) -> None: | |
| """Compile kernels for ``record`` and attach a runner to ``model_id``. | |
| Returns the number of KV cache pages allocated on the device so the | |
| caller can synchronise host-side block metadata. | |
| """ | |
| print("[register_model] compiling kernels …", flush=True) | |
| def register_model(self, model_id: str, record: ModelRecord) -> None: | |
| """Compile kernels for ``record`` and attach a runner to ``model_id``.""" | |
| def register_model(self, model_id: str, record: ModelRecord) -> int: | |
| """Compile kernels for ``record`` and attach a runner to ``model_id``. | |
| Returns the number of KV cache pages allocated on the device so the | |
| caller can synchronise host-side block metadata. | |
| """ | |
| print("[register_model] compiling kernels …", flush=True) |
| if len(output_ids) >= 2: | ||
| prev_token = output_ids[-2] | ||
| elif output_ids and prompt_ids: | ||
| prev_token = prompt_ids[-1] | ||
| else: | ||
| prev_token = last_token |
There was a problem hiding this comment.
这里的逻辑旨在获取绝对位置为 seq_len - 2 的 Token。如果 output_ids 为空（即 len(output_ids) == 0），此时 seq_len 等于 len(prompt_ids)。如果 prompt_ids 长度大于等于 2，那么 seq_len - 2 对应的 Token 应该是 prompt_ids[-2]，而不是 prompt_ids[-1]（即 last_token）。当前的 elif output_ids and prompt_ids: 限制了 output_ids 必须非空，导致当 output_ids 为空时会直接走到 else 分支，将 prev_token 设为 last_token（即 prompt_ids[-1]）。建议将逻辑调整为更精确地处理 output_ids 为空且 prompt_ids 长度充足的情况。
| if len(output_ids) >= 2: | |
| prev_token = output_ids[-2] | |
| elif output_ids and prompt_ids: | |
| prev_token = prompt_ids[-1] | |
| else: | |
| prev_token = last_token | |
| if len(output_ids) >= 2: | |
| prev_token = output_ids[-2] | |
| elif len(output_ids) == 1: | |
| prev_token = prompt_ids[-1] | |
| elif len(prompt_ids) >= 2: | |
| prev_token = prompt_ids[-2] | |
| else: | |
| prev_token = last_token |
| prev_embeddings = torch.zeros_like(decode_embeddings) | ||
| else: | ||
| decode_embeddings = self.executor.lookup_embeddings(runtime_model, decode_token_tensor) | ||
| prev_token_tensor = torch.tensor(prev_tokens, dtype=torch.long, device=device) | ||
| prev_embeddings = self.executor.lookup_embeddings(runtime_model, prev_token_tensor) | ||
|
|
||
| if self.executor.supports_device_embedding: | ||
| prev_token_tensor = torch.tensor(prev_tokens, dtype=torch.long, device=device) |
There was a problem hiding this comment.
在 self.executor.supports_device_embedding 为 True 和 False 的分支中，prev_token_tensor 都可以直接在各自的分支内初始化。这样可以避免在最外层进行重复的 if self.executor.supports_device_embedding: 判断，使代码结构更加清晰、紧凑。
| prev_embeddings = torch.zeros_like(decode_embeddings) | |
| else: | |
| decode_embeddings = self.executor.lookup_embeddings(runtime_model, decode_token_tensor) | |
| prev_token_tensor = torch.tensor(prev_tokens, dtype=torch.long, device=device) | |
| prev_embeddings = self.executor.lookup_embeddings(runtime_model, prev_token_tensor) | |
| if self.executor.supports_device_embedding: | |
| prev_token_tensor = torch.tensor(prev_tokens, dtype=torch.long, device=device) | |
| prev_embeddings = torch.zeros_like(decode_embeddings) | |
| prev_token_tensor = torch.tensor(prev_tokens, dtype=torch.long, device=device) | |
| else: | |
| decode_embeddings = self.executor.lookup_embeddings(runtime_model, decode_token_tensor) | |
| prev_token_tensor = torch.tensor(prev_tokens, dtype=torch.long, device=device) | |
| prev_embeddings = self.executor.lookup_embeddings(runtime_model, prev_token_tensor) |
| import time | ||
|
|
||
| pack_t0 = time.perf_counter() |
There was a problem hiding this comment.
变量 pack_t0 在第 617 行被赋值，但在后续代码中完全没有被使用。此外，import time 也是多余的。建议将它们删除以保持代码整洁。
| if actual is not None and actual != expected: | ||
| mismatched.append(f"{name}={actual} expected {expected}") | ||
| expected_module_constants = { |
There was a problem hiding this comment.
在 _validate_kernel_contract 中，如果 _int_constant_from_file 返回 None（表示常量未找到或无法解析），当前的逻辑会默默跳过该校验（actual is not None）。这会导致如果常量缺失或格式不符合预期时，校验无法起到“快速失败（fail-fast）”的作用，可能会在后续编译或运行阶段抛出更难排查的错误。建议在 actual is None 时也将其加入到 mismatched 列表中，或者显式提示该常量缺失。
actual = _int_constant_from_file(config_path, name)
if actual is None:
mismatched.append(f"{name} is missing or unparseable")
elif actual != expected:
mismatched.append(f"{name}={actual} expected {expected}")| actual = _int_constant_from_file(module_path, name) | ||
| if actual is not None and actual != expected: | ||
| mismatched.append(f"{filename}:{name}={actual} expected {expected}") |
There was a problem hiding this comment.
在 _validate_kernel_contract 中，如果 _int_constant_from_file 返回 None（表示常量未找到或无法解析），当前的逻辑会默默跳过该校验（actual is not None）。这会导致如果常量缺失或格式不符合预期时，校验无法起到“快速失败（fail-fast）”的作用，可能会在后续编译或运行阶段抛出更难排查的错误。建议在 actual is None 时也将其加入到 mismatched 列表中，或者显式提示该常量缺失。
| actual = _int_constant_from_file(module_path, name) | |
| if actual is not None and actual != expected: | |
| mismatched.append(f"{filename}:{name}={actual} expected {expected}") | |
| actual = _int_constant_from_file(module_path, name) | |
| if actual is None: | |
| mismatched.append(f"{filename}:{name} is missing or unparseable") | |
| elif actual != expected: | |
| mismatched.append(f"{filename}:{name}={actual} expected {expected}") |
| if not tokenizer_file.exists(): | ||
| raise FileNotFoundError(f"Missing tokenizer.json in {model_path}") | ||
| config_path = model_path / "tokenizer_config.json" | ||
| tokenizer_config = json.loads(config_path.read_text()) if config_path.exists() else {} |
There was a problem hiding this comment.
在读取 tokenizer_config.json 时，建议显式指定 encoding="utf-8"，以避免在不同操作系统（如 Windows）上因系统默认编码不同而导致读取失败或解析错误。
| tokenizer_config = json.loads(config_path.read_text()) if config_path.exists() else {} | |
| tokenizer_config = json.loads(config_path.read_text(encoding="utf-8")) if config_path.exists() else {} |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 9
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
python/core/async_engine.py (1)
210-217: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winAvoid blocking shutdown on the full worker-step timeout.
With DeepSeek’s 1200s step timeout,
stop()can wait up to 20 minutes at Line 214 before_shutdown_worker()runs if_engine_loop()is blocked inoutput_queue.get(). Cancel the loop task during shutdown so stop can proceed promptly.Proposed fix
self._running = False if self._loop_task is not None: - await self._loop_task + self._loop_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await self._loop_task self._loop_task = NoneAlso applies to: 343-348
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@python/core/async_engine.py` around lines 210 - 217, The shutdown path in stop() is waiting for the engine loop to finish naturally before calling _shutdown_worker(), which can block for the full step timeout when _engine_loop() is stuck on output_queue.get(). Update stop() to cancel the running loop task before awaiting it, and make sure _engine_loop() handles cancellation cleanly so shutdown can proceed immediately; apply the same fix to the other stop/shutdown path referenced by the matching async engine methods.
🧹 Nitpick comments (2)
tests/test_deepseek_v4.py (1)
791-792: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueUnused variable
sparse0The variable
sparse0is unpacked but never used. Prefix it with_to signal intentional discard.♻️ Proposed fix
- sparse0, lens0 = prepared.sparse_inputs_for_ratio(0) + _sparse0, lens0 = prepared.sparse_inputs_for_ratio(0)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/test_deepseek_v4.py` around lines 791 - 792, The test in sparse_inputs_for_ratio unpacking creates an unused value named sparse0, so update the assignment to intentionally discard that result. Keep the existing prepared.sparse_inputs_for_ratio calls in test_deepseek_v4.py, but rename the unused variable to start with an underscore so the intent is clear and linting no longer flags it.Source: Linters/SAST tools
examples/model/deepseek_v4/runner/weight_loader.py (1)
93-94: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsolidate duplicated compress-ratio constants.
Two sets of constants hold identical values under different names:
_DEEPSEEK_V4_CSA_COMPRESS_RATIO/_DEEPSEEK_V4_HCA_COMPRESS_RATIO(lines 93-94, used in packing) and_DEEPSEEK_V4_CSA_COMPRESS_RATIO_VALUE/_DEEPSEEK_V4_HCA_COMPRESS_RATIO_VALUE(lines 186-187, used in stacking). If one set is changed without the other, weight-name generation and layer-stacking grouping will silently diverge.♻️ Proposed fix: use a single set of constants
_DEEPSEEK_V4_CSA_COMPRESS_RATIO_VALUE = 4 _DEEPSEEK_V4_HCA_COMPRESS_RATIO_VALUE = 128Then update references in
_attention_suffixes_for_compress_ratio(lines 267, 269) and_pack_deepseek_v4_optional_attention(lines 820-821) to use_DEEPSEEK_V4_CSA_COMPRESS_RATIO_VALUE/_DEEPSEEK_V4_HCA_COMPRESS_RATIO_VALUE, and remove the duplicate lines 93-94.Also applies to: 186-187
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@examples/model/deepseek_v4/runner/weight_loader.py` around lines 93 - 94, The DeepSeek V4 compress-ratio values are duplicated in two constant pairs, which can diverge over time; consolidate them into a single source of truth. Remove the redundant _DEEPSEEK_V4_CSA_COMPRESS_RATIO and _DEEPSEEK_V4_HCA_COMPRESS_RATIO definitions and update _attention_suffixes_for_compress_ratio and _pack_deepseek_v4_optional_attention to reference _DEEPSEEK_V4_CSA_COMPRESS_RATIO_VALUE and _DEEPSEEK_V4_HCA_COMPRESS_RATIO_VALUE instead, keeping the packing and stacking logic aligned.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@examples/model/deepseek_v4/runner/weight_loader.py`:
- Around line 615-635: The weight-loading loop in `load_packed_layer_weights`
starts `pack_t0` but never uses it, so add a final completion log after
`stack_deepseek_v4_layer_weights` that reports the elapsed time since
`time.perf_counter()` started. Also move the `import time` out of the function
and into the module-level imports to keep imports consistent.
In `@pypto-lib`:
- Line 1: The pypto-lib submodule is pinned to a commit that no longer exists in
the upstream repository, so checkout fails. Update the submodule reference to a
valid commit that is present in hw-native-sys/pypto-lib.git, and make sure the
recorded pin for pypto-lib in the parent repo points to an accessible revision
so submodule initialization succeeds.
In `@python/core/async_engine.py`:
- Line 509: The shutdown path using asyncio.gather over reversed(self._cores)
currently stops on the first core.stop() error and may leave other cores still
shutting down; update this gather call to wait for every core.stop() to complete
even if some fail, by handling exceptions with return_exceptions=True and then
processing any failures after all stops have been observed in the async engine
shutdown logic.
- Around line 35-45: The _positive_env_timeout_seconds helper currently accepts
non-finite values like nan and inf because it only checks timeout <= 0 after
float conversion. Update this function to explicitly reject non-finite results
before returning, alongside the existing positive check, so environment timeouts
validated here only allow finite positive seconds and keep the behavior
consistent with the current error handling in _positive_env_timeout_seconds.
In `@python/core/model_loader.py`:
- Around line 91-92: The exception handling in model_loader’s import fallback
drops the original ImportError context. Update the except ImportError block to
re-raise the RuntimeError while preserving the caught exception as the cause, so
the failure in the safetensors import path remains traceable. Locate this in the
checkpoint loading logic around the safetensors import inside the model loader
and keep the original exception chain intact.
In `@python/core/pypto_executor.py`:
- Around line 72-81: The preflight path in PyptoExecutor leaves allocated runner
resources behind if runner.preflight(record) raises after init_kv_cache
succeeds. Update the try/finally handling around the init_kv_cache and preflight
sequence so that any failure in either step triggers the same close cleanup used
for init_kv_cache failures, using the existing runner.close hook and the
PyptoExecutor.preflight flow to locate the change. Make sure the runner is
cleaned up before re-raising so unregistered runners do not leak resources.
- Around line 69-87: Restore compiled artifact registration in PyptoExecutor
after creating the runner: the compiled model returned by _compile_model() is
currently never stored, so lookup_embeddings() can later fail when it reads
self._compiled[model_id]. Update the load path in PyptoExecutor to save the
compiled artifact under the current model_id before or alongside runner
initialization, keeping the existing init_kv_cache(), preflight(), and _runners
registration flow intact.
- Around line 56-64: The duplicate register_model definition in PyptoExecutor is
overriding the first implementation, so collapse it into a single method and
keep the int return contract used by serving_worker.py. Preserve the existing
compile/logging behavior in register_model, ensure the method still returns
num_pages, and remove the redundant second definition so the progress log is not
dead code.
In `@python/core/serving_worker.py`:
- Around line 150-153: The request-release cleanup in busy_loop() can fail
before the main try/except path, which prevents any StepOutput from being
emitted. Move or wrap the release_finished_requests() call so failures are
captured and reported through the existing step error handling, using the
busy_loop() flow and the release_finished symbol to locate the change. Ensure a
release failure still results in an error StepOutput instead of letting the
worker exit silently.
---
Outside diff comments:
In `@python/core/async_engine.py`:
- Around line 210-217: The shutdown path in stop() is waiting for the engine
loop to finish naturally before calling _shutdown_worker(), which can block for
the full step timeout when _engine_loop() is stuck on output_queue.get(). Update
stop() to cancel the running loop task before awaiting it, and make sure
_engine_loop() handles cancellation cleanly so shutdown can proceed immediately;
apply the same fix to the other stop/shutdown path referenced by the matching
async engine methods.
---
Nitpick comments:
In `@examples/model/deepseek_v4/runner/weight_loader.py`:
- Around line 93-94: The DeepSeek V4 compress-ratio values are duplicated in two
constant pairs, which can diverge over time; consolidate them into a single
source of truth. Remove the redundant _DEEPSEEK_V4_CSA_COMPRESS_RATIO and
_DEEPSEEK_V4_HCA_COMPRESS_RATIO definitions and update
_attention_suffixes_for_compress_ratio and _pack_deepseek_v4_optional_attention
to reference _DEEPSEEK_V4_CSA_COMPRESS_RATIO_VALUE and
_DEEPSEEK_V4_HCA_COMPRESS_RATIO_VALUE instead, keeping the packing and stacking
logic aligned.
In `@tests/test_deepseek_v4.py`:
- Around line 791-792: The test in sparse_inputs_for_ratio unpacking creates an
unused value named sparse0, so update the assignment to intentionally discard
that result. Keep the existing prepared.sparse_inputs_for_ratio calls in
test_deepseek_v4.py, but rename the unused variable to start with an underscore
so the intent is clear and linting no longer flags it.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 7e045509-2b66-4ee6-b55b-7b1fa644a6d8
📒 Files selected for processing (18)
docs/dev/model/deepseek-v4.mdexamples/model/deepseek_v4/__init__.pyexamples/model/deepseek_v4/runner/__init__.pyexamples/model/deepseek_v4/runner/npu_executor.pyexamples/model/deepseek_v4/runner/npu_runner.pyexamples/model/deepseek_v4/runner/weight_loader.pypypto-libpython/cli/main.pypython/core/async_engine.pypython/core/model_loader.pypython/core/model_runner.pypython/core/pypto_executor.pypython/core/serving_worker.pypython/core/tokenizer.pypython/core/types.pytests/test_batching.pytests/test_deepseek_v4.pytests/test_parallel.py
💤 Files with no reviewable changes (1)
- tests/test_parallel.py
| import time | ||
|
|
||
| pack_t0 = time.perf_counter() | ||
| for layer_id in range(num_hidden_layers): | ||
| per_layer.append( | ||
| self.load_packed_layer_weights( | ||
| layer_id, | ||
| ranks=ranks, | ||
| n_routed_experts=n_routed_experts, | ||
| compress_ratio=int(compress_ratios[layer_id]), | ||
| include_tid2eid=layer_id < num_hash_layers, | ||
| include_gate_bias=layer_id >= num_hash_layers, | ||
| ) | ||
| ) | ||
| if layer_id % 5 == 0 or layer_id == num_hidden_layers - 1: | ||
| logger.info( | ||
| "DeepSeekV4 weight load progress: layer %d/%d", | ||
| layer_id + 1, | ||
| num_hidden_layers, | ||
| ) | ||
| return stack_deepseek_v4_layer_weights(per_layer, compress_ratios=compress_ratios) |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
pack_t0 timer is started but never used.
time.perf_counter() is called at line 617 but the elapsed time is never logged. Since the PR objective is to add progress logging during weight loading, add a completion log after stacking finishes to make the timer useful.
⏱️ Proposed fix: add completion elapsed-time log
return stack_deepseek_v4_layer_weights(per_layer, compress_ratios=compress_ratios)
+ stacked = stack_deepseek_v4_layer_weights(per_layer, compress_ratios=compress_ratios)
+ pack_elapsed = time.perf_counter() - pack_t0
+ logger.info("DeepSeekV4 weight load complete: %d layers in %.2fs", num_hidden_layers, pack_elapsed)
+ return stackedAlso move import time to the module-level imports at the top of the file.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@examples/model/deepseek_v4/runner/weight_loader.py` around lines 615 - 635,
The weight-loading loop in `load_packed_layer_weights` starts `pack_t0` but
never uses it, so add a final completion log after
`stack_deepseek_v4_layer_weights` that reports the elapsed time since
`time.perf_counter()` started. Also move the `import time` out of the function
and into the module-level imports to keep imports consistent.
| @@ -1 +1 @@ | |||
| Subproject commit 57772f304bbcaee927b51227f6aa495a5591debf | |||
| Subproject commit c159c325a8d4279d0af65fbd528c9096bfb6a57a | |||
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
expected_sha="c159c325a8d4279d0af65fbd528c9096bfb6a57a"
actual_sha="$(git ls-tree HEAD pypto-lib | awk '{print $3}')"
test "$actual_sha" = "$expected_sha" || {
echo "pypto-lib pointer is $actual_sha, expected $expected_sha" >&2
exit 1
}
url="$(git config -f .gitmodules --get submodule.pypto-lib.url)"
git ls-remote "$url" "$expected_sha" | grep -q "$expected_sha" || {
echo "Commit $expected_sha was not found at $url" >&2
exit 1
}
test -d pypto-lib || {
echo "pypto-lib submodule directory is missing; CI must checkout submodules for this PR." >&2
exit 1
}
python - <<'PY'
from pathlib import Path
root = Path("pypto-lib")
required = {
"config.py",
"prefill_attention_hca.py",
"prefill_attention_csa.py",
"prefill_layer.py",
"prefill_fwd.py",
"decode_layer.py",
"decode_fwd.py",
}
candidate_dirs = {}
for path in root.rglob("*.py"):
if path.name in required:
candidate_dirs.setdefault(path.parent, set()).add(path.name)
complete = [directory for directory, names in candidate_dirs.items() if required <= names]
if not complete:
print("No pypto-lib directory contains the full DeepSeekV4 required kernel file set:", sorted(required))
raise SystemExit(1)
print("Found DeepSeekV4 kernel contract candidate(s):")
for directory in complete:
print(f" - {directory}")
PYRepository: hw-native-sys/pypto-serving
Length of output: 277
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf 'Top-level files:\n'
git ls-files | sed -n '1,120p'
printf '\nSearch for pypto-lib and submodule-related references:\n'
rg -n --hidden -S 'pypto-lib|submodule|git submodule|checkout.*submodule|actions/checkout|submodules|skip.*pypto-lib' .github . || trueRepository: hw-native-sys/pypto-serving
Length of output: 11721
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf 'Tree for .github:\n'
find .github -maxdepth 3 -type f | sort | sed -n '1,200p'
printf '\nRelevant workflow excerpts:\n'
for f in $(find .github -maxdepth 3 -type f | sort); do
echo "### $f"
rg -n -S 'submodules|checkout|pypto-lib|git submodule|skip' "$f" || true
doneRepository: hw-native-sys/pypto-serving
Length of output: 2128
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "Submodule config:"
cat -n .gitmodules
echo
echo "Workflow files mentioning checkout/submodules:"
find .github -type f | sort | while read -r f; do
if rg -n -S 'actions/checkout|submodules:|git submodule|pypto-lib' "$f" >/tmp/matches.txt; then
echo "### $f"
cat /tmp/matches.txt
fi
doneRepository: hw-native-sys/pypto-serving
Length of output: 1551
Update the pypto-lib submodule pin pypto-lib points to c159c325a8d4279d0af65fbd528c9096bfb6a57a, which is absent from https://github.com/hw-native-sys/pypto-lib.git, so submodule checkout will fail.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pypto-lib` at line 1, The pypto-lib submodule is pinned to a commit that no
longer exists in the upstream repository, so checkout fails. Update the
submodule reference to a valid commit that is present in
hw-native-sys/pypto-lib.git, and make sure the recorded pin for pypto-lib in the
parent repo points to an accessible revision so submodule initialization
succeeds.
| def _positive_env_timeout_seconds(name: str, default: float) -> float: | ||
| raw = os.environ.get(name) | ||
| if raw is None or raw.strip() == "": | ||
| return default | ||
| try: | ||
| timeout = float(raw) | ||
| except ValueError as exc: | ||
| raise ValueError(f"{name} must be a positive number of seconds") from exc | ||
| if timeout <= 0: | ||
| raise ValueError(f"{name} must be a positive number of seconds") | ||
| return timeout |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Reject non-finite timeout values.
float("nan") bypasses timeout <= 0, and float("inf") makes bounded waits effectively unbounded. Validate finiteness before returning.
Proposed fix
+import math
import os- if timeout <= 0:
+ if not math.isfinite(timeout) or timeout <= 0:
raise ValueError(f"{name} must be a positive number of seconds")📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| def _positive_env_timeout_seconds(name: str, default: float) -> float: | |
| raw = os.environ.get(name) | |
| if raw is None or raw.strip() == "": | |
| return default | |
| try: | |
| timeout = float(raw) | |
| except ValueError as exc: | |
| raise ValueError(f"{name} must be a positive number of seconds") from exc | |
| if timeout <= 0: | |
| raise ValueError(f"{name} must be a positive number of seconds") | |
| return timeout | |
| import math | |
| import os | |
| def _positive_env_timeout_seconds(name: str, default: float) -> float: | |
| raw = os.environ.get(name) | |
| if raw is None or raw.strip() == "": | |
| return default | |
| try: | |
| timeout = float(raw) | |
| except ValueError as exc: | |
| raise ValueError(f"{name} must be a positive number of seconds") from exc | |
| if not math.isfinite(timeout) or timeout <= 0: | |
| raise ValueError(f"{name} must be a positive number of seconds") | |
| return timeout |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@python/core/async_engine.py` around lines 35 - 45, The
_positive_env_timeout_seconds helper currently accepts non-finite values like
nan and inf because it only checks timeout <= 0 after float conversion. Update
this function to explicitly reject non-finite results before returning,
alongside the existing positive check, so environment timeouts validated here
only allow finite positive seconds and keep the behavior consistent with the
current error handling in _positive_env_timeout_seconds.
| """Stop all DP engine cores.""" | ||
| for core in reversed(self._cores): | ||
| await core.stop() | ||
| await asyncio.gather(*(core.stop() for core in reversed(self._cores))) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Wait for every core stop even when one fails.
asyncio.gather() without return_exceptions=True can propagate the first core.stop() failure before the remaining stop operations are observed, leaving workers shutting down in the background.
Proposed fix
- await asyncio.gather(*(core.stop() for core in reversed(self._cores)))
+ results = await asyncio.gather(
+ *(core.stop() for core in reversed(self._cores)),
+ return_exceptions=True,
+ )
+ failures = [result for result in results if isinstance(result, BaseException)]
+ if failures:
+ raise RuntimeError("Failed to stop one or more replica cores") from failures[0]📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| await asyncio.gather(*(core.stop() for core in reversed(self._cores))) | |
| results = await asyncio.gather( | |
| *(core.stop() for core in reversed(self._cores)), | |
| return_exceptions=True, | |
| ) | |
| failures = [result for result in results if isinstance(result, BaseException)] | |
| if failures: | |
| raise RuntimeError("Failed to stop one or more replica cores") from failures[0] |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@python/core/async_engine.py` at line 509, The shutdown path using
asyncio.gather over reversed(self._cores) currently stops on the first
core.stop() error and may leave other cores still shutting down; update this
gather call to wait for every core.stop() to complete even if some fail, by
handling exceptions with return_exceptions=True and then processing any failures
after all stops have been observed in the async engine shutdown logic.
| except ImportError: | ||
| raise RuntimeError("safetensors is required to read weight names from a single-shard checkpoint.") |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Preserve the original import error context.
Ruff flags this except block because the replacement RuntimeError drops the ImportError cause.
Proposed fix
- except ImportError:
- raise RuntimeError("safetensors is required to read weight names from a single-shard checkpoint.")
+ except ImportError as exc:
+ raise RuntimeError(
+ "safetensors is required to read weight names from a single-shard checkpoint."
+ ) from exc📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| except ImportError: | |
| raise RuntimeError("safetensors is required to read weight names from a single-shard checkpoint.") | |
| except ImportError as exc: | |
| raise RuntimeError( | |
| "safetensors is required to read weight names from a single-shard checkpoint." | |
| ) from exc |
🧰 Tools
🪛 Ruff (0.15.20)
[warning] 92-92: Within an except clause, raise exceptions with raise ... from err or raise ... from None to distinguish them from errors in exception handling
(B904)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@python/core/model_loader.py` around lines 91 - 92, The exception handling in
model_loader’s import fallback drops the original ImportError context. Update
the except ImportError block to re-raise the RuntimeError while preserving the
caught exception as the cause, so the failure in the safetensors import path
remains traceable. Locate this in the checkpoint loading logic around the
safetensors import inside the model loader and keep the original exception chain
intact.
Source: Linters/SAST tools
| def register_model(self, model_id: str, record: ModelRecord) -> None: | ||
| """Compile kernels for ``record`` and attach a runner to ``model_id``. | ||
|
|
||
| Returns the number of KV cache pages allocated on the device so the | ||
| caller can synchronise host-side block metadata. | ||
| """ | ||
| print("[register_model] compiling kernels …", flush=True) | ||
| def register_model(self, model_id: str, record: ModelRecord) -> None: | ||
| """Compile kernels for ``record`` and attach a runner to ``model_id``.""" |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Collapse the duplicate register_model definitions and keep the int return contract.
Line 63 overrides the method started on Line 56, so the Line 62 progress log is dead code. The method also still returns num_pages, which serving_worker.py consumes.
Proposed fix
- def register_model(self, model_id: str, record: ModelRecord) -> None:
+ def register_model(self, model_id: str, record: ModelRecord) -> int:
"""Compile kernels for ``record`` and attach a runner to ``model_id``.
Returns the number of KV cache pages allocated on the device so the
caller can synchronise host-side block metadata.
"""
- print("[register_model] compiling kernels …", flush=True)
- def register_model(self, model_id: str, record: ModelRecord) -> None:
- """Compile kernels for ``record`` and attach a runner to ``model_id``."""
import time
+ logger.info("PyptoExecutor %s: compiling kernels …", model_id)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| def register_model(self, model_id: str, record: ModelRecord) -> None: | |
| """Compile kernels for ``record`` and attach a runner to ``model_id``. | |
| Returns the number of KV cache pages allocated on the device so the | |
| caller can synchronise host-side block metadata. | |
| """ | |
| print("[register_model] compiling kernels …", flush=True) | |
| def register_model(self, model_id: str, record: ModelRecord) -> None: | |
| """Compile kernels for ``record`` and attach a runner to ``model_id``.""" | |
| def register_model(self, model_id: str, record: ModelRecord) -> int: | |
| """Compile kernels for ``record`` and attach a runner to ``model_id``. | |
| Returns the number of KV cache pages allocated on the device so the | |
| caller can synchronise host-side block metadata. | |
| """ | |
| logger.info("PyptoExecutor %s: compiling kernels …", model_id) | |
| import time |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@python/core/pypto_executor.py` around lines 56 - 64, The duplicate
register_model definition in PyptoExecutor is overriding the first
implementation, so collapse it into a single method and keep the int return
contract used by serving_worker.py. Preserve the existing compile/logging
behavior in register_model, ensure the method still returns num_pages, and
remove the redundant second definition so the progress log is not dead code.
| compiled = self._compile_model(record.runtime_model) | ||
| runner = self._create_runner(model_id, compiled) | ||
|
|
||
| try: | ||
| num_pages = runner.init_kv_cache(model_id, record.config, record.runtime) | ||
| except Exception: | ||
| close = getattr(runner, "close", None) | ||
| if callable(close): | ||
| close() | ||
| raise | ||
| self._compiled[model_id] = compiled | ||
|
|
||
| with profile_span("PyptoExecutor.preflight", cat="executor", args={"model_id": model_id}): | ||
| runner.preflight(record) | ||
| logger.info( | ||
| "PyptoExecutor %s: model loaded (%.1fs total)", | ||
| model_id, | ||
| time.perf_counter() - start_t0, | ||
| ) | ||
| self._runners[model_id] = runner |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Restore compiled artifact registration.
DeepSeekV4PyptoExecutor.lookup_embeddings() reads self._compiled[model_id]; without storing compiled here, registered DeepSeek models can fail later as “not registered”.
Proposed fix
logger.info(
"PyptoExecutor %s: model loaded (%.1fs total)",
model_id,
time.perf_counter() - start_t0,
)
+ self._compiled[model_id] = compiled
self._runners[model_id] = runner📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| compiled = self._compile_model(record.runtime_model) | |
| runner = self._create_runner(model_id, compiled) | |
| try: | |
| num_pages = runner.init_kv_cache(model_id, record.config, record.runtime) | |
| except Exception: | |
| close = getattr(runner, "close", None) | |
| if callable(close): | |
| close() | |
| raise | |
| self._compiled[model_id] = compiled | |
| with profile_span("PyptoExecutor.preflight", cat="executor", args={"model_id": model_id}): | |
| runner.preflight(record) | |
| logger.info( | |
| "PyptoExecutor %s: model loaded (%.1fs total)", | |
| model_id, | |
| time.perf_counter() - start_t0, | |
| ) | |
| self._runners[model_id] = runner | |
| compiled = self._compile_model(record.runtime_model) | |
| runner = self._create_runner(model_id, compiled) | |
| try: | |
| num_pages = runner.init_kv_cache(model_id, record.config, record.runtime) | |
| except Exception: | |
| close = getattr(runner, "close", None) | |
| if callable(close): | |
| close() | |
| raise | |
| with profile_span("PyptoExecutor.preflight", cat="executor", args={"model_id": model_id}): | |
| runner.preflight(record) | |
| logger.info( | |
| "PyptoExecutor %s: model loaded (%.1fs total)", | |
| model_id, | |
| time.perf_counter() - start_t0, | |
| ) | |
| self._compiled[model_id] = compiled | |
| self._runners[model_id] = runner |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@python/core/pypto_executor.py` around lines 69 - 87, Restore compiled
artifact registration in PyptoExecutor after creating the runner: the compiled
model returned by _compile_model() is currently never stored, so
lookup_embeddings() can later fail when it reads self._compiled[model_id].
Update the load path in PyptoExecutor to save the compiled artifact under the
current model_id before or alongside runner initialization, keeping the existing
init_kv_cache(), preflight(), and _runners registration flow intact.
| try: | ||
| num_pages = runner.init_kv_cache(model_id, record.config, record.runtime) | ||
| except Exception: | ||
| close = getattr(runner, "close", None) | ||
| if callable(close): | ||
| close() | ||
| raise | ||
| self._compiled[model_id] = compiled | ||
|
|
||
| with profile_span("PyptoExecutor.preflight", cat="executor", args={"model_id": model_id}): | ||
| runner.preflight(record) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Clean up runner resources when preflight fails.
runner.preflight(record) now materializes deferred weights after KV cache allocation; if it raises, the runner is not registered and the existing cleanup path is skipped.
Proposed fix
- with profile_span("PyptoExecutor.preflight", cat="executor", args={"model_id": model_id}):
- runner.preflight(record)
+ try:
+ with profile_span("PyptoExecutor.preflight", cat="executor", args={"model_id": model_id}):
+ runner.preflight(record)
+ except Exception:
+ close = getattr(runner, "close", None)
+ if callable(close):
+ close()
+ raise📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| try: | |
| num_pages = runner.init_kv_cache(model_id, record.config, record.runtime) | |
| except Exception: | |
| close = getattr(runner, "close", None) | |
| if callable(close): | |
| close() | |
| raise | |
| self._compiled[model_id] = compiled | |
| with profile_span("PyptoExecutor.preflight", cat="executor", args={"model_id": model_id}): | |
| runner.preflight(record) | |
| try: | |
| num_pages = runner.init_kv_cache(model_id, record.config, record.runtime) | |
| except Exception: | |
| close = getattr(runner, "close", None) | |
| if callable(close): | |
| close() | |
| raise | |
| try: | |
| with profile_span("PyptoExecutor.preflight", cat="executor", args={"model_id": model_id}): | |
| runner.preflight(record) | |
| except Exception: | |
| close = getattr(runner, "close", None) | |
| if callable(close): | |
| close() | |
| raise |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@python/core/pypto_executor.py` around lines 72 - 81, The preflight path in
PyptoExecutor leaves allocated runner resources behind if
runner.preflight(record) raises after init_kv_cache succeeds. Update the
try/finally handling around the init_kv_cache and preflight sequence so that any
failure in either step triggers the same close cleanup used for init_kv_cache
failures, using the existing runner.close hook and the PyptoExecutor.preflight
flow to locate the change. Make sure the runner is cleaned up before re-raising
so unregistered runners do not leak resources.
| if cmd.finished_request_ids: | ||
| pass # No allocation cleanup needed | ||
| release_finished = getattr(self.executor, "release_finished_requests", None) | ||
| if callable(release_finished): | ||
| release_finished(cmd.finished_request_ids) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Include request-release failures in the step error path.
release_finished_requests() runs before the try, so a cleanup failure exits busy_loop() without putting a StepOutput; the engine then waits until the worker-step timeout.
Proposed fix
elif cmd.type == "step":
- if cmd.finished_request_ids:
- release_finished = getattr(self.executor, "release_finished_requests", None)
- if callable(release_finished):
- release_finished(cmd.finished_request_ids)
-
try:
+ if cmd.finished_request_ids:
+ release_finished = getattr(self.executor, "release_finished_requests", None)
+ if callable(release_finished):
+ release_finished(cmd.finished_request_ids)
result = self._execute_step(cmd.scheduler_output)
self.output_queue.put(result)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| if cmd.finished_request_ids: | |
| pass # No allocation cleanup needed | |
| release_finished = getattr(self.executor, "release_finished_requests", None) | |
| if callable(release_finished): | |
| release_finished(cmd.finished_request_ids) | |
| elif cmd.type == "step": | |
| try: | |
| if cmd.finished_request_ids: | |
| release_finished = getattr(self.executor, "release_finished_requests", None) | |
| if callable(release_finished): | |
| release_finished(cmd.finished_request_ids) | |
| result = self._execute_step(cmd.scheduler_output) | |
| self.output_queue.put(result) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@python/core/serving_worker.py` around lines 150 - 153, The request-release
cleanup in busy_loop() can fail before the main try/except path, which prevents
any StepOutput from being emitted. Move or wrap the release_finished_requests()
call so failures are captured and reported through the existing step error
handling, using the busy_loop() flow and the release_finished symbol to locate
the change. Ensure a release failure still results in an error StepOutput
instead of letting the worker exit silently.
2 participants
