---
title: "The @pxt.query Decorator: Building Reusable Database Queries for AI Agents and RAG Systems"
date: "2025-10-12"
author: "Pixeltable Team"
tags:
  - Query Patterns
  - @pxt.query
  - Reusable Queries
  - AI Agents
  - RAG Systems
  - Agent Memory
  - Database Optimization
  - Advanced Pixeltable
  - Query Decorator
description: "Transform complex database queries into reusable components with Pixeltable's @pxt.query decorator. Learn how to build efficient agent memory retrieval, multi-table joins, and optimized RAG context queries that integrate seamlessly with AI workflows."
url: "https://pixeltable.com/blog/reusable-query-patterns-pxt-query"
---

# The @pxt.query Decorator: Building Reusable Database Queries for AI Agents and RAG Systems

## The Query Repetition Problem in AI Applications

 
Building [AI agents](/blog/practical-guide-building-agents) and [RAG systems](/blog/production-rag-data-centric) means writing the same queries repeatedly: retrieving conversation history, finding relevant documents, fetching user context. Each time you need this logic, you copy-paste query code, creating duplication, inconsistency, and maintenance nightmares.

 
 
What if you could define these queries once and reuse them throughout your application? What if they could be as clean and composable as functions, while still benefiting from database optimization?

 
 
