2026 年 AI 开发工具链全景图：从模型训练到部署上线的完整工具推荐

AI 开发听起来高大上，其实就是在搭积木。今天给你一张完整的 AI 开发工具链地图，从数据处理到模型训练、从评测到部署，每个环节都有靠谱工具。

先上全景图

AI 开发全流程
├── 数据处理
│   ├── 数据采集：Scrapy、Playwright、Data-centric AI
│   ├── 数据清洗：Great Expectations、Pandas
│   └── 数据标注：Label Studio、CVAT、Doccano
│
├── 模型训练
│   ├── 框架：PyTorch、TensorFlow、JAX
│   ├── 分布式训练：DeepSpeed、ColossalAI、FSDP
│   ├── 超参调优：Ray Tune、Optuna、Weights & Biases
│   └── 实验管理：MLflow、Neptune、W&B
│
├── 模型优化
│   ├── 量化：GPTQ、AWQ、llama.cpp、TensorRT-LLM
│   ├── 剪枝：TorchPruning、SparseGPT
│   └── 蒸馏：Knowledge Distillation、 TinyBERT
│
├── 推理部署
│   ├── 推理框架：vLLM、Text Generation Inference、Ollama
│   ├── 模型服务：Ray Serve、TensorFlow Serving、Triton
│   └── 云部署：SageMaker、Vertex AI、腾讯云 TI-ONE
│
└── AI 应用开发
    ├── 应用框架：LangChain、LlamaIndex、AutoGen、Dify
    ├── 向量数据库：Milvus、Pinecone、Weaviate、Chroma
    └── 监控系统：LangSmith、AgentOps、MLflow

数据处理：好模型从好数据开始

采集 + 清洗：Pandas + Great Expectations

# data_pipeline.py
import pandas as pd
from great_expectations.dataset import PandasDataset
from great_expectations.core import ExpectationSuite

# 加载数据
df = pd.read_csv("training_data.csv")

# 定义数据质量预期
ge_df = PandasDataset(df)
suite = ExpectationSuite("training_data_quality")

# 基础检查
ge_df.expect_column_to_exist("prompt")
ge_df.expect_column_to_exist("completion")
ge_df.expect_column_values_to_not_be_null("prompt")
ge_df.expect_column_values_to_be_of_type("prompt", "str")
ge_df.expect_column_value_lengths_to_be_between("completion", min_value=10, max_value=4096)

# 质量检查报告
results = ge_df.validate(expectation_suite=suite)
if not results["success"]:
    print("数据质量问题：", results["results"])
    raise ValueError("数据质量不达标，停止训练")

# 清洗
df = df.dropna()
df = df[df["completion"].str.len() >= 10]
df.to_parquet("cleaned_training_data.parquet")

数据标注：Label Studio

# Docker 一键启动 Label Studio
docker run -d \
  -p 8080:8080 \
  -v $(pwd)/label_studio_data:/label-studio/data \
  --name label-studio \
  heartexlabs/label-studio:latest

# API 导出标注数据
curl -X GET http://localhost:8080/api/projects/1/export \
  -H "Authorization: Token your-token-here" \
  > annotations.json

模型训练：分布式 + 实验管理

分布式训练：DeepSpeed + FSDP

# train_distributed.py - DeepSpeed ZeRO-3 分布式训练
import deepspeed
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer

# DeepSpeed 配置
ds_config = {
    "train_batch_size": "auto",
    "gradient_accumulation_steps": "auto",
    "fp16": {"enabled": False},
    "bf16": {"enabled": True},                    # 使用 BF16 精度
    "zero_optimization": {
        "stage": 3,                                 # ZeRO-3：极致显存优化
        "offload_param": {"device": "cpu"},        # 卸载参数到 CPU
        "offload_optimizer": {"device": "cpu"},
    },
    "gradient_checkpointing": {"enable": True},    # 激活检查点
}

# 模型
model = AutoModelForCausalLM.from_pretrained(
    "Qwen/Qwen2.5-7B-Instruct",
    torch_dtype=torch.bfloat16,
)

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

# 训练循环
for batch in dataloader:
    batch = {k: v.to(model_engine.device) for k, v in batch.items()}
    outputs = model_engine(**batch)
    model_engine.backward(outputs.loss)
    model_engine.step()

# 保存
model_engine.save_checkpoint("output/checkpoint")

实验管理：Weights & Biases

# train_with_wandb.py
import wandb
from transformers import Trainer, TrainingArguments

# 初始化
wandb.init(
    project="llm-finetuning",
    name="qwen2.5-7b-sft-exp-001",
    config={
        "learning_rate": 2e-5,
        "batch_size": 8,
        "epochs": 3,
        "model": "Qwen2.5-7B-Instruct",
    }
)

training_args = TrainingArguments(
    output_dir="output",
    eval_strategy="steps",
    logging_steps=10,
    report_to="wandb",              # 对接 W&B
    save_steps=500,
    per_device_train_batch_size=8,
    gradient_accumulation_steps=4,
)

trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=train_dataset,
    eval_dataset=eval_dataset,
)

trainer.train()
wandb.finish()

模型优化：让模型又小又快

量化：AWQ + vLLM 部署

