用 50 行 Python 搭一个本地 LLM 推理服务——vLLM 原理拆解 + 可运行代码
用 50 行 Python 搭一个本地 LLM 推理服务——vLLM 原理拆解 + 可运行代码
很多人以为"高性能 LLM 推理"是大厂专属，自己跑个本地服务就只能用 llama.cpp 凑合。
其实不然。vLLM 的核心思路——PagedAttention——并不复杂，理解之后你会发现，哪怕不自己实现，也能在调用 API 时做出更聪明的工程决策。
为什么普通推理这么慢？
传统推理框架的瓶颈在 KV Cache。
每个 token 生成时，模型需要保存之前所有 token 的 Key/Value 矩阵。问题是：你不知道这次对话最终会有多长，所以框架只能预先分配最大长度的显存。
结果就是：
- 一个 4096 token 的上下文窗口，哪怕你只用了 200 token，剩下 3896 token 的显存也被占着
- 并发 10 个请求，显存直接爆
vLLM 的 PagedAttention 借鉴了操作系统的虚拟内存分页思路：KV Cache 不再连续分配，而是按需分配固定大小的"页"，用一张页表做映射。显存利用率从 20-40% 直接拉到 90%+。
动手：用 OpenAI 兼容接口做一个流式推理客户端
不需要自己部署 vLLM（那需要 GPU 服务器），我们直接用兼容 OpenAI 格式的 API 来演示流式输出 + 并发请求的工程实践。
原理是一样的：理解了 streaming 和 async 的配合，你就理解了推理服务的核心调用模式。
环境准备
pip install openai httpx asyncio
完整代码（50 行）
import asyncio
import time
from openai import AsyncOpenAI
# 换一行 base_url，国内直连，不需要代理
client = AsyncOpenAI(
api_key="your-api-key",
base_url="https://api2everything.xyz/v1"
)
async def stream_completion(prompt: str, request_id: int) -> dict:
"""单个流式请求，记录首 token 延迟和总耗时"""
start = time.perf_counter()
first_token_time = None
full_response = []
stream = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
stream=True,
max_tokens=200,
)
async for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
if first_token_time is None:
first_token_time = time.perf_counter() - start
full_response.append(delta)
total_time = time.perf_counter() - start
text = "".join(full_response)
return {
"id": request_id,
"ttft": round(first_token_time or 0, 3), # Time To First Token
"total": round(total_time, 3),
"tokens": len(text.split()),
"preview": text[:60] + "..."
}
async def benchmark_concurrent(prompts: list[str]):
"""并发发送多个请求，模拟真实负载"""
print(f"并发发送 {len(prompts)} 个请求...\n")
tasks = [stream_completion(p, i) for i, p in enumerate(prompts)]
results = await asyncio.gather(*tasks)
for r in results:
print(f"[{r['id']}] TTFT={r['ttft']}s | 总耗时={r['total']}s | "
f"词数≈{r['tokens']} | {r['preview']}")
avg_ttft = sum(r['ttft'] for r in results) / len(results)
print(f"\n平均首 token 延迟: {avg_ttft:.3f}s")
if __name__ == "__main__":
prompts = [
"用一句话解释 PagedAttention 的核心思想",
"Python asyncio 和多线程的本质区别是什么",
"KV Cache 为什么会成为 LLM 推理的瓶颈",
"解释 speculative decoding 的工作原理",
]
asyncio.run(benchmark_concurrent(prompts))
跑起来之后你会看到什么
并发发送 4 个请求...
[0] TTFT=0.312s | 总耗时=1.847s | 词数≈38 | PagedAttention 将 KV Cache 拆分为固定大小的...
[2] TTFT=0.298s | 总耗时=1.923s | 词数≈41 | KV Cache 存储每个 token 的注意力键值对，随序列...
[1] TTFT=0.334s | 总耗时=2.105s | 词数≈44 | asyncio 是单线程事件循环，多线程依赖 OS 调度...
[3] TTFT=0.445s | 总耗时=2.341s | 词数≈52 | Speculative decoding 用小模型草稿，大模型验证...
平均首 token 延迟: 0.347s
4 个请求并发，总耗时和单个请求差不多——这就是异步 IO 的价值，也是推理服务能扛高并发的基础。
几个值得注意的工程细节
TTFT vs 总延迟：用户感知的"快不快"主要看 TTFT（首 token 延迟），而不是总耗时。流式输出的意义就在这里——让用户感觉响应立刻开始了。
asyncio.gather vs asyncio.wait：gather 会等所有任务完成再返回，适合 benchmark；生产环境如果需要"先完成先处理"，用 as_completed。
连接池复用：AsyncOpenAI 客户端内部维护了 HTTP 连接池，不要在循环里反复初始化 client，那会把连接池优势全部抵消。
错误处理：高并发下偶发 429（限流）很正常，加指数退避重试：
import random
async def retry_with_backoff(prompt, request_id, max_retries=3):
for attempt in range(max_retries):
try:
return await stream_completion(prompt, request_id)
except Exception as e:
if attempt == max_retries - 1:
raise
wait = (2 ** attempt) + random.uniform(0, 1)
await asyncio.sleep(wait)
关于模型选择
代码里用的是 gpt-4o-mini，做 benchmark 和原型验证够用。如果你要测试推理能力上限，可以换 deepseek-r1 或 claude-3-5-sonnet，接口格式完全一样，改一个字段就行。
国内访问官方 API 需要代理，而且 Claude/Gemini 的充值本身就是个麻烦事。我自己现在用的是无量Api，300+ 模型国内直连，价格比官方便宜 60% 左右，OpenAI 格式兼容，就是把上面代码里的 base_url 换成他们的地址，其他一行不改。注册还送 ¥1 余额，拿来跑这个 benchmark 完全够。
如果你在跑这段代码时遇到问题，或者想聊聊 PagedAttention 的具体实现细节，评论区见。
觉得有用的话点个赞，后续还会写 speculative decoding 和 continuous batching 的实战拆解。