Lesson 03: Request Lifecycle#

"Understanding the journey of a request is understanding the system itself."

📚 Table of Contents#

1. Introduction
2. What is a Request Lifecycle?
3. The Complete Journey
4. Step-by-Step Walkthrough
5. Example 1: Simple Chat Request
6. Example 2: Email Analysis with Tasks
7. Example 3: RAG Query
8. Error Handling in the Lifecycle
9. Performance Considerations
10. Key Takeaways
11. Interview Questions
12. Exercises
13. Next Steps

Introduction#

In Lessons 01 and 02, you learned what VeyraOps AI is and how it's architected. Now you'll learn what actually happens when a user makes a request.

Every button click, every API call, every database query is part of a request lifecycle — the complete journey from client to server and back.

Understanding this lifecycle is critical because:

• Debugging: Know where things can go wrong
• Performance: Identify bottlenecks
• Architecture: See how components interact
• Monitoring: Track what matters

In this lesson, you'll trace a request through every layer, every function, every line of code that executes.

What is a Request Lifecycle?#

A request lifecycle is the complete sequence of events from when a request enters the system until a response is returned.

The Basic Flow#

The Reality#

The Complete Journey#

Let's map out all the stages a request passes through:

Stage 1: Client Initiation#

• User triggers action (button click, form submit)
• Frontend constructs HTTP request
• Request sent over network

Stage 2: Server Reception#

• FastAPI receives request
• Routes to correct endpoint
• Begins processing

Stage 3: Validation#

• Pydantic validates request body
• Type checking
• Required field verification

Stage 4: Dependency Injection#

• FastAPI resolves dependencies
• Database session created
• Other dependencies injected

Stage 5: Business Logic#

• Router calls service functions
• Services orchestrate operations
• External APIs called (OpenAI, etc.)

Stage 6: Data Persistence#

• ORM queries executed
• Database writes performed
• Relationships updated

Stage 7: Response Construction#

• Data formatted
• Pydantic serialization
• HTTP response prepared

Stage 8: Client Reception#

• Response returned to client
• Frontend processes data
• UI updates

Step-by-Step Walkthrough#

Let's trace a real request with actual code:

Example Request#

POST /agents/1/chat HTTP/1.1
Host: localhost:8001
Content-Type: application/json
 
{
  "message": "What is machine learning?"
}

Step 1: FastAPI Receives Request#

# FastAPI framework automatically:
# 1. Parses HTTP headers
# 2. Reads request body
# 3. Routes to correct endpoint
 
@router.post("/{agent_id}/chat", response_model=schemas.ChatResponse)
def chat_with_agent(
    agent_id: int,                      # ← Extracted from URL path
    chat_data: schemas.ChatRequest,     # ← Parsed from request body
    db: Session = Depends(get_db)       # ← Dependency injection
):
    # Function starts executing...

What happens here:

• agent_id extracted from URL: /agents/1/chat → agent_id=1
• chat_data parsed from JSON: {"message": "..."} → ChatRequest object
• db injected by FastAPI using get_db() dependency

Step 2: Pydantic Validation#

# Before the function executes, Pydantic validates:
class ChatRequest(BaseModel):
    message: str  # Must be a string, cannot be empty
 
# If validation fails:
# {
#   "message": 123  ← Wrong type!
# }
# FastAPI returns 422 Unprocessable Entity automatically

What Pydantic checks:

• ✅ message is present
• ✅ message is a string
• ✅ message is not null

If validation fails:

{
  "detail": [
    {
      "loc": ["body", "message"],
      "msg": "str type expected",
      "type": "type_error.str"
    }
  ]
}

Step 3: Database Session Creation#

def get_db():
    db = SessionLocal()  # Create new session
    try:
        yield db          # Provide to endpoint
    finally:
        db.close()       # Always close, even if error

Why this matters:

• Each request gets its own session (thread-safe)
• Session automatically closes (no memory leaks)
• Transactions managed properly

Step 4: Query Database for Agent#

agent = db.query(models.Agent).filter(models.Agent.id == agent_id).first()

