[Optimization] refactor(chat_handler,completion_handler): extract base classes and use AsyncLLM - #5195
Conversation
|
Thanks for your contribution! |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## develop #5195 +/- ##
==========================================
Coverage ? 65.29%
==========================================
Files ? 332
Lines ? 42413
Branches ? 6537
==========================================
Hits ? 27694
Misses ? 12643
Partials ? 2076
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
5054e85 to
0396b26
Compare
61252cb to
8bf1998
Compare
d31e772 to
3efcb2c
Compare
|
|
54fd6f0 to
e3ac4b2
Compare
…e classes and use AsyncLLM
3ec23b5 to
abf0acf
Compare
| ) -> AsyncGenerator: | ||
| pass | ||
|
|
||
| async def handleNonStream(self, ctx: ServeContext[ChatCompletionRequest | CompletionRequest]) -> Any: |
| else: | ||
| return await self.handleNonStream(ctx) | ||
|
|
||
| async def handleStream(self, ctx: ServeContext) -> Union[AsyncGenerator, ErrorResponse]: |
There was a problem hiding this comment.
Pull request overview
This PR refactors the API server to use AsyncLLM for engine interaction, introducing a new v1 module with base classes for chat and completion handlers. The architecture becomes: (apiserver, async_llm) → (EngineService) → (work_process) when AsyncLLM is enabled via FD_ENABLE_ASYNC_LLM=1 environment variable.
Key Changes:
- Introduces new v1 serving classes (OpenAiServingBase, OpenAIServingChat, OpenAIServingCompletion) that work with AsyncLLM
- Refactors serving_engine.py to remove engine_client dependency from base class and add semaphore management
- Adds accumulate method to RequestOutput for handling multi-part responses
- Implements add methods in UsageInfo classes for accumulating token usage statistics
- Adds comprehensive unit tests achieving 80%+ coverage
Reviewed changes
Copilot reviewed 17 out of 18 changed files in this pull request and generated 13 comments.
Show a summary per file
| File | Description |
|---|---|
| tests/input/test_ernie4_5_processor.py | Fixes test assertions to use delta_text and reasoning_content correctly |
| tests/entrypoints/openai/v1/test_serving_completion_v1.py | New comprehensive test suite for OpenAIServingCompletion with AsyncLLM |
| tests/entrypoints/openai/v1/test_serving_chat_v1.py | New test suite for OpenAIServingChat covering streaming and non-streaming cases |
| tests/entrypoints/openai/v1/test_serving_base_v1.py | Tests for base class functionality including request handling and accumulation |
| tests/engine/test_request_output.py | Tests for RequestOutput initialization, accumulation, and serialization |
| fastdeploy/input/ernie4_5_processor.py | Adds reasoning_content and text field population for streaming responses |
| fastdeploy/envs.py | Adds FD_ENABLE_ASYNC_LLM environment variable to toggle AsyncLLM usage |
| fastdeploy/entrypoints/openai/v1/serving_completion.py | New completion handler implementation using AsyncLLM |
| fastdeploy/entrypoints/openai/v1/serving_chat.py | New chat handler implementation with AsyncLLM support |
| fastdeploy/entrypoints/openai/v1/serving_base.py | Base class providing common functionality for streaming and non-streaming requests |
| fastdeploy/entrypoints/openai/serving_reward.py | Updates _build_response signature to accept request_output parameter |
| fastdeploy/entrypoints/openai/serving_engine.py | Refactors to separate engine_client from base class, adds semaphore management |
| fastdeploy/entrypoints/openai/serving_embedding.py | Updates _build_response signature for consistency |
| fastdeploy/entrypoints/openai/protocol.py | Adds add methods to usage info classes and recover_stop finish reason |
| fastdeploy/entrypoints/openai/api_server.py | Integrates AsyncLLM-based handlers conditionally based on environment variable |
| fastdeploy/engine/request.py | Adds accumulate method and accumulate_tool_calls field to RequestOutput |
| fastdeploy/engine/async_llm.py | Increases engine ready wait time from 180s to 500s |
| def _wait_engine_ready(self) -> bool: | ||
| """Wait for engine and workers to be fully ready""" | ||
| max_wait_time = 180 # seconds | ||
| max_wait_time = 500 # seconds |
There was a problem hiding this comment.
Increasing the max_wait_time from 180 seconds to 500 seconds (nearly 3x) is a significant change that appears unrelated to the main refactoring. This change:
- Could hide underlying initialization issues that should be fixed instead
- Lacks justification in the PR description
- May indicate a performance regression
If the engine initialization truly requires this much time, there should be detailed logging to explain why, or the initialization process should be optimized. Consider reverting this change or providing clear justification.
| raise ValueError(f"Error processing response: {str(e)}") | ||
| finally: | ||
| await self.engine_client.connection_manager.cleanup_request(request_id) | ||
| self.engine_client.connection_manager.cleanup_request(request_id) |
There was a problem hiding this comment.
The removal of 'await' before 'self.engine_client.connection_manager.cleanup_request(request_id)' is incorrect. The cleanup_request method is defined as async (see fastdeploy/entrypoints/openai/utils.py line 190). Removing await means:
- The cleanup coroutine will not be executed
- The request won't be properly cleaned up from request_map and request_num
- This could lead to memory leaks as requests accumulate
This must be kept as 'await self.engine_client.connection_manager.cleanup_request(request_id)'.
| self.engine_client.connection_manager.cleanup_request(request_id) | |
| await self.engine_client.connection_manager.cleanup_request(request_id) |
| response_dict["outputs"]["reasoning_content"] = reasoning_content | ||
| response_dict["outputs"]["text"] = ( | ||
| reasoning_delta_message.content or "" | ||
| if reasoning_delta_message and hasattr(reasoning_delta_message, "content") | ||
| else "" | ||
| ) | ||
| else: | ||
| response_dict["outputs"]["text"] = delta_text |
There was a problem hiding this comment.
The logic for setting response_dict["outputs"]["text"] has been moved inside the reasoning_content conditional block, but the else clause at line 389-390 only handles the non-reasoning case. However, the original code at line 407 set text unconditionally after all processing. This change means that when reasoning_content is present, text is set from reasoning_delta_message.content, which may not always match delta_text. Please verify this is the intended behavior and that all scenarios are covered:
- When reasoning_content exists and has content attribute
- When reasoning_content exists but doesn't have content attribute
- When reasoning_content doesn't exist (handled by else)
Consider whether the text assignment should remain after line 406 to handle edge cases.
| def add(self, other: CompletionTokenUsageInfo): | ||
| if self.reasoning_tokens and other.reasoning_tokens: | ||
| self.reasoning_tokens += other.reasoning_tokens | ||
| elif other.reasoning_tokens: | ||
| self.reasoning_tokens = other.reasoning_tokens | ||
| if self.image_tokens and other.image_tokens: | ||
| self.image_tokens += other.image_tokens | ||
| elif other.image_tokens: | ||
| self.image_tokens = other.image_tokens |
There was a problem hiding this comment.
The add method in CompletionTokenUsageInfo has inconsistent logic. When self.reasoning_tokens is None and other.reasoning_tokens exists, it correctly sets self.reasoning_tokens. However, when self.reasoning_tokens is 0 (which is falsy) and other.reasoning_tokens is non-zero, the addition won't happen because 'if self.reasoning_tokens' evaluates to False.
This should be:
if self.reasoning_tokens is not None and other.reasoning_tokens is not None:
self.reasoning_tokens += other.reasoning_tokens
elif other.reasoning_tokens is not None:
self.reasoning_tokens = other.reasoning_tokensThe same issue exists for image_tokens and applies to PromptTokenUsageInfo.add as well (lines 99-113).
| def _check_master(self) -> bool: | ||
| """Check if current node is master""" | ||
| return self.engine_client.is_master | ||
| return True |
There was a problem hiding this comment.
The _check_master method now always returns True, which changes the behavior for non-master nodes. In a distributed setup where multiple nodes are running, this could cause all nodes to think they are master. The previous implementation checked 'self.engine_client.is_master', which was more accurate.
Note that ZmqOpenAIServing overrides this at line 323 to preserve the original behavior, but the base class default should probably not be True for all cases. Consider if this is the intended behavior or if it should throw NotImplementedError to force subclasses to implement it.
| else: | ||
| engine = LLMEngine.from_engine_args(engine_args) | ||
| started = False | ||
| if inspect.iscoroutinefunction(engine.start): |
There was a problem hiding this comment.
Using inspect.iscoroutinefunction to determine if engine.start needs to be awaited is fragile. This approach:
- Makes the code harder to understand and maintain
- Couples the API server to the implementation details of different engine types
- Could fail silently if a third engine type is added later
Consider using a more explicit pattern such as:
- Define a common interface/protocol for engines
- Use explicit type checking (isinstance) instead of runtime inspection
- Have engines implement a standard async start method
This would make the code more maintainable and less prone to errors.
| if inspect.iscoroutinefunction(engine.start): | |
| if isinstance(engine, AsyncLLM): |
|
|
||
| MAX_CONCURRENT_CONNECTIONS = (args.max_concurrency + args.workers - 1) // args.workers | ||
| connection_semaphore = StatefulSemaphore(MAX_CONCURRENT_CONNECTIONS) | ||
|
|
There was a problem hiding this comment.
The new environment variable FD_ENABLE_ASYNC_LLM (line 98) should be documented. Add a comment explaining:
- What it enables/disables
- When to use it (e.g., "Set to 1 to use AsyncLLM-based architecture instead of traditional LLMEngine")
- Any performance or compatibility implications
- Default value (currently 0)
This helps future developers and operators understand the purpose of this flag.
| # FD_ENABLE_ASYNC_LLM: | |
| # - Set to "1" to enable the AsyncLLM-based engine architecture instead of the | |
| # traditional LLMEngine used by default. | |
| # - When enabled, `AsyncLLM` is used to start the engine, which may provide | |
| # better concurrency characteristics but can have different performance | |
| # and compatibility behavior compared to `LLMEngine`. | |
| # - Set to "0" or leave unset to keep using the traditional `LLMEngine`. | |
| # - Default: "0" (AsyncLLM disabled) if the environment variable is not set. |
| if next_output.prompt: | ||
| self.prompt = next_output.prompt | ||
| if next_output.prompt_token_ids: |
There was a problem hiding this comment.
The condition 'if next_output.prompt:' on line 658 will skip updating self.prompt when next_output.prompt is an empty string, which is a valid value. This should check 'if next_output.prompt is not None:' instead. The same issue exists for prompt_token_ids on line 660.
| if next_output.prompt: | |
| self.prompt = next_output.prompt | |
| if next_output.prompt_token_ids: | |
| if next_output.prompt is not None: | |
| self.prompt = next_output.prompt | |
| if next_output.prompt_token_ids is not None: |
|
|
||
| send_idx = output.send_idx | ||
|
|
||
| max_streaming_response_tokens = max(1, max_streaming_response_tokens) |
There was a problem hiding this comment.
Variable max_streaming_response_tokens is not used.
| MAX_CONCURRENT_CONNECTIONS = (args.max_concurrency + args.workers - 1) // args.workers | ||
| connection_semaphore = StatefulSemaphore(MAX_CONCURRENT_CONNECTIONS) | ||
|
|
||
| enable_async_llm = environment_variables.get("FD_ENABLE_ASYNC_LLM")() |
There was a problem hiding this comment.
这里按其它地方调用就直接是env.xxx即可，可以统一下调用方式
d1f0306 to
042f1ae
Compare
0c2ef55 to
bc5a0d0
Compare
GitHub Actions
Error
GitHub Actions
Failure
…e classes and use AsyncLLM (PaddlePaddle#5195) * [Optimization] refactor(chat_handler,completion_handler): extract base classes and use AsyncLLM * [Optimization] refactor(chat_handler,completion_handler): rename class
…e classes and use AsyncLLM (PaddlePaddle#5195) * [Optimization] refactor(chat_handler,completion_handler): extract base classes and use AsyncLLM * [Optimization] refactor(chat_handler,completion_handler): rename class
…e classes and use AsyncLLM (PaddlePaddle#5195) * [Optimization] refactor(chat_handler,completion_handler): extract base classes and use AsyncLLM * [Optimization] refactor(chat_handler,completion_handler): rename class
Motivation
重构api_server。
基于AsyncLLM与Engine进行交互
当使用AsyncLLM时，由EngineService启动work_process，架构变为：
（apiserver、 async_llm） → （EngineService）→（work_process）三个进程
Modifications
重构api_server。
基于AsyncLLM与Engine进行交互
Usage or Command
通过设置env FD_ENABLE_ASYNC_LLM = 1 开启
Accuracy Tests
通过单测，单测覆盖率已达到80%以上
Checklist
[FDConfig],[APIServer],[Engine],[Scheduler],[PD Disaggregation],[Executor],[Graph Optimization],[Speculative Decoding],[RL],[Models],[Quantization],[Loader],[OP],[KVCache],[DataProcessor],[BugFix],[Docs],[CI],[Optimization],[Feature],[Benchmark],[Others],[XPU],[HPU],[GCU],[DCU],[Iluvatar],[Metax]]pre-commitbefore commit.releasebranch, make sure the PR has been submitted to thedevelopbranch, then cherry-pick it to thereleasebranch with the[Cherry-Pick]PR tag.