Xata Serverless Database
PostgreSQL with built-in search, analytics, and file storage for AI agents
Xata provides PostgreSQL with built-in full-text search, vector search, analytics, and file storage, perfect for AI agents that need comprehensive data capabilities.
import os
from praisonaiagents import Agent, MemoryConfig
agent = Agent(
name="data-assistant",
instructions="Query and store records with Xata.",
memory=MemoryConfig(
db={"database_url": os.environ["XATA_DATABASE_URL"], "api_key": os.environ["XATA_API_KEY"]},
))
agent.start("Save this note about our launch plan")The user stores data; Xata persists rows, search indexes, and vectors for the agent.
How It Works
Quick Start
- Set Up Xata Database
- Create account at xata.io
- Create a new database
- Get PostgreSQL connection string from Settings
- Set environment variable:
bash export XATA_DATABASE_URL="postgresql://workspace:api_key@us-east-1.sql.xata.sh:5432/mydb:main?sslmode=require" - Create Analytics-Ready Agent
python from praisonaiagents import Agent, MemoryConfig agent = Agent( name="Xata Agent", instructions="You are an AI assistant with built-in search and analytics.", memory=MemoryConfig( db={"database_url": "postgresql://ws:key@us-east-1.sql.xata.sh:5432/db:main?sslmode=require"}, )) # Conversations automatically indexed for search result = agent.start("I'm working on a machine learning project with customer data") print(result) - Test Search Capabilities
python # Later conversation - search works across all messages result = agent.start("Find our previous discussion about machine learning") print(result) # Agent can search conversation history # Built-in analytics track conversation patterns result = agent.start("What topics do we discuss most frequently?") print(result) # Xata analytics can power these insights
Installation
pip
# Xata uses PostgreSQL driver
pip install "praisonai[xata]"Environment Variables
# Required - includes workspace and API key
export XATA_DATABASE_URL="postgresql://workspace:api_key@us-east-1.sql.xata.sh:5432/db:main?sslmode=require"
# Optional
export OPENAI_API_KEY="sk-..."Configuration Options
| Option | Type | Default | Description |
|---|---|---|---|
database_url | str | None | Xata PostgreSQL connection URL |
max_retries | int | 3 | Retries for serverless cold starts |
retry_delay | float | 0.5 | Base delay between retries |
auto_create_tables | bool | True | Create conversation tables |
enable_search | bool | True | Enable full-text search indexing |
Usage Patterns
Using Convenience Class
from praisonai.db.adapter import XataDB
from praisonaiagents import Agent, MemoryConfig
# Auto-reads from XATA_DATABASE_URL environment variable
db = XataDB()
agent = Agent(name="Xata Agent",
memory=MemoryConfig(
db=db,
))Manual Configuration with Search
from praisonai.db.adapter import PraisonAIDB
from praisonaiagents import Agent, MemoryConfig
db = PraisonAIDB(
database_url="postgresql://ws:key@us-east-1.sql.xata.sh:5432/db:main?sslmode=require",
max_retries=5, # Extra retries for cold starts
auto_create_tables=True
)
agent = Agent(
name="Search-Enabled Agent",
instructions="You can search through our entire conversation history.",
memory=MemoryConfig(
db=db,
))Full Lifecycle with Search & Analytics
import os
from praisonai import ManagedAgent, LocalManagedConfig
from praisonai.db.adapter import XataDB
from praisonaiagents import Agent, MemoryConfig
# Phase 1: Create agent with Xata
db = XataDB(database_url=os.environ["XATA_DATABASE_URL"])
managed = ManagedAgent(
provider="local",
db=db,
config=LocalManagedConfig(
model="gpt-4o-mini",
name="Xata Analytics Agent",
system="You are an AI assistant with powerful search and analytics capabilities."
)
)
agent = Agent(name="User", backend=managed)
# Build rich conversation data
conversations = [
"I'm working on a customer segmentation ML model using clustering",
"The model will help target marketing campaigns more effectively",
"I need to analyze purchase history and demographic data",
"Performance metrics show 85% accuracy on test dataset",
"Next step is deploying to production with A/B testing"
]
for message in conversations:
result = agent.run(message)
print(f"Agent: {result[:100]}...")
print(f"Session: {managed.session_id}")
# Phase 2: Search and analytics
search_queries = [
"What machine learning technique am I using?",
"What's the accuracy of my model?",
"What data sources do I need?",
"Summarize my entire ML project"
]
for query in search_queries:
result = agent.run(query)
print(f"\nQuery: {query}")
print(f"Answer: {result[:200]}...")
# Phase 3: Save and resume
session_data = managed.save_ids()
del agent, managed, db
# Phase 4: Resume with search intact
db2 = XataDB(database_url=os.environ["XATA_DATABASE_URL"])
managed2 = ManagedAgent(provider="local", db=db2)
managed2.resume_session(session_data["session_id"])
agent2 = Agent(name="User", backend=managed2)
result = agent2.run("Search for everything related to accuracy and testing")
print(f"\nResumed search: {result[:300]}...")Xata-Specific Features
Built-in Full-Text Search
All conversation data is automatically indexed for search:
# Conversations are searchable by default
agent = Agent(
name="Searchable Agent",
instructions="You can search our conversation history using natural language.",
memory=MemoryConfig(
db=XataDB(),
))
# Search works across all messages
result = agent.run("Find mentions of 'database performance' in our chats")
print(result) # Returns relevant conversation contextVector Search for Similarity
Store and search embeddings alongside conversations:
-- Add vector column to messages table (via Xata Console)
ALTER TABLE praison_messages ADD COLUMN embedding VECTOR(1536);
-- Enable vector search
CREATE INDEX ON praison_messages USING ivfflat (embedding vector_cosine_ops);from openai import OpenAI
# Custom agent with vector search capability
class VectorSearchAgent(Agent):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.openai = OpenAI()
def search_similar(self, query: str, top_k: int = 5):
# Get query embedding
response = self.openai.embeddings.create(
model="text-embedding-ada-002",
input=query
)
query_vector = response.data[0].embedding
# Search similar messages (custom SQL via Xata API)
# Implementation would use Xata's vector search
return f"Found {top_k} similar conversations about: {query}"
agent = VectorSearchAgent(name="Vector Agent", db=XataDB())
similar = agent.search_similar("machine learning accuracy")
print(similar)Real-Time Analytics
Track conversation patterns and metrics:
# Analytics queries via Xata (pseudo-code)
def get_conversation_analytics(session_id: str):
# Use Xata's analytics API or custom aggregation
analytics = {
"total_messages": "SELECT COUNT(*) FROM praison_messages WHERE session_id = ?",
"avg_response_time": "SELECT AVG(response_time) FROM praison_messages",
"topic_distribution": "SELECT topics, COUNT(*) FROM praison_messages GROUP BY topics",
"sentiment_analysis": "SELECT sentiment, COUNT(*) FROM praison_messages GROUP BY sentiment"
}
return analytics
# Agent can provide analytics about itself
agent = Agent(
name="Self-Aware Agent",
instructions="You can analyze your own conversation patterns and provide insights.",
memory=MemoryConfig(
db=XataDB(),
))File Storage Integration
Store conversation-related files directly in Xata:
# File upload example (using Xata API)
import requests
def upload_conversation_file(file_path: str, session_id: str):
# Upload file to Xata storage
with open(file_path, 'rb') as f:
response = requests.post(
"https://workspace.xata.sh/db/mydb/tables/conversation_files/data",
headers={"Authorization": "Bearer api_key"},
files={"file": f},
data={"session_id": session_id}
)
return response.json()
# Agent with file handling
agent = Agent(
name="File-Aware Agent",
instructions="You can reference uploaded files and documents in our conversations.",
memory=MemoryConfig(
db=XataDB(),
))Best Practices
Optimize Search Performance
Structure conversation data for effective search:
# Store rich metadata for better search
agent = Agent(
name="Structured Agent",
instructions="You organize information with clear topics and context.",
memory=MemoryConfig(
db=XataDB(),
))
# Conversations with good structure improve search results
result = agent.run("Topic: Machine Learning - Discussing model accuracy metrics")Use Analytics for Insights
Leverage Xata’s analytics to improve agent performance:
# Track conversation quality metrics
class AnalyticsAgent(Agent):
def get_conversation_stats(self):
return {
"messages_count": "Query conversation length",
"response_quality": "Analyze response patterns",
"user_satisfaction": "Track positive/negative sentiment"
}
agent = AnalyticsAgent(name="Metrics Agent", db=XataDB())Manage Data Growth
Monitor database size and implement archiving:
# Archive old conversations to manage costs
def archive_old_conversations(days_old: int = 90):
# Use Xata branching or backup features
# Move old data to cold storage
pass
# Implement in agent lifecycle
agent = Agent(
name="Lifecycle Agent",
instructions="You manage conversation history efficiently.",
memory=MemoryConfig(
db=XataDB(),
))Leverage Multi-Modal Data
Store different data types together:
# Agent that handles text, files, and structured data
agent = Agent(
name="Multi-Modal Agent",
instructions="""
You work with:
- Text conversations (PostgreSQL)
- Document files (Xata storage)
- Search across both
- Analytics on usage patterns
""",
memory=MemoryConfig(
db=XataDB(),
))Environment Variables
| Variable | Required | Format | Example |
|---|---|---|---|
XATA_DATABASE_URL | ✅ | postgresql://workspace:key@region.sql.xata.sh:5432/db:branch | postgresql://ws_123:xau_abc@us-east-1.sql.xata.sh:5432/chatdb:main?sslmode=require |
OPENAI_API_KEY | ✅ | sk-... | sk-1234567890abcdef... |
Feature Comparison
| Feature | Xata | Standard PostgreSQL | Benefit |
|---|---|---|---|
| Full-Text Search | ✅ Built-in | Manual setup | Instant search capability |
| Vector Search | ✅ Native | Extension required | AI/ML integration |
| File Storage | ✅ Integrated | Separate service | Unified data platform |
| Analytics | ✅ Built-in | Custom queries | Conversation insights |
| Branching | ✅ Git-like | Complex | Easy development workflow |
Troubleshooting
Search Index Issues
If search isn’t working:
# Ensure tables have proper text indexing
# Check Xata console for search configuration
# May need to recreate search indexesFile Upload Limits
Xata has file size limits:
# Check file size before upload
import os
file_size = os.path.getsize("large_file.pdf")
if file_size > 10 * 1024 * 1024: # 10MB limit
print("File too large for Xata storage")Connection String Format
Ensure proper connection string format:
# Correct format includes workspace, API key, region, database, and branch
export XATA_DATABASE_URL="postgresql://workspace:apikey@region.sql.xata.sh:5432/database:branch?sslmode=require"Branch Management
If using multiple branches:
# Switch between development and production branches
dev_db = XataDB(database_url="postgresql://ws:key@region.sql.xata.sh:5432/db:dev")
prod_db = XataDB(database_url="postgresql://ws:key@region.sql.xata.sh:5432/db:main")
dev_agent = Agent(name="Dev Agent", memory={"db": dev_db})
prod_agent = Agent(name="Prod Agent", memory={"db": prod_db})