What happens under the hood:

1. SQLAlchemy generates SQL:

SELECT * FROM agents WHERE id = 1 LIMIT 1;

2. SQLite executes query:

• Searches agents table
• Finds row with id=1
• Returns data

3. SQLAlchemy creates Python object:

Agent(
    id=1,
    name="Support Agent",
    type="customer_support",
    system_prompt="You are a helpful assistant...",
    language="English",
    tone="Professional"
)

Step 5: Error Checking#

if agent is None:
    raise HTTPException(status_code=404, detail="Agent not found")

If agent doesn't exist:

• HTTPException raised
• FastAPI catches it
• Returns 404 response immediately
• Rest of function doesn't execute

Step 6: Save User Message#

user_message = models.Message(
    agent_id=agent.id,
    role="user",
    content=chat_data.message
)
db.add(user_message)

What happens:

• Creates Message object in memory
• db.add() stages it for insertion
• NOT saved to database yet (waits for commit)

Database state: PENDING

Step 7: Call Service Layer (Business Logic)#

assistant_text = generate_agent_response(
    system_prompt=agent.system_prompt,
    user_message=chat_data.message,
    agent_name=agent.name,
    language=agent.language,
    tone=agent.tone
)

Service function executes:

# services/openai_service.py
def generate_agent_response(...):
    # 1. Construct prompt
    final_prompt = f"""
You are {agent_name}.
Instructions: {system_prompt}
Language: {language}
Tone: {tone}
"""
    
    # 2. Call OpenAI API
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": final_prompt},
            {"role": "user", "content": user_message}
        ],
        temperature=0.3
    )
    
    # 3. Extract and return response
    return response.choices[0].message.content

External API call:

• HTTP request to api.openai.com
• Waits for LLM generation (500-3000ms typically)
• Receives JSON response
• Extracts text content

Step 8: Save Assistant Message#

assistant_message = models.Message(
    agent_id=agent.id,
    role="assistant",
    content=assistant_text
)
db.add(assistant_message)

Database state: PENDING (2 messages waiting)

Step 9: Calculate Metrics#

latency_ms = int((time.time() - start_time) * 1000)

Tracks request duration:

• Measures time from request start to now
• Converts seconds to milliseconds
• Used for monitoring

Step 10: Save Agent Run#

run = models.AgentRun(
    agent_id=agent.id,
    input_text=chat_data.message,
    output_text=assistant_text,
    latency_ms=latency_ms,
    status="success"
)
db.add(run)

Database state: PENDING (2 messages + 1 run waiting)

Step 11: Commit Transaction#

db.commit()

What happens:

• All staged objects saved to database in single transaction
• If any fails, entire transaction rolls back
• SQLite file updated

SQL executed:

BEGIN TRANSACTION;
 
INSERT INTO messages (agent_id, role, content, created_at)
VALUES (1, 'user', 'What is machine learning?', '2026-07-17 16:35:00');
 
INSERT INTO messages (agent_id, role, content, created_at)
VALUES (1, 'assistant', 'Machine learning is...', '2026-07-17 16:35:02');
 
INSERT INTO agent_runs (agent_id, input_text, output_text, latency_ms, status, created_at)
VALUES (1, 'What is machine learning?', 'Machine learning is...', 2150, 'success', '2026-07-17 16:35:02');
 
COMMIT;

Step 12: Refresh Objects#

db.refresh(run)

Why?

• Database generated run.id (auto-increment)
• Database set run.created_at (server_default)
• Refresh loads these values back into Python object

Step 13: Construct Response#

return {
    "agent_id": agent.id,
    "user_message": chat_data.message,
    "assistant_response": assistant_text,
    "run_id": run.id,
    "latency_ms": latency_ms
}

Python dictionary created

Step 14: Pydantic Serialization#

# FastAPI sees response_model=schemas.ChatResponse
class ChatResponse(BaseModel):
    agent_id: int
    user_message: str
    assistant_response: str
    run_id: int
    latency_ms: int

Pydantic:

