大模型时代的Python全能实战：从数据处理到模型调用与API开发

一篇博文带你打通Python在AI应用开发中的三大核心技能链

前言

在当今大模型和AI应用爆发的时代，Python早已不只是“胶水语言”，而是连接数据、算法与产品的核心纽带。无论是准备训练数据集、调用开源大模型，还是将模型能力封装成稳定的API服务，Python都扮演着不可替代的角色。

本文将围绕数据处理、模型调用、API开发三大板块，结合大模型应用的真实场景，带你构建一套完整的技术实战体系。无论你是想微调自己的LoRA模型，还是搭建基于RAG的智能问答系统，这套技能链都是你的必经之路。

---

一、数据处理：从原始数据到高质量语料

1.1 为什么数据处理是AI项目的命脉？

Garbage in, garbage out —— 这是AI领域永恒的真理。

在大模型时代，数据处理的复杂度被推向了新的高度：

• 需要处理PB级别的文本数据
• 需要从PDF、Word、网页等异构来源提取信息
• 需要为RAG系统准备高质量的向量化分块
• 需要为模型微调准备结构化指令数据

1.2 Python数据处理核心工具链

import pandas as pd
import numpy as np
import json
import re
from typing import List, Dict, Any
from pathlib import Path

1.2.1 文本清洗与预处理

def clean_text(text: str) -> str:
    """
    大模型文本数据清洗最佳实践
    """
    # 1. 去除不可见字符
    text = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]', '', text)
    
    # 2. 统一换行符
    text = text.replace('\r\n', '\n').replace('\r', '\n')
    
    # 3. 压缩多余空白（保留段落结构）
    text = re.sub(r' +', ' ', text)
    
    # 4. 移除过多空行（最多保留两个连续换行）
    text = re.sub(r'\n{3,}', '\n\n', text)
    
    # 5. 全角转半角（英文、数字）
    text = text.replace('．', '.').replace('，', ',').replace('（', '(').replace('）', ')')
    
    return text.strip()

def process_large_text_file(
    input_path: str, 
    output_path: str, 
    chunk_size: int = 10000
):
    """
    流式处理大规模文本文件，防止OOM
    """
    with open(input_path, 'r', encoding='utf-8') as f_in, \
         open(output_path, 'w', encoding='utf-8') as f_out:
        
        buffer = []
        for line in f_in:
            if line.strip():
                cleaned = clean_text(line)
                buffer.append(cleaned)
            
            if len(buffer) >= chunk_size:
                f_out.write('\n'.join(buffer) + '\n')
                buffer = []
        
        # 写入剩余数据
        if buffer:
            f_out.write('\n'.join(buffer))

1.2.2 为RAG系统准备数据：智能分块策略

from typing import List, Tuple
import tiktoken

class SmartChunker:
    """
    语义感知的智能分块器
    比传统固定长度分块检索效果提升30%+
    """
    
    def __init__(self, model_name: str = "gpt-4", max_tokens: int = 512):
        self.encoder = tiktoken.encoding_for_model(model_name)
        self.max_tokens = max_tokens
    
    def chunk_by_semantic(self, text: str) -> List[str]:
        """
        按语义边界进行分块
        """
        # 按段落、句子边界识别语义单元
        sentences = self._split_sentences(text)
        
        chunks = []
        current_chunk = []
        current_tokens = 0
        
        for sent in sentences:
            sent_tokens = len(self.encoder.encode(sent))
            
            if current_tokens + sent_tokens > self.max_tokens:
                # 保存当前chunk
                if current_chunk:
                    chunks.append(' '.join(current_chunk))
                # 如果单个句子超长，强制截断
                if sent_tokens > self.max_tokens:
                    chunks.append(self._truncate_by_tokens(sent))
                    current_chunk = []
                    current_tokens = 0
                else:
                    current_chunk = [sent]
                    current_tokens = sent_tokens
            else:
                current_chunk.append(sent)
                current_tokens += sent_tokens
        
        if current_chunk:
            chunks.append(' '.join(current_chunk))
        
        return chunks
    
    def _split_sentences(self, text: str) -> List[str]:
        """
        中英文混合句子切分
        """
        # 使用正则捕获中英文句子边界
        pattern = r'([。！？\.\?\!…\n]+)'
        parts = re.split(pattern, text)
        
        sentences = []
        for i in range(0, len(parts)-1, 2):
            sent = parts[i] + (parts[i+1] if i+1 < len(parts) else '')
            if sent.strip():
                sentences.append(sent.strip())
        
        return sentences
    
    def _truncate_by_tokens(self, text: str) -> str:
        tokens = self.encoder.encode(text)
        truncated = tokens[:self.max_tokens]
        return self.encoder.decode(truncated)

