2026年本地部署大语言模型完全指南：用Python运行自己的AI模型
为什么要本地部署大语言模型？
每次调用GPT-4、Claude这些商业API，你都在为别人的服务器买单。更关键的是，你的数据——商业计划、代码、客户信息——全部经过第三方服务器。
本地部署LLM的核心优势：
- 数据隐私：所有数据留在本地，不出你的机器
- 零API成本：一次部署，无限调用，不再按token计费
- 离线可用：断网也能正常使用
- 完全可控：模型选择、参数调整、输出格式，全由你说了算
- 低延迟：省去网络往返，响应速度更快（尤其在GPU加速下）
2026年，开源大模型的质量已经非常接近商业模型。Llama 3.3、Qwen 2.5、DeepSeek V3、Mistral等模型在很多任务上已经达到GPT-4级别。这意味着本地部署的实用性已经不再是问题。
本文将带你用Python从零搭建本地LLM服务，覆盖三大主流方案：Ollama、vLLM和llama.cpp，每个方案都有完整可运行的代码。
一、硬件需求与选型建议
1.1 GPU显存是关键
本地部署LLM最大的瓶颈是GPU显存。不同参数量的模型需要的显存差异巨大：
| 模型规模 | FP16显存需求 | 4bit量化后 | 代表模型 |
|---|---|---|---|
| 1-3B | 2-6GB | 1-2GB | Qwen2.5-1.5B, Phi-3-mini |
| 7-8B | 14-16GB | 4-5GB | Llama-3.1-8B, Qwen2.5-7B |
| 14B | 28GB | 8-9GB | Qwen2.5-14B |
| 32-70B | 64-140GB | 20-40GB | Qwen2.5-72B, Llama-3.3-70B |
实用建议：
- 入门级（8GB显存，如RTX 4060）：跑7B量化模型，日常问答、代码生成够用
- 中端（16-24GB显存，如RTX 4080/4090）：跑14B-32B模型，质量明显提升
- 高端（双卡或A100）：跑70B模型，接近商业API效果
- 纯CPU方案：llama.cpp支持CPU推理，速度慢但零硬件门槛
1.2 检查你的GPU环境
#!/usr/bin/env python3
"""检查本地GPU环境，确定适合的模型规模"""
import subprocess
import sys
def check_nvidia_gpu():
"""检查NVIDIA GPU信息"""
try:
result = subprocess.run(
["nvidia-smi", "--query-gpu=name,memory.total,memory.free,driver_version",
"--format=csv,noheader,nounits"],
capture_output=True, text=True
)
if result.returncode == 0:
gpus = []
for line in result.stdout.strip().split('\n'):
parts = [p.strip() for p in line.split(',')]
gpus.append({
'name': parts[0],
'total_mb': int(parts[1]),
'free_mb': int(parts[2]),
'driver': parts[3]
})
return gpus
except FileNotFoundError:
pass
return None
def recommend_model(gpus):
"""根据GPU配置推荐模型"""
if not gpus:
print("⚠️ 未检测到NVIDIA GPU，建议使用llama.cpp CPU推理方案")
print(" 推荐模型：Qwen2.5-3B-Instruct (GGUF Q4_K_M)")
return
total_free = sum(g['free_mb'] for g in gpus)
gpu_info = gpus[0]
print(f"🖥️ GPU: {gpu_info['name']}")
print(f"📊 显存: {gpu_info['total_mb']}MB 总计, {total_free}MB 可用")
print(f"🔧 驱动版本: {gpu_info['driver']}")
print()
if total_free >= 40000:
print("✅ 推荐：70B模型 (Qwen2.5-72B / Llama-3.3-70B)")
print(" 方案：vLLM + AWQ/GPTQ 4bit量化")
elif total_free >= 16000:
print("✅ 推荐：14B-32B模型 (Qwen2.5-14B / DeepSeek-V2-Lite)")
print(" 方案：Ollama 或 vLLM")
elif total_free >= 8000:
print("✅ 推荐：7B-8B模型 (Llama-3.1-8B / Qwen2.5-7B)")
print(" 方案：Ollama（最简单）")
else:
print("⚠️ 显存较小，推荐：3B以下量化模型")
print(" 方案：llama.cpp + GGUF量化格式")
if __name__ == "__main__":
gpus = check_nvidia_gpu()
recommend_model(gpus)
二、方案一：Ollama——最简单的本地LLM方案
Ollama是2024-2026年最火的本地LLM运行工具，一条命令就能跑起来，非常适合入门。
2.1 安装Ollama
# Linux / WSL
curl -fsSL https://ollama.com/install.sh | sh
# macOS
brew install ollama
# Windows：直接下载安装包 https://ollama.com/download
2.2 用命令行快速体验
# 拉取并运行 Qwen2.5 7B 模型
ollama run qwen2.5:7b
# 拉取 Llama 3.1 8B
ollama run llama3.1:8b
# 查看本地已下载的模型
ollama list
# 查看运行中的模型
ollama ps
2.3 用Python调用Ollama API
Ollama启动后会自动暴露一个REST API（默认端口11434），我们可以用Python直接调用：
#!/usr/bin/env python3
"""用Python调用Ollama本地LLM API - 基础用法"""
import requests
import json
import time
class OllamaClient:
"""Ollama API 客户端"""
def __init__(self, base_url="http://localhost:11434"):
self.base_url = base_url
def list_models(self):
"""列出本地所有已下载的模型"""
resp = requests.get(f"{self.base_url}/api/tags")
return resp.json().get("models", [])
def generate(self, model, prompt, system=None, stream=False):
"""向模型发送请求并获取回复"""
payload = {
"model": model,
"prompt": prompt,
"stream": stream,
}
if system:
payload["system"] = system
start = time.time()
resp = requests.post(
f"{self.base_url}/api/generate",
json=payload,
stream=stream
)
if stream:
return self._handle_stream(resp)
result = resp.json()
elapsed = time.time() - start
return {
"response": result["response"],
"total_duration_s": result.get("total_duration", 0) / 1e9,
"eval_count": result.get("eval_count", 0),
"tokens_per_second": result.get("eval_count", 0) / (result.get("eval_duration", 1) / 1e9),
}
def chat(self, model, messages, stream=False):
"""对话模式（支持多轮对话）"""
payload = {
"model": model,
"messages": messages,
"stream": stream,
}
resp = requests.post(
f"{self.base_url}/api/chat",
json=payload,
stream=stream
)
if stream:
return self._handle_stream(resp)
result = resp.json()
return result["message"]["content"]
def _handle_stream(self, resp):
"""处理流式响应"""
full_text = ""
for line in resp.iter_lines():
if line:
chunk = json.loads(line)
token = chunk.get("response", "") or chunk.get("message", {}).get("content", "")
print(token, end="", flush=True)
full_text += token
if chunk.get("done"):
break
print()
return full_text
# ==================== 使用示例 ====================
if __name__ == "__main__":
client = OllamaClient()
model = "qwen2.5:7b"
# 1. 列出本地模型
models = client.list_models()
print("📦 本地模型列表：")
for m in models:
print(f" - {m['name']} ({m['size'] / 1e9:.1f}GB)")
# 2. 简单生成
print("\n" + "=" * 50)
print("📝 简单生成测试：")
result = client.generate(
model=model,
prompt="用Python写一个快速排序算法，要求支持自定义比较函数",
)
print(f"回复：{result['response']}")
print(f"速度：{result['tokens_per_second']:.1f} tokens/s")
# 3. 带系统提示的对话
print("\n" + "=" * 50)
print("💬 对话模式测试：")
messages = [
{
"role": "system",
"content": "你是一位资深Python开发专家，回答简洁实用，包含代码示例。"
},
{
"role": "user",
"content": "如何用asyncio实现一个异步任务队列？"
}
]
reply = client.chat(model=model, messages=messages)
print(f"回复：{reply}")
# 4. 多轮对话
print("\n" + "=" * 50)
print("🔄 多轮对话测试：")
conversation = [
{"role": "system", "content": "你是一个有帮助的助手。"},
{"role": "user", "content": "解释一下Python的GIL是什么"},
]
reply1 = client.chat(model=model, messages=conversation)
print(f"Q: Python的GIL是什么")
print(f"A: {reply1}")
# 追加历史消息，继续对话
conversation.append({"role": "assistant", "content": reply1})
conversation.append({"role": "user", "content": "那在什么场景下GIL会成为瓶颈？"})
reply2 = client.chat(model=model, messages=conversation)
print(f"\nQ: 什么场景下GIL会成为瓶颈？")
print(f"A: {reply2}")
2.4 搭建Ollama API网关——对外提供标准化接口
Ollama自带的API格式比较特殊，如果你想用OpenAI兼容的格式（方便切换后端），可以搭一个适配层：
#!/usr/bin/env python3
"""
将Ollama包装为OpenAI兼容API
这样任何支持OpenAI SDK的工具都能直接接入本地模型
"""
from flask import Flask, request, Response, jsonify
import requests
import json
import time
import uuid
app = Flask(__name__)
OLLAMA_URL = "http://localhost:11434"
@app.route("/v1/chat/completions", methods=["POST"])
def chat_completions():
"""OpenAI兼容的chat completions接口"""
data = request.json
model = data.get("model", "qwen2.5:7b")
messages = data.get("messages", [])
stream = data.get("stream", False)
# 转换为Ollama格式
ollama_payload = {
"model": model,
"messages": messages,
"stream": stream,
"options": {}
}
if "temperature" in data:
ollama_payload["options"]["temperature"] = data["temperature"]
if "max_tokens" in data:
ollama_payload["options"]["num_predict"] = data["max_tokens"]
if stream:
return Response(
stream_ollama(ollama_payload),
content_type="text/event-stream"
)
resp = requests.post(f"{OLLAMA_URL}/api/chat", json=ollama_payload)
result = resp.json()
# 转换为OpenAI格式
return jsonify({
"id": f"chatcmpl-{uuid.uuid4().hex[:8]}",
"object": "chat.completion",
"created": int(time.time()),
"model": model,
"choices": [{
"index": 0,
"message": {
"role": "assistant",
"content": result["message"]["content"]
},
"finish_reason": "stop"
}],
"usage": {
"prompt_tokens": result.get("prompt_eval_count", 0),
"completion_tokens": result.get("eval_count", 0),
"total_tokens": result.get("prompt_eval_count", 0) + result.get("eval_count", 0)
}
})
def stream_ollama(payload):
"""流式响应转换"""
resp = requests.post(
f"{OLLAMA_URL}/api/chat",
json=payload,
stream=True
)
for line in resp.iter_lines():
if line:
chunk = json.loads(line)
content = chunk.get("message", {}).get("content", "")
sse_data = {
"id": f"chatcmpl-{uuid.uuid4().hex[:8]}",
"object": "chat.completion.chunk",
"choices": [{
"index": 0,
"delta": {"content": content} if content else {},
"finish_reason": "stop" if chunk.get("done") else None
}]
}
yield f"data: {json.dumps(sse_data)}\n\n"
if chunk.get("done"):
yield "data: [DONE]\n\n"
@app.route("/v1/models", methods=["GET"])
def list_models():
"""列出可用模型"""
resp = requests.get(f"{OLLAMA_URL}/api/tags")
models = resp.json().get("models", [])
return jsonify({
"object": "list",
"data": [{
"id": m["name"],
"object": "model",
"owned_by": "local"
} for m in models]
})
if __name__ == "__main__":
print("🚀 OpenAI兼容API网关启动在 http://localhost:8000")
app.run(host="0.0.0.0", port=8000)
启动后，任何支持OpenAI SDK的代码都能无缝切换到本地模型：
from openai import OpenAI
# 指向本地网关
client = OpenAI(
base_url="http://localhost:8000/v1",
api_key="not-needed" # 本地不需要真实key
)
response = client.chat.completions.create(
model="qwen2.5:7b",
messages=[
{"role": "system", "content": "你是一个Python专家"},
{"role": "user", "content": "写一个装饰器，实现函数调用的自动重试"}
],
temperature=0.7,
max_tokens=1000
)
print(response.choices[0].message.content)
三、方案二：vLLM——高性能推理引擎
vLLM是目前最流行的高性能LLM推理引擎，支持PagedAttention等先进技术，吞吐量是普通推理的2-24倍。适合需要高并发、低延迟的生产环境。
3.1 安装vLLM
# 确保CUDA版本 >= 11.8
pip install vllm
# 或者用Docker（推荐，环境隔离更好）
docker pull vllm/vllm-openai:latest
3.2 启动vLLM服务
# 启动一个兼容OpenAI API的服务
python -m vllm.entrypoints.openai.api_server \
--model Qwen/Qwen2.5-7B-Instruct \
--host 0.0.0.0 \
--port 8000 \
--max-model-len 8192 \
--gpu-memory-utilization 0.85
# 使用4bit量化模型（显存减半）
python -m vllm.entrypoints.openai.api_server \
--model Qwen/Qwen2.5-14B-Instruct-AWQ \
--host 0.0.0.0 \
--port 8000 \
--quantization awq \
--max-model-len 4096
3.3 用Python调用vLLM
#!/usr/bin/env python3
"""vLLM高性能推理客户端 - 支持批量处理和异步调用"""
import asyncio
import aiohttp
import json
import time
from dataclasses import dataclass
from typing import List
@dataclass
class LLMRequest:
"""LLM请求结构"""
prompt: str
max_tokens: int = 512
temperature: float = 0.7
class VLLMClient:
"""vLLM API客户端，支持并发批量调用"""
def __init__(self, base_url="http://localhost:8000", model="Qwen/Qwen2.5-7B-Instruct"):
self.base_url = base_url
self.model = model
async def complete(self, session, prompt, max_tokens=512, temperature=0.7):
"""异步单次调用"""
payload = {
"model": self.model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": max_tokens,
"temperature": temperature,
}
async with session.post(
f"{self.base_url}/v1/chat/completions",
json=payload
) as resp:
result = await resp.json()
return result["choices"][0]["message"]["content"]
async def batch_complete(self, requests: List[LLMRequest], max_concurrent=10):
"""批量并发调用"""
semaphore = asyncio.Semaphore(max_concurrent)
results = []
async def _process(req):
async with semaphore:
async with aiohttp.ClientSession() as session:
return await self.complete(
session, req.prompt,
req.max_tokens, req.temperature
)
tasks = [_process(req) for req in requests]
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
def check_health(self):
"""检查服务状态"""
import requests
try:
resp = requests.get(f"{self.base_url}/v1/models", timeout=5)
return resp.status_code == 200
except:
return False
# ==================== 实战示例 ====================
async def main():
client = VLLMClient()
# 1. 检查服务
if not client.check_health():
print("❌ vLLM服务未启动，请先运行启动命令")
return
print("✅ vLLM服务正常")
# 2. 单次调用
async with aiohttp.ClientSession() as session:
result = await client.complete(
session,
"用Python实现一个简单的LRU缓存，要求支持TTL过期",
max_tokens=1000
)
print(f"\n📝 单次调用结果：\n{result}")
# 3. 批量处理——同时处理多个任务
print("\n" + "=" * 50)
print("🚀 批量并发测试（5个任务）：")
tasks = [
LLMRequest("写一个Python函数，计算两个日期之间的工作日数量"),
LLMRequest("用Python实现生产者-消费者模式，使用asyncio"),
LLMRequest("写一个装饰器，实现API请求的指数退避重试"),
LLMRequest("用Python解析CSV文件并统计每列的数据分布"),
LLMRequest("实现一个简单的Markdown解析器，支持标题、粗体和链接"),
]
start = time.time()
results = await client.batch_complete(tasks, max_concurrent=5)
elapsed = time.time() - start
for i, result in enumerate(results):
if isinstance(result, Exception):
print(f"\n❌ 任务{i+1}失败: {result}")
else:
print(f"\n✅ 任务{i+1}:")
print(result[:200] + "..." if len(result) > 200 else result)
print(f"\n⏱️ 5个任务总耗时: {elapsed:.2f}s")
if __name__ == "__main__":
asyncio.run(main())
3.4 vLLM性能优化技巧
#!/usr/bin/env python3
"""vLLM性能调优配置示例"""
import subprocess
# 针对不同场景的启动配置
configs = {
"high_throughput": {
"desc": "高吞吐量场景：批量处理、数据标注",
"args": [
"--model", "Qwen/Qwen2.5-7B-Instruct",
"--max-model-len", "4096", # 缩短上下文，提升并发数
"--gpu-memory-utilization", "0.92",
"--max-num-seqs", "64", # 最大并发序列数
"--max-num-batched-tokens", "8192",
"--enforce-eager", # 禁用CUDA Graph，减少显存占用
]
},
"low_latency": {
"desc": "低延迟场景：实时对话、API服务",
"args": [
"--model", "Qwen/Qwen2.5-7B-Instruct",
"--max-model-len", "8192",
"--gpu-memory-utilization", "0.85",
"--max-num-seqs", "16", # 较少并发，降低排队延迟
"--enable-chunked-prefill", # 分块预填充，降低首token延迟
]
},
"memory_efficient": {
"desc": "省显存场景：显存有限时跑更大模型",
"args": [
"--model", "Qwen/Qwen2.5-14B-Instruct-AWQ",
"--quantization", "awq",
"--max-model-len", "2048",
"--gpu-memory-utilization", "0.95",
"--enforce-eager",
"--max-num-seqs", "8",
]
}
}
for name, cfg in configs.items():
print(f"\n{'='*50}")
print(f"📋 {name}: {cfg['desc']}")
cmd = "python -m vllm.entrypoints.openai.api_server " + " ".join(cfg["args"])
print(f"命令: {cmd}")
四、方案三：llama.cpp——纯CPU也能跑
llama.cpp是C++实现的LLM推理引擎，最大特点是不需要GPU也能运行，而且支持GGUF量化格式，显存占用极低。
4.1 安装与配置
# 通过Python绑定安装（最简单）
pip install llama-cpp-python
# 如果需要GPU加速（CUDA）
CMAKE_ARGS="-DGGML_CUDA=on" pip install llama-cpp-python --force-reinstall --no-cache-dir
# 如果需要Metal加速（macOS Apple Silicon）
CMAKE_ARGS="-DGGML_METAL=on" pip install llama-cpp-python --force-reinstall --no-cache-dir
4.2 下载GGUF模型
#!/usr/bin/env python3
"""自动下载GGUF格式的量化模型"""
import os
import requests
from pathlib import Path
MODELS = {
"qwen2.5-7b-q4": {
"url": "https://huggingface.co/Qwen/Qwen2.5-7B-Instruct-GGUF/resolve/main/qwen2.5-7b-instruct-q4_k_m.gguf",
"size": "4.7GB",
"desc": "Qwen2.5-7B 4bit量化，平衡质量和速度"
},
"qwen2.5-3b-q4": {
"url": "https://huggingface.co/Qwen/Qwen2.5-3B-Instruct-GGUF/resolve/main/qwen2.5-3b-instruct-q4_k_m.gguf",
"size": "2.1GB",
"desc": "Qwen2.5-3B 4bit量化，CPU友好"
},
"llama3.1-8b-q4": {
"url": "https://huggingface.co/bartowski/Meta-Llama-3.1-8B-Instruct-GGUF/resolve/main/Meta-Llama-3.1-8B-Instruct-Q4_K_M.gguf",
"size": "4.9GB",
"desc": "Llama 3.1 8B 4bit量化"
},
"phi-3.5-mini-q4": {
"url": "https://huggingface.co/bartowski/Phi-3.5-mini-instruct-GGUF/resolve/main/Phi-3.5-mini-instruct-Q4_K_M.gguf",
"size": "2.3GB",
"desc": "微软Phi-3.5 Mini，轻量高效"
},
}
def download_model(name, target_dir="./models"):
"""下载指定模型"""
if name not in MODELS:
print(f"❌ 未知模型: {name}")
print(f"可用模型: {', '.join(MODELS.keys())}")
return
info = MODELS[name]
os.makedirs(target_dir, exist_ok=True)
filename = info["url"].split("/")[-1]
filepath = os.path.join(target_dir, filename)
if os.path.exists(filepath):
print(f"✅ 模型已存在: {filepath}")
return filepath
print(f"📥 下载 {name} ({info['size']})")
print(f" {info['desc']}")
resp = requests.get(info["url"], stream=True)
total = int(resp.headers.get("content-length", 0))
downloaded = 0
with open(filepath, "wb") as f:
for chunk in resp.iter_content(chunk_size=8192):
f.write(chunk)
downloaded += len(chunk)
if total > 0:
pct = downloaded / total * 100
print(f"\r 进度: {pct:.1f}% ({downloaded/1e9:.2f}GB)", end="")
print(f"\n✅ 下载完成: {filepath}")
return filepath
if __name__ == "__main__":
# 下载适合CPU运行的轻量模型
download_model("qwen2.5-3b-q4")
4.3 用llama-cpp-python进行推理
#!/usr/bin/env python3
"""llama.cpp Python推理——CPU和GPU都支持"""
from llama_cpp import Llama
import time
class LocalLLM:
"""本地LLM推理封装"""
def __init__(self, model_path, n_ctx=4096, n_gpu_layers=-1, n_threads=None):
"""
初始化模型
Args:
model_path: GGUF模型文件路径
n_ctx: 上下文窗口大小
n_gpu_layers: GPU卸载层数，-1=全部GPU，0=纯CPU
n_threads: CPU线程数，None=自动检测
"""
print(f"⏳ 加载模型: {model_path}")
start = time.time()
self.llm = Llama(
model_path=model_path,
n_ctx=n_ctx,
n_gpu_layers=n_gpu_layers,
n_threads=n_threads,
verbose=False
)
elapsed = time.time() - start
print(f"✅ 模型加载完成 ({elapsed:.1f}s)")
def generate(self, prompt, system_prompt=None, max_tokens=512, temperature=0.7):
"""生成回复"""
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
messages.append({"role": "user", "content": prompt})
start = time.time()
output = self.llm.create_chat_completion(
messages=messages,
max_tokens=max_tokens,
temperature=temperature,
top_p=0.9,
)
elapsed = time.time() - start
content = output["choices"][0]["message"]["content"]
tokens = output["usage"]["completion_tokens"]
tokens_per_sec = tokens / elapsed if elapsed > 0 else 0
return {
"text": content,
"tokens": tokens,
"time_s": elapsed,
"speed": tokens_per_sec
}
def stream_generate(self, prompt, system_prompt=None, max_tokens=512, temperature=0.7):
"""流式生成"""
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
messages.append({"role": "user", "content": prompt})
stream = self.llm.create_chat_completion(
messages=messages,
max_tokens=max_tokens,
temperature=temperature,
stream=True
)
full_text = ""
for chunk in stream:
delta = chunk["choices"][0].get("delta", {})
content = delta.get("content", "")
if content:
print(content, end="", flush=True)
full_text += content
print()
return full_text
# ==================== 使用示例 ====================
if __name__ == "__main__":
# 初始化（CPU模式，适合没有GPU的机器）
llm = LocalLLM(
model_path="./models/qwen2.5-3b-instruct-q4_k_m.gguf",
n_ctx=4096,
n_gpu_layers=0, # 纯CPU推理
n_threads=4 # 使用4个CPU线程
)
# 1. 代码生成
print("=" * 50)
print("🔧 代码生成测试：")
result = llm.generate(
system_prompt="你是一位Python专家，只输出代码，不要解释",
prompt="写一个Python函数，实现布隆过滤器(Bloom Filter)，支持add和check操作",
max_tokens=800
)
print(result["text"])
print(f"⏱️ {result['tokens']} tokens, {result['speed']:.1f} tokens/s")
# 2. 流式输出
print("\n" + "=" * 50)
print("📡 流式输出测试：")
llm.stream_generate(
system_prompt="用中文回答，简洁明了",
prompt="Python的__slots__有什么用？什么场景下应该使用它？",
max_tokens=300
)
# 3. 文本分析
print("\n" + "=" * 50)
print("📊 文本分析测试：")
code_sample = '''
def fibonacci(n):
if n <= 1:
return n
a, b = 0, 1
for _ in range(2, n + 1):
a, b = b, a + b
return b
'''
result = llm.generate(
prompt=f"分析以下代码的时间复杂度和空间复杂度，并提出优化建议：\n```python\n{code_sample}\n```",
max_tokens=400
)
print(result["text"])
print(f"⏱️ {result['tokens']} tokens, {result['speed']:.1f} tokens/s")
五、构建完整的本地AI应用
光会调用模型还不够，真正的价值在于把LLM集成到实际应用中。下面是一个完整的实战项目：本地AI代码审查助手。
5.1 项目架构
#!/usr/bin/env python3
"""
本地AI代码审查助手
功能：自动扫描Git仓库的代码变更，用本地LLM进行审查，生成审查报告
架构：
Git Diff -> 代码分块 -> LLM审查 -> 结果聚合 -> 报告生成
"""
import subprocess
import json
import os
import re
from dataclasses import dataclass, field
from typing import List, Optional
from datetime import datetime
import requests
@dataclass
class CodeChange:
"""代码变更结构"""
file_path: str
change_type: str # added, modified, deleted
diff_content: str
language: str = ""
@dataclass
class ReviewIssue:
"""审查发现的问题"""
severity: str # critical, warning, suggestion
file_path: str
line_hint: str
description: str
fix_suggestion: str
@dataclass
class ReviewReport:
"""审查报告"""
repo: str
branch: str
commit: str
timestamp: str
issues: List[ReviewIssue] = field(default_factory=list)
summary: str = ""
score: int = 0
class CodeReviewer:
"""AI代码审查助手"""
SYSTEM_PROMPT = """你是一个资深代码审查专家。你的任务是审查代码变更，发现潜在问题。
请关注以下方面：
1. **安全性**：SQL注入、XSS、硬编码密钥、不安全的加密
2. **性能**：不必要的循环、内存泄漏、N+1查询
3. **代码质量**：命名规范、代码重复、复杂度过高
4. **Bug风险**：边界条件、异常处理、并发问题
5. **最佳实践**：类型提示、文档字符串、测试覆盖
对每个发现的问题，请用以下JSON格式输出：
```json
{
"severity": "critical|warning|suggestion",
"description": "问题描述",
"line_hint": "大致在哪个位置",
"fix_suggestion": "修复建议"
}
如果代码没有问题，返回空数组 []。
只输出JSON，不要其他内容。"""
def __init__(self, model_url="http://localhost:11434/api/generate", model="qwen2.5:7b"):
self.model_url = model_url
self.model = model
def get_git_diff(self, repo_path, commit="HEAD"):
"""获取Git变更"""
os.chdir(repo_path)
# 获取变更的文件列表
result = subprocess.run(
["git", "diff", "--name-status", f"{commit}~1", commit],
capture_output=True, text=True
)
changes = []
for line in result.stdout.strip().split('\n'):
if not line:
continue
parts = line.split('\t')
status = parts[0]
filepath = parts[1]
# 获取具体diff
diff_result = subprocess.run(
["git", "diff", f"{commit}~1", commit, "--", filepath],
capture_output=True, text=True
)
lang = self._detect_language(filepath)
change_type = {"A": "added", "M": "modified", "D": "deleted"}.get(status[0], "modified")
changes.append(CodeChange(
file_path=filepath,
change_type=change_type,
diff_content=diff_result.stdout,
language=lang
))
return changes
def _detect_language(self, filepath):
"""根据文件扩展名判断语言"""
ext_map = {
'.py': 'Python', '.js': 'JavaScript', '.ts': 'TypeScript',
'.go': 'Go', '.rs': 'Rust', '.java': 'Java',
'.cpp': 'C++', '.c': 'C', '.rb': 'Ruby',
'.php': 'PHP', '.sql': 'SQL', '.sh': 'Shell',
}
for ext, lang in ext_map.items():
if filepath.endswith(ext):
return lang
return "Unknown"
def review_change(self, change: CodeChange) -> List[ReviewIssue]:
"""审查单个代码变更"""
if not change.diff_content.strip():
return []
# 限制diff大小，避免超出上下文窗口
diff_text = change.diff_content[:6000]
prompt = f"""审查以下代码变更：
文件: {change.file_path} 语言: {change.language} 变更类型: {change.change_type}
{diff_text}
```"""
try:
resp = requests.post(self.model_url, json={
"model": self.model,
"prompt": prompt,
"system": self.SYSTEM_PROMPT,
"stream": False,
"options": {"temperature": 0.1}
}, timeout=120)
result = resp.json()
response_text = result["response"]
# 解析JSON响应
issues = self._parse_issues(response_text, change.file_path)
return issues
except Exception as e:
print(f"⚠️ 审查 {change.file_path} 时出错: {e}")
return []
def _parse_issues(self, text, file_path):
"""从LLM响应中解析问题列表"""
issues = []
# 提取JSON数组
json_match = re.search(r'\[[\s\S]*?\]', text)
if not json_match:
return []
try:
items = json.loads(json_match.group())
for item in items:
issues.append(ReviewIssue(
severity=item.get("severity", "suggestion"),
file_path=file_path,
line_hint=item.get("line_hint", ""),
description=item.get("description", ""),
fix_suggestion=item.get("fix_suggestion", "")
))
except json.JSONDecodeError:
pass
return issues
def generate_report(self, repo, branch, commit, issues) -> str:
"""生成Markdown格式的审查报告"""
critical = [i for i in issues if i.severity == "critical"]
warnings = [i for i in issues if i.severity == "warning"]
suggestions = [i for i in issues if i.severity == "suggestion"]
# 计算评分（100分制）
score = max(0, 100 - len(critical) * 20 - len(warnings) * 5 - len(suggestions) * 1)
severity_icons = {"critical": "🔴", "warning": "🟡", "suggestion": "💡"}
report = f"""# 🔍 代码审查报告
- **仓库**: {repo}
- **分支**: {branch}
- **提交**: {commit[:8]}
- **时间**: {datetime.now().strftime('%Y-%m-%d %H:%M')}
- **评分**: {score}/100
## 📊 统计
| 级别 | 数量 |
|------|------|
| 🔴 严重问题 | {len(critical)} |
| 🟡 警告 | {len(warnings)} |
| 💡 建议 | {len(suggestions)} |
"""
if not issues:
report += "✅ **未发现问题，代码质量良好！**\n"
return report
report += "## 📋 问题详情\n\n"
for severity in ["critical", "warning", "suggestion"]:
group = [i for i in issues if i.severity == severity]
if not group:
continue
report += f"### {severity_icons[severity]} {severity.upper()}\n\n"
for i, issue in enumerate(group, 1):
report += f"**{i}. {issue.file_path}** {issue.line_hint}\n\n"
report += f"> {issue.description}\n\n"
report += f"**修复建议**: {issue.fix_suggestion}\n\n---\n\n"
return report
def review_repo(self, repo_path, commit="HEAD"):
"""完整审查流程"""
print(f"🔍 开始审查: {repo_path}")
# 获取变更
changes = self.get_git_diff(repo_path, commit)
print(f"📂 发现 {len(changes)} 个文件变更")
# 逐文件审查
all_issues = []
for change in changes:
print(f" 审查: {change.file_path} ({change.language})")
issues = self.review_change(change)
all_issues.extend(issues)
if issues:
for issue in issues:
icon = {"critical": "🔴", "warning": "🟡", "suggestion": "💡"}[issue.severity]
print(f" {icon} {issue.description[:60]}")
# 生成报告
repo_name = os.path.basename(os.path.abspath(repo_path))
result = subprocess.run(
["git", "rev-parse", "--abbrev-ref", "HEAD"],
capture_output=True, text=True, cwd=repo_path
)
branch = result.stdout.strip()
report = self.generate_report(repo_name, branch, commit, all_issues)
# 保存报告
report_path = os.path.join(repo_path, "REVIEW_REPORT.md")
with open(report_path, "w") as f:
f.write(report)
print(f"\n📝 审查报告已保存: {report_path}")
print(f"📊 发现 {len(all_issues)} 个问题")
return report
# ==================== 使用 ====================
if __name__ == "__main__":
import sys
repo_path = sys.argv[1] if len(sys.argv) > 1 else "."
reviewer = CodeReviewer(
model_url="http://localhost:11434/api/generate",
model="qwen2.5:7b"
)
report = reviewer.review_repo(repo_path)
print("\n" + "=" * 60)
print(report)
5.2 集成到CI/CD流水线
# .github/workflows/ai-review.yml
# 在GitHub Actions中使用本地LLM进行代码审查
name: AI Code Review
on:
pull_request:
branches: [main]
jobs:
ai-review:
runs-on: self-hosted # 需要有GPU的自托管runner
steps:
- uses: actions/checkout@v4
with:
fetch-depth: 2
- name: Run AI Review
run: |
python scripts/ai_reviewer.py . > review_output.md
- name: Comment on PR
uses: actions/github-script@v7
with:
script: |
const fs = require('fs');
const review = fs.readFileSync('review_output.md', 'utf8');
github.rest.issues.createComment({
owner: context.repo.owner,
repo: context.repo.repo,
issue_number: context.issue.number,
body: review
});
六、模型微调：让通用模型变成领域专家
6.1 用LoRA进行低成本微调
#!/usr/bin/env python3
"""
使用LoRA微调开源模型，使其精通特定领域
适合在单卡24GB显存上微调7B模型
"""
from datasets import Dataset
from transformers import AutoTokenizer, AutoModelForCausalLM
from peft import LoraConfig, get_peft_model, TaskType
from trl import SFTTrainer, SFTConfig
import torch
def prepare_dataset():
"""准备训练数据——以代码审查为例"""
# 实际项目中应该有几百到几千条高质量数据
examples = [
{
"instruction": "审查以下Python代码的安全问题",
"input": "query = f\"SELECT * FROM users WHERE name = '{user_input}'\"",
"output": "🔴 严重安全问题：SQL注入漏洞。\n\n问题：直接将用户输入拼接到SQL语句中，攻击者可以通过构造恶意输入执行任意SQL命令。\n\n修复建议：使用参数化查询\n```python\nquery = \"SELECT * FROM users WHERE name = %s\"\ncursor.execute(query, (user_input,))\n```"
},
{
"instruction": "审查以下Python代码的性能问题",
"input": "result = []\nfor item in large_list:\n if item not in another_list:\n result.append(item)",
"output": "🟡 性能警告：O(n*m)时间复杂度的列表查找。\n\n问题：`item not in another_list`对列表的查找操作是O(n)，嵌套循环导致整体复杂度为O(n*m)。\n\n修复建议：将列表转换为集合\n```python\nanother_set = set(another_list)\nresult = [item for item in large_list if item not in another_set]\n```\n优化后复杂度降为O(n+m)。"
},
]
# 转换为对话格式
formatted = []
for ex in examples:
formatted.append({
"messages": [
{"role": "system", "content": "你是一个代码审查专家"},
{"role": "user", "content": f"{ex['instruction']}\n\n{ex['input']}"},
{"role": "assistant", "content": ex["output"]}
]
})
return Dataset.from_list(formatted)
def train_lora():
"""LoRA微调"""
model_name = "Qwen/Qwen2.5-7B-Instruct"
# 加载模型和分词器
tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)
model = AutoModelForCausalLM.from_pretrained(
model_name,
torch_dtype=torch.bfloat16,
device_map="auto",
trust_remote_code=True
)
# LoRA配置
lora_config = LoraConfig(
task_type=TaskType.CAUSAL_LM,
r=16, # LoRA秩，越大拟合能力越强
lora_alpha=32, # 缩放系数
lora_dropout=0.05,
target_modules=["q_proj", "k_proj", "v_proj", "o_proj",
"gate_proj", "up_proj", "down_proj"],
bias="none",
)
model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
# 输出类似：trainable params: 13,631,488 || all params: 7,628,000,000 || 0.18%
# 训练配置
training_config = SFTConfig(
output_dir="./lora-output",
num_train_epochs=3,
per_device_train_batch_size=2,
gradient_accumulation_steps=8,
learning_rate=2e-4,
warmup_ratio=0.1,
logging_steps=10,
save_strategy="epoch",
bf16=True,
max_seq_length=2048,
)
# 准备数据
dataset = prepare_dataset()
# 开始训练
trainer = SFTTrainer(
model=model,
args=training_config,
train_dataset=dataset,
processing_class=tokenizer,
)
trainer.train()
# 保存LoRA权重（只保存adapter，通常几十MB）
model.save_pretrained("./lora-output/final")
tokenizer.save_pretrained("./lora-output/final")
print("✅ LoRA微调完成，权重已保存")
if __name__ == "__main__":
train_lora()
6.2 合并LoRA权重并导出GGUF
# 合并LoRA权重到基础模型
python -c "
from peft import PeftModel
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch
base = AutoModelForCausalLM.from_pretrained('Qwen/Qwen2.5-7B-Instruct', torch_dtype=torch.float16)
model = PeftModel.from_pretrained(base, './lora-output/final')
model = model.merge_and_unload()
model.save_pretrained('./merged-model')
print('✅ 合并完成')
"
# 转换为GGUF格式（用于llama.cpp/Ollama）
python llama.cpp/convert_hf_to_gguf.py ./merged-model --outtype q4_k_m --outfile ./my-custom-model.gguf
# 在Ollama中注册自定义模型
cat > Modelfile << 'EOF'
FROM ./my-custom-model.gguf
SYSTEM "你是一个专业的代码审查AI，精通Python安全和性能优化"
PARAMETER temperature 0.1
EOF
ollama create code-reviewer -f Modelfile
ollama run code-reviewer
七、性能对比与选型建议
7.1 三大方案对比
| 特性 | Ollama | vLLM | llama.cpp |
|---|---|---|---|
| 安装难度 | ⭐ 最简单 | ⭐⭐⭐ 需要CUDA | ⭐⭐ 中等 |
| GPU推理速度 | 中等 | ⭐⭐⭐ 最快 | 中等偏上 |
| CPU推理 | 支持 | 不支持 | ⭐⭐⭐ 最优 |
| 并发能力 | 一般 | ⭐⭐⭐ 极强 | 一般 |
| API兼容性 | 自有格式 | OpenAI兼容 | 自有格式 |
| 量化支持 | GGUF | AWQ/GPTQ/FP8 | GGUF |
| 适用场景 | 开发测试、个人使用 | 生产环境、API服务 | 资源受限、CPU环境 |
| 社区活跃度 | ⭐⭐⭐ | ⭐⭐⭐ | ⭐⭐⭐ |
7.2 选型决策树
你的需求是什么？
│
├── 个人开发/学习/测试
│ └── 选 Ollama → 最快上手，一条命令搞定
│
├── 生产环境API服务
│ ├── 需要高并发 → 选 vLLM
│ └── 并发要求不高 → 选 Ollama + OpenAI适配层
│
├── 没有GPU / 只有CPU
│ └── 选 llama.cpp → CPU推理性能最优
│
└── 嵌入式/边缘设备
└── 选 llama.cpp + 超小模型 (1-3B)
八、常见问题排查
8.1 显存不足 (CUDA out of memory)
# 方案1：使用更小的量化模型
# 原始: Qwen2.5-14B-Instruct (FP16, ~28GB)
# 替换: Qwen2.5-14B-Instruct-AWQ (4bit, ~8GB)
# 方案2：减少上下文长度
# --max-model-len 2048 (默认可能是32768)
# 方案3：降低GPU内存占用率
# --gpu-memory-utilization 0.80
8.2 模型输出乱码或重复
# 调整采样参数
params = {
"temperature": 0.7, # 太低会重复，太高会乱说
"top_p": 0.9, # 核采样
"top_k": 40, # 限制候选token数
"repeat_penalty": 1.1, # 重复惩罚
"max_tokens": 1024, # 防止无限生成
}
8.3 速度太慢
# 检查是否真的在用GPU
nvidia-smi # 看看GPU利用率
# llama.cpp: 确保编译时开启了CUDA
CMAKE_ARGS="-DGGML_CUDA=on" pip install llama-cpp-python --force-reinstall
# vLLM: 确保启用了连续批处理
# --enable-chunked-prefill --max-num-seqs 32
# 通用: 使用更小的模型或更激进的量化
九、总结与下一步
本地部署LLM已经不再是高门槛的技术挑战。2026年的开源工具链已经非常成熟：
- 想最快上手 → 装Ollama，
ollama run qwen2.5:7b一条命令搞定 - 要生产级性能 → 用vLLM，支持高并发和低延迟
- 没有GPU → 用llama.cpp，CPU也能跑
下一步建议：
- 结合本文的代码审查助手示例，构建自己的AI辅助开发工具
- 尝试微调模型，让AI精通你的业务领域
- 搭建OpenAI兼容的API网关，让团队共用本地模型
本地AI不是未来，是现在。数据安全、零成本、完全可控——这些优势在企业环境中越来越重要。现在就开始部署你的第一个本地LLM吧。
本文代码均已在实际环境中测试通过。如果部署过程中遇到问题，欢迎在评论区留言交流。
评论