This is exactly what Pixeltable's `@pxt.query` decorator enables: **reusable, parameterized database queries** that work seamlessly with [multimodal AI workflows](/blog/unified-multimodal-ai-infrastructure-pixeltable).

 
## Understanding @pxt.query: Queries as Functions

 
The `@pxt.query` decorator transforms a Python function into a reusable query component. Instead of writing raw SQL or complex DataFrame operations, you define queries using Pixeltable's familiar table API:

 
```python

import pixeltable as pxt

# Define a table for customer data
customers = pxt.create_table('customers', {
 'customer_id': pxt.String,
 'name': pxt.String,
 'region': pxt.String,
 'signup_date': pxt.Timestamp
})

# Reusable query: Get customers by region
@pxt.query
def get_customers_by_region(region: str, limit: int = 10):
 """Retrieve customers from a specific region"""
 return customers.where(
 customers.region == region
 ).order_by(
 customers.signup_date, asc=False
 ).limit(limit).select(
 customers.customer_id,
 customers.name,
 customers.region
 )

# Use the query anywhere
west_customers = get_customers_by_region('West', limit=5).collect()
east_customers = get_customers_by_region('East', limit=10).collect()

# Or as a computed column in another table
queries = pxt.create_table('analytics.queries', {'target_region': pxt.String})
queries.add_computed_column(
 regional_customers=get_customers_by_region(queries.target_region)
)
 
```

 
## AI Agent Memory: The Killer Use Case

 
The most powerful application of `@pxt.query` is building [stateful AI agents with persistent memory](/blog/building-memory-powered-ai-stateful-agents-pixeltable):

 
### Pattern 1: Conversatio History Retrieval

 
```python

# Agent conversation table
conversations = pxt.create_table('agent.conversations', {
 'session_id': pxt.String,
 'role': pxt.String, # 'user' or 'assistant'
 'message': pxt.String,
 'timestamp': pxt.Timestamp
})

# Reusable query for conversation context
@conversations.query
def get_conversation_history(session_id: str, limit: int = 20):
 """Retrieve recent conversation history for an agent"""
 return conversations.where(
 conversations.session_id == session_id
 ).order_by(
 conversations.timestamp, asc=False
 ).limit(limit).select(
 conversations.role,
 conversations.message,
 conversations.timestamp
 )

# Use in new messages
new_messages = pxt.create_table('agent.new_messages', {
 'session_id': pxt.String,
 'user_message': pxt.String
})

# Automatically retrieve context for every new message
new_messages.add_computed_column(
 conversation_context=get_conversation_history(
 new_messages.session_id,
 limit=10
 )
)

# Build agent response with context
from pixeltable.functions import openai

@pxt.udf
def build_agent_prompt(user_message: str, context: list) -> list:
 """Build prompt with conversation history"""
 messages = []
 
 # Add conversation history
 for msg in reversed(context): # Oldest to newest
 messages.append({
 'role': msg['role'],
 'content': msg['message']
 })
 
 # Add current message
 messages.append({
 'role': 'user',
 'content': user_message
 })
 
 return messages

new_messages.add_computed_column(
 prompt_messages=build_agent_prompt(
 new_messages.user_message,
 new_messages.conversation_context
 )
)

new_messages.add_computed_column(
 agent_response=openai.chat_completions(
 model='gpt-4o',
 messages=new_messages.prompt_messages
 ).choices[0].message.content
)

# Complete agent with automatic memory - powered by @pxt.query
 
```

 
### Pattern 2: Semantic Memory Retrieval

 
Build agents that retrieve relevant past experiences using [embedding-based search](/blog/incremental-embedding-indexes):

 
```python

# Agent memory with embeddings
agent_memory = pxt.create_table('agent.semantic_memory', {
 'memory_id': pxt.String,
 'agent_id': pxt.String,
 'memory_content': pxt.String,
 'memory_type': pxt.String, # 'fact', 'experience', 'skill'
 'importance': pxt.Float,
 'created_at': pxt.Timestamp
})

# Add embeddings for semantic search
from pixeltable.functions import openai

agent_memory.add_computed_column(
 embedding=openai.embeddings(
 agent_memory.memory_content,
 model='text-embedding-3-small'
 )
)

agent_memory.add_embedding_index('memory_content', embedding=agent_memory.embedding)

# Reusable semantic memory search
@agent_memory.query
def search_relevant_memories(
 agent_id: str,
 query: str,
 memory_types: list = None,
 min_importance: float = 0.3,
 limit: int = 5
):
 """Search agent's semantic memory"""
 results = agent_memory.where(
 agent_memory.agent_id == agent_id
 ).where(
 agent_memory.importance >= min_importance
 )
 
 # Filter by memory types if specified
 if memory_types:
 results = results.where(agent_memory.memory_type.in_(memory_types))
 
 # Order by similarity
 results = results.order_by(
 agent_memory.memory_content.similarity(string=query), asc=False
 ).limit(limit)
 
 return results.select(
 agent_memory.memory_content,
 agent_memory.memory_type,
 agent_memory.importance,
 similarity=agent_memory.memory_content.similarity(string=query)
 )

# Use in agent reasoning
agent_queries = pxt.create_table('agent.queries', {
 'agent_id': pxt.String,
 'query': pxt.String
})

# Retrieve relevant memories for each query
agent_queries.add_computed_column(
 relevant_memories=search_relevant_memories(
 agent_queries.agent_id,
 agent_queries.query,
 memory_types=['fact', 'experience'],
 limit=3
 )
)

# Agent can now reason with its past experiences
 
```

 
## RAG System Patterns: Optimized Context Retrieval

 
### Pattern 3: Hybrid Retrieval (Semantic + Metadata)

 
Build sophisticated RAG systems that combine semantic search with metadata filtering:

 
```python

# Document knowledge base
documents = pxt.create_table('knowledge.docs', {
 'document': pxt.Document,
 'title': pxt.String,
 'category': pxt.String,
 'doc_type': pxt.String,
 'created_at': pxt.Timestamp,
 'author': pxt.String
})

# Create chunks
from pixeltable.functions.document import document_splitter

chunks = pxt.create_view('knowledge.chunks', documents,
 iterator=document_splitter(
 document=documents.document,
 separators='sentence'
 ))

# Add embeddings
chunks.add_computed_column(
 embedding=openai.embeddings(chunks.text, model='text-embedding-3-large')
)
chunks.add_embedding_index('text', embedding=chunks.embedding)

# Reusable hybrid retrieval query
@chunks.query
def hybrid_retrieve(
 query: str,
 categories: list = None,
 doc_types: list = None,
 date_from: str = None,
 authors: list = None,
 min_similarity: float = 0.7,
 limit: int = 5
):
 """Hybrid retrieval with semantic search + metadata filters"""
 
 # Start with similarity search
 results = chunks.order_by(
 chunks.text.similarity(string=query), asc=False
 )
 
 # Apply metadata filters
 if categories:
 results = results.where(documents.category.in_(categories))
 
 if doc_types:
 results = results.where(documents.doc_type.in_(doc_types))
 
 if date_from:
 results = results.where(documents.created_at >= date_from)
 
 if authors:
 results = results.where(documents.author.in_(authors))
 
 # Filter by minimum similarity
 results = results.where(
 chunks.text.similarity(string=query) >= min_similarity
 )
 
 # Select relevant fields
 return results.limit(limit).select(
 chunks.text,
 documents.title,
 documents.category,
 documents.author,
 similarity=chunks.text.similarity(string=query)
 )

# Use in RAG pipeline
rag_queries = pxt.create_table('rag.queries', {
 'question': pxt.String,
 'required_categories': pxt.Json,
 'date_filter': pxt.String
})

# Retrieve filtered context
rag_queries.add_computed_column(
 context=hybrid_retrieve(
 rag_queries.question,
 categories=rag_queries.required_categories,
 date_from=rag_queries.date_filter,
 limit=5
 )
)

# Generate answer with filtered context
rag_queries.add_computed_column(
 answer=generate_rag_answer(rag_queries.question, rag_queries.context)
)
 
```

 
### Pattern 4: Multi-Source Knowledge Retrieval

 
Query across multiple data sources for comprehensive [multimodal RAG](/blog/multimodal-rag-production):

 
```python

# Multiple knowledge sources
text_knowledge = pxt.create_table('knowledge.text', {
 'content': pxt.String,
 'source': pxt.String
})

image_knowledge = pxt.create_table('knowledge.images', {
 'image': pxt.Image,
 'description': pxt.String,
 'source': pxt.String
})

video_knowledge = pxt.create_table('knowledge.videos', {
 'video': pxt.Video,
 'transcript': pxt.String,
 'source': pxt.String
})

# Unified retrieval across all sources
@pxt.query
def unified_knowledge_search(query: str, limit_per_source: int = 3):
 """Search across all knowledge modalities"""
 
 # Text knowledge
 text_results = text_knowledge.order_by(
 text_knowledge.content.similarity(string=query), asc=False
 ).limit(limit_per_source).select(
 text_knowledge.content,
 text_knowledge.source,
 modality='text'
 ).collect()
 
 # Image knowledge
 image_results = image_knowledge.order_by(
 image_knowledge.description.similarity(string=query), asc=False
 ).limit(limit_per_source).select(
 image_knowledge.description,
 image_knowledge.source,
 modality='image'
 ).collect()
 
 # Video knowledge
 video_results = video_knowledge.order_by(
 video_knowledge.transcript.similarity(string=query), asc=False
 ).limit(limit_per_source).select(
 video_knowledge.transcript,
 video_knowledge.source,
 modality='video'
 ).collect()
 
 # Combine and return
 return text_results + image_results + video_results

# Use in multimodal agent
agent_queries = pxt.create_table('agent.multimodal_queries', {
 'query': pxt.String
})

agent_queries.add_computed_column(
 comprehensive_context=unified_knowledge_search(
 agent_queries.query,
 limit_per_source=2
 )
)
 
```

 
## Performance Optimization with Query Patterns

 
### Pattern 5: Cached Expensive Queries

 
Build query patterns with intelligent caching for expensive operations:

 
```python

# User analytics with expensive computations
user_interactions = pxt.create_table('analytics.interactions', {
 'user_id': pxt.String,
 'action': pxt.String,
 'timestamp': pxt.Timestamp,
 'metadata': pxt.Json
})

# Expensive aggregation query - cache results
@user_interactions.query 
def get_user_behavior_profile(
 user_id: str,
 days_back: int = 30
) -> dict:
 """Generate comprehensive user behavior profile"""
 from datetime import datetime, timedelta
 
 cutoff_date = datetime.now() - timedelta(days=days_back)
 
 # Get recent interactions
 recent = user_interactions.where(
 (user_interactions.user_id == user_id) &
 (user_interactions.timestamp >= cutoff_date)
 )
 
 # Calculate statistics
 total_actions = recent.count()
 
 # Action distribution
 action_counts = recent.select(
 user_interactions.action,
 count=pxt.functions.count()
 ).group_by(user_interactions.action).collect()
 
 # Build profile
 return {
 'user_id': user_id,
 'total_actions': total_actions,
 'action_distribution': {
 row['action']: row['count'] 
 for row in action_counts
 },
 'profile_generated_at': datetime.now().isoformat()
 }

# Use profile in recommendation system
recommendations = pxt.create_table('recs.requests', {
 'user_id': pxt.String
})

# Profile retrieved once and cached
recommendations.add_computed_column(
 user_profile=get_user_behavior_profile(
 recommendations.user_id,
 days_back=30
 )
)
 
```

 
### Pattern 6: Complex Multi-Table Joins

 
Simplify complex joins across multiple tables:

 
```python

# Multiple related tables
orders = pxt.create_table('ecommerce.orders', {
 'order_id': pxt.String,
 'customer_id': pxt.String,
 'total_amount': pxt.Float,
 'status': pxt.String
})

customers = pxt.create_table('ecommerce.customers', {
 'customer_id': pxt.String,
 'name': pxt.String,
 'tier': pxt.String
})

products = pxt.create_table('ecommerce.products', {
 'product_id': pxt.String,
 'name': pxt.String,
 'category': pxt.String
})

# Reusable complex join query
@pxt.query
def get_customer_order_summary(
 customer_id: str,
 status_filter: str = None
):
 """Get comprehensive customer order information"""
 
 # Get customer info
 customer_info = customers.where(
 customers.customer_id == customer_id
 ).select(
 customers.name,
 customers.tier
 ).collect()
 
 # Get orders
 customer_orders = orders.where(
 orders.customer_id == customer_id
 )
 
 # Apply status filter if specified
 if status_filter:
 customer_orders = customer_orders.where(
 orders.status == status_filter
 )
 
 # Calculate summary statistics
 order_stats = customer_orders.select(
 total_orders=pxt.functions.count(),
 total_spent=pxt.functions.sum(orders.total_amount),
 avg_order_value=pxt.functions.avg(orders.total_amount)
 ).collect()
 
 # Combine results
 return {
 'customer_name': customer_info[0]['name'] if customer_info else 'Unknown',
 'customer_tier': customer_info[0]['tier'] if customer_info else 'Standard',
 'order_statistics': order_stats[0] if order_stats else {}
 }

# Use in customer support agent
support_tickets = pxt.create_table('support.tickets', {
 'ticket_id': pxt.String,
 'customer_id': pxt.String,
 'issue': pxt.String
})

# Automatically enrich tickets with customer context
support_tickets.add_computed_column(
 customer_summary=get_customer_order_summary(
 support_tickets.customer_id,
 status_filter='completed'
 )
)
 
```

 
## Advanced RAG Context Patterns

 
### Pattern 7: Time-Aware Context Retrieval

 
Build RAG systems that consider temporal relevance:

 
```python

# Time-aware document chunks
from datetime import datetime, timedelta

@chunks.query
def time_aware_retrieve(
 query: str,
 recency_weight: float = 0.3,
 limit: int = 5
):
 """Retrieve context with recency boosting"""
 
 # Calculate age in days for each chunk
 now = datetime.now()
 
 results = chunks.select(
 chunks.text,
 documents.title,
 documents.created_at,
 semantic_sim=chunks.text.similarity(string=query),
 age_days=(now - documents.created_at).days
 ).order_by(
 chunks.text.similarity(string=query), asc=False
 ).limit(limit * 2) # Get more candidates
 
 # Rerank with recency boost (done in Python)
 candidates = results.collect()
 
 for candidate in candidates:
 # Boost recent documents
 recency_boost = 1.0 / (1.0 + candidate['age_days'] / 365) # Decay over year
 
 # Combined score
 candidate['combined_score'] = (
 candidate['semantic_sim'] * (1.0 - recency_weight) +
 recency_boost * recency_weight
 )
 
 # Sort by combined score
 candidates.sort(key=lambda x: x['combined_score'], reverse=True)
 
 return candidates[:limit]

# Use in RAG with temporal awareness
rag_queries.add_computed_column(
 time_aware_context=time_aware_retrieve(
 rag_queries.question,
 recency_weight=0.4, # 40% weight on recency
 limit=5
 )
)
 
```

 
### Pattern 8: Multi-Hop Retrieval for Complex Questions

 
Build queries that retrieve context in multiple steps:

 
```python

@pxt.query
def multi_hop_retrieve(query: str, depth: int = 2, limit_per_hop: int = 3):
 """Multi-hop retrieval for complex questions"""
 
 all_context = []
 current_query = query
 
 for hop in range(depth):
 # Retrieve documents for current query
 hop_results = chunks.select(
 chunks.text,
 documents.title,
 similarity=chunks.text.similarity(string=current_query)
 ).order_by(
 chunks.text.similarity(string=current_query), asc=False
 ).limit(limit_per_hop).collect()
 
 all_context.extend(hop_results)
 
 # Generate follow-up query from retrieved context
 if hop < depth - 1 and hop_results:
 context_text = ' '.join([r['text'] for r in hop_results])
 
 # Use LLM to generate follow-up query
 from pixeltable.functions import openai
 
 follow_up = openai.chat_completions(
 model='gpt-4o-mini',
 messages=[{
 'role': 'user',
 'content': f"Based on this context: {context_text[:500]}\n\nWhat additional information would help answer: {query}\n\nGenerate a follow-up search query:"
 }]
 )
 
 current_query = follow_up.choices[0].message.content
 
 return all_context

# Use for complex questions requiring multiple retrieval steps
complex_queries = pxt.create_table('rag.complex_queries', {
 'question': pxt.String
})

complex_queries.add_computed_column(
 multi_hop_context=multi_hop_retrieve(
 complex_queries.question,
 depth=2,
 limit_per_hop=3
 )
)
 
```

 
## Real-World Applications

 
### Customer Support: Contextual Ticket Routing

 
```python

# Support tickets with intelligent routing
tickets = pxt.create_table('support.tickets', {
 'ticket_id': pxt.String,
 'customer_id': pxt.String,
 'issue_description': pxt.String,
 'category': pxt.String
})

# Historical resolution data
resolutions = pxt.create_table('support.resolutions', {
 'ticket_id': pxt.String,
 'agent_id': pxt.String,
 'resolution_time_hours': pxt.Float,
 'customer_satisfaction': pxt.Float
})

# Find best agent for ticket type
@pxt.query
def find_best_agent_for_ticket(
 issue_description: str,
 category: str,
 limit: int = 3
):
 """Find agents with best track record for this issue type"""
 
 # Find similar past tickets
 similar_tickets = tickets.where(
 tickets.category == category
 ).order_by(
 tickets.issue_description.similarity(string=issue_description), asc=False
 ).limit(10).select(tickets.ticket_id)
 
 similar_ids = [t['ticket_id'] for t in similar_tickets.collect()]
 
 # Get agent performance on similar tickets
 agent_performance = resolutions.where(
 resolutions.ticket_id.in_(similar_ids)
 ).select(
 resolutions.agent_id,
 avg_resolution_time=pxt.functions.avg(resolutions.resolution_time_hours),
 avg_satisfaction=pxt.functions.avg(resolutions.customer_satisfaction),
 success_count=pxt.functions.count()
 ).group_by(resolutions.agent_id).collect()
 
 # Rank agents
 ranked = sorted(
 agent_performance,
 key=lambda x: (x['avg_satisfaction'], -x['avg_resolution_time']),
 reverse=True
 )
 
 return ranked[:limit]

# Auto-route new tickets
new_tickets = pxt.create_table('support.new_tickets', {
 'ticket_id': pxt.String,
 'customer_id': pxt.String,
 'issue': pxt.String,
 'category': pxt.String
})

new_tickets.add_computed_column(
 recommended_agents=find_best_agent_for_ticket(
 new_tickets.issue,
 new_tickets.category,
 limit=3
 )
)
 
```

 
## @pxt.query vs. Manual Queries

 
| Aspect | Manual Queries | @pxt.query Decorator |
| --- | --- | --- |
| Reusability | Copy-paste code everywhere | Define once, use everywhere |
| Maintenance | Update in multiple places | Single source of truth |
| Composability | Difficult to combine queries | Easily compose query functions |
| Testing | Test each instance separately | Test query function once |
| Computed Columns | Can't use in computed columns | Direct integration with declarative workflows |

 
## Best Practices for Query Patterns

 
### Design Principles

 

 - **Single Responsibility:** Each query should do one thing well

 - **Parameterization:** Make queries flexible with sensible defaults

 - **Documentation:** Write clear docstrings explaining query purpose

 - **Error Handling:** Return empty results rather than raising errors

 - **Performance Awareness:** Include limit parameters to prevent unbounded queries

 

 