1.2.3 构建指令微调数据集

def build_instruction_dataset(
    raw_data: List[Dict[str, Any]],
    instruction_template: str,
    output_path: str
):
    """
    将原始数据转换为LLaMA-Factory / HuggingFace支持的指令格式
    """
    processed = []
    
    for item in raw_data:
        # 构建instruction-input-output三元组
        entry = {
            "instruction": instruction_template.format(**item),
            "input": item.get("input", ""),
            "output": item.get("output", ""),
            "history": item.get("history", [])  # 支持多轮对话
        }
        processed.append(entry)
    
    # 输出为jsonl格式（支持流式读取）
    with open(output_path, 'w', encoding='utf-8') as f:
        for entry in processed:
            f.write(json.dumps(entry, ensure_ascii=False) + '\n')
    
    print(f"✅ 数据已保存至 {output_path}，共 {len(processed)} 条样本")
    return processed

---

二、模型调用：从加载到推理的艺术

2.1 模型加载的工业化实践

import torch
from transformers import (
    AutoModelForCausalLM, 
    AutoTokenizer,
    BitsAndBytesConfig,
    pipeline
)

class LLMLoader:
    """
    支持多模式加载的大模型加载器
    兼顾灵活性与性能
    """
    
    @staticmethod
    def load_for_inference(
        model_name: str,
        use_4bit: bool = True,
        use_flash_attn: bool = True,
        device_map: str = "auto"
    ):
        """
        加载推理模型的最佳实践
        """
        # 量化配置（节省显存）
        if use_4bit:
            bnb_config = BitsAndBytesConfig(
                load_in_4bit=True,
                bnb_4bit_use_double_quant=True,
                bnb_4bit_quant_type="nf4",
                bnb_4bit_compute_dtype=torch.bfloat16,
                bnb_4bit_quant_storage_dtype=torch.uint8,
            )
        else:
            bnb_config = None
        
        # 模型加载参数
        model_kwargs = {
            "device_map": device_map,
            "torch_dtype": torch.bfloat16,
            "attn_implementation": "flash_attention_2" if use_flash_attn else None,
            "quantization_config": bnb_config,
        }
        
        # 加载模型
        model = AutoModelForCausalLM.from_pretrained(
            model_name,
            **model_kwargs
        )
        
        # 加载tokenizer（统一设置padding_side）
        tokenizer = AutoTokenizer.from_pretrained(
            model_name,
            trust_remote_code=True,
            padding_side="left"
        )
        
        # 确保有pad_token
        if tokenizer.pad_token is None:
            tokenizer.pad_token = tokenizer.eos_token
        
        return model, tokenizer
    
    @staticmethod
    def load_for_training(
        model_name: str,
        use_gradient_checkpointing: bool = True
    ):
        """
        加载用于训练/微调的模型
        """
        model, tokenizer = LLMLoader.load_for_inference(
            model_name, 
            use_4bit=True
        )
        
        # 启用梯度检查点（节省显存）
        if use_gradient_checkpointing:
            model.gradient_checkpointing_enable()
        
        # 冻结部分层（可选）
        # for param in model.model.embed_tokens.parameters():
        #     param.requires_grad = False
        
        return model, tokenizer

2.2 高性能推理：批处理与流式输出

import asyncio
from transformers import TextIteratorStreamer
from threading import Thread

