AI EngineeringAdvanced

Deploy a Model with vLLM on Kubernetes

Azure OpenAI + AI Search + embeddings — real-world architecture for production AI systems, including legacy data, orchestration, hybrid retrieval, cost control, and failure modes.

By Amit UpadhyayPublished 18 FebUpdated 30 Aug 202615 min

Following one scenario: “What’s our parental leave policy in Germany?”An internal knowledge assistant, moving from a 20-person pilot to 2,000 employeesWhy it matters →

What you will build

A small FastAPI gateway in front of the model with per-user budgets, a versioned cache, simple routing and a fallback.

The same scenario as Read and See. You are done when every line below is true. Tick them off as you go.

Done when

A user over budget gets a clear error, and no model call is made
A repeated question is served from the cache until the data version changes
Simple and complex questions go to different models, and the route is logged
A simulated provider failure falls back to the second model

0 of 4 done

0/1 steps

Build Steps

Progress

0/1

◆ Before you start

Prerequisites

Python 3.11+Required
python --version
fastapi, uvicorn, openaiRequired
pip install fastapi uvicorn openai
OPENAI_API_KEYRequired
export OPENAI_API_KEY=<your-api-key>

Step 1 of 1

Expose an LLM as a FastAPI endpoint

Wrap the OpenAI API in a FastAPI service with request validation, streaming support, and basic rate limiting.

Tasks

Define a ChatRequest Pydantic model with model, messages, and stream fields

Create POST /chat endpoint that proxies to OpenAI

Add a GET /health endpoint that returns service status

Run uvicorn and test with curl or the FastAPI docs UI

Add an X-RateLimit-Remaining response header

🎯 What You'll Learn

▸Use Pydantic models for request/response validation
▸Implement Server-Sent Events (SSE) for token streaming
▸Add a simple in-memory rate limiter

Code

Pythonmain.py
1 from fastapi import FastAPI, HTTPException
2 from fastapi.responses import StreamingResponse
3 from pydantic import BaseModel
4 from openai import AsyncOpenAI
5 import asyncio
6  
7 app = FastAPI(title="LLM Serving API")
8 client = AsyncOpenAI()
9  
10 class Message(BaseModel):
11     role: str
12     content: str
13  
14 class ChatRequest(BaseModel):
15     model: str = "gpt-6-luna"
16     messages: list[Message]
17     stream: bool = False
18  
19 @app.get("/health")
20 async def health():
21     return {"status": "ok"}
22  
23 @app.post("/chat")
24 async def chat(req: ChatRequest):
25     messages = [m.model_dump() for m in req.messages]
26  
27     if not req.stream:
28         resp = await client.chat.completions.create(
29             model=req.model, messages=messages
30         )
31         return {"content": resp.choices[0].message.content}
32  
33     async def event_generator():
34         async with await client.chat.completions.create(
35             model=req.model, messages=messages, stream=True
36         ) as stream:
37             async for chunk in stream:
38                 delta = chunk.choices[0].delta.content
39                 if delta:
40                     yield f"data: {delta}\n\n"
41         yield "data: [DONE]\n\n"
42  
43     return StreamingResponse(event_generator(), media_type="text/event-stream")
44

💡 Pro Tips

▸Use StreamingResponse with media_type="text/event-stream" for SSE
▸For production: add auth middleware before this endpoint

Evaluate it

Prove it works before you call it done

Working once on your machine is a demo. Measure it against a fixed set of cases.

Replay 200 realistic questions, including 60 repeats, then simulate a document update and a provider outage. Report cost, cache hits, routes and errors.

Pass bar

No stale cache hits after the update, and no user-visible errors during the outage

Ship and observe

What changes when it is live

Before it ships, in order

Make every app call models through the gateway, and remove direct provider keys from apps
Set per-app budgets and alerts before announcing the launch

Put on a dashboard

Cost per app and route
Fresh cache hit rate
Fallback activations

Break it on purpose

Learn the failure modes while they are cheap

Make one change at a time, predict what will happen, then run it.

ChangeWhat you should see

Remove the data version from cache keys and update a documentThe old answer keeps being served for the updated question.

Route everything to the small modelCost drops and some complex answers get noticeably worse in the eval.

Make the primary provider always fail with no fallbackEvery question errors. With the fallback, answers continue.

What this build costs

Replaying 200 questions costs under $1 with routing in place.

Where this leads

LLM monitoring

The gateway produces the data. Monitoring turns it into alerts and dashboards before users or finance notice a problem.

