大模型AI服务工程化部署与运维实践：推理性能调优、接口开发、监控告警全流程实操25.0

# 大模型推理优化部署完整代码
# 安装依赖：pip install torch transformers accelerate bitsandbytes sentencepiece
import torch
from transformers import AutoTokenizer, AutoModelForCausalLM, BitsAndBytesConfig
from accelerate import dispatch_model, infer_auto_device_map

# 1. 配置量化参数（INT4量化，显存最优优化）
bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,  # 开启4位量化
    bnb_4bit_use_double_quant=True,  # 二次量化，进一步压缩显存
    bnb_4bit_quant_type="nf4",  # 最优量化类型，适配大模型
    bnb_4bit_compute_dtype=torch.bfloat16  # 计算精度，平衡速度与效果
)

# 2. 模型与分词器加载
model_path = "./qwen-7b-chat"  # 本地模型路径
tokenizer = AutoTokenizer.from_pretrained(
    model_path,
    trust_remote_code=True,
    padding_side="right"
)
tokenizer.pad_token = tokenizer.eos_token

# 3. 加载模型并开启推理优化
model = AutoModelForCausalLM.from_pretrained(
    model_path,
    quantization_config=bnb_config,
    device_map="auto",  # 自动设备分配
    torch_dtype=torch.bfloat16,
    trust_remote_code=True,
    use_cache=True  # 开启KV Cache缓存
)

# 4. 推理生成核心函数（支持批量推理、上下文缓存）
def llm_infer(prompt_list, max_new_tokens=512, temperature=0.7):
    """
    批量推理函数
    :param prompt_list: 对话请求列表
    :param max_new_tokens: 最大生成token数
    :param temperature: 随机性系数
    :return: 推理结果列表
    """
    # 批量编码输入
    inputs = tokenizer(
        prompt_list,
        return_tensors="pt",
        padding=True,
        truncation=True,
        max_length=2048
    ).to("cuda")
    
    # 推理生成（开启优化参数）
    with torch.no_grad():  # 关闭梯度计算，节省显存
        outputs = model.generate(
            **inputs,
            max_new_tokens=max_new_tokens,
            temperature=temperature,
            top_p=0.95,
            do_sample=True,
            pad_token_id=tokenizer.eos_token_id,
            use_cache=True,
            num_return_sequences=1
        )
    
    # 批量解码结果
    res_list = [tokenizer.decode(output, skip_special_tokens=True) for output in outputs]
    return res_list

# 5. 测试推理效果
if __name__ == "__main__":
    test_prompts = ["什么是大模型工程化部署？", "AI服务运维核心工作有哪些？"]
    results = llm_infer(test_prompts)
    for idx, res in enumerate(results):
        print(f"请求{idx+1}结果：{res}")

# 大模型生产级API服务开发完整代码
# 安装依赖：pip install fastapi uvicorn slowapi python-multipart
from fastapi import FastAPI, HTTPException, Request, Header
from fastapi.responses import StreamingResponse
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
import time
import logging
from typing import Optional, Dict, Any

# 1. 初始化日志、限流、服务实例
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
limiter = Limiter(key_func=get_remote_address)
app = FastAPI(title="大模型AI服务接口", version="1.0")
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

# 2. 全局配置
API_TOKEN = "llm-service-2026"  # 接口鉴权密钥
LIMITS = "20/minute"  # 单IP每分钟20次请求限制

# 3. 鉴权依赖函数
async def verify_token(authorization: Optional[str] = Header(None)):
    if not authorization or authorization != f"Bearer {API_TOKEN}":
        raise HTTPException(status_code=401, detail="接口鉴权失败，无权访问")
    return True

# 4. 继承前文的推理函数（此处复用优化后的llm_infer）
from infer_opt import llm_infer

# 5. 普通问答接口（非流式）
@app.post("/api/llm/chat")
@limiter.limit(LIMITS)
async def llm_chat(request: Request, token: bool = Depends(verify_token)):
    try:
        # 解析请求参数
        data = await request.json()
        prompt = data.get("prompt", "")
        max_tokens = data.get("max_tokens", 512)
        temperature = data.get("temperature", 0.7)
        
        if not prompt:
            raise HTTPException(status_code=400, detail="请求参数不能为空")
        
        # 执行推理
        start_time = time.time()
        result = llm_infer([prompt], max_tokens, temperature)[0]
        cost_time = round(time.time() - start_time, 2)
        
        # 记录请求日志
        logging.info(f"请求成功，耗时：{cost_time}s，prompt：{prompt[:50]}")
        
        return {
            "code": 200,
            "msg": "success",
            "data": {
                "result": result,
                "cost_time": cost_time
            }
        }
    except Exception as e:
        logging.error(f"请求异常：{str(e)}")
        raise HTTPException(status_code=500, detail=f"服务推理异常：{str(e)}")