class InferenceEngine:
    """
    高性能推理引擎
    支持批处理、流式输出、异步调用
    """
    
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer
        
    def batch_generate(
        self,
        prompts: List[str],
        max_new_tokens: int = 512,
        temperature: float = 0.7,
        top_p: float = 0.9,
        batch_size: int = 4
    ) -> List[str]:
        """
        批量生成（显著提升吞吐量）
        """
        results = []
        
        for i in range(0, len(prompts), batch_size):
            batch_prompts = prompts[i:i+batch_size]
            
            # Tokenize批量输入
            inputs = self.tokenizer(
                batch_prompts,
                return_tensors="pt",
                padding=True,
                truncation=True,
                max_length=2048
            )
            
            # 移到GPU
            inputs = {k: v.to(self.model.device) for k, v in inputs.items()}
            
            # 生成
            with torch.no_grad():
                outputs = self.model.generate(
                    **inputs,
                    max_new_tokens=max_new_tokens,
                    temperature=temperature,
                    top_p=top_p,
                    do_sample=True,
                    pad_token_id=self.tokenizer.pad_token_id,
                    eos_token_id=self.tokenizer.eos_token_id,
                )
            
            # 解码
            batch_results = self.tokenizer.batch_decode(
                outputs, 
                skip_special_tokens=True
            )
            
            # 提取新生成的文本
            for prompt, result in zip(batch_prompts, batch_results):
                # 移除原始prompt
                new_text = result[len(prompt):].lstrip()
                results.append(new_text)
        
        return results
    
    def stream_generate(self, prompt: str, **kwargs):
        """
        流式生成（实时输出，提升用户体验）
        """
        streamer = TextIteratorStreamer(
            self.tokenizer,
            skip_prompt=True,
            skip_special_tokens=True
        )
        
        inputs = self.tokenizer(
            prompt, 
            return_tensors="pt"
        ).to(self.model.device)
        
        # 在单独线程中执行生成
        generation_kwargs = dict(
            inputs,
            streamer=streamer,
            max_new_tokens=kwargs.get('max_new_tokens', 512),
            temperature=kwargs.get('temperature', 0.7),
            top_p=kwargs.get('top_p', 0.9),
            do_sample=True,
            pad_token_id=self.tokenizer.pad_token_id,
            eos_token_id=self.tokenizer.eos_token_id,
        )
        
        def generate():
            self.model.generate(**generation_kwargs)
        
        thread = Thread(target=generate)
        thread.start()
        
        # 实时yield生成的文本片段
        for text in streamer:
            yield text

2.3 模型调用的工程化封装

from functools import lru_cache
from datetime import datetime
import logging

class ModelService:
    """
    生产级模型服务封装
    包含：缓存、日志、重试、降级
    """
    
    def __init__(self, model_name: str):
        self.model_name = model_name
        self.logger = self._setup_logger()
        
        # 加载模型
        self.model, self.tokenizer = LLMLoader.load_for_inference(model_name)
        self.engine = InferenceEngine(self.model, self.tokenizer)
        
    def _setup_logger(self):
        logger = logging.getLogger(f"ModelService_{self.model_name}")
        logger.setLevel(logging.INFO)
        handler = logging.StreamHandler()
        handler.setFormatter(logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        ))
        logger.addHandler(handler)
        return logger
    
    @lru_cache(maxsize=1000)
    def predict_with_cache(self, prompt: str, **kwargs) -> str:
        """
        带缓存的预测（相同输入直接返回缓存结果）
        """
        return self._predict(prompt, **kwargs)
    
    def _predict(self, prompt: str, **kwargs) -> str:
        """
        核心预测逻辑（带重试机制）
        """
        max_retries = 3
        for attempt in range(max_retries):
            try:
                start_time = datetime.now()
                
                # 执行生成
                results = self.engine.batch_generate(
                    [prompt], 
                    **kwargs
                )
                
                elapsed = (datetime.now() - start_time).total_seconds()
                self.logger.info(
                    f"Generation completed in {elapsed:.2f}s, "
                    f"output_length={len(results[0])}"
                )
                
                return results[0]
            
            except Exception as e:
                self.logger.warning(f"Attempt {attempt+1} failed: {str(e)}")
                if attempt == max_retries - 1:
                    # 降级方案：返回友好的错误提示
                    return "抱歉，我暂时无法处理您的请求，请稍后再试。"
                continue
    
    async def apredict(self, prompt: str, **kwargs) -> str:
        """
        异步预测接口
        """
        return await asyncio.to_thread(self._predict, prompt, **kwargs)

