feat: Complete vLLM-omni implementation with conda environment setup
- Add comprehensive PRD, architecture design, and test design documents - Implement core modules: OmniLLM, AsyncOmniLLM, stage configurations - Add DiT scheduler and cache manager for diffusion models - Implement CLI integration with --omni flag support - Add API server and plugin system for vLLM integration - Create comprehensive test suite with fixtures - Update dependencies to vLLM 0.10.2 and PyTorch 2.8.0 - Add conda environment setup and package installation - Implement stage-based processing architecture - Add multimodal output processing capabilities This commit establishes the foundation for multi-modality models inference and serving with non-autoregressive structures.
This commit is contained in:
1 parent
c1da072744
commit
51d2570464
25 files changed
+4065
-214
No files matched your search
@@ -71,39 +71,6 @@ from typing import List, Optional, Dict, Any
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
|
||||
class SchedulerType(Enum):
|
||||
"""Types of schedulers available."""
|
||||
FIFO = "fifo"
|
||||
PRIORITY = "priority"
|
||||
MULTIMODAL = "multimodal"
|
||||
DIFFUSION = "diffusion"
|
||||
|
||||
@dataclass
|
||||
class SchedulerConfig:
|
||||
"""Configuration for scheduler."""
|
||||
scheduler_type: SchedulerType = SchedulerType.FIFO
|
||||
max_queue_size: int = 1000
|
||||
priority_weights: Dict[str, float] = None
|
||||
timeout_seconds: int = 300
|
||||
|
||||
class BaseScheduler(ABC):
|
||||
"""Abstract base class for all schedulers."""
|
||||
|
||||
@abstractmethod
|
||||
async def schedule_request(self, request: OmniRequest) -> bool:
|
||||
"""Schedule a request for processing."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
async def get_next_request(self) -> Optional[OmniRequest]:
|
||||
"""Get the next request to process."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
async def remove_request(self, request_id: str) -> bool:
|
||||
"""Remove a request from the queue."""
|
||||
pass
|
||||
|
||||
class DiTCacheManager:
|
||||
"""Manages DiT cache for diffusion models."""
|
||||
|
||||
@@ -129,7 +96,7 @@ class DiTCacheManager:
|
||||
|
||||
#### Initialization
|
||||
```python
|
||||
class CoreScheduler:
|
||||
class OmniScheduler():
|
||||
def __init__(self, config: SchedulerConfig):
|
||||
"""Initialize the core scheduler."""
|
||||
self.config = config
|
||||
@@ -148,7 +115,7 @@ class CoreScheduler:
|
||||
|
||||
#### Core Operations
|
||||
```python
|
||||
async def schedule_request(self, request: OmniRequest) -> bool:
|
||||
async def schedule(self, request: OmniRequest) -> bool:
|
||||
"""
|
||||
Schedule a request for processing.
|
||||
|
||||
@@ -163,33 +130,6 @@ class CoreScheduler:
|
||||
QueueFullError: If queue is at capacity
|
||||
"""
|
||||
pass
|
||||
|
||||
async def get_next_request(self) -> Optional[OmniRequest]:
|
||||
"""
|
||||
Get the next request to process.
|
||||
|
||||
Returns:
|
||||
OmniRequest or None: Next request to process
|
||||
|
||||
Raises:
|
||||
SchedulerError: If retrieval fails
|
||||
"""
|
||||
pass
|
||||
|
||||
async def process_request(self, request: OmniRequest) -> Any:
|
||||
"""
|
||||
Process a request through the appropriate pipeline.
|
||||
|
||||
Args:
|
||||
request: The request to process
|
||||
|
||||
Returns:
|
||||
Any: Processing result
|
||||
|
||||
Raises:
|
||||
ProcessingError: If processing fails
|
||||
"""
|
||||
pass
|
||||
```
|
||||
|
||||
#### Configuration
|
||||
@@ -240,7 +180,7 @@ class CoreScheduler:
|
||||
|
||||
```python
|
||||
@dataclass
|
||||
class CoreModuleConfig:
|
||||
class OmniConfig:
|
||||
"""Configuration for the core module."""
|
||||
|
||||
# Scheduler configuration
|
||||
|
||||
Reference in new issue
Block a user