312 lines
11 KiB
Python
Executable file
312 lines
11 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Test script for adaptive prompting system.
|
|
|
|
Demonstrates user stats tracking and adaptive prompt building with synthetic data.
|
|
|
|
Usage:
|
|
python scripts/test-adaptive-prompting.py
|
|
python scripts/test-adaptive-prompting.py --user-id custom-user
|
|
python scripts/test-adaptive-prompting.py --clear-cache
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
# Add packages to path
|
|
sys.path.insert(0, str(Path.home() / "Code/@applications/@ml/knowledge-platform"))
|
|
sys.path.insert(
|
|
0,
|
|
str(
|
|
Path(__file__).parent.parent
|
|
/ "operations/platform-knowledge/crystal-ai/src"
|
|
),
|
|
)
|
|
|
|
from knowledge_platform.feedback import (
|
|
AdaptivePromptBuilder,
|
|
UserStatsTracker,
|
|
)
|
|
|
|
|
|
def create_test_feedback_data(storage_dir: Path, user_id: str) -> None:
|
|
"""Create synthetic feedback data for testing."""
|
|
print(f"📝 Creating test feedback data for user: {user_id}")
|
|
|
|
corrections_dir = storage_dir / "corrections"
|
|
validations_dir = storage_dir / "validations"
|
|
searches_dir = storage_dir / "searches"
|
|
|
|
for directory in [corrections_dir, validations_dir, searches_dir]:
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Create correction logs (last 7 days)
|
|
correction_logs = []
|
|
|
|
# Frequent pattern: "escort" → "creator" (10 times)
|
|
for i in range(10):
|
|
timestamp = datetime.now() - timedelta(days=i % 7, hours=i)
|
|
correction_logs.append(
|
|
{
|
|
"timestamp": timestamp.isoformat(),
|
|
"conversation_id": f"{user_id}-session-{i // 3}",
|
|
"original": "escort platform features",
|
|
"corrected": "creator platform features",
|
|
"changes": [
|
|
{
|
|
"type": "terminology",
|
|
"original": "escort",
|
|
"replacement": "creator",
|
|
}
|
|
],
|
|
"confidence": 0.92 + (i * 0.01),
|
|
}
|
|
)
|
|
|
|
# Frequent pattern: "blockchain" → "database" (8 times)
|
|
for i in range(8):
|
|
timestamp = datetime.now() - timedelta(days=i % 5, hours=i + 2)
|
|
correction_logs.append(
|
|
{
|
|
"timestamp": timestamp.isoformat(),
|
|
"conversation_id": f"{user_id}-session-{i // 3}",
|
|
"original": "blockchain storage system",
|
|
"corrected": "database storage system",
|
|
"changes": [
|
|
{
|
|
"type": "factual",
|
|
"original": "blockchain",
|
|
"replacement": "database",
|
|
}
|
|
],
|
|
"confidence": 0.88 + (i * 0.01),
|
|
}
|
|
)
|
|
|
|
# Frequent pattern: "5% fee" → "0% fee" (6 times)
|
|
for i in range(6):
|
|
timestamp = datetime.now() - timedelta(days=i % 4, hours=i + 4)
|
|
correction_logs.append(
|
|
{
|
|
"timestamp": timestamp.isoformat(),
|
|
"conversation_id": f"{user_id}-session-{i // 2}",
|
|
"original": "Lilith charges 5% fee",
|
|
"corrected": "Lilith charges 0% fee",
|
|
"changes": [
|
|
{
|
|
"type": "factual",
|
|
"original": "5%",
|
|
"replacement": "0%",
|
|
}
|
|
],
|
|
"confidence": 0.95,
|
|
}
|
|
)
|
|
|
|
# Write corrections to date-based JSONL files
|
|
corrections_by_date: dict[str, list] = {}
|
|
for log in correction_logs:
|
|
date_str = datetime.fromisoformat(log["timestamp"]).strftime("%Y%m%d")
|
|
if date_str not in corrections_by_date:
|
|
corrections_by_date[date_str] = []
|
|
corrections_by_date[date_str].append(log)
|
|
|
|
for date_str, logs in corrections_by_date.items():
|
|
log_file = corrections_dir / f"{date_str}.jsonl"
|
|
with open(log_file, "w") as f:
|
|
for log in logs:
|
|
f.write(json.dumps(log) + "\n")
|
|
|
|
print(f" ✓ Created {len(correction_logs)} correction events across {len(corrections_by_date)} days")
|
|
|
|
# Create validation logs (topic focus and low confidence)
|
|
validation_logs = []
|
|
|
|
# High focus on "marketplace" topic (15 interactions)
|
|
for i in range(15):
|
|
timestamp = datetime.now() - timedelta(days=i % 7, hours=i)
|
|
validation_logs.append(
|
|
{
|
|
"timestamp": timestamp.isoformat(),
|
|
"conversation_id": f"{user_id}-session-{i // 3}",
|
|
"subjects": ["marketplace", "booking"],
|
|
"confidence": 0.85 + (i * 0.01),
|
|
}
|
|
)
|
|
|
|
# Some interactions on "legal" topic (low confidence)
|
|
for i in range(5):
|
|
timestamp = datetime.now() - timedelta(days=i % 3, hours=i + 1)
|
|
validation_logs.append(
|
|
{
|
|
"timestamp": timestamp.isoformat(),
|
|
"conversation_id": f"{user_id}-session-{i}",
|
|
"subjects": ["legal", "compliance"],
|
|
"confidence": 0.42 + (i * 0.02), # Low confidence
|
|
}
|
|
)
|
|
|
|
# Some interactions on "payments" topic (low confidence)
|
|
for i in range(4):
|
|
timestamp = datetime.now() - timedelta(days=i % 3, hours=i + 3)
|
|
validation_logs.append(
|
|
{
|
|
"timestamp": timestamp.isoformat(),
|
|
"conversation_id": f"{user_id}-session-{i}",
|
|
"subjects": ["payments", "billing"],
|
|
"confidence": 0.45 + (i * 0.02), # Low confidence
|
|
}
|
|
)
|
|
|
|
# Write validations to date-based JSONL files
|
|
validations_by_date: dict[str, list] = {}
|
|
for log in validation_logs:
|
|
date_str = datetime.fromisoformat(log["timestamp"]).strftime("%Y%m%d")
|
|
if date_str not in validations_by_date:
|
|
validations_by_date[date_str] = []
|
|
validations_by_date[date_str].append(log)
|
|
|
|
for date_str, logs in validations_by_date.items():
|
|
log_file = validations_dir / f"{date_str}.jsonl"
|
|
with open(log_file, "w") as f:
|
|
for log in logs:
|
|
f.write(json.dumps(log) + "\n")
|
|
|
|
print(f" ✓ Created {len(validation_logs)} validation events across {len(validations_by_date)} days")
|
|
|
|
|
|
def test_user_stats(storage_dir: Path, user_id: str) -> None:
|
|
"""Test user statistics computation."""
|
|
print(f"\n📊 Computing user statistics...")
|
|
|
|
tracker = UserStatsTracker(storage_dir)
|
|
stats = tracker.get_user_stats(user_id, days=30, min_correction_count=3)
|
|
|
|
print(f"\n📋 User Stats Summary:")
|
|
print(f" User ID: {stats.user_id}")
|
|
print(f" Period: {stats.period_start[:10]} to {stats.period_end[:10]}")
|
|
print(f" Total Interactions: {stats.total_interactions}")
|
|
|
|
if stats.corrections:
|
|
print(f"\n✏️ Corrections:")
|
|
print(f" Total: {stats.corrections.total_corrections}")
|
|
print(f" Avg Confidence: {stats.corrections.avg_confidence:.2f}")
|
|
print(f" Frequent Patterns:")
|
|
for original, replacement, count in stats.corrections.frequent_patterns[:5]:
|
|
print(f" • {original} → {replacement} ({count}x)")
|
|
print(f" Common Error Types:")
|
|
for error_type, count in stats.corrections.common_error_types[:5]:
|
|
print(f" • {error_type}: {count} occurrences")
|
|
|
|
if stats.topics:
|
|
print(f"\n📚 Topics:")
|
|
print(f" Primary Topics:")
|
|
for topic, count in stats.topics.primary_topics[:5]:
|
|
print(f" • {topic}: {count} interactions")
|
|
print(f" Low Confidence Topics:")
|
|
for topic, confidence in stats.topics.low_confidence_topics[:5]:
|
|
print(f" • {topic}: {confidence:.0%} avg confidence")
|
|
|
|
|
|
def test_adaptive_prompt(storage_dir: Path, user_id: str) -> None:
|
|
"""Test adaptive prompt building."""
|
|
print(f"\n🔧 Building adaptive prompt...")
|
|
|
|
base_prompt = """You are Crystal, an expert on the Lilith Platform.
|
|
|
|
You have access to comprehensive documentation about Lilith's features,
|
|
architecture, and best practices. Use your knowledge to provide accurate,
|
|
helpful responses to user questions."""
|
|
|
|
builder = AdaptivePromptBuilder(
|
|
user_id=user_id,
|
|
storage_dir=storage_dir,
|
|
stats_days=30,
|
|
)
|
|
|
|
# Build with context
|
|
context = {
|
|
"recent_low_confidence": True,
|
|
"current_topic": "marketplace features",
|
|
"session_corrections": 2,
|
|
}
|
|
|
|
enhanced_prompt = builder.build(base_prompt, context)
|
|
|
|
print(f"\n📝 Base Prompt Length: {len(base_prompt)} chars")
|
|
print(f"📝 Enhanced Prompt Length: {len(enhanced_prompt)} chars")
|
|
print(f"📝 Added: {len(enhanced_prompt) - len(base_prompt)} chars in adaptive sections")
|
|
|
|
print(f"\n🎯 Enhanced Prompt:")
|
|
print("=" * 80)
|
|
print(enhanced_prompt)
|
|
print("=" * 80)
|
|
|
|
# Get summary
|
|
summary = builder.get_stats_summary()
|
|
print(f"\n📊 Stats Summary (for debugging):")
|
|
print(json.dumps(summary, indent=2, default=str))
|
|
|
|
|
|
def main() -> None:
|
|
"""Main test execution."""
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(
|
|
description="Test adaptive prompting system"
|
|
)
|
|
parser.add_argument(
|
|
"--user-id",
|
|
default="test-user-123",
|
|
help="User ID for testing (default: test-user-123)",
|
|
)
|
|
parser.add_argument(
|
|
"--clear-cache",
|
|
action="store_true",
|
|
help="Clear cached stats before running",
|
|
)
|
|
parser.add_argument(
|
|
"--storage-dir",
|
|
type=Path,
|
|
default=Path.home() / ".cache/crystal/feedback-test",
|
|
help="Storage directory for test data",
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
storage_dir: Path = args.storage_dir
|
|
user_id: str = args.user_id
|
|
|
|
print("=" * 80)
|
|
print("🧪 Adaptive Prompting System Test")
|
|
print("=" * 80)
|
|
|
|
# Clear cache if requested
|
|
if args.clear_cache:
|
|
cache_dir = storage_dir / "user-stats"
|
|
if cache_dir.exists():
|
|
cache_file = cache_dir / f"{user_id}.json"
|
|
if cache_file.exists():
|
|
cache_file.unlink()
|
|
print(f"🗑️ Cleared cache for user: {user_id}")
|
|
|
|
# Create test data
|
|
create_test_feedback_data(storage_dir, user_id)
|
|
|
|
# Test user stats
|
|
test_user_stats(storage_dir, user_id)
|
|
|
|
# Test adaptive prompt
|
|
test_adaptive_prompt(storage_dir, user_id)
|
|
|
|
print(f"\n✅ Test complete!")
|
|
print(f"\n📂 Test data location: {storage_dir}")
|
|
print(f" You can inspect the generated files:")
|
|
print(f" - Corrections: {storage_dir / 'corrections'}/*.jsonl")
|
|
print(f" - Validations: {storage_dir / 'validations'}/*.jsonl")
|
|
print(f" - Cached stats: {storage_dir / 'user-stats'}/{user_id}.json")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|