大模型部署¶

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig

# FP16量化
model_fp16 = AutoModelForCausalLM.from_pretrained(
    "gpt2",
    torch_dtype=torch.float16
)

# INT8量化
model_int8 = AutoModelForCausalLM.from_pretrained(
    "gpt2",
    load_in_8bit=True,
    device_map="auto"
)

# INT4量化
bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_compute_dtype=torch.float16
)

model_int4 = AutoModelForCausalLM.from_pretrained(
    "gpt2",
    quantization_config=bnb_config,
    device_map="auto"
)

# 比较显存占用
def get_model_size(model):
    """获取模型大小"""
    param_size = 0
    for param in model.parameters():
        param_size += param.nelement() * param.element_size()
    return param_size / (1024 ** 2)  # MB

print(f"FP16模型大小: {get_model_size(model_fp16):.2f} MB")
print(f"FP32模型大小(估算): {get_model_size(model_fp16) * 2:.2f} MB")  # 估算值: FP16字节数 × 2，仅当所有参数均为FP16时准确
print(f"INT8模型大小: {get_model_size(model_int8):.2f} MB")
print(f"INT4模型大小: {get_model_size(model_int4):.2f} MB")

from transformers import AutoModelForCausalLM, AutoTokenizer
from auto_gptq import AutoGPTQForCausalLM, BaseQuantizeConfig

# 量化配置（注意：BaseQuantizeConfig 不接受 model_name_or_path 参数）
quantize_config = BaseQuantizeConfig(
    bits=4,  # 量化位数
    group_size=128,  # 组大小
    damp_percent=0.01,
    desc_act=False,
    sym=True,
    true_sequential=True,
)

# 加载并量化模型
model = AutoGPTQForCausalLM.from_pretrained(
    "gpt2",
    quantize_config=quantize_config,
    device_map="auto"
)

# 保存量化模型
model.save_quantized("./gpt2-gptq")

from transformers import AutoModelForCausalLM, AutoTokenizer
import torch

# PyTorch 2.0+ 原生SDPA优化（替代已废弃的BetterTransformer）
# 方式1: 自动启用SDPA（PyTorch 2.0+默认）
model = AutoModelForCausalLM.from_pretrained(
    "gpt2",
    torch_dtype=torch.float16,
    device_map="auto"
)

# 方式2: 显式指定SDPA注意力实现
model = AutoModelForCausalLM.from_pretrained(
    "gpt2",
    torch_dtype=torch.float16,
    attn_implementation="sdpa",  # 使用PyTorch原生SDPA
    device_map="auto"
)

# 使用bitsandbytes进行4-bit量化
model = AutoModelForCausalLM.from_pretrained(
    "gpt2",
    load_in_4bit=True,
    device_map="auto"
)

# 保存优化模型
model.save_pretrained("./gpt2-optimized")

from vllm import LLM, SamplingParams

# 初始化LLM
llm = LLM(
    model="gpt2",
    tensor_parallel_size=1,  # 张量并行大小
    gpu_memory_utilization=0.9,  # GPU显存利用率
    max_model_len=2048,  # 最大序列长度
)

# 采样参数
sampling_params = SamplingParams(
    temperature=0.7,
    top_p=0.95,
    max_tokens=100
)

# 生成
prompts = ["你好!", "介绍一下人工智能"]
outputs = llm.generate(prompts, sampling_params)

# 打印结果
for output in outputs:
    print(f"Prompt: {output.prompt}")
    print(f"Output: {output.outputs[0].text}\n")

import requests

# API端点
api_url = "http://localhost:8080/generate"

# 请求参数
payload = {
    "inputs": "你好!",
    "parameters": {
        "max_new_tokens": 100,
        "temperature": 0.7,
        "top_p": 0.95
    }
}

# 发送请求
response = requests.post(api_url, json=payload)
result = response.json()

# 打印结果
print(f"生成文本: {result['generated_text']}")

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch

# 初始化FastAPI
app = FastAPI(title="大模型API")

# 加载模型
model_name = "gpt2"
model = AutoModelForCausalLM.from_pretrained(model_name)
tokenizer = AutoTokenizer.from_pretrained(model_name)

# 设置设备
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model.to(device)  # .to(device)将数据移至GPU/CPU

