Performance monitoring and caching utilities for AI systems.
- QueryProfiler: Database query profiling and performance monitoring with support for sync and async functions
- TTLCache: Time-based function result caching decorator with thread-safe cache management
- Performance Analytics: Track slow queries, get performance summaries, and analyze bottlenecks
- Cache Statistics: Monitor cache hits, misses, and hit rates
pip install socratic-performanceTrack database query performance with automatic slow query detection:
from socratic_performance import QueryProfiler
profiler = QueryProfiler(slow_query_threshold_ms=100)
@profiler.profile("get_user")
async def load_user(user_id: str):
# Your async database query here
return await db.get(User, user_id)
# Execute the function
user = await load_user("user_123")
# Get statistics
stats = profiler.get_stats()
print(f"get_user avg time: {stats['get_user']['avg_time_ms']:.2f}ms")
# Get slowest queries
slowest = profiler.get_slowest_queries(limit=5)
for query in slowest:
print(f"{query['name']}: {query['avg_time_ms']:.1f}ms")Cache expensive function results with automatic expiration:
from socratic_performance import cached
@cached(ttl_minutes=5)
def expensive_computation(project_id: str) -> dict:
# Simulate expensive operation
return analyze_project(project_id)
result1 = expensive_computation("proj_123") # Computed (slow)
result2 = expensive_computation("proj_123") # Cached (fast)
# Check cache statistics
stats = expensive_computation.cache_stats()
print(f"Cache hit rate: {stats['hit_rate']}")profiler = QueryProfiler(slow_query_threshold_ms=100.0)Parameters:
slow_query_threshold_ms(float): Threshold above which queries are considered slow (default: 100ms)
@profiler.profile("query_name")
async def my_query():
pass
@profiler.profile("custom_threshold", slow_query_threshold_ms=50)
def fast_query():
passstart = time.time()
try:
result = perform_operation()
duration = time.time() - start
profiler.manual_track("operation", duration)
except Exception:
duration = time.time() - start
profiler.manual_track("operation", duration, error=True)
raise# Get all statistics
stats = profiler.get_stats()
# Get slow queries (with slow executions)
slow_queries = profiler.get_slow_queries(min_slow_count=1)
# Get slowest queries (by average time)
slowest = profiler.get_slowest_queries(limit=10)
# Print summary
profiler.print_summary(limit=5)
# Reset statistics
profiler.reset_stats() # Reset all
profiler.reset_stats("specific_query") # Reset onecache = TTLCache(ttl_minutes=5)
@cache
def expensive_operation(x, y):
return x + yfrom socratic_performance import cached
@cached(ttl_minutes=10)
def my_function(param):
return expensive_operation(param)# Clear cache
my_function.cache_clear()
# Get statistics
stats = my_function.cache_stats()
# Returns: {'hits': 5, 'misses': 3, 'total_calls': 8, 'hit_rate': '62.5%', ...}
# Get human-readable info
info = my_function.cache_info()
# Returns: "Cache: 3 entries, 62.5% hit rate, TTL: 10 minutes"cache = TTLCache(ttl_minutes=5)
# Clean up expired entries
removed_count = cache.cleanup_expired()
# Reset statistics only (keep cache)
cache.reset_stats()
# Clear all cache
cache.clear()See the examples/ directory for complete examples:
- 01_query_profiling.py - Query profiling with decorators, manual tracking, async functions
- 02_ttl_cache.py - TTL cache usage, multiple functions, kwargs handling
Run examples:
python examples/01_query_profiling.py
python examples/02_ttl_cache.pyAccess a global profiler instance without managing your own:
from socratic_performance import get_profiler, profile_query
@profile_query("operation_name")
async def my_operation():
pass
# Get global profiler
profiler = get_profiler()
stats = profiler.get_stats()- Set appropriate thresholds: Tune
slow_query_threshold_msbased on your performance requirements - Use meaningful names: Give queries descriptive names for easy identification
- Review slow queries regularly: Use
get_slow_queries()andget_slowest_queries()to identify bottlenecks - Print summaries: Use
print_summary()to log performance reports
- Choose appropriate TTLs: Balance freshness vs. performance
- Monitor hit rates: Check
cache_stats()to ensure caching is effective - Handle unhashable arguments: The cache gracefully skips caching for unhashable arguments
- Clean up expired entries: Call
cleanup_expired()periodically for long-running processes
QueryProfiler provides low-overhead query performance tracking:
- Works with both sync and async functions
- Configurable slow query thresholds
- Automatic error tracking
- Minimal performance impact
TTLCache implements thread-safe result caching:
- Automatic expiration after TTL
- Thread-safe operations using RLock
- Graceful handling of unhashable arguments
- Detailed cache statistics
MIT
Contributions welcome! Please ensure tests pass and code is formatted with black.
pytest tests/
black .This package is a component of Socrates AI, a production-ready platform for building intelligent multi-agent systems with constitutional governance.
pip install socratic-performancepip install socrates-ai # Includes 37+ modules + all 11 packagesSee the Socrates ECOSYSTEM.md for detailed integration examples showing how to use socratic-performance with other Socratic packages.
Related packages you might use together: