Lesson 06: Database Models#

"Your database schema is your application's DNA."

📚 Table of Contents#

1. Introduction
2. What is models.py?
3. SQLAlchemy ORM Basics
4. Model 1: Agent
5. Model 2: Message
6. Model 3: AgentRun
7. Model 4: Task
8. Model 5: ToolCall
9. Model 6: Document
10. Model 7: DocumentChunk
11. Model 8: WorkflowRun
12. Model 9: WorkflowStep
13. Relationships Explained
14. Cascade Deletes
15. Key Takeaways
16. Interview Questions
17. Exercises

Introduction#

models.py defines your database schema using SQLAlchemy ORM. Every model class becomes a table, every attribute becomes a column.

In this lesson, you'll learn:

• All 9 database models in VeyraOps AI
• Every column and its purpose
• All relationships and foreign keys
• Why the schema is designed this way
• How to query and modify data

What is models.py?#

Location: backend/app/models.py
Size: 6,359 bytes
Tables: 9 models
Purpose: Define database schema

The 9 Models#

1. Agent - AI agent configurations
2. Message - Chat conversation history
3. AgentRun - Execution tracking and metrics
4. Task - Generated action items
5. ToolCall - LLM operation logging
6. Document - Uploaded file metadata
7. DocumentChunk - Document segments for RAG
8. WorkflowRun - Multi-agent workflow tracking
9. WorkflowStep - Individual workflow operations

SQLAlchemy ORM Basics#

What is ORM?#

ORM = Object-Relational Mapping

Without ORM (Raw SQL):

cursor.execute("SELECT * FROM agents WHERE id = ?", (1,))
row = cursor.fetchone()
agent_id = row[0]
agent_name = row[1]

With ORM (SQLAlchemy):

agent = db.query(Agent).filter(Agent.id == 1).first()
agent_name = agent.name  # Python object!

Model Structure#

class Agent(Base):                    # Inherit from Base
    __tablename__ = "agents"          # Table name in database
    
    id = Column(Integer, primary_key=True)  # Column definition
    name = Column(String(100))              # Type and constraints
    
    messages = relationship("Message")      # Foreign key relationship

Model 1: Agent#

class Agent(Base):  
    __tablename__ = "agents"
    
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(100), nullable=False)
    type = Column(String(50), nullable=False)
    description = Column(Text, nullable=True)
    system_prompt = Column(Text, nullable=False)
    language = Column(String(50), default="English")
    tone = Column(String(50), default="Professional")
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    enabled_tools = Column(Text, default="")
    
    messages = relationship("Message", back_populates="agent", cascade="all, delete")
    runs = relationship("AgentRun", back_populates="agent", cascade="all, delete")
    tasks = relationship("Task", cascade="all, delete")
    tool_calls = relationship("ToolCall", cascade="all, delete")
    documents = relationship("Document", cascade="all, delete")
    workflow_runs = relationship("WorkflowRun", cascade="all, delete")

Column Breakdown#

id - Primary key

• Integer: Numeric type
• primary_key=True: Unique identifier
• index=True: Faster lookups
• Auto-increments: 1, 2, 3...

name - Agent display name

• String(100): Max 100 characters
• nullable=False: Required field
• Example: "Customer Support Agent"

type - Agent category

• String(50): Max 50 characters
• nullable=False: Required
• Example: "customer_support", "sales", "technical"

description - Optional details

• Text: Unlimited length
• nullable=True: Optional
• Example: "Handles customer inquiries..."

system_prompt - Core AI instructions

• Text: Can be very long
• nullable=False: Required
• Example: "You are a helpful customer support agent who..."

language - Response language

• String(50)
• default="English": If not provided
• Example: "English", "Spanish", "French"

tone - Communication style

• String(50)
• default="Professional"
• Example: "Professional", "Friendly", "Technical"

created_at - Creation timestamp

• DateTime(timezone=True): Timezone-aware
• server_default=func.now(): Database sets this automatically
• Example: 2026-07-17 17:02:43+00:00

enabled_tools - Available capabilities

• Text: Comma-separated list
• default="": Empty by default
• Example: "email_analyzer,task_creator,document_search"

