Prompt Intelligence System - Implementation Plan

✅ Test 1: Retrieval - Retrieved workflow with 9 stages
✅ Test 2: Catalog - 1 workflow cached
✅ Test 3: Semantic Search - Found 1 result
✅ Test 4: Update Prompt - Version incremented
✅ Test 5: Performance - <1ms cached retrieval
✅ Test 6: API Integration - All 5 endpoints registered
✅ Test 7: Database Sync - Adapters initialized

📊 VALIDATION SUMMARY: 7/7 tests passed
🎉 System ready for production

# Pre-build workflows at startup
try:
    from stores.prompts import get_prompt_store
    prompt_store = get_prompt_store()
    await prompt_store.initialize()
    
    # Build critical workflows
    await prompt_store.get_workflow("questionnaire_to_contract")
    logger.info("✅ Prompt Intelligence initialized - workflows cached")
except Exception as e:
    logger.warning(f"⚠️  Prompt Intelligence initialization failed: {e} - will build on-demand")

# In stores/prompts.py
from apscheduler.schedulers.asyncio import AsyncIOScheduler
 
scheduler = AsyncIOScheduler()
 
# Daily 9am: Rebuild embeddings
scheduler.add_job(rebuild_embeddings, 'cron', hour=9)
 
# Hourly: Update analytics
scheduler.add_job(update_analytics, 'interval', hours=1)
 
scheduler.start()

// Add to lib/api.ts
export async function searchPrompts(query: string) {
  return fetch('/api/prompts/search', {
    method: 'POST',
    body: JSON.stringify({query, namespace: 'workflows'})
  });
}
 
export async function executeWorkflow(workflow: string, data: any) {
  return fetch('/api/prompts/execute', {
    method: 'POST',
    body: JSON.stringify({workflow, input_data: data})
  });
}

# Add to app.py
import requests
 
def search_workflows(query):
    response = requests.post(
        f"{BACKEND_URL}/api/prompts/search",
        json={"query": query, "namespace": "workflows"}
    )
    return response.json()
 
def execute_workflow(workflow_name, input_data):
    response = requests.post(
        f"{BACKEND_URL}/api/prompts/execute",
        json={"workflow": workflow_name, "input_data": input_data}
    )
    return response.json()

-- Top performing prompts
SELECT 
  prompt_name,
  version,
  suggestions_count,
  updated_at
FROM prompt_catalog
ORDER BY suggestions_count DESC
LIMIT 10;
 
-- Workflow execution stats
SELECT 
  workflow_name,
  COUNT(*) as total_executions,
  AVG(execution_time) as avg_time_seconds,
  COUNT(*) FILTER (WHERE success = true)::FLOAT / COUNT(*) as success_rate
FROM workflow_executions
GROUP BY workflow_name
ORDER BY total_executions DESC;
 
-- League-specific example coverage
SELECT 
  sport,
  COUNT(*) as example_count,
  COUNT(DISTINCT tier) as tier_coverage
FROM league_examples
GROUP BY sport
ORDER BY example_count DESC;

// Monitor user activity
prompts/
  user_metrics/
    total_users: 156
    active_today: 42
    top_workflow: "questionnaire_to_contract"
    avg_queries_per_user: 8.5

# Local testing with mock DBs
export FIREBASE_SERVICE_ACCOUNT_PATH=""  # Use mock
export SUPABASE_URL=""                   # Use mock
 
python database/scripts/build.py
python database/scripts/validate.py
cd apps/backend && python server.py

# Build for Cloud Run
docker build -t gcr.io/PROJECT_ID/altsportsdata-backend:latest .
 
# Test locally
docker run -p 8080:8080 \
  -e OPENAI_API_KEY=$OPENAI_API_KEY \
  gcr.io/PROJECT_ID/altsportsdata-backend:latest

# Push image
docker push gcr.io/PROJECT_ID/altsportsdata-backend:latest
 
# Deploy
gcloud run deploy altsportsdata-backend \
  --image gcr.io/PROJECT_ID/altsportsdata-backend:latest \
  --platform managed \
  --region us-central1 \
  --memory 2Gi \
  --cpu 2 \
  --timeout 300 \
  --set-env-vars OPENAI_API_KEY=$OPENAI_API_KEY \
  --set-env-vars FIREBASE_SERVICE_ACCOUNT_PATH=./config/firebase.json \
  --set-env-vars SUPABASE_URL=$SUPABASE_URL \
  --set-env-vars SUPABASE_SERVICE_KEY=$SUPABASE_SERVICE_KEY

# Get Cloud Run URL
SERVICE_URL=$(gcloud run services describe altsportsdata-backend --format='value(status.url)')
 
# Test catalog
curl $SERVICE_URL/api/prompts/catalog
 
# Test search
curl -X POST $SERVICE_URL/api/prompts/search \
  -H "Content-Type: application/json" \
  -d '{"query": "questionnaire to contract"}'
 
# Test execution
curl -X POST $SERVICE_URL/api/prompts/execute \
  -H "Content-Type: application/json" \
  -d '{
    "workflow": "questionnaire_to_contract",
    "input_data": {"questionnaire_text": "Sample questionnaire..."}
  }'

1. User/Analyst identifies improvement
   ↓
2. POST /api/prompts/update with suggestions
   ↓
3. System updates InMemoryStore (immediate)
   ↓
4. Background sync to Firebase (persistent)
   ↓
5. Background track in Supabase (analytics)
   ↓
6. All future queries use new version
   ↓
7. All Cloud Run instances sync via Firebase

Traffic spike → Cloud Run spins up new instances
  ↓
New instance starts
  ↓
Runs database/scripts/build.py
  ↓
Loads from Firebase (gets latest prompt versions)
  ↓
Populates InMemoryStore
  ↓
Instance ready (<5s startup)
  ↓
Handles requests with <1ms prompt retrieval

# Process 100 questionnaires
POST /api/prompts/batch
{
  "workflow": "questionnaire_to_contract",
  "inputs": [... 100 questionnaires ...],
  "max_parallel": 10
}
 
# Cloud Run automatically:
# - Scales to handle load
# - Processes 10 concurrent workflows
# - Each workflow uses cached prompts
# - Returns all results in ~3 minutes (vs 50 minutes sequential)