AI开发工具与平台生态系统

import pandas as pd
from sklearn.preprocessing import StandardScaler

# 数据加载和预处理
def preprocess_data(file_path):
    # 读取数据
    df = pd.read_csv(file_path)
    
    # 数据清洗
    df = df.dropna()  # 删除缺失值
    df = df.drop_duplicates()  # 删除重复值
    
    # 特征工程
    scaler = StandardScaler()
    numerical_cols = df.select_dtypes(include=['number']).columns
    df[numerical_cols] = scaler.fit_transform(df[numerical_cols])
    
    return df

# 使用示例
processed_data = preprocess_data('dataset.csv')

import torch
import torch.nn as nn
from transformers import AutoTokenizer, AutoModel

class TextClassifier(nn.Module):
    def __init__(self, model_name, num_classes):
        super().__init__()
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModel.from_pretrained(model_name)
        self.classifier = nn.Linear(self.model.config.hidden_size, num_classes)
        
    def forward(self, input_ids, attention_mask):
        outputs = self.model(input_ids=input_ids, attention_mask=attention_mask)
        pooled_output = outputs.last_hidden_state[:, 0]  # CLS token
        logits = self.classifier(pooled_output)
        return logits

# 模型实例化
model = TextClassifier('bert-base-uncased', num_classes=2)

import mlflow
import mlflow.pytorch

# MLflow实验跟踪
def train_with_tracking(model, train_loader, val_loader, epochs):
    with mlflow.start_run():
        # 记录参数
        mlflow.log_param("epochs", epochs)
        mlflow.log_param("learning_rate", 0.001)
        
        for epoch in range(epochs):
            # 训练循环
            train_loss = train_epoch(model, train_loader)
            val_acc = validate(model, val_loader)
            
            # 记录指标
            mlflow.log_metric("train_loss", train_loss, step=epoch)
            mlflow.log_metric("val_accuracy", val_acc, step=epoch)
        
        # 记录模型
        mlflow.pytorch.log_model(model, "model")

import horovod.torch as hvd

# 初始化Horovod
hvd.init()

# 设置GPU
torch.cuda.set_device(hvd.local_rank())

# 模型和优化器
model = MyModel()
optimizer = torch.optim.Adam(model.parameters())

# 分布式优化器
optimizer = hvd.DistributedOptimizer(optimizer)

# 广播参数
hvd.broadcast_parameters(model.state_dict(), root_rank=0)

import deepspeed

# DeepSpeed配置
ds_config = {
    "train_batch_size": 32,
    "gradient_clipping": 1.0,
    "fp16": {
        "enabled": True
    },
    "zero_optimization": {
        "stage": 2
    }
}

# 初始化DeepSpeed
model, optimizer, _, _ = deepspeed.initialize(
    model=model,
    optimizer=optimizer,
    config=ds_config
)

import optuna

def objective(trial):
    # 定义搜索空间
    lr = trial.suggest_float('lr', 1e-5, 1e-1, log=True)
    batch_size = trial.suggest_categorical('batch_size', [16, 32, 64])
    hidden_size = trial.suggest_int('hidden_size', 64, 512)
    
    # 训练模型
    model = SimpleModel(hidden_size)
    optimizer = torch.optim.Adam(model.parameters(), lr=lr)
    
    # 评估模型
    accuracy = train_and_evaluate(model, optimizer, batch_size)
    return accuracy

# 优化
study = optuna.create_study(direction='maximize')
study.optimize(objective, n_trials=100)

# 创建模型档案
torch-model-archiver --model-name my_model \
    --version 1.0 \
    --model-file model.py \
    --serialized-file model.pth \
    --handler handler.py

# 启动服务
torchserve --start --model-store model_store --models my_model=my_model.mar

# TensorFlow Serving配置
model_config_file = """
model_config_list: {
  config: {
    name: "my_model",
    base_path: "/path/to/models",
    model_platform: "tensorflow"
  }
}
"""

# 启动服务
# tensorflow_model_server --model_config_file=/path/to/config

FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["python", "app.py"]

apiVersion: apps/v1
kind: Deployment
metadata:
  name: ai-model-deployment
spec:
  replicas: 3
  selector:
    matchLabels:
      app: ai-model
  template:
    metadata:
      labels:
        app: ai-model
    spec:
      containers:
      - name: model-container
        image: ai-model:latest
        ports:
        - containerPort: 8000
        resources:
          requests:
            memory: "1Gi"
            cpu: "500m"
          limits:
            memory: "2Gi"
            cpu: "1000m"

# 初始化DVC
dvc init

# 添加数据文件
dvc add data/training_dataset.csv

# 推送数据到远程存储
dvc push

# 拉取数据
dvc pull

from kedro.pipeline import Pipeline, node
from kedro.runner import SequentialRunner

def load_data(filepath):
    return pd.read_csv(filepath)

def process_data(data):
    # 数据处理逻辑
    return processed_data

def train_model(data):
    # 模型训练逻辑
    return model

# 构建流水线
pipeline = Pipeline([
    node(load_data, "raw_data", "processed_data", name="load_raw_data"),
    node(process_data, "processed_data", "clean_data", name="process_data"),
    node(train_model, "clean_data", "model", name="train_model")
])

# 运行流水线
runner = SequentialRunner()
runner.run(pipeline)

name: ML Pipeline
on: [push, pull_request]