Relationships#

messages - All chat messages

• Links to Message model
• back_populates="agent": Bidirectional
• cascade="all, delete": Delete messages when agent deleted

runs - All executions

• Links to AgentRun model
• Tracks performance metrics

tasks - All generated tasks

• Links to Task model

tool_calls - All LLM operations

• Links to ToolCall model
• For observability

documents - All uploaded files

• Links to Document model

workflow_runs - All multi-agent workflows

• Links to WorkflowRun model

Model 2: Message#

class Message(Base):
    __tablename__ = "messages"
 
    id = Column(Integer, primary_key=True, index=True)
    agent_id = Column(Integer, ForeignKey("agents.id"), nullable=False)
    role = Column(String(20), nullable=False)  # user or assistant
    content = Column(Text, nullable=False)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
 
    agent = relationship("Agent", back_populates="messages")

Purpose#

Store conversation history between users and agents.

Column Breakdown#

agent_id - Foreign key

• ForeignKey("agents.id"): References agents table
• Links message to its agent
• Why needed: Know which agent sent/received message

role - Message sender

• String(20)
• Values: "user" or "assistant"
• user: Human message
• assistant: AI response

content - Message text

• Text: Can be long
• The actual message content

Example Rows#

Model 3: AgentRun#

class AgentRun(Base):
    __tablename__ = "agent_runs"
 
    id = Column(Integer, primary_key=True, index=True)
    agent_id = Column(Integer, ForeignKey("agents.id"), nullable=False)
    input_text = Column(Text, nullable=False)
    output_text = Column(Text, nullable=True)
    latency_ms = Column(Integer, nullable=True)
    status = Column(String(30), default="success")
    error_message = Column(Text, nullable=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
 
    agent = relationship("Agent", back_populates="runs")
    tool_calls = relationship("ToolCall", cascade="all, delete")

Purpose#

Track every agent execution with metrics and status.

Column Breakdown#

input_text - User's message output_text - Agent's response latency_ms - Response time in milliseconds status - Success or failure

• Values: "success", "failed" error_message - If failed, why?

Example Usage#

run = AgentRun(
    agent_id=1,
    input_text="Hello",
    output_text="Hi! How can I help?",
    latency_ms=850,
    status="success"
)
db.add(run)
db.commit()

Model 4: Task#

class Task(Base):
    __tablename__ = "tasks"
 
    id = Column(Integer, primary_key=True, index=True)
    agent_id = Column(Integer, ForeignKey("agents.id"), nullable=False)
    title = Column(String(200), nullable=False)
    description = Column(Text, nullable=True)
    priority = Column(String(20), default="medium")
    status = Column(String(30), default="open")
    due_date = Column(String(100), nullable=True)
    source_type = Column(String(50), default="email")
    created_at = Column(DateTime(timezone=True), server_default=func.now())
 
    agent = relationship("Agent")

Purpose#

Store action items extracted from emails or created manually.

Column Breakdown#

title - Task summary

• Max 200 characters
• Example: "Fix database backup issue"

priority - Urgency level

• Values: "low", "medium", "high", "urgent"
• Default: "medium"

status - Current state

• Values: "open", "in_progress", "completed"
• Default: "open"

due_date - Deadline

• Stored as string (flexible format)
• Example: "Friday", "2026-07-20", "next week"

source_type - Where it came from

• Default: "email"
• Could be: "manual", "workflow", etc.

Model 5: ToolCall#

class ToolCall(Base):
    __tablename__ = "tool_calls"
 
    id = Column(Integer, primary_key=True, index=True)
    agent_id = Column(Integer, ForeignKey("agents.id"), nullable=False)
    agent_run_id = Column(Integer, ForeignKey("agent_runs.id"), nullable=True)
    tool_name = Column(String(100), nullable=False)
    tool_input = Column(Text, nullable=True)
    tool_output = Column(Text, nullable=True)
    success = Column(String(10), default="true")
    error_message = Column(Text, nullable=True)
    latency_ms = Column(Integer, nullable=True)
    model_name = Column(String(100), nullable=True)
    estimated_tokens = Column(Integer, nullable=True)
    estimated_cost = Column(String(30), nullable=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
 
    agent = relationship("Agent")
    agent_run = relationship("AgentRun")

Purpose#

Observability - Track every LLM operation with cost and performance.

Why This Exists#

Answers critical questions:

• How much are we spending on AI?
• Which operations are slow?
• What prompts are we using?
• Which models are being called?

Column Breakdown#

tool_name - What operation

• Example: "openai_chat", "email_analyzer", "document_search"

tool_input - Input data

• The prompt or query sent

tool_output - Result

• The response received

success - Did it work?

• Values: "true" or "false" (stored as string)

latency_ms - How long it took

model_name - Which AI model

• Example: "gpt-4o-mini", "gpt-4"

estimated_tokens - Token count

• Approximate (1 token ≈ 4 characters)

estimated_cost - Dollar amount

• Example: "0.000150" ($0.00015)

Model 6: Document#

class Document(Base):
    __tablename__ = "documents"
 
    id = Column(Integer, primary_key=True, index=True)
    agent_id = Column(Integer, ForeignKey("agents.id"), nullable=False)
    filename = Column(String(255), nullable=False)
    file_type = Column(String(50), default="txt")
    status = Column(String(50), default="uploaded")
    chunks_count = Column(Integer, default=0)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
 
    agent = relationship("Agent")
    chunks = relationship("DocumentChunk", cascade="all, delete")

Purpose#

Track uploaded documents for RAG queries.

Column Breakdown#

filename - Original file name

• Example: "company_policies.txt"

file_type - File extension

• Example: "txt", "pdf", "docx"

status - Processing state

• Values: "uploaded", "processing", "indexed", "failed"

chunks_count - How many segments

• Documents are split into chunks for better retrieval

Model 7: DocumentChunk#

class DocumentChunk(Base):
    __tablename__ = "document_chunks"
 
    id = Column(Integer, primary_key=True, index=True)
    document_id = Column(Integer, ForeignKey("documents.id"), nullable=False)
    agent_id = Column(Integer, ForeignKey("agents.id"), nullable=False)
    chunk_text = Column(Text, nullable=False)
    chunk_index = Column(Integer, nullable=False)
    vector_id = Column(String(255), nullable=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
 
    document = relationship("Document")
    agent = relationship("Agent")

Purpose#

Store document segments for semantic search.

Why Chunk Documents?#

Problem: LLMs have context limits (4K-128K tokens)

Solution: Split documents into smaller pieces

Example:

Document: "Company Policy Handbook.txt" (50,000 words)
↓ Split into chunks
Chunk 1: "Vacation Policy: Employees receive..." (800 words)
Chunk 2: "Sick Leave: In case of illness..." (800 words)
Chunk 3: "Remote Work: Employees may work..." (800 words)
...

Column Breakdown#

chunk_text - The actual text segment

chunk_index - Position in document

• 0, 1, 2, 3...
• Preserves order

vector_id - ChromaDB identifier

• Links SQL record to vector database
• Format: "agent_1_document_5_chunk_3"

Model 8: WorkflowRun#

class WorkflowRun(Base):
    __tablename__ = "workflow_runs"
 
    id = Column(Integer, primary_key=True, index=True)
    agent_id = Column(Integer, ForeignKey("agents.id"), nullable=False)
    workflow_name = Column(String(100), nullable=False)
    input_text = Column(Text, nullable=False)
    final_output = Column(Text, nullable=True)
    status = Column(String(30), default="success")
    quality_score = Column(String(20), nullable=True)
    error_message = Column(Text, nullable=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
 
    agent = relationship("Agent")
    steps = relationship("WorkflowStep", cascade="all, delete")

Purpose#

Track multi-agent workflows (e.g., email analysis → reply → review).

Column Breakdown#

workflow_name - Which workflow

• Example: "multi_agent_email_workflow"

input_text - Starting data

• The email being analyzed

final_output - End result

• JSON with analysis, tasks, reply, review

quality_score - Review score

• Example: "0.85" (85% quality)

Model 9: WorkflowStep#

class WorkflowStep(Base):
    __tablename__ = "workflow_steps"
 
    id = Column(Integer, primary_key=True, index=True)
    workflow_run_id = Column(Integer, ForeignKey("workflow_runs.id"), nullable=False)
    step_name = Column(String(100), nullable=False)
    input_data = Column(Text, nullable=True)
    output_data = Column(Text, nullable=True)
    status = Column(String(30), default="success")
    latency_ms = Column(Integer, nullable=True)
    error_message = Column(Text, nullable=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
 
    workflow_run = relationship("WorkflowRun")

Purpose#

Track individual steps within a workflow.

Example Steps#

For email workflow:

1. "analysis_agent" - Extract data
2. "task_agent" - Create tasks
3. "reply_agent" - Generate reply
4. "reviewer_agent" - Check quality

Each step tracked separately with its own metrics.

Relationships Explained#

One-to-Many#

One Agent → Many Messages

# Agent model
messages = relationship("Message", back_populates="agent")
 
# Message model
agent = relationship("Agent", back_populates="messages")

Usage:

agent = db.query(Agent).first()
for message in agent.messages:
    print(message.content)

Cascade Deletes#

messages = relationship("Message", cascade="all, delete")

What it means: When you delete an agent, all its messages are automatically deleted.

Example:

agent = db.query(Agent).filter(Agent.id == 1).first()
db.delete(agent)
db.commit()
# All messages with agent_id=1 are also deleted

Key Takeaways#

🎯 9 models define the schema: Agent, Message, AgentRun, Task, ToolCall, Document, DocumentChunk, WorkflowRun, WorkflowStep

🎯 Relationships connect tables through foreign keys and SQLAlchemy relationships

🎯 Cascade deletes ensure referential integrity when deleting parent records

🎯 Observability is built-in through ToolCall, AgentRun, and WorkflowStep tracking

🎯 RAG is enabled through Document and DocumentChunk models

🎯 Every timestamp is timezone-aware with DateTime(timezone=True)

Interview Questions#

Junior Level#

Q1: What is the purpose of primary_key=True?
A: Makes the column a unique identifier for each row.

Q2: What does ForeignKey("agents.id") mean?
A: References the id column in the agents table, creating a relationship.

Q3: Why do we need both Document and DocumentChunk models?
A: Documents are split into chunks for better RAG retrieval within context limits.

Mid-Level#

Q4: Explain cascade delete with an example.
A: When cascade="all, delete" is set, deleting a parent automatically deletes children. Example: Deleting an Agent deletes all its Messages.

Q5: Why is estimated_cost a String instead of Float?
A: To avoid floating-point precision issues with currency. Stored as string like "0.000150".

Q6: What's the difference between AgentRun and ToolCall?
A: AgentRun tracks complete user interactions. ToolCall tracks individual LLM operations (one AgentRun can have multiple ToolCalls).

Senior Level#

Q7: Design an indexing strategy for this schema to optimize common queries.
A: Add compound indexes: (agent_id, created_at) on messages/runs, (agent_id, status) on tasks, (document_id, chunk_index) on chunks.

Q8: How would you implement soft deletes instead of cascade deletes?
A: Add deleted_at column (nullable DateTime). Override delete() to set timestamp instead. Filter queries with filter(deleted_at == None).

Q9: This schema stores quality_score as String. Redesign for analytics.
A: Create WorkflowMetrics table with Float columns for scores, separate from operational data. Use materialized views for aggregations.

Exercises#

Exercise 1: Query Practice#

Write queries to:

• Get all messages for agent_id=1
• Count tasks by status
• Find slowest ToolCalls (top 10)

Exercise 2: Add Column#

Add priority_score (Integer) to Task model and create a migration.

Exercise 3: New Model#

Design a User model for authentication with email, password_hash, and created_at.

Exercise 4: Relationship#

Add a many-to-many relationship: Agents can have multiple Tags, Tags can apply to multiple Agents.

Exercise 5: Cascade Analysis#

Draw a diagram showing what gets deleted when you delete an Agent.

Next Steps#

You now understand every database model in VeyraOps AI.

Next: Pydantic Schemas - validation and serialization.

👉 Lesson 07: Schemas →

← Previous: Database | ↑ Index | Next: Schemas →

Module 2 · Data Layer