Quick Start Guide: Prompt Management System

# Run comprehensive tests (proves both use cases)
python data_layer/scripts/test_prompt_retrieval.py
 
# Expected: ✅ All tests passing

from data_layer.scripts.test_prompt_retrieval import SimplePromptRetriever
 
# Initialize
retriever = SimplePromptRetriever()
 
# Find prompts for league onboarding
results = retriever.search_by_keywords(
    keywords=["league", "questionnaire", "extraction", "database"],
    top_k=5
)
 
# Print results
for i, r in enumerate(results, 1):
    print(f"{i}. {r['title']} (score: {r['score']:.1f})")
    print(f"   ID: {r['id']}")
    print(f"   Type: {r['type']}")
    print()

# Get by ID
prompt = retriever.get_by_id("workflows.league-questionnaire-extraction.v1")
 
print(f"Title: {prompt['title']}")
print(f"Type: {prompt['type']}")
print(f"Confidence: {prompt['confidence']*100:.0f}%")
print(f"Tags: {', '.join(prompt['tags'][:5])}")

from data_layer.scripts.test_prompt_retrieval import SimplePromptRetriever
from data_layer.scripts.generate_adapters import (
    LeagueQuestionnaireSchema,
    TierClassificationSchema
)
 
# 1. Find relevant prompts
retriever = SimplePromptRetriever()
prompts = retriever.search_by_keywords(
    keywords=["league", "questionnaire", "extraction", "database", "upsert"],
    top_k=5
)
 
# 2. Build workflow
print("League Onboarding Workflow:")
print()
print("Step 1: Extract Questionnaire")
print(f"  Prompt: {prompts[0]['id']}")
print(f"  Output: LeagueQuestionnaireSchema")
print()
 
print("Step 2: Enrich Data")
print(f"  Prompt: {prompts[1]['id']}")
print(f"  Output: EnrichedLeagueDataSchema")
print()
 
print("Step 3: Classify Tier")
print(f"  Prompt: {prompts[2]['id']}")
print(f"  Output: TierClassificationSchema")
print()
 
print("Step 4: Upsert to Database")
print("  Output: DatabaseUpsertResultSchema")

from data_layer.scripts.test_prompt_retrieval import SimplePromptRetriever
from data_layer.scripts.generate_adapters import (
    ContractTermsSchema,
    NegotiationPackageSchema
)
 
# 1. Find contract templates
retriever = SimplePromptRetriever()
prompts = retriever.search_by_keywords(
    keywords=["tier", "contract", "partnership", "pricing", "premium"],
    top_k=5
)
 
# 2. Build workflow
print("Contract Generation Workflow:")
print()
print("Step 1: Load Profile")
print("  Input: league_id")
print(f"  Output: LeagueProfileSchema")
print()
 
print("Step 2: Generate Terms")
print(f"  Prompt: {prompts[0]['id']}")
print(f"  Output: ContractTermsSchema")
print()
 
print("Step 3: Create Variants")
print(f"  Prompt: {prompts[1]['id']}")
print(f"  Output: PricingVariantsSchema")
print()
 
print("Step 4: Generate Documents")
print(f"  Prompt: {prompts[2]['id']}")
print(f"  Output: NegotiationPackageSchema")
print()
 
print("Step 5: Save Outputs")
print("  Location: ./output/contracts/League_Name_TIMESTAMP/")
print("  Files: contract_deal.md, contract_list.md, contract_ceiling.md")

# Search only workflows
workflow_results = retriever.search_by_keywords(
    keywords=["email", "processing"],
    top_k=3,
    filter_type="workflow"
)
 
# Search only contract templates
contract_results = retriever.search_by_keywords(
    keywords=["tier", "partnership"],
    top_k=3,
    filter_type="contract_template"
)
 
# Search with minimum confidence
high_quality = retriever.search_by_keywords(
    keywords=["league", "analysis"],
    top_k=5,
    min_confidence=0.80  # Only prompts with 80%+ confidence
)

# Step 1: Scan prompts and rebuild registry
python data_layer/scripts/scan_prompts.py
 
# Step 2: Regenerate enriched documentation
python data_layer/scripts/generate_prompt_docs.py
 