# quantize_awq.py - AWQ 量化（精度比 GPTQ 更高）
from awq import AutoAWQForCausalLM
from transformers import AutoTokenizer

model_path = "Qwen/Qwen2.5-7B-Instruct"
quant_path = "Qwen2.5-7B-Instruct-AWQ"

# 加载模型
model = AutoAWQForCausalLM.from_pretrained(
    model_path,
    torch_dtype=torch.float16,
)

tokenizer = AutoTokenizer.from_pretrained(model_path)

# 量化配置
quant_config = {
    "zero_point": True,
    "q_group_size": 128,      # 量化组大小
    "w_bit": 4,               # 权重量化位数
    "version": "GEMM",        # GEMM / GEMV
}

# 量化（需要校准数据）
model.quantize(tokenizer, quant_config=quant_config)

# 保存
model.save_quantized(quant_path)
tokenizer.save_pretrained(quant_path)

# 用 vLLM 部署量化模型
python -m vllm.entrypoints.openai.api_server \
    --model Qwen2.5-7B-Instruct-AWQ \
    --quantization awq \
    --tensor-parallel-size 1 \
    --gpu-memory-utilization 0.9

模型大小对比

模型版本	精度	参数量	FP16 显存	INT4 量化显存	加速比
Qwen2.5-7B	BF16	7.7B	15.4 GB	4.5 GB	1x
Qwen2.5-7B-AWQ	INT4	7.7B	–	4.5 GB	2.3x
Qwen2.5-14B	BF16	14B	28 GB	8.2 GB	1x
Qwen2.5-14B-AWQ	INT4	14B	–	8.2 GB	2.1x
Llama3-8B	BF16	8B	16 GB	4.9 GB	1x
Llama3-8B-GPTQ	INT4	8B	–	4.9 GB	1.9x

推理部署：让模型跑起来

vLLM：高吞吐推理服务

# serve_vllm.py - 用 vLLM 启动推理服务
from vllm import LLM

# 一行启动生产级推理
llm = LLM(
    model="Qwen/Qwen2.5-14B-Instruct",
    tensor_parallel_size=2,          # 2 卡并行
    gpu_memory_utilization=0.92,     # 显存占用
    max_num_seqs=256,                # 最大并发
    enable_chunked_prefill=True,     # 分块预填充
    use_flash_attention=True,         # Flash Attention
    quantization="awq",              # 量化部署
)

# OpenAI 兼容 API
from vllm import SamplingParams
sampling = SamplingParams(temperature=0.7, top_p=0.9, max_tokens=512)

outputs = llm.generate(["解释量子计算", "Python异步编程原理"], sampling)
for o in outputs:
    print(o.outputs[0].text)

Ollama：本地一键跑大模型

# 安装
curl -fsSL https://ollama.com/install.sh | sh

# 拉取并运行模型
ollama pull qwen2.5:14b
ollama pull llama3:8b
ollama pull deepseek-r1:7b

# 运行
ollama run qwen2.5:14b "用Python实现快速排序"

# API 调用
curl http://localhost:11434/api/generate -d '{
  "model": "qwen2.5:14b",
  "prompt": "解释什么是RESTful API",
  "stream": false
}'

AI 应用开发：搭应用必备

向量数据库选型

数据库	优势	适用场景	部署方式
Milvus	功能全面、性能强	亿级向量、生产级	K8s/Docker
Pinecone	托管服务、零运维	快速上线	云托管
Weaviate	原生多模态支持	图文混合检索	Docker/K8s
Chroma	轻量、Python 原生	快速原型	本地/Pip
Qdrant	Rust 实现、高性能	实时推荐	Docker/K8s

RAG 应用开发：LangChain + Milvus

# rag_app.py - 完整 RAG 流水线
from langchain_community.vectorstores import Milvus
from langchain_community.embeddings import HuggingFaceBgeEmbeddings
from langchain_community.llms import Ollama
from langchain.prompts import PromptTemplate
from langchain.chains import RetrievalQA

# 1. Embedding 模型
embeddings = HuggingFaceBgeEmbeddings(
    model_name="BAAI/bge-large-zh-v1.5",
    model_kwargs={"device": "cuda"},
    encode_kwargs={"normalize_embeddings": True},
)

# 2. 连接 Milvus 向量数据库
vector_db = Milvus(
    embedding_function=embeddings,
    connection_args={"host": "localhost", "port": "19530"},
    collection_name="knowledge_base",
)

# 3. LLM
llm = Ollama(model="qwen2.5:14b", temperature=0.1)

# 4. 构建 RAG Chain
prompt = PromptTemplate(
    template="""使用以下上下文来回答问题。
    上下文：{context}
    问题：{question}
    回答：""",
    input_variables=["context", "question"],
)

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vector_db.as_retriever(search_kwargs={"k": 3}),
    return_source_documents=True,
)

# 5. 查询
result = qa_chain.invoke({"query": "公司年假政策是什么？"})
print(result["result"])

---

👤 作者简介

一枚在大中原腹地（河南）卖公有云的女/男士，主营腾讯云/阿里云/华为云，曾踩坑无数，现专注AI大模型应用落地。关注公众号「AI热点日报」，围观AI前沿动态~

博客：y1cloud.cn