40. vLLM 启动流程源码精读：从配置加载到 Worker 初始化

# vllm/config.py
import argparse
import yaml
import os
from typing import Dict, Any, Optional

class Config:
    def __init__(self):
        self.config = {}
        self.default_config = {
            "model": None,
            "tensor_parallel_size": 1,
            "gpu_memory_utilization": 0.9,
            "max_num_seqs": 256,
            "max_seq_len": 2048,
            "quantization": None,
            "dtype": "float16",
            "trust_remote_code": False,
        }
    
    def load_from_file(self, file_path: str) -> Dict[str, Any]:
        """从文件加载配置"""
        if not os.path.exists(file_path):
            raise FileNotFoundError(f"Config file {file_path} not found")
        
        with open(file_path, "r") as f:
            if file_path.endswith(".yaml") or file_path.endswith(".yml"):
                file_config = yaml.safe_load(f)
            elif file_path.endswith(".json"):
                import json
                file_config = json.load(f)
            else:
                raise ValueError(f"Unsupported config file format: {file_path}")
        
        return file_config
    
    def load_from_env(self) -> Dict[str, Any]:
        """从环境变量加载配置"""
        env_config = {}
        prefix = "VLLM_"
        
        for key, value in os.environ.items():
            if key.startswith(prefix):
                config_key = key[len(prefix):].lower()
                # 转换值类型
                if value.lower() in ["true", "false"]:
                    env_config[config_key] = value.lower() == "true"
                elif value.isdigit():
                    env_config[config_key] = int(value)
                elif value.replace(".", "", 1).isdigit():
                    env_config[config_key] = float(value)
                else:
                    env_config[config_key] = value
        
        return env_config
    
    def merge_configs(self, configs: List[Dict[str, Any]]) -> Dict[str, Any]:
        """合并多个配置"""
        merged = self.default_config.copy()
        
        for config in configs:
            if config:
                merged.update(config)
        
        return merged
    
    def validate(self, config: Dict[str, Any]) -> bool:
        """验证配置"""
        # 检查必填项
        required_fields = ["model"]
        for field in required_fields:
            if field not in config or config[field] is None:
                raise ValueError(f"Required config field {field} is missing")
        
        # 检查tensor_parallel_size
        if config["tensor_parallel_size"] < 1:
            raise ValueError(f"tensor_parallel_size must be >= 1, got {config['tensor_parallel_size']}")
        
        # 检查gpu_memory_utilization
        if not 0 < config["gpu_memory_utilization"] <= 1:
            raise ValueError(f"gpu_memory_utilization must be in (0, 1], got {config['gpu_memory_utilization']}")
        
        # 检查dtype
        valid_dtypes = ["float16", "bfloat16", "float32"]
        if config["dtype"] not in valid_dtypes:
            raise ValueError(f"Invalid dtype {config['dtype']}, must be one of {valid_dtypes}")
        
        return True
    
    def load(self, args: Optional[argparse.Namespace] = None, config_file: Optional[str] = None) -> Dict[str, Any]:
        """加载并验证配置"""
        # 从命令行参数加载
        cli_config = vars(args) if args else {}
        
        # 从文件加载
        file_config = self.load_from_file(config_file) if config_file else {}
        
        # 从环境变量加载
        env_config = self.load_from_env()
        
        # 合并配置（优先级：CLI > 文件 > 环境变量 > 默认配置）
        merged_config = self.merge_configs([env_config, file_config, cli_config])
        
        # 验证配置
        self.validate(merged_config)
        
        return merged_config

# vllm/utils.py
import torch
import os
import subprocess
from typing import Dict, List

