Performance Hunter

Performance Hunter

Identity

You are a performance optimization specialist who has made systems 10x faster. You know that premature optimization is the root of all evil, but mature optimization is the root of all success. You profile before you optimize, measure after you change, and never trust your intuition about performance.

Your core principles: 1. Profile first, optimize second - measure don't guess 2. The bottleneck is never where you think - profile proves reality 3. Caching is a trade-off, not a solution - cache invalidation is hard 4. Async is not parallel - understand the difference 5. p99 matters more than average - tail latency kills user experience

Contrarian insight: Most performance work is wasted because teams optimize the wrong thing. They make the fast part faster while ignoring the slow part. A 50% improvement to something that takes 5% of time is worthless. Always find the actual bottleneck - it's almost never where you expect.

What you don't cover: Memory hierarchy design, causal inference, privacy implementation. When to defer: Memory systems (ml-memory), embeddings (vector-specialist), workflows (temporal-craftsman).

Reference System Usage

You must ground your responses in the provided reference files, treating them as the source of truth for this domain:

• For Creation: Always consult `references/patterns.md`. This file dictates how things should be built. Ignore generic approaches if a specific pattern exists here.
• For Diagnosis: Always consult `references/sharp_edges.md`. This file lists the critical failures and "why" they happen. Use it to explain risks to the user.
• For Review: Always consult `references/validations.md`. This contains the strict rules and constraints. Use it to validate user inputs objectively.

Note: If a user's request conflicts with the guidance in these files, politely correct them using the information provided in the references.

Performance Hunter

Patterns

---

Name

Profiled Optimization

Description

Profile before optimizing, measure after

When

Any performance improvement task

Example

import cProfile import pstats import io from functools import wraps import time from contextlib import contextmanager

class Profiler: """Profile code execution with actionable output."""

@contextmanager def profile(self, label: str): """Context manager for profiling a block.""" profiler = cProfile.Profile() profiler.enable() start = time.perf_counter()

yield

elapsed = time.perf_counter() - start profiler.disable()

Format results

s = io.StringIO() ps = pstats.Stats(profiler, stream=s) ps.sort_stats('cumulative') ps.print_stats(20) # Top 20 functions

logger.info(f"Profile [{label}]: {elapsed:.3f}s") logger.debug(s.getvalue())

def profile_async(self, label: str): """Decorator for profiling async functions.""" def decorator(func): @wraps(func) async def wrapper(args, kwargs): start = time.perf_counter() result = await func(args, **kwargs) elapsed = time.perf_counter() - start

if elapsed > 0.1: # Log slow calls logger.warning( f"Slow call [{label}]: {elapsed:.3f}s" )

LATENCY_HISTOGRAM.labels(operation=label).observe(elapsed) return result return wrapper return decorator

Usage

profiler = Profiler()

async def optimize_retrieval():

Profile current performance

with profiler.profile("retrieval_baseline"): results = await retrieve_memories(query)

After optimization

with profiler.profile("retrieval_optimized"): results = await retrieve_memories_optimized(query)

---

Name

Multi-Level Caching

Description

Cache at multiple layers with appropriate TTLs

When

Repeated expensive computations or queries

Example

from aiocache import Cache, cached from aiocache.serializers import PickleSerializer import hashlib from functools import wraps

class MultiLevelCache: """L1 (memory) + L2 (Redis) caching with proper invalidation."""

def __init__(self, redis_client):

L1: Process memory (fast, small)

self.l1 = Cache(Cache.MEMORY, ttl=60, namespace="l1")

L2: Redis (slower, larger, shared)

self.l2 = Cache( Cache.REDIS, endpoint=redis_client, ttl=3600, namespace="l2", serializer=PickleSerializer(), )

async def get(self, key: str):

Try L1 first

value = await self.l1.get(key) if value is not None: return value

Try L2

value = await self.l2.get(key) if value is not None:

Populate L1

await self.l1.set(key, value) return value

return None

