AI Engineer Master Course
Trang chủ / Module 2
🧱 Module 2 / 17 · 4-6 tuần

Kiến thức nền tảng bắt buộc

Python hiện đại (OOP, AsyncIO, Type Hint), Git/Docker/Linux, HTTP/REST/WebSocket, SQL nâng cao.

PythonDockerSQLHTTP
📌 Vì sao module này quan trọng

AI Engineer không viết ML từ đầu — họ viết hệ thống bọc quanh model. 95% code trong một dự án RAG/Agent thực tế là code hạ tầng bình thường: gọi API, xử lý bất đồng bộ, quản lý kết nối DB, retry, logging. Nếu nền Python/hạ tầng của bạn yếu, bạn sẽ luôn "demo được nhưng không chạy được ở production".

🎯 Mục tiêu học tập

  • Viết được Python bất đồng bộ (async/await, asyncio.gather, semaphore) để gọi nhiều LLM API song song
  • Container hoá một service Python bằng Docker, viết docker-compose cho stack nhiều service (API + DB + vector DB)
  • Thiết kế REST API chuẩn, hiểu khi nào dùng WebSocket/SSE cho streaming response
  • Viết SQL nâng cao trên PostgreSQL: window function, CTE, index, EXPLAIN ANALYZE

Python hiện đại cho AI Engineer

Không cần "học lại Python từ đầu" — tập trung vào phần khác biệt với code PHP truyền thống:

  • Type hints (def f(x: int) -> str) + Pydantic: bắt buộc vì hầu hết SDK AI (OpenAI, Anthropic, Google) dùng Pydantic để validate structured output.
  • AsyncIO: gọi LLM API là I/O-bound (chờ mạng), giống hệt việc gọi HTTP client trong Laravel Queue — nhưng Python cho phép chạy hàng chục request song song trong 1 process qua asyncio.gather() thay vì cần nhiều worker.
  • Context manager (async with) để quản lý kết nối DB/HTTP client đúng cách, tránh rò rỉ connection — lỗi rất phổ biến khi build AI service chạy lâu dài.
import asyncio
from anthropic import AsyncAnthropic

client = AsyncAnthropic()

async def ask(prompt: str) -> str:
    resp = await client.messages.create(
        model="claude-sonnet-5",
        max_tokens=512,
        messages=[{"role": "user", "content": prompt}],
    )
    return resp.content[0].text

async def main():
    prompts = ["Tóm tắt X", "Tóm tắt Y", "Tóm tắt Z"]
    results = await asyncio.gather(*(ask(p) for p in prompts))
    print(results)

asyncio.run(main())

Git, Docker, Linux — vận hành hằng ngày

Không có gì mới về bản chất so với PHP dev, nhưng AI project có vài thói quen riêng:

  • Docker Compose nhiều service: API service (FastAPI) + PostgreSQL/pgvector + Redis (cache) + đôi khi Qdrant. Học cách dùng healthcheckdepends_on để service khởi động đúng thứ tự.
  • Quản lý secret: API key của OpenAI/Anthropic/Google tuyệt đối không commit — dùng .env + python-dotenv, hoặc secret manager khi lên cloud.
  • Git cho notebook: nếu dùng Jupyter, cấu hình nbstripout để không commit output/ảnh nặng vào git history.

HTTP, REST, WebSocket, SSE cho AI app

Điểm khác biệt lớn nhất so với REST API truyền thống: streaming response. Khi LLM sinh văn bản, người dùng muốn thấy chữ xuất hiện dần (giống ChatGPT) thay vì chờ 10-20 giây rồi nhận toàn bộ. Hai kỹ thuật chính:

  • Server-Sent Events (SSE): đơn giản, một chiều (server → client), là lựa chọn mặc định cho streaming chat response, được toàn bộ SDK OpenAI/Claude/Gemini hỗ trợ sẵn.
  • WebSocket: cần khi có tương tác hai chiều liên tục (voice agent, giao diện cộng tác real-time nhiều người dùng cùng agent).
💡 Mẹo

FastAPI hỗ trợ StreamingResponse cho SSE rất tự nhiên — đây là framework Python được dùng phổ biến nhất cho AI backend năm 2026 nhờ type-safety (Pydantic) và hiệu năng async.

SQL nâng cao & PostgreSQL

PostgreSQL là database "mặc định" của hệ sinh thái AI hiện đại — không chỉ lưu dữ liệu quan hệ mà còn lưu vector (qua extension pgvector, xem Module 7). Cần vững:

  • Window functions, CTE đệ quy cho phân tích dữ liệu/log hội thoại
  • JSONB: lưu metadata linh hoạt (structured output, tool-call history) mà không cần migrate schema liên tục
  • EXPLAIN ANALYZE và index (B-tree, GIN, HNSW cho vector) — vì vector search chậm là nguyên nhân số 1 khiến RAG app "cảm giác chậm"

🏋️ Bài tập thực hành

FastAPI streaming service

Xây một service FastAPI với 1 endpoint /chat nhận prompt, gọi một LLM API bất kỳ (OpenAI/Claude/Gemini) và trả về response dạng SSE streaming. Container hoá bằng Docker, thêm PostgreSQL qua docker-compose để lưu lịch sử hội thoại dạng JSONB. Deploy local, test bằng curl với -N để xem stream.

📚 Tài nguyên học tập

✅ Tự đánh giá hoàn thành

  • Viết được hàm async gọi song song 5+ API request bằng asyncio.gather
  • Tạo được docker-compose.yml chạy API + PostgreSQL + Redis
  • Xây được 1 endpoint streaming (SSE) trả response dần từng token
  • Viết được 1 câu SQL dùng window function trên PostgreSQL