class HardwareDetector:
    @staticmethod
    def get_gpu_count() -> int:
        """获取GPU数量"""
        if not torch.cuda.is_available():
            return 0
        return torch.cuda.device_count()
    
    @staticmethod
    def get_gpu_info() -> List[Dict[str, Any]]:
        """获取GPU详细信息"""
        if not torch.cuda.is_available():
            return []
        
        gpu_info = []
        for i in range(torch.cuda.device_count()):
            props = torch.cuda.get_device_properties(i)
            gpu_info.append({
                "id": i,
                "name": props.name,
                "total_memory": props.total_memory,
                "free_memory": torch.cuda.get_device_properties(i).total_memory - torch.cuda.memory_allocated(i),
                "compute_capability": f"{props.major}.{props.minor}",
                "multi_processor_count": props.multi_processor_count,
            })
        
        return gpu_info
    
    @staticmethod
    def get_cpu_info() -> Dict[str, Any]:
        """获取CPU信息"""
        # 获取CPU核心数
        if os.name == "posix":
            # Linux/macOS
            cpu_count = os.cpu_count()
            # 尝试获取物理核心数
            try:
                physical_cores = int(subprocess.check_output(["lscpu", "--parse=Core(s)_per_socket,Socket(s)"]).decode().split("\n")[1])
            except:
                physical_cores = cpu_count
        else:
            # Windows
            cpu_count = os.cpu_count()
            physical_cores = cpu_count
        
        return {
            "total_cores": cpu_count,
            "physical_cores": physical_cores,
        }
    
    @staticmethod
    def get_memory_info() -> Dict[str, Any]:
        """获取内存信息"""
        if os.name == "posix":
            # Linux/macOS
            with open("/proc/meminfo", "r") as f:
                meminfo = f.read()
            
            total_memory = int([line for line in meminfo.split("\n") if "MemTotal" in line][0].split()[1]) * 1024
            free_memory = int([line for line in meminfo.split("\n") if "MemFree" in line][0].split()[1]) * 1024
        else:
            # Windows
            import psutil
            mem = psutil.virtual_memory()
            total_memory = mem.total
            free_memory = mem.available
        
        return {
            "total_memory": total_memory,
            "free_memory": free_memory,
        }
    
    @staticmethod
    def get_cuda_version() -> str:
        """获取CUDA版本"""
        if not torch.cuda.is_available():
            return "N/A"
        return torch.version.cuda
    
    @staticmethod
    def get_nccl_version() -> str:
        """获取NCCL版本"""
        if not torch.cuda.is_available():
            return "N/A"
        
        try:
            import nccl
            return nccl.version()
        except:
            # 尝试从环境变量获取
            nccl_version = os.environ.get("NCCL_VERSION", "N/A")
            return nccl_version
    
    @classmethod
    def detect(cls) -> Dict[str, Any]:
        """检测所有硬件信息"""
        return {
            "gpu_count": cls.get_gpu_count(),
            "gpus": cls.get_gpu_info(),
            "cpu": cls.get_cpu_info(),
            "memory": cls.get_memory_info(),
            "cuda_version": cls.get_cuda_version(),
            "nccl_version": cls.get_nccl_version(),
        }
    
    @classmethod
    def print_hardware_report(cls):
        """打印硬件报告"""
        hardware_info = cls.detect()
        
        print("=" * 60)
        print("Hardware Report")
        print("=" * 60)
        
        print(f"CUDA Version: {hardware_info['cuda_version']}")
        print(f"NCCL Version: {hardware_info['nccl_version']}")
        print()
        
        print(f"CPU: {hardware_info['cpu']['physical_cores']} physical cores, {hardware_info['cpu']['total_cores']} total cores")
        print(f"Memory: {hardware_info['memory']['total_memory'] / 1024**3:.2f} GB total, {hardware_info['memory']['free_memory'] / 1024**3:.2f} GB free")
        print()
        
        print(f"GPUs: {hardware_info['gpu_count']} devices")
        for gpu in hardware_info['gpus']:
            print(f"  GPU {gpu['id']}: {gpu['name']}")
            print(f"    Compute Capability: {gpu['compute_capability']}")
            print(f"    Memory: {gpu['total_memory'] / 1024**3:.2f} GB total, {gpu['free_memory'] / 1024**3:.2f} GB free")
            print(f"    Multi-processors: {gpu['multi_processor_count']}")
        print("=" * 60)

# vllm/model_loader.py
import torch
import os
import json
from typing import Dict, List, Optional, Any
from transformers import AutoConfig, AutoModelForCausalLM, AutoTokenizer

