Source: data_layer/docs/PROMPT_INTELLIGENCE_README.md

Prompt Intelligence System

Production-grade prompt retrieval and workflow execution for Google Cloud Run

🎯 Core Capabilities

Catalog: List all 135+ available prompts and workflows
Search: Find closest match using semantic vector search
Update: Improve prompts with suggestions → auto-sync to databases
Execute: Run pre-built workflows with LangGraph state machines

🏗️ Architecture (lesson_5.py Pattern)

InMemoryStore (Fast &lt;1ms)
  ↓ On startup: Load existing
Firebase (User data, real-time)
  - User prompt preferences
  - Prompt history/feedback
  ↓ On update: Sync
Supabase (League data, analytics)
  - League-specific examples
  - Workflow performance metrics
  - SQL analytics queries
  ↓ Build if missing
Files (Guaranteed fallback)
  - database/prompts/ (135 .md files)
  - database/output-styles/ (7-stage pipeline)

📂 File Structure

apps/backend/
├── stores/
│   └── prompts.py          # InMemoryStore + DB sync
├── services/
│   └── prompts.py          # Business logic + workflow execution
├── api/
│   └── prompts.py          # REST API endpoints
└── server.py               # Router registration (line 574-580)

database/
├── prompts/                # 135 .md source files
├── output-styles/
│   └── league_questionnaire_to_contract/  # 7-stage pipeline
└── scripts/
    ├── build.py            # Pre-build workflows
    └── validate.py         # End-to-end tests

🚀 API Endpoints

1. GET /api/prompts/catalog

List all available prompts:

GET /api/prompts/catalog
 
Response:
{
  "status": "success",
  "catalog": {
    "workflows": [
      {
        "name": "questionnaire_to_contract",
        "namespace": "workflows",
        "metadata": {"total_stages": 7, "avg_execution_time": 22.5},
        "version": 2
      }
    ],
    "prompts": [...],
    "components": [...],
    "recipes": [...]
  },
  "total": 158
}

2. POST /api/prompts/search

Semantic search for closest match:

POST /api/prompts/search
{
  "query": "convert basketball questionnaire to contract",
  "namespace": "workflows",
  "limit": 5
}
 
Response:
{
  "status": "success",
  "query": "convert basketball questionnaire to contract",
  "results": [
    {
      "name": "questionnaire_to_contract",
      "namespace": "workflows",
      "score": 0.95,
      "metadata": {...},
      "version": 2
    }
  ]
}

3. POST /api/prompts/update

Update prompt with improvement suggestions:

POST /api/prompts/update
{
  "prompt_type": "workflow",
  "prompt_name": "questionnaire_to_contract",
  "suggestions": [
    "Add more basketball-specific examples",
    "Improve NBA tier classification",
    "Include luxury tax considerations"
  ]
}
 
Response:
{
  "status": "updated",
  "new_version": 3,
  "suggestions_applied": 3,
  "synced_to_db": true
}

4. POST /api/prompts/execute

Execute workflow:

POST /api/prompts/execute
{
  "workflow": "questionnaire_to_contract",
  "input_data": {
    "questionnaire_text": "...",
    "file_path": "s3://uploads/nba.pdf"
  },
  "user_id": "analyst@altsportsdata.com"
}
 
Response:
{
  "status": "completed",
  "workflow": "questionnaire_to_contract",
  "result": {...},
  "execution_time": 18.5,
  "stages_completed": 7
}

5. POST /api/prompts/batch

Parallel batch execution:

POST /api/prompts/batch
{
  "workflow": "questionnaire_to_contract",
  "inputs": [
    {"questionnaire_text": "NBA..."},
    {"questionnaire_text": "MLB..."},
    {"questionnaire_text": "NHL..."}
  ],
  "max_parallel": 10
}
 
Response:
{
  "status": "batch_completed",
  "total_processed": 3,
  "successful": 3,
  "results": [...]
}

💾 Database Strategy (Google Cloud Production)

Firebase (User Data)

prompts/
  workflows/
    questionnaire_to_contract: {...}  // Critical workflows
  
  user_preferences/
    {userId}/
      preferred_detail_level: "technical"
      favorite_workflows: [...]
 
  user_history/
    {userId}/
      {promptId}: {uses, success_rate, feedback}

Supabase (League Data + Analytics)

-- Workflow definitions and performance
CREATE TABLE workflow_definitions (
  workflow_name TEXT PRIMARY KEY,
  total_stages INT,
  version INT,
  metadata JSONB,
  updated_at TIMESTAMP
);
 
-- Prompt catalog for search
CREATE TABLE prompt_catalog (
  prompt_type TEXT,
  prompt_name TEXT,
  version INT,
  suggestions_count INT,
  metadata JSONB,
  updated_at TIMESTAMP,
  PRIMARY KEY (prompt_type, prompt_name)
);
 
-- Execution tracking
CREATE TABLE workflow_executions (
  id SERIAL PRIMARY KEY,
  workflow_name TEXT,
  execution_time FLOAT,
  success BOOLEAN,
  executed_at TIMESTAMP
);
 
-- League-specific examples
CREATE TABLE league_examples (
  league_name TEXT,
  sport TEXT,
  tier TEXT,
  examples JSONB,
  updated_at TIMESTAMP
);

🔧 Deployment (Google Cloud Run)