# 6. 流式问答接口（核心生产接口）
@app.post("/api/llm/chat/stream")
@limiter.limit(LIMITS)
async def llm_chat_stream(request: Request, token: bool = Depends(verify_token)):
    try:
        data = await request.json()
        prompt = data.get("prompt", "")
        if not prompt:
            raise HTTPException(status_code=400, detail="请求参数不能为空")

        # 流式生成器
        def stream_generator():
            inputs = tokenizer(prompt, return_tensors="pt").to("cuda")
            for output in model.generate_stream(
                **inputs, max_new_tokens=512, temperature=0.7, use_cache=True
            ):
                chunk = tokenizer.decode(output, skip_special_tokens=True)
                yield f"data: {chunk}\n\n"
        
        return StreamingResponse(stream_generator(), media_type="text/event-stream")
    except Exception as e:
        logging.error(f"流式请求异常：{str(e)}")
        raise HTTPException(status_code=500, detail="流式推理服务异常")

# 7. 健康检查接口（运维必备）
@app.get("/health")
async def health_check():
    return {"code": 200, "status": "running", "time": time.strftime("%Y-%m-%d %H:%M:%S")}

if __name__ == "__main__":
    import uvicorn
    # 启动服务：支持多进程、高并发
    uvicorn.run(app, host="0.0.0.0", port=8000, workers=2)

# 大模型AI服务自定义监控+告警代码
# 安装依赖：pip install prometheus-client psutil pynvml requests
import time
import psutil
import pynvml
import requests
from prometheus_client import Counter, Gauge, Histogram, start_http_server

# 1. 初始化GPU监控工具
pynvml.nvmlInit()
gpu_handle = pynvml.nvmlDeviceGetHandleByIndex(0)

# 2. 定义Prometheus监控指标
# 请求总量
LLM_REQUEST_TOTAL = Counter("llm_request_total", "大模型总请求数", ["status"])
# GPU资源指标
LLM_GPU_MEM = Gauge("llm_gpu_mem_usage", "GPU显存使用率")
LLM_GPU_UTIL = Gauge("llm_gpu_util", "GPU算力利用率")
# 推理延迟指标
LLM_INFER_LATENCY = Histogram("llm_infer_latency", "大模型推理延迟(秒)")

# 3. 钉钉告警推送函数
def ding_alert(msg: str):
    """异常告警推送至钉钉群"""
    webhook = "https://oapi.dingtalk.com/robot/send?access_token=你的钉钉机器人token"
    headers = {"Content-Type": "application/json"}
    data = {
        "msgtype": "text",
        "text": {"content": f"【大模型服务告警】{msg}"}
    }
    try:
        requests.post(webhook, json=data, headers=headers, timeout=5)
    except Exception as e:
        print("告警推送失败：", e)

# 4. 资源指标采集函数
def collect_resource_metric():
    while True:
        # 采集GPU指标
        mem_info = pynvml.nvmlDeviceGetMemoryInfo(gpu_handle)
        mem_usage = mem_info.used / mem_info.total * 100
        gpu_util = pynvml.nvmlDeviceGetUtilizationRates(gpu_handle).gpu

        # 更新监控指标
        LLM_GPU_MEM.set(mem_usage)
        LLM_GPU_UTIL.set(gpu_util)

        # 异常告警判断
        if mem_usage > 90:
            ding_alert(f"GPU显存使用率过高！当前使用率：{round(mem_usage,2)}%，临近溢出风险")
        if gpu_util < 10:
            ding_alert(f"GPU算力利用率过低！当前利用率：{gpu_util}%，资源严重浪费")
        
        time.sleep(5)  # 5秒采集一次

# 5. 推理延迟监控装饰器
def monitor_infer_latency(func):
    def wrapper(*args, **kwargs):
        start = time.time()
        res = func(*args, **kwargs)
        latency = time.time() - start
        LLM_INFER_LATENCY.observe(latency)
        # 延迟过高告警
        if latency > 10:
            ding_alert(f"推理延迟过高！当前延迟：{round(latency,2)}s")
        return res
    return wrapper

# 6. 请求状态监控
def update_request_status(status: str):
    """status: success/fail/timeout"""
    LLM_REQUEST_TOTAL.labels(status=status).inc()

# 7. 启动监控服务
if __name__ == "__main__":
    # 开启prometheus指标端口
    start_http_server(8090)
    print("监控服务启动成功，端口：8090")
    # 启动资源采集线程
    import threading
    threading.Thread(target=collect_resource_metric, daemon=True).start()
    while True:
        time.sleep(1)