Async Agent Frameworks: LangChain and Beyond
ainvoke(), astream(), and async chains in LangChain and LangGraph.
Async Execution in LangChain
LangChain provides async versions of all its interfaces. Every component that has an invoke() also has ainvoke(), and every stream() has an astream(). Async is the recommended approach for production agents.
ainvoke() for Async LLM Calls
ainvoke() is the async equivalent of invoke(). Use it inside async functions to make non-blocking LLM calls. This allows multiple agents or requests to share the event loop.
import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
llm = ChatOpenAI(model='gpt-4o-mini', api_key='sk-...')
async def async_agent_call(question: str) -> str:
# ainvoke: non-blocking, releases event loop while waiting for OpenAI
response = await llm.ainvoke([HumanMessage(content=question)])
return response.content
async def handle_multiple_users(questions: list) -> list:
# All three LLM calls run concurrently
results = await asyncio.gather(*[async_agent_call(q) for q in questions])
return results
questions = [
'What is Python?',
'What is TypeScript?',
'What is Rust?'
]
results = asyncio.run(handle_multiple_users(questions))
for q, a in zip(questions, results):
print(f'Q: {q[:30]}... A: {a[:50]}...')All lessons in this course
- Async Python for Agent Developers
- Event Queues and Message Brokers
- Non-Blocking Parallel Tool Execution
- Async Agent Frameworks: LangChain and Beyond