Build Workflows at Container Startup

Add to Dockerfile or startup script:

# In Dockerfile
CMD python database/scripts/build.py && uvicorn apps.backend.server:app --host 0.0.0.0 --port $PORT

Or in server.py lifespan:

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup - build workflows
    try:
        from stores.prompts import get_prompt_store
        store = get_prompt_store()
        await store.initialize()
        await store.get_workflow("questionnaire_to_contract")
        logger.info("✅ Workflows pre-built and cached")
    except Exception as e:
        logger.warning(f"Workflow pre-build failed: {e}")
    
    yield
    # Shutdown

Environment Variables

# Required for embeddings
OPENAI_API_KEY=sk-...
 
# Optional - for database sync
FIREBASE_SERVICE_ACCOUNT_PATH=./config/firebase-service-account.json
SUPABASE_URL=https://...
SUPABASE_SERVICE_KEY=...

📊 Performance Characteristics

Operation	Time	Notes
First query	~9ms	Build from files + cache
Cached query	<1ms	InMemoryStore retrieval
Firebase sync	~20ms	Background async
Supabase sync	~15ms	Background async
Workflow execution	15-30s	7-stage LangGraph pipeline
Batch (10 parallel)	~30s	vs 5min sequential

🧪 Testing

Run Validation Suite

cd /Users/kbselander/Developer/Notebook/mcp-servers/servers/mcp-server-altsportsleagues.ai/2.1-cloud-run-docker-mcp
python database/scripts/validate.py

Tests:

✅ Retrieval from InMemoryStore
✅ Catalog listing
✅ Semantic search
✅ Prompt updates
✅ Performance (<1ms cached)
✅ API endpoints
✅ Database sync

Manual Testing

# Start server
cd apps/backend
python server.py
 
# Test catalog
curl http://localhost:8080/api/prompts/catalog
 
# Test search
curl -X POST http://localhost:8080/api/prompts/search \
  -H "Content-Type: application/json" \
  -d '{"query": "questionnaire to contract", "namespace": "workflows"}'
 
# Test update
curl -X POST http://localhost:8080/api/prompts/update \
  -H "Content-Type: application/json" \
  -d '{"prompt_type": "workflow", "prompt_name": "questionnaire_to_contract", "suggestions": ["Add NBA examples"]}'

🎯 Usage Patterns

Pattern 1: Retrieve and Execute

from services.prompts import PromptService
 
service = PromptService()
 
# Execute workflow
result = await service.execute_workflow(
    "questionnaire_to_contract",
    {"questionnaire_text": "..."}
)

Pattern 2: Search and Update

from stores.prompts import get_prompt_store
 
store = get_prompt_store()
await store.initialize()
 
# Search for workflow
results = await store.search_prompts("basketball contract generation")
 
# Update with improvements
updated = await store.update_prompt(
    "workflow",
    results[0].key,
    ["Add more NBA examples", "Improve tier logic"]
)

Pattern 3: Batch Processing

# Process 50 questionnaires in parallel
result = await service.batch_execute(
    "questionnaire_to_contract",
    [{"questionnaire_text": q} for q in questionnaires],
    max_parallel=10
)

🔄 Update Workflow

When you have new examples or improvements:

Update via API:

POST /api/prompts/update
{
  "prompt_type": "workflow",
  "prompt_name": "questionnaire_to_contract",
  "suggestions": ["Your improvements here"]
}

System automatically:
- Increments version
- Updates InMemoryStore (immediate)
- Syncs to Firebase (user data)
- Syncs to Supabase (analytics)
Next execution uses updated version

🎓 Best Practices (Production)

Data Separation

Firebase: User-specific (preferences, history, feedback)
Supabase: League-specific (examples, analytics, performance)
InMemoryStore: All prompts (fast cache)

Sync Strategy

Reads: Always from InMemoryStore (fast)
Writes: InMemoryStore + async background sync (doesn't block)
Startup: Load from Firebase (restore state across container restarts)

Error Handling

InMemoryStore always works (no network)
Firebase/Supabase failures don't break system
Graceful degradation to in-memory only

Scaling (Cloud Run)

Each container instance has own InMemoryStore
Firebase syncs state between instances
Supabase tracks aggregate analytics
Multi-instance safe

📈 Monitoring

Track in Supabase:

-- Most used workflows
SELECT workflow_name, COUNT(*) as executions
FROM workflow_executions
GROUP BY workflow_name
ORDER BY executions DESC;
 
-- Average execution time by workflow
SELECT workflow_name, AVG(execution_time) as avg_time
FROM workflow_executions
WHERE success = true
GROUP BY workflow_name;
 
-- Prompt update frequency
SELECT prompt_name, MAX(version) as current_version, COUNT(*) as total_updates
FROM prompt_catalog
GROUP BY prompt_name
ORDER BY total_updates DESC;

🎉 What This Enables

Dynamic Agent Spawning: Query → Retrieve workflow → Spawn LangGraph
Continuous Improvement: Update prompts based on feedback
Multi-Frontend: Next.js + Streamlit both use same intelligence
Horizontal Scaling: Parallel batch processing
Production Reliability: Multi-tier fallback, DB sync, monitoring

Built following DeepLearning.AI lesson_5.py patterns for production Google Cloud Run deployment.

Production Deployment Prompt Intelligence System - COMPLETE ✅