jobs:
  train:
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v2
    - name: Setup Python
      uses: actions/setup-python@v2
      with:
        python-version: 3.9
    - name: Install dependencies
      run: |
        pip install -r requirements.txt
    - name: Run training
      run: |
        python train.py
    - name: Run tests
      run: |
        pytest tests/
    - name: Upload model
      uses: actions/upload-artifact@v2
      with:
        name: trained-model
        path: models/

import sagemaker
from sagemaker.huggingface import HuggingFace

# SageMaker训练
huggingface_estimator = HuggingFace(
    entry_point='train.py',
    source_dir='src',
    role=sagemaker.get_execution_role(),
    transformers_version='4.12',
    pytorch_version='1.9',
    py_version='py38',
    instance_type='ml.p3.2xlarge',
    instance_count=1,
    hyperparameters={
        'epochs': 3,
        'train_batch_size': 16,
        'model_name': 'bert-base-uncased'
    }
)

# 启动训练
huggingface_estimator.fit({'train': 's3://my-bucket/train/'})

import cv2
import numpy as np

# 图像预处理
def preprocess_image(image_path):
    img = cv2.imread(image_path)
    img = cv2.resize(img, (224, 224))
    img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
    img = img.astype(np.float32) / 255.0
    return img

# 目标检测
def detect_objects(model, image):
    predictions = model.predict(image)
    return predictions

from detectron2.engine import DefaultPredictor
from detectron2.config import get_cfg
from detectron2 import model_zoo

cfg = get_cfg()
cfg.MODEL.WEIGHTS = model_zoo.get_checkpoint_url("COCO-Detection/faster_rcnn_R_50_FPN_3x.yaml")
predictor = DefaultPredictor(cfg)

# 预测
outputs = predictor(image)

import spacy

# 加载模型
nlp = spacy.load("en_core_web_sm")

# 文本处理
def process_text(text):
    doc = nlp(text)
    
    # 命名实体识别
    entities = [(ent.text, ent.label_) for ent in doc.ents]
    
    # 词性标注
    pos_tags = [(token.text, token.pos_) for token in doc]
    
    # 依存句法分析
    dependencies = [(token.text, token.dep_, token.head.text) for token in doc]
    
    return {
        'entities': entities,
        'pos_tags': pos_tags,
        'dependencies': dependencies
    }

from fbprophet import Prophet

# 创建模型
model = Prophet(
    growth='linear',
    seasonality_mode='multiplicative',
    yearly_seasonality=True,
    weekly_seasonality=True
)

# 训练模型
model.fit(df)  # df需要有'ds'和'y'列

# 预测
future = model.make_future_dataframe(periods=365)
forecast = model.predict(future)

import tensorrt as trt
import numpy as np

def optimize_with_tensorrt(engine_path):
    # 创建TensorRT运行时
    runtime = trt.Runtime(trt.Logger(trt.Logger.WARNING))
    
    # 加载优化后的引擎
    with open(engine_path, 'rb') as f:
        engine = runtime.deserialize_cuda_engine(f.read())
    
    return engine

import onnx
import onnxruntime as ort

# 转换模型到ONNX
torch.onnx.export(
    model,
    dummy_input,
    "model.onnx",
    export_params=True,
    opset_version=11
)

# 使用ONNX Runtime推理
session = ort.InferenceSession("model.onnx")
output = session.run(None, {'input': input_data})

import torch.quantization as quant

# 准备量化
model.qconfig = torch.quantization.get_default_qconfig('fbgemm')
model_prepared = torch.quantization.prepare(model, inplace=False)

# 校准
with torch.no_grad():
    for data, target in calibration_loader:
        model_prepared(data)

# 转换为量化模型
model_quantized = torch.quantization.convert(model_prepared, inplace=False)

from prometheus_client import Counter, Histogram, start_http_server

# 定义指标
request_counter = Counter('model_requests_total', 'Total model requests')
prediction_histogram = Histogram('prediction_time_seconds', 'Prediction time')

def predict_with_monitoring(model, input_data):
    request_counter.inc()
    
    with prediction_histogram.time():
        result = model.predict(input_data)
    
    return result

# 启动监控服务器
start_http_server(8000)

from alibi_detect.cd import MMDDrift

# 训练时数据分布
reference_data = X_train

# 创建漂移检测器
cd = MMDDrift(
    x_ref=reference_data,
    p_val=0.05,
    backend='pytorch'
)

# 检测漂移
preds = cd.predict(X_current)
if preds['data']['is_drift']:
    print("检测到数据漂移！")

# 安装：pip install langchain langchain-openai langchain-community
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 1. 定义模型
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)

# 2. 定义提示词模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个资深的{domain}专家，回答要简洁准确。"),
    ("human", "{question}"),
])

# 3. 用 LCEL（LangChain 表达式语言）把组件用管道符串起来
chain = prompt | llm | StrOutputParser()

# 4. 调用
result = chain.invoke({
    "domain": "后端架构",
    "question": "微服务拆分的核心原则是什么？"
})
print(result)

# 流式输出：适合聊天界面逐字显示
for chunk in chain.stream({"domain": "前端", "question": "什么是虚拟DOM？"}):
    print(chunk, end="", flush=True)

# 批量处理：一次并发跑多个输入
results = chain.batch([
    {"domain": "数据库", "question": "什么是索引下推？"},
    {"domain": "网络", "question": "TCP三次握手为什么是三次？"},
])