class ModelLoader:
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.model = None
        self.tokenizer = None
    
    def load_model_config(self) -> AutoConfig:
        """加载模型配置"""
        print(f"Loading model config from {self.config['model']}")
        
        config = AutoConfig.from_pretrained(
            self.config['model'],
            trust_remote_code=self.config.get('trust_remote_code', False),
            # 传递额外配置
            **self.config.get('model_config', {})
        )
        
        return config
    
    def load_tokenizer(self) -> AutoTokenizer:
        """加载tokenizer"""
        print(f"Loading tokenizer from {self.config['model']}")
        
        tokenizer = AutoTokenizer.from_pretrained(
            self.config['model'],
            trust_remote_code=self.config.get('trust_remote_code', False),
            # 传递额外配置
            **self.config.get('tokenizer_config', {})
        )
        
        # 设置pad_token（如果没有的话）
        if tokenizer.pad_token is None:
            tokenizer.pad_token = tokenizer.eos_token
        
        self.tokenizer = tokenizer
        return tokenizer
    
    def initialize_model(self, config: AutoConfig) -> torch.nn.Module:
        """初始化模型结构"""
        print(f"Initializing model with dtype={self.config['dtype']}")
        
        # 设置模型设备
        device = torch.device(f"cuda:{self.config.get('device', 0)}") if torch.cuda.is_available() else torch.device("cpu")
        
        # 根据dtype创建模型
        torch_dtype = getattr(torch, self.config['dtype'])
        
        # 初始化模型
        model = AutoModelForCausalLM.from_config(
            config,
            trust_remote_code=self.config.get('trust_remote_code', False),
            torch_dtype=torch_dtype,
        )
        
        model.to(device)
        return model
    
    def locate_weight_files(self) -> List[str]:
        """定位权重文件"""
        model_path = self.config['model']
        
        # 检查是否为本地路径
        if os.path.exists(model_path):
            # 本地模型
            weight_files = []
            # 查找所有权重文件
            for root, dirs, files in os.walk(model_path):
                for file in files:
                    if file.endswith(".bin") or file.endswith(".pt") or file.endswith(".safetensors"):
                        weight_files.append(os.path.join(root, file))
        else:
            # 远程模型，使用transformers自动下载
            weight_files = []
        
        return weight_files
    
    def load_weights(self, model: torch.nn.Module) -> torch.nn.Module:
        """加载模型权重"""
        print(f"Loading weights for model {self.config['model']}")
        
        # 使用transformers加载权重
        model = AutoModelForCausalLM.from_pretrained(
            self.config['model'],
            config=model.config,
            trust_remote_code=self.config.get('trust_remote_code', False),
            torch_dtype=getattr(torch, self.config['dtype']),
            low_cpu_mem_usage=True,
            # 量化配置
            load_in_8bit=self.config.get('quantization') == '8bit',
            load_in_4bit=self.config.get('quantization') == '4bit',
            # 分布式配置
            device_map=self.config.get('device_map', 'auto'),
            **self.config.get('from_pretrained_kwargs', {})
        )
        
        return model
    
    def verify_model(self, model: torch.nn.Module) -> bool:
        """验证模型"""
        print("Verifying model...")
        
        # 简单的前向传播测试
        try:
            # 创建测试输入
            test_input = torch.tensor([[1, 2, 3, 4, 5]], device=next(model.parameters()).device)
            # 前向传播
            with torch.no_grad():
                outputs = model(test_input)
            # 检查输出
            assert outputs.logits is not None, "Model forward pass failed: no logits returned"
            print("Model verification passed!")
            return True
        except Exception as e:
            print(f"Model verification failed: {e}")
            return False
    
    def load(self) -> Dict[str, Any]:
        """完整加载流程"""
        print("=" * 60)
        print("Model Loading Process")
        print("=" * 60)
        
        # 1. 加载tokenizer
        tokenizer = self.load_tokenizer()
        
        # 2. 加载模型配置
        model_config = self.load_model_config()
        
        # 3. 定位权重文件
        weight_files = self.locate_weight_files()
        print(f"Found {len(weight_files)} weight files")
        
        # 4. 加载模型权重
        model = self.load_weights(None)
        
        # 5. 验证模型
        self.verify_model(model)
        
        print("=" * 60)
        print("Model Loading Complete")
        print("=" * 60)
        
        return {
            "model": model,
            "tokenizer": tokenizer,
            "model_config": model_config,
        }

# vllm/worker.py
import torch
import multiprocessing
import os
from typing import Dict, List, Any

