{"title":"Exercise: Async Python","description":null,"content":"# Exercise 1: Run API Calls Concurrently\n\nTake three simulated API calls that currently run one after another and make them run concurrently with `asyncio.gather`, so the total time is set by the slowest call rather than the sum of all three.\n\nWhen an application makes several independent model calls, running them sequentially wastes time: each call spends most of its life waiting on the network while the CPU sits idle. `asyncio.gather` schedules the calls together and lets the event loop overlap that waiting, which is one of the simplest, highest-impact speedups in an AI application.\n\n### Requirements\n\n- Keep the `call_llm` coroutine, which simulates a network call with `asyncio.sleep`.\n- Run all three calls concurrently with `asyncio.gather` instead of awaiting them one at a time.\n- Print each result and the total elapsed time.\n- The elapsed time should be close to the longest single delay (about 0.3s), not their sum (0.6s).\n\n\n**main.py**\n\n```python\nimport asyncio\nimport time\n\nasync def call_llm(prompt: str, delay: float) -> str:\n # Simulates the network wait of a real API call\n await asyncio.sleep(delay)\n return f\"Response to: {prompt}\"\n\nasync def main():\n start = time.perf_counter()\n\n # TODO: run these three calls concurrently with asyncio.gather\n # call_llm(\"What is Python?\", 0.2)\n # call_llm(\"What is async?\", 0.3)\n # call_llm(\"What is an LLM?\", 0.1)\n\n # TODO: print each result\n\n elapsed = time.perf_counter() - start\n print(f\"Total time: {elapsed:.2f}s\")\n\nasyncio.run(main())\n```\n\n\n### Expected Output\n\nAll three responses print, and the total time is close to the single longest delay rather than the sum.\n\n\n```plaintext\nResponse to: What is Python?\nResponse to: What is async?\nResponse to: What is an LLM?\nTotal time: 0.30s\n```\n\n\n### Solution\n\n\n**main.py**\n\n```python\nimport asyncio\nimport time\n\nasync def call_llm(prompt: str, delay: float) -> str:\n await asyncio.sleep(delay)\n return f\"Response to: {prompt}\"\n\nasync def main():\n start = time.perf_counter()\n\n results = await asyncio.gather(\n call_llm(\"What is Python?\", 0.2),\n call_llm(\"What is async?\", 0.3),\n call_llm(\"What is an LLM?\", 0.1),\n )\n\n for result in results:\n print(result)\n\n elapsed = time.perf_counter() - start\n print(f\"Total time: {elapsed:.2f}s\")\n\nasyncio.run(main())\n```\n\n\n`asyncio.gather` schedules all three coroutines on the event loop at once and waits for them together. While each one is parked on `await asyncio.sleep`, the loop runs the others, so the three waits overlap and the total time collapses to the longest single delay. The returned list preserves the order of the arguments, which is why you can match results back to their prompts without extra bookkeeping.\n\n---\n\n# Exercise 2: Handle Failures in Concurrent Calls\n\nWhen you fan out many calls at once, some will fail while others succeed. By default `asyncio.gather` cancels everything if one task raises. Use `return_exceptions=True` so the gather completes, collecting each task's result or its exception, and then handle them individually.\n\nThis is the difference between a fragile batch (one failure loses every result) and a resilient one (you keep the successes and deal with the failures). It is the standard pattern for running independent model or API calls in parallel.\n\n### Requirements\n\n- Keep the `call_llm` coroutine, where one call is set to fail.\n- Run all three concurrently with `asyncio.gather(..., return_exceptions=True)`.\n- Iterate the results, printing successes and failures separately by checking `isinstance(result, Exception)`.\n\n\n**main.py**\n\n```python\nimport asyncio\n\nasync def call_llm(prompt: str, delay: float, fail: bool = False) -> str:\n await asyncio.sleep(delay)\n if fail:\n raise RuntimeError(f\"call failed for: {prompt}\")\n return f\"ok: {prompt}\"\n\nasync def main():\n tasks = [\n call_llm(\"a\", 0.1),\n call_llm(\"b\", 0.1, fail=True),\n call_llm(\"c\", 0.1),\n ]\n\n # TODO: run all tasks concurrently with asyncio.gather(..., return_exceptions=True)\n\n # TODO: print each result, labeling exceptions as failures and the rest as successes\n\nasyncio.run(main())\n```\n\n\n### Expected Output\n\nThe two good calls return their results; the failing one is reported, not raised, and does not lose the others.\n\n\n```plaintext\nSUCCESS: ok: a\nFAILED: call failed for: b\nSUCCESS: ok: c\n```\n\n\n### Solution\n\n\n**main.py**\n\n```python\nimport asyncio\n\nasync def call_llm(prompt: str, delay: float, fail: bool = False) -> str:\n await asyncio.sleep(delay)\n if fail:\n raise RuntimeError(f\"call failed for: {prompt}\")\n return f\"ok: {prompt}\"\n\nasync def main():\n tasks = [\n call_llm(\"a\", 0.1),\n call_llm(\"b\", 0.1, fail=True),\n call_llm(\"c\", 0.1),\n ]\n\n results = await asyncio.gather(*tasks, return_exceptions=True)\n\n for result in results:\n if isinstance(result, Exception):\n print(f\"FAILED: {result}\")\n else:\n print(f\"SUCCESS: {result}\")\n\nasyncio.run(main())\n```\n\n\n`return_exceptions=True` changes gather from all-or-nothing to collect-everything: a raised exception becomes an item in the results list instead of propagating and cancelling the siblings. Checking `isinstance(result, Exception)` lets you separate the successes from the failures and decide what to do with each, such as retrying only the failed calls. The results stay in the order you submitted the tasks, so you can map them back to their inputs.","pageType":"ai-engineering"}

Get Premium