Xata Serverless Database

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)

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

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

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]}...")

# 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

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

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

# Ensure tables have proper text indexing
# Check Xata console for search configuration
# May need to recreate search indexes

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

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

# 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", db=dev_db)
prod_agent = Agent(name="Prod Agent", db=prod_db)