# Step 3: Sync to Google Drive
python data_layer/scripts/sync_to_drive.py
 
# Step 4: Re-index for semantic search (optional)
python data_layer/scripts/index_prompts.py

# View registry statistics
python -c "
import json
from pathlib import Path
 
registry_file = Path('data_layer/kb_catalog/manifests/prompt_registry.json')
registry = json.loads(registry_file.read_text())
 
print(f'Total Prompts: {len(registry[\"prompts\"])}')
print(f'Last Updated: {registry.get(\"generated_at\", \"N/A\")}')
 
# Count by type
types = {}
for p in registry['prompts']:
    t = p['type']
    types[t] = types.get(t, 0) + 1
 
print('\\nBy Type:')
for t, count in sorted(types.items()):
    print(f'  {t}: {count}')
"

from data_layer.scripts.generate_adapters import (
    # League schemas
    LeagueQuestionnaireSchema,
    LeagueProfileSchema,
    TierClassificationSchema,
 
    # Contract schemas
    ContractTermsSchema,
    NegotiationPackageSchema,
    PricingVariantsSchema,
 
    # Database schemas
    DatabaseUpsertResultSchema,
 
    # And many more...
)

# Get prompt with schema info
prompt = retriever.get_by_id("workflows.league-questionnaire-extraction.v1")
 
print(f"Required Input Schemas: {prompt['requires_schemas']}")
# Output: ['LeagueQuestionnaireSchema']
 
print(f"Output Schema: {prompt['output_schema']}")
# Output: 'ExtractedLeagueDataSchema'
 
# Use schema for validation
from data_layer.scripts.generate_adapters import LeagueQuestionnaireSchema
 
data = extract_questionnaire("./questionnaire.pdf")
validated = LeagueQuestionnaireSchema(**data)  # Validates structure

prompt = retriever.get_by_id("workflows.league-questionnaire-extraction.v1")
 
print(f"Suggested Agents: {prompt['agents_suggested']}")
# Output: ['document-processor', 'email-handler', 'ocr-agent']
 
# Use agents from kb_catalog
from pathlib import Path
import json
 
agents_file = Path("data_layer/kb_catalog/manifests/agents.json")
agents = json.loads(agents_file.read_text())
 
for agent_id in prompt['agents_suggested']:
    agent = next((a for a in agents if a['id'] == agent_id), None)
    if agent:
        print(f"\nAgent: {agent['name']}")
        print(f"Description: {agent['description']}")
        print(f"Tools: {', '.join(agent['tools'][:3])}")

data_layer/
├── scripts/
│   ├── scan_prompts.py              # Build registry
│   ├── generate_prompt_docs.py      # Generate docs
│   ├── sync_to_drive.py             # Sync to Drive
│   ├── index_prompts.py             # LangMem indexing
│   ├── test_prompt_retrieval.py     # Test suite
│   └── demo_prompt_workflows.py     # Demo workflows
│
├── kb_catalog/manifests/
│   ├── prompt_registry.json         # Main registry (116 prompts)
│   └── agents.json                  # Agent catalog
│
├── storage/
│   ├── prompts/docs/                # Enriched documentation
│   ├── prompts/drive_sync/          # Drive sync state
│   └── embeddings/langmem_index/    # Semantic search index
│
└── prompts/                         # Source .md files (source of truth)
    ├── workflows/
    ├── agents/
    ├── specs/contracts/
    ├── specs/legal/
    └── components/

# List all prompts by type
ls data_layer/storage/prompts/docs/workflow/
ls data_layer/storage/prompts/docs/contract_template/
ls data_layer/storage/prompts/docs/agent/
 
# Read specific prompt doc
cat "data_layer/storage/prompts/docs/workflow/workflows.league-questionnaire-extraction.v1.md"
 
# View source prompt
cat "data_layer/prompts/workflows/league_questionnaire_extraction_v1.md"

python data_layer/scripts/test_prompt_retrieval.py

✅ Found 5 relevant prompts (League Onboarding)
✅ Found 5 relevant prompts (Contract Generation)
✅ All workflows generated successfully
✅ SYSTEM STATUS: FULLY OPERATIONAL