Python 人工智能开发的核心范式：从"调包侠"到"系统架构师"

import asyncio
from dataclasses import dataclass
from typing import List, Optional
import chromadb
from sentence_transformers import SentenceTransformer
from transformers import pipeline

@dataclass
class Document:
    content: str
    metadata: dict
    embedding: Optional[List[float]] = None

class RAGSystem:
    def __init__(self, embed_model: str = "all-MiniLM-L6-v2"):
        # 1. 嵌入模型（本地运行，无网络依赖）
        self.embedder = SentenceTransformer(embed_model)
        
        # 2. 向量数据库（持久化存储）
        self.client = chromadb.PersistentClient(path="./vectordb")
        self.collection = self.client.get_or_create_collection("knowledge")
        
        # 3. 生成模型（可选用开源或 API）
        self.generator = pipeline(
            "text-generation",
            model="Qwen/Qwen2-7B-Instruct",
            device="cuda",
            max_new_tokens=512
        )
    
    def add_documents(self, docs: List[Document]):
        """批量插入文档"""
        embeddings = self.embedder.encode([d.content for d in docs])
        
        self.collection.add(
            documents=[d.content for d in docs],
            embeddings=embeddings.tolist(),
            metadatas=[d.metadata for d in docs],
            ids=[f"doc_{i}" for i in range(len(docs))]
        )
    
    def query(self, question: str, top_k: int = 3) -> str:
        """检索 + 生成"""
        # Step 1: 向量检索（确定性查询）
        query_embedding = self.embedder.encode([question])[0]
        results = self.collection.query(
            query_embeddings=[query_embedding.tolist()],
            n_results=top_k
        )
        
        # Step 2: 构建上下文
        context = "\n\n".join(results['documents'][0])
        
        # Step 3: 调用 LLM 生成（概率性输出）
        prompt = f"""基于以下参考信息回答问题：
        
参考信息：
{context}

问题：{question}
回答："""
        
        response = self.generator(prompt)[0]['generated_text']
        return response.replace(prompt, "").strip()

# 使用示例
rag = RAGSystem()
rag.add_documents([
    Document("Python 是解释型语言", {"source": "wiki"}),
    Document("AI 工程需要关注可观测性", {"source": "blog"})
])
print(rag.query("Python 是什么类型的语言？"))

import asyncio
from concurrent.futures import ThreadPoolExecutor

class AsyncAIService:
    def __init__(self):
        self.executor = ThreadPoolExecutor(max_workers=4)
        # 假设已加载多个模型
        self.models = {
            "classifier": load_classifier(),
            "summarizer": load_summarizer(),
            "translator": load_translator()
        }
    
    async def process_pipeline(self, text: str) -> dict:
        """并行执行多个 AI 任务"""
        loop = asyncio.get_event_loop()
        
        # 并发调用三个模型
        tasks = [
            loop.run_in_executor(self.executor, self.models["classifier"], text),
            loop.run_in_executor(self.executor, self.models["summarizer"], text),
            loop.run_in_executor(self.executor, self.models["translator"], text)
        ]
        
        # 等待所有任务完成
        classification, summary, translation = await asyncio.gather(*tasks)
        
        return {
            "classification": classification,
            "summary": summary,
            "translation": translation
        }

# 使用
service = AsyncAIService()
result = asyncio.run(service.process_pipeline("AI 正在改变世界"))

import time
import hashlib
import json
from functools import lru_cache
from typing import Optional
import redis

class ResilientLLM:
    def __init__(self, api_key: str):
        self.client = OpenAI(api_key=api_key)
        self.redis = redis.Redis(decode_responses=True)
        self.max_retries = 3
    
    def _cache_key(self, prompt: str, model: str) -> str:
        """基于内容和模型生成缓存键"""
        content = f"{prompt}|{model}"
        return f"llm_cache:{hashlib.md5(content.encode()).hexdigest()}"
    
    def generate(self, prompt: str, model: str = "gpt-3.5-turbo") -> Optional[str]:
        """带缓存 + 重试 + 超时的 LLM 调用"""
        
        # 1. 缓存命中
        cache_key = self._cache_key(prompt, model)
        cached = self.redis.get(cache_key)
        if cached:
            return json.loads(cached)
        
        # 2. 重试机制
        for attempt in range(self.max_retries):
            try:
                response = self.client.chat.completions.create(
                    model=model,
                    messages=[{"role": "user", "content": prompt}],
                    timeout=10.0  # 超时控制
                )
                result = response.choices[0].message.content
                
                # 存入缓存（TTL 1小时）
                self.redis.setex(cache_key, 3600, json.dumps(result))
                return result
                
            except Exception as e:
                wait = 2 ** attempt
                print(f"尝试 {attempt+1}/{self.max_retries} 失败: {e}")
                if attempt == self.max_retries - 1:
                    raise
                time.sleep(wait)
        
        return None

my_ai_project/
├── src/
│   ├── models/          # 模型封装
│   │   ├── embedder.py
│   │   └── generator.py
│   ├── pipelines/       # 业务流水线
│   │   ├── rag.py
│   │   └── agent.py
│   ├── services/        # API 服务
│   │   └── app.py
│   └── utils/           # 工具函数
│       ├── cache.py
│       └── metrics.py
├── tests/
│   ├── test_embedder.py
│   └── test_pipeline.py
├── configs/
│   └── model_config.yaml
├── requirements.txt
└── pyproject.toml       # 现代 Python 包管理

# 模型量化示例
from transformers import AutoModelForCausalLM, BitsAndBytesConfig

# 4-bit 量化加载
quant_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_use_double_quant=True
)

model = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Llama-2-7b-hf",
    quantization_config=quant_config,
    device_map="auto"
)
# 原本 13GB 显存需求，降到约 4GB

from vllm import LLM, SamplingParams

llm = LLM(model="Qwen/Qwen2-7B-Instruct", tensor_parallel_size=2)
outputs = llm.generate(["Hello world"], SamplingParams(temperature=0.7))