---

三、API开发：将模型能力产品化

3.1 基于FastAPI的现代API架构

from fastapi import FastAPI, HTTPException, BackgroundTasks
from fastapi.responses import StreamingResponse, JSONResponse
from pydantic import BaseModel, Field
from typing import Optional, List
import uvicorn
import time
from contextlib import asynccontextmanager

# ---------- 数据模型定义 ----------
class GenerateRequest(BaseModel):
    prompt: str = Field(..., description="输入提示词", min_length=1)
    max_tokens: int = Field(512, ge=1, le=4096, description="最大生成token数")
    temperature: float = Field(0.7, ge=0.0, le=2.0, description="温度参数")
    top_p: float = Field(0.9, ge=0.0, le=1.0, description="Top-p采样")
    stream: bool = Field(False, description="是否流式输出")
    cache: bool = Field(True, description="是否使用缓存")

class GenerateResponse(BaseModel):
    text: str
    tokens_used: int
    processing_time: float
    model: str

class HealthResponse(BaseModel):
    status: str
    model_name: str
    device: str
    gpu_memory: float

# ---------- 应用生命周期管理 ----------
@asynccontextmanager
async def lifespan(app: FastAPI):
    """
    管理应用的生命周期
    - 启动时加载模型
    - 关闭时释放资源
    """
    # 启动时加载模型
    print("🚀 Loading model...")
    app.state.model_service = ModelService("Qwen/Qwen2.5-7B-Instruct")
    print("✅ Model loaded successfully!")
    
    yield
    
    # 关闭时清理
    print("🔄 Cleaning up resources...")
    # 这里可以添加显存清理等操作
    if hasattr(app.state.model_service, 'model'):
        del app.state.model_service.model
    torch.cuda.empty_cache()
    print("✅ Cleanup complete!")

# ---------- 创建FastAPI应用 ----------
app = FastAPI(
    title="LLM API Service",
    description="高性能大模型推理API服务",
    version="1.0.0",
    lifespan=lifespan
)

# ---------- API端点 ----------
@app.get("/health", response_model=HealthResponse)
async def health_check():
    """
    健康检查接口
    """
    service = app.state.model_service
    return HealthResponse(
        status="healthy",
        model_name=service.model_name,
        device=str(service.model.device),
        gpu_memory=torch.cuda.memory_allocated(service.model.device) / 1024**3 if torch.cuda.is_available() else 0
    )

@app.post("/generate", response_model=GenerateResponse)
async def generate(
    request: GenerateRequest,
    background_tasks: BackgroundTasks
):
    """
    同步生成接口
    """
    service = app.state.model_service
    
    try:
        start_time = time.time()
        
        # 根据缓存参数选择调用方式
        if request.cache:
            text = service.predict_with_cache(
                request.prompt,
                max_new_tokens=request.max_tokens,
                temperature=request.temperature,
                top_p=request.top_p
            )
        else:
            text = service._predict(
                request.prompt,
                max_new_tokens=request.max_tokens,
                temperature=request.temperature,
                top_p=request.top_p
            )
        
        elapsed = time.time() - start_time
        
        # 记录使用日志（后台任务）
        background_tasks.add_task(
            log_usage,
            request.prompt,
            text,
            elapsed
        )
        
        return GenerateResponse(
            text=text,
            tokens_used=len(service.tokenizer.encode(text)),
            processing_time=elapsed,
            model=service.model_name
        )
    
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/generate/stream")
async def generate_stream(request: GenerateRequest):
    """
    流式生成接口（SSE）
    """
    if not request.stream:
        return JSONResponse(
            status_code=400,
            content={"error": "请设置 stream=true 以使用流式接口"}
        )
    
    service = app.state.model_service
    
    def event_stream():
        """生成SSE事件流"""
        try:
            for text_chunk in service.engine.stream_generate(
                request.prompt,
                max_new_tokens=request.max_tokens,
                temperature=request.temperature,
                top_p=request.top_p
            ):
                # SSE格式
                yield f"data: {json.dumps({'text': text_chunk}, ensure_ascii=False)}\n\n"
            
            yield f"data: {json.dumps({'done': True})}\n\n"
            
        except Exception as e:
            yield f"data: {json.dumps({'error': str(e)})}\n\n"
    
    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream"
    )

