Z
Zappio Team
AI & Real Estate Experts · 6 July 2026 · 13 min read
Loading article...
Zappio Team
AI & Real Estate Experts · 6 July 2026 · 13 min read
A real estate developer committing to their first AI Calling deployment has one non-negotiable requirement: the system must not fail during a campaign launch. A ₹4 lakh weekend Meta campaign generating 300 leads on a Saturday morning — with buyers who expressed property interest 47 minutes ago — is not the moment to discover that the LLM inference layer saturates at 25 concurrent calls, or that the telephony streaming API returns connection errors above 50 simultaneous sessions, or that the CRM webhook times out under batch load and loses disposition data.
These are documented failure modes in AI Calling deployments that skipped structured pre-launch load testing. The cost of each failure: lead pool degradation, wasted ad spend, and permanent brand damage in a market where developer reputation is a primary purchase driver. This article provides the complete pre-launch load testing framework for a real estate AI Calling system — the test categories, the specific scripts to run, the pass/fail thresholds, and the go-live decision criteria that determine whether a system is production-ready.
Standard API load testing sends HTTP requests at scale and measures response time and error rate. AI Calling load testing must simulate the full conversational pipeline under concurrent load — not just the API endpoints, but the entire sequential chain: telephony audio stream, ASR transcription, NLU entity extraction, LLM inference, TTS synthesis, audio playback, and CRM write-back — running simultaneously across 50, 100, or 200 concurrent calls.
Before testing the full pipeline, validate each component's throughput ceiling individually.
async def test_asr_throughput(
concurrent_streams: int,
duration_seconds: int = 60,
audio_sample_path: str = "test_audio_8khz_30sec.wav"
):
"""
Sends concurrent ASR streams and measures:
- Success rate
- p50/p95/p99 latency
- Error rate by type
"""
with open(audio_sample_path, "rb") as f:
audio_bytes = f.read()
async def single_asr_call(session_id: int):
start = time.monotonic()
try:
result = await asr_client.transcribe_async(
audio=audio_bytes,
language="hinglish",
session_id=f"loadtest-{session_id}"
)
latency = (time.monotonic() - start) * 1000
return {"success": True, "latency_ms": latency, "confidence": result.confidence}
except Exception as e:
return {"success": False, "error": str(e), "latency_ms": None}
tasks = [single_asr_call(i) for i in range(concurrent_streams)]
batch_results = await asyncio.gather(*tasks)
success_count = sum(1 for r in batch_results if r["success"])
latencies = [r["latency_ms"] for r in batch_results if r["latency_ms"]]
print(f"ASR Throughput Test @ {concurrent_streams} concurrent:")
print(f" Success rate: {success_count}/{concurrent_streams}")
print(f" Latency p50: {sorted(latencies)[len(latencies)//2]:.0f}ms")
print(f" Latency p95: {sorted(latencies)[int(len(latencies)*0.95)]:.0f}ms")
return batch_results
# Run at: 10, 25, 50, 100, 150 concurrent streams
for concurrency in [10, 25, 50, 100, 150]:
asyncio.run(test_asr_throughput(concurrency))Pass criteria: ASR success rate ≥ 99% at target concurrency with P95 latency ≤ 350ms. LLM inference: P95 ≤ 600ms, error rate under 0.5%, no rate limit errors. TTS: P95 first-byte ≤ 250ms at target concurrency. CRM write-back: 100% write success with no rate-limit errors at expected peak batch size.
The end-to-end test runs the full AI Calling pipeline using recorded test audio — not live PSTN calls — at increasing concurrency levels, covering scenarios like an ideal buyer, budget mismatch, soft objection, hard transfer trigger, Hinglish buyer, and poor audio quality.
class SimulatedCallLoadTest:
"""
Runs full AI Calling pipeline simulation:
Audio playback → ASR → NLU → LLM → TTS → CRM write
Without initiating actual PSTN calls.
"""
TEST_SCENARIOS = [
("test_qualified_3bhk_1cr.wav", "qualified_site_visit_booked", "Ideal buyer"),
("test_budget_mismatch.wav", "disqualified_budget", "Budget below floor"),
("test_objection_call_later.wav", "qualified_callback_scheduled", "Soft objection"),
("test_human_request.wav", "transferred_human_class1", "Hard transfer trigger"),
("test_hinglish_mixed.wav", "qualified_site_visit_booked", "Hinglish buyer"),
("test_silent_pause_heavy.wav", "no_answer", "Poor audio quality"),
]
async def run_concurrent_scenario_batch(
self,
concurrency: int,
scenario_distribution: dict = None
) -> LoadTestReport:
if not scenario_distribution:
scenario_distribution = {
"test_qualified_3bhk_1cr.wav": 0.35,
"test_budget_mismatch.wav": 0.20,
"test_objection_call_later.wav": 0.25,
"test_human_request.wav": 0.08,
"test_hinglish_mixed.wav": 0.10,
"test_silent_pause_heavy.wav": 0.02,
}
scenarios = self._sample_scenarios(concurrency, scenario_distribution)
start_time = time.monotonic()
tasks = [self._simulate_single_call(s) for s in scenarios]
results = await asyncio.gather(*tasks, return_exceptions=True)
total_duration = time.monotonic() - start_time
return LoadTestReport(
concurrency=concurrency,
total_duration_seconds=total_duration,
results=results,
timestamp=datetime.utcnow()
)Pass criteria at target concurrency: end-to-end success rate (call started to disposition written) at least 99%, correct disposition rate at least 92%, end-to-end P95 latency under 1,200ms, zero Redis key collisions, and CRM write success rate at 100%.
Real estate campaign launches generate non-linear load spikes — 300 leads arriving over 4 hours from a Saturday morning Meta campaign, with the first 90 minutes handling 70% of the volume. The burst test simulates ramp-up, sustained peak, and ramp-down phases.
async def burst_load_test(
peak_concurrency: int = 150,
ramp_up_minutes: int = 15,
sustained_peak_minutes: int = 30,
ramp_down_minutes: int = 15
):
phases = [
("ramp_up", ramp_up_minutes * 60, 0, peak_concurrency),
("sustained", sustained_peak_minutes * 60, peak_concurrency, peak_concurrency),
("ramp_down", ramp_down_minutes * 60, peak_concurrency, 10)
]
for phase_name, duration_sec, start_conc, end_conc in phases:
print(f"Starting phase: {phase_name}")
steps = 10
for step in range(steps):
current_conc = int(start_conc + (end_conc - start_conc) * (step / steps))
await run_concurrent_batch(current_conc)
await asyncio.sleep(duration_sec / steps)Pass criteria: zero system errors during ramp-up, P95 latency at or under 1,400ms at sustained peak, no telephony concurrency ceiling errors, Redis memory usage under 80% of allocated capacity during peak, and a CRM write queue depth under 50 pending writes.
Production systems encounter failures. The failure recovery test verifies the system degrades gracefully and recovers automatically across four scenarios: an ASR provider timeout at 40% of calls (expect fallback to a secondary ASR provider within 500ms, no call drop); an LLM inference rate limit hit at 80% of target concurrency (expect queued inference with a holding phrase and no dropped calls); a CRM webhook returning errors for 5 minutes (expect disposition data buffered locally and written successfully after recovery, with zero data loss); and a telephony stream disconnect mid-call (expect the session marked interrupted with state preserved for reconnect and no orphaned sessions).
Pass criteria for all four failure tests: zero data loss, zero call drops due to infrastructure failure (buyer-initiated drops excluded), and auto-recovery within defined time bounds.
Memory leaks, Redis TTL expiry edge cases, and connection pool exhaustion manifest only under sustained operation. The soak test runs the system at 60% of target concurrency for 8 consecutive hours, monitoring process memory of the call orchestrator (should be flat — rising indicates a memory leak), Redis connection pool utilization (should stay under 60% steady-state), database connection pool for leaked connections, and LLM API token consumption rate (should be proportional to call count — drift indicates prompt length creep).
Pass criteria: no metric trending upward over 8 hours, with P95 latency at hour 8 within 10% of the hour 1 baseline.
After all load tests pass, the go-live decision requires sign-off on a checklist spanning three readiness dimensions.
| Criterion | Go | No-Go |
|---|---|---|
| E2E pipeline test success rate | ≥ 99% | < 99% |
| CRM write success rate | 100% | < 100% |
| Burst test zero errors | Pass | Any error |
| Failure recovery tests | All 4 pass | Any failure |
| Soak test memory stable | Pass | Any upward trend |
| Monitoring alerts wired | All configured | Any gap |
| Human fallback tested live | Pass | Not tested |
If any No-Go criterion is present, do not launch. Delay by one business day, address the failing criterion, and re-run only the affected test category.
A production-ready AI Calling system is not defined by whether it works in a demo call — it is defined by whether it holds up under the exact concurrency, burst pattern, and failure conditions a real campaign launch will throw at it. The five load testing categories and the three-part go-live checklist in this article are the minimum bar for any deployment handling live buyer leads at meaningful volume. Treat the go/no-go table as a hard gate, not a guideline — the cost of delaying a launch by a day is trivial compared to the cost of degrading a lead pool that will never re-warm to the same intent level.
Disclaimer: Load testing thresholds, go-live criteria, and infrastructure recommendations in this article are based on production real estate AI Calling deployments as of Q2 2026. Target concurrency, acceptable latency bounds, and failure recovery time requirements vary based on deployment scale, telephony provider capabilities, lead volume, and organizational risk tolerance. All code samples are illustrative and require adaptation to specific infrastructure environments. Pre-launch testing does not guarantee zero production incidents — monitoring and alerting remain essential post-launch.