LLM - Day 7 - Async & Virtual Environment

Sync

• One after another.
• f1() , f2() , f3() & f4()
• Once f1() completes then only f2() starts , f1() I/O Operation ( DB qwery / API qwery response / File operations ).

• There are some operation running on top of CPU.
• I/O operation won't occupy more CPU.
• But below function ,

def add(a,b):
return a+b

• But the above function keeps the CPU is active and its utilized.

def getweatherdetail()
CPU --> Request --> HTTP API --> x secs we are waiting for my reply. --> this is not controlled by the CPU.

f2()
f3()

All these functions are in main function ()
weather()
f2()
f3()

• One step after the another , f2() & f3() எதுக்கு wait பண்றாங்கன்னு தெரியாது. இருந்து வெயிட் பண்ணி முடிச்சிட்ட பிறகு அடுத்து நடக்கும்.

Async

• weather() --> Request sent and waiting. In the mean time f2() and f3() will be called.
• No wait.

Example

• Looks like parallel execution but its context switching.
• Overall total wait is only 3 secs NOT 5 secs.
• await --> tells like , to wait for the wait for the function to finish.
• Eg.,

import asyncio
async def make_tea():
    print("Boiling water...")
    await asyncio.sleep(3)  # time taking process.
    print("Tea is ready!")
async def make_toast():
    print("Toasting bread...")
    await asyncio.sleep(2)
    print("Toast is ready!")
async def main():
    await asyncio.gather(make_toast(), make_tea())
asyncio.run(main())

gather --> its going to call sequentially --> each functions is called " Coroutine "

Sync Code

import time
def make_tea():
    print("Boiling water...")
    time.sleep(3)
    print("Tea is ready!")
def make_toast():
    print("Toasting bread...")
    time.sleep(3)
    print("Toast is ready!")
def main():
    make_toast()
    make_tea()
main()

Async Code

import asyncio
async def hello():
print("Hello")
await asyncio.sleep(2)
print("World")
asyncio.run(hello())

Sequential_async

import asyncio
import time
async def task(name, seconds):
    print(f"{name} started")
    await asyncio.sleep(seconds)
    print(f"{name} completed")
async def main():
    start = time.perf_counter()
    await task("Task 1", 2)
    await task("Task 2", 2)
    await task("Task 3", 2)
    end = time.perf_counter()
    print(f"Total time: {end - start:.2f} seconds")
asyncio.run(main())

Gather

import asyncio
import time
async def task(name, seconds):
    print(f"{name} started")
    await asyncio.sleep(seconds)
    print(f"{name} completed")
async def main():
    start = time.perf_counter()
    await asyncio.gather(
        task("Task 1", 2),
        task("Task 2", 2),
        task("Task 3", 2),
    )
    end = time.perf_counter()
    print(f"Total time: {end - start:.2f} seconds")
asyncio.run(main())

Gather return value

import asyncio
async def get_user():
    await asyncio.sleep(2)
    return "User"
async def get_orders():
    await asyncio.sleep(3)
    return "Orders"
async def get_products():
    await asyncio.sleep(1)
    return "Products"
async def main():
    user, orders, products = await asyncio.gather(
        get_user(),
        get_orders(),
        get_products(),
    )
    print(user)
    print(orders)
    print(products)
asyncio.run(main())

• WAIT will be PARALLEL but EXECUTION will be CONCURRENT.

Using API Call

import asyncio
import httpx
async def fetch(client, url):
    print(f"Requesting {url}")
    response = await client.get(url)
    print(f"{url} → {response.status_code}")
    return response.status_code
async def main():
    urls = [
        "https://example.com",
        "https://httpbin.org/get",
        "https://httpbin.org/uuid",
    ]
    async with httpx.AsyncClient() as client:
        results = await asyncio.gather(*(fetch(client, url) for url in urls))
    print(results)
asyncio.run(main())

IC Query

import asyncio
from dotenv import load_dotenv
from langchain_groq import ChatGroq
load_dotenv()
llm = ChatGroq(
    model="openai/gpt-oss-120b",
    temperature=0,
)
async def main():
    response = await llm.ainvoke("What is LangChain in one sentence?")
    print(response.content)
asyncio.run(main())

crm

