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.

python
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.

Diagram

Xata Multi-Feature Platform

Conversation

Search

Vectors

Files

🤖 Agent

🔀 Data Type

🐘 PostgreSQL

🔍 Full-Text

🧠 Vector DB

📁 File Storage

📊 Analytics

⚡ Auto-Scale

How It Works

Diagram
Xata Serverless DatabaseAgentUserXata Serverless DatabaseAgentUserRequestProcess requestResult Agent–>>User: Response

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:
    bash
    export XATA_DATABASE_URL="postgresql://workspace:api_key@us-east-1.sql.xata.sh:5432/mydb:main?sslmode=require"
  2. 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)
  3. 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

bash
# Xata uses PostgreSQL driver
pip install "praisonai[xata]"

Environment Variables

bash
# 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

OptionTypeDefaultDescription
database_urlstrNoneXata PostgreSQL connection URL
max_retriesint3Retries for serverless cold starts
retry_delayfloat0.5Base delay between retries
auto_create_tablesboolTrueCreate conversation tables
enable_searchboolTrueEnable full-text search indexing

Usage Patterns

Using Convenience Class

python
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

python
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

python
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:

python
# 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 context

Vector Search for Similarity

Store and search embeddings alongside conversations:

sql
-- 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);
python
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:

python
# 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:

python
# 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:

python
# 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:

python
# 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:

python
# 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:

python
# 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

VariableRequiredFormatExample
XATA_DATABASE_URL✅postgresql://workspace:key@region.sql.xata.sh:5432/db:branchpostgresql://ws_123:xau_abc@us-east-1.sql.xata.sh:5432/chatdb:main?sslmode=require
OPENAI_API_KEY✅sk-...sk-1234567890abcdef...

Feature Comparison

FeatureXataStandard PostgreSQLBenefit
Full-Text Search✅ Built-inManual setupInstant search capability
Vector Search✅ NativeExtension requiredAI/ML integration
File Storage✅ IntegratedSeparate serviceUnified data platform
Analytics✅ Built-inCustom queriesConversation insights
Branching✅ Git-likeComplexEasy development workflow

Troubleshooting

Search Index Issues

If search isn’t working:

python
# 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:

python
# 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:

bash
# 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:

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

Related