• Validates response matches schema
• Converts to JSON
• Sets HTTP headers (Content-Type: application/json)

Step 15: HTTP Response Sent#

HTTP/1.1 200 OK
Content-Type: application/json
Content-Length: 234
 
{
  "agent_id": 1,
  "user_message": "What is machine learning?",
  "assistant_response": "Machine learning is a branch of AI...",
  "run_id": 42,
  "latency_ms": 2150
}

Step 16: Database Session Cleanup#

# FastAPI automatically calls:
finally:
    db.close()

Session closed, resources freed

Example 1: Simple Chat Request#

Complete Timeline#

Total: 1,872ms (most time in OpenAI call)

Example 2: Email Analysis with Tasks#

Request#

POST /agents/1/email/analyze
{
  "email_text": "Need database fixed by Friday..."
}

Extended Lifecycle#

Additional Steps#

Step A: Tool Verification

if not agent_has_tool(agent, "email_analyzer"):
    raise HTTPException(403, "Tool not enabled")

Step B: Email Analysis

result = analyze_email(email_text)
# Returns: {summary, intent, priority, deadline, action_items, suggested_reply}

Step C: Log Tool Call

log_tool_call(
    db=db,
    agent_id=agent.id,
    tool_name="email_analyzer",
    tool_input=email_text,
    tool_output=json.dumps(result),
    success=True,
    latency_ms=email_latency
)

Step D: Conditional Task Creation

if agent_has_tool(agent, "task_creator"):
    for item in result["action_items"]:
        task = models.Task(...)
        db.add(task)
        created_tasks.append(task)

Step E: Second Tool Call Log

log_tool_call(
    db=db,
    tool_name="task_creator",
    tool_input=json.dumps(result["action_items"]),
    tool_output=json.dumps([task.title for task in created_tasks])
)

Example 3: RAG Query#

Request#

POST /agents/1/documents/ask
{
  "question": "What is the vacation policy?",
  "top_k": 3
}

RAG Lifecycle#

LangGraph State Machine#

# State flows through nodes:
initial_state = {
    "agent_id": 1,
    "question": "What is the vacation policy?",
    "top_k": 3,
    "retrieved_chunks": [],
    "route": ""
}
 
# Node 1: Retrieve
state = retrieve_node(state)
# state["retrieved_chunks"] = [chunk1, chunk2, chunk3]
 
# Node 2: Evaluate
state = evaluate_node(state)
# state["route"] = "answer" or "refuse"
 
# Node 3: Conditional routing
if state["route"] == "answer":
    state = answer_node(state)
else:
    state = refusal_node(state)
 
# state["answer"] = "According to policy..."

Error Handling in the Lifecycle#

Where Errors Can Occur#

1. Validation Errors (Pydantic)#

# Request: {"message": 123}  ← Wrong type
# Pydantic catches before function executes
# Returns: 422 Unprocessable Entity

2. Database Errors (SQLAlchemy)#

agent = db.query(Agent).filter(Agent.id == 999).first()
if agent is None:
    raise HTTPException(404, "Agent not found")

3. External API Errors (OpenAI)#

try:
    response = client.chat.completions.create(...)
except Exception as e:
    status = "failed"
    error_message = str(e)
    # Still save run with error status

4. Business Logic Errors#

if not agent_has_tool(agent, "email_analyzer"):
    raise HTTPException(403, "Tool not enabled")

Error Recovery Pattern#

try:
    # Business logic
    result = perform_operation()
    status = "success"
    error_message = None
except Exception as e:
    # Graceful degradation
    result = "Sorry, operation failed"
    status = "failed"
    error_message = str(e)
finally:
    # Always log
    run = AgentRun(status=status, error_message=error_message)
    db.add(run)
    db.commit()

Performance Considerations#

Bottleneck Analysis#

# Typical latency breakdown for chat request:
 
Database Query:         10ms   (0.5%)
Service Logic:           5ms   (0.25%)
OpenAI API Call:     1,800ms  (98%)
Database Save:          10ms   (0.5%)
Serialization:           5ms   (0.25%)
────────────────────────────
Total:              1,830ms   (100%)