# 请求模型
class GenerationRequest(BaseModel):  # Pydantic BaseModel：自动数据验证和序列化
    text: str
    max_length: int = 100
    temperature: float = 0.7
    top_p: float = 0.95

# 响应模型
class GenerationResponse(BaseModel):
    generated_text: str

@app.post("/generate", response_model=GenerationResponse)
async def generate(request: GenerationRequest):  # async def定义协程函数
    """生成文本"""
    try:  # try/except捕获异常，防止程序崩溃
        # 编码输入
        inputs = tokenizer(
            request.text,
            return_tensors="pt",
            truncation=True,
            max_length=512
        ).to(device)

        # 生成
        with torch.no_grad():  # 禁用梯度计算，节省内存（推理时使用）
            outputs = model.generate(
                **inputs,
                max_length=request.max_length,
                temperature=request.temperature,
                top_p=request.top_p,
                do_sample=True
            )

        # 解码输出
        generated_text = tokenizer.decode(
            outputs[0],
            skip_special_tokens=True
        )

        return GenerationResponse(generated_text=generated_text)

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health():
    """健康检查"""
    return {"status": "healthy"}

# 启动服务
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

from fastapi import FastAPI
from concurrent.futures import ThreadPoolExecutor
import asyncio  # Python标准异步库

app = FastAPI()

# 线程池
executor = ThreadPoolExecutor(max_workers=4)  # ThreadPoolExecutor线程池并发执行

# 缓存
cache = {}

async def generate_async(request: GenerationRequest):
    """异步生成"""
    # 检查缓存
    cache_key = f"{request.text}_{request.max_length}"
    if cache_key in cache:
        return cache[cache_key]

    # 在线程池中执行
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(  # await等待异步操作完成
        executor,
        generate_sync,
        request
    )

    # 缓存结果
    cache[cache_key] = result

    return result

@app.post("/generate")
async def generate(request: GenerationRequest):
    """生成接口"""
    result = await generate_async(request)
    return result

from kubernetes import client, config

# Kubernetes自动扩展
config.load_kube_config()
autoscaling_api = client.AutoscalingV2Api()

# 配置自动扩展
scaling_policy = {
    "apiVersion": "autoscaling/v2",
    "kind": "HorizontalPodAutoscaler",
    "metadata": {
        "name": "llm-api-hpa"
    },
    "spec": {
        "scaleTargetRef": {
            "apiVersion": "apps/v1",
            "kind": "Deployment",
            "name": "llm-api"
        },
        "minReplicas": 1,
        "maxReplicas": 10,
        "metrics": [
            {
                "type": "Resource",
                "resource": {
                    "name": "cpu",
                    "target": {
                        "type": "Utilization",
                        "averageUtilization": 70
                    }
                }
            }
        ]
    }
}

autoscaling_api.create_namespaced_horizontal_pod_autoscaler(
    namespace="default",
    body=scaling_policy
)

from prometheus_client import Counter, Histogram, Gauge, start_http_server

# 定义指标
request_count = Counter('llm_requests_total', 'Total requests')  # Counter统计元素出现次数
request_latency = Histogram('llm_request_latency_seconds', 'Request latency')
gpu_utilization = Gauge('llm_gpu_utilization_percent', 'GPU utilization')

# 记录指标
@app.post("/generate")
async def generate(request: GenerationRequest):
    start_time = time.time()

    # 处理请求
    result = await generate_async(request)

    # 记录指标
    request_count.inc()
    request_latency.observe(time.time() - start_time)
    gpu_utilization.set(get_gpu_utilization())

    return result

# 启动Prometheus服务器
start_http_server(8001)

from transformers import AutoModelForCausalLM, BitsAndBytesConfig

# 量化配置
bnb_config = BitsAndBytesConfig(
    load_in_8bit=True
)

# 加载量化模型
model = AutoModelForCausalLM.from_pretrained(
    "gpt2",
    quantization_config=bnb_config
)

from fastapi import FastAPI
from transformers import AutoModelForCausalLM, AutoTokenizer

app = FastAPI()

# 加载模型
model = AutoModelForCausalLM.from_pretrained("gpt2")
tokenizer = AutoTokenizer.from_pretrained("gpt2")

@app.post("/generate")
async def generate(text: str):
    inputs = tokenizer(text, return_tensors="pt")
    outputs = model.generate(**inputs)
    return {"text": tokenizer.decode(outputs[0])}