FastAPI: AI xizmatini veb-servisga aylantirish

Notebook'dagi model yoki skriptdagi LLM chaqiruvi — hali mahsulot emas. Uni boshqa dasturlar (veb-sayt, mobil ilova, boshqa servislar) ishlata olishi uchun HTTP API kerak. Python'da buning standart vositasi — FastAPI: u type hint'lar va Pydantic ustiga qurilgan (15-bob), async'ni to'liq qo'llaydi (17-bob) va avtomatik hujjat yaratadi. AI servislari uchun eng ko'p ishlatiladigan freymvork.

uv add "fastapi[standard]" anthropic pydantic-settings

Python FastAPI Tutorial (Part 1): Getting Started - Web App + REST API
FastAPI bilan birinchi veb-ilova va REST API — Corey Schafer'ning yangi seriyasining birinchi qismi.
YouTube'da ochish ↗

1. Birinchi endpoint#

# app/main.py
from fastapi import FastAPI
 
app = FastAPI(title="Sharh tahlili servisi")
 
 
@app.get("/health")
async def health() -> dict:
    return {"status": "ok"}
 
 
@app.get("/items/{item_id}")
async def get_item(item_id: int, verbose: bool = False) -> dict:
    return {"id": item_id, "verbose": verbose}

terminal

$ uv run fastapi dev app/main.py
INFO     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO     Started reloader process using WatchFiles

@app.get(...) — 14-bobdagi ro'yxatga oluvchi dekorator: funksiyani URL va HTTP metodiga bog'laydi.
item_id: int — yo'l parametri (path): /items/42. Type hint tufayli FastAPI uni intga aylantiradi; /items/abc → avtomatik 422 xato.
verbose: bool = False — so'rov parametri (query): /items/42?verbose=true.
Qaytarilgan lug'at avtomatik JSON'ga aylanadi.

Brauzerda http://127.0.0.1:8000/docs ni oching — FastAPI type hint'lardan yaratgan interaktiv hujjat (Swagger UI): har endpointni shu yerdan sinab ko'rish mumkin.

2. So'rov tanasi va javob — Pydantic modellari#

from typing import Literal
 
from pydantic import BaseModel, Field
 
 
class SummarizeRequest(BaseModel):
    text: str = Field(min_length=1, max_length=20_000)
    style: Literal["qisqa", "batafsil"] = "qisqa"
 
 
class SummarizeResponse(BaseModel):
    summary: str
    input_tokens: int
    output_tokens: int
    latency_ms: int
 
 
@app.post("/summarize", response_model=SummarizeResponse)
async def summarize(req: SummarizeRequest) -> SummarizeResponse:
    ...

Parametr Pydantic modeli bo'lsa — FastAPI uni JSON tanadan o'qiydi va tekshiradi. Noto'g'ri so'rov sizning kodingizgacha yetib kelmaydi:

POST /summarize  {"text": ""}
 
422 Unprocessable Entity
{"detail": [{"type": "string_too_short", "loc": ["body", "text"],
             "msg": "String should have at least 1 character", "input": "", ...}]}

response_model — javob ham sxema bo'yicha tekshiriladi va hujjatda ko'rinadi; ichki maydonlar (masalan, xarajat) tasodifan tashqariga chiqib ketmaydi.

3. Dependency injection — `Depends`#

Endpointlarga kerakli narsalarni (sozlamalar, LLM klienti, ma'lumotlar bazasi, joriy foydalanuvchi) parametr sifatida berish mexanizmi. 16-bobda o'rgangan g'oya: bog'liqlik ichkarida yaratilmasa, testda uni almashtirish oson.

from functools import cache
 
from fastapi import Depends
from pydantic_settings import BaseSettings
 
 
class Settings(BaseSettings):            # 15-bob: muhit o'zgaruvchilari va .env dan
    model: str = "claude-opus-5-5"
    service_api_keys: set[str] = {"dev-key"}
    max_tokens: int = 1024
 
 
@cache
def get_settings() -> Settings:          # bir marta yaratiladi
    return Settings()
 
 
@app.get("/config")
async def config(settings: Settings = Depends(get_settings)) -> dict:
    return {"model": settings.model}

Autentifikatsiya dependency sifatida#

from typing import Annotated
 
from fastapi import Header, HTTPException
 
 
def require_api_key(
    x_api_key: Annotated[str | None, Header()] = None,     # X-API-Key sarlavhasidan
    settings: Settings = Depends(get_settings),
) -> None:
    if x_api_key not in settings.service_api_keys:
        raise HTTPException(status_code=401, detail="API kaliti noto'g'ri")
 
 
@app.post("/summarize", dependencies=[Depends(require_api_key)])
async def summarize(...): ...

Dependency'lar zanjir bo'lishi mumkin (require_api_key o'zi get_settingsga bog'liq) — FastAPI ularni o'zi hal qiladi.

4. LLM klienti: bitta, ilova bilan birga#

LLM klienti (ulanishlar puli bilan) har so'rovda yaratilmasligi kerak (17-bob). Ilova ishga tushganda bir marta yaratib, app.stateda saqlaymiz. Va testda almashtirish uchun — Protocol (13, 15-boblar):

# app/llm.py
from collections.abc import AsyncIterator
from typing import Protocol
 
from anthropic import AsyncAnthropic
 
 
class LLM(Protocol):
    async def complete(self, system: str, prompt: str, max_tokens: int) -> tuple[str, int, int]: ...
    def stream(self, system: str, prompt: str, max_tokens: int) -> AsyncIterator[str]: ...
 
 
class ClaudeLLM:
    def __init__(self, model: str) -> None:
        self.client = AsyncAnthropic(max_retries=3, timeout=60)
        self.model = model
 
    async def complete(self, system: str, prompt: str, max_tokens: int) -> tuple[str, int, int]:
        msg = await self.client.messages.create(
            model=self.model, max_tokens=max_tokens, system=system,
            messages=[{"role": "user", "content": prompt}],
        )
        text = "".join(b.text for b in msg.content if b.type == "text")
        return text, msg.usage.input_tokens, msg.usage.output_tokens
 
    async def stream(self, system: str, prompt: str, max_tokens: int) -> AsyncIterator[str]:
        async with self.client.messages.stream(
            model=self.model, max_tokens=max_tokens, system=system,
            messages=[{"role": "user", "content": prompt}],
        ) as s:
            async for text in s.text_stream:
                yield text

stream — async generator (async def + yield): 7-bobdagi generatorlarning async versiyasi, async for bilan aylaniladi.

# app/main.py
from contextlib import asynccontextmanager
 
from fastapi import FastAPI, Request
 
from app.llm import LLM, ClaudeLLM
 
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.llm = ClaudeLLM(get_settings().model)    # ishga tushishda — bir marta
    yield
    # bu yerda — to'xtashda tozalash (masalan, ulanishlarni yopish)
 
 
app = FastAPI(title="Sharh tahlili servisi", lifespan=lifespan)
 
 
def get_llm(request: Request) -> LLM:
    return request.app.state.llm

lifespan — 14-bobdagi @contextmanagerning async versiyasi: yieldgacha — ishga tushish, keyin — to'xtash.

5. Asosiy endpoint#

import logging
import time
 
import anthropic
 
log = logging.getLogger("app")
SYSTEM = "Sen matnlarni o'zbek tilida xulosa qiluvchi yordamchisan."
 
 
def build_prompt(req: SummarizeRequest) -> str:
    length = "1-2 gap" if req.style == "qisqa" else "5-7 gap"
    return f"Quyidagi matnni {length}da xulosa qil:\n\n<matn>\n{req.text}\n</matn>"
 
 
@app.post("/summarize", response_model=SummarizeResponse, dependencies=[Depends(require_api_key)])
async def summarize(
    req: SummarizeRequest,
    llm: LLM = Depends(get_llm),
    settings: Settings = Depends(get_settings),
) -> SummarizeResponse:
    start = time.perf_counter()
    try:
        text, tin, tout = await llm.complete(SYSTEM, build_prompt(req), settings.max_tokens)
    except anthropic.RateLimitError:
        raise HTTPException(status_code=503, detail="Model band, birozdan keyin urinib ko'ring")
    except anthropic.APIError:
        log.exception("LLM xatosi")
        raise HTTPException(status_code=502, detail="Model xizmatida xato")
    return SummarizeResponse(
        summary=text, input_tokens=tin, output_tokens=tout,
        latency_ms=int((time.perf_counter() - start) * 1000),
    )

E'tibor bering:

endpoint async def va LLM chaqiruvi await — kutish paytida server boshqa so'rovlarga xizmat qiladi (17-bob). Async endpoint ichida bloklovchi chaqiruv (sinxron anthropic.Anthropic(), requests, time.sleep) qilmang — butun server to'xtaydi;
tashqi xizmat xatolari mijozga tushunarli HTTP kodiga aylantiriladi (502/503), ichki tafsilotlar (traceback, kalitlar) esa faqat logga yoziladi;
foydalanuvchi matni <matn> teglari ichida — 3-bobdagi prompt injection'dan qisman himoya.

6. Streaming: Server-Sent Events (SSE)#

LLM javobini token-token brauzerga uzatish — chat interfeyslarining standart usuli:

import json
from collections.abc import AsyncIterator
 
from fastapi.responses import StreamingResponse
 
 
@app.post("/summarize/stream", dependencies=[Depends(require_api_key)])
async def summarize_stream(
    req: SummarizeRequest,
    llm: LLM = Depends(get_llm),
    settings: Settings = Depends(get_settings),
) -> StreamingResponse:
    async def events() -> AsyncIterator[str]:
        try:
            async for chunk in llm.stream(SYSTEM, build_prompt(req), settings.max_tokens):
                yield f"data: {json.dumps({'text': chunk}, ensure_ascii=False)}\n\n"
            yield "data: [DONE]\n\n"
        except anthropic.APIError:
            log.exception("stream xatosi")
            yield f"data: {json.dumps({'error': 'model xatosi'})}\n\n"
 
    return StreamingResponse(
        events(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

SSE formati oddiy: har bir hodisa — data: ... qatori va bo'sh qator. X-Accel-Buffering: no — nginx orqasida ishlaganda javobni bufferlamasdan darhol uzatish uchun (aks holda foydalanuvchi butun javobni oxirida birdan ko'radi). Diqqat: oqim boshlanganidan keyin HTTP status kodini o'zgartirib bo'lmaydi — xatolar oqim ichida hodisa sifatida yuboriladi.

7. Testlash#

16-bobdagi yondashuv: haqiqiy LLM'siz, tez va deterministik. FastAPI'da dependency_overrides — bog'liqlikni testda almashtirishning rasmiy usuli:

# tests/test_api.py
import pytest
from fastapi.testclient import TestClient
 
from app.main import app, get_llm
 
 
class FakeLLM:
    def __init__(self) -> None:
        self.prompts: list[str] = []
 
    async def complete(self, system, prompt, max_tokens):
        self.prompts.append(prompt)
        return "Qisqa xulosa.", 120, 8
 
    async def stream(self, system, prompt, max_tokens):
        for part in ["Qisqa ", "xulosa."]:
            yield part
 
 
@pytest.fixture
def fake_llm():
    return FakeLLM()
 
 
@pytest.fixture
def client(fake_llm):
    app.dependency_overrides[get_llm] = lambda: fake_llm
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()
 
 
HEADERS = {"X-API-Key": "dev-key"}
 
 
def test_summarize_ok(client, fake_llm):
    r = client.post("/summarize", json={"text": "Uzun matn..."}, headers=HEADERS)
    assert r.status_code == 200
    assert r.json()["summary"] == "Qisqa xulosa."
    assert "1-2 gap" in fake_llm.prompts[0]
 
 
def test_requires_api_key(client):
    assert client.post("/summarize", json={"text": "x"}).status_code == 401
 
 
@pytest.mark.parametrize("payload", [{"text": ""}, {"text": "ok", "style": "o'rta"}, {}])
def test_validation(client, payload):
    assert client.post("/summarize", json=payload, headers=HEADERS).status_code == 422
 
 
def test_stream(client):
    with client.stream("POST", "/summarize/stream", json={"text": "x"}, headers=HEADERS) as r:
        body = "".join(r.iter_text())
    assert '"text": "Qisqa "' in body
    assert body.rstrip().endswith("[DONE]")

terminal

$ uv run pytest -q
.......                                                                  [100%]
7 passed in 0.74s

Bir soniyadan kam vaqtda butun API — validatsiya, autentifikatsiya, javob formati, streaming — tekshirildi, bitta ham haqiqiy so'rov yuborilmasdan.

8. Ishga tushirish va deploy#

Ishlab chiqish va production#

uv run fastapi dev app/main.py                      # ishlab chiqish: avtomatik qayta yuklash
uv run fastapi run app/main.py --workers 4          # production: bir nechta jarayon
# yoki to'g'ridan-to'g'ri:
uv run uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4

--workers — bir nechta process (17-bob): har biri o'z event loop'i bilan. Odatda yadrolar soni atrofida.

Docker#

FROM python:3.13-slim
COPY --from=ghcr.io/astral-sh/uv:latest /uv /bin/uv
 
WORKDIR /app
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev --no-install-project      # bog'liqliklar — alohida qatlam (kesh uchun)
 
COPY app ./app
RUN uv sync --frozen --no-dev
 
ENV PATH="/app/.venv/bin:$PATH"
EXPOSE 8000
CMD ["fastapi", "run", "app/main.py", "--port", "8000", "--workers", "2"]

docker build -t summarizer .
docker run -p 8000:8000 -e ANTHROPIC_API_KEY=sk-ant-... -e SERVICE_API_KEYS='["prod-key"]' summarizer

uv.lock + --frozen — konteynerda aynan sinalgan versiyalar (9-bob). Kalitlar image ichiga emas, ishga tushirishda muhit o'zgaruvchisi sifatida beriladi.

Production ro'yxati#

Masala	Yechim
HTTPS, statik fayllar	Oldida reverse proxy (nginx, Caddy) yoki bulut load balancer
Sirlar	Muhit o'zgaruvchilari / secret manager; logga yozilmasin
Rate limiting	Foydalanuvchi/kalit bo'yicha limit (Redis) — LLM xarajatini nazorat qilish
Xarajat nazorati	Har so'rov token va narxini logga/bazaga yozish, kunlik limitlar
Timeout'lar	LLM klienti, proxy va server timeout'lari mos bo'lsin (streaming uzun davom etadi)
Kuzatuv	Strukturali loglar (14-bob `span`), metrikalar: kechikish p95, xatolar ulushi
CORS	Brauzerdan chaqirilsa — `CORSMiddleware` faqat kerakli domenlar bilan
Health check	`/health` — orkestrator (Docker, Kubernetes) uchun

9. Django, Flask yoki FastAPI?#

	FastAPI	Django	Flask
Asosiy kuchi	API, async, type hint'lar, avtomatik hujjat	To'liq veb-ilova: ORM, admin, auth, migratsiyalar	Minimal, moslashuvchan
AI servislari	Eng ko'p tanlanadi	Katta mahsulot ichida AI funksiya (masalan, aidoc.uz backend'i)	Kichik servislar, eski loyihalar
Async	To'liq	Qisman (yaxshilanmoqda)	Cheklangan

Qoida: alohida AI mikroservisi — FastAPI; foydalanuvchilar, admin paneli, murakkab ma'lumotlar bazasi bo'lgan to'liq mahsulot — Django (+ DRF).

Nima noto'g'ri ketishi mumkin#

Xato	Oqibat	Tuzatish
`async def` ichida sinxron LLM/HTTP chaqiruvi	Butun server bloklanadi	Async klientlar yoki oddiy `def` endpoint (FastAPI uni threadda ishlatadi)
Har so'rovda yangi LLM klienti	Sekin, ulanishlar tugaydi	`lifespan`da bitta klient
Kirish uzunligi cheklanmagan	Katta xarajat, DoS	`Field(max_length=...)`
Ichki xato tafsilotlari javobda	Ma'lumot sizib chiqishi	Log'ga — to'liq; mijozga — umumiy xabar
Testlarda haqiqiy LLM	Sekin, qimmat, beqaror	`dependency_overrides` + fake
nginx streaming'ni bufferlaydi	Javob oxirida birdan keladi	`X-Accel-Buffering: no` yoki `proxy_buffering off`
Kalit Docker image'da	Sizib chiqadi	Ishga tushirishda muhit o'zgaruvchisi
`--workers` + xotiradagi holat	Har worker'da boshqa holat	Holatni Redis/bazada saqlang

Amaliy topshiriq

1. Tasniflash endpointi

POST /classify endpointini qo'shing: sharh matnini qabul qilib, 23-bobdagi Review (sentiment, topics, summary) sxemasida javob qaytarsin (messages.parse). LLM protokoliga yangi metod qo'shing va fake bilan kamida 3 ta test yozing.

Maslahatni ko'rsatish

23-bobdagi Review modeli + messages.parse; FakeLLM'ga parse natijasini qaytaruvchi metod qo'shing.

Amaliy topshiriq

2. Rate limit dependency

Har bir API kaliti uchun daqiqasiga 10 ta so'rov limitini qo'yuvchi dependency yozing (avval xotirada, keyin Redis bilan). Limitdan oshganda 429 va Retry-After sarlavhasi qaytsin. Test yozing.

Maslahatni ko'rsatish

Lug'at {kalit: deque(maxlen=N)} yoki Redis INCR + EXPIRE; limitdan oshsa HTTPException(429, headers={'Retry-After': ...}).

Amaliy topshiriq

3. Batch endpoint

POST /summarize/batch — 50 tagacha matn ro'yxatini qabul qilib, ularni bir vaqtda ko'pi bilan 5 tadan parallel xulosa qilsin (17-bob); bitta matn xatosi butun javobni buzmasin — har element uchun ok yoki error qaytarsin.

Maslahatni ko'rsatish

async def; asyncio.Semaphore(5); asyncio.gather(..., return_exceptions=True); har element uchun ok/error.

Amaliy topshiriq

4. To'liq deploy

Servisni Docker image'ga yig'ing, mahalliy ishga tushiring va curl bilan /health, /summarize va /summarize/stream ni sinang (curl -N — streaming'ni bufferlamasdan ko'rsatadi). Keyin GitHub Actions'ga image'ni yig'ish va testlarni ishga tushirish qadamini qo'shing (16-bob mashqining davomi).

Yodda saqlang

01FastAPI: dekoratorlar bilan routing, type hint'lar bilan parametrlar, /docs — avtomatik hujjat.
02So'rov va javob — Pydantic modellari: validatsiya avtomatik, noto'g'ri so'rov → 422.
03Depends — sozlamalar, klientlar, autentifikatsiya; testda dependency_overrides bilan almashtiriladi.
04LLM klienti lifespan'da bir marta yaratiladi; endpointlar async va bloklovchi chaqiruvsiz.
05Streaming: StreamingResponse + SSE (data: ...\n\n); nginx orqasida X-Accel-Buffering: no.
06Kirishni cheklang, xatolarni tushunarli HTTP kodlariga aylantiring, tafsilotlarni faqat logga yozing.
07Deploy: uv.lock + Docker, sirlar muhit o'zgaruvchisida, rate limit va xarajat nazorati.

FastAPI: AI xizmatini veb-servisga aylantirish

1. Birinchi endpoint#

2. So'rov tanasi va javob — Pydantic modellari#

3. Dependency injection — `Depends`#

Autentifikatsiya dependency sifatida#

4. LLM klienti: bitta, ilova bilan birga#

5. Asosiy endpoint#

6. Streaming: Server-Sent Events (SSE)#

7. Testlash#

8. Ishga tushirish va deploy#

Ishlab chiqish va production#

Docker#

Production ro'yxati#

9. Django, Flask yoki FastAPI?#

Nima noto'g'ri ketishi mumkin#

1. Tasniflash endpointi

2. Rate limit dependency

3. Batch endpoint

4. To'liq deploy

Test va amaliy masalalar

Bob bo'yicha test

Amaliy masalalar

01SSE formatlovchi

02Rate limiter

03So'rov validatsiyasi (Pydantic)

1. Birinchi endpoint#

2. So'rov tanasi va javob — Pydantic modellari#

3. Dependency injection — Depends#

Autentifikatsiya dependency sifatida#

4. LLM klienti: bitta, ilova bilan birga#

5. Asosiy endpoint#

6. Streaming: Server-Sent Events (SSE)#

7. Testlash#

8. Ishga tushirish va deploy#

Ishlab chiqish va production#

Docker#

Production ro'yxati#

9. Django, Flask yoki FastAPI?#

Nima noto'g'ri ketishi mumkin#

1. Tasniflash endpointi

2. Rate limit dependency

3. Batch endpoint

4. To'liq deploy

Test va amaliy masalalar

Bob bo'yicha test

Amaliy masalalar

3. Dependency injection — `Depends`#