GenAI Application Engineering
10 episodes — short audio overviews on genai application engineering.
Chat Completion API with Streaming — chapter audio overview
Build FastAPI streaming chat endpoints using SSE for token-by-token delivery across four LLM providers.
Implement ChatRequest and ChatMessage Pydantic models with SSE frame formatter — lab audio overview
Build the Pydantic data models that define the chat request schema and SSE frame formatting. You will create ChatMessage, ChatRequest, and StreamChunk models with field validators for provider names and temperature ranges, plus an SSE frame formatter that outputs W3C-compliant event-stream strings.
Build async generator streaming endpoint with StreamingResponse — lab audio overview
Implement the core streaming endpoint at POST /api/v1/chat/stream using an async generator that yields SSE-formatted token strings. You will build the stream_tokens() async generator, wire it into a FastAPI StreamingResponse with media_type text/event-stream, and create the FastAPI application with
Add CORS middleware configuration and streaming healthcheck endpoint — lab audio overview
Configure CORSMiddleware on the FastAPI application for browser EventSource clients and implement a GET /api/v1/chat/stream/health endpoint that returns the streaming service status. You will also add startup and shutdown lifespan events for graceful resource management.
Create GeminiStreamAdapter with google.genai.Client initialization — lab audio overview
Build a GeminiStreamAdapter class that initializes google.genai.Client with proxy configuration and defines Pydantic models for streaming chat messages and response chunks.
Implement thinking_config toggle for Gemini 2.5 Flash reasoning budget — lab audio overview
Implement the stream_chat async generator method with ThinkingConfig toggle that controls Gemini 2.5 Flash extended reasoning budget and streams GenerateContentResponse chunks as unified StreamChunk objects.
Add safety rating handling and role normalization logic — lab audio overview
Add safety block detection by checking candidate finish_reason against FinishReason.SAFETY, implement role normalization from Gemini 'model' to 'assistant', and emit StreamError objects when content is blocked.
Create AnthropicStreamAdapter with AsyncAnthropic client setup — lab audio overview
Build the foundational AnthropicStreamAdapter class that wraps anthropic.AsyncAnthropic with proxy-based configuration. Define the ChatMessage, StreamChunk, TokenUsage, and FinishReason models that form the unified SSE interface, and initialize the async client with proper base_url and api_key setti
Implement event-type dispatch for MessageStream delta iteration — lab audio overview
Build the streaming event dispatch logic for the AnthropicStreamAdapter. Implement the stream_chat async generator that calls client.messages.stream() as an async context manager, iterates over MessageStream events, and dispatches message_start, content_block_delta, and message_stop events into unif
Add system prompt extraction and stop_reason mapping to unified enum — lab audio overview
Implement system prompt extraction that separates system-role messages into the Anthropic top-level system parameter, and map Anthropic stop_reason values (end_turn, max_tokens, stop_sequence) to the unified FinishReason enum. Build a message preparation pipeline and a complete stream_chat method th