---
title: "The @pxt.query Decorator: Building Reusable Database Queries for AI Agents and RAG Systems"
date: "2025-10-12"
dateModified: "2026-11-07"
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 lookups, and optimized RAG context queries that integrate seamlessly with AI workflows."
url: "https://www.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:

 
The decorated function runs once, when it is decorated, with placeholders for its parameters. Parameters can appear in query expressions, but not in Python logic such as an `if` statement, and the function returns the query rather than calling `.collect()`. You then call it inside a `select()` or a computed column, where it runs once per row with that row's values.

 
```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 in any select(): it runs once per row
pxt.create_dir('analytics', if_exists='ignore')
queries = pxt.create_table('analytics.queries', {'target_region': pxt.String})
queries.insert([{'target_region': 'West'}, {'target_region': 'East'}])
queries.select(
 queries.target_region,
 top_customers=get_customers_by_region(queries.target_region, limit=5)
).collect()

# Or as a computed column, stored with each row
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: Conversation History Retrieval

 
```python

# Agent conversation table
pxt.create_dir('agent', if_exists='ignore')
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
@pxt.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
 )

# 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/pixeltable-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 an embedding index for semantic search
from pixeltable.functions import openai

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

# Reusable semantic memory search
@pxt.query
def search_relevant_memories(
 agent_id: str,
 query: str,
 memory_types: list[str],
 min_importance: float = 0.3,
 limit: int = 5
):
 """Search agent's semantic memory"""
 sim = agent_memory.memory_content.similarity(string=query)

 # One where() per query: combine the filters with &
 return agent_memory.where(
 (agent_memory.agent_id == agent_id)
 & (agent_memory.importance >= min_importance)
 & agent_memory.memory_type.isin(memory_types)
 ).order_by(sim, asc=False).limit(limit).select(
 agent_memory.memory_content,
 agent_memory.memory_type,
 agent_memory.importance,
 similarity=sim
 )

# 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

from datetime import datetime

# Document knowledge base
pxt.create_dir('knowledge', if_exists='ignore')
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 an embedding index
embed_fn = openai.embeddings.using(model='text-embedding-3-large')
chunks.add_embedding_index('text', embedding=embed_fn)

# Reusable hybrid retrieval query
@pxt.query
def hybrid_retrieve(
 query: str,
 categories: list[str],
 date_from: datetime,
 min_similarity: float = 0.7,
 limit: int = 5
):
 """Hybrid retrieval with semantic search + metadata filters"""
 sim = chunks.text.similarity(string=query)

 # A view has its base table's columns: chunks.category is the document's category
 return chunks.where(
 chunks.category.isin(categories)
 & (chunks.created_at >= date_from)
 & (sim >= min_similarity)
 ).order_by(sim, asc=False).limit(limit).select(
 chunks.text,
 chunks.title,
 chunks.category,
 chunks.author,
 similarity=sim
 )

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

# 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=openai.chat_completions(
 model='gpt-4o-mini',
 messages=[{
 'role': 'user',
 'content': 'Context: ' + pxt.functions.json.dumps(rag_queries.context)
 + '\n\nQuestion: ' + rag_queries.question
 }]
 ).choices[0].message.content
)
 
```

 
### 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
})

# Index the text that each source carries
embed_fn = openai.embeddings.using(model='text-embedding-3-small')
text_knowledge.add_embedding_index('content', embedding=embed_fn)
image_knowledge.add_embedding_index('description', embedding=embed_fn)
video_knowledge.add_embedding_index('transcript', embedding=embed_fn)

# A query reads one table, so write one query per source
@pxt.query
def search_text(query: str, limit: int = 3):
 sim = text_knowledge.content.similarity(string=query)
 return text_knowledge.order_by(sim, asc=False).limit(limit).select(
 text_knowledge.content,
 text_knowledge.source,
 modality='text'
 )

@pxt.query
def search_images(query: str, limit: int = 3):
 sim = image_knowledge.description.similarity(string=query)
 return image_knowledge.order_by(sim, asc=False).limit(limit).select(
 image_knowledge.description,
 image_knowledge.source,
 modality='image'
 )

@pxt.query
def search_videos(query: str, limit: int = 3):
 sim = video_knowledge.transcript.similarity(string=query)
 return video_knowledge.order_by(sim, asc=False).limit(limit).select(
 video_knowledge.transcript,
 video_knowledge.source,
 modality='video'
 )

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

agent_queries.add_computed_column(text_context=search_text(agent_queries.query, limit=2))
agent_queries.add_computed_column(image_context=search_images(agent_queries.query, limit=2))
agent_queries.add_computed_column(video_context=search_videos(agent_queries.query, limit=2))
 
```

 
## Performance Optimization with Query Patterns

 
### Pattern 5: Stored Results for Expensive Queries

 
Call an expensive query from a computed column, and its result is stored with each row, so reading it later does not run the query again. Pixeltable does not cache query results: each new row runs the query, even for a user it has seen before.

 
```python

# User analytics with expensive computations
from datetime import datetime

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

# Expensive aggregation query: a user's actions, counted by type
@pxt.query
def get_user_behavior_profile(user_id: str, since: datetime):
 """Generate a user behavior profile: action counts since a given time"""
 return user_interactions.where(
 (user_interactions.user_id == user_id) &
 (user_interactions.timestamp >= since)
 ).group_by(user_interactions.action).select(
 user_interactions.action,
 count=pxt.functions.count(user_interactions.action)
 )

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

# Profile computed once per row and stored with it
recommendations.add_computed_column(
 user_profile=get_user_behavior_profile(
 recommendations.user_id,
 recommendations.since
 )
)
 
```

 
### Pattern 6: Context From Several Tables

 
Pull context from several related tables into one row, with one query per table:

 
```python

# Multiple related tables
pxt.create_dir('ecommerce', if_exists='ignore')
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 queries over related tables: one query per table
@pxt.query
def get_customer_info(customer_id: str):
 """Get the customer's name and tier"""
 return customers.where(
 customers.customer_id == customer_id
 ).select(
 customers.name,
 customers.tier
 ).limit(1)

@pxt.query
def get_order_stats(customer_id: str, status_filter: str):
 """Get order statistics for one customer"""
 return orders.where(
 (orders.customer_id == customer_id) & (orders.status == status_filter)
 ).select(
 total_orders=pxt.functions.count(orders.order_id),
 total_spent=pxt.functions.sum(orders.total_amount),
 avg_order_value=pxt.functions.mean(orders.total_amount)
 )

# Use in customer support agent
pxt.create_dir('support', if_exists='ignore')
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_info=get_customer_info(support_tickets.customer_id)
)
support_tickets.add_computed_column(
 order_stats=get_order_stats(
 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, timezone

# Get more candidates than you need, with each chunk's document date
@pxt.query
def dated_candidates(query: str, limit: int = 10):
 """Top chunks by similarity, with the date of their document"""
 sim = chunks.text.similarity(string=query)
 return chunks.order_by(sim, asc=False).limit(limit).select(
 chunks.text,
 chunks.title,
 created_at=chunks.created_at.isoformat(),
 semantic_sim=sim
 )

# Rerank with recency boost (done in Python, in a UDF)
@pxt.udf
def rerank_by_recency(candidates: list, recency_weight: float, limit: int) -> list:
 """Retrieve context with recency boosting"""
 now = datetime.now(timezone.utc)

 for candidate in candidates:
 # Boost recent documents
 age_days = (now - datetime.fromisoformat(candidate['created_at'])).days
 recency_boost = 1.0 / (1.0 + 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(
 candidates=dated_candidates(rag_queries.question, limit=10)
)
rag_queries.add_computed_column(
 time_aware_context=rerank_by_recency(
 rag_queries.candidates,
 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

# One retrieval step as a reusable query
@pxt.query
def retrieve(query: str, limit: int = 3):
 """Retrieve the chunks most similar to a query"""
 sim = chunks.text.similarity(string=query)
 return chunks.order_by(sim, asc=False).limit(limit).select(
 chunks.text,
 chunks.title,
 similarity=sim
 )

# Each hop is a computed column, so both hops run for every new question
complex_queries = pxt.create_table('rag.complex_queries', {
 'question': pxt.String
})

# Hop 1: retrieve documents for the question
complex_queries.add_computed_column(
 hop_1=retrieve(complex_queries.question)
)

# Use LLM to generate follow-up query from retrieved context
complex_queries.add_computed_column(
 follow_up=openai.chat_completions(
 model='gpt-4o-mini',
 messages=[{
 'role': 'user',
 'content': 'Based on this context: ' + pxt.functions.json.dumps(complex_queries.hop_1)
 + '\n\nWhat additional information would help answer: ' + complex_queries.question
 + '\n\nGenerate a follow-up search query:'
 }]
 ).choices[0].message.content.astype(pxt.String)
)

# Hop 2: retrieve documents for the follow-up query
complex_queries.add_computed_column(
 hop_2=retrieve(complex_queries.follow_up)
)
 
```

 
## Real-World Applications

 
### Customer Support: Contextual Ticket Routing

 
```python

# Support tickets with intelligent routing
tickets = pxt.create_table('support.past_tickets', {
 'ticket_id': pxt.String,
 'customer_id': pxt.String,
 'issue_description': pxt.String,
 'category': pxt.String
})
tickets.add_embedding_index(
 'issue_description',
 embedding=openai.embeddings.using(model='text-embedding-3-small')
)

# 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 similar past tickets
@pxt.query
def find_similar_tickets(issue_description: str, category: str, limit: int = 10):
 """Past tickets in the same category, most similar first"""
 sim = tickets.issue_description.similarity(string=issue_description)
 return tickets.where(
 tickets.category == category
 ).order_by(sim, asc=False).limit(limit).select(tickets.ticket_id)

# Get agent performance on similar tickets
@pxt.query
def get_agent_performance(similar_ids: list[str]):
 """Resolution stats per agent on the given tickets"""
 return resolutions.where(
 resolutions.ticket_id.isin(similar_ids)
 ).group_by(resolutions.agent_id).select(
 resolutions.agent_id,
 avg_resolution_time=pxt.functions.mean(resolutions.resolution_time_hours),
 avg_satisfaction=pxt.functions.mean(resolutions.customer_satisfaction),
 success_count=pxt.functions.count(resolutions.ticket_id)
 )

# Rank agents
@pxt.udf
def rank_agents(agent_performance: list, limit: int) -> list:
 """Find agents with best track record for this issue type"""
 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(
 similar_tickets=find_similar_tickets(new_tickets.issue, new_tickets.category)
)
new_tickets.add_computed_column(
 agent_stats=get_agent_performance(new_tickets.similar_tickets)
)
new_tickets.add_computed_column(
 recommended_agents=rank_agents(new_tickets.agent_stats, 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
 pxt.create_dir('test', if_exists='ignore')
 test_table = pxt.create_table('test.data', {
 'id': pxt.String,
 'value': pxt.Float,
 'category': pxt.String
 }, if_exists='replace')
 
 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
 @pxt.query
 def get_by_category(category: str):
 return test_table.where(
 test_table.category == category
 ).order_by(test_table.id).select(test_table.value)
 
 # Run the query from a one-row table of test inputs
 inputs = pxt.create_table('test.inputs', {'category': pxt.String}, if_exists='replace')
 inputs.insert([{'category': 'X'}])
 results = inputs.select(values=get_by_category(inputs.category)).collect()
 
 # Verify
 assert len(results[0]['values']) == 2, "Should return 2 items"
 assert results[0]['values'][0] == 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 = pxt.get_table('customers') # the table from the first example

@pxt.query
def search_customers(name_pattern: str, region: str):
 """Search customers by name pattern and region"""
 return customers.where(
 customers.name.contains(name_pattern) & (customers.region == region)
 ).select(
 customers.customer_id,
 customers.name,
 customers.region
 ).limit(10)

# A query function is already a tool. For exact-match lookups,
# pxt.retrieval_udf() builds one from the table's columns
customer_lookup = pxt.retrieval_udf(
 customers,
 name='lookup_customer',
 parameters=['customer_id'],
 description='Look up a customer by ID'
)
tools = pxt.tools(search_customers, customer_lookup)

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

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

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

# Run the tool calls the model asked for
agent_queries.add_computed_column(
 tool_output=openai.invoke_tools(tools, agent_queries.response)
)
 
```

 
## 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:** Lookups and aggregations across tables 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.* 🔍