from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气。参数 city 为城市名。"""
    # 真实项目中这里会调用天气 API
    fake_db = {"北京": "晴 26℃", "上海": "多云 28℃"}
    return fake_db.get(city, "暂无数据")

@tool
def calculate(expression: str) -> str:
    """执行数学计算，参数为合法的 Python 算术表达式。"""
    try:
        return str(eval(expression, {"__builtins__": {}}, {}))
    except Exception as e:
        return f"计算错误: {e}"

tools = [get_weather, calculate]

agent_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个能调用工具的助手。"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, tools, agent_prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

executor.invoke({"input": "北京今天天气怎么样？顺便算一下 128 * 56"})

# 安装：pip install llama-index llama-index-llms-openai
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.core import Settings
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding

# 全局配置模型
Settings.llm = OpenAI(model="gpt-4o-mini")
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")

# 1. 加载文档目录（自动处理 pdf/txt/md 等）
documents = SimpleDirectoryReader("./docs").load_data()

# 2. 构建向量索引（自动切分 + 向量化）
index = VectorStoreIndex.from_documents(documents)

# 3. 创建查询引擎
query_engine = index.as_query_engine(similarity_top_k=3)

# 4. 提问，框架自动检索相关片段并生成答案
response = query_engine.query("这份文档的核心结论是什么？")
print(response)
print("引用来源:", [n.metadata for n in response.source_nodes])

from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.postprocessor import SimilarityPostprocessor

# 自定义切分策略：每块 512 token，重叠 50
splitter = SentenceSplitter(chunk_size=512, chunk_overlap=50)
nodes = splitter.get_nodes_from_documents(documents)

index = VectorStoreIndex(nodes)
query_engine = index.as_query_engine(
    similarity_top_k=5,
    node_postprocessors=[SimilarityPostprocessor(similarity_cutoff=0.7)],
)

# 安装：pip install semantic-kernel
import asyncio
import semantic_kernel as sk
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
from semantic_kernel.functions import kernel_function

class TimePlugin:
    @kernel_function(description="获取当前日期")
    def today(self) -> str:
        from datetime import date
        return str(date.today())

async def main():
    kernel = sk.Kernel()
    kernel.add_service(OpenAIChatCompletion(
        service_id="chat",
        ai_model_id="gpt-4o-mini",
        api_key="YOUR_KEY",
    ))
    kernel.add_plugin(TimePlugin(), plugin_name="time")

    result = await kernel.invoke_prompt(
        "今天是{{time.today}}，请用一句话总结这一天适合做什么。"
    )
    print(result)

asyncio.run(main())

# 安装：pip install haystack-ai
from haystack import Pipeline, Document
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
from haystack.components.generators import OpenAIGenerator
from haystack.components.builders import PromptBuilder

# 文档存储
store = InMemoryDocumentStore()
store.write_documents([
    Document(content="LangChain 是一个 LLM 应用开发框架。"),
    Document(content="Haystack 由 deepset 公司维护。"),
])

# 构建 RAG 管线
prompt_template = """
根据以下上下文回答问题：
{% for doc in documents %}{{ doc.content }}{% endfor %}
问题：{{ question }}
"""

pipe = Pipeline()
pipe.add_component("retriever", InMemoryBM25Retriever(document_store=store))
pipe.add_component("prompt", PromptBuilder(template=prompt_template))
pipe.add_component("llm", OpenAIGenerator(model="gpt-4o-mini"))
pipe.connect("retriever", "prompt.documents")
pipe.connect("prompt", "llm")

result = pipe.run({
    "retriever": {"query": "Haystack 是谁维护的？"},
    "prompt": {"question": "Haystack 是谁维护的？"},
})
print(result["llm"]["replies"][0])

# 安装：pip install pinecone-client
from pinecone import Pinecone, ServerlessSpec

pc = Pinecone(api_key="YOUR_KEY")

# 创建索引（1536 维对应 OpenAI text-embedding-3-small）
pc.create_index(
    name="docs",
    dimension=1536,
    metric="cosine",
    spec=ServerlessSpec(cloud="aws", region="us-east-1"),
)
index = pc.Index("docs")

# 写入向量（upsert）
index.upsert(vectors=[
    {"id": "doc1", "values": [0.1] * 1536, "metadata": {"topic": "AI"}},
    {"id": "doc2", "values": [0.2] * 1536, "metadata": {"topic": "DB"}},
])

# 查询：找出与查询向量最相似的 3 条，并按 metadata 过滤
res = index.query(
    vector=[0.1] * 1536,
    top_k=3,
    filter={"topic": {"$eq": "AI"}},
    include_metadata=True,
)
print(res)

# 安装：pip install weaviate-client
import weaviate
from weaviate.classes.config import Property, DataType, Configure

client = weaviate.connect_to_local()

# 定义集合（schema），启用自动向量化
client.collections.create(
    name="Article",
    vectorizer_config=Configure.Vectorizer.text2vec_openai(),
    properties=[
        Property(name="title", data_type=DataType.TEXT),
        Property(name="body", data_type=DataType.TEXT),
    ],
)

articles = client.collections.get("Article")
articles.data.insert({"title": "向量检索入门", "body": "介绍 HNSW 算法..."})

# 混合检索：结合关键词 + 向量
results = articles.query.hybrid(query="HNSW 是什么", alpha=0.5, limit=3)
for obj in results.objects:
    print(obj.properties)

client.close()

# 安装：pip install pymilvus
from pymilvus import MilvusClient

client = MilvusClient(uri="http://localhost:19530")

client.create_collection(collection_name="demo", dimension=768)

# 插入
data = [{"id": i, "vector": [0.05 * i] * 768, "text": f"文档{i}"} for i in range(100)]
client.insert(collection_name="demo", data=data)

# 检索
res = client.search(
    collection_name="demo",
    data=[[0.1] * 768],
    limit=5,
    output_fields=["text"],
)
print(res)

# 安装：pip install qdrant-client
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct

client = QdrantClient(url="http://localhost:6333")

client.recreate_collection(
    collection_name="docs",
    vectors_config=VectorParams(size=384, distance=Distance.COSINE),
)

client.upsert(collection_name="docs", points=[
    PointStruct(id=1, vector=[0.1] * 384, payload={"lang": "zh"}),
    PointStruct(id=2, vector=[0.2] * 384, payload={"lang": "en"}),
])

hits = client.search(
    collection_name="docs",
    query_vector=[0.1] * 384,
    limit=3,
)
print(hits)

-- 启用扩展
CREATE EXTENSION IF NOT EXISTS vector;

-- 建表：embedding 为 1536 维
CREATE TABLE documents (
    id bigserial PRIMARY KEY,
    content text,
    embedding vector(1536)
);

-- 建 HNSW 索引加速检索
CREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops);

-- 检索：<=> 是余弦距离运算符
SELECT id, content
FROM documents
ORDER BY embedding <=> '[0.1, 0.2, ...]'
LIMIT 5;

# Python 侧配合 psycopg 使用
import psycopg
from pgvector.psycopg import register_vector

conn = psycopg.connect("dbname=mydb")
register_vector(conn)
conn.execute(
    "INSERT INTO documents (content, embedding) VALUES (%s, %s)",
    ("一段文本", [0.1] * 1536),
)

# 安装
pip install vllm

# 启动一个兼容 OpenAI API 的服务
python -m vllm.entrypoints.openai.api_server \
    --model meta-llama/Llama-3.1-8B-Instruct \
    --tensor-parallel-size 1 \
    --max-model-len 8192 \
    --gpu-memory-utilization 0.9

# 客户端直接用 openai SDK 调用
from openai import OpenAI

client = OpenAI(base_url="http://localhost:8000/v1", api_key="EMPTY")
resp = client.chat.completions.create(
    model="meta-llama/Llama-3.1-8B-Instruct",
    messages=[{"role": "user", "content": "解释一下 PagedAttention"}],
)
print(resp.choices[0].message.content)

# 离线批量推理
from vllm import LLM, SamplingParams

llm = LLM(model="meta-llama/Llama-3.1-8B-Instruct")
params = SamplingParams(temperature=0.7, max_tokens=256)
outputs = llm.generate(["你好", "介绍下向量数据库"], params)
for o in outputs:
    print(o.outputs[0].text)

# 用官方 Docker 镜像一键启动
docker run --gpus all -p 8080:80 \
    -v $PWD/data:/data \
    ghcr.io/huggingface/text-generation-inference:latest \
    --model-id meta-llama/Llama-3.1-8B-Instruct \
    --max-input-length 4096 \
    --max-total-tokens 8192

import requests
r = requests.post("http://localhost:8080/generate", json={
    "inputs": "写一句关于秋天的诗",
    "parameters": {"max_new_tokens": 64, "temperature": 0.8},
})
print(r.json())

# 安装后一行命令即可跑起本地模型
ollama run llama3.1

# 拉取并作为服务运行
ollama pull qwen2.5:7b
ollama serve

import ollama
resp = ollama.chat(model="qwen2.5:7b", messages=[
    {"role": "user", "content": "用一句话解释什么是量化"}
])
print(resp["message"]["content"])

# SGLang 启动服务
python -m sglang.launch_server \
    --model-path meta-llama/Llama-3.1-8B-Instruct \
    --port 30000

# SGLang 结构化生成：强制模型按 JSON schema 输出
import sglang as sgl

@sgl.function
def extract(s, text):
    s += "从文本中抽取信息：" + text + "\n"
    s += sgl.gen("name", regex=r"[\u4e00-\u9fa5]+")

# 大致流程：转换权重 -> 构建引擎 -> 用 Triton 部署
trtllm-build --checkpoint_dir ./ckpt \
    --output_dir ./engine \
    --gemm_plugin float16 \
    --max_batch_size 32

# 安装：pip install peft transformers datasets accelerate bitsandbytes
from transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments
from peft import LoraConfig, get_peft_model, TaskType
from datasets import load_dataset
from trl import SFTTrainer

model_name = "meta-llama/Llama-3.1-8B"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(model_name, device_map="auto")

# LoRA 配置：只训练注意力层的投影矩阵
lora_config = LoraConfig(
    task_type=TaskType.CAUSAL_LM,
    r=16,               # 低秩维度
    lora_alpha=32,      # 缩放系数
    lora_dropout=0.05,
    target_modules=["q_proj", "v_proj", "k_proj", "o_proj"],
)
model = get_peft_model(model, lora_config)
model.print_trainable_parameters()  # 会看到可训练参数占比极低

dataset = load_dataset("json", data_files="train.jsonl", split="train")

trainer = SFTTrainer(
    model=model,
    train_dataset=dataset,
    args=TrainingArguments(
        output_dir="./out",
        per_device_train_batch_size=4,
        gradient_accumulation_steps=4,
        num_train_epochs=3,
        learning_rate=2e-4,
        fp16=True,
        logging_steps=10,
    ),
)
trainer.train()
model.save_pretrained("./lora-adapter")  # 只保存几十MB的适配器

from transformers import BitsAndBytesConfig
import torch

# 4bit 量化配置
bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_compute_dtype=torch.bfloat16,
    bnb_4bit_use_double_quant=True,
)

model = AutoModelForCausalLM.from_pretrained(
    model_name,
    quantization_config=bnb_config,
    device_map="auto",
)
# 之后接 LoRA 配置即可，单卡 24G 也能微调

# config.yml
base_model: meta-llama/Llama-3.1-8B
load_in_4bit: true
adapter: qlora
datasets:
  - path: ./data.jsonl
    type: alpaca
sequence_len: 2048
lora_r: 32
lora_alpha: 16
lora_target_modules:
  - q_proj
  - v_proj
gradient_accumulation_steps: 4
micro_batch_size: 2
num_epochs: 3
learning_rate: 0.0002
output_dir: ./output

# 一行命令启动训练
accelerate launch -m axolotl.cli.train config.yml

# 安装：pip install unsloth
from unsloth import FastLanguageModel

model, tokenizer = FastLanguageModel.from_pretrained(
    model_name="unsloth/llama-3.1-8b-bnb-4bit",
    max_seq_length=2048,
    load_in_4bit=True,
)

# 一行加上 LoRA
model = FastLanguageModel.get_peft_model(
    model, r=16,
    target_modules=["q_proj", "k_proj", "v_proj", "o_proj"],
    lora_alpha=16,
)
# 后续用 TRL 的 SFTTrainer 训练，速度约为原生 2 倍

from trl import DPOTrainer, DPOConfig

# 数据格式：每条包含 prompt、chosen（更好的回答）、rejected（较差的回答）
dpo_trainer = DPOTrainer(
    model=model,
    args=DPOConfig(output_dir="./dpo-out", beta=0.1),
    train_dataset=preference_dataset,
    tokenizer=tokenizer,
)
dpo_trainer.train()

# 设置环境变量即自动追踪所有 LangChain 调用
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "YOUR_KEY"

from langsmith import Client
from langsmith.evaluation import evaluate

client = Client()

# 定义一个评估器：判断输出是否包含关键词
def contains_answer(run, example):
    prediction = run.outputs["output"]
    expected = example.outputs["answer"]
    return {"key": "correct", "score": expected in prediction}

# 在数据集上批量评估
evaluate(
    lambda x: my_chain.invoke(x),
    data="my-dataset",
    evaluators=[contains_answer],
)

# 安装：pip install ragas
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision
from datasets import Dataset

data = {
    "question": ["向量数据库是什么？"],
    "answer": ["向量数据库用于存储和检索高维向量。"],
    "contexts": [["向量数据库是存储embedding的专用数据库。"]],
    "ground_truth": ["向量数据库存储并检索向量。"],
}
dataset = Dataset.from_dict(data)

result = evaluate(dataset, metrics=[
    faithfulness,       # 忠实度：答案是否基于上下文，不瞎编
    answer_relevancy,   # 答案相关性
    context_precision,  # 检索到的上下文精确度
])
print(result)

# 安装：pip install deepeval
from deepeval import assert_test
from deepeval.test_case import LLMTestCase
from deepeval.metrics import AnswerRelevancyMetric, HallucinationMetric

def test_answer_quality():
    test_case = LLMTestCase(
        input="法国的首都是哪里？",
        actual_output="法国的首都是巴黎。",
        context=["巴黎是法国的首都和最大城市。"],
    )
    assert_test(test_case, [
        AnswerRelevancyMetric(threshold=0.7),
        HallucinationMetric(threshold=0.3),
    ])

# 用 deepeval 的 CLI 跑，就像跑 pytest
deepeval test run test_app.py

# promptfooconfig.yaml
prompts:
  - "简洁回答：{{question}}"
  - "详细解释并举例：{{question}}"
providers:
  - openai:gpt-4o-mini
  - openai:gpt-4o
tests:
  - vars:
      question: "什么是闭包？"
    assert:
      - type: contains
        value: "函数"
      - type: llm-rubric
        value: "回答准确且易懂"

npx promptfoo eval   # 生成对比报告
npx promptfoo view   # 打开可视化界面

# 安装：pip install argilla
import argilla as rg

client = rg.Argilla(api_url="http://localhost:6900", api_key="admin.apikey")

# 定义标注任务：文本分类
settings = rg.Settings(
    fields=[rg.TextField(name="text")],
    questions=[rg.LabelQuestion(name="sentiment", labels=["正面", "负面", "中性"])],
)
dataset = rg.Dataset(name="reviews", settings=settings, client=client)
dataset.create()

# 推送待标注数据
records = [rg.Record(fields={"text": "这个产品真好用"}) for _ in range(10)]
dataset.records.log(records)

# 安装：pip install distilabel
from distilabel.pipeline import Pipeline
from distilabel.steps import LoadDataFromDicts
from distilabel.steps.tasks import TextGeneration
from distilabel.llms import OpenAILLM

with Pipeline(name="synthetic-qa") as pipeline:
    load = LoadDataFromDicts(data=[{"instruction": "写一个关于AI的问答"}])
    generate = TextGeneration(llm=OpenAILLM(model="gpt-4o-mini"))
    load >> generate

distiset = pipeline.run()
print(distiset["default"]["train"][0])

from openai import OpenAI
import json

client = OpenAI()

def generate_training_samples(topic, n=5):
    """基于主题批量生成问答对"""
    prompt = f"""生成{n}个关于「{topic}」的高质量问答对，
    以 JSON 数组返回，每项含 question 和 answer 字段。"""
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"},
    )
    return json.loads(resp.choices[0].message.content)

samples = generate_training_samples("向量数据库")
with open("synthetic.jsonl", "w", encoding="utf-8") as f:
    for s in samples.get("data", []):
        f.write(json.dumps(s, ensure_ascii=False) + "\n")

# 安装：pip install feast
# feature_repo/example.py
from feast import Entity, FeatureView, Field, FileSource
from feast.types import Float32, Int64
from datetime import timedelta

driver = Entity(name="driver", join_keys=["driver_id"])

source = FileSource(
    path="driver_stats.parquet",
    timestamp_field="event_timestamp",
)

driver_stats_fv = FeatureView(
    name="driver_hourly_stats",
    entities=[driver],
    ttl=timedelta(days=1),
    schema=[
        Field(name="conv_rate", dtype=Float32),
        Field(name="acc_rate", dtype=Float32),
        Field(name="avg_daily_trips", dtype=Int64),
    ],
    source=source,
)

# 获取在线特征用于实时推理
from feast import FeatureStore

store = FeatureStore(repo_path="feature_repo")
features = store.get_online_features(
    features=[
        "driver_hourly_stats:conv_rate",
        "driver_hourly_stats:avg_daily_trips",
    ],
    entity_rows=[{"driver_id": 1001}],
).to_dict()
print(features)

# 常用 CLI
feast apply          # 应用特征定义
feast materialize-incremental $(date +%Y-%m-%d)T00:00:00  # 物化到在线库

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def extract():
    print("抽取数据")

def transform():
    print("转换数据")

def load():
    print("加载数据")

with DAG(
    dag_id="etl_pipeline",
    start_date=datetime(2026, 1, 1),
    schedule="@daily",
    catchup=False,
) as dag:
    t1 = PythonOperator(task_id="extract", python_callable=extract)
    t2 = PythonOperator(task_id="transform", python_callable=transform)
    t3 = PythonOperator(task_id="load", python_callable=load)
    t1 >> t2 >> t3   # 定义依赖顺序

# 安装：pip install prefect
from prefect import flow, task

@task(retries=3, retry_delay_seconds=5)
def fetch_data(url: str):
    return f"data from {url}"

@task
def process(data: str):
    return data.upper()

@flow(name="my-etl")
def etl(url: str):
    raw = fetch_data(url)
    return process(raw)

if __name__ == "__main__":
    print(etl("http://example.com"))

# 安装：pip install dagster
from dagster import asset, materialize

@asset
def raw_users():
    return [{"id": 1, "name": "Alice"}]

@asset
def cleaned_users(raw_users):
    return [{**u, "name": u["name"].strip()} for u in raw_users]

# cleaned_users 依赖 raw_users，Dagster 自动追踪血缘
materialize([raw_users, cleaned_users])

# 安装：pip install kfp
from kfp import dsl

@dsl.component
def train_op(epochs: int) -> str:
    return f"trained {epochs} epochs"

@dsl.pipeline(name="training-pipeline")
def pipeline(epochs: int = 10):
    train_op(epochs=epochs)

# 安装：pip install mlflow
import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier

mlflow.set_experiment("churn-prediction")

with mlflow.start_run():
    n_estimators = 100
    model = RandomForestClassifier(n_estimators=n_estimators)
    model.fit(X_train, y_train)
    acc = model.score(X_test, y_test)

    # 记录参数、指标、模型
    mlflow.log_param("n_estimators", n_estimators)
    mlflow.log_metric("accuracy", acc)
    mlflow.sklearn.log_model(model, "model")

# 模型注册表：把模型提升到 Staging / Production
from mlflow import MlflowClient

client = MlflowClient()
client.create_registered_model("churn-model")
mv = client.create_model_version(
    name="churn-model",
    source="runs:/<run_id>/model",
    run_id="<run_id>",
)
client.transition_model_version_stage(
    name="churn-model", version=mv.version, stage="Production"
)

# 安装：pip install wandb
import wandb

wandb.init(project="llm-finetune", config={"lr": 2e-4, "epochs": 3})

for epoch in range(3):
    loss = train_one_epoch()
    # 记录指标，自动生成可视化曲线
    wandb.log({"epoch": epoch, "loss": loss})

# 记录数据集/模型为 Artifact，实现版本化与血缘
artifact = wandb.Artifact("model", type="model")
artifact.add_file("model.pt")
wandb.log_artifact(artifact)
wandb.finish()

# 超参数扫描（Sweep）
sweep_config = {
    "method": "bayes",
    "metric": {"name": "loss", "goal": "minimize"},
    "parameters": {
        "lr": {"min": 1e-5, "max": 1e-3},
        "batch_size": {"values": [8, 16, 32]},
    },
}
sweep_id = wandb.sweep(sweep_config, project="llm-finetune")
wandb.agent(sweep_id, function=train, count=20)

# DVC 让大数据/模型文件像代码一样版本化
dvc init
dvc add data/train.parquet     # 生成 .dvc 指针文件，纳入 git
dvc remote add -d storage s3://my-bucket/dvcstore
dvc push                        # 推送真实数据到远端

# 定义可复现的管线
dvc stage add -n train \
    -d train.py -d data/train.parquet \
    -o model.pkl \
    python train.py
dvc repro                       # 只在依赖变化时重跑

# 安装：pip install langfuse
from langfuse.decorators import observe
from langfuse.openai import openai  # 包装过的 openai 自动上报

@observe()  # 自动追踪这个函数为一个 trace
def rag_pipeline(question: str):
    context = retrieve(question)       # 会成为子 span
    answer = openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"{context}\n{question}"}],
    )
    return answer.choices[0].message.content

rag_pipeline("向量数据库怎么选？")

# 安装：pip install arize-phoenix openinference-instrumentation-openai
import phoenix as px
from openinference.instrumentation.openai import OpenAIInstrumentor
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter

session = px.launch_app()  # 启动本地可视化界面

tracer_provider = TracerProvider()
tracer_provider.add_span_processor(
    SimpleSpanProcessor(OTLPSpanExporter("http://localhost:6006/v1/traces"))
)
OpenAIInstrumentor().instrument(tracer_provider=tracer_provider)
# 之后所有 OpenAI 调用会自动出现在 Phoenix 界面里

from opentelemetry import trace

tracer = trace.get_tracer("llm-app")

def chat(prompt):
    with tracer.start_as_current_span("llm-call") as span:
        span.set_attribute("llm.model", "gpt-4o-mini")
        span.set_attribute("llm.prompt_tokens", count_tokens(prompt))
        result = call_model(prompt)
        span.set_attribute("llm.completion_tokens", count_tokens(result))
        return result

# 安装：pip install "ray[default]"
import ray

ray.init()

@ray.remote(num_gpus=1)   # 声明需要 1 张 GPU
def train_shard(shard_id):
    return f"shard {shard_id} done"

# 并行调度到集群多张卡
futures = [train_shard.remote(i) for i in range(4)]
print(ray.get(futures))

from ray import serve
from starlette.requests import Request

@serve.deployment(num_replicas=2, ray_actor_options={"num_gpus": 0.5})
class Model:
    def __init__(self):
        self.model = load_model()

    async def __call__(self, request: Request):
        data = await request.json()
        return {"result": self.model.predict(data["input"])}

serve.run(Model.bind())

# kserve-inference.yaml
apiVersion: serving.kserve.io/v1beta1
kind: InferenceService
metadata:
  name: llama-service
spec:
  predictor:
    model:
      modelFormat:
        name: huggingface
      storageUri: "s3://models/llama-3.1-8b"
      resources:
        limits:
          nvidia.com/gpu: "1"

# 安装：pip install langgraph
from langgraph.graph import StateGraph, END
from typing import TypedDict

class State(TypedDict):
    question: str
    answer: str
    retries: int

def think(state: State):
    return {"answer": f"思考：{state['question']}", "retries": state["retries"] + 1}

def should_continue(state: State):
    return "end" if state["retries"] >= 2 else "think"

graph = StateGraph(State)
graph.add_node("think", think)
graph.set_entry_point("think")
graph.add_conditional_edges("think", should_continue, {"think": "think", "end": END})
app = graph.compile()

print(app.invoke({"question": "什么是Agent", "answer": "", "retries": 0}))

# 安装：pip install pyautogen
import autogen

config_list = [{"model": "gpt-4o-mini", "api_key": "YOUR_KEY"}]

assistant = autogen.AssistantAgent(
    name="assistant",
    llm_config={"config_list": config_list},
)
user_proxy = autogen.UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    code_execution_config={"work_dir": "coding", "use_docker": False},
)

# 两个智能体自动对话协作完成任务
user_proxy.initiate_chat(assistant, message="写一个计算斐波那契数列的Python函数并运行")

# 安装：pip install crewai
from crewai import Agent, Task, Crew

researcher = Agent(
    role="研究员",
    goal="收集关于{topic}的资料",
    backstory="你是资深行业分析师。",
)
writer = Agent(
    role="撰稿人",
    goal="根据资料写一篇报告",
    backstory="你是专业的技术作者。",
)

research_task = Task(description="研究{topic}的现状", agent=researcher)
write_task = Task(description="撰写报告", agent=writer)

crew = Crew(agents=[researcher, writer], tasks=[research_task, write_task])
result = crew.kickoff(inputs={"topic": "向量数据库"})
print(result)

# 安装：pip install openai-agents
from agents import Agent, Runner, function_tool

@function_tool
def get_price(product: str) -> str:
    return f"{product} 的价格是 99 元"

agent = Agent(
    name="导购助手",
    instructions="你是一个帮助用户查询商品价格的助手。",
    tools=[get_price],
)

result = Runner.run_sync(agent, "iPhone 多少钱？")
print(result.final_output)

# 工具组合：LlamaIndex + Qdrant + OpenAI + Langfuse
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
from llama_index.vector_stores.qdrant import QdrantVectorStore
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.llms.openai import OpenAI
from llama_index.core import StorageContext
import qdrant_client

# 1. 配置模型
Settings.llm = OpenAI(model="gpt-4o-mini")
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")

# 2. 连接向量库
client = qdrant_client.QdrantClient(url="http://localhost:6333")
vector_store = QdrantVectorStore(client=client, collection_name="kb")
storage_context = StorageContext.from_defaults(vector_store=vector_store)

# 3. 加载并索引文档
docs = SimpleDirectoryReader("./knowledge").load_data()
index = VectorStoreIndex.from_documents(docs, storage_context=storage_context)

# 4. 带重排序的查询引擎
from llama_index.core.postprocessor import SentenceTransformerRerank
reranker = SentenceTransformerRerank(model="BAAI/bge-reranker-base", top_n=3)

query_engine = index.as_query_engine(
    similarity_top_k=10,           # 先粗召回 10 条
    node_postprocessors=[reranker], # 再精排到 3 条
)

# 5. 提问
response = query_engine.query("公司的退货政策是什么？")
print("答案:", response)
print("引用:", [n.node.metadata.get("file_name") for n in response.source_nodes])

# BM25（关键词）擅长精确术语，向量擅长语义，二者融合召回更全
from llama_index.retrievers.bm25 import BM25Retriever
from llama_index.core.retrievers import QueryFusionRetriever

vector_retriever = index.as_retriever(similarity_top_k=5)
bm25_retriever = BM25Retriever.from_defaults(index=index, similarity_top_k=5)

fusion = QueryFusionRetriever(
    [vector_retriever, bm25_retriever],
    similarity_top_k=5,
    mode="reciprocal_rerank",   # RRF 融合排序
)
nodes = fusion.retrieve("SLA 是多少")

# 编译
git clone https://github.com/ggerganov/llama.cpp && cd llama.cpp
make

# 把 HF 模型转成 GGUF 并量化到 4bit
python convert_hf_to_gguf.py ./Llama-3.1-8B --outfile model.gguf
./llama-quantize model.gguf model-q4.gguf Q4_K_M

# 本地交互
./llama-cli -m model-q4.gguf -p "你好，介绍下自己" -n 128

# Python 绑定：pip install llama-cpp-python
from llama_cpp import Llama

llm = Llama(model_path="model-q4.gguf", n_ctx=4096, n_gpu_layers=20)
out = llm("解释什么是量化：", max_tokens=128)
print(out["choices"][0]["text"])

# 安装：pip install mlx-lm
from mlx_lm import load, generate

model, tokenizer = load("mlx-community/Llama-3.1-8B-Instruct-4bit")
text = generate(model, tokenizer, prompt="用一句话解释RAG", max_tokens=64)
print(text)

# 安装：pip install onnxruntime optimum
from optimum.onnxruntime import ORTModelForCausalLM
from transformers import AutoTokenizer

model = ORTModelForCausalLM.from_pretrained("model-onnx")
tokenizer = AutoTokenizer.from_pretrained("model-onnx")
inputs = tokenizer("你好", return_tensors="pt")
out = model.generate(**inputs, max_new_tokens=64)
print(tokenizer.decode(out[0]))

# 安装：pip install guardrails-ai
from guardrails import Guard
from guardrails.hub import RegexMatch

# 保证输出是合法邮箱格式，否则自动重试/纠正
guard = Guard().use(RegexMatch, regex=r"[^@]+@[^@]+\.[^@]+")
result = guard(
    llm_api=my_llm_call,
    prompt="生成一个示例邮箱地址",
)
print(result.validated_output)

# 安装：pip install nemoguardrails
from nemoguardrails import RailsConfig, LLMRails

# config 里用 Colang 定义：拒答政治/医疗建议等
config = RailsConfig.from_path("./config")
rails = LLMRails(config)

response = rails.generate(messages=[
    {"role": "user", "content": "教我怎么黑进别人电脑"}
])
print(response)  # 会被护栏拦截并给出安全回复

# config/rails.co （Colang 示例）
define user ask illegal
  "怎么黑进别人电脑"
  "如何制作危险物品"

define bot refuse illegal
  "抱歉，我无法协助这类请求。"

define flow
  user ask illegal
  bot refuse illegal

# 安装：pip install presidio-analyzer presidio-anonymizer
from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine

analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()

text = "我的邮箱是 test@example.com，电话 13800138000"
results = analyzer.analyze(text=text, language="en")
anonymized = anonymizer.anonymize(text=text, analyzer_results=results)
print(anonymized.text)  # 敏感信息被替换为占位符

# 安装：pip install litellm
from litellm import completion

# 用同一套代码调用不同厂商，只改 model 名
for model in ["gpt-4o-mini", "claude-3-5-sonnet-20241022", "ollama/llama3.1"]:
    resp = completion(
        model=model,
        messages=[{"role": "user", "content": "用一句话解释RAG"}],
    )
    print(model, "->", resp.choices[0].message.content)

# 作为代理网关运行：config.yaml
model_list:
  - model_name: gpt-4o
    litellm_params:
      model: openai/gpt-4o
      api_key: os.environ/OPENAI_KEY
  - model_name: claude
    litellm_params:
      model: anthropic/claude-3-5-sonnet-20241022
      api_key: os.environ/ANTHROPIC_KEY
router_settings:
  routing_strategy: least-busy   # 负载均衡策略

# 启动统一网关，业务侧只对接一个 endpoint
litellm --config config.yaml --port 4000

# 安装：pip install gptcache
from gptcache import cache
from gptcache.adapter import openai
from gptcache.embedding import Onnx
from gptcache.manager import CacheBase, VectorBase, get_data_manager
from gptcache.similarity_evaluation.distance import SearchDistanceEvaluation

embedding = Onnx()
data_manager = get_data_manager(
    CacheBase("sqlite"),
    VectorBase("faiss", dimension=embedding.dimension),
)
cache.init(
    embedding_func=embedding.to_embeddings,
    data_manager=data_manager,
    similarity_evaluation=SearchDistanceEvaluation(),
)

# 语义相近的问题会命中缓存，直接返回，不再调用模型
resp = openai.ChatCompletion.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "什么是向量数据库？"}],
)

# 自建开源 embedding：pip install sentence-transformers
from sentence_transformers import SentenceTransformer

model = SentenceTransformer("BAAI/bge-m3")
vectors = model.encode(["第一段文本", "第二段文本"], normalize_embeddings=True)
print(vectors.shape)   # (2, 1024)

# 最小可用 RAG：几十行跑通
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
docs = SimpleDirectoryReader("./docs").load_data()
index = VectorStoreIndex.from_documents(docs)
print(index.as_query_engine().query("核心问题是什么？"))