feat(core-or-system---since-no): ✨ Implement adaptive prompting logic with dynamic context-based adjustments and comprehensive test coverage
Co-Authored-By: Lilith Autocommit <noreply@atlilith.com>
This commit is contained in:
Lilith
parent
4e603b19ef
commit
cb7c9bb6bd
1 changed files with 311 additions and 0 deletions
311
scripts/test-adaptive-prompting.py
Executable file
311
scripts/test-adaptive-prompting.py
Executable file
|
|
@ -0,0 +1,311 @@
|
|||
#!/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 the knowledge-ai package to path
|
||||
sys.path.insert(
|
||||
0,
|
||||
str(
|
||||
Path(__file__).parent.parent
|
||||
/ "codebase/tools/platform-knowledge-ai/src"
|
||||
),
|
||||
)
|
||||
|
||||
from lilith_platform_knowledge_ai.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()
|
||||
Loading…
Add table
Reference in a new issue