# ---------- 辅助函数 ----------
async def log_usage(prompt: str, response: str, elapsed: float):
    """
    异步记录使用日志（可对接MySQL/ClickHouse）
    """
    # 这里可以写入数据库或日志系统
    print(f"[USAGE] prompt_len={len(prompt)}, response_len={len(response)}, elapsed={elapsed:.2f}s")

3.2 API的工程化加固

from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from slowapi import Limiter, _rate_limit_exceeded
from slowapi.util import get_remote_address
import asyncio

# ---------- 安全与性能中间件 ----------
# CORS配置
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 生产环境请限定具体域名
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 限流器
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

@app.post("/generate")
@limiter.limit("5/minute")  # 每IP每分钟最多5次请求
async def generate_with_rate_limit(request: GenerateRequest):
    """
    带限流的生成接口
    """
    return await generate(request)

# ---------- 生产级启动脚本 ----------
if __name__ == "__main__":
    uvicorn.run(
        "main:app",
        host="0.0.0.0",
        port=8000,
        workers=4,  # 多worker部署
        reload=False,
        log_level="info",
        limit_concurrency=100,  # 最大并发连接数
        timeout_keep_alive=60,
    )

3.3 客户端SDK封装（对外提供）

import requests
import time
from typing import Iterator, Optional

class LLMClient:
    """
    API客户端封装
    提供同步、异步、流式三种调用方式
    """
    
    def __init__(self, base_url: str = "http://localhost:8000", timeout: int = 60):
        self.base_url = base_url.rstrip('/')
        self.timeout = timeout
        self.session = requests.Session()
    
    def generate(
        self,
        prompt: str,
        max_tokens: int = 512,
        temperature: float = 0.7,
        top_p: float = 0.9,
        stream: bool = False,
        cache: bool = True
    ) -> str:
        """同步生成"""
        endpoint = "/generate/stream" if stream else "/generate"
        
        response = self.session.post(
            f"{self.base_url}{endpoint}",
            json={
                "prompt": prompt,
                "max_tokens": max_tokens,
                "temperature": temperature,
                "top_p": top_p,
                "stream": stream,
                "cache": cache
            },
            timeout=self.timeout,
            stream=stream
        )
        
        if stream:
            # 流式响应，返回生成器
            return self._handle_stream_response(response)
        else:
            response.raise_for_status()
            data = response.json()
            return data.get("text", "")
    
    def _handle_stream_response(self, response):
        """处理流式响应"""
        for line in response.iter_lines():
            if line:
                line = line.decode('utf-8')
                if line.startswith('data: '):
                    data = json.loads(line[6:])
                    if data.get('done'):
                        break
                    if 'text' in data:
                        yield data['text']
    
    async def async_generate(self, prompt: str, **kwargs) -> str:
        """异步生成"""
        import aiohttp
        
        async with aiohttp.ClientSession() as session:
            async with session.post(
                f"{self.base_url}/generate",
                json={"prompt": prompt, **kwargs},
                timeout=aiohttp.ClientTimeout(total=self.timeout)
            ) as response:
                data = await response.json()
                return data.get("text", "")

# 使用示例
if __name__ == "__main__":
    client = LLMClient()
    
    # 同步调用
    result = client.generate("请用Python写一个快速排序算法")
    print(f"结果: {result}")
    
    # 流式调用
    print("流式输出:")
    for chunk in client.generate("讲一个笑话", stream=True):
        print(chunk, end="", flush=True)

---

