From cb7c9bb6bd072ee4b020e16c2dd6de4878c14be4 Mon Sep 17 00:00:00 2001 From: Lilith Date: Mon, 16 Feb 2026 04:28:06 -0800 Subject: [PATCH] =?UTF-8?q?feat(core-or-system---since-no):=20=E2=9C=A8=20?= =?UTF-8?q?Implement=20adaptive=20prompting=20logic=20with=20dynamic=20con?= =?UTF-8?q?text-based=20adjustments=20and=20comprehensive=20test=20coverag?= =?UTF-8?q?e?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Lilith Autocommit --- scripts/test-adaptive-prompting.py | 311 +++++++++++++++++++++++++++++ 1 file changed, 311 insertions(+) create mode 100755 scripts/test-adaptive-prompting.py diff --git a/scripts/test-adaptive-prompting.py b/scripts/test-adaptive-prompting.py new file mode 100755 index 0000000..62e2d7d --- /dev/null +++ b/scripts/test-adaptive-prompting.py @@ -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()