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.

Quick Start

1. Set Up Xata Database

   1. Create account at xata.io
   2. Create a new database
   3. Get PostgreSQL connection string from Settings
   4. Set environment variable:

   export XATA_DATABASE_URL="postgresql://workspace:[e-mail]:5432/mydb:main?sslmode=require"
   
2. Create Analytics-Ready Agent

   from praisonaiagents import Agent
   
   agent = Agent(
       name="Xata Agent",
       instructions="You are an AI assistant with built-in search and analytics.",
       db={"database_url": "postgresql://ws:[e-mail]: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)
   
3. Test Search Capabilities

   # 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:[e-mail]: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

# Auto-reads from XATA_DATABASE_URL environment variable
db = XataDB()
agent = Agent(name="Xata Agent", db=db)

Manual Configuration with Search

from praisonai.db.adapter import PraisonAIDB
from praisonaiagents import Agent

db = PraisonAIDB(
    database_url="postgresql://ws:[e-mail]: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.",
    db=db
)

Full Lifecycle with Search & Analytics

import os
from praisonai import ManagedAgent, LocalManagedConfig  
from praisonai.db.adapter import XataDB
from praisonaiagents import Agent

# 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.",
    db=XataDB()
)

# Search works across all messages
result = agent.run("Find mentions of 'database performance' in our chats")
print(result)  # Returns relevant conversation context

Vector 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.",
    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.",
    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.",
    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.",
    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
    """,
    db=XataDB()
)

---

Environment Variables

Variable	Required	Format	Example
XATA_DATABASE_URL	✅	postgresql://workspace:[e-mail]:5432/db:branch	postgresql://ws_123:[e-mail]: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 indexes

File 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:[e-mail]:5432/database:branch?sslmode=require"

Branch Management

If using multiple branches:

# Switch between development and production branches
dev_db = XataDB(database_url="postgresql://ws:[e-mail]:5432/db:dev")
prod_db = XataDB(database_url="postgresql://ws:[e-mail]:5432/db:main")

dev_agent = Agent(name="Dev Agent", db=dev_db)
prod_agent = Agent(name="Prod Agent", db=prod_db)

---

Related

Cloud Databases Overview

Compare all cloud database providers

Search & Analytics

Advanced search and analytics features