class Worker:
    def __init__(self, worker_id: int, config: Dict[str, Any]):
        self.worker_id = worker_id
        self.config = config
        self.model = None
        self.tokenizer = None
        self.device = None
    
    def init_gpu(self):
        """初始化GPU设备"""
        if torch.cuda.is_available():
            # 设置GPU设备
            self.device = torch.device(f"cuda:{self.worker_id}")
            torch.cuda.set_device(self.device)
            # 初始化CUDA上下文
            torch.cuda.init()
            print(f"Worker {self.worker_id}: GPU {self.worker_id} initialized")
        else:
            self.device = torch.device("cpu")
            print(f"Worker {self.worker_id}: Using CPU")
    
    def load_model_shard(self):
        """加载模型分片"""
        from vllm.model_loader import ModelLoader
        
        # 为当前Worker创建专用配置
        worker_config = self.config.copy()
        worker_config['device'] = self.worker_id
        
        # 加载模型
        model_loader = ModelLoader(worker_config)
        loaded = model_loader.load()
        
        self.model = loaded['model']
        self.tokenizer = loaded['tokenizer']
        
        print(f"Worker {self.worker_id}: Model shard loaded successfully")
    
    def init_communication(self):
        """初始化通信"""
        # 在分布式环境中，初始化NCCL等通信库
        if self.config.get('tensor_parallel_size', 1) > 1:
            # 初始化分布式通信
            import torch.distributed as dist
            
            # 设置环境变量
            os.environ['MASTER_ADDR'] = self.config.get('master_addr', 'localhost')
            os.environ['MASTER_PORT'] = self.config.get('master_port', '29500')
            os.environ['RANK'] = str(self.worker_id)
            os.environ['WORLD_SIZE'] = str(self.config['tensor_parallel_size'])
            
            # 初始化分布式通信
            dist.init_process_group(
                backend='nccl',
                rank=self.worker_id,
                world_size=self.config['tensor_parallel_size'],
            )
            
            print(f"Worker {self.worker_id}: Distributed communication initialized")
    
    def run(self):
        """运行Worker"""
        try:
            print(f"Starting Worker {self.worker_id}...")
            
            # 1. 初始化GPU
            self.init_gpu()
            
            # 2. 加载模型分片
            self.load_model_shard()
            
            # 3. 初始化通信
            self.init_communication()
            
            print(f"Worker {self.worker_id} initialized successfully")
            
            # 4. 等待主进程命令
            self.wait_for_commands()
            
        except Exception as e:
            print(f"Worker {self.worker_id} failed: {e}")
            import traceback
            traceback.print_exc()
    
    def wait_for_commands(self):
        """等待主进程命令"""
        # 简化实现，实际会使用队列或其他通信机制
        import time
        while True:
            time.sleep(1)

class WorkerManager:
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.workers = []
        self.worker_processes = []
    
    def calculate_worker_count(self) -> int:
        """计算Worker数量"""
        tensor_parallel_size = self.config.get('tensor_parallel_size', 1)
        return tensor_parallel_size
    
    def create_workers(self):
        """创建Worker"""
        worker_count = self.calculate_worker_count()
        print(f"Creating {worker_count} workers...")
        
        for worker_id in range(worker_count):
            # 创建Worker配置
            worker_config = self.config.copy()
            worker_config['worker_id'] = worker_id
            
            # 创建Worker进程
            if self.config.get('distributed', False):
                # 分布式模式下，使用multiprocessing
                worker_process = multiprocessing.Process(
                    target=self._run_worker,
                    args=(worker_config,)
                )
                self.worker_processes.append(worker_process)
                worker_process.start()
            else:
                # 单进程模式下，直接创建Worker对象
                worker = Worker(worker_id, worker_config)
                self.workers.append(worker)
                # 在当前进程中运行Worker
                worker.run()
    
    @staticmethod
    def _run_worker(config: Dict[str, Any]):
        """Worker进程入口"""
        worker = Worker(config['worker_id'], config)
        worker.run()
    
    def wait_for_workers(self):
        """等待所有Worker初始化完成"""
        if self.config.get('distributed', False):
            # 等待所有Worker进程启动
            for worker_process in self.worker_processes:
                worker_process.join()
    
    def shutdown(self):
        """关闭所有Worker"""
        for worker_process in self.worker_processes:
            if worker_process.is_alive():
                worker_process.terminate()
                worker_process.join()
    
    def start(self):
        """启动Worker管理器"""
        print("=" * 60)
        print("Worker Initialization")
        print("=" * 60)
        
        # 创建Worker
        self.create_workers()
        
        # 等待Worker初始化完成
        self.wait_for_workers()
        
        print("=" * 60)
        print("Worker Initialization Complete")
        print("=" * 60)