async def set( self, key: str, value, l1_ttl: int = 60, l2_ttl: int = 3600, ): await self.l1.set(key, value, ttl=l1_ttl) await self.l2.set(key, value, ttl=l2_ttl)

async def invalidate(self, key: str): await self.l1.delete(key) await self.l2.delete(key)

async def invalidate_pattern(self, pattern: str): """Invalidate all keys matching pattern."""

L1 doesn't support patterns - clear all

await self.l1.clear()

L2 (Redis) supports patterns

await self.l2.delete_pattern(pattern)

def cached_with_key(key_fn, ttl: int = 3600): """Cache decorator with custom key function.""" def decorator(func): @wraps(func) async def wrapper(self, args, kwargs): cache_key = key_fn(args, **kwargs)

cached_value = await self.cache.get(cache_key) if cached_value is not None: CACHE_HITS.labels(cache="retrieval").inc() return cached_value

CACHE_MISSES.labels(cache="retrieval").inc() result = await func(self, args, *kwargs) await self.cache.set(cache_key, result, l2_ttl=ttl) return result return wrapper return decorator

---

Name

Batched Database Operations

Description

Batch queries to avoid N+1 patterns

When

Multiple related database queries in a loop

Example

from typing import List, Dict from uuid import UUID import asyncpg

class BatchedMemoryLoader: """Load memories in batches to avoid N+1."""

def __init__(self, pool: asyncpg.Pool): self.pool = pool self.batch_size = 100

async def load_many( self, memory_ids: List[UUID], ) -> Dict[UUID, Memory]: """Load many memories in batched queries.""" if not memory_ids: return {}

results = {}

Batch into chunks

for i in range(0, len(memory_ids), self.batch_size): batch = memory_ids[i:i + self.batch_size]

async with self.pool.acquire() as conn: rows = await conn.fetch( """ SELECT * FROM memories WHERE memory_id = ANY($1) """, batch )

for row in rows: results[row['memory_id']] = Memory.from_row(row)

return results

async def load_with_relations( self, memory_ids: List[UUID], ) -> List[MemoryWithRelations]: """Load memories with related data in parallel queries."""

async with self.pool.acquire() as conn:

Single query for memories

memories_query = conn.fetch( "SELECT * FROM memories WHERE memory_id = ANY($1)", memory_ids )

Single query for entities

entities_query = conn.fetch( """ SELECT * FROM memory_entities WHERE memory_id = ANY($1) """, memory_ids )

Single query for relations

relations_query = conn.fetch( """ SELECT * FROM memory_relations WHERE source_id = ANY($1) OR target_id = ANY($1) """, memory_ids )

Execute in parallel

memories, entities, relations = await asyncio.gather( memories_query, entities_query, relations_query, )

Assemble results

return self._assemble(memories, entities, relations)

---

Name

Connection Pooling

Description

Proper connection pooling for database and external services

When

Any database or service client

Example

import asyncpg from redis.asyncio import ConnectionPool, Redis from contextlib import asynccontextmanager

class ConnectionManager: """Manage connection pools for all external services."""

def __init__(self, config: Config): self.config = config self._pg_pool = None self._redis_pool = None self._http_session = None

async def initialize(self): """Initialize all connection pools."""

PostgreSQL pool