四、三大技能链的融合实战：构建一个完整的智能问答系统

4.1 系统架构全景

┌─────────────┐      ┌─────────────┐      ┌─────────────┐
│  数据层      │─────▶│  模型层      │─────▶│  服务层      │
│             │      │             │      │             │
│ • MySQL     │      │ • LLM       │      │ • FastAPI   │
│ • Crawler   │      │ • Embedding │      │ • 限流/缓存 │
│ • Parser    │      │ • RAG       │      │ • 日志监控 │
│ • Datasets  │      │ • LoRA      │      │ • SDK      │
└─────────────┘      └─────────────┘      └─────────────┘

4.2 代码实战：RAG + LLM API

from langchain_community.vectorstores import FAISS
from langchain_huggingface import HuggingFaceEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document
import sqlalchemy as sa
from sqlalchemy import create_engine, text

class SmartQASystem:
    """
    融合数据处理 + 模型调用 + API的三位一体系统
    """
    
    def __init__(self, model_api_url: str):
        self.embedding_model = HuggingFaceEmbeddings(
            model_name="BAAI/bge-small-zh-v1.5",
            encode_kwargs={'normalize_embeddings': True}
        )
        self.vector_store = None
        self.api_client = LLMClient(model_api_url)
        self.db_engine = create_engine("mysql+pymysql://user:pass@localhost/qa_system")
    
    # ---------- 数据处理能力 ----------
    def load_and_index_documents(self, file_paths: List[str]):
        """
        加载文档并建立向量索引（RAG核心）
        """
        documents = []
        for path in file_paths:
            # 使用前面的SmartChunker进行分块
            with open(path, 'r', encoding='utf-8') as f:
                text = f.read()
            
            chunker = SmartChunker(max_tokens=512)
            chunks = chunker.chunk_by_semantic(text)
            
            for chunk in chunks:
                documents.append(Document(
                    page_content=chunk,
                    metadata={"source": path}
                ))
        
        # 建立向量存储
        self.vector_store = FAISS.from_documents(
            documents, 
            self.embedding_model
        )
        return len(documents)
    
    def sync_with_mysql(self, table_name: str):
        """
        从MySQL同步数据到知识库
        """
        with self.db_engine.connect() as conn:
            result = conn.execute(text(f"SELECT * FROM {table_name}"))
            rows = result.fetchall()
            
            for row in rows:
                # 将数据库记录转换为文档
                doc_text = f"{row.title}\n{row.content}"
                self.vector_store.add_texts(
                    [doc_text],
                    metadatas=[{"id": row.id, "source": "mysql"}]
                )
        
        self.vector_store.save_local("faiss_index")
    
    # ---------- 模型调用能力 ----------
    def answer_with_rag(self, question: str) -> str:
        """
        使用RAG增强的问答
        """
        # 1. 检索相关文档
        if self.vector_store is None:
            self.vector_store = FAISS.load_local(
                "faiss_index", 
                self.embedding_model,
                allow_dangerous_deserialization=True
            )
        
        docs = self.vector_store.similarity_search(question, k=3)
        context = "\n\n".join([doc.page_content for doc in docs])
        
        # 2. 构建增强提示
        enhanced_prompt = f"""基于以下参考信息回答问题：
        [参考信息]
        {context}
        
        [问题]
        {question}
        
        [回答]
        """
        
        # 3. 调用LLM API
        response = self.api_client.generate(enhanced_prompt, max_tokens=1024)
        return response
    
    # ---------- API开发能力 ----------
    def start_api(self):
        """
        将这个问答系统本身封装为API
        """
        from fastapi import FastAPI
        app = FastAPI()
        
        @app.post("/qa")
        async def qa_endpoint(question: str):
            answer = self.answer_with_rag(question)
            return {"question": question, "answer": answer}
        
        return app

---

五、进阶：LoRA微调的数据准备 + 模型训练 + API部署全流程