Insight: 98% of time is waiting for OpenAI

Optimization Strategies#

1. Caching#

# Cache frequent prompts
@lru_cache(maxsize=100)
def generate_agent_response(system_prompt, user_message):
    ...

2. Async Processing#

# Don't make client wait for logging
background_tasks.add_task(log_expensive_operation)

3. Database Optimization#

# Add indexes for frequent queries
CREATE INDEX idx_messages_agent_id ON messages(agent_id);

4. Connection Pooling#

# Reuse database connections
engine = create_engine(DATABASE_URL, pool_size=10)

Key Takeaways#

🎯 A request lifecycle spans from client HTTP request to server processing to database operations to external APIs and back.

🎯 FastAPI handles routing, validation, dependency injection, and serialization automatically.

🎯 Pydantic validates requests before your code runs and responses before they're returned.

🎯 Database sessions are created per request and always cleaned up, even on errors.

🎯 Service layer contains business logic isolated from HTTP concerns.

🎯 Most latency comes from external API calls (OpenAI), not internal processing.

🎯 Error handling should be graceful — log failures but still complete the request lifecycle.

🎯 LangGraph workflows add state management for complex multi-step operations.

Interview Questions#

Junior Level#

Q1: What are the main stages of a request lifecycle?
A: Reception → Validation → Business Logic → Data Persistence → Response

Q2: When does Pydantic validation happen?
A: Before the endpoint function executes (request validation) and after it returns (response validation).

Q3: What is the purpose of db.commit()?
A: Saves all staged changes to the database in a single transaction.

Mid-Level#

Q4: Explain what happens between db.add(message) and db.commit().
A: The message is staged in the session (in-memory), but not yet saved to the database. When commit() is called, SQLAlchemy generates INSERT SQL and executes it.

Q5: Why is most request latency from OpenAI and not database operations?
A: OpenAI API requires LLM generation (1-3 seconds), while local database operations are sub-10ms.

Q6: How does FastAPI ensure database sessions are always closed?
A: Using dependency injection with yield in a try-finally block: try: yield db finally: db.close()

Senior Level#

Q7: Design a caching strategy that reduces OpenAI calls without sacrificing response quality.
A: Use Redis to cache (hash of system_prompt + user_message) → response for 24 hours. Invalidate cache when agent's system_prompt changes. For user-specific queries, don't cache. For FAQ-type questions, cache aggressively.

Q8: A request is taking 10 seconds instead of 2 seconds. How do you diagnose the bottleneck?
A: Add timing logs at each stage: (1) before/after DB query, (2) before/after OpenAI call, (3) before/after commit. Most likely culprit is OpenAI timeout or rate limiting. Check OpenAI dashboard for errors.

Q9: How would you implement request tracing across services for debugging?
A: Add request ID to every log entry. Generate UUID at router entry, pass to all service calls, include in database logs. Use structured logging (JSON) for easy parsing. Tools: OpenTelemetry, Datadog APM.

Exercises#

Exercise 1: Trace Your Own Request#

Task: Send a chat request to the backend with detailed logging. Calculate latency at each stage.

Exercise 2: Add Timing Middleware#

Task: Create FastAPI middleware that logs request duration for all endpoints.

Exercise 3: Error Simulation#

Task: Simulate an OpenAI API failure. Verify the request still completes and logs the error.

Exercise 4: Lifecycle Diagram#

Task: Draw the complete request lifecycle for multi-agent workflow from memory.

Exercise 5: Optimization Challenge#

Task: Identify 3 ways to reduce latency in the chat endpoint without changing OpenAI.

Next Steps#

You now understand the complete journey of a request through VeyraOps AI.

Next, you'll explore the folder structure — every file, every directory, and why it's organized this way.

👉 Lesson 04: Folder Structure →

← Previous: Backend Architecture | ↑ Back to Index | Next: Folder Structure →

Module 1 · Foundation

Last Updated: 2026-07-17