从命令行到代码封装：Python 调用 vLLM 实战

环境搭建完毕，服务成功启动后，很多开发者习惯直接用 curl 在终端里敲两行命令测试一下，看到返回结果就心满意足地结束了。但在实际的生产集成或应用开发中，我们不可能让业务系统去执行 shell 命令。如何将 vLLM 提供的 OpenAI 兼容接口优雅地融入 Python 项目，处理流式输出、异常重试以及耗时监控，才是工程落地的关键一步。

既然我们的 Instinct GPU 已经跑起了 vLLM 服务（假设监听在 http://localhost:8000），接下来就跳过繁琐的环境配置，直接聚焦于应用层的集成实践。我们将使用标准的 requests 库，构建一个健壮、可复用的客户端脚本，涵盖单轮对话、流式响应以及基础的错误处理逻辑。

为什么不要只依赖 Curl 测试

在调试阶段，curl 确实是神器。一条命令就能验证服务是否存活：

curl http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{
    "model": "meta-llama/Meta-Llama-3-8B-Instruct",
    "messages": [{"role": "user", "content": "你好"}],
    "max_tokens": 50
  }'

但这仅仅证明了“路是通的”。在真实业务场景中，我们需要处理网络波动导致的超时、服务端返回的 5xx 错误、JSON 解析失败以及各种边界情况。如果把这些逻辑散落在业务的各个角落，代码将难以维护。因此，封装一个统一的 API 客户端类，将 HTTP 细节、重试机制和日志记录收敛起来，是迈向生产级的必要操作。

构建健壮的 Python 调用客户端

下面是一个完整的 Python 脚本示例。它不仅仅发送请求，还包含了连接超时设置、自动重试逻辑以及对流式输出的支持。你可以直接复制并根据实际模型名称微调。

import requests
import json
import time
from typing import Generator, Optional, Dict, Any

class VLLMClient:
    def __init__(self, base_url: str, model_name: str, timeout: int = 30):
        self.base_url = base_url.rstrip("/")
        self.model_name = model_name
        self.timeout = timeout
        self.session = requests.Session()
        # 设置重试策略，应对短暂的网络抖动或服务重启
        from requests.adapters import HTTPAdapter
        from urllib3.util.retry import Retry
        
        retry_strategy = Retry(
            total=3,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["POST"]
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        self.session.mount("http://", adapter)
        self.session.mount("https://", adapter)

    def chat_completion(self, messages: list[dict], max_tokens: int = 512, temperature: float = 0.7) -> Optional[str]:
        """非流式调用，直接返回完整生成内容"""
        url = f"{self.base_url}/v1/chat/completions"
        payload = {
            "model": self.model_name,
            "messages": messages,
            "max_tokens": max_tokens,
            "temperature": temperature,
            "stream": False
        }
        
        start_time = time.time()
        try:
            response = self.session.post(url, json=payload, timeout=self.timeout)
            response.raise_for_status()
            data = response.json()
            
            elapsed = time.time() - start_time
            # 简单日志：记录耗时和 token 使用情况
            usage = data.get("usage", {})
            print(f"[Info] Request finished in {elapsed:.2f}s. Prompt: {usage.get('prompt_tokens', 0)}, Completion: {usage.get('completion_tokens', 0)}")
            
            return data["choices"][0]["message"]["content"]
        except requests.exceptions.RequestException as e:
            print(f"[Error] Request failed: {e}")
            return None

    def chat_completion_stream(self, messages: list[dict], max_tokens: int = 512) -> Generator[str, None, None]:
        """流式调用，逐字yield生成内容"""
        url = f"{self.base_url}/v1/chat/completions"
        payload = {
            "model": self.model_name,
            "messages": messages,
            "max_tokens": max_tokens,
            "stream": True
        }

        try:
            with self.session.post(url, json=payload, timeout=self.timeout, stream=True) as response:
                response.raise_for_status()
                for line in response.iter_lines():
                    if line:
                        decoded_line = line.decode("utf-8")
                        if decoded_line.startswith("data: "):
                            data_str = decoded_line[6:]
                            if data_str.strip() == "[DONE]":
                                break
                            try:
                                data = json.loads(data_str)
                                content = data["choices"][0]["delta"].get("content", "")
                                if content:
                                    yield content
                            except json.JSONDecodeError:
                                continue
        except requests.exceptions.RequestException as e:
            print(f"[Error] Stream request failed: {e}")

if __name__ == "__main__":
    # 初始化客户端，请根据实际部署情况修改 model_name
    client = VLLMClient(
        base_url="http://localhost:8000", 
        model_name="meta-llama/Meta-Llama-3-8B-Instruct"
    )

    messages = [
        {"role": "system", "content": "你是一个专业的 AI 助手。"},
        {"role": "user", "content": "请用一句话解释 ROCm 在大模型推理中的作用。"}
    ]

    print("=== 测试非流式调用 ===")
    result = client.chat_completion(messages)
    if result:
        print(f"回答：{result}\n")

    print("=== 测试流式调用 ===")
    stream_messages = [
        {"role": "user", "content": "写一段 Python 代码计算斐波那契数列的前 10 项。"}
    ]
    
    print("正在生成...", end="", flush=True)
    for chunk in client.chat_completion_stream(stream_messages):
        print(chunk, end="", flush=True)
    print("\n生成结束。")

这段代码做了几个关键动作：首先，它利用 requests.Session 和 Retry 机制处理了临时性的网络故障，避免因为一次偶发的超时导致整个任务失败；其次，它将非流式和流式调用封装成两个清晰的方法，流式部分通过 iter_lines() 逐行解析 SSE（Server-Sent Events）数据，实现了类似打字机的实时输出效果；最后，我们在非流式调用中加入了一个简单的耗时打印，这对于后续的性能分析至关重要。

生产环境的可观测性建议

代码跑通只是第一步。当你的服务接入真实流量后，如何知道它是否健康？上面的脚本中我们打印了 elapsed 时间，但这远远不够。

在生产环境中，建议在客户端或服务端网关层收集更细致的指标。重点关注 首字延迟（TTFT） 和 端到端延迟。对于流式请求，TTFT 是用户体验的核心，如果用户发出请求后超过 2 秒才看到第一个字，通常会认为服务卡死。你可以在客户端记录发出请求的时间戳与收到第一个 yield 内容的时间戳之差，将其上报到监控系统（如 Prometheus）。

此外，留意 vLLM 返回的 usage 字段。它不仅包含 token 数量，还能间接反映显存压力和批处理效率。如果发现某些请求的 completion_tokens 极少但耗时极长，可能是遇到了显存碎片化导致的调度延迟，或者是后端发生了 OOM 重启。通过将这类日志结构化（例如输出为 JSON 格式并接入 ELK 栈），你可以快速定位是特定长度的输入导致了性能抖动，还是某个特定的模型算子在 Instinct GPU 上存在兼容性问题。

从简单的 curl 测试到封装健壮的客户端，再到建立监控闭环，这一步跨越了从“Demo"到“产品”的鸿沟。在 AMD ROCm 生态日益成熟的今天，掌握这些工程化细节，能让你的大模型应用在本地或云端跑得更加平稳高效。

200小时GPU算力已就位，快来领取：https://marketing.csdn.net/questions/Q2604140858304426315?utm_source=AIpaper