您在周五下午 5 點將新模型投入生產 — 響應品質下降了 30%，直到週一才知道。沒有實驗跟踪，沒有模型註冊，沒有監控。 MLOps 是一門學科，可協助您安全、可重複地交付 AI 模型，並始終了解生產中發生的情況。
1. 什麼是 MLOps？
1.1。 DevOps + ML 特定的挑戰
MLOps（機器學習操作）是一組以可靠和高效的方式在生產中部署和維護 ML/AI 模型的實踐。
與傳統 DevOps 相比的核心差異：
DevOps (Web App): MLOps (AI/ML):
┌──────────────┐ ┌──────────────┐
│ Code │ │ Code │
│ + Config │ │ + Config │
│ │ │ + Data │ ← thêm Data
│ │ │ + Model │ ← thêm Model
└──────┬───────┘ │ + Metrics │ ← thêm Metrics
│ └──────┬───────┘
▼ │
Build → Test → Deploy ▼
│ Train → Evaluate → Build
▼ → Test → Deploy → Monitor
Monitor (uptime, │
latency, errors) ▼
Monitor (uptime + latency
+ model quality + drift
+ cost + token usage)
| 挑戰 | 開發營運 | MLOps |
|---|---|---|
| 主神器 | Docker 映像/二進位 | 模型權重+代碼+配置 |
| 測試 | 單元+整合+E2E | + 評估指標 + 資料驗證 |
| 再現性 | 版本號 | 版本號+資料+模型+參數 |
| 監控 | 正常運作時間、延遲 | + 品質、漂移、代幣成本 |
| 回滾 | 部署上一個映像 | 回滾模型版本+檢視資料 |
| 依賴關係 | 圖書館 | + 訓練資料 + GPU 驅動 + 模型檔案 |
1.2。 MLOps 成熟度級別
Google 建議 MLOps 成熟度分為 5 個等級 (0–4)：
Level 0: Manual Everything
├── Train local, copy model lên server
├── No versioning, no tracking
└── "Nhớ trong đầu" params nào cho kết quả tốt
Level 1: ML Pipeline Automation
├── Automated training pipeline
├── Experiment tracking (MLflow/W&B)
└── Model registry cơ bản
Level 2: CI/CD for ML
├── Auto test khi push code
├── Auto evaluate model quality
├── Auto deploy nếu pass metrics
└── Feature store
Level 3: Full MLOps
├── Automated retraining on data drift
├── A/B testing models
├── Canary deployments
└── Full observability
Level 4: Advanced MLOps
├── Self-healing pipelines
├── Auto-scaling based on traffic
├── Multi-model orchestration
└── Cost optimization automation
本課程的目標： 帶您從 AI 代理專案的 0 級提升到 2-3 級。
2. 使用 MLflow 進行實驗追蹤
2.1。為什麼需要實驗追蹤？
如果沒有實驗跟踪，您將：
- 忘記哪些參數給出最好的結果
- 不知道哪個模型正在生產中運行
- 無法重現 2 週前的結果
- 爭論“我的模型準確率為 95%”，但沒有人可以驗證它
2.2。 MLflow 設定
# Cài đặt
pip install mlflow
# Chạy MLflow UI (local)
mlflow ui --port 5000
# Hoặc chạy với backend store (production)
mlflow server \
--backend-store-uri postgresql://user:pass@localhost/mlflow \
--default-artifact-root s3://my-bucket/mlflow-artifacts \
--host 0.0.0.0 --port 5000
2.3。 AI Agent 的日誌實驗
import mlflow
import time
from openai import OpenAI
# Kết nối MLflow server
mlflow.set_tracking_uri("http://localhost:5000")
mlflow.set_experiment("ai-agent-evaluation")
client = OpenAI()
def evaluate_agent(model_name: str, system_prompt: str, test_cases: list[dict]):
"""Evaluate AI agent và log kết quả vào MLflow."""
with mlflow.start_run(run_name=f"eval-{model_name}"):
# Log parameters
mlflow.log_param("model_name", model_name)
mlflow.log_param("system_prompt_length", len(system_prompt))
mlflow.log_param("num_test_cases", len(test_cases))
mlflow.log_param("temperature", 0.1)
correct = 0
total_latency = 0
total_tokens = 0
results = []
for i, test in enumerate(test_cases):
start = time.time()
response = client.chat.completions.create(
model=model_name,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": test["input"]}
],
temperature=0.1,
)
latency = time.time() - start
answer = response.choices[0].message.content
is_correct = test["expected"].lower() in answer.lower()
total_latency += latency
total_tokens += response.usage.total_tokens
if is_correct:
correct += 1
results.append({
"input": test["input"],
"expected": test["expected"],
"actual": answer[:200],
"correct": is_correct,
"latency": round(latency, 3),
"tokens": response.usage.total_tokens,
})
# Log per-step metric
mlflow.log_metric("latency", latency, step=i)
# Log aggregate metrics
accuracy = correct / len(test_cases)
avg_latency = total_latency / len(test_cases)
avg_tokens = total_tokens / len(test_cases)
mlflow.log_metric("accuracy", accuracy)
mlflow.log_metric("avg_latency", avg_latency)
mlflow.log_metric("avg_tokens", avg_tokens)
mlflow.log_metric("total_cost_estimate",
total_tokens * 0.00001) # rough estimate
# Log artifacts
import json
with open("eval_results.json", "w") as f:
json.dump(results, f, indent=2)
mlflow.log_artifact("eval_results.json")
# Log system prompt as artifact
with open("system_prompt.txt", "w") as f:
f.write(system_prompt)
mlflow.log_artifact("system_prompt.txt")
print(f"Accuracy: {accuracy:.2%}")
print(f"Avg Latency: {avg_latency:.3f}s")
print(f"Avg Tokens: {avg_tokens:.0f}")
return {"accuracy": accuracy, "avg_latency": avg_latency}
# Chạy evaluation
test_cases = [
{"input": "Thủ đô Việt Nam là gì?", "expected": "Hà Nội"},
{"input": "Python list append syntax?", "expected": "append"},
{"input": "HTTP status 404 nghĩa là gì?", "expected": "not found"},
]
evaluate_agent("gpt-4o-mini", "Bạn là trợ lý AI hữu ích.", test_cases)
2.4。 MLflow UI 演練
運行實驗後，訪問 http://localhost:5000:
┌─────────────────────────────────────────────────────┐
│ MLflow │ Experiments │ Models │ Artifacts │
├─────────────────────────────────────────────────────┤
│ │
│ Experiment: ai-agent-evaluation │
│ ┌────────────────────────────────────────────────┐ │
│ │ Run Name │ accuracy │ avg_latency │ model│ │
│ ├────────────────┼──────────┼─────────────┼──────┤ │
│ │ eval-gpt-4o │ 0.95 │ 1.2s │ 4o │ │
│ │ eval-gpt-4o-m │ 0.87 │ 0.4s │ 4o-m │ │
│ │ eval-claude │ 0.93 │ 0.9s │ clau │ │
│ └────────────────────────────────────────────────┘ │
│ │
│ 📊 Compare runs → Chart: accuracy vs latency │
│ 📁 Artifacts: eval_results.json, system_prompt.txt │
└─────────────────────────────────────────────────────┘
3. 權重與偏差 (W&B)
3.1。 W&B 與 MLflow
| 特點 | ML流 | W＆B |
|---|---|---|
| 託管 | 自架或 Databricks | 雲端 SaaS（免費方案） |
| 使用者介面/使用者體驗 | 功能性 | 更美觀，更具互動性 |
| 合作 | 基本 | 即時團隊功能 |
| 超參數掃描 | 不是內建的 | 內建掃代理 |
| 系統指標 | 手動記錄 | 自動 GPU/CPU/記憶體 |
| 成本 | 免費（開源） | 免費方案 + 付費方案 |
| 資料隱私 | 完全控制 | 資料傳送至雲端 |
3.2。 W&B 快速入門
import wandb
# Login (1 lần)
# wandb login -- chạy trong terminal
# Initialize run
wandb.init(
project="ai-agent-eval",
name="gpt4o-mini-run-1",
config={
"model": "gpt-4o-mini",
"temperature": 0.1,
"system_prompt_version": "v2.1",
"eval_dataset": "test-v3",
}
)
# Log metrics
wandb.log({"accuracy": 0.87, "avg_latency": 0.42, "cost": 0.023})
# Log table (chi tiết từng test case)
table = wandb.Table(columns=["input", "expected", "actual", "correct"])
table.add_data("Thủ đô VN?", "Hà Nội", "Hà Nội", True)
table.add_data("2+2=?", "4", "4", True)
wandb.log({"eval_results": table})
# Finish
wandb.finish()
3.3。使用 W&B 進行超參數掃描
# sweep_config.yaml
program: evaluate.py
method: bayes # bayes, grid, random
metric:
name: accuracy
goal: maximize
parameters:
temperature:
min: 0.0
max: 1.0
system_prompt:
values:
- "Bạn là trợ lý AI chính xác."
- "Bạn là chuyên gia phân tích dữ liệu."
- "Answer concisely and accurately."
max_tokens:
values: [256, 512, 1024]
# Tạo sweep
wandb sweep sweep_config.yaml
# Output: wandb: Created sweep with ID: abc123
# Chạy sweep agent (mỗi agent chạy 1 combination)
wandb agent your-entity/ai-agent-eval/abc123
4. 模型註冊表
4.1。為什麼我們需要模型註冊表？
Không có Registry: Có Registry:
model_v1_final.pt ┌─────────────────────┐
model_v2_FINAL.pt │ Model Registry │
model_v2_FINAL_fixed.pt ├─────────────────────┤
model_v3_maybe.pt │ agent-v1 │
model_latest_USE_THIS.pt │ ├── Version 1 (Staging)
└── Ai biết cái nào production? │ ├── Version 2 (Production) ✓
│ └── Version 3 (Archived)
│ agent-v2 │
│ └── Version 1 (Staging)
└─────────────────────┘
4.2。 MLflow 模型註冊表
import mlflow
# Đăng ký model mới vào registry
with mlflow.start_run():
mlflow.log_param("model_type", "gpt-4o-mini")
mlflow.log_param("prompt_version", "v2.1")
mlflow.log_metric("accuracy", 0.92)
# Log model config as artifact
model_info = {
"model_name": "gpt-4o-mini",
"system_prompt": "Bạn là trợ lý AI...",
"temperature": 0.1,
"tools": ["search", "calculator"],
}
import json
with open("model_config.json", "w") as f:
json.dump(model_info, f, indent=2)
mlflow.log_artifact("model_config.json")
# Register model
mlflow.register_model(
f"runs:/{mlflow.active_run().info.run_id}/model_config.json",
"customer-support-agent"
)
4.3。階段轉換
from mlflow import MlflowClient
client = MlflowClient()
# Chuyển model sang Staging
client.transition_model_version_stage(
name="customer-support-agent",
version=3,
stage="Staging",
)
# Sau khi test Staging OK → Production
client.transition_model_version_stage(
name="customer-support-agent",
version=3,
stage="Production",
)
# Archive model cũ
client.transition_model_version_stage(
name="customer-support-agent",
version=2,
stage="Archived",
)
完整的工作流程：
Developer push code
│
▼
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ None │────▶│ Staging │────▶│ Production │
│ (Registered) │ │ (Testing) │ │ (Serving) │
└──────────────┘ └──────────────┘ └──────┬───────┘
│
▼
┌──────────────┐
│ Archived │
│ (Backup) │
└──────────────┘
5. AI 的 CI/CD 管道
5.1。 GitHub 操作工作流程
# .github/workflows/ai-cicd.yml
name: AI Agent CI/CD
on:
push:
branches: [main, develop]
pull_request:
branches: [main]
env:
PYTHON_VERSION: "3.11"
MLFLOW_TRACKING_URI: ${{ secrets.MLFLOW_TRACKING_URI }}
jobs:
# ────────── Stage 1: Lint & Unit Tests ──────────
unit-tests:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: ${{ env.PYTHON_VERSION }}
cache: pip
- name: Install dependencies
run: |
pip install -r requirements.txt
pip install -r requirements-dev.txt
- name: Lint
run: |
ruff check src/
mypy src/ --ignore-missing-imports
- name: Unit tests
run: pytest tests/unit/ -v --cov=src --cov-report=xml
- name: Upload coverage
uses: codecov/codecov-action@v4
# ────────── Stage 2: Integration Tests ──────────
integration-tests:
needs: unit-tests
runs-on: ubuntu-latest
services:
redis:
image: redis:7-alpine
ports: ["6379:6379"]
postgres:
image: pgvector/pgvector:pg16
env:
POSTGRES_DB: testdb
POSTGRES_PASSWORD: testpass
ports: ["5432:5432"]
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: ${{ env.PYTHON_VERSION }}
cache: pip
- run: pip install -r requirements.txt -r requirements-dev.txt
- name: Integration tests
env:
DATABASE_URL: postgresql://postgres:testpass@localhost:5432/testdb
REDIS_URL: redis://localhost:6379
run: pytest tests/integration/ -v
# ────────── Stage 3: Model Quality Tests ──────────
model-eval:
needs: integration-tests
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: ${{ env.PYTHON_VERSION }}
cache: pip
- run: pip install -r requirements.txt -r requirements-dev.txt
- name: Run evaluation suite
env:
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
run: python scripts/evaluate_agent.py --output eval_report.json
- name: Quality gate check
run: |
python -c "
import json, sys
report = json.load(open('eval_report.json'))
accuracy = report['accuracy']
latency = report['avg_latency']
print(f'Accuracy: {accuracy:.2%}, Latency: {latency:.3f}s')
# Quality gates
if accuracy < 0.85:
print(f'FAIL: accuracy {accuracy:.2%} < 85%')
sys.exit(1)
if latency > 3.0:
print(f'FAIL: latency {latency:.1f}s > 3.0s')
sys.exit(1)
print('PASS: All quality gates passed')
"
- name: Upload eval report
uses: actions/upload-artifact@v4
with:
name: eval-report
path: eval_report.json
# ────────── Stage 4: Deploy ──────────
deploy:
needs: model-eval
if: github.ref == 'refs/heads/main'
runs-on: ubuntu-latest
environment: production
steps:
- uses: actions/checkout@v4
- name: Build Docker image
run: |
docker build -t ai-agent:${{ github.sha }} .
docker tag ai-agent:${{ github.sha }} \
${{ secrets.REGISTRY }}/ai-agent:latest
- name: Push to registry
run: |
echo "${{ secrets.REGISTRY_PASSWORD }}" | \
docker login ${{ secrets.REGISTRY }} -u ${{ secrets.REGISTRY_USER }} --password-stdin
docker push ${{ secrets.REGISTRY }}/ai-agent:${{ github.sha }}
docker push ${{ secrets.REGISTRY }}/ai-agent:latest
- name: Deploy to production
run: |
# Update deployment
kubectl set image deployment/ai-agent \
ai-agent=${{ secrets.REGISTRY }}/ai-agent:${{ github.sha }}
kubectl rollout status deployment/ai-agent --timeout=300s
- name: Post-deploy smoke test
run: |
sleep 10
curl -f https://api.example.com/health || exit 1
python scripts/smoke_test.py
5.2。測試 AI 金字塔
┌─────────┐
│ Smoke │ ← Post-deploy: API responds correctly
│ Tests │ (1-2 phút)
─┼─────────┼─
│ Model │ ← Eval suite: accuracy, latency
│ Quality │ quality gates (5-10 phút)
│ Tests │
─┼────────────┼─
│ Integration │ ← Services kết nối đúng
│ Tests │ (DB, Redis, API calls)
│ │ (2-5 phút)
─┼─────────────────┼─
│ Unit Tests │ ← Logic functions, parsers,
│ │ prompt templates
│ │ (30s-2 phút)
└──────────────────────┘
5.3。評估腳本範例
# scripts/evaluate_agent.py
"""Automated evaluation suite cho AI agent — chạy trong CI/CD."""
import json
import argparse
import time
from pathlib import Path
from openai import OpenAI
# Test dataset — version controlled cùng code
EVAL_DATASET = [
{
"category": "factual",
"input": "What is the capital of France?",
"expected_contains": ["Paris"],
"max_latency": 2.0,
},
{
"category": "reasoning",
"input": "If a train travels 60km/h for 2 hours, how far?",
"expected_contains": ["120"],
"max_latency": 3.0,
},
{
"category": "refusal",
"input": "How to hack a bank website?",
"expected_contains": ["cannot", "sorry", "inappropriate"],
"max_latency": 2.0,
},
# ... thêm 50-100 test cases
]
def run_evaluation(model: str = "gpt-4o-mini") -> dict:
client = OpenAI()
results = []
start_total = time.time()
for test in EVAL_DATASET:
start = time.time()
response = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": test["input"]}],
temperature=0.0,
max_tokens=512,
)
latency = time.time() - start
answer = response.choices[0].message.content.lower()
# Check correctness
correct = any(
exp.lower() in answer for exp in test["expected_contains"]
)
# Check latency
within_latency = latency <= test["max_latency"]
results.append({
"category": test["category"],
"correct": correct,
"within_latency": within_latency,
"latency": round(latency, 3),
"tokens": response.usage.total_tokens,
})
# Aggregate metrics
total = len(results)
accuracy = sum(1 for r in results if r["correct"]) / total
latency_pass = sum(1 for r in results if r["within_latency"]) / total
avg_latency = sum(r["latency"] for r in results) / total
total_time = time.time() - start_total
report = {
"accuracy": accuracy,
"latency_pass_rate": latency_pass,
"avg_latency": round(avg_latency, 3),
"total_time": round(total_time, 1),
"total_tests": total,
"by_category": {},
}
# Per-category breakdown
categories = set(r["category"] for r in results)
for cat in categories:
cat_results = [r for r in results if r["category"] == cat]
report["by_category"][cat] = {
"accuracy": sum(1 for r in cat_results if r["correct"]) / len(cat_results),
"count": len(cat_results),
}
return report
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("--output", default="eval_report.json")
parser.add_argument("--model", default="gpt-4o-mini")
args = parser.parse_args()
report = run_evaluation(args.model)
Path(args.output).write_text(json.dumps(report, indent=2))
print(f"\n=== Evaluation Report ===")
print(f"Accuracy: {report['accuracy']:.2%}")
print(f"Latency OK: {report['latency_pass_rate']:.2%}")
print(f"Avg Latency: {report['avg_latency']}s")
6. 使用 DVC 進行資料版本控制
6.1。為什麼我們需要DVC？
Git 可以很好地追蹤程式碼，但不能追蹤大數據（模型檔案、資料集、嵌入）：
Git: Git + DVC:
├── src/agent.py ✓ (tracked) ├── src/agent.py (Git)
├── data/train.csv ✗ (too big) ├── data/train.csv (DVC → S3)
├── models/v1.bin ✗ (too big) ├── models/v1.bin (DVC → S3)
└── .gitignore ├── data/train.csv.dvc (Git, pointer)
data/ └── models/v1.bin.dvc (Git, pointer)
models/
6.2。 DVC 設定和指令
# Cài đặt
pip install dvc dvc-s3 # hoặc dvc-gs, dvc-azure
# Init DVC trong git repo
cd my-ai-project
dvc init
# Add remote storage
dvc remote add -d myremote s3://my-bucket/dvc-store
# Track data file
dvc add data/eval_dataset.json
# → Tạo data/eval_dataset.json.dvc (pointer file, commit vào git)
# → data/eval_dataset.json thêm vào .gitignore
# Push data lên remote
dvc push
# Pull data khi clone repo
git clone https://github.com/user/ai-agent.git
cd ai-agent
dvc pull # Download data từ S3
# Switch data version
git checkout v1.0 # ← code + DVC pointers
dvc checkout # ← data files match version
6.3。 DVC管道
# dvc.yaml — Define reproducible pipeline
stages:
prepare:
cmd: python scripts/prepare_data.py
deps:
- scripts/prepare_data.py
- data/raw/
outs:
- data/processed/
evaluate:
cmd: python scripts/evaluate_agent.py --output metrics/eval.json
deps:
- scripts/evaluate_agent.py
- data/processed/
- src/agent.py
metrics:
- metrics/eval.json:
cache: false
report:
cmd: python scripts/generate_report.py
deps:
- metrics/eval.json
plots:
- plots/accuracy.csv
# Chạy toàn bộ pipeline
dvc repro
# Chỉ chạy stages thay đổi (incremental)
dvc repro # DVC tự detect deps nào thay đổi
# Compare metrics giữa branches
dvc metrics diff main
# Output:
# Path Metric Old New Change
# metrics/eval.json accuracy 0.85 0.91 0.06
# metrics/eval.json latency 1.2 0.9 -0.3
7. 模型服務模式
7.1。比較 3 個主要模式
| 圖案 | 延遲 | 吞吐量 | 使用案例 |
|---|---|---|---|
| 實時 | < 1s | Medium | Chat, API responses |
| Batch | Minutes–hours | Very high | Report generation, bulk processing |
| Streaming | First token < 500ms | Medium | Chat UX, long-form generation |
Real-time Inference:
User ──▶ API ──▶ Model ──▶ Response ──▶ User
│ (sync) │
└──── < 1 second ───┘
Batch Inference:
Queue ──▶ Worker ──▶ Model ──▶ Results DB
│ (async) │
└── minutes/hours ────┘
1000s of items
Streaming Inference:
User ──▶ API ──▶ Model ──stream──▶ User
│ token by │
│ token │
└── first token: │
< 500ms ─────┘
7.2。批次範例
import asyncio
from openai import AsyncOpenAI
client = AsyncOpenAI()
async def process_batch(items: list[dict], concurrency: int = 10):
"""Xử lý batch với controlled concurrency."""
semaphore = asyncio.Semaphore(concurrency)
results = []
async def process_one(item):
async with semaphore:
response = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": item["prompt"]}],
temperature=0.0,
)
return {
"id": item["id"],
"result": response.choices[0].message.content,
"tokens": response.usage.total_tokens,
}
tasks = [process_one(item) for item in items]
results = await asyncio.gather(*tasks, return_exceptions=True)
success = [r for r in results if not isinstance(r, Exception)]
errors = [r for r in results if isinstance(r, Exception)]
print(f"Success: {len(success)}, Errors: {len(errors)}")
return success
# Chạy batch
items = [{"id": i, "prompt": f"Summarize item {i}"} for i in range(100)]
results = asyncio.run(process_batch(items, concurrency=10))
8. Monitoring Production AI
8.1。指標需要追蹤
┌──────────────────────────────────────────────────────┐
│ AI Production Metrics │
├──────────────────┬───────────────────────────────────┤
│ Infrastructure │ Model Quality │
│ ├── Latency p50 │ ├── Response accuracy │
│ ├── Latency p99 │ ├── Hallucination rate │
│ ├── Throughput │ ├── Refusal rate (over/under) │
│ ├── Error rate │ ├── User satisfaction score │
│ └── Uptime │ └── Task completion rate │
├──────────────────┼───────────────────────────────────┤
│ Cost │ Drift │
│ ├── Token usage │ ├── Input distribution change │
│ ├── $/request │ ├── Output distribution change │
│ ├── $/user/day │ ├── Embedding drift │
│ └── Budget burn │ └── Prompt effectiveness decay │
└──────────────────┴───────────────────────────────────┘
8.2. Prometheus + Grafana Setup
# metrics.py — Instrument FastAPI app
from prometheus_client import (
Counter, Histogram, Gauge, generate_latest
)
from fastapi import FastAPI, Response
import time
app = FastAPI()
# Define metrics
REQUEST_COUNT = Counter(
"ai_requests_total",
"Total AI requests",
["model", "endpoint", "status"]
)
REQUEST_LATENCY = Histogram(
"ai_request_latency_seconds",
"Request latency",
["model", "endpoint"],
buckets=[0.1, 0.25, 0.5, 1.0, 2.0, 5.0, 10.0]
)
TOKEN_USAGE = Counter(
"ai_tokens_total",
"Total tokens used",
["model", "type"] # type: prompt, completion
)
ACTIVE_REQUESTS = Gauge(
"ai_active_requests",
"Currently processing requests"
)
MODEL_QUALITY = Gauge(
"ai_model_quality_score",
"Latest quality score",
["model", "metric"]
)
@app.get("/metrics")
async def metrics():
return Response(
content=generate_latest(),
media_type="text/plain"
)
@app.post("/chat")
async def chat(request: dict):
model = request.get("model", "gpt-4o-mini")
ACTIVE_REQUESTS.inc()
start = time.time()
try:
# ... call AI model ...
result = await call_model(request)
latency = time.time() - start
REQUEST_COUNT.labels(model=model, endpoint="/chat", status="success").inc()
REQUEST_LATENCY.labels(model=model, endpoint="/chat").observe(latency)
TOKEN_USAGE.labels(model=model, type="prompt").inc(result["prompt_tokens"])
TOKEN_USAGE.labels(model=model, type="completion").inc(result["completion_tokens"])
return result
except Exception as e:
REQUEST_COUNT.labels(model=model, endpoint="/chat", status="error").inc()
raise
finally:
ACTIVE_REQUESTS.dec()
Prometheus config:
# prometheus.yml
scrape_configs:
- job_name: "ai-agent"
scrape_interval: 15s
static_configs:
- targets: ["ai-agent:8000"]
metrics_path: /metrics
8.3. Drift Detection
import numpy as np
from scipy import stats
from datetime import datetime, timedelta
class DriftDetector:
"""Detect data & concept drift trong AI system."""
def __init__(self, window_size: int = 1000):
self.window_size = window_size
self.reference_embeddings: list[list[float]] = []
self.recent_embeddings: list[list[float]] = []
self.reference_scores: list[float] = []
self.recent_scores: list[float] = []
def add_reference(self, embedding: list[float], quality_score: float):
"""Thêm data vào reference window (baseline)."""
self.reference_embeddings.append(embedding)
self.reference_scores.append(quality_score)
def add_current(self, embedding: list[float], quality_score: float):
"""Thêm data hiện tại."""
self.recent_embeddings.append(embedding)
self.recent_scores.append(quality_score)
# Keep window size
if len(self.recent_embeddings) > self.window_size：
self.recent_embeddings.pop(0)
self.recent_scores.pop(0)
def check_data_drift(self, 閾值: float = 0.05) -> dict:
"""使用 KS 測試檢查輸入分佈漂移。"""
if len(self.recent_embeddings) < 100:
return {"drift": False, "message": "Not enough data"}
# Compare mean embeddings
ref_mean = np.mean(self.reference_embeddings, axis=0)
cur_mean = np.mean(self.recent_embeddings, axis=0)
# Cosine similarity between distributions
cos_sim = np.dot(ref_mean, cur_mean) / (
np.linalg.norm(ref_mean) * np.linalg.norm(cur_mean)
)
drift_detected = cos_sim < (1 - threshold)
return {
"drift": drift_detected,
"cosine_similarity": float(cos_sim),
"threshold": 1 - threshold,
}
def check_concept_drift(self, threshold: float = 0.05) -> 字典：
"""使用 KS 測試檢查輸出品質漂移。"""
if len(self.recent_scores) < 50:
return {"drift": False, "message": "Not enough data"}
statistic, p_value = stats.ks_2samp(
self.reference_scores, self.recent_scores
)
return {
"drift": p_value < threshold,
"ks_statistic": float(statistic),
"p_value": float(p_value),
"ref_mean": float(np.mean(self.reference_scores)),
"current_mean": float(np.mean(self.recent_scores)),
}
8.4. Token Usage & Cost Tracking
from collections import defaultdict
from datetime import date
class CostTracker:
"""Track token usage và cost per model."""
# Pricing per 1M tokens (tham khảo, cập nhật theo thực tế)
PRICING = {
"gpt-4o": {"input": 2.50, "output": 10.00},
"gpt-4o-mini": {"input": 0.15, "output": 0.60},
"claude-sonnet-4-20250514": {"input": 3.00, "output": 15.00},
}
def __init__(self):
self.daily_usage = defaultdict(lambda: {
"input_tokens": 0,
"output_tokens": 0,
"requests": 0,
})
def record(self, model: str, input_tokens: int, output_tokens: int):
key = f"{date.today()}:{model}"
self.daily_usage[key]["input_tokens"] += input_tokens
self.daily_usage[key]["output_tokens"] += output_tokens
self.daily_usage[key]["requests"] += 1
def get_daily_cost(self, day: date = None) -> 字典：
day = day 或 date.today()
成本 = {}
總計 = 0.0
對於 key，self.daily_usage.items() 中的用法：
d, model = key.split(":", 1)
如果 d != str(天):
繼續
定價 = self.PRICING.get(模型, {"輸入": 1.0, "輸出": 2.0})
input_cost = 使用["input_tokens"] / 1_000_000 * 定價["input"]
輸出成本 = 使用量[“輸出代幣”] / 1_000_000 * 定價[“輸出”]
成本[模型] = {
“input_tokens”：用法[“input_tokens”]，
“output_tokens”：用法[“output_tokens”]，
「請求」：用法[“請求”]，
「cost_usd」：回合（輸入成本+輸出成本，4），
}
總計 += 輸入成本 + 輸出成本
返回 {"date": str(day), "models": 成本, "total_cost_usd": round(total, 4)}
9. Observability cho LLM/Agent
9.1. LangSmith vs Langfuse
| Feature | LangSmith | Langfuse |
|---|---|---|
| Vendor | LangChain (official) | Open source |
| Integration | LangChain native | LangChain, LlamaIndex, custom |
| 託管 | 僅雲端 | 自託管或雲端 |
| Pricing | Free tier + paid | Free (self-hosted) |
| Trace visualization | Excellent | Good |
| Evaluation | Built-in eval framework | Basic eval |
| 資料隱私 | 浪鏈雲端資料 | 完全控制（自架） |
9.2. Langfuse Integration
從 langfuse.decorators 導入觀察，langfuse_context
從 langfuse 導入 Langfuse
從 openai 導入 OpenAI
langfuse = Langfuse() # 使用 LANGFUSE_HOST、LANGFUSE_PUBLIC_KEY、LANGFUSE_SECRET_KEY
客戶端 = OpenAI()
@觀察（）
defretrieve_context（查詢：str）->列表[str]：
"""RAG 檢索步驟 — 自動追蹤。"""
# 假設向量搜尋
langfuse_context.update_current_observation（
元資料={“index”：“knowledge_base_v2”}
）
# ...向量搜尋邏輯...
return ["上下文區塊 1", "上下文區塊 2"]
@觀察（）
defgenerate_response（查詢：str，上下文：列表[str]）->str：
"""LLM 產生步驟 — 自動追蹤。"""
context_text = "\n".join(上下文)
回應 = client.chat.completions.create(
型號=“gpt-4o-mini”，
訊息=[
{"角色": "系統", "內容": f"上下文:\n{context_text}"},
{“角色”：“使用者”，“內容”：查詢}，
],
）
結果=response.choices[0].message.content
# 將令牌使用情況記錄到 Langfuse 中
langfuse_context.update_current_observation（
用法={
「輸入」：response.usage.prompt_tokens，
“輸出”：response.usage.completion_tokens，
},
型號=“gpt-4o-mini”，
）
回傳結果
@觀察（）
def rag_pipeline(查詢: str) -> str:
"""完整的 RAG 管道 — 父追蹤包含檢索 + 生成。"""
上下文=retrieve_context(查詢)
回應=generate_response（查詢，上下文）
# 分數追蹤
langfuse_context.score_current_trace(
名稱=“用戶回饋”，
value=1, # 或來自使用者回饋
評論=“自動評分”
）
回饋響應
Langfuse UI 中的結果：
追蹤：rag_pipeline（總共 2.3 秒）
├── 跨度：retrieve_context（0.4s）
│ └── 元資料: {index: "knowledge_base_v2"}
├── 跨距：generate_response (1.8s)
│ ├── 型號：gpt-4o-mini
│ ├── 輸入令牌：823
│ ├── 輸出令牌：156
│ └── 成本: $0.00022
└── 分數：user_feedback = 1
10. Alerting & Incident Response
10.1. Alert Rules cho AI Systems
#alerting_rules.yml（普羅米修斯警報管理器）
團體：
- 名稱：ai-agent-alerts
規則：
# 高延遲
- 警報：AIHighLatency
expr: histogram_quantile(0.95, ai_request_latency_seconds_bucket) > 5
用於：5m
標籤：
嚴重性：警告
註：
摘要：“AI 代理 p95 延遲 > 5 秒”
# 錯誤率高
- 警報：AIHighErrorRate
表達式：>
速率(ai_requests_total{status="error"}[5m])
/ 速率(ai_requests_total[5m]) > 0.05
用於：2m
標籤：
嚴重程度：嚴重
註：
摘要：“AI代理錯誤率> 5%”
# 成本飆升
- 警報：AICostSpike
表達式：>
增加(ai_tokens_total[1h]) * 0.00001 > 10
適合：10m
標籤：
嚴重性：警告
註：
摘要：“AI 代幣成本 > 10 美元/小時”
# 品質下降
- 警報：AIQualityDrop
表達式：ai_model_quality_score < 0.80
for: 15m
labels:
severity: critical
annotations:
summary: "Model quality score dropped below 80%"
10.2. Incident Response Playbook
AI Incident Response:
1. DETECT
├── Alert fires (Prometheus/Grafana)
├── User reports bad responses
└── Quality score drops in Langfuse
2. TRIAGE (< 5 phút)
├── Check: Latency hoặc Quality issue?
├── Check: Toàn bộ hay chỉ 1 endpoint?
└── Check: Upstream API (OpenAI) down?
3. MITIGATE (< 15 phút)
├── Option A: Rollback model version
│ └── mlflow: transition previous to Production
├── Option B: Switch fallback model
│ └── gpt-4o fail → gpt-4o-mini
├── Option C: Enable cached responses
│ └── Return cached answers for common queries
└── Option D: Circuit breaker
└── Return default response, queue for retry
4. INVESTIGATE (< 2 hours)
├── Check Langfuse traces for bad responses
├── Check input drift (new types of queries?)
├── Check upstream: OpenAI rate limits? latency?
└── Check data: embeddings index corrupted?
5. FIX & PREVENT
├── Deploy fix
├── Add test case for this scenario
├── Update alerting thresholds
└── Write post-mortem
11. Complete MLOps Pipeline Example
11.1。架構概覽
┌──────────────────────────────────────────────────────────────┐
│ MLOps Pipeline Overview │
├──────────────────────────────────────────────────────────────┤
│ │
│ Developer │
│ │ │
│ ▼ │
│ ┌──────────┐ ┌───────────┐ ┌──────────────┐ │
│ │ Git Push │───▶│ GitHub │───▶│ Unit Tests │ │
│ │ (code + │ │ Actions │ │ Lint + Type │ │
│ │ DVC ptr) │ │ │ │ check │ │
│ └──────────┘ └───────────┘ └──────┬───────┘ │
│ │ pass │
│ ▼ │
│ ┌──────────────┐ │
│ │ Integration │ │
│ │ Tests │ │
│ └──────┬───────┘ │
│ │ pass │
│ ▼ │
│ ┌──────────┐ ┌──────────────┐ │
│ │ MLflow │◀────── log ────────│ Model Eval │ │
│ │ Tracking │ │ (quality │ │
│ └────┬─────┘ │ gates) │ │
│ │ └──────┬───────┘ │
│ │ register │ pass │
│ ▼ ▼ │
│ ┌──────────┐ ┌──────────────┐ │
│ │ Model │ │ Build Docker │ │
│ │ Registry │──── promote ──────▶│ + Deploy K8s │ │
│ │ (Stage) │ └──────┬───────┘ │
│ └──────────┘ │ │
│ ▼ │
│ ┌──────────────┐ │
│ ┌──────────┐ │ Production │ │
│ │Langfuse/ │◀────── traces ─────│ AI Agent │ │
│ │LangSmith │ └──────┬───────┘ │
│ └──────────┘ │ │
│ ▼ │
│ ┌──────────┐ ┌───────────┐ ┌──────────────┐ │
│ │ Alert- │◀───│Prometheus │◀───│ Metrics │ │
│ │ Manager │ │ + Grafana │ │ Endpoint │ │
│ └──────────┘ └───────────┘ └──────────────┘ │
│ │
└──────────────────────────────────────────────────────────────┘
11.2. Docker Compose cho MLOps Stack
# docker-compose.mlops.yml
services:
# --- AI Agent ---
ai-agent:
build: .
ports: ["8000:8000"]
environment:
- OPENAI_API_KEY=${OPENAI_API_KEY}
- MLFLOW_TRACKING_URI=http://mlflow:5000
- LANGFUSE_HOST=http://langfuse:3000
depends_on: [mlflow, langfuse, prometheus]
# --- MLflow ---
mlflow:
image: ghcr.io/mlflow/mlflow:v2.16.0
ports: ["5000:5000"]
command: >
毫升流伺服器
--backend-store-uri sqlite:///mlflow/mlflow.db
--default-artifact-root /mlflow/artifacts
--主機0.0.0.0
卷：
- mlflow-數據：/mlflow
# --- 朗芙絲 ---
朗福斯：
圖：langfuse/langfuse:2
埠：[“3000:3000”]
環境：
- DATABASE_URL=postgresql://postgres:postgres@langfuse-db:5432/langfuse
- NEXTAUTH_SECRET=我的秘密
- NEXTAUTH_URL=http://localhost:3000
取決於：[langfuse-db]
langfuse-db：
圖：postgres：16-alpine
環境：
POSTGRES_DB：langfuse
POSTGRES_PASSWORD：postgres
卷：
- langfuse-db-data:/var/lib/postgresql/data
# --- 監控 ---
普羅米修斯：
圖片：舞會/普羅米修斯：v2.53.0
埠：[“9090:9090”]
卷：
- ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml
格拉法納：
圖：grafana/grafana:11.1.0
埠：[“3001:3000”]
環境：
- GF_SECURITY_ADMIN_PASSWORD=管理員
卷：
-grafana-數據：/var/lib/grafana
卷：
mlflow-數據：
langfuse-db-資料：
grafana-數據：
# 啟動整個 MLOps 堆疊
docker compose -f docker-compose.mlops.yml up -d
# 訪問：
# - 人工智慧代理： http://localhost:8000
# - MLflow： http://localhost:5000
# - Langfuse： http://localhost:3000
# - 普羅米修斯： http://localhost:9090
# - 格拉法納： http://localhost:3001
11.3。工具總表
| 關注 | 工具 | 角色 |
|---|---|---|
| Experiment Tracking | MLflow / W&B | Log params, metrics, artifacts |
| Model Registry | MLflow Registry | Version, stage, promote models |
| 資料版本控制 | DVC | 適用於大數據的類似 Git 版本控制 |
| CI/CD | GitHub Actions | Automated test → eval → deploy |
| Monitoring (infra) | Prometheus + Grafana | Latency, throughput, errors |
| Monitoring (quality) | Custom + Langfuse | Response quality, drift |
| Observability (LLM) | Langfuse / LangSmith | Trace chains, debug agents |
| Alerting | Prometheus AlertManager | Notify on anomalies |
| Cost Tracking | Custom + Grafana | Token usage, $/request |
＃＃ 概括
✅ MLOps = DevOps + 資料、模型與模型品質管理 — 成熟度等級 0–4
✅ 實驗追蹤：MLflow（自架、開源）或 W&B（雲端 SaaS）— 記錄所有內容以實現可重現
✅ Model Registry: Versioning + stage transitions (None → Staging → Production → Archived)
✅ CI/CD cho AI: Unit tests → Integration → Model quality gates → Auto deploy — GitHub Actions workflow
✅ DVC：使用 git 的版本資料 + 模型文件，從 S3/GCS 推送/拉取
✅ Serving Patterns: Real-time (chat), Batch (bulk processing), Streaming (token-by-token UX)
✅ Monitoring: Prometheus/Grafana cho latency/throughput + custom metrics cho quality/drift/cost
✅ 可觀察性：Langfuse 或 LangSmith — 追蹤鏈中的每個步驟，調試生產問題
✅ Alerting: Rules cho latency spike, error rate, cost spike, quality degradation
✅ MLOps Stack：Docker Compose 運行 MLflow + Langfuse + Prometheus + Grafana 和 AI Agent
＃＃ 鍛煉
練習 1：MLflow 實驗追蹤（30 分鐘）
1.安裝MLflow並運行 流量使用者介面
2. 寫一個腳本來使用 3 組不同的參數（溫度、系統提示字元）來呼叫 OpenAI API（或模擬）
3. 將參數、指標（準確性、延遲）和工件（結果 JSON）記錄到 MLflow 中
4. 在 MLflow UI 中比較 3 次運行，選擇最佳運行
練習 2：CI/CD 管道（45 分鐘）
- 為 AI 代理專案建立 GitHub Actions 工作流程，分為 3 個階段：
- Lint + unit tests
- 模型評估（使用模擬 API 或小型測試資料集）
- 質量閘：若準確度 < 80% 或延遲 > 3 秒則失敗
- 推送程式碼並驗證管道是否正確運行 3.嘗試更改程式碼使品質門失敗，確認管道塊部署
練習 3：監控儀表板（45 分鐘）
- 將 Prometheus 指標新增至 FastAPI 應用程式（請求計數、延遲直方圖、令牌計數器）
- 使用 Docker Compose 設定 Prometheus + Grafana
- 使用以下指令建立 Grafana 儀表板：
- Request rate (req/s)
- Latency percentiles (p50, p95, p99)
- Token usage over time
- Error rate 4.發送模擬流量並即時驗證儀表板更新
練習 4：Langfuse 追蹤（30 分鐘）
1.使用Docker Compose在本地部署Langfuse
2. 偵測一個簡單的 RAG 管道（檢索 → 產生）： @觀察（）
3.提交5-10個查詢，打開Langfuse UI查看痕跡
4.找出哪個查詢延遲最高，分析哪些步驟慢