# 完整流程示例
def full_lora_pipeline():
    """
    从数据到API的完整LoRA微调流程
    """
    
    # ==================== 第一步：数据处理 ====================
    print("📊 Step 1: 准备训练数据")
    
    # 从各个来源收集数据
    from crawler import WebCrawler  # 假设自定义爬虫
    crawler = WebCrawler()
    raw_data = crawler.fetch_domain_data("https://your-domain.com/docs")
    
    # 清洗并转换为指令格式
    instruction_data = build_instruction_dataset(
        raw_data,
        instruction_template="请回答以下问题：{input}",
        output_path="train_data.jsonl"
    )
    
    # ==================== 第二步：模型训练 ====================
    print("🚀 Step 2: 执行LoRA微调")
    
    # 使用PEFT进行LoRA训练（这是简化示例）
    # 实际项目中会配合DeepSpeed等技术
    from peft import LoraConfig, get_peft_model
    from transformers import TrainingArguments, Trainer
    
    model, tokenizer = LLMLoader.load_for_training("Qwen/Qwen2.5-7B")
    
    lora_config = LoraConfig(
        r=16,
        lora_alpha=32,
        target_modules=["q_proj", "v_proj", "k_proj", "o_proj"],
        lora_dropout=0.1,
        bias="none",
        task_type="CAUSAL_LM"
    )
    
    model = get_peft_model(model, lora_config)
    model.print_trainable_parameters()
    
    # 训练（简化）
    training_args = TrainingArguments(
        output_dir="./lora_checkpoints",
        per_device_train_batch_size=4,
        gradient_accumulation_steps=4,
        num_train_epochs=3,
        learning_rate=2e-4,
        fp16=True,
        save_strategy="epoch",
    )
    
    # trainer = Trainer(...)
    # trainer.train()
    # model.save_pretrained("./final_lora_model")
    
    # ==================== 第三步：API部署 ====================
    print("🔧 Step 3: 部署为API服务")
    
    # 加载微调后的模型
    from peft import PeftModel
    
    base_model, tokenizer = LLMLoader.load_for_inference("Qwen/Qwen2.5-7B")
    finetuned_model = PeftModel.from_pretrained(base_model, "./final_lora_model")
    
    # 启动API服务（使用前面定义的FastAPI应用）
    # uvicorn.run(app, host="0.0.0.0", port=8000)
    
    print("✅ 全流程完成！")

---

六、性能优化与最佳实践清单

6.1 数据处理优化

• ✅ 使用pandas.read_csv()的chunksize参数处理大文件
• ✅ 对于超大数据集，使用pyarrow或parquet格式
• ✅ 文本清洗使用re编译正则对象以提高性能

6.2 模型推理优化

• ✅ 使用torch.compile()对模型进行JIT编译（PyTorch 2.0+）
• ✅ 开启Flash Attention 2支持
• ✅ 使用vLLM或TGI框架处理高并发场景
• ✅ 使用continuous batching提升吞吐量

6.3 API服务优化

• ✅ 使用Gunicorn + Uvicorn组合部署
• ✅ 配置Nginx反向代理和负载均衡
• ✅ 使用Redis做分布式缓存
• ✅ 引入Prometheus + Grafana做监控告警

---

七、总结

本文从数据处理、模型调用、API开发三个维度，系统地展示了Python在大模型时代的全能实战能力：

技能维度	核心能力	典型工具/技术
数据处理	清洗、分块、格式化、向量化	pandas, re, tiktoken, LangChain
模型调用	加载、推理、流式输出、缓存	transformers, PEFT, vLLM
API开发	接口设计、限流、部署、SDK	FastAPI, Uvicorn, Nginx, Redis

这三个技能环环相扣：数据处理为模型提供高质量输入，模型调用是核心智能引擎，而API开发则将这一切转化为可用的产品服务。掌握这三者的结合，就掌握了从AI idea到product的全链路能力。

---

📚 参考资料与扩展阅读

1. HuggingFace Transformers 文档
2. FastAPI 官方文档
3. PEFT 微调技术
4. LangChain 开发指南
5. Flash Attention 论文

原创声明：本文为CSDN博主原创文章，未经允许不得转载。如有问题或建议，欢迎在评论区交流讨论。

---

最后更新：2026年6月