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

agent = Agent(
    name="data-assistant",
    instructions="Query and store records with Xata.",
    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

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