当我们谈论 AI 应用开发时，很多人习惯性地把注意力集中在"模型"本身——选一个最强的模型，调一调 Prompt，觉得这就够了。
但真正做过生产级 AI 系统的工程师都知道：模型只是冰山一角。
笔者在过去一年经历了从"调模型"到"建系统"的认知升级，深刻体会到：AI 工程化的核心挑战不在于模型能力，而在于如何把模型能力可靠、可维护、可扩展地交付给用户。
这篇文章，我将分享我认为最合理的 AI 工程化三层架构设计：
模型层（Model Layer）
平台层（Platform Layer）
应用层（Application Layer）
这个架构不是凭空想象，而是来自笔者在多个项目中的实战总结。
一、为什么需要三层架构？
1.1 传统软件架构的启示
在传统后端开发中，我们早就习惯了分层架构：
┌─────────────────┐
│ 网关层 (API) │ ← 路由、鉴权、限流
├─────────────────┤
│ 服务层 (Logic) │ ← 业务逻辑、数据处理
├─────────────────┤
│ 数据层 (Data) │ ← 数据库、缓存、存储
└─────────────────┘每一层都有明确的职责边界：网关负责入口、服务负责逻辑、数据负责持久化。这种分离带来了几个关键价值：
可替换性：换数据库不用改业务逻辑
可扩展性：每层可以独立扩容
可维护性：问题定位更清晰
团队分工：不同团队负责不同层
1.2 AI 系统的特殊性
但是，AI 系统有几个传统系统没有的特性：
这些特性决定了 AI 系统不能简单套用传统分层，需要针对 AI 特点重新设计架构层次。
1.3 三层架构的核心理念
经过多个项目的实践，我总结出 AI 工程化的三层架构：
核心理念：
模型层：解决"模型怎么来"的问题
平台层：解决"模型怎么跑"的问题
应用层：解决"模型怎么用"的问题
三层之间松耦合、高内聚，每层都可以独立演进。
二、模型层
2.1 模型选型：从"选最强"到"选最合适"
模型选型是 AI 工程化的第一步，但很多团队犯了一个错误：盲目追求最强模型。
笔者的教训：曾经为了追求效果，用 GPT-4 处理所有任务，结果：
成本是 GPT-3.5 的 20 倍
延迟是 GPT-3.5 的 3 倍
很多简单任务根本不需要 GPT-4 的能力
正确的选型策略：根据任务复杂度分级
# config/model_routing.yaml
model_routing:
# 简单分类任务 → 小模型
task_type: classification
models:
- name: qwen2.5-3b
score_threshold: 0.85
cost_per_1k_tokens: 0.001
- name: qwen2.5-7b
fallback: true
score_threshold: 0.75
cost_per_1k_tokens: 0.003
# 复杂推理任务 → 大模型
task_type: reasoning
models:
- name: qwen3.5-32b
score_threshold: 0.80
cost_per_1k_tokens: 0.015
- name: deepseek-v3
fallback: true
score_threshold: 0.75
cost_per_1k_tokens: 0.0122.2 模型微调：小样本学习的艺术
当预训练模型不能满足业务需求时，需要微调（Fine-tuning）。但微调不是银弹：
适合微调的场景：
特定领域术语和表达方式
特定输出格式要求
低延迟要求的本地部署场景
不适合微调的场景：
数据量太少（< 1000 条）
任务变化频繁
需要快速迭代
# scripts/finetune_llama.py
"""
LoRA 微调脚本 - 使用 QLoRA 降低微调成本
"""
import torch
from transformers import (
AutoModelForCausalLM,
AutoTokenizer,
TrainingArguments,
DataCollatorForLanguageModeling,
)
from peft import LoraConfig, get_peft_model, TaskType
from datasets import load_dataset
def setup_lora_model(model_name: str):
"""加载模型并配置 LoRA"""
# 加载分词器
tokenizer = AutoTokenizer.from_pretrained(model_name)
tokenizer.pad_token = tokenizer.eos_token
# 加载模型（4-bit 量化）
model = AutoModelForCausalLM.from_pretrained(
model_name, torch_dtype=torch.float16, load_in_4bit=True, device_map="auto"
)
# 配置 LoRA
lora_config = LoraConfig(
task_type=TaskType.CAUSAL_LM,
r=64, # LoRA 秩，越大效果越好但参数越多
lora_alpha=128, # LoRA 缩放因子
target_modules=[ # 要应用 LoRA 的模块
"q_proj",
"k_proj",
"v_proj",
"o_proj",
"gate_proj",
"up_proj",
"down_proj",
],
lora_dropout=0.05,
bias="none",
inference_mode=False,
)
# 应用 LoRA
model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
return model, tokenizer
def prepare_training_data(data_path: str, tokenizer, max_length: int = 2048):
"""准备训练数据"""
# 加载数据集
dataset = load_dataset("json", data_files=data_path)
def tokenize_function(examples):
# 格式化输入输出对
texts = [
f"### 指令: {instr}\n### 回答: {resp}"
for instr, resp in zip(examples["instruction"], examples["response"])
]
# 分词
result = tokenizer(
texts, truncation=True, max_length=max_length, padding="max_length"
)
# 标签与输入相同（用于语言模型训练）
result["labels"] = result["input_ids"].copy()
return result
return dataset.map(
tokenize_function, batched=True, remove_columns=dataset["train"].column_names
)
def train(model, tokenizer, train_dataset, output_dir: str = "./output_model"):
"""执行训练"""
training_args = TrainingArguments(
output_dir=output_dir,
num_train_epochs=3,
per_device_train_batch_size=4,
gradient_accumulation_steps=4, # 累加梯度
learning_rate=2e-4,
warmup_ratio=0.03,
lr_scheduler_type="cosine",
logging_steps=10,
save_strategy="epoch",
save_total_limit=3,
fp16=True, # 混合精度训练
optim="paged_adamw_32bit", # 节省显存
report_to="tensorboard",
push_to_hub=False,
)
trainer = Trainer(
model=model,
args=training_args,
train_dataset=train_dataset,
# 因果语言模型不用 MLM
data_collator=DataCollatorForLanguageModeling(tokenizer=tokenizer, mlm=False),
)
trainer.train()
trainer.save_model()
return model2.3 模型评测：没有评测就没有优化
核心观点：模型评测不是"上线前的检查"，而是"持续迭代的基础"。
# eval/model_evaluator.py
"""
模型评测框架 - 支持多维度评估
"""
from dataclasses import dataclass
from typing import List, Dict, Callable
import json
from collections import defaultdict
@dataclass
class EvalResult:
"""评测结果"""
model_name: str
task_name: str
metrics: Dict[str, float]
samples: List[Dict] # 具体样例
def summary(self) -> str:
lines = [f"📊 {self.task_name} - {self.model_name}"]
for metric, value in self.metrics.items():
lines.append(f" {metric}: {value:.4f}")
return "\n".join(lines)
class ModelEvaluator:
"""模型评测器"""
def __init__(self):
self.results = {}
def add_metric(self, name: str, func: Callable[[str, str], float]):
"""添加评测指标"""
self.metrics[name] = func
def evaluate(
self, model_name: str, task_name: str, test_cases: List[Dict]
) -> EvalResult:
"""
执行评测
Args:
model_name: 模型名称
task_name: 任务名称
test_cases: 测试用例 [{input, reference, metadata}]
"""
metrics = defaultdict(list)
samples = []
for case in test_cases:
# 获取模型输出
model_output = self.call_model(model_name, case["input"])
reference = case["reference"]
# 计算每个指标
for metric_name, metric_func in self.metrics.items():
score = metric_func(model_output, reference)
metrics[metric_name].append(score)
# 保存样例
samples.append(
{
"input": case["input"],
"reference": reference,
"model_output": model_output,
"scores": {
name: func(model_output, reference)
for name, func in self.metrics.items()
},
}
)
# 聚合指标
aggregated = {
name: sum(values) / len(values) for name, values in metrics.items()
}
return EvalResult(
model_name=model_name,
task_name=task_name,
metrics=aggregated,
samples=samples,
)
def call_model(self, model_name: str, prompt: str) -> str:
"""调用模型（需要对接实际的推理服务）"""
# 实际实现中对接推理服务
raise NotImplementedError
# 预置的评测指标
def exact_match(pred: str, ref: str) -> float:
"""精确匹配"""
return 1.0 if pred.strip() == ref.strip() else 0.0
def rouge_l(pred: str, ref: str) -> float:
"""ROUGE-L 分数（简化版）"""
# 实际实现使用 rouge-score 库
pred_tokens = pred.split()
ref_tokens = ref.split()
lcs = longest_common_subsequence(pred_tokens, ref_tokens)
if not lcs:
return 0.0
precision = lcs / len(pred_tokens)
recall = lcs / len(ref_tokens)
if precision + recall == 0:
return 0.0
return 2 * precision * recall / (precision + recall)
def longest_common_subsequence(a: List, b: List) -> int:
"""计算 LCS 长度"""
m, n = len(a), len(b)
dp = [[0] * (n + 1) for _ in range(m + 1)]
for i in range(1, m + 1):
for j in range(1, n + 1):
if a[i - 1] == b[j - 1]:
dp[i][j] = dp[i - 1][j - 1] + 1
else:
dp[i][j] = max(dp[i - 1][j], dp[i][j - 1])
return dp[m][n]2.4 模型版本管理：可重复的 AI
模型版本管理是被严重低估的能力。很多团队遇到这样的问题：
模型更新后效果变差，但回滚困难
无法精确复现某个历史结果
A/B 实验无法保证模型一致性
# config/model_registry.yaml
model_registry:
version: "1.0.0"
models:
# 生产模型
production:
intent_classification:
model_id: "qwen2.5-7b:v3.2.1"
deployed_at: "2026-09-15"
status: "active"
metrics:
accuracy: 0.923
latency_p99_ms: 45
text_generation:
model_id: "deepseek-v3:v2.1.0"
deployed_at: "2026-08-20"
status: "active"
metrics:
rouge_l: 0.452
latency_p99_ms: 180
# 候选模型
candidate:
- model_id: "qwen3.5-32b:v1.0.0"
stage: "staging"
metrics:
accuracy: 0.935
compare_to: "qwen2.5-7b:v3.2.1"# model_registry.py
"""
模型注册表 - 管理模型版本和部署
"""
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from datetime import datetime
import json
from pathlib import Path
@dataclass
class ModelVersion:
"""模型版本"""
model_id: str # 格式: name:version
model_path: str # 模型文件路径
config: Dict # 模型配置
metrics: Dict # 评测指标
created_at: datetime
created_by: str
description: str = ""
class ModelRegistry:
"""模型注册表"""
def __init__(self, registry_path: str):
self.registry_path = Path(registry_path)
self.versions: Dict[str, List[ModelVersion]] = {}
self._load()
def _load(self):
"""加载注册表"""
if self.registry_path.exists():
with open(self.registry_path) as f:
data = json.load(f)
for model_name, versions in data["models"].items():
self.versions[model_name] = [ModelVersion(**v) for v in versions]
def _save(self):
"""保存注册表"""
data = {
"models": {
name: [
{
"model_id": v.model_id,
"model_path": v.model_path,
"config": v.config,
"metrics": v.metrics,
"created_at": v.created_at.isoformat(),
"created_by": v.created_by,
"description": v.description,
}
for v in versions
]
for name, versions in self.versions.items()
}
}
with open(self.registry_path, "w") as f:
json.dump(data, f, indent=2, ensure_ascii=False)
def register(
self,
name: str,
version: str,
model_path: str,
config: Dict,
metrics: Dict,
created_by: str,
description: str = "",
) -> ModelVersion:
"""注册新模型版本"""
model_id = f"{name}:{version}"
model_version = ModelVersion(
model_id=model_id,
model_path=model_path,
config=config,
metrics=metrics,
created_at=datetime.now(),
created_by=created_by,
description=description,
)
if name not in self.versions:
self.versions[name] = []
self.versions[name].append(model_version)
self._save()
return model_version
def get_latest(self, name: str) -> Optional[ModelVersion]:
"""获取最新版本"""
if name not in self.versions:
return None
return self.versions[name][-1]
def get_by_version(self, model_id: str) -> Optional[ModelVersion]:
"""根据版本 ID 获取模型"""
name, version = model_id.rsplit(":", 1)
if name not in self.versions:
return None
for v in self.versions[name]:
if v.model_id == model_id:
return v
return None三、平台层
3.1 训练服务：从"能训练"到"高效训练"
训练服务解决的核心问题：如何高效、可靠地训练模型。
┌─────────────────────────────────────────────────────────────────┐
│ 训练服务架构 │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ 任务 │───▶│ 调度器 │───▶│ 执行器 │ │
│ │ 提交 │ │ (Priority)│ │ (GPU Pool)│ │
│ └──────────┘ └──────────┘ └──────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ 任务 │ │ 资源 │ │ Checkpoint│ │
│ │ 队列 │ │ 管理 │ │ 存储 │ │
│ └──────────┘ └──────────┘ └──────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘# config/training_job.yaml
apiVersion: mlops/v1
kind: TrainingJob
metadata:
name: intent-classification-v4
namespace: ml-team
spec:
# 模型配置
model:
base_model: "qwen2.5-7b"
training_type: "lora" # full / lora / qlora
# 数据配置
data:
train_path: "s3://data/train.jsonl"
val_path: "s3://data/val.jsonl"
max_seq_length: 2048
# 超参数
hyperparameters:
learning_rate: 2.0e-4
batch_size: 8
epochs: 3
warmup_ratio: 0.03
lora_r: 64
lora_alpha: 128
# 资源请求
resources:
gpu: 4 # 需要的 GPU 数量
gpu_type: "A100-80G"
memory: "64Gi"
# 调度配置
scheduling:
priority: 80 # 优先级 1-100
max_running_time: "24h" # 最大运行时间
preemptible: false # 是否可抢占
# 输出配置
output:
model_path: "s3://models/intent-v4"
checkpoint_interval: 100 # 每 N 步保存一次
# 回调配置
callbacks:
- type: "tensorboard"
log_dir: "s3://logs/intent-v4"
- type: "early_stopping"
patience: 2
monitor: "val_loss"3.2 推理服务：生产级部署的关键
推理服务是 AI 工程化最复杂也最关键的组件。核心挑战：
高并发：同时处理成百上千请求
低延迟：TTFT < 1s，TPOT < 50ms
高吞吐：最大化 GPU 利用率
成本控制：GPU 资源昂贵
# Dockerfile.vllm
FROM nvidia/cuda:12.4.0-devel-ubuntu22.04
# 安装 Python 和依赖
RUN apt-get update && apt-get install -y \
python3.11 python3-pip git curl \
&& rm -rf /var/lib/apt/lists/*
# 安装 vLLM（2026年推荐 0.20.0+）
RUN pip install vllm==0.20.0 \
transformers==4.48.0 \
accelerate==0.36.0
# 复制启动脚本
COPY entrypoint.sh /entrypoint.sh
RUN chmod +x /entrypoint.sh
# 暴露端口
EXPOSE 8000
# 健康检查
HEALTHCHECK --interval=30s --timeout=10s --start-period=60s \
CMD curl -f http://localhost:8000/health || exit 1
ENTRYPOINT ["/entrypoint.sh"]# config/vllm_deployment.yaml
# Kubernetes 部署配置
apiVersion: apps/v1
kind: Deployment
metadata:
name: vllm-inference
labels:
app: vllm
model: qwen2.5-7b
spec:
replicas: 3
selector:
matchLabels:
app: vllm
template:
metadata:
labels:
app: vllm
model: qwen2.5-7b
spec:
containers:
- name: vllm
image: registry.company.com/vllm:0.20.0
# 端口
ports:
- containerPort: 8000
name: http
# 资源限制
resources:
limits:
nvidia.com/gpu: 1
memory: "60Gi"
requests:
nvidia.com/gpu: 1
memory: "40Gi"
# 环境变量
env:
- name: MODEL_PATH
value: "/models/qwen2.5-7b"
- name: TENSOR_PARALLEL_SIZE
value: "1"
- name: QUANTIZATION
value: "fp16"
# 挂载模型
volumeMounts:
- name: model-storage
mountPath: /models
# 启动参数
args:
- "--model=$(MODEL_PATH)"
- "--tensor-parallel-size=$(TENSOR_PARALLEL_SIZE)"
- "--quantization=$(QUANTIZATION)"
- "--max-model-len=32768"
- "--gpu-memory-utilization=0.9"
- "--trust-remote-code"
- "--enforce-eager" # 禁用 CUDA graph（兼容性）
- "--enable-chunked-prefill"
- "--max-num-batched-tokens=8192"
# 健康检查
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 60
periodSeconds: 30
readinessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 30
periodSeconds: 10
# 亲和性配置（让同一个模型的实例分散在不同节点）
affinity:
podAntiAffinity:
preferredDuringSchedulingIgnoredDuringExecution:
- weight: 100
podAffinityTerm:
labelSelector:
matchLabels:
model: qwen2.5-7b
topologyKey: kubernetes.io/hostname
volumes:
- name: model-storage
persistentVolumeClaim:
claimName: model-storage-pvc
---
# Service 配置
apiVersion: v1
kind: Service
metadata:
name: vllm-inference-service
spec:
type: ClusterIP
ports:
- port: 8000
targetPort: 8000
name: http
selector:
app: vllm
---
# HPA 配置
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: vllm-inference-hpa
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: vllm-inference
minReplicas: 2
maxReplicas: 10
metrics:
- type: GPU
gpu:
averageUtilization: 70
averageValue: null
- type: Resource
resource:
name: memory
target:
type: Utilization
averageUtilization: 80
behavior:
scaleUp:
stabilizationWindowSeconds: 60
policies:
- type: Percent
value: 100
periodSeconds: 60
scaleDown:
stabilizationWindowSeconds: 300# scripts/inference_client.py
"""
推理服务客户端 - 支持流式调用和重试
"""
import asyncio
import aiohttp
from typing import AsyncIterator, Optional
import json
from dataclasses import dataclass
import time
@dataclass
class InferenceConfig:
"""推理配置"""
base_url: str
model_name: str
max_tokens: int = 2048
temperature: float = 0.7
top_p: float = 0.9
timeout: int = 120
class InferenceClient:
"""推理服务客户端"""
def __init__(self, config: InferenceConfig):
self.config = config
self.session: Optional[aiohttp.ClientSession] = None
async def __aenter__(self):
self.session = aiohttp.ClientSession()
return self
async def __aexit__(self, *args):
if self.session:
await self.session.close()
async def generate(
self, prompt: str, system_prompt: str = "", stream: bool = True
) -> str:
"""同步生成"""
url = f"{self.config.base_url}/v1/completions"
headers = {"Content-Type": "application/json"}
payload = {
"model": self.config.model_name,
"prompt": prompt,
"max_tokens": self.config.max_tokens,
"temperature": self.config.temperature,
"top_p": self.config.top_p,
"stream": stream,
}
if system_prompt:
payload["prompt"] = (
f"System: {system_prompt}\n\nUser: {prompt}\n\nAssistant:"
)
async with self.session.post(
url,
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(total=self.config.timeout),
) as response:
if response.status != 200:
error = await response.text()
raise Exception(f"Inference failed: {error}")
if stream:
return await self._handle_stream(response)
else:
result = await response.json()
return result["choices"][0]["text"]
async def _handle_stream(self, response: aiohttp.ClientResponse) -> str:
"""处理流式响应"""
full_text = ""
async for line in response.content:
line = line.decode("utf-8").strip()
if not line or line == "data: [DONE]":
continue
if line.startswith("data: "):
data = json.loads(line[6:])
if "choices" in data and len(data["choices"]) > 0:
delta = data["choices"][0].get("delta", {})
if "content" in delta:
full_text += delta["content"]
return full_text
async def batch_generate(
self, prompts: list[str], max_concurrency: int = 5
) -> list[str]:
"""批量生成（带并发控制）"""
semaphore = asyncio.Semaphore(max_concurrency)
async def generate_with_semaphore(prompt: str, idx: int):
async with semaphore:
try:
return await self.generate(prompt)
except Exception as e:
print(f"Prompt {idx} failed: {e}")
return ""
tasks = [
generate_with_semaphore(prompt, idx) for idx, prompt in enumerate(prompts)
]
return await asyncio.gather(*tasks)
3.3 数据管理：AI 的燃料
数据是 AI 系统的"燃料"，但管理不好就是"消防隐患"。
# data/data_manager.py
"""
数据管理模块 - 支持训练数据和知识库管理
"""
from dataclasses import dataclass
from typing import List, Optional, Dict
from enum import Enum
import json
from pathlib import Path
import hashlib
from datetime import datetime
class DataType(Enum):
"""数据类型"""
TRAINING = "training"
VALIDATION = "validation"
TEST = "test"
KNOWLEDGE = "knowledge"
@dataclass
class DataRecord:
"""数据记录"""
id: str
content: str
metadata: Dict
data_type: DataType
embedding: Optional[List[float]] = None
created_at: datetime = None
def __post_init__(self):
if self.created_at is None:
self.created_at = datetime.now()
class DataManager:
"""数据管理器"""
def __init__(self, storage_path: str, vector_store_path: str = None):
self.storage_path = Path(storage_path)
self.vector_store_path = vector_store_path
self.storage_path.mkdir(parents=True, exist_ok=True)
# 内存索引（生产环境应该用数据库）
self.records: Dict[str, DataRecord] = {}
self._load_index()
def _load_index(self):
"""加载索引"""
index_file = self.storage_path / "index.json"
if index_file.exists():
with open(index_file) as f:
data = json.load(f)
for record_data in data.get("records", []):
record_data["data_type"] = DataType(record_data["data_type"])
record_data["created_at"] = datetime.fromisoformat(
record_data["created_at"]
)
self.records[record_data["id"]] = DataRecord(**record_data)
def _save_index(self):
"""保存索引"""
index_file = self.storage_path / "index.json"
data = {
"records": [
{
"id": r.id,
"content": r.content,
"metadata": r.metadata,
"data_type": r.data_type.value,
"created_at": r.created_at.isoformat(),
}
for r in self.records.values()
]
}
with open(index_file, "w") as f:
json.dump(data, f, indent=2, ensure_ascii=False)
def add_record(
self, content: str, metadata: Dict, data_type: DataType
) -> DataRecord:
"""添加数据记录"""
record_id = hashlib.md5(
f"{content}{datetime.now().isoformat()}".encode()
).hexdigest()[:16]
record = DataRecord(
id=record_id, content=content, metadata=metadata, data_type=data_type
)
self.records[record_id] = record
self._save_index()
return record
def add_batch(self, records: List[Dict], data_type: DataType) -> List[DataRecord]:
"""批量添加记录"""
created = []
for record in records:
created.append(
self.add_record(
content=record["content"],
metadata=record.get("metadata", {}),
data_type=data_type,
)
)
return created
def get_records(
self, data_type: Optional[DataType] = None, limit: int = 100
) -> List[DataRecord]:
"""获取记录"""
records = list(self.records.values())
if data_type:
records = [r for r in records if r.data_type == data_type]
return records[:limit]
def search_by_content(self, query: str, limit: int = 10) -> List[DataRecord]:
"""简单关键词搜索（生产环境应该用向量检索）"""
query_lower = query.lower()
scored = []
for record in self.records.values():
# 计算关键词匹配分数
content_lower = record.content.lower()
words = query_lower.split()
score = sum(1 for w in words if w in content_lower)
if score > 0:
scored.append((score, record))
scored.sort(key=lambda x: x[0], reverse=True)
return [r for _, r in scored[:limit]]3.4 可观测性：理解 AI 行为
AI 系统的可观测性比传统系统更复杂：
┌─────────────────────────────────────────────────────────────────┐
│ AI 可观测性指标体系 │
│ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ 请求级指标 │ │
│ │ TTFT (Time To First Token) 首 token 延迟 │ │
│ │ TPOT (Time Per Output Token) 每个 token 延迟 │ │
│ │ E2E Latency 端到端延迟 │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ 系统级指标 │ │
│ │ GPU Utilization GPU 利用率 │ │
│ │ KV Cache Hit Rate 缓存命中率 │ │
│ │ Batch Size 批大小 │ │
│ │ Queue Length 请求队列长度 │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ 业务级指标 │ │
│ │ Token 消耗 成本 │ │
│ │ 错误率 稳定性 │ │
│ │ 质量评分 效果 │ │
│ │ 用户满意度 体验 │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘# observability/metrics_collector.py
"""
AI 可观测性指标收集器
"""
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from datetime import datetime, timedelta
from enum import Enum
import time
class MetricType(Enum):
"""指标类型"""
COUNTER = "counter" # 计数器
GAUGE = "gauge" # 瞬时值
HISTOGRAM = "histogram" # 分布
@dataclass
class MetricPoint:
"""指标点"""
name: str
value: float
timestamp: datetime
labels: Dict[str, str] = field(default_factory=dict)
class MetricsCollector:
"""指标收集器"""
def __init__(self):
self.metrics: Dict[str, List[MetricPoint]] = {}
self._request_start: Dict[str, float] = {}
def record_request_start(self, request_id: str):
"""记录请求开始"""
self._request_start[request_id] = time.time()
def record_request_end(
self, request_id: str, labels: Optional[Dict[str, str]] = None
):
"""记录请求结束"""
if request_id not in self._request_start:
return
duration = time.time() - self._request_start[request_id]
labels = labels or {}
self.record(
name="request_duration_seconds",
value=duration,
metric_type=MetricType.HISTOGRAM,
labels=labels,
)
del self._request_start[request_id]
def record_token_usage(
self,
prompt_tokens: int,
completion_tokens: int,
labels: Optional[Dict[str, str]] = None,
):
"""记录 Token 消耗"""
labels = labels or {}
self.record("prompt_tokens", prompt_tokens, MetricType.COUNTER, labels)
self.record("completion_tokens", completion_tokens, MetricType.COUNTER, labels)
self.record(
"total_tokens",
prompt_tokens + completion_tokens,
MetricType.COUNTER,
labels,
)
def record_ttft(self, ttft_seconds: float, labels: Optional[Dict[str, str]] = None):
"""记录 TTFT"""
self.record("ttft_seconds", ttft_seconds, MetricType.HISTOGRAM, labels or {})
def record_gpu_utilization(self, gpu_id: int, utilization: float):
"""记录 GPU 利用率"""
self.record(
"gpu_utilization", utilization, MetricType.GAUGE, {"gpu_id": str(gpu_id)}
)
def record(
self, name: str, value: float, metric_type: MetricType, labels: Dict = None
):
"""通用记录方法"""
point = MetricPoint(
name=name, value=value, timestamp=datetime.now(), labels=labels or {}
)
if name not in self.metrics:
self.metrics[name] = []
self.metrics[name].append(point)
# 保留最近 24 小时的数据
cutoff = datetime.now() - timedelta(hours=24)
self.metrics[name] = [p for p in self.metrics[name] if p.timestamp > cutoff]
def get_stats(self, name: str, window: timedelta = None) -> Dict:
"""获取指标统计"""
if name not in self.metrics:
return {}
points = self.metrics[name]
if window:
cutoff = datetime.now() - window
points = [p for p in points if p.timestamp > cutoff]
values = [p.value for p in points]
if not values:
return {}
values_sorted = sorted(values)
n = len(values_sorted)
return {
"count": n,
"sum": sum(values),
"mean": sum(values) / n,
"min": values_sorted[0],
"max": values_sorted[-1],
"p50": values_sorted[int(n * 0.5)],
"p95": values_sorted[int(n * 0.95)],
"p99": values_sorted[int(n * 0.99)],
}
def export_prometheus(self) -> str:
"""导出 Prometheus 格式"""
lines = []
for name, points in self.metrics.items():
if not points:
continue
stats = self.get_stats(name)
labels_str = ""
if points[0].labels:
labels_str = (
"{"
+ ",".join(f'{k}="{v}"' for k, v in points[0].labels.items())
+ "}"
)
for stat_name, value in stats.items():
if stat_name == "count":
continue
metric_name = f"{name}_{stat_name}"
lines.append(f"{metric_name}{labels_str} {value}")
return "\n".join(lines)四、应用层
4.1 Agent 编排：让 AI 做事
Agent 是 AI 应用的"执行引擎"，负责：
理解用户意图
规划执行步骤
调用工具
处理异常
# agent/agent_builder.py
"""
基于 LangGraph 的 Agent 编排框架
"""
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
# 定义 Agent 状态
class AgentState(TypedDict):
"""Agent 状态"""
messages: Annotated[list[BaseMessage], operator.add]
intent: str # 识别的意图
entities: dict # 提取的实体
next_action: str # 下一步动作
context: dict # 额外上下文
loop_count: int # 循环次数（防止死循环）
def create_agent_flow(tools: list):
"""创建 Agent 工作流"""
# 构建状态图
workflow = StateGraph(AgentState)
# 添加节点
workflow.add_node("intent_classifier", intent_classifier_node)
workflow.add_node("entity_extractor", entity_extractor_node)
workflow.add_node("tool_executor", tool_executor_node)
workflow.add_node("response_generator", response_generator_node)
# 设置入口
workflow.set_entry_point("intent_classifier")
# 定义边
workflow.add_edge("intent_classifier", "entity_extractor")
# 条件边：根据意图决定下一步
workflow.add_conditional_edges(
"entity_extractor",
decide_next_step,
{"execute_tool": "tool_executor", "generate_response": "response_generator"},
)
# 循环边：工具执行后可能需要再次判断
workflow.add_edge("tool_executor", "intent_classifier")
workflow.add_edge("response_generator", END)
return workflow.compile()
def intent_classifier_node(state: AgentState) -> AgentState:
"""意图分类节点"""
llm = ChatOpenAI(model="gpt-4o", temperature=0)
last_message = state["messages"][-1].content
# 简单意图分类
prompt = f"""分析用户消息的意图，分类到以下类别之一：
- search: 需要搜索信息
- calculate: 需要计算
- query: 查询数据
- general: 一般对话
用户消息: {last_message}
只返回一个词：search, calculate, query, 或 general
"""
response = llm.invoke([HumanMessage(content=prompt)])
intent = response.content.strip().lower()
return {
**state,
"intent": intent,
"next_action": intent,
"loop_count": state.get("loop_count", 0) + 1,
}
def entity_extractor_node(state: AgentState) -> AgentState:
"""实体提取节点"""
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
last_message = state["messages"][-1].content
prompt = f"""从用户消息中提取关键实体，以 JSON 格式返回：
{{
"entities": {{...}}, // 提取的实体
"requires_tools": true/false // 是否需要调用工具
}}
用户消息: {last_message}
"""
response = llm.invoke([HumanMessage(content=prompt)])
# 解析响应（简化处理）
import json
try:
result = json.loads(response.content)
entities = result.get("entities", {})
requires_tools = result.get("requires_tools", False)
except:
entities = {}
requires_tools = False
return {
**state,
"entities": entities,
"next_action": "execute_tool" if requires_tools else "generate_response",
}
def decide_next_step(state: AgentState) -> str:
"""决定下一步"""
# 检查循环次数
if state.get("loop_count", 0) > 10:
return "generate_response"
return state.get("next_action", "generate_response")
def tool_executor_node(state: AgentState) -> AgentState:
"""工具执行节点"""
# 执行具体工具
# 这里需要根据 state["intent"] 和 state["entities"] 调用对应工具
# 简化处理
executed_result = f"已执行 {state['intent']} 相关操作"
return {
**state,
"messages": state["messages"] + [AIMessage(content=executed_result)],
}
def response_generator_node(state: AgentState) -> AgentState:
"""响应生成节点"""
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
# 收集上下文
context = {
"intent": state.get("intent", "general"),
"entities": state.get("entities", {}),
"messages": [m.content for m in state["messages"][-5:]],
}
prompt = f"""基于以下上下文生成回复：
{context}
"""
response = llm.invoke([HumanMessage(content=prompt)])
return {**state, "messages": state["messages"] + [response]}
4.2 RAG：让 AI 有知识
RAG（检索增强生成）是 AI 应用的核心能力之一。
# rag/rag_pipeline.py
"""
RAG 流程实现
"""
from dataclasses import dataclass
from typing import List, Optional, Dict
import numpy as np
@dataclass
class RetrievedChunk:
"""检索到的片段"""
content: str
score: float
source: str
metadata: Dict
class RAGPipeline:
"""RAG 流水线"""
def __init__(self, embedding_model, vector_store, reranker=None, llm=None):
self.embedding_model = embedding_model
self.vector_store = vector_store
self.reranker = reranker
self.llm = llm
def retrieve(
self, query: str, top_k: int = 10, rerank: bool = True
) -> List[RetrievedChunk]:
"""检索相关片段"""
# 向量化查询
query_embedding = self.embedding_model.encode(query)
# 向量检索
candidates = self.vector_store.search(
query_embedding, top_k * 3 if rerank else top_k # 多检索一些用于重排
)
if not candidates:
return []
if rerank and self.reranker:
# 重排
results = self.reranker.rerank(
query=query, documents=[c["content"] for c in candidates], top_n=top_k
)
return [
RetrievedChunk(
content=candidates[r["index"]]["content"],
score=r["score"],
source=candidates[r["index"]].get("source", ""),
metadata=candidates[r["index"]].get("metadata", {}),
)
for r in results
]
else:
return [
RetrievedChunk(
content=c["content"],
score=c["score"],
source=c.get("source", ""),
metadata=c.get("metadata", {}),
)
for c in candidates[:top_k]
]
def generate(
self, query: str, context_chunks: List[RetrievedChunk], stream: bool = False
):
"""基于检索结果生成回答"""
if not self.llm:
raise ValueError("LLM not configured")
# 构建提示词
context_text = "\n\n".join(
f"[来源 {i+1}: {chunk.source}]\n{chunk.content}"
for i, chunk in enumerate(context_chunks)
)
prompt = f"""基于以下参考资料回答问题。如果资料中没有相关信息，请说明不知道。
参考资料:
{context_text}
问题: {query}
回答（引用来源编号）:
"""
if stream:
return self.llm.stream(prompt)
else:
return self.llm.invoke(prompt)
def query(self, query: str, top_k: int = 5, use_rerank: bool = True) -> Dict:
"""完整 RAG 查询"""
# 检索
chunks = self.retrieve(query, top_k, use_rerank)
# 生成
response = self.generate(query, chunks)
return {
"query": query,
"chunks": chunks,
"response": response.content if hasattr(response, "content") else response,
"chunk_count": len(chunks),
}4.3 工具调用：扩展 AI 能力
# tools/tool_registry.py
"""
工具注册与管理
"""
from typing import Dict, List, Callable, Any
from dataclasses import dataclass
from enum import Enum
import json
class ToolCategory(Enum):
"""工具类别"""
SEARCH = "search" # 搜索
DATABASE = "database" # 数据库
API = "api" # API 调用
FILE = "file" # 文件操作
COMPUTE = "compute" # 计算
@dataclass
class ToolDefinition:
"""工具定义"""
name: str
description: str
parameters: Dict # JSON Schema 格式
category: ToolCategory
handler: Callable
class ToolRegistry:
"""工具注册表"""
def __init__(self):
self.tools: Dict[str, ToolDefinition] = {}
def register(
self,
name: str,
description: str,
parameters: Dict,
category: ToolCategory,
handler: Callable,
):
"""注册工具"""
self.tools[name] = ToolDefinition(
name=name,
description=description,
parameters=parameters,
category=category,
handler=handler,
)
def get_openapi_spec(self) -> Dict:
"""生成 OpenAPI 格式的工具定义（用于 function calling）"""
tools = []
for tool in self.tools.values():
tools.append(
{
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.parameters,
},
}
)
return {"tools": tools}
def execute(self, tool_name: str, parameters: Dict) -> Any:
"""执行工具"""
if tool_name not in self.tools:
raise ValueError(f"Tool not found: {tool_name}")
tool = self.tools[tool_name]
return tool.handler(**parameters)
# 预置工具示例
def create_search_tool(search_function: Callable) -> ToolDefinition:
"""创建搜索工具"""
return ToolDefinition(
name="web_search",
description="搜索互联网获取最新信息",
parameters={
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"},
"max_results": {
"type": "integer",
"description": "最大结果数",
"default": 5,
},
},
"required": ["query"],
},
category=ToolCategory.SEARCH,
handler=search_function,
)
# 工具执行器
class ToolExecutor:
"""工具执行器"""
def __init__(self, registry: ToolRegistry):
self.registry = registry
async def execute_tool_call(self, tool_call: Dict) -> str:
"""执行单个工具调用"""
function = tool_call.get("function", {})
name = function.get("name")
arguments = json.loads(function.get("arguments", "{}"))
try:
result = self.registry.execute(name, arguments)
return json.dumps({"success": True, "result": result})
except Exception as e:
return json.dumps({"success": False, "error": str(e)})
async def execute_tools(self, tool_calls: List[Dict]) -> List[str]:
"""批量执行工具"""
return [await self.execute_tool_call(tc) for tc in tool_calls]五、架构演进
5.1 项目结构示例
ai-project/
├── config/
│ ├── models.yaml # 模型配置
│ ├── deployment.yaml # 部署配置
│ └── prompts.yaml # Prompt 模板
│
├── model_layer/ # 模型层
│ ├── registry.py # 模型注册表
│ ├── evaluator.py # 模型评测
│ └── fine_tune/
│ ├── data_prep.py # 数据准备
│ └── trainer.py # 训练脚本
│
├── platform_layer/ # 平台层
│ ├── inference/
│ │ ├── server.py # 推理服务
│ │ ├── client.py # 客户端
│ │ └── docker/
│ │ └── Dockerfile
│ ├── training/
│ │ ├── job.py # 训练任务
│ │ └── scheduler.py # 任务调度
│ ├── data/
│ │ ├── manager.py # 数据管理
│ │ └── vector_store.py # 向量存储
│ └── observability/
│ ├── metrics.py # 指标收集
│ └── tracing.py # 链路追踪
│
├── application_layer/ # 应用层
│ ├── agent/
│ │ ├── workflow.py # Agent 工作流
│ │ └── nodes/ # 节点实现
│ ├── rag/
│ │ ├── pipeline.py # RAG 流水线
│ │ ├── chunker.py # 分块策略
│ │ └── retriever.py # 检索器
│ ├── tools/ # 工具定义
│ │ ├── registry.py
│ │ └── implementations/
│ └── ui/ # 用户界面
│
├── tests/
│ ├── unit/
│ ├── integration/
│ └── eval/
│
├── scripts/
│ ├── deploy.sh
│ └── eval.sh
│
└── docs/
├── architecture.md
└── deployment.md5.2 三层依赖关系
┌─────────────────────────────────────────────────────────────────┐
│ 依赖关系图 │
│ │
│ 应用层 │
│ │ │
│ │ 调用 │
│ ▼ │
│ 平台层 ◄────────────────────────────────┐ │
│ │ │ │
│ │ 调用 │ 抽象/配置 │
│ ▼ │ │
│ 模型层 │ │
│ │
│ 依赖方向: 应用层 → 平台层 → 模型层 │
│ 变更影响: 模型层变更 < 平台层变更 < 应用层变更 │
│ │
└─────────────────────────────────────────────────────────────────┘关键原则：
应用层不知道模型的具体实现
平台层不知道业务逻辑
每层只依赖下层的抽象，不依赖具体实现
六、与传统架构的类比
总结
AI 工程化的三层架构是笔者在多个项目中的实战总结。核心要点：
模型层：解决"模型从哪来"，包括选型、微调、评测、版本管理
平台层：解决"模型怎么跑"，包括训练、推理、数据、可观测性
应用层：解决"模型怎么用"，包括 Agent、RAG、工具、交互
三层之间松耦合、高内聚，可以独立演进。每一层都有其独特的技术挑战，需要不同的专业知识。
AI 工程化不是"给 AI 代码加一层包装"，而是需要从根本上重新思考软件架构。
随着 AI 应用的普及，这种三层架构（或其变体）将成为事实标准。
评论区