add deepseek-v4-flash offline infer - #143
Merged
Merged
Conversation
|
Important Review skippedAuto incremental reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
📝 WalkthroughWalkthroughAdds DeepSeek V4 Flash W8A8 offline generation for Ascend NPUs. The change includes an eight-device engine configuration, structured generation APIs, a CLI, documentation, and unit tests. ChangesDeepSeek V4 offline generation
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant CLI as npu_generate.py
participant Config as build_deepseek_v4_offline_engine_config
participant Engine as AsyncLLMEngine
CLI->>Config: Build validated DeepSeek V4 EngineConfig
Config-->>CLI: Return offline engine configuration
CLI->>Engine: Start engine and submit prompts
Engine-->>CLI: Stream outputs or return GenerateResult values
CLI->>Engine: Stop profiling and engine
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 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 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 `@pypto_serving/serving/engine/async_engine.py`:
- Around line 1042-1045: Update generate_batch to create explicit tasks for each
generate_result call, then gather those tasks; on any gather failure, cancel
unfinished sibling tasks, await them with return_exceptions enabled so cleanup
completes, and re-raise the original exception. Add a regression test covering
one failing request and one long-blocking request, asserting the blocker is
cancelled and its cleanup finishes.
🪄 Autofix
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 Plus
Run ID: 86d572f9-c810-4447-a365-ec092b1d5d01
📒 Files selected for processing (8)
README.mddocs/dev/model/deepseek-v4.mdexamples/model/deepseek_v4/npu_generate.pypypto_serving/model/deepseek/offline.pypypto_serving/serving/engine/async_engine.pytests/unit/model/deepseek/test_offline.pytests/unit/serving/engine/test_async_engine_replicas.pytests/unit/serving/engine/test_output_delivery.py
Comment on lines
+1042
to
+1045
| return list( | ||
| await asyncio.gather( | ||
| *(self.generate_result(prompt, generate_config) for prompt in prompts) | ||
| ) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
ast-grep outline pypto_serving/serving/engine/async_engine.py --items all
rg -n -C 8 'asyncio\.gather|async def generate_batch|finish_reason == "error"|async def add_request' \
pypto_serving/serving/engine/async_engine.py \
tests/unit/serving/engine/test_async_engine_replicas.pyRepository: hw-native-sys/pypto-serving
Length of output: 16741
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== async_engine around generate_result/generate_batch/add_request =="
sed -n '900,1175p' pypto_serving/serving/engine/async_engine.py
echo
echo "== request cleanup/abort implementations =="
rg -n -C 6 'async def abort_request|def \.abort|abort_request\(|def add_request|def _set_profile_active|cleanup|finished|final_output|finish_reason == "error"' pypto_serving/serving
echo
echo "== existing async engine tests around batching/errors =="
fd -t f . tests | xargs rg -n -C 5 'generate_batch|generate_result|cancel|return_exceptions|on_queued|queued' || true
echo
echo "== asyncio gather cancellation probe =="
python3 - <<'PY'
import asyncio
async def task(idx, event):
if idx == 0:
await event.wait()
raise RuntimeError("sibling blocked forever unless gather cancels")
await asyncio.sleep(0.001)
return idx
async def main():
events = [asyncio.Event() for _ in range(2)]
gathered = asyncio.gather(task(0, events[0]), task(1, events[1]))
try:
await gathered
except Exception as exc:
print(type(exc).__name__, str(exc))
# Give event loop a chance to run any cleanup/continuation from cancellation.
await asyncio.sleep(0.01)
print("task0 done", asyncio.isfuture(asyncio.current_task()), "cancelled" if gathered.cancelled() else str(gathered))
try:
asyncio.run(main(), debug=True)
except Exception as exc:
print(type(exc).__name__, str(exc))
PYRepository: hw-native-sys/pypto-serving
Length of output: 157
🏁 Script executed:
#!/bin/bash
set -euo pipefail
timeout 60 bash -c '
echo "== async_engine generate_result/generate_batch/add_request =="
sed -n "1020,1060p" pypto_serving/serving/engine/async_engine.py
echo
echo "== request cleanup candidates =="
rg -n -C 4 'abort_request|add_request|on_queued|request_id|context|cleanup|finish_reason == "error"|final_output is None' pypto_serving/serving/engine/async_engine.py pypto_serving/serving -g "*.py" || true
echo
echo "== existing batch/error cancellation tests =="
fd -t f . tests | xargs rg -n -C 4 'generate_batch|generate_result|cancel|return_exceptions|on_queued|queued' || true
' || true
python3 - <<'PY'
import asyncio
async def task(idx, event):
if idx == 0:
await event.wait()
raise RuntimeError("sibling failed")
await asyncio.sleep(0.01)
return idx
async def main():
events = [asyncio.Event() for _ in range(2)]
gathered = asyncio.gather(task(0, events[0]), task(1, events[1]))
try:
await gathered
except Exception as exc:
print(type(exc).__name__, str(exc))
await asyncio.sleep(0.02)
print("gathered done", gathered.done(), "gathered cancelled", gathered.cancelled())
asyncio.run(main())
PYRepository: hw-native-sys/pypto-serving
Length of output: 157
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== async_engine generate_result/generate_batch/add_request =="
sed -n '1000,1075p' pypto_serving/serving/engine/async_engine.py
echo
echo "== request cleanup/abort implementations =="
rg -n -C 5 'async def abort_request|async def add_request|function add_request|class .*Context|final_output is None|finish_reason == "error"|def cleanup|context' pypto_serving/serving/engine/async_engine.py pypto_serving/serving -g "*.py" || true
echo
echo "== existing async engine tests around batching/errors/cancellation =="
fd -t f . tests | xargs rg -n -C 4 'generate_batch|generate_result|cancel|return_exceptions|on_queued|queued|aborted|cleanup|finished' || true
echo
echo "== asyncio.gather propagation/probe =="
python3 - <<'PY'
import asyncio
async def blocking(idx):
await asyncio.sleep(10)
async def failing(idx):
await asyncio.sleep(0)
raise RuntimeError(f"sibling {idx} failed")
async def main():
gathered = asyncio.gather(failing(0), blocking(1))
try:
await gathered
except Exception as exc:
print("exception_type", type(exc).__name__)
print("exception_msg", str(exc))
# Let the event loop try to schedule continuation from any pending cancellation.
await asyncio.sleep(0.01)
print("gathered done", gathered.done())
print("gathered cancelled", gathered.cancelled())
asyncio.run(main(), debug=True)
PYRepository: hw-native-sys/pypto-serving
Length of output: 50383
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== add_request body =="
sed -n '1055,1092p' pypto_serving/serving/engine/async_engine.py
echo
echo "== abort_request bodies =="
sed -n '351,430p' pypto_serving/serving/engine/async_engine.py
sed -n '1092,1104p' pypto_serving/serving/engine/async_engine.py
echo
echo "== focused async engine replica tests =="
sed -n '130,200p' tests/unit/serving/engine/test_async_engine_replicas.pyRepository: hw-native-sys/pypto-serving
Length of output: 7884
Cancel sibling batch requests when one fails.
generate_batch directly passes generate_result coroutines to asyncio.gather(), so a failure propagates without cancelling the remaining coroutines. If a sibling remains blocked, it can continue occupying scheduler/state while the caller receives an error.
Create tasks for each generate_result(), catch the gather exception, cancel unfinished tasks awaiting return_exceptions, then re-raise. Add a regression test with one failing request and one long-blocking request asserting the blocker is cancelled and its cleanup completes.
🤖 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_serving/serving/engine/async_engine.py` around lines 1042 - 1045,
Update generate_batch to create explicit tasks for each generate_result call,
then gather those tasks; on any gather failure, cancel unfinished sibling tasks,
await them with return_exceptions enabled so cleanup completes, and re-raise the
original exception. Add a regression test covering one failing request and one
long-blocking request, asserting the blocker is cancelled and its cleanup
finishes.
Source: Coding guidelines
superxf
changed the title
bumble0918
reviewed
Aug 6, 2026
| max_new_tokens: int = 32, | ||
| max_num_seqs: int = 32, | ||
| max_num_batched_tokens: int = 512, | ||
| long_prefill_token_threshold: int = 2048, |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Add offline generation support for DeepSeek V4 Flash W8A8 without starting an HTTP server. The offline entry reuses the serving scheduler, distributed worker, grouped KV-cache, and MTP acceptance paths so its execution topology stays aligned with online serving.
Key changes
AsyncLLMEnginewith structuredgenerate_resultandgenerate_batchAPIs, final token IDs, normalized finish reasons, and sibling-task cleanup when a batch request fails.Current constraints
temperature=0).--long-prefill-token-threshold 128.Verification
platform-buildpassed.pre-commitpassed.unit-testspassed.