Next topic →

Azure-OpenAI AI-Search production RAG architecture

← PreviousToken Economics: Understanding and Optimizing LLM Costs Next →Embeddings: From Theory to Production Choice

Continue reading

All AI Engineering articles →

AI Engineering13 min

Guardrails: Input/Output Safety in Production

Input, output, schema, tool, and cost guardrails — the layered defence system that catches LLM failures before they reach a user or downstream system.

AI Engineering11 min

Embeddings: From Theory to Production Choice

Convert text, images, and code into dense vectors that capture semantic meaning — the foundational primitive of modern AI systems, and the choice that dominates RAG quality.

AI Engineering10 min

Token Economics: Understanding and Optimizing LLM Costs

A practical guide to understanding token pricing, measuring real costs, and implementing optimization strategies — caching, prompt compression, model routing.

Discussion

A small FastAPI gateway in front of the model with per-user budgets, a versioned cache, simple routing and a fallback.

The same scenario as Read and See. You are done when every line below is true. Tick them off as you go.

Done when

A user over budget gets a clear error, and no model call is made
A repeated question is served from the cache until the data version changes
Simple and complex questions go to different models, and the route is logged
A simulated provider failure falls back to the second model

0 of 4 done

1 from fastapi import FastAPI, HTTPException 2 from fastapi.responses import StreamingResponse 3 from pydantic import BaseModel 4 from openai import AsyncOpenAI 5 import asyncio 6 7 app = FastAPI(title="LLM Serving API") 8 client = AsyncOpenAI() 9 10 class Message(BaseModel): 11 role: str 12 content: str 13 14 class ChatRequest(BaseModel): 15 model: str = "gpt-6-luna" 16 messages: list[Message] 17 stream: bool = False 18 19 @app.get("/health") 20 async def health(): 21 return {"status": "ok"} 22 23 @app.post("/chat") 24 async def chat(req: ChatRequest): 25 messages = [m.model_dump() for m in req.messages] 26 27 if not req.stream: 28 resp = await client.chat.completions.create( 29 model=req.model, messages=messages 30 ) 31 return {"content": resp.choices[0].message.content} 32 33 async def event_generator(): 34 async with await client.chat.completions.create( 35 model=req.model, messages=messages, stream=True 36 ) as stream: 37 async for chunk in stream: 38 delta = chunk.choices[0].delta.content 39 if delta: 40 yield f"data: {delta}\n\n" 41 yield "data: [DONE]\n\n" 42 43 return StreamingResponse(event_generator(), media_type="text/event-stream") 44

Prove it works before you call it done

Working once on your machine is a demo. Measure it against a fixed set of cases.

Replay 200 realistic questions, including 60 repeats, then simulate a document update and a provider outage. Report cost, cache hits, routes and errors.

Pass bar

No stale cache hits after the update, and no user-visible errors during the outage

Learn the failure modes while they are cheap

Make one change at a time, predict what will happen, then run it.

ChangeWhat you should see

Remove the data version from cache keys and update a documentThe old answer keeps being served for the updated question.

Route everything to the small modelCost drops and some complex answers get noticeably worse in the eval.

Make the primary provider always fail with no fallbackEvery question errors. With the fallback, answers continue.

1	from fastapi import FastAPI, HTTPException
2	from fastapi.responses import StreamingResponse
3	from pydantic import BaseModel
4	from openai import AsyncOpenAI
5	import asyncio
6
7	app = FastAPI(title="LLM Serving API")
8	client = AsyncOpenAI()
9
10	class Message(BaseModel):
11	role: str
12	content: str
13
14	class ChatRequest(BaseModel):
15	model: str = "gpt-6-luna"
16	messages: list[Message]
17	stream: bool = False
18
19	@app.get("/health")
20	async def health():
21	return {"status": "ok"}
22
23	@app.post("/chat")
24	async def chat(req: ChatRequest):
25	messages = [m.model_dump() for m in req.messages]
26
27	if not req.stream:
28	resp = await client.chat.completions.create(
29	model=req.model, messages=messages
30	)
31	return {"content": resp.choices[0].message.content}
32
33	async def event_generator():
34	async with await client.chat.completions.create(
35	model=req.model, messages=messages, stream=True
36	) as stream:
37	async for chunk in stream:
38	delta = chunk.choices[0].delta.content
39	if delta:
40	yield f"data: {delta}\n\n"
41	yield "data: [DONE]\n\n"
42
43	return StreamingResponse(event_generator(), media_type="text/event-stream")
44