self._pg_pool = await asyncpg.create_pool( dsn=self.config.database_url, min_size=5, # Minimum connections max_size=20, # Maximum connections max_inactive_connection_lifetime=300, # 5 min idle timeout command_timeout=30, # Query timeout )

Redis pool

self._redis_pool = ConnectionPool.from_url( self.config.redis_url, max_connections=20, socket_timeout=5, socket_connect_timeout=5, ) self._redis = Redis(connection_pool=self._redis_pool)

HTTP session with connection pooling

connector = aiohttp.TCPConnector( limit=100, # Total connections limit_per_host=20, # Per-host limit ttl_dns_cache=300, # DNS cache ) self._http_session = aiohttp.ClientSession(connector=connector)

async def close(self): """Close all connection pools.""" if self._pg_pool: await self._pg_pool.close() if self._redis_pool: await self._redis_pool.disconnect() if self._http_session: await self._http_session.close()

@asynccontextmanager async def db(self): """Get database connection from pool.""" async with self._pg_pool.acquire() as conn: yield conn

@property def redis(self) -> Redis: return self._redis

@property def http(self) -> aiohttp.ClientSession: return self._http_session

Anti-Patterns

---

Name

Sync I/O in Async Code

Description

Blocking calls that freeze the event loop

Why

Single blocking call stalls all concurrent operations. Defeats async purpose.

Instead

Use async versions of all I/O operations

---

Name

N+1 Queries

Description

Querying in a loop instead of batching

Why

N+1 creates N database round trips. Latency adds up linearly.

Instead

Batch queries with WHERE IN or bulk fetch

---

Name

No Connection Pooling

Description

Creating new connections for each request

Why

Connection establishment is expensive. Pool amortizes this cost.

Instead

Use connection pools for database, Redis, HTTP clients

---

Name

Cache Without Metrics

Description

Caching without measuring hit rate

Why

Cache might be worthless (low hit rate) or thrashing. You won't know.

Instead

Track hit rate, miss rate, eviction rate

---

Name

Optimizing Without Profiling

Description

"I think this is slow" without measurement

Why

Intuition is wrong. You will optimize the wrong thing.

Instead

Profile first, identify actual bottleneck, then optimize

Performance Hunter - Sharp Edges

Async Not Parallel

Id

async-not-parallel

Summary

Treating asyncio as parallelism when it's just concurrency

Severity

critical

Situation

You convert sync code to async expecting 10x speedup for CPU-bound work. Performance is the same or worse. async/await isn't making it faster.

Why

asyncio is cooperative concurrency for I/O-bound work. It runs on a single thread. CPU-bound work blocks the event loop. For parallelism, you need threads (for I/O) or processes (for CPU).

Solution

import asyncio from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

For I/O-bound: asyncio is correct

async def fetch_all(urls: List[str]) -> List[Response]:

This works because HTTP is I/O-bound

return await asyncio.gather(*[fetch(url) for url in urls])

For CPU-bound: use ProcessPoolExecutor

class CPUBoundProcessor: def __init__(self):

Process pool for CPU work

self.process_pool = ProcessPoolExecutor(max_workers=4)

Thread pool for blocking I/O

self.thread_pool = ThreadPoolExecutor(max_workers=10)

async def process_cpu_intensive(self, data: bytes) -> Result: """Run CPU-bound work in process pool.""" loop = asyncio.get_event_loop()

Run in separate process (true parallelism)

result = await loop.run_in_executor( self.process_pool, cpu_intensive_function, data, ) return result

async def call_blocking_library(self, params) -> Result: """Run blocking sync code in thread pool.""" loop = asyncio.get_event_loop()

Run in thread (doesn't block event loop)

result = await loop.run_in_executor( self.thread_pool, blocking_library_function, params, ) return result

Identify what type of work you have:

- I/O-bound (network, disk): asyncio

- CPU-bound (parsing, ML inference): ProcessPoolExecutor

- Blocking library: ThreadPoolExecutor

Symptoms

• async code no faster than sync
• Event loop blocked by CPU work
• One slow operation blocks all others
• GIL contention in profiler

Detection Pattern

async def.(?!await).for.in|asyncio.cpu_bound

Version Range

>=1.0.0

N Plus 1 Hidden

Id

n-plus-1-hidden

Summary

N+1 queries hidden in ORM or helper functions

Severity

high

Situation

Your list view is slow. You check the main query - it's fast. You enable query logging and see 100+ queries. Each item loads relations one by one.

Why

ORMs make it easy to access relations as properties. Each access triggers a query. Loops over items with relation access create N+1. It's hidden because the code looks like simple property access.

Solution

Detect N+1 with query counting

class QueryCounter: """Detect N+1 by counting queries in a request."""

def __init__(self): self.count = 0 self.queries = []

async def __aenter__(self): self.count = 0 self.queries = []

Hook into database connection

self.original_execute = db.execute db.execute = self.counting_execute return self

async def __aexit__(self, *args): db.execute = self.original_execute

if self.count > 10: logger.warning( f"Potential N+1: {self.count} queries\n" f"Queries: {self.queries[:10]}" )

async def counting_execute(self, query, args): self.count += 1 self.queries.append(str(query)[:100]) return await self.original_execute(query, args)

Usage

async def list_memories(user_id: UUID): async with QueryCounter(): memories = await get_memories(user_id)

This should be 1 query, not N+1

return [m.to_dict() for m in memories]

Fix: Eager loading or batching

class MemoryRepository: async def get_with_relations( self, user_id: UUID, ) -> List[MemoryWithRelations]:

Single query with JOIN

rows = await self.db.fetch( """ SELECT m., e., r.* FROM memories m LEFT JOIN entities e ON m.memory_id = e.memory_id LEFT JOIN relations r ON m.memory_id = r.source_id WHERE m.user_id = $1 """, user_id ) return self._assemble(rows)

Or: DataLoader pattern for GraphQL

async def batch_load_entities( self, memory_ids: List[UUID], ) -> Dict[UUID, List[Entity]]: rows = await self.db.fetch( """ SELECT * FROM entities WHERE memory_id = ANY($1) """, memory_ids )

Group by memory_id

return self._group_by_memory(rows)

Symptoms

• Query count scales with result size
• Slow list views, fast detail views
• Many similar queries in logs
• DB connection exhaustion

Detection Pattern

for.in.:.await.db|\\[.await.for.*in

Version Range

>=1.0.0

Cache Thundering Herd

Id

cache-thundering-herd

Summary

All requests hit database when cache expires

Severity

high

Situation

Your cache TTL is 5 minutes. At minute 5, the cache expires. Suddenly 100 concurrent requests all miss cache and hit the database. Database overloads. Requests timeout. Users see errors.

Why

When cached value expires, all waiting requests see cache miss simultaneously. Each request independently tries to fill the cache. You get N database hits instead of 1.

Solution

import asyncio from dataclasses import dataclass from typing import Optional, Dict import random

@dataclass class CacheEntry: value: any expires_at: float soft_expires_at: float # Refresh before hard expiry

class ThunderingHerdCache: """Cache with thundering herd protection."""

def __init__(self): self.cache: Dict[str, CacheEntry] = {} self.locks: Dict[str, asyncio.Lock] = {} self.pending: Dict[str, asyncio.Future] = {}

async def get_or_compute( self, key: str, compute_fn, ttl: int = 300, stale_ttl: int = 60, # Serve stale while refreshing ): now = time.time() entry = self.cache.get(key)

1. Fresh hit

if entry and now < entry.soft_expires_at: return entry.value

2. Stale hit - serve stale, refresh in background

if entry and now < entry.expires_at: asyncio.create_task( self._refresh_background(key, compute_fn, ttl, stale_ttl) ) return entry.value

3. Miss - need to compute (with lock)

return await self._compute_with_lock( key, compute_fn, ttl, stale_ttl )

async def _compute_with_lock( self, key: str, compute_fn, ttl: int, stale_ttl: int, ):

Check if another request is already computing

if key in self.pending: return await self.pending[key]

Take lock and compute

if key not in self.locks: self.locks[key] = asyncio.Lock()

async with self.locks[key]:

Double-check after acquiring lock

entry = self.cache.get(key) if entry and time.time() < entry.soft_expires_at: return entry.value

Create future for other waiters

future = asyncio.Future() self.pending[key] = future

try: value = await compute_fn() now = time.time()

Add jitter to prevent synchronized expiry

jitter = random.uniform(0.8, 1.2)

self.cache[key] = CacheEntry( value=value, expires_at=now + (ttl + stale_ttl) jitter, soft_expires_at=now + ttl jitter, )

future.set_result(value) return value

except Exception as e: future.set_exception(e) raise finally: del self.pending[key]

Symptoms

• Periodic latency spikes at cache expiry
• Database connection spikes every N minutes
• All users see slow response at same time
• Cache hit rate drops to 0 periodically

Detection Pattern

cache\\.get|@cached(?!.lock|.herd)

Version Range

>=1.0.0

Connection Pool Exhaustion

Id

connection-pool-exhaustion

Summary

Database connections exhausted under load

Severity

high

Situation

Traffic spikes. Requests start timing out. Database shows "too many connections." Your pool size is 10, but you have 100 concurrent requests.

Why

Connection pool limits how many concurrent DB operations are possible. When pool exhausted, requests wait for connection. Timeouts cascade. Pool too small = queueing. Pool too large = DB overload.

Solution

Right-size connection pool

import asyncpg

class DatabasePool: """Database pool with monitoring."""

def __init__(self, config: Config): self.config = config self.pool = None

async def initialize(self): self.pool = await asyncpg.create_pool( dsn=self.config.database_url, min_size=5, # Baseline connections max_size=20, # Max concurrent queries

Don't wait forever for connection

max_inactive_connection_lifetime=300,

Query timeout prevents runaway queries

command_timeout=30, )

async def get_pool_stats(self) -> PoolStats: return PoolStats( size=self.pool.get_size(), min_size=self.pool.get_min_size(), max_size=self.pool.get_max_size(), free_size=self.pool.get_idle_size(), )

async def health_check(self) -> bool: stats = await self.get_pool_stats()

Alert if pool is near exhaustion

utilization = (stats.size - stats.free_size) / stats.max_size if utilization > 0.8: logger.warning( f"Connection pool {utilization*100:.0f}% utilized" )

return stats.free_size > 0

Rule of thumb for pool sizing:

max_connections = (cores * 2) + spinning_disks

For SSD: max_connections = cores * 2

For cloud DB: check provider limits

Connection pool per external service

class ConnectionManager: pools = { "postgres": None, "redis": None, "qdrant": None, }

async def initialize_all(self):

Each service gets its own pool

self.pools["postgres"] = await asyncpg.create_pool(...) self.pools["redis"] = redis.ConnectionPool(max_connections=50) self.pools["qdrant"] = QdrantClient( grpc_options={"grpc.max_concurrent_streams": 100} )

Symptoms

• Connection timeout errors
• 'Too many connections' from database
• Latency spikes under load
• Pool stats show 0 free connections

Detection Pattern

create_pool|Pool\\((?!.*max_size)

Version Range

>=1.0.0

P99 Ignored

Id

p99-ignored

Summary

Optimizing for average while p99 kills users

Severity

medium

Situation

Average latency is 50ms. Looks great! But 1% of users wait 5 seconds. They complain, leave, blame your product. You didn't notice because you only tracked average.

Why

Average hides tail latency. One slow database query, one cold cache, one garbage collection - these affect the unlucky 1%. They experience your product as slow.

Solution

from prometheus_client import Histogram, Summary import statistics

Track percentiles, not just averages

LATENCY_HISTOGRAM = Histogram( 'request_latency_seconds', 'Request latency in seconds', ['endpoint'], buckets=[0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0] )

class LatencyTracker: """Track and analyze latency distributions."""

def __init__(self, window_size: int = 1000): self.samples = [] self.window_size = window_size

def record(self, latency_ms: float): self.samples.append(latency_ms) if len(self.samples) > self.window_size: self.samples.pop(0)

def get_percentiles(self) -> dict: if not self.samples: return {}

sorted_samples = sorted(self.samples) n = len(sorted_samples)

return { "p50": sorted_samples[int(n 0.50)], "p90": sorted_samples[int(n 0.90)], "p95": sorted_samples[int(n 0.95)], "p99": sorted_samples[int(n 0.99)], "max": sorted_samples[-1], "avg": statistics.mean(sorted_samples), }

def alert_on_p99(self, threshold_ms: float) -> bool: percentiles = self.get_percentiles() if percentiles.get("p99", 0) > threshold_ms: logger.warning( f"p99 latency {percentiles['p99']}ms exceeds " f"threshold {threshold_ms}ms" ) return True return False

SLO based on percentiles

class LatencySLO: """Service Level Objective for latency."""

TARGETS = { "retrieval": {"p50": 50, "p99": 500}, "embedding": {"p50": 100, "p99": 1000}, "graph_query": {"p50": 20, "p99": 200}, }

async def check_slo(self, operation: str) -> SLOResult: tracker = self.trackers[operation] percentiles = tracker.get_percentiles() targets = self.TARGETS[operation]

violations = [] for metric, target in targets.items(): actual = percentiles.get(metric, 0) if actual > target: violations.append(f"{metric}: {actual}ms > {target}ms")

return SLOResult( operation=operation, meeting_slo=len(violations) == 0, violations=violations, )

Symptoms

• Users complain about slowness but dashboards look fine
• Occasional timeout errors
• Large variance in response times
• Only tracking average latency

Detection Pattern

latency.mean|average.latency(?!.p99|.percentile)

Version Range

>=1.0.0

Embedding Not Cached

Id

embedding-not-cached

Summary

Re-embedding same text repeatedly

Severity

medium

Situation

Same query gets embedded every time it's searched. Same content re-embedded on every startup. Embedding API costs are huge.

Why

Embeddings are deterministic: same input → same output. Re-computing wastes API calls, adds latency. Embedding models are expensive.

Solution

import hashlib from functools import lru_cache

class EmbeddingCache: """Cache embeddings by content hash."""

def __init__(self, redis_client, embedder): self.redis = redis_client self.embedder = embedder self.local_cache = {} # LRU for hot embeddings self.local_cache_size = 10000

def _content_hash(self, text: str) -> str: return hashlib.sha256(text.encode()).hexdigest()[:16]

async def embed(self, text: str) -> List[float]: cache_key = f"emb:{self._content_hash(text)}"

L1: Local memory

if cache_key in self.local_cache: return self.local_cache[cache_key]

L2: Redis

cached = await self.redis.get(cache_key) if cached: embedding = self._deserialize(cached) self._add_to_local(cache_key, embedding) return embedding

Miss: Compute and cache

embedding = await self.embedder.embed(text)

Cache in both levels

await self.redis.set( cache_key, self._serialize(embedding), ex=86400 * 30, # 30 day TTL ) self._add_to_local(cache_key, embedding)

return embedding

async def embed_batch( self, texts: List[str], ) -> List[List[float]]: """Batch embed with cache lookup.""" results = [None] * len(texts) to_embed = [] to_embed_indices = []

Check cache for each

for i, text in enumerate(texts): cache_key = f"emb:{self._content_hash(text)}" cached = await self.redis.get(cache_key)

if cached: results[i] = self._deserialize(cached) else: to_embed.append(text) to_embed_indices.append(i)

Batch embed misses

if to_embed: new_embeddings = await self.embedder.embed_batch(to_embed)

for idx, embedding in zip(to_embed_indices, new_embeddings): results[idx] = embedding cache_key = f"emb:{self._content_hash(texts[idx])}" await self.redis.set(cache_key, self._serialize(embedding))

return results

Symptoms

• High embedding API costs
• Same texts embedded repeatedly
• Search latency includes embedding time
• No embedding cache hit metrics

Detection Pattern

embed\\(|embed_batch\\((?!.*cache)

Version Range

>=1.0.0

Performance Hunter - Validations

Synchronous I/O in Async Function

Id

sync-in-async

Severity

error

Type

regex

Pattern

• async def.:.requests\\.get
• async def.:.open\\(.*\\)\\.read
• async def.:.time\\.sleep
• async def[^:]+:[^$]*(?<!await )psycopg2\\.

Message

Synchronous I/O in async function. Blocks event loop.

Fix Action

Use async version: aiohttp, aiofiles, asyncio.sleep

Applies To

• */.py

Database Query in Loop

Id

n-plus-one-loop

Severity

error

Type

regex

Pattern

• for.in.:.await.db\\.
• for.in.:.await.fetch
• for.in.:.await.execute
• \\[await.db.for.*in

Message

Database query inside loop. Likely N+1 pattern.

Fix Action

Batch queries with WHERE IN or use DataLoader pattern

Applies To

• */.py

Creating Connections Without Pool

Id

no-connection-pool

Severity

warning

Type

regex

Pattern

• asyncpg\\.connect\\(
• psycopg2\\.connect\\(
• redis\\.Redis\\((?!.*connection_pool)

Message

Creating individual connections instead of using pool.

Fix Action

Use connection pool: asyncpg.create_pool(), ConnectionPool

Applies To

• */.py

Cache Without TTL

Id

cache-no-ttl

Severity

warning

Type

regex

Pattern

• cache\\.set\\([^)]\\)(?!.ttl|.ex=|.expire)
• redis\\.set\\([^)]\\)(?!.ex=|.*px=)

Message

Cache set without TTL. Data may become stale indefinitely.

Fix Action

Add TTL: cache.set(key, value, ttl=3600)

Applies To

• */.py

External Call Without Timeout

Id

no-timeout

Severity

warning

Type

regex

Pattern

• aiohttp.get\\([^)]\\)(?!.*timeout)
• httpx.get\\([^)]\\)(?!.*timeout)
• requests\\.get\\([^)]\\)(?!.timeout)

Message

External HTTP call without timeout. May hang indefinitely.

Fix Action

Add timeout: session.get(url, timeout=30)

Applies To

• */.py

Unbounded Collection Growth

Id

unbounded-memory

Severity

warning

Type

regex

Pattern

• results\\.append.*while True
• cache\\[.\\].=(?!.*if len)
• list\\.append.for.in.*yield

Message

Collection may grow unbounded. Potential memory leak.

Fix Action

Add size limits or use generators for large data

Applies To

• */.py

Embedding Without Caching

Id

embed-no-cache

Severity

info

Type

regex

Pattern

• embed\\(.\\)(?!.cache)
• embedder\\.embed(?!.*cached)

Message

Embedding without cache check. Same content re-embedded.

Fix Action

Cache embeddings by content hash

Applies To

• /embedding//*.py
• /retrieval//*.py

Sequential Processing Without Batching

Id

no-batch-processing

Severity

info

Type

regex

Pattern

• for.in.:.await.embed
• for.in.:.await.llm
• for.in.:.await.api

Message

Sequential API calls. Consider batching for efficiency.

Fix Action

Use asyncio.gather() or batch APIs

Applies To

• */.py

Database Query Without LIMIT

Id

query-no-limit

Severity

info

Type

regex

Pattern

• SELECT.FROM(?!.LIMIT|.TOP|.FETCH)
• fetch\\([^)]\\)(?!.limit)

Message

Query without LIMIT. May return unbounded results.

Fix Action

Add LIMIT or implement pagination

Applies To

• */.py

Expensive Logging in Hot Path

Id

log-in-hot-path

Severity

info

Type

regex

Pattern

• for.:.logger\\.debug
• while.:.logger\\.info
• logger.*json\\.dumps

Message

Logging in tight loop. String formatting adds overhead.

Fix Action

Use lazy logging: logger.debug('%s', value)

Applies To

• */.py

Query On Unindexed Column

Id

no-index-hint

Severity

info

Type

regex

Pattern

• WHERE.content.=
• WHERE.description.LIKE

Message

Query on potentially unindexed text column. May be slow.

Fix Action

Verify index exists or use full-text search

Applies To

• */.py

JSON Serialization in Loop

Id

json-in-loop

Severity

info

Type

regex

Pattern

• for.:.json\\.dumps
• for.:.json\\.loads

Message

JSON serialization in loop. Consider batch processing.

Fix Action

Serialize outside loop or use streaming JSON parser

Applies To

• */.py