Dan 11/28: Validacija sa Pydantic i provider-specific modeli
Vreme potrebno: 90 – 180 minuta
Gde je Učitelj Vasa danas?
Učitelj Vasa ima funkcionalan web API sa strukturiranim zahtevima koji prepoznaju različite tipove pitanja i inteligentno rutiraju ka najboljem AI provideru. Ali trenutno validacija podataka je ručna – proveravamo da li postoji polje “pitanje”, da li je prazno, itd. Takođe, svaki provider (OpenAI, Gemini) ima svoje specifičnosti koje trenutno ignorišemo. Danas menjamo to sa Pydantic bibliotekom!
Cilj današnje lekcije
Danas ćeš naučiti kako da koristiš Pydantic za automatsku validaciju svih podataka koji ulaze i izlaze iz API-ja, kreiraš type-safe modele koji garantuju ispravnost podataka, i implementiraš provider-specifične modele koji omogućavaju fine-grained kontrolu nad svakim AI servisom. Nakon ove lekcije, Učitelj Vasa će imati profesionalnu validaciju koja automatski proverava i formatira sve podatke!
Predznanja
- Funkcionalan FastAPI server sa routing sistemom (Dan 8-10)
- Razumevanje dictionary i class struktura u Python-u
- Osnovno razumevanje type hints u Python-u
- StructuredRequest sistem iz Dana 10
- Razumevanje try/except blokova
Glavni sadržaj
Zašto je Pydantic važan za API razvoj?
Pre nego što krenemo sa implementacijom, razumimo zašto nam treba Pydantic.
📚 NIVO 1 – Osnovno objašnjenje
Pydantic je kao strogi kontrolor na ulazu u klub – proverava da li imaš sve što treba (ime, godine, kartu) i da li su podaci u ispravnom formatu. Bez njega, morao bi ručno da proveravaš:
- Da li postoji svako polje
- Da li je tačnog tipa (broj, tekst…)
- Da li je u dozvoljenom opsegu
- Da li je u ispravnom formatu
Pydantic sve ovo radi automatski! Samo mu kažeš kakvi podaci treba da budu, a on proverava i javlja ako nešto nije u redu.
💡 PRO TIP: Zamislite Pydantic kao aerodromski skener za vaše podatke: svaki kofer (JSON) prolazi kroz rendgen pre nego što ga pustite u avion (business-logiku). Ako skener nađe “nedozvoljen predmet” (pogrešan tip ili prazno polje), odmah diže uzbunu — čuvate aplikaciju od “opasnog tereta” pre nego što poleti.
🚀 NIVO 2 – Dublje razumevanje
Pydantic koristi Python type hints za:
- Runtime validation – proverava podatke kada stignu
- Serialization – pretvara objekte u JSON
- Documentation – automatski generiše OpenAPI shemu
- IDE support – autocomplete i type checking
- Error messages – jasne poruke šta nije u redu
FastAPI i Pydantic rade zajedno – FastAPI automatski validira request/response koristeći Pydantic modele.
🔍 UVID: Pydantic implementira moćan sistemski princip poznat kao “Parse, Don’t Validate”. Umesto da stalno proveravaš sirove podatke (npr. dictionary) u različitim delovima koda, ti ih na samom ulazu u sistem parsiraš u strogo definisan Pydantic model. Ako parsiranje uspe, imaš garanciju da je objekat sa kojim radiš ispravan u ostatku aplikacije. Ovo premešta sve provere na “granicu” sistema i drastično smanjuje broj grešaka, čineći kod čistijim i predvidljivijim. Tretiraš validaciju kao transformaciju iz nestrukturiranog u strukturirani, pouzdan podatak.
🎈 ZABAVNA ČINJENICA: Ime “Pydantic” je igra reči – kombinacija “Python” i “pedantic” (što znači biti vrlo precizan oko detalja). Samuel Colvin, kreator biblioteke, hteo je ime koje sugeriše “Python biblioteku koja je pedantna oko tipova”!
Instalacija i osnovni Pydantic modeli
Počnimo sa instalacijom i prvim modelima.
📚 NIVO 1 – Osnovno objašnjenje
Pydantic model je kao formular sa pravilima:
- Svako polje ima ime i tip
- Može imati default vrednost
- Može imati dodatna pravila (minimum, maksimum…)
Kada napraviš model, Pydantic automatski:
- Proverava da li su svi podaci ispravni
- Konvertuje tipove ako može (npr. “123” u 123)
- Vraća grešku sa objašnjenjem ako nešto nije u redu
💡 PRO TIP: Ako treba brzo da “olabavite” pravila — recimo, tokom razvoja — podesite model_config = {'extra': 'allow'} u klasi. Time govorite skeneru: “Pusti i koferčiće bez deklaracije; samo ih stavi u lost-and-found polje model_extra“. Korisno je za prototipe, ali u produkciji vratite na 'forbid' da sprečite iznenađenja!
🚀 NIVO 2 – Dublje razumevanje
Pydantic modeli nasleđuju BaseModel klasu i koriste:
- Type annotations za definisanje tipova
- Field() funkciju za dodatna ograničenja
- Validators za custom provere
- Config klasu za podešavanja
Pydantic podržava sve Python tipove plus:
EmailStr,HttpUrl,UUIDconint(),constr()sa ograničenjimaUnion,Optional,List,Dict- Custom tipove i regex validaciju
📊 DIJAGRAM: Anatomija Pydantic Modela
[Zahtev u JSON formatu] → [Pydantic Model: MyRequest(BaseModel)]
"pitanje": "..." |
"broj": 123 ├─ pitanje: str = Field(..., min_length=1) <-- Pravilo (Rule)
| ↑ ↑
└─ Tip (Type) Ograničenje (Constraint)
|
├─ broj: int = Field(..., gt=0)
|
└─ @validator('pitanje') <-- Custom Logika
│
↓
[Validiran Python Objekat]
my_request.pitanje (tipa 'str')
my_request.broj (tipa 'int')
🎯 ALTERNATIVNO REŠENJE: Umesto zasebnih SimpleQuestionRequest i StructuredQuestionRequest modela, možete definisati jedan model sa opcionalnim poljima i koristiti prilagođeni @root_validator da proveri “mode” (simple vs. structured). Prednost je manje klasa; mana je složenija validacija:
class QuestionRequest(BaseModel):
pitanje: str = Field(..., min_length=1, max_length=2000)
context: Optional[RequestContext]
preferences: Optional[AIPreferences]
@root_validator
def check_mode(cls, values):
if values.get("context") is None and values.get("preferences") is None:
values["mode"] = "simple"
else:
values["mode"] = "structured"
return values
Instaliraj Pydantic (već je instaliran sa FastAPI):
# Pydantic dolazi sa FastAPI, ali možeš eksplicitno instalirati najnoviju verziju
pip install pydantic[email]
Ažuriraj requirements.txt:
python-dotenv
openai
google-generativeai
fastapi
uvicorn[standard]
pydantic[email]
Kreiraj src/web_api/models/validation.py:
"""
Pydantic modeli za validaciju podataka u Učitelju Vasi
Osigurava type safety i automatsku validaciju
"""
from pydantic import BaseModel, Field, field_validator, model_validator
from typing import Optional, Dict, Any, List, Literal, Union
from datetime import datetime
from enum import Enum
from web_api.models.request_types import RequestType
# Osnovni modeli za request/response
class BaseRequest(BaseModel):
"""Bazni model za sve zahteve."""
class Config:
# Dozvoli enum values umesto samo enum objekata
use_enum_values = True
# Prikaži primer u dokumentaciji
json_schema_extra = {
"example": {
"pitanje": "Kako da sortiram listu u Python-u?"
}
}
class SimpleQuestionRequest(BaseRequest):
"""Jednostavan zahtev sa samo pitanjem."""
pitanje: str = Field(
..., # ... znači obavezno polje
min_length=1,
max_length=2000,
description="Pitanje za AI asistenta"
)
@field_validator('pitanje')
@classmethod
def pitanje_not_empty(cls, v):
"""Proveri da pitanje nije samo whitespace."""
if not v.strip():
raise ValueError('Pitanje ne može biti prazno')
return v.strip()
class ProgrammingLanguage(str, Enum):
"""Podržani programski jezici."""
PYTHON = "python"
JAVASCRIPT = "javascript"
JAVA = "java"
CPP = "cpp"
CSHARP = "csharp"
GO = "go"
RUST = "rust"
OTHER = "other"
class UserLevel(str, Enum):
"""Nivoi znanja korisnika."""
BEGINNER = "beginner"
INTERMEDIATE = "intermediate"
ADVANCED = "advanced"
EXPERT = "expert"
class RequestContext(BaseModel):
"""Kontekst zahteva sa dodatnim informacijama."""
programming_language: Optional[ProgrammingLanguage] = Field(
None,
description="Programski jezik za koji se traži pomoć"
)
error_message: Optional[str] = Field(
None,
max_length=1000,
description="Poruka o grešci ako postoji"
)
code_snippet: Optional[str] = Field(
None,
max_length=5000,
description="Deo koda relevantan za pitanje"
)
user_level: UserLevel = Field(
UserLevel.INTERMEDIATE,
description="Nivo znanja korisnika"
)
previous_attempts: List[str] = Field(
default_factory=list,
max_items=5,
description="Prethodni pokušaji rešavanja"
)
@field_validator('code_snippet')
@classmethod
def clean_code_snippet(cls, v):
"""Očisti code snippet od ekstra whitespace."""
if v:
return v.strip()
return v
class AIPreferences(BaseModel):
"""Preference za AI odgovor."""
temperature: float = Field(
0.7,
ge=0.0, # greater or equal
le=2.0, # less or equal
description="Kreativnost odgovora (0=deterministički, 2=vrlo kreativan)"
)
max_tokens: int = Field(
150,
ge=50,
le=2000,
description="Maksimalan broj tokena u odgovoru"
)
response_style: Literal["concise", "detailed", "tutorial"] = Field(
"concise",
description="Stil odgovora"
)
include_examples: bool = Field(
True,
description="Da li uključiti primere koda"
)
language: Literal["sr", "en"] = Field(
"sr",
description="Jezik odgovora"
)
class StructuredQuestionRequest(BaseRequest):
"""Strukturiran zahtev sa svim opcijama."""
pitanje: str = Field(
...,
min_length=1,
max_length=2000,
description="Glavno pitanje"
)
tip: Optional[RequestType] = Field(
None,
description="Eksplicitni tip zahteva"
)
context: Optional[RequestContext] = Field(
None,
description="Dodatni kontekst"
)
preferences: Optional[AIPreferences] = Field(
None,
description="AI preference"
)
force_provider: Optional[Literal["openai", "gemini"]] = Field(
None,
description="Forsiraj specifičan provider"
)
session_id: Optional[str] = Field(
None,
pattern="^[a-zA-Z0-9_-]+$", # Promenjen regex u pattern
max_length=50,
description="ID sesije za praćenje konverzacije"
)
@field_validator('pitanje')
@classmethod
def clean_question(cls, v):
"""Očisti pitanje."""
return v.strip()
@model_validator(mode='after')
def validate_context_with_type(self):
"""Proveri da li kontekst odgovara tipu."""
if self.tip == RequestType.CODE_DEBUG and self.context:
if not self.context.code_snippet and not self.context.error_message:
raise ValueError(
'Debug tip zahteva kod ili error poruku u kontekstu'
)
return self
class Config:
json_schema_extra = { # Promenjeno iz schema_extra
"example": {
"pitanje": "Zašto mi ova funkcija vraća None?",
"tip": "debug",
"context": {
"programming_language": "python",
"code_snippet": "def suma(a, b):\n print(a + b)",
"user_level": "beginner"
},
"preferences": {
"temperature": 0.3,
"response_style": "detailed",
"include_examples": True
}
}
}
# Response modeli
class ProviderInfo(BaseModel):
"""Informacije o korišćenom provideru."""
selected: str = Field(..., description="Izabrani provider")
reason: str = Field(..., description="Razlog izbora")
strategy: str = Field(..., description="Korišćena strategija")
available_providers: List[str] = Field(
...,
description="Svi dostupni provideri"
)
class OptimizationInfo(BaseModel):
"""Informacije o optimizaciji."""
temperature: float = Field(..., description="Korišćena temperatura")
max_tokens: int = Field(..., description="Max tokena")
adjusted_for_type: bool = Field(
...,
description="Da li su parametri prilagođeni tipu"
)
class QuestionResponse(BaseModel):
"""Standardni odgovor na pitanje."""
pitanje: str = Field(..., description="Originalno pitanje")
odgovor: str = Field(..., description="AI odgovor")
tip_zahteva: str = Field(..., description="Prepoznat tip zahteva")
provider: ProviderInfo = Field(..., description="Provider info")
optimization: Optional[OptimizationInfo] = Field(
None,
description="Optimizacija info"
)
context_used: bool = Field(
False,
description="Da li je korišćen kontekst"
)
session_id: Optional[str] = Field(
None,
description="ID sesije"
)
response_time_ms: Optional[int] = Field(
None,
description="Vreme odgovora u milisekundama"
)
class Config:
json_schema_extra = { # Promenjeno iz schema_extra
"example": {
"pitanje": "Kako da sortiram listu?",
"odgovor": "U Python-u možeš sortirati listu na nekoliko načina...",
"tip_zahteva": "explain",
"provider": {
"selected": "gemini",
"reason": "Best for explain requests",
"strategy": "static",
"available_providers": ["openai", "gemini"]
},
"optimization": {
"temperature": 0.6,
"max_tokens": 400,
"adjusted_for_type": True
},
"context_used": False,
"response_time_ms": 1250
}
}
class ErrorResponse(BaseModel):
"""Standardni error response."""
error: str = Field(..., description="Tip greške")
detail: str = Field(..., description="Detaljan opis")
suggestion: Optional[str] = Field(
None,
description="Predlog za rešavanje"
)
error_code: Optional[str] = Field(
None,
description="Interni kod greške"
)
class Config:
json_schema_extra = { # Promenjeno iz schema_extra
"example": {
"error": "validation_error",
"detail": "Pitanje je predugačko (max 2000 karaktera)",
"suggestion": "Skrati pitanje ili ga podeli na više manjih",
"error_code": "VAL001"
}
}
# Provider-specifični modeli
class OpenAISpecificRequest(BaseModel):
"""OpenAI specifične opcije."""
model: Literal["gpt-3.5-turbo", "gpt-4", "gpt-4-turbo"] = Field(
"gpt-3.5-turbo",
description="OpenAI model"
)
system_prompt: Optional[str] = Field(
None,
max_length=1000,
description="Custom system prompt"
)
presence_penalty: float = Field(
0.0,
ge=-2.0,
le=2.0,
description="Penalizuje ponavljanje tema"
)
frequency_penalty: float = Field(
0.0,
ge=-2.0,
le=2.0,
description="Penalizuje ponavljanje reči"
)
top_p: float = Field(
1.0,
ge=0.0,
le=1.0,
description="Nucleus sampling"
)
stop_sequences: List[str] = Field(
default_factory=list,
max_items=4,
description="Sekvence za prekid generisanja"
)
class GeminiSpecificRequest(BaseModel):
"""Gemini specifične opcije."""
model: Literal["gemini-pro", "gemini-pro-vision"] = Field(
"gemini-pro",
description="Gemini model"
)
safety_settings: Dict[str, str] = Field(
default_factory=dict,
description="Safety filter settings"
)
candidate_count: int = Field(
1,
ge=1,
le=8,
description="Broj kandidat odgovora"
)
stop_sequences: List[str] = Field(
default_factory=list,
max_items=5,
description="Stop sekvence"
)
class ProviderSpecificRequest(BaseModel):
"""Wrapper za provider-specifične opcije."""
provider: Literal["openai", "gemini"] = Field(
...,
description="Target provider"
)
options: Union[OpenAISpecificRequest, GeminiSpecificRequest] = Field(
...,
description="Provider-specific options"
)
@model_validator(mode='after')
def validate_options_match_provider(self):
"""Proveri da opcije odgovaraju provideru."""
if self.provider == 'openai' and not isinstance(self.options, OpenAISpecificRequest):
raise ValueError('OpenAI provider zahteva OpenAISpecificRequest opcije')
if self.provider == 'gemini' and not isinstance(self.options, GeminiSpecificRequest):
raise ValueError('Gemini provider zahteva GeminiSpecificRequest opcije')
return self
# Validation utility funkcije
def validate_api_key(api_key: str, provider: str) -> bool:
"""
Validira format API ključa za provider.
Args:
api_key: API ključ za validaciju
provider: Ime providera
Returns:
True ako je valjan format
"""
if provider == "openai":
# OpenAI ključevi počinju sa 'sk-'
return api_key.startswith("sk-") and len(api_key) > 20
elif provider == "gemini":
# Gemini ključevi su obično 39 karaktera
return len(api_key) == 39
return False
def sanitize_code_snippet(code: str) -> str:
"""
Sanitizuje kod snippet za sigurno procesiranje.
Args:
code: Sirovi kod
Returns:
Očišćen kod
"""
# Ukloni potencijalno opasne stringove
dangerous_patterns = [
"__import__",
"eval(",
"exec(",
"compile(",
"globals(",
"locals("
]
sanitized = code
for pattern in dangerous_patterns:
sanitized = sanitized.replace(pattern, f"[REMOVED: {pattern}]")
return sanitized.strip()
# Test funkcionalnost
if __name__ == "__main__":
print("🧪 Test Pydantic modela")
print("=" * 50)
# Test jednostavnog zahteva
print("\n1. Test jednostavnog zahteva:")
try:
simple = SimpleQuestionRequest(pitanje=" Kako da sortiram listu? ")
print(f"✅ Validno: {simple.pitanje}")
except Exception as e:
print(f"❌ Greška: {e}")
# Test praznog pitanja
print("\n2. Test praznog pitanja:")
try:
empty = SimpleQuestionRequest(pitanje=" ")
print(f"✅ Validno: {empty.pitanje}")
except Exception as e:
print(f"❌ Greška: {e}")
# Test strukturiranog zahteva
print("\n3. Test strukturiranog zahteva:")
structured_data = {
"pitanje": "Debug ovaj kod",
"tip": "debug",
"context": {
"programming_language": "python",
"code_snippet": "def test():\n pass",
"user_level": "beginner"
},
"preferences": {
"temperature": 0.3,
"response_style": "detailed"
}
}
try:
structured = StructuredQuestionRequest(**structured_data)
print(f"✅ Validno:")
print(f" Tip: {structured.tip}")
print(f" Jezik: {structured.context.programming_language}")
print(f" Stil: {structured.preferences.response_style}")
except Exception as e:
print(f"❌ Greška: {e}")
# Test JSON Schema generisanja
print("\n4. JSON Schema:")
print(f"Schema ima {len(StructuredQuestionRequest.model_json_schema()['properties'])} polja")
# Test provider-specific
print("\n5. Test provider-specific modela:")
openai_req = ProviderSpecificRequest(
provider="openai",
options=OpenAISpecificRequest(
model="gpt-4",
temperature=0.5,
top_p=0.9
)
)
print(f"✅ OpenAI request: model={openai_req.options.model}")
💡 PRO TIP: Koristi Pydantic Field() sa regex parametrom za validaciju formata. Na primer, za session ID: Field(regex="^sess_[a-zA-Z0-9]{16}$") garantuje format “sess_” + 16 alfanumeričkih karaktera. Ovo je moćnije od običnog string tipa!
⚡ SAVET ZA OPTIMIZACIJU: Zašto koristiti default_factory=list umesto default=[]? Korišćenje default=[] stvara jedan te isti objekat liste kada se klasa definiše. Svi Pydantic modeli bi delili tu jednu istu listu, što bi dovelo do neočekivanih bugova gde jedna instanca menja listu drugoj. default_factory=list osigurava da se za svaku novu instancu modela kreira nova, prazna lista pozivanjem list(). Ovo je malo sporije, ali esencijalno za ispravnost podataka i sprečavanje sistemskih grešaka.
🎈 ZABAVNA ČINJENICA: Pydantic internu validaciju obavlja pomoću ujson ili rapidjson kada su dostupni, pa su “skeneri” zapravo turbo-brzi!
Integracija Pydantic modela u FastAPI
Sada ćemo integrisati ove modele u postojeće endpoint-e.
📚 NIVO 1 – Osnovno objašnjenje
FastAPI automatski koristi Pydantic modele za:
- Validaciju ulaza – proverava podatke pre nego što stignu do tvoje funkcije
- Serijalizaciju izlaza – pretvara tvoje objekte u JSON
- Dokumentaciju – generiše primere i opise u Swagger UI
Sve što trebaš je da umesto Dict koristiš Pydantic model u funkciji!
💡 PRO TIP: FastAPI automatski generiše OpenAPI šemu iz Pydantic modela. To znači da svaki put kada promenite model (polje dodato, opis izmenjen), Swagger UI se “samodokumentuje”. Razmislite o tome kao o Google Docs-u koji se dopunjuje dok kucate kod — nikad više zastarele API specifikacije!
🚀 NIVO 2 – Dublje razumevanje
FastAPI integracija omogućava:
- Automatske 422 error response za nevalidne podatke
- Type hints u IDE sa autocomplete
- Automatsku konverziju tipova
- Response model validation
- Dependency injection sa Pydantic modelima
🌐 GOOGLE CONNECTION: Ovakav način definisanja modela podataka je izuzetno koristan kada vaš servis treba da komunicira sa drugim Google Cloud servisima. Na primer, kada Učitelj Vasa postane dovoljno napredan da šalje logove u Google Cloud Logging (deo Operations Suite-a), možete lako serijalizovati QuestionResponse ili ErrorResponse modele u JSON format. Ovo omogućava strukturirano logovanje, gde svaki log zapis ima ista polja. To kasnije drastično olakšava pretragu, analizu i kreiranje dashboarda u Google Cloud-u, jer možete da filtrirate logove po provider.selected ili tip_zahteva umesto da pretražujete običan tekst.
📊 DIJAGRAM: Tok validacije u FastAPI
[HTTP Request]
|
v
[FastAPI Router]
|
v
[Pydantic Validation] <--- Automatska validacija
|
OK? |
/ \
YES NO
| |
v v
[Handler] [422 Error]
| |
v v
[Response] [Error Details]
🎈 ZABAVNA ČINJENICA: FastAPI + Pydantic kombinacija je toliko popularna da je inspirisala slične framework-e u drugim jezicima. Rust ima Poem, Go ima Fiber, TypeScript ima tsoa – svi pokušavaju da repliciraju magiju automatske validacije i dokumentacije!
🎯 ALTERNATIVNO REŠENJE: Ako želite personalizovane poruke na srpskom za sve 422 greške, dodajte “globalni” exception handler:
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
return JSONResponse(
status_code=422,
content={
"error": "nevalidan_zahtev",
"detail": exc.errors(),
"poruka": "Proveri format i probaj ponovo 😉"
}
)
Ažuriraj src/web_api/app.py sa Pydantic modelima:
# Dodaj nove import-e na početak
from web_api.models.validation import (
SimpleQuestionRequest,
StructuredQuestionRequest,
QuestionResponse,
ErrorResponse,
ProviderInfo,
OptimizationInfo
)
from fastapi import Query
from fastapi.responses import JSONResponse
import time
# Zameni postojeći /pitaj endpoint sa ovim
@app.post("/pitaj",
summary="Postavi pitanje sa naprednom validacijom",
description="""
Endpoint sa Pydantic validacijom koji podržava:
- Jednostavne zahteve (samo pitanje)
- Strukturirane zahteve sa kontekstom
- Provider-specifične opcije
- Automatsku validaciju svih podataka
""",
response_model=QuestionResponse,
responses={
422: {
"description": "Validation Error",
"model": ErrorResponse
},
500: {
"description": "Internal Server Error",
"model": ErrorResponse
}
},
tags=["Questions"]
)
async def pitaj_vasu(
request: Union[SimpleQuestionRequest, StructuredQuestionRequest],
analyze_request: bool = Query(
True,
description="Da li analizirati tip zahteva"
)
):
"""
Postavlja pitanje Učitelju Vasi sa Pydantic validacijom.
Request body se automatski validira prema modelu.
Podržava jednostavne i strukturirane zahteve.
"""
start_time = time.time()
# Konvertuj u strukturiran zahtev ako je jednostavan
if isinstance(request, SimpleQuestionRequest):
structured_request = RequestAnalyzer.create_structured_request(
request.pitanje
)
else:
# Već je strukturiran
if analyze_request and not request.tip:
# Analiziraj tip ako nije eksplicitno postavljen
analyzed_type = RequestAnalyzer.analyze(request.pitanje)
request_type = analyzed_type
else:
request_type = request.tip or RequestType.CHAT
# Kreiraj interni structured request
from web_api.models.request_types import (
StructuredRequest as InternalStructuredRequest,
RequestContext as InternalContext
)
# Konvertuj Pydantic context u interni
internal_context = None
if request.context:
internal_context = InternalContext(
programming_language=request.context.programming_language,
error_message=request.context.error_message,
code_snippet=request.context.code_snippet,
user_level=request.context.user_level,
previous_attempts=request.context.previous_attempts
)
structured_request = InternalStructuredRequest(
content=request.pitanje,
request_type=request_type,
context=internal_context,
preferences=request.preferences.dict() if request.preferences else {}
)
# Proveri AI servis
if not ai_service:
return JSONResponse(
status_code=503,
content=ErrorResponse(
error="service_unavailable",
detail="AI servis trenutno nije dostupan",
suggestion="Pokušaj ponovo za nekoliko sekundi",
error_code="AI001"
).dict()
)
try:
# Rutiraj zahtev
force_provider = None
if isinstance(request, StructuredQuestionRequest):
force_provider = request.force_provider
selected_provider, routing_metadata = smart_router.route_request(
structured_request,
override_provider=force_provider
)
# Promeni provider ako je potrebno
original_provider = Config.AI_PROVIDER
if selected_provider != original_provider and selected_provider != "simulation":
Config.AI_PROVIDER = selected_provider
AIServiceFactory.reset()
current_service = AIServiceFactory.create_resilient_service()
else:
current_service = ai_service
# Dobij optimizovane parametre
optimized_params = structured_request.get_optimized_params()
# Primeni parametre
current_service.apply_settings(optimized_params)
# Generiši poboljšan prompt
enhanced_prompt = structured_request.get_enhanced_prompt()
# Pozovi AI
odgovor = current_service.pozovi_ai(enhanced_prompt, VASA_LICNOST)
# Vrati originalni provider
if selected_provider != original_provider:
Config.AI_PROVIDER = original_provider
AIServiceFactory.reset()
# Računaj vreme odgovora
response_time_ms = int((time.time() - start_time) * 1000)
# Pripremi response koristeći Pydantic model
return QuestionResponse(
pitanje=request.pitanje,
odgovor=odgovor,
tip_zahteva=structured_request.request_type.value,
provider=ProviderInfo(
selected=selected_provider,
reason=routing_metadata.get("selected_reason", "Default selection"),
strategy=routing_metadata.get("strategy", "unknown"),
available_providers=routing_metadata.get("available_providers", [])
),
optimization=OptimizationInfo(
temperature=optimized_params.get("temperature", 0.7),
max_tokens=optimized_params.get("max_tokens", 150),
adjusted_for_type=True
),
context_used=structured_request.context.has_code_context(),
session_id=request.session_id if isinstance(request, StructuredQuestionRequest) else None,
response_time_ms=response_time_ms
)
except ValidationError as e:
# Pydantic validation error
return JSONResponse(
status_code=422,
content=ErrorResponse(
error="validation_error",
detail=str(e),
suggestion="Proveri format podataka",
error_code="VAL002"
).dict()
)
except Exception as e:
# Opšta greška
print(f"❌ Greška pri obradi pitanja: {e}")
return JSONResponse(
status_code=500,
content=ErrorResponse(
error="processing_error",
detail="Dogodila se greška pri obradi pitanja",
suggestion="Pokušaj ponovo ili promeni formulaciju",
error_code="PROC001"
).dict()
)
# Dodaj novi endpoint za validaciju
@app.post("/validate-request",
summary="Validira zahtev bez slanja AI-ju",
description="Koristi za proveru da li je zahtev valjan pre slanja",
tags=["Validation"]
)
async def validate_request(
request: StructuredQuestionRequest
) -> Dict[str, Any]:
"""
Validira strukturiran zahtev i vraća analizu.
Korisno za frontend da proveri podatke pre slanja.
"""
# Analiziraj tip ako nije postavljen
detected_type = None
if not request.tip:
detected_type = RequestAnalyzer.analyze(request.pitanje)
# Proveri kontekst
context_complete = False
context_warnings = []
if request.context:
if request.tip == "debug" or detected_type == RequestType.CODE_DEBUG:
if not request.context.code_snippet and not request.context.error_message:
context_warnings.append(
"Debug zahtevi obično trebaju kod ili error poruku"
)
if request.context.code_snippet:
# Proveri da li je kod bezbedan
from web_api.models.validation import sanitize_code_snippet
sanitized = sanitize_code_snippet(request.context.code_snippet)
if sanitized != request.context.code_snippet:
context_warnings.append(
"Kod sadrži potencijalno nesigurne elemente"
)
context_complete = bool(
request.context.programming_language or
request.context.code_snippet
)
# Analiza preferencija
preference_notes = []
if request.preferences:
if request.preferences.temperature < 0.3:
preference_notes.append("Vrlo niska temperatura - deterministički odgovori")
elif request.preferences.temperature > 1.5:
preference_notes.append("Vrlo visoka temperatura - kreativni ali možda nestabilni odgovori")
if request.preferences.max_tokens < 100:
preference_notes.append("Mali broj tokena - kratak odgovor")
elif request.preferences.max_tokens > 1000:
preference_notes.append("Veliki broj tokena - dugačak odgovor")
return {
"valid": True,
"analysis": {
"detected_type": detected_type.value if detected_type else request.tip,
"has_context": request.context is not None,
"context_complete": context_complete,
"context_warnings": context_warnings,
"has_preferences": request.preferences is not None,
"preference_notes": preference_notes,
"estimated_tokens": len(request.pitanje.split()) * 2, # Gruba procena
"force_provider": request.force_provider,
"session_tracking": request.session_id is not None
},
"suggestions": [
w for w in context_warnings
] + preference_notes
}
# Ažuriraj i ostale endpoint-e da koriste modele
from pydantic import create_model
from typing import Type
# Dinamički kreiraj response modele za postojeće endpoint-e
ProviderListResponse = create_model(
'ProviderListResponse',
providers=(List[Dict[str, Any]], ...),
active_provider=(str, ...),
total_configured=(int, ...)
)
@app.get("/providers",
response_model=ProviderListResponse,
tags=["Providers"]
)
async def get_providers():
"""Vraća informacije o dostupnim AI providerima."""
# Postojeća logika ostaje ista
providers = []
if Config.OPENAI_API_KEY:
providers.append({
"name": "openai",
"display_name": "OpenAI GPT",
"available": True,
"is_active": Config.AI_PROVIDER == "openai",
"features": ["chat", "code_generation", "analysis"]
})
if Config.GEMINI_API_KEY:
providers.append({
"name": "gemini",
"display_name": "Google Gemini",
"available": True,
"is_active": Config.AI_PROVIDER == "gemini",
"features": ["chat", "multimodal", "fast_responses"]
})
if not providers:
providers.append({
"name": "simulation",
"display_name": "Lokalna simulacija",
"available": True,
"is_active": True,
"features": ["basic_responses"]
})
return ProviderListResponse(
providers=providers,
active_provider=Config.AI_PROVIDER,
total_configured=len([p for p in providers if p["name"] != "simulation"])
)
Kreiranje provider-specific servisa sa validacijom
Sada ćemo kreirati servise koji koriste provider-specifične modele.
📚 NIVO 1 – Osnovno objašnjenje
Svaki AI provider (OpenAI, Gemini) ima svoje posebne opcije:
- OpenAI ima
presence_penaltyza kontrolu ponavljanja - Gemini ima
safety_settingsza filtriranje sadržaja
Provider-specifični servisi omogućavaju da koristiš ove posebne funkcije dok zadržavaš jedinstveni interfejs za osnovne operacije.
🚀 NIVO 2 – Dublje razumevanje
Provider-specific implementacija omogućava:
- Fine-grained kontrolu nad svakim servisom
- Validaciju opcija specifičnih za provider
- Optimizaciju za karakteristike svakog modela
- Fallback na default vrednosti
- Type safety za provider-specific features
📊 DIJAGRAM: Dinamička Validacija Providera
[Strukturiran Zahtev]
|
└─ provider_specific: {
"provider": "openai",
"options": { "model": "gpt-4", ... }
}
|
↓
[ProviderSpecificRequest (Pydantic Model)]
|
├─ provider: Literal["openai", "gemini"]
|
└─ options: Union[OpenAISpecificRequest, GeminiSpecificRequest]
|
↓
[ @root_validator ] <-- Sistemska provera
|
└─ "Da li 'provider' polje (openai) odgovara tipu 'options' objekta (OpenAISpecificRequest)?"
|
/ \
DA NE
| |
↓ ↓
[✅ OK] [❌ ValueError]
💡 PRO TIP: Umesto da hardkoduješ provider opcije, koristi “capability discovery”. Svaki provider servis može da vrati get_capabilities() sa listom podržanih features. Frontend može dinamički da prikaže samo relevantne opcije!
Kreiraj src/ai_services/validated_services.py:
"""
Provider servisi sa Pydantic validacijom
Omogućava type-safe rad sa AI servisima
"""
from typing import Dict, Any, Optional, List
from pydantic import BaseModel, ValidationError
import logging
from ai_services.base_service import BaseAIService
from ai_services.openai_service import OpenAIService
from ai_services.gemini_service import GeminiService
from web_api.models.validation import (
OpenAISpecificRequest,
GeminiSpecificRequest,
ProviderSpecificRequest
)
logger = logging.getLogger(__name__)
class ValidatedOpenAIService(OpenAIService):
"""OpenAI servis sa Pydantic validacijom."""
def __init__(self):
super().__init__()
self._custom_settings: Optional[OpenAISpecificRequest] = None
def apply_validated_settings(self, settings: OpenAISpecificRequest):
"""
Primenjuje validirane postavke.
Args:
settings: OpenAI specifične postavke
"""
self._custom_settings = settings
# Konvertuj u standardne postavke
standard_settings = {
"model": settings.model,
"temperature": settings.temperature if hasattr(settings, 'temperature') else 0.7,
"max_tokens": settings.max_tokens if hasattr(settings, 'max_tokens') else 150,
"top_p": settings.top_p,
"frequency_penalty": settings.frequency_penalty,
"presence_penalty": settings.presence_penalty
}
# Primeni kroz baznu metodu
self.apply_settings(standard_settings)
logger.info(f"Primenjene OpenAI postavke: model={settings.model}, top_p={settings.top_p}")
def pozovi_ai(
self,
poruka: str,
system_prompt: Optional[str] = None,
**kwargs
) -> str:
"""
Poziva OpenAI sa validiranim postavkama.
Args:
poruka: Korisnička poruka
system_prompt: System prompt
**kwargs: Dodatne opcije
Returns:
AI odgovor
"""
# Ako imamo custom settings, koristi ih
if self._custom_settings:
# Override system prompt ako je dat
if self._custom_settings.system_prompt:
system_prompt = self._custom_settings.system_prompt
# Dodaj stop sekvence ako postoje
if self._custom_settings.stop_sequences:
kwargs['stop'] = self._custom_settings.stop_sequences
# Pozovi parent metodu
return super().pozovi_ai(poruka, system_prompt, **kwargs)
def get_capabilities(self) -> Dict[str, Any]:
"""Vraća mogućnosti ovog servisa."""
return {
"provider": "openai",
"models": ["gpt-3.5-turbo", "gpt-4", "gpt-4-turbo"],
"features": {
"streaming": True,
"function_calling": True,
"vision": False, # Zavisi od modela
"max_context": 4096, # Zavisi od modela
"fine_tuning": True
},
"parameters": {
"temperature": {"min": 0, "max": 2, "default": 0.7},
"max_tokens": {"min": 1, "max": 4096, "default": 150},
"top_p": {"min": 0, "max": 1, "default": 1},
"frequency_penalty": {"min": -2, "max": 2, "default": 0},
"presence_penalty": {"min": -2, "max": 2, "default": 0}
}
}
class ValidatedGeminiService(GeminiService):
"""Gemini servis sa Pydantic validacijom."""
def __init__(self):
super().__init__()
self._custom_settings: Optional[GeminiSpecificRequest] = None
def apply_validated_settings(self, settings: GeminiSpecificRequest):
"""
Primenjuje validirane Gemini postavke.
Args:
settings: Gemini specifične postavke
"""
self._custom_settings = settings
# Gemini koristi drugačije nazive
standard_settings = {
"model": settings.model,
"temperature": 0.7, # Default, Gemini možda ne podržava
"max_tokens": 150, # Default
"candidate_count": settings.candidate_count
}
self.apply_settings(standard_settings)
logger.info(f"Primenjene Gemini postavke: model={settings.model}, candidates={settings.candidate_count}")
def pozovi_ai(
self,
poruka: str,
system_prompt: Optional[str] = None,
**kwargs
) -> str:
"""
Poziva Gemini sa validiranim postavkama.
Args:
poruka: Korisnička poruka
system_prompt: System prompt
**kwargs: Dodatne opcije
Returns:
AI odgovor
"""
# Pripremi generation config
if self._custom_settings:
generation_config = {
"candidate_count": self._custom_settings.candidate_count,
"stop_sequences": self._custom_settings.stop_sequences,
"temperature": kwargs.get('temperature', 0.7),
"max_output_tokens": kwargs.get('max_tokens', 150)
}
# Primeni safety settings
if self._custom_settings.safety_settings:
kwargs['safety_settings'] = self._custom_settings.safety_settings
kwargs['generation_config'] = generation_config
return super().pozovi_ai(poruka, system_prompt, **kwargs)
def get_capabilities(self) -> Dict[str, Any]:
"""Vraća mogućnosti Gemini servisa."""
return {
"provider": "gemini",
"models": ["gemini-pro", "gemini-pro-vision"],
"features": {
"streaming": True,
"function_calling": False, # Trenutno
"vision": True, # gemini-pro-vision
"max_context": 32768,
"fine_tuning": False
},
"parameters": {
"temperature": {"min": 0, "max": 1, "default": 0.7},
"max_tokens": {"min": 1, "max": 2048, "default": 150},
"candidate_count": {"min": 1, "max": 8, "default": 1}
},
"safety_categories": [
"HARM_CATEGORY_HARASSMENT",
"HARM_CATEGORY_HATE_SPEECH",
"HARM_CATEGORY_SEXUALLY_EXPLICIT",
"HARM_CATEGORY_DANGEROUS_CONTENT"
]
}
class ValidatedAIServiceFactory:
"""Factory za kreiranje validiranih AI servisa."""
@staticmethod
def create_validated_service(
provider: str,
specific_request: Optional[ProviderSpecificRequest] = None
) -> BaseAIService:
"""
Kreira validiran AI servis.
Args:
provider: Ime providera
specific_request: Provider-specifične opcije
Returns:
Validiran AI servis
Raises:
ValueError: Ako provider nije podržan
"""
if provider == "openai":
service = ValidatedOpenAIService()
if specific_request and isinstance(specific_request.options, OpenAISpecificRequest):
service.apply_validated_settings(specific_request.options)
elif provider == "gemini":
service = ValidatedGeminiService()
if specific_request and isinstance(specific_request.options, GeminiSpecificRequest):
service.apply_validated_settings(specific_request.options)
else:
raise ValueError(f"Provider '{provider}' nije podržan")
return service
@staticmethod
def get_provider_schema(provider: str) -> Dict[str, Any]:
"""
Vraća JSON Schema za provider-specifične opcije.
Args:
provider: Ime providera
Returns:
JSON Schema
"""
if provider == "openai":
return OpenAISpecificRequest.schema()
elif provider == "gemini":
return GeminiSpecificRequest.schema()
else:
return {}
# Dodaj endpoint za provider-specific pozive
def add_provider_specific_endpoint(app):
"""Dodaje provider-specific endpoint u FastAPI app."""
@app.post("/providers/{provider}/ask",
summary="Provider-specifični AI poziv",
description="Poziva specifičan provider sa custom opcijama",
tags=["Providers"]
)
async def provider_specific_ask(
provider: str,
request: ProviderSpecificRequest
):
"""Poziva specifičan provider sa validiranim opcijama."""
try:
# Validuj da provider odgovara request-u
if provider != request.provider:
raise HTTPException(
status_code=400,
detail=f"Provider '{provider}' ne odgovara request provideru '{request.provider}'"
)
# Kreiraj servis sa opcijama
service = ValidatedAIServiceFactory.create_validated_service(
provider,
request
)
# Osnovno pitanje mora postojati
if not hasattr(request, 'question'):
raise HTTPException(
status_code=400,
detail="Request mora imati 'question' polje"
)
# Pozovi AI
response = service.pozovi_ai(request.question)
return {
"provider": provider,
"question": request.question,
"response": response,
"options_applied": request.options.dict()
}
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
logger.error(f"Provider-specific error: {e}")
raise HTTPException(
status_code=500,
detail="Greška pri pozivanju provider-specific servisa"
)
# Test validiranih servisa
if __name__ == "__main__":
print("🧪 Test Validated Services")
print("=" * 50)
# Test OpenAI sa validacijom
print("\n1. Test ValidatedOpenAIService:")
try:
openai_settings = OpenAISpecificRequest(
model="gpt-4",
temperature=0.5,
top_p=0.9,
presence_penalty=0.1
)
service = ValidatedOpenAIService()
service.apply_validated_settings(openai_settings)
print(f"✅ OpenAI servis konfigurisan")
print(f" Model: {openai_settings.model}")
print(f" Capabilities: {service.get_capabilities()['features']}")
except Exception as e:
print(f"❌ Greška: {e}")
# Test schema generisanja
print("\n2. Test Schema generisanja:")
schema = ValidatedAIServiceFactory.get_provider_schema("openai")
print(f"✅ OpenAI schema ima {len(schema['properties'])} propertija")
schema = ValidatedAIServiceFactory.get_provider_schema("gemini")
print(f"✅ Gemini schema ima {len(schema['properties'])} propertija")
Praktična implementacija
Testiranje Pydantic validacije
Kreiraj src/test_validation.py:
"""
Test scenariji za Pydantic validaciju
"""
import requests
import json
from typing import Dict, Any
BASE_URL = "http://localhost:8000"
def test_simple_validation():
"""Testira osnovnu validaciju pitanja."""
print("\n🧪 TEST 1: Osnovna validacija")
print("=" * 50)
test_cases = [
{
"name": "Validno pitanje",
"data": {"pitanje": "Kako da sortiram listu?"},
"should_pass": True
},
{
"name": "Prazno pitanje",
"data": {"pitanje": " "},
"should_pass": False
},
{
"name": "Bez pitanja",
"data": {},
"should_pass": False
},
{
"name": "Predugačko pitanje",
"data": {"pitanje": "x" * 2001},
"should_pass": False
}
]
for test in test_cases:
response = requests.post(f"{BASE_URL}/pitaj", json=test["data"])
print(f"\n{test['name']}:")
print(f" Status: {response.status_code}")
if test["should_pass"]:
if response.status_code == 200:
print(" ✅ Prošao kako je očekivano")
else:
print(" ❌ Trebalo je da prođe!")
print(f" Greška: {response.json()}")
else:
if response.status_code == 422:
print(" ✅ Validacija radila kako treba")
error = response.json()
print(f" Detalj: {error.get('detail', 'N/A')}")
else:
print(" ❌ Trebalo je da failuje!")
def test_structured_validation():
"""Testira strukturiranu validaciju."""
print("\n🧪 TEST 2: Strukturirana validacija")
print("=" * 50)
# Validan strukturiran zahtev
valid_request = {
"pitanje": "Debug ovaj Python kod",
"tip": "debug",
"context": {
"programming_language": "python",
"code_snippet": "def test():\n return None",
"user_level": "beginner"
},
"preferences": {
"temperature": 0.3,
"response_style": "detailed",
"include_examples": True
}
}
response = requests.post(f"{BASE_URL}/pitaj", json=valid_request)
if response.status_code == 200:
data = response.json()
print("✅ Strukturiran zahtev prošao")
print(f" Tip: {data.get('tip_zahteva')}")
print(f" Kontekst korišćen: {data.get('context_used')}")
print(f" Optimizacija: {data.get('optimization')}")
else:
print(f"❌ Greška: {response.status_code}")
print(response.json())
# Test sa nevalidnim context
invalid_context = valid_request.copy()
invalid_context["context"]["user_level"] = "super-expert" # Nije valjan enum
response = requests.post(f"{BASE_URL}/pitaj", json=invalid_context)
print(f"\nNevalidan user level:")
print(f" Status: {response.status_code}")
if response.status_code == 422:
print(" ✅ Validacija uhvatila grešku")
# Test sa nevalidnim preferences
invalid_prefs = valid_request.copy()
invalid_prefs["preferences"]["temperature"] = 3.0 # Preko maksimuma
response = requests.post(f"{BASE_URL}/pitaj", json=invalid_prefs)
print(f"\nNevalidna temperatura:")
print(f" Status: {response.status_code}")
if response.status_code == 422:
print(" ✅ Validacija uhvatila grešku")
def test_validation_endpoint():
"""Testira /validate-request endpoint."""
print("\n🧪 TEST 3: Validation endpoint")
print("=" * 50)
test_request = {
"pitanje": "Zašto ovaj kod ne radi?",
"context": {
"programming_language": "python"
}
}
response = requests.post(f"{BASE_URL}/validate-request", json=test_request)
if response.status_code == 200:
analysis = response.json()
print("✅ Validacija analize:")
print(f" Detektovan tip: {analysis['analysis']['detected_type']}")
print(f" Kontekst kompletan: {analysis['analysis']['context_complete']}")
if analysis['analysis']['context_warnings']:
print(f" Upozorenja: {analysis['analysis']['context_warnings']}")
if analysis['suggestions']:
print(f" Sugestije: {analysis['suggestions']}")
else:
print(f"❌ Greška: {response.status_code}")
def test_response_validation():
"""Testira da li response odgovara modelu."""
print("\n🧪 TEST 4: Response validacija")
print("=" * 50)
response = requests.post(
f"{BASE_URL}/pitaj",
json={"pitanje": "Test validacije response-a"}
)
if response.status_code == 200:
data = response.json()
# Proveri da li postoje sva očekivana polja
expected_fields = [
"pitanje", "odgovor", "tip_zahteva",
"provider", "optimization", "response_time_ms"
]
missing = [f for f in expected_fields if f not in data]
if not missing:
print("✅ Sva polja prisutna u response-u")
# Proveri strukture
if isinstance(data.get("provider"), dict):
provider_fields = ["selected", "reason", "strategy"]
provider_ok = all(f in data["provider"] for f in provider_fields)
print(f" Provider info kompletna: {'✅' if provider_ok else '❌'}")
if isinstance(data.get("optimization"), dict):
opt_fields = ["temperature", "max_tokens", "adjusted_for_type"]
opt_ok = all(f in data["optimization"] for f in opt_fields)
print(f" Optimization info kompletna: {'✅' if opt_ok else '❌'}")
else:
print(f"❌ Nedostaju polja: {missing}")
else:
print(f"❌ Request failed: {response.status_code}")
def test_provider_schemas():
"""Testira provider-specific scheme."""
print("\n🧪 TEST 5: Provider schemas")
print("=" * 50)
# Ovaj test bi trebalo da bude endpoint, ali možemo simulirati
print("📌 Provider-specific schemas bi trebalo da budu dostupne kroz API")
print(" Predlog: GET /providers/{provider}/schema")
# Test provider info sa capabilities
response = requests.get(f"{BASE_URL}/providers")
if response.status_code == 200:
data = response.json()
print(f"\n✅ Dostupni provideri: {len(data['providers'])}")
for provider in data['providers']:
print(f"\n {provider['display_name']}:")
print(f" Ime: {provider['name']}")
print(f" Features: {provider.get('features', [])}")
def run_all_tests():
"""Pokreće sve testove."""
print("🚀 PYDANTIC VALIDATION TEST SUITE")
print("=" * 60)
print("\n⚠️ Proveri da li je server pokrenut na http://localhost:8000")
input("Pritisni ENTER za početak testiranja...")
try:
# Proveri da li server radi
health = requests.get(f"{BASE_URL}/health")
if health.status_code != 200:
print("❌ Server ne odgovara!")
return
test_simple_validation()
test_structured_validation()
test_validation_endpoint()
test_response_validation()
test_provider_schemas()
print("\n✅ Svi testovi završeni!")
except requests.exceptions.ConnectionError:
print("❌ Ne mogu da se povežem sa serverom!")
print(" Pokreni server sa: python src/web_api/run_server.py")
except Exception as e:
print(f"❌ Neočekivana greška: {e}")
if __name__ == "__main__":
run_all_tests()
🔄 VEŽBA: Pre nego što pokrenete skriptu za testiranje, izmenite temperature u test-case-u na -0.5. Pogodite kakvu će grešku Pydantic baciti i zašto. Zatim ispravite vrednost i pokrenite test ponovo.
Kreiranje interaktivnog test UI-ja
Za lakše testiranje, kreiraj jednostavan HTML interface.
Kreiraj src/web_api/static/validation_test.html:
<!DOCTYPE html>
<html lang="sr">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Učitelj Vasa - Pydantic Validation Test</title>
<style>
body {
font-family: Arial, sans-serif;
max-width: 1200px;
margin: 0 auto;
padding: 20px;
background-color: #f5f5f5;
}
.container {
background: white;
padding: 20px;
border-radius: 8px;
box-shadow: 0 2px 4px rgba(0,0,0,0.1);
margin-bottom: 20px;
}
h1, h2 {
color: #333;
}
.form-group {
margin-bottom: 15px;
}
label {
display: block;
margin-bottom: 5px;
font-weight: bold;
color: #555;
}
input[type="text"],
textarea,
select {
width: 100%;
padding: 8px;
border: 1px solid #ddd;
border-radius: 4px;
box-sizing: border-box;
}
textarea {
min-height: 100px;
resize: vertical;
}
button {
background-color: #007bff;
color: white;
padding: 10px 20px;
border: none;
border-radius: 4px;
cursor: pointer;
margin-right: 10px;
}
button:hover {
background-color: #0056b3;
}
button.secondary {
background-color: #6c757d;
}
button.secondary:hover {
background-color: #545b62;
}
.response {
background-color: #f8f9fa;
padding: 15px;
border-radius: 4px;
margin-top: 20px;
white-space: pre-wrap;
font-family: 'Courier New', monospace;
font-size: 14px;
max-height: 400px;
overflow-y: auto;
}
.error {
background-color: #f8d7da;
color: #721c24;
padding: 10px;
border-radius: 4px;
margin-top: 10px;
}
.success {
background-color: #d4edda;
color: #155724;
padding: 10px;
border-radius: 4px;
margin-top: 10px;
}
.flex-row {
display: flex;
gap: 20px;
}
.flex-col {
flex: 1;
}
.checkbox-group {
margin: 10px 0;
}
.checkbox-group label {
display: inline;
margin-left: 5px;
font-weight: normal;
}
.slider-group {
display: flex;
align-items: center;
gap: 10px;
}
.slider {
flex: 1;
}
.slider-value {
min-width: 50px;
text-align: right;
}
.json-input {
font-family: 'Courier New', monospace;
font-size: 13px;
}
</style>
</head>
<body>
<h1>🎓 Učitelj Vasa - Pydantic Validation Test</h1>
<div class="container">
<h2>Strukturirani Zahtev</h2>
<div class="form-group">
<label for="question">Pitanje: *</label>
<textarea id="question" placeholder="Upiši svoje pitanje...">Kako da sortiram listu u Python-u?</textarea>
</div>
<div class="flex-row">
<div class="flex-col">
<h3>Tip Zahteva</h3>
<div class="form-group">
<label for="requestType">Tip:</label>
<select id="requestType">
<option value="">Automatski prepoznaj</option>
<option value="chat">Chat - Obična konverzacija</option>
<option value="code">Code - Generisanje koda</option>
<option value="debug">Debug - Pomoć sa greškama</option>
<option value="explain">Explain - Objašnjenje koncepata</option>
<option value="review">Review - Pregled koda</option>
<option value="translate">Translate - Prevod koda</option>
<option value="optimize">Optimize - Optimizacija</option>
</select>
</div>
<h3>Kontekst</h3>
<div class="form-group">
<label for="language">Programski jezik:</label>
<select id="language">
<option value="">Nije specificirano</option>
<option value="python">Python</option>
<option value="javascript">JavaScript</option>
<option value="java">Java</option>
<option value="cpp">C++</option>
<option value="csharp">C#</option>
<option value="go">Go</option>
<option value="rust">Rust</option>
</select>
</div>
<div class="form-group">
<label for="userLevel">Nivo znanja:</label>
<select id="userLevel">
<option value="beginner">Početnik</option>
<option value="intermediate" selected>Srednji</option>
<option value="advanced">Napredan</option>
<option value="expert">Ekspert</option>
</select>
</div>
<div class="form-group">
<label for="errorMessage">Error poruka (za debug):</label>
<input type="text" id="errorMessage" placeholder="npr. IndexError: list index out of range">
</div>
</div>
<div class="flex-col">
<h3>Preference</h3>
<div class="form-group">
<label>Temperatura (kreativnost):</label>
<div class="slider-group">
<input type="range" id="temperature" class="slider"
min="0" max="2" step="0.1" value="0.7">
<span id="tempValue" class="slider-value">0.7</span>
</div>
</div>
<div class="form-group">
<label>Max tokena:</label>
<div class="slider-group">
<input type="range" id="maxTokens" class="slider"
min="50" max="2000" step="50" value="150">
<span id="tokensValue" class="slider-value">150</span>
</div>
</div>
<div class="form-group">
<label for="responseStyle">Stil odgovora:</label>
<select id="responseStyle">
<option value="concise">Sažet</option>
<option value="detailed">Detaljan</option>
<option value="tutorial">Tutorial</option>
</select>
</div>
<div class="checkbox-group">
<input type="checkbox" id="includeExamples" checked>
<label for="includeExamples">Uključi primere</label>
</div>
<div class="form-group">
<label for="forceProvider">Forsiraj provider:</label>
<select id="forceProvider">
<option value="">Automatski izbor</option>
<option value="openai">OpenAI</option>
<option value="gemini">Gemini</option>
</select>
</div>
</div>
</div>
<div class="form-group">
<label for="codeSnippet">Kod (opciono):</label>
<textarea id="codeSnippet" class="json-input" placeholder="Paste your code here..."></textarea>
</div>
<div class="form-group">
<label for="sessionId">Session ID (opciono):</label>
<input type="text" id="sessionId" placeholder="sess_abc123" pattern="^[a-zA-Z0-9_-]+$">
</div>
<button onclick="sendRequest()">Pošalji Pitanje</button>
<button onclick="validateOnly()" class="secondary">Samo Validiraj</button>
<button onclick="showJSON()" class="secondary">Prikaži JSON</button>
<button onclick="clearForm()" class="secondary">Očisti</button>
<div id="result"></div>
</div>
<div class="container">
<h2>Raw JSON Editor</h2>
<div class="form-group">
<label for="rawJson">JSON Request:</label>
<textarea id="rawJson" class="json-input" rows="15">{
"pitanje": "Kako da sortiram listu?",
"tip": "explain",
"context": {
"programming_language": "python",
"user_level": "beginner"
},
"preferences": {
"temperature": 0.7,
"response_style": "detailed"
}
}</textarea>
</div>
<button onclick="sendRawJSON()">Pošalji Raw JSON</button>
</div>
<script>
// Update slider values
document.getElementById('temperature').addEventListener('input', function(e) {
document.getElementById('tempValue').textContent = e.target.value;
});
document.getElementById('maxTokens').addEventListener('input', function(e) {
document.getElementById('tokensValue').textContent = e.target.value;
});
function buildRequest() {
const request = {
pitanje: document.getElementById('question').value
};
// Tip
const tip = document.getElementById('requestType').value;
if (tip) request.tip = tip;
// Context
const context = {};
const lang = document.getElementById('language').value;
if (lang) context.programming_language = lang;
const errorMsg = document.getElementById('errorMessage').value;
if (errorMsg) context.error_message = errorMsg;
const code = document.getElementById('codeSnippet').value;
if (code) context.code_snippet = code;
context.user_level = document.getElementById('userLevel').value;
if (Object.keys(context).length > 1) { // Samo user_level uvek postoji
request.context = context;
}
// Preferences
const preferences = {
temperature: parseFloat(document.getElementById('temperature').value),
max_tokens: parseInt(document.getElementById('maxTokens').value),
response_style: document.getElementById('responseStyle').value,
include_examples: document.getElementById('includeExamples').checked
};
request.preferences = preferences;
// Force provider
const forceProvider = document.getElementById('forceProvider').value;
if (forceProvider) request.force_provider = forceProvider;
// Session ID
const sessionId = document.getElementById('sessionId').value;
if (sessionId) request.session_id = sessionId;
return request;
}
async function sendRequest() {
const resultDiv = document.getElementById('result');
resultDiv.innerHTML = '<div class="success">Šalje se zahtev...</div>';
try {
const request = buildRequest();
const response = await fetch('http://localhost:8000/pitaj', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify(request)
});
const data = await response.json();
if (response.ok) {
resultDiv.innerHTML = `
<div class="success">✅ Uspešno!</div>
<div class="response">
<strong>ODGOVOR:</strong>
${data.odgovor}
<strong>METADATA:</strong>
Tip zahteva: ${data.tip_zahteva}
Provider: ${data.provider.selected} (${data.provider.reason})
Strategija: ${data.provider.strategy}
Vreme odgovora: ${data.response_time_ms}ms
Kontekst korišćen: ${data.context_used ? 'Da' : 'Ne'}
<strong>OPTIMIZACIJA:</strong>
Temperature: ${data.optimization.temperature}
Max tokena: ${data.optimization.max_tokens}
</div>
`;
} else {
resultDiv.innerHTML = `
<div class="error">❌ Greška ${response.status}</div>
<div class="response">${JSON.stringify(data, null, 2)}</div>
`;
}
} catch (error) {
resultDiv.innerHTML = `
<div class="error">❌ Network greška: ${error.message}</div>
`;
}
}
async function validateOnly() {
const resultDiv = document.getElementById('result');
resultDiv.innerHTML = '<div class="success">Validiram zahtev...</div>';
try {
const request = buildRequest();
const response = await fetch('http://localhost:8000/validate-request', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify(request)
});
const data = await response.json();
if (response.ok) {
resultDiv.innerHTML = `
<div class="success">✅ Validacija prošla!</div>
<div class="response">${JSON.stringify(data, null, 2)}</div>
`;
} else {
resultDiv.innerHTML = `
<div class="error">❌ Validacija neuspešna</div>
<div class="response">${JSON.stringify(data, null, 2)}</div>
`;
}
} catch (error) {
resultDiv.innerHTML = `
<div class="error">❌ Network greška: ${error.message}</div>
`;
}
}
function showJSON() {
const request = buildRequest();
document.getElementById('rawJson').value = JSON.stringify(request, null, 2);
document.getElementById('result').innerHTML = `
<div class="success">JSON prikazan u editoru ispod</div>
`;
}
async function sendRawJSON() {
const resultDiv = document.getElementById('result');
try {
const json = JSON.parse(document.getElementById('rawJson').value);
const response = await fetch('http://localhost:8000/pitaj', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify(json)
});
const data = await response.json();
resultDiv.innerHTML = `
<div class="${response.ok ? 'success' : 'error'}">
Status: ${response.status}
</div>
<div class="response">${JSON.stringify(data, null, 2)}</div>
`;
} catch (error) {
resultDiv.innerHTML = `
<div class="error">❌ JSON Parse greška: ${error.message}</div>
`;
}
}
function clearForm() {
document.getElementById('question').value = '';
document.getElementById('requestType').value = '';
document.getElementById('language').value = '';
document.getElementById('userLevel').value = 'intermediate';
document.getElementById('errorMessage').value = '';
document.getElementById('codeSnippet').value = '';
document.getElementById('temperature').value = 0.7;
document.getElementById('tempValue').textContent = '0.7';
document.getElementById('maxTokens').value = 150;
document.getElementById('tokensValue').textContent = '150';
document.getElementById('responseStyle').value = 'concise';
document.getElementById('includeExamples').checked = true;
document.getElementById('forceProvider').value = '';
document.getElementById('sessionId').value = '';
document.getElementById('result').innerHTML = '';
}
</script>
</body>
</html>
Dodaj u FastAPI da servira statičke fajlove. U src/web_api/app.py dodaj:
from fastapi.staticfiles import StaticFiles
import os
# Nakon kreiranja app instance
app.mount(
"/static",
StaticFiles(directory=os.path.join(os.path.dirname(__file__), "static")),
name="static"
)
Sada možeš testirati kroz browser: http://localhost:8000/static/validation_test.html
Dodatni primeri i zabavne vežbe
- Validiraj datum rođenja: Napravite Pydantic model
UserProfilesa poljembirth_date: datei validatorom koji vraća grešku ako je korisnik mlađi od 13 godina. - Pretvori string u enum: Dodajte polje
preferred_idekoje prihvata stringove “pycharm”, “vscode”, “thony”, ali ih pretvara u EnumIDE.
🔄 VEŽBA: Implementirajte oba zahteva, stavite ih u validation.py, pa kroz Swagger proverite da li se pojavljuju odgovarajuće greške.
🎈 ZABAVNA ČINJENICA: Najmlađi poznati Pydantic contributor imao je 13 godina — taman uz restrikciju iz zadatka!
Česte greške i rešenja
❌ GREŠKA: ValidationError: field required
💡 REŠENJE:
- Proveri da li šalješ sva obavezna polja (označena sa … u modelu)
- Koristi Swagger UI da vidiš tačan format
- Proveri da li je JSON pravilno formatiran
🔬 DETALJNIJE: Greška ValidationError je najvažnija karakteristika Pydantic-a. Ona se dešava jer Pydantic radi po principu “fail fast” (padni brzo). Sistem je dizajniran da zaustavi izvršavanje čim detektuje neispravan podatak na samom ulazu. Zašto je ovo dobro? Zato što sprečava da neispravni ili neočekivani podaci (npr. None umesto stringa) stignu duboko u logiku tvoje aplikacije, gde bi mogli da izazovu mnogo teže za pronalaženje greške (npr. AttributeError: 'NoneType' object has no attribute 'strip'). Ova greška nije problem, već tvoj najraniji i najbolji signal da podaci koje dobijaš nisu u skladu sa “ugovorom” (schemom) koji si definisao.
❌ GREŠKA: 422 Unprocessable Entity sa Pydantic modelima
💡 REŠENJE:
- Pročitaj detail u error response-u – Pydantic daje jasne poruke
- Proveri tipove podataka (broj vs string)
- Proveri enum vrednosti – moraju biti tačne
❌ GREŠKA: Response ne odgovara response_model
💡 REŠENJE:
- Proveri da li vraćaš sve required fields
- Koristi Pydantic model umesto dict-a
- Dodaj default vrednosti za optional polja
❌ GREŠKA: Enum validation fails
💡 REŠENJE:
- Koristi .value kada vraćaš enum
- Dodaj use_enum_values = True u Config
- Proveri case-sensitive vrednosti
❌ GREŠKA: Custom validator ne radi
💡 REŠENJE:
- Validator mora da vrati vrednost, ne samo da validira
- Koristi @validator(‘field_name’) sa tačnim imenom
- Za više polja koristi @root_validator
Proveri svoje razumevanje
[NIVO 1]:
- Šta je Pydantic i čemu služi?
- Kako FastAPI koristi Pydantic modele?
- Šta znači … u Field definiciji?
- Zašto je validacija važna za API?
[NIVO 2]:
- Koja je razlika između @validator i @root_validator?
- Kako bi dodao custom tip sa regex validacijom?
- Šta je prednost response_model u FastAPI?
- Kako Pydantic pomaže sa API dokumentacijom?
- Kada koristiti Union tipove u modelima?
🤔 MINI-KVIZ:
- Šta će se desiti ako polju sa
constr(min_length=5)pošaljete reč “chat”? - Kako biste u Pydantic-u validirali da je
max_tokensveći odtemperature × 50? - Da li
validatormože da pristupi dva različita polja istovremeno? Objasnite.
Ažuriranje dokumentacije
Ažuriraj README.md:
## 🚀 Trenutni Status
- ✅ Dan -3: Python 3.13+ instaliran
- ✅ Dan -2: PyCharm unified edition podešen
- ✅ Dan -1: GitHub repository kreiran
- ✅ Dan 0: Profesionalna struktura projekta
- ✅ Dan 1: Prvi Python moduli - Vasa može da pozdravi!
- ✅ Dan 2: Razumevanje AI API-ja - simulacija komunikacije
- ✅ Dan 3: Multi-provider podrška - OpenAI i Gemini
- ✅ Dan 4: Prvi AI poziv - univerzalni sistem sa SSL fix-om
- ✅ Dan 5: Profilisanje i optimizacija - automatski izbor najboljih postavki
- ✅ Dan 6: Resilience sistem - retry, circuit breaker, fallback i graceful degradation
- ✅ Dan 7: Napredna personalizacija - profili, preference i adaptivno učenje
- ✅ Dan 8: Uvod u FastAPI - Učitelj Vasa je sada web servis!
- ✅ Dan 9: Multi-provider web endpoint-i - transparentnost i monitoring
- ✅ Dan 10: Strukturirani zahtevi i inteligentno rutiranje
- ✅ Dan 11: Pydantic validacija i type safety! 🛡️
- ⏳ Dan 12: Naprednije testiranje multi-provider API-ja (sutra)
## 🛡️ Validacija i Type Safety
### Pydantic modeli:
- `SimpleQuestionRequest` - Osnovno pitanje
- `StructuredQuestionRequest` - Pitanje sa kontekstom
- `QuestionResponse` - Standardizovan odgovor
- `ErrorResponse` - Standardizovane greške
### Provider-specifični modeli:
- `OpenAISpecificRequest` - OpenAI opcije
- `GeminiSpecificRequest` - Gemini opcije
- Validacija svih parametara
### Novi endpoint-i:
- `POST /validate-request` - Validira bez slanja
- Provider-specific pozivi sa custom opcijama
### Test UI:
- http://localhost:8000/static/validation_test.html
Dodaj u docs/development_log.md:
## Dan 11: Validacija sa Pydantic i provider-specific modeli (21.06.2025)
### Šta je urađeno:
- ✅ Instaliran i konfigurisan Pydantic
- ✅ Kreirani modeli za sve tipove zahteva
- ✅ Response modeli sa potpunim type safety
- ✅ Provider-specifični modeli (OpenAI, Gemini)
- ✅ Validacija konteksta i preferencija
- ✅ /validate-request endpoint
- ✅ Validated service wrapper-i
- ✅ Interaktivni test UI
- ✅ Custom validatori za biznis logiku
### Naučene lekcije:
- Pydantic drastično poboljšava sigurnost API-ja
- Automatska validacija štedi mnogo koda
- Type hints omogućavaju bolji IDE support
- Response modeli garantuju konzistentnost
- FastAPI + Pydantic = automatska dokumentacija
### Problemi i rešenja:
- **Problem**: Kako validirati enum vrednosti?
- **Rešenje**: use_enum_values = True u Config
- **Problem**: Nested modeli i validacija
- **Rešenje**: root_validator za cross-field validaciju
### Testiranje:
- Svi test case-ovi prolaze
- Validacija hvata sve neispravne podatke
- Error poruke su jasne i korisne
- UI omogućava lako testiranje
### Za sutra (Dan 12):
- Naprednije testiranje sa pytest
- Mock-ovanje AI servisa
- Load testing
- Integration testovi
Git commit za danas
git add .
git commit -m "Dan 11: Implementirana Pydantic validacija sa provider-specific modelima i test UI!"
git push
ČESTITAM! 🎉 Učitelj Vasa sada ima profesionalnu validaciju podataka! Implementirao si:
- Type-safe modele za sve podatke
- Automatsku validaciju svih request/response objekata
- Provider-specifične opcije sa validacijom
- Jasne error poruke za sve slučajeve
- Interaktivni UI za testiranje
API je sada sigurniji, lakši za korišćenje i održavanje!
Sutra Učitelj Vasa uči
Sutra ćemo se fokusirati na naprednije testiranje multi-provider API-ja. Naučićeš kako da pišeš unit i integration testove sa pytest, mock-uješ AI servise za brže testiranje, implementiraš load testing sa locust, i kreiraš CI/CD pipeline za automatsko testiranje. Gradimo na današnjoj validaciji da osiguramo kvalitet koda!
📚 REČNIK DANAŠNJE LEKCIJE:
- Pydantic: Python biblioteka za data validation korišćenjem type hints
- BaseModel: Osnovna klasa za sve Pydantic modele
- Field: Funkcija za definisanje dodatnih ograničenja polja
- Validator: Funkcija za custom validaciju pojedinačnih polja
- Root Validator: Validator koji ima pristup svim poljima modela
- Type Hints: Python anotacije tipova (str, int, List[str])
- Runtime Validation: Provera tipova tokom izvršavanja programa
- Serialization: Pretvaranje objekata u JSON format
- Schema: Opis strukture podataka (JSON Schema)
- Union Type: Tip koji može biti jedan od više tipova
- Literal Type: Tip koji može imati samo specifične vrednosti
- Response Model: Pydantic model koji definiše format API odgovora