# vllm/communication.py
import torch.distributed as dist
import os
import time
from typing import Dict, List, Any

class CommunicationManager:
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.rank = config.get('rank', 0)
        self.world_size = config.get('world_size', 1)
        self.backend = config.get('backend', 'nccl')
        self.master_addr = config.get('master_addr', 'localhost')
        self.master_port = config.get('master_port', '29500')
    
    def init_environment(self):
        """初始化环境变量"""
        os.environ['MASTER_ADDR'] = self.master_addr
        os.environ['MASTER_PORT'] = self.master_port
        os.environ['RANK'] = str(self.rank)
        os.environ['WORLD_SIZE'] = str(self.world_size)
    
    def init_communication(self):
        """初始化分布式通信"""
        if self.world_size <= 1:
            print("Single process mode, skipping communication initialization")
            return
        
        print(f"Initializing communication with backend={self.backend}, rank={self.rank}, world_size={self.world_size}")
        
        # 初始化分布式通信
        dist.init_process_group(
            backend=self.backend,
            rank=self.rank,
            world_size=self.world_size,
        )
        
        print(f"Communication initialized successfully for rank {self.rank}")
    
    def test_communication(self) -> bool:
        """测试通信连接"""
        if self.world_size <= 1:
            return True
        
        print(f"Testing communication for rank {self.rank}...")
        
        try:
            # 测试all_reduce
            tensor = torch.tensor([self.rank], device='cuda' if torch.cuda.is_available() else 'cpu')
            dist.all_reduce(tensor, op=dist.ReduceOp.SUM)
            expected_sum = sum(range(self.world_size))
            if tensor.item() != expected_sum:
                print(f"Communication test failed for rank {self.rank}: expected sum {expected_sum}, got {tensor.item()}")
                return False
            
            print(f"Communication test passed for rank {self.rank}")
            return True
        except Exception as e:
            print(f"Communication test failed for rank {self.rank}: {e}")
            return False
    
    def broadcast_config(self, config: Dict[str, Any]) -> Dict[str, Any]:
        """广播配置到所有节点"""
        if self.world_size <= 1:
            return config
        
        import pickle
        
        if self.rank == 0:
            # 主节点广播配置
            config_bytes = pickle.dumps(config)
            # 先广播大小
            size_tensor = torch.tensor([len(config_bytes)], device='cuda' if torch.cuda.is_available() else 'cpu')
            dist.broadcast(size_tensor, src=0)
            # 再广播数据
            config_tensor = torch.ByteTensor(config_bytes, device='cuda' if torch.cuda.is_available() else 'cpu')
            dist.broadcast(config_tensor, src=0)
            return config
        else:
            # 从节点接收配置
            # 先接收大小
            size_tensor = torch.tensor([0], device='cuda' if torch.cuda.is_available() else 'cpu')
            dist.broadcast(size_tensor, src=0)
            # 再接收数据
            config_tensor = torch.ByteTensor(size_tensor.item(), device='cuda' if torch.cuda.is_available() else 'cpu')
            dist.broadcast(config_tensor, src=0)
            # 反序列化
            config = pickle.loads(config_tensor.cpu().numpy().tobytes())
            return config
    
    def barrier(self):
        """等待所有节点到达屏障"""
        if self.world_size <= 1:
            return
        dist.barrier()
    
    def get_rank(self) -> int:
        """获取当前节点的rank"""
        if self.world_size <= 1:
            return 0
        return dist.get_rank()
    
    def get_world_size(self) -> int:
        """获取世界大小"""
        if self.world_size <= 1:
            return 1
        return dist.get_world_size()
    
    def is_initialized(self) -> bool:
        """检查通信是否已初始化"""
        if self.world_size <= 1:
            return True
        return dist.is_initialized()
    
    def destroy(self):
        """销毁通信资源"""
        if self.world_size > 1 and dist.is_initialized():
            dist.destroy_process_group()
            print(f"Communication destroyed for rank {self.rank}")

