Guidelines for profiling, measuring, and optimizing async Python performance across VoxAgent's voice pipeline. Covers event loop health, pipeline instrumentation, provider latency, and benchmarking.
-
perf-no-sync-in-async — NEVER run synchronous/blocking code in async functions. Common violations:
data = open("file.txt").read()
result = requests.get(url)
time.sleep(1)
json.loads(huge_string)
subprocess.run(["cmd"])
async with aiofiles.open("file.txt") as f:
data = await f.read()
-
perf-loop-monitor — Detect event loop blocking with a monitor in development:
import asyncio
LOOP_BLOCK_THRESHOLD_MS = 100
def enable_loop_monitor() -> None:
loop = asyncio.get_event_loop()
loop.slow_callback_duration = LOOP_BLOCK_THRESHOLD_MS / 1000
loop.set_debug(True)
-
perf-gather-bounded — Use asyncio.gather() for parallel I/O, but limit concurrency to avoid overwhelming external services:
MAX_CONCURRENT_REQUESTS = 5
async def check_all_providers(providers: list[Provider]) -> list[bool]:
semaphore = asyncio.Semaphore(MAX_CONCURRENT_REQUESTS)
async def check_with_limit(p: Provider) -> bool:
async with semaphore:
return await p.health_check()
return await asyncio.gather(*[check_with_limit(p) for p in providers])
-
perf-task-cancellation — Always handle asyncio.CancelledError. Clean up resources on cancellation:
async def long_running_task() -> None:
try:
while True:
await process_next()
except asyncio.CancelledError:
await cleanup()
raise
-
perf-timing-decorator — Use a reusable timing decorator for all pipeline stages:
import time
import functools
import logging
logger = logging.getLogger(__name__)
def timed(stage: str):
def decorator(fn):
@functools.wraps(fn)
async def wrapper(*args, **kwargs):
start = time.perf_counter()
try:
return await fn(*args, **kwargs)
finally:
elapsed_ms = (time.perf_counter() - start) * 1000
logger.info("%s completed in %.1fms", stage, elapsed_ms)
return wrapper
return decorator
@timed("stt")
async def transcribe(audio: bytes) -> str:
...
-
perf-pipeline-trace — Collect timing for all pipeline stages into a single trace object:
@dataclass
class PipelineTrace:
request_id: str
stages: dict[str, float] = field(default_factory=dict)
def record(self, stage: str, elapsed_ms: float) -> None:
self.stages[stage] = elapsed_ms
@property
def total_ms(self) -> float:
return sum(self.stages.values())
def over_budget(self, budgets: dict[str, float]) -> list[str]:
return [s for s, ms in self.stages.items() if ms > budgets.get(s, float("inf"))]
-
perf-structured-metrics — Log metrics in structured format for analysis:
logger.info(
"pipeline_complete",
extra={
"request_id": trace.request_id,
"total_ms": trace.total_ms,
"stages": trace.stages,
"over_budget": trace.over_budget(LATENCY_BUDGETS),
},
)
-
perf-budget — Define and enforce latency budgets. Warn when a stage exceeds its budget:
LATENCY_BUDGETS: dict[str, float] = {
"stt": 500.0,
"intent": 50.0,
"llm": 3000.0,
"skill": 2000.0,
"tts": 800.0,
}
async def check_budget(trace: PipelineTrace) -> None:
violations = trace.over_budget(LATENCY_BUDGETS)
if violations:
logger.warning("Latency budget exceeded: %s", violations)
-
perf-provider-timing — Track per-provider latency with rolling averages:
from collections import deque
ROLLING_WINDOW = 100
class ProviderMetrics:
def __init__(self) -> None:
self._latencies: deque[float] = deque(maxlen=ROLLING_WINDOW)
def record(self, latency_ms: float) -> None:
self._latencies.append(latency_ms)
@property
def avg_ms(self) -> float:
return sum(self._latencies) / len(self._latencies) if self._latencies else 0.0
@property
def p95_ms(self) -> float:
if not self._latencies:
return 0.0
sorted_vals = sorted(self._latencies)
idx = int(len(sorted_vals) * 0.95)
return sorted_vals[idx]
-
perf-connection-pool — Reuse HTTP connections. Never create a new httpx.AsyncClient per request:
class Provider:
def __init__(self):
self._client = httpx.AsyncClient(limits=httpx.Limits(max_connections=10))
async def call_api(self):
async with httpx.AsyncClient() as client:
...
-
perf-streaming-prefer — Use streaming responses when available. Start processing before the full response arrives:
async def stream_llm(messages: list[Message]) -> AsyncGenerator[str, None]:
async with client.stream("POST", url, json=payload) as response:
async for chunk in response.aiter_text():
yield chunk
-
perf-cache-responses — Cache deterministic responses (embeddings, classifications) with TTL:
from functools import lru_cache
from asyncio import Lock
_cache: dict[str, tuple[float, Any]] = {}
_cache_lock = Lock()
CACHE_TTL_SECONDS = 300
async def cached_classify(text: str) -> str:
async with _cache_lock:
if text in _cache:
ts, result = _cache[text]
if time.time() - ts < CACHE_TTL_SECONDS:
return result
result = await classify(text)
async with _cache_lock:
_cache[text] = (time.time(), result)
return result
-
perf-benchmark-isolated — Benchmark one thing at a time. Isolate network, disk, and CPU benchmarks:
@pytest.mark.benchmark
async def test_intent_classification_speed():
classifier = IntentClassifier()
inputs = [make_test_input() for _ in range(100)]
start = time.perf_counter()
for inp in inputs:
await classifier.classify(inp)
elapsed = time.perf_counter() - start
avg_ms = (elapsed / len(inputs)) * 1000
assert avg_ms < 50, f"Intent classification too slow: {avg_ms:.1f}ms (budget: 50ms)"
-
perf-regression-baseline — Store baseline measurements. Fail CI if regression exceeds 20%:
BASELINE_MS = {
"stt_transcribe": 450.0,
"intent_classify": 35.0,
"llm_chat": 2500.0,
}
REGRESSION_THRESHOLD = 1.2
-
perf-profile-async — Use pyinstrument for async profiling (not cProfile, which misses await time):
from pyinstrument import Profiler
async def profile_pipeline():
profiler = Profiler(async_mode="enabled")
profiler.start()
await run_pipeline(test_input)
profiler.stop()
print(profiler.output_text(unicode=True))