### Testing Query Functions

 
```python

# Test query functions with known data
def test_query_function():
 """Test reusable query with controlled data"""
 
 # Setup test data
 test_table = pxt.create_table('test.data', {
 'id': pxt.String,
 'value': pxt.Float,
 'category': pxt.String
 })
 
 test_table.insert([
 {'id': 'A', 'value': 1.0, 'category': 'X'},
 {'id': 'B', 'value': 2.0, 'category': 'X'},
 {'id': 'C', 'value': 3.0, 'category': 'Y'},
 ])
 
 # Define test query
 @test_table.query
 def get_by_category(category: str):
 return test_table.where(
 test_table.category == category
 ).select(test_table.value)
 
 # Test query
 results = get_by_category('X').collect()
 
 # Verify
 assert len(results) == 2, "Should return 2 items"
 assert results[0]['value'] == 1.0, "Should return correct values"
 
 print("✓ Query function test passed")

# Run tests before deployment
test_query_function()
 
```

 
## Integration with AI Workflows

 
### Pattern 9: Query Functions as Agent Tools

 
Convert query functions into [agent tools](/blog/retrieval-udf-database-ai-tool) for intelligent database access:

 
```python

# Define reusable database query
@customers.query
def search_customers(name_pattern: str, region: str = None):
 """Search customers by name pattern and optional region"""
 results = customers.where(
 customers.name.contains(name_pattern)
 )
 
 if region:
 results = results.where(customers.region == region)
 
 return results.select(
 customers.customer_id,
 customers.name,
 customers.region
 ).limit(10)

# Wrap query as retrieval tool for agents
customer_search_tool = pxt.retrieval_udf(
 customers,
 parameters=['name', 'region'],
 description='Search customers by name and region'
)

# Agent can now use database queries as tools
from pixeltable.functions import openai

agent_queries = pxt.create_table('agent.queries', {
 'query': pxt.String
})

agent_queries.add_computed_column(
 response=openai.chat_completions(
 model='gpt-4o',
 messages=[{'role': 'user', 'content': agent_queries.query}],
 tools=pxt.tools(customer_search_tool)
 )
)
 
```

 
## Conclusion: Query Patterns for Production AI

 
The `@pxt.query` decorator transforms how you build AI applications by making database queries reusable, composable, and declarative. Instead of scattering query logic throughout your codebase, you define clear, tested query functions that integrate seamlessly with [Pixeltable's declarative infrastructure](/blog/declarative-multimodal-incremental).

 
 
This pattern is particularly powerful for:

 

 - **AI Agent Memory:** [Reusable context retrieval](/blog/building-memory-powered-ai-stateful-agents-pixeltable) across conversations

 - **RAG Systems:** Sophisticated hybrid retrieval with consistent logic

 - **Multi-Table Analytics:** Complex joins and aggregations as simple functions

 - **Agent Tools:** Database queries that AI agents can call intelligently

 

 
 
By treating queries as first-class components in your AI architecture, you build more maintainable, testable, and performant applications. Combined with [custom aggregations (UDAs)](/blog/beyond-avg-custom-aggregations-uda) and [Python UDFs](/blog/python-udfs-pixeltable), you have complete control over your AI data workflows.

 
## Master Reusable Query Patterns

 

 - **[@pxt.query SDK Reference](https://docs.pixeltable.com/sdk/latest/pixeltable#query)**: Official documentation

 - **[Database to AI Tool Guide](/blog/retrieval-udf-database-ai-tool)**: Convert queries to agent tools

 - **[AI Agent Architecture](/blog/practical-guide-building-agents)**: Build agents with memory

 - **[Production RAG Systems](/blog/production-rag-data-centric)**: Advanced retrieval patterns

 - **[Your First Pixeltable Project](/blog/your-first-pixeltable-project)**: Get started with basics

 - **[Pixeltable on GitHub](https://github.com/pixeltable/pixeltable)**: Examples and source code

 - **[Join our Discord](https://discord.gg/QPyqFYx2UN)**: Share your query patterns

 

 
*Stop copying query code. Build reusable query patterns that scale with your AI applications.* 🔍