import asyncio
from dotenv import load_dotenv
from langchain_core.tools import tool
from langchain_groq import ChatGroq
from langgraph.prebuilt import create_react_agent
load_dotenv(
@tool
def get_weather(city: str) -> str:
    """Get the weather for a city."""
    return f"It's sunny and 32°C in {city}"
llm = ChatGroq(model="openai/gpt-oss-120b", temperature=0)
agent = create_react_agent(model=llm, tools=[get_weather])
# async def main():
#     response = await agent.ainvoke(
#         {"messages": [{"role": "user", "content": "What's the weather in Coimbatore?"}]}
#     )
#     print(response["messages"][-1].content)
# async def main():
#     async for chunk in agent.astream(
#         {
#             "messages": [
#                 {"role": "user", "content": "Explain async in Python simply in detail "}
#             ]
#         }
#     ):
#         print(chunk)
#         print("#" * 30)
async def main():
    results = await asyncio.gather(
        agent.ainvoke(
            {"messages": [{"role": "user", "content": "Weather in Chennai?"}]}
        ),
        agent.ainvoke(
            {"messages": [{"role": "user", "content": "Weather in Bangalore?"}]}
        ),
        agent.ainvoke(
            {"messages": [{"role": "user", "content": "Weather in Mumbai?"}]}
        ),
    )
    for r in results:
        print(r["messages"][-1].content)
asyncio.run(main())

{"role": "user", "content": "Explain async in Python simply in detail "} --> Lets stream this , Explain async in Python simply ... ( No tool involved here )

API

app.py

from dotenv import load_dotenv
from fastapi import FastAPI
from langchain_core.tools import tool
from langchain_groq import ChatGroq
from langgraph.prebuilt import create_react_agent
from pydantic import BaseModel

load_dotenv()
app = FastAPI()

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"It's sunny and 32°C in {city}"

@tool
def get_time(city: str) -> str:
    """Get the current time for a city (mocked)."""
    return f"It's 4:30 PM in {city}"
llm = ChatGroq(model="openai/gpt-oss-120b", temperature=0)
agent = create_react_agent(model=llm, tools=[get_weather, get_time])
class ChatRequest(BaseModel):
    message: str
class ChatResponse(BaseModel):
    reply: str
@app.post("/chat", response_model=ChatResponse)
async def chat(req: ChatRequest):
    response = await agent.ainvoke(
        {"messages": [{"role": "user", "content": req.message}]}
    )
    reply = response["messages"][-1].content
    return ChatResponse(reply=reply)

app_sync.py

from dotenv import load_dotenv
from fastapi import FastAPI
from langchain_core.tools import tool
from langchain_groq import ChatGroq
from langgraph.prebuilt import create_react_agent
from pydantic import BaseModel
load_dotenv()
app = FastAPI()
# --- Tools ---
@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"It's sunny and 32°C in {city}"

@tool
def get_time(city: str) -> str:
    """Get the current time for a city (mocked)."""
    return f"It's 4:30 PM in {city}"

llm = ChatGroq(model="openai/gpt-oss-120b", temperature=0)
agent = create_react_agent(model=llm, tools=[get_weather, get_time])
class ChatRequest(BaseModel):
    message: str
class ChatResponse(BaseModel):
    reply: str

@app.post("/chat", response_model=ChatResponse)
def chat(req: ChatRequest):
    response = agent.invoke({"messages": [{"role": "user", "content": req.message}]})
    reply = response["messages"][-1].content
    return ChatResponse(reply=reply)

load.py

import asyncio
import time
import httpx

async def send_request(client, msg):
    resp = await client.post("http://localhost:8000/chat", json={"message": msg})
    print(resp.json())

async def main():
    start = time.time()
    async with httpx.AsyncClient(timeout=30) as client:
        await asyncio.gather(
            send_request(client, "Weather in Chennai?"),
            send_request(client, "Weather in Mumbai?"),
            send_request(client, "Time in Delhi?"),
        )
    print(f"\nTotal time: {time.time() - start:.2f}s")

asyncio.run(main())

Load_sync.py

import time
import requests

def send_request(msg):
    start = time.time()
    resp = requests.post("http://localhost:8000/chat", json={"message": msg})
    elapsed = time.time() - start
    print(f"[{msg}] -> {resp.json()['reply']}  ({elapsed:.2f}s)")

messages = [
    "Weather in Chennai?",
    "Weather in Mumbai?",
    "Time in Delhi?",
]

start = time.time()
for message in messages:
    send_request(message)

print(f"\nTotal time: {time.time() - start:.2f}s")

Virtual Environment

• Consider we have machine with python 3.10 and now a code which is getting deployed and its running with 3.10.
• Later we need to upgrade to 3.20 , now the code with 3.10 may not work, to avoid this , we can create virtual environment to avoid this issue.
• conda , venv , UV , Poetry Tools are available
• venv comes along with python.
• why DOT ? Hidden folder , so that we can run the project.

python -m venv .env ( created an environment )
source .venv/bin/activate.fish ( activate the environment )
python -m pip install -r requirements.txt

gitignore

Commands

source .venv/bin/activate.fish
python -m pip freeze > requirements.txt ( its like an update , but u need to push this command )
deactivate
rm -rf .venv

python -m pip install django
python -m pip freeze > requirements.txt ( now you are updating the file , to make entry in the requirements file )

Notes

• Async --> Single thread.
• Multi-thread is like using multiple threads.
• Coroutine --> https://www.geeksforgeeks.org/python/coroutine-in-python/

• what is the purpose of Async ? While creating Agent , mostly it will be asybc only. Because lot amount wait time is there or involved . so while creating AGENT then obviously , all needs to go with ASYNC.

• What is React ? --> Reason & Action.

• App Sync & Load Sync ae important.

Comments

[Marcus Kim]

The async vs sync examples show how gathering tasks reduces total wait time from 5s to 3s by overlapping I/O operations. I noticed the API app uses ainvoke for async calls while the sync version uses invoke-that's a subtle but critical difference when handling multiple tool responses. For real-time apps, the async pattern avoids blocking the main thread during weather lookups, but the 3-second latency in the gather example might be too slow for users expecting instant replies.