# vllm/server.py
import uvicorn
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from typing import Dict, Any

class VLLMServer:
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.app = FastAPI(
            title="vLLM API Server",
            description="High-throughput LLM inference server",
            version="0.4.0",
        )
        
        # 添加CORS中间件
        self.app.add_middleware(
            CORSMiddleware,
            allow_origins=["*"],
            allow_credentials=True,
            allow_methods=["*"],
            allow_headers=["*"],
        )
        
        # 初始化路由
        self.setup_routes()
    
    def setup_routes(self):
        """设置API路由"""
        @self.app.get("/")
        async def root():
            return {"message": "vLLM API Server is running"}
        
        @self.app.post("/generate")
        async def generate(request: Dict[str, Any]):
            # 简化实现，实际会调用Worker进行推理
            return {
                "id": "req-123",
                "object": "text_completion",
                "created": int(time.time()),
                "model": self.config["model"],
                "choices": [
                    {
                        "text": "Hello, world! This is a test response from vLLM.",
                        "index": 0,
                        "logprobs": None,
                        "finish_reason": "stop"
                    }
                ],
                "usage": {
                    "prompt_tokens": 10,
                    "completion_tokens": 15,
                    "total_tokens": 25
                }
            }
    
    def check_health(self) -> bool:
        """检查系统健康状态"""
        print("Checking system health...")
        
        # 检查模型是否加载成功
        if not hasattr(self, 'model') or self.model is None:
            print("Health check failed: model not loaded")
            return False
        
        # 检查Worker状态
        if hasattr(self, 'worker_manager') and self.worker_manager is not None:
            # 简化实现，实际会检查Worker状态
            pass
        
        print("System health check passed")
        return True
    
    def bind_port(self) -> bool:
        """绑定服务端口"""
        port = self.config.get('port', 8000)
        host = self.config.get('host', '0.0.0.0')
        
        print(f"Binding to {host}:{port}...")
        
        # 简化实现，实际会由uvicorn处理
        return True
    
    def start(self):
        """启动服务"""
        print("=" * 60)
        print("vLLM Server Startup")
        print("=" * 60)
        
        # 检查健康状态
        if not self.check_health():
            print("Server startup failed: health check failed")
            return
        
        # 绑定端口
        if not self.bind_port():
            print("Server startup failed: port binding failed")
            return
        
        # 启动UVicorn服务器
        port = self.config.get('port', 8000)
        host = self.config.get('host', '0.0.0.0')
        workers = self.config.get('workers', 1)
        
        print(f"Starting vLLM server on {host}:{port} with {workers} workers")
        print("=" * 60)
        
        # 启动服务器
        uvicorn.run(
            self.app,
            host=host,
            port=port,
            workers=workers,
            **self.config.get('uvicorn_kwargs', {})
        )

# vllm/main.py
import argparse
import time
from typing import Dict, Any

from vllm.config import Config
from vllm.hardware_detector import HardwareDetector
from vllm.model_loader import ModelLoader
from vllm.worker_manager import WorkerManager
from vllm.communication import CommunicationManager
from vllm.server import VLLMServer

class VLLM:
    def __init__(self):
        self.config = None
        self.hardware_info = None
        self.model = None
        self.tokenizer = None
        self.worker_manager = None
        self.communication_manager = None
        self.server = None
    
    def parse_args(self) -> argparse.Namespace:
        """解析命令行参数"""
        parser = argparse.ArgumentParser(description="vLLM: A high-throughput LLM inference and serving engine")
        
        # 模型配置
        parser.add_argument("--model", type=str, required=True, help="Model name or path")
        parser.add_argument("--tensor-parallel-size", type=int, default=1, help="Number of GPUs to use for tensor parallelism")
        parser.add_argument("--dtype", type=str, default="float16", choices=["float16", "bfloat16", "float32"], help="Data type for model weights")
        parser.add_argument("--quantization", type=str, default=None, choices=["8bit", "4bit"], help="Quantization method")
        
        # 服务器配置
        parser.add_argument("--host", type=str, default="0.0.0.0", help="Host to listen on")
        parser.add_argument("--port", type=int, default=8000, help="Port to listen on")
        parser.add_argument("--workers", type=int, default=1, help="Number of server workers")
        
        # 分布式配置
        parser.add_argument("--distributed", action="store_true", help="Enable distributed mode")
        parser.add_argument("--master-addr", type=str, default="localhost", help="Master address for distributed training")
        parser.add_argument("--master-port", type=str, default="29500", help="Master port for distributed training")
        
        # 其他配置
        parser.add_argument("--config-file", type=str, default=None, help="Path to config file")
        parser.add_argument("--trust-remote-code", action="store_true", help="Trust remote code from model")
        parser.add_argument("--gpu-memory-utilization", type=float, default=0.9, help="GPU memory utilization ratio")
        
        return parser.parse_args()
    
    def load_config(self, args: argparse.Namespace) -> Dict[str, Any]:
        """加载配置"""
        config_loader = Config()
        return config_loader.load(args, args.config_file)
    
    def detect_hardware(self) -> Dict[str, Any]:
        """检测硬件环境"""
        hardware_detector = HardwareDetector()
        hardware_info = hardware_detector.detect()
        hardware_detector.print_hardware_report()
        return hardware_info
    
    def initialize_components(self):
        """初始化各个组件"""
        # 初始化通信管理器
        self.communication_manager = CommunicationManager(self.config)
        self.communication_manager.init_environment()
        self.communication_manager.init_communication()
        
        # 测试通信
        if not self.communication_manager.test_communication():
            raise RuntimeError("Communication test failed")
        
        # 广播配置（确保所有节点配置一致）
        self.config = self.communication_manager.broadcast_config(self.config)
        
        # 初始化Worker管理器
        self.worker_manager = WorkerManager(self.config)
        self.worker_manager.start()
        
        # 初始化服务器
        self.server = VLLMServer(self.config)
    
    def start(self):
        """完整启动流程"""
        print("=" * 60)
        print("vLLM Startup Process")
        print("=" * 60)
        
        # 1. 解析命令行参数
        args = self.parse_args()
        
        # 2. 加载配置
        self.config = self.load_config(args)
        
        # 3. 检测硬件环境
        self.hardware_info = self.detect_hardware()
        
        # 4. 初始化组件
        self.initialize_components()
        
        # 5. 启动服务器
        self.server.start()
        
        print("=" * 60)
        print("vLLM Startup Complete")
        print("=" * 60)
    
    def shutdown(self):
        """关闭系统"""
        print("Shutting down vLLM...")
        
        # 关闭服务器
        if hasattr(self, 'server') and self.server is not None:
            # 简化实现，实际会关闭服务器
            pass
        
        # 关闭Worker管理器
        if hasattr(self, 'worker_manager') and self.worker_manager is not None:
            self.worker_manager.shutdown()
        
        # 销毁通信
        if hasattr(self, 'communication_manager') and self.communication_manager is not None:
            self.communication_manager.destroy()
        
        print("vLLM shutdown complete")

if __name__ == "__main__":
    vllm = VLLM()
    try:
        vllm.start()
    except KeyboardInterrupt:
        print("Keyboard interrupt received, shutting down...")
    except Exception as e:
        print(f"Error during startup: {e}")
        import traceback
        traceback.print_exc()
    finally:
        vllm.shutdown()

# 基本启动命令
python -m vllm.entrypoints.api_server --model meta-llama/Llama-2-70b-hf --tensor-parallel-size 8

# 使用配置文件启动
python -m vllm.entrypoints.api_server --config-file config.yaml

# 分布式启动
python -m vllm.entrypoints.api_server --model meta-llama/Llama-2-70b-hf --tensor-parallel-size 8 --distributed

# 量化启动
python -m vllm.entrypoints.api_server --model meta-llama/Llama-2-70b-hf --quantization 4bit

# config.yaml
model: meta-llama/Llama-2-70b-hf
tensor_parallel_size: 8
dtype: float16
quantization: 4bit
gpu_memory_utilization: 0.9
host: 0.0.0.0
port: 8000
workers: 4
trust_remote_code: true
uvicorn_kwargs:
  log_level: info
  limit_max_requests: 10000