Estruturas assíncronas para agentes: LangChain e além
ainvoke(), astream() e cadeias assíncronas em LangChain e LangGraph.
Estruturas assíncronas para agentes: LangChain e além é uma aula grátis de AI Agents no CoddyKit. Esta é a aula 4 de 4. Você pode ler a aula completa abaixo gratuitamente — depois pratica ao vivo no navegador com um editor de código integrado e um tutor de IA 24/7. Faz parte do caminho de aprendizado de AI Agents, e seu progresso é sincronizado entre a web e o app CoddyKit. O curso de AI Agents inclui 4 aulas no total.
Execução assíncrona no LangChain
O LangChain oferece versões assíncronas de todas as suas interfaces. Todo componente que possui um invoke() também possui um ainvoke(), e todo stream() possui um astream(). A abordagem assíncrona é recomendada para agentes em produção.
ainvoke() para chamadas assíncronas de LLM
ainvoke() é o equivalente assíncrono de invoke(). Use-o dentro de funções assíncronas para fazer chamadas não bloqueantes ao LLM. Isso permite que vários agentes ou solicitações compartilhem o laço de eventos.
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]}...')astream() para transmissão de tokens
astream() fornece tokens à medida que chegam do LLM. Isso permite transmitir respostas ao usuário em tempo real, sem esperar que a conclusão completa chegue.
import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
llm = ChatOpenAI(model='gpt-4o-mini', api_key='sk-...')
async def stream_response(question: str):
print(f'Streaming answer to: {question}\n')
full_response = ''
async for chunk in llm.astream([HumanMessage(content=question)]):
token = chunk.content
if token:
print(token, end='', flush=True) # Print each token as it arrives
full_response += token
print() # New line after streaming
return full_response
async def main():
await stream_response('List 3 benefits of async programming in Python')
asyncio.run(main())AsyncCallbackHandler
Os retornos de chamada do LangChain são acionados em eventos específicos: início do LLM, fim do LLM, início da ferramenta e erro da cadeia. O AsyncCallbackHandler gerencia esses eventos de forma assíncrona, sem bloquear o laço do agente.
from langchain_core.callbacks import AsyncCallbackHandler
from typing import Any, Dict, List
import time
class LatencyCallbackHandler(AsyncCallbackHandler):
def __init__(self):
self.step_times = {}
self.step_counts = {}
async def on_llm_start(self, serialized: Dict, prompts: List[str], **kwargs):
run_id = str(kwargs.get('run_id', ''))
self.step_times[run_id] = time.perf_counter()
async def on_llm_end(self, response, **kwargs):
run_id = str(kwargs.get('run_id', ''))
if run_id in self.step_times:
elapsed_ms = (time.perf_counter() - self.step_times[run_id]) * 1000
print(f'LLM call completed in {elapsed_ms:.0f}ms')
async def on_tool_start(self, serialized: Dict, input_str: str, **kwargs):
tool_name = serialized.get('name', 'unknown')
print(f'Tool starting: {tool_name}')
async def on_tool_error(self, error: Exception, **kwargs):
print(f'Tool error: {error}')
handler = LatencyCallbackHandler()
print('Async callback handler created')
# Use: llm.ainvoke([...], config={'callbacks': [handler]})Funções de nós assíncronas do LangGraph
Os nós do LangGraph podem ser funções assíncronas. Quando você define um nó como async def, o LangGraph o aguarda durante a execução do grafo. Esse é o padrão recomendado para grafos em produção.
import asyncio
from langgraph.graph import StateGraph, END
from typing import TypedDict, List
class AgentState(TypedDict):
question: str
entities: List[str]
context: str
answer: str
async def extract_entities_node(state: AgentState) -> AgentState:
await asyncio.sleep(0.1) # Simulate async NLP call
entities = state['question'].split()[:3] # Simplified
return {'entities': entities}
async def retrieve_context_node(state: AgentState) -> AgentState:
await asyncio.sleep(0.2) # Simulate async vector search
context = f'Context for entities: {state["entities"]}'
return {'context': context}
async def generate_answer_node(state: AgentState) -> AgentState:
await asyncio.sleep(0.3) # Simulate async LLM call
answer = f'Answer based on: {state["context"]}'
return {'answer': answer}
# Build async graph
graph = StateGraph(AgentState)
graph.add_node('extract', extract_entities_node)
graph.add_node('retrieve', retrieve_context_node)
graph.add_node('generate', generate_answer_node)
graph.set_entry_point('extract')
graph.add_edge('extract', 'retrieve')
graph.add_edge('retrieve', 'generate')
graph.add_edge('generate', END)
app = graph.compile()
print('Async LangGraph compiled')Transmissão assíncrona do LangGraph
O LangGraph oferece suporte à transmissão assíncrona de estados intermediários enquanto o grafo é executado. Use astream() para ver a saída de cada nó assim que ele for concluído, em vez de esperar a execução completa.
import asyncio
async def stream_graph_execution(graph_app, initial_state: dict):
print('Graph execution streaming:')
async for step_output in graph_app.astream(initial_state):
for node_name, state_delta in step_output.items():
print(f' Node [{node_name}] completed:')
for key, value in state_delta.items():
print(f' {key}: {value}')
# Run the async graph
initial = {
'question': 'What is machine learning?',
'entities': [],
'context': '',
'answer': ''
}
asyncio.run(stream_graph_execution(app, initial))Limitação de taxa com semáforos
A OpenAI e outras APIs de LLM têm limites de taxa para solicitações por minuto. Use um semáforo assíncrono para garantir que o limite de taxa nunca seja excedido, mesmo ao executar muitas tarefas de agentes simultaneamente.
import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
llm = ChatOpenAI(model='gpt-4o-mini', api_key='sk-...')
# Limit to 10 concurrent LLM calls
LLM_SEMAPHORE = asyncio.Semaphore(10)
async def rate_limited_llm_call(question: str) -> str:
async with LLM_SEMAPHORE:
response = await llm.ainvoke([HumanMessage(content=question)])
return response.content
async def process_large_batch(questions: list) -> list:
print(f'Processing {len(questions)} questions with max 10 concurrent LLM calls')
tasks = [rate_limited_llm_call(q) for q in questions]
results = await asyncio.gather(*tasks, return_exceptions=True)
successes = [r for r in results if not isinstance(r, Exception)]
failures = [r for r in results if isinstance(r, Exception)]
print(f'Success: {len(successes)}, Failed: {len(failures)}')
return results
# Process 50 questions with max 10 concurrent calls
questions = [f'Question {i}: What is concept number {i}?' for i in range(20)]
asyncio.run(process_large_batch(questions))Definições de ferramentas assíncronas
No LangChain, as funções de ferramentas podem ser assíncronas. As ferramentas assíncronas são aguardadas durante a execução do agente, permitindo chamadas de API não bloqueantes dentro da própria ferramenta.
import asyncio
import httpx
from langchain.tools import tool
@tool
async def async_web_search(query: str) -> str:
'''Search the web for information about the query.'''
async with httpx.AsyncClient() as client:
# Real implementation would use a search API
response = await client.get(
'https://api.search.example.com/search',
params={'q': query, 'api_key': 'your-key'},
timeout=10.0
)
response.raise_for_status()
results = response.json()
return '\n'.join([r['snippet'] for r in results.get('items', [])[:3]])
@tool
async def async_fetch_document(url: str) -> str:
'''Fetch and return the text content of a URL.'''
async with httpx.AsyncClient() as client:
response = await client.get(url, timeout=15.0)
return response.text[:3000] # Limit content size
print('Async tools defined')
print('Use with: agent.ainvoke({"input": "your question"})')Agente assíncrono diretamente com OpenAI
Você pode criar um laço de agente totalmente assíncrono diretamente com o SDK da OpenAI, sem usar o LangChain. Isso oferece máximo controle e sobrecarga mínima.
import asyncio
import openai
import json
client = openai.AsyncOpenAI(api_key='sk-...')
TOOLS = [
{'type': 'function', 'function': {
'name': 'web_search',
'description': 'Search the web',
'parameters': {'type': 'object', 'properties': {'query': {'type': 'string'}}, 'required': ['query']}
}}
]
async def async_tool_call(tool_name: str, args: dict) -> str:
if tool_name == 'web_search':
await asyncio.sleep(0.3) # Simulate search
return f'Search results for: {args["query"]}'
return 'Unknown tool'
async def async_agent_loop(question: str, max_turns: int = 5) -> str:
messages = [{'role': 'user', 'content': question}]
for turn in range(max_turns):
response = await client.chat.completions.create(
model='gpt-4o-mini', messages=messages, tools=TOOLS
)
msg = response.choices[0].message
messages.append(msg)
if not msg.tool_calls:
return msg.content
# Execute tool calls in parallel
tool_results = await asyncio.gather(*[
async_tool_call(tc.function.name, json.loads(tc.function.arguments))
for tc in msg.tool_calls
])
for tc, result in zip(msg.tool_calls, tool_results):
messages.append({'role': 'tool', 'tool_call_id': tc.id, 'content': result})
return 'Max turns reached'
result = asyncio.run(async_agent_loop('What is the latest news on AI?'))
print(result)Cancelamento e limpeza
As tarefas assíncronas podem ser canceladas. Trate asyncio.CancelledError corretamente para garantir que os recursos sejam liberados quando uma execução do agente for cancelada (por exemplo, por solicitação do usuário ou por tempo limite).
import asyncio
async def cancellable_agent(question: str):
try:
print('Agent starting')
await asyncio.sleep(0.5) # Step 1
print('Step 1 done')
await asyncio.sleep(0.5) # Step 2 - may be cancelled here
print('Step 2 done')
return 'Completed'
except asyncio.CancelledError:
print('Agent was cancelled - cleaning up')
# Clean up resources: close connections, log cancellation
raise # Always re-raise CancelledError
finally:
print('Cleanup always runs')
async def run_with_timeout(question: str, timeout: float):
task = asyncio.create_task(cancellable_agent(question))
try:
result = await asyncio.wait_for(task, timeout=timeout)
return result
except asyncio.TimeoutError:
print(f'Agent exceeded {timeout}s timeout')
task.cancel()
return None
# Run with 0.7s timeout (not enough for both steps)
result = asyncio.run(run_with_timeout('test', timeout=0.7))
print('Final result:', result)Testes de código assíncrono de agentes
Teste as funções assíncronas dos agentes usando pytest-asyncio. Marque as funções de teste com @pytest.mark.asyncio para executá-las em um laço de eventos.
import pytest
import asyncio
from unittest.mock import AsyncMock, patch
# Install: pip install pytest-asyncio
# pytest.ini: [pytest] asyncio_mode = auto
@pytest.mark.asyncio
async def test_async_agent_call():
with patch('openai.AsyncOpenAI') as mock_openai:
mock_client = AsyncMock()
mock_openai.return_value = mock_client
mock_response = AsyncMock()
mock_response.choices[0].message.content = 'Mocked answer'
mock_response.choices[0].message.tool_calls = None
mock_client.chat.completions.create.return_value = mock_response
# Test the async function
result = await async_agent_call('What is Python?')
assert isinstance(result, str)
print('Async test passed')
@pytest.mark.asyncio
async def test_parallel_execution():
start = asyncio.get_event_loop().time()
results = await asyncio.gather(
asyncio.sleep(0.1),
asyncio.sleep(0.1),
asyncio.sleep(0.1)
)
elapsed = asyncio.get_event_loop().time() - start
assert elapsed < 0.3, 'Should complete in parallel'
print(f'Parallel test passed: {elapsed:.2f}s')Verificação de conhecimento: estruturas assíncronas
Teste sua compreensão das estruturas assíncronas para agentes.
Resumo das estruturas assíncronas
O LangChain assíncrono oferece ainvoke(), astream() e AsyncCallbackHandler para agentes assíncronos prontos para produção. O LangGraph oferece suporte nativo a funções de nós assíncronas. Use semáforos para limitação de taxa, o cliente AsyncOpenAI para chamadas diretas e pytest-asyncio para testes. O tratamento adequado do cancelamento garante a limpeza correta dos recursos quando os agentes são interrompidos.
Aprenda AI Agents com um tutor de IA — grátis
Escreva e execute código real no seu navegador, obtenha ajuda instantânea de um tutor de IA 24/7 e continue de onde parou na web ou no app.
- Cursos
- 60
- Aulas
- 239
Perguntas Frequentes
A aula “Estruturas assíncronas para agentes: LangChain e além” é grátis?
Sim — o texto completo de “Estruturas assíncronas para agentes: LangChain e além” é grátis para ler aqui na web. Para praticá-la interativamente (um editor de código integrado e um tutor de IA 24/7) e desbloquear o restante do curso de AI Agents, atualize para CoddyKit PRO. O curso de AI Agents inclui 4 aulas no total.
O que vou aprender em “Estruturas assíncronas para agentes: LangChain e além”?
ainvoke(), astream() e cadeias assíncronas em LangChain e LangGraph. Você pratica AI Agents com código prático que executa diretamente no navegador, e um tutor de IA 24/7 responde suas dúvidas enquanto trabalha na aula.
Preciso ter experiência prévia para começar AI Agents?
Nenhuma experiência prévia é necessária. AI Agents no CoddyKit é estruturado para alunos iniciantes até avançados, então você pode começar aqui ou desde o início e aprender no seu ritmo. Esta é a aula 4 de 4.
Quanto tempo leva a aula “Estruturas assíncronas para agentes: LangChain e além”?
A maioria das aulas CoddyKit leva cerca de 5–10 minutos. Cada uma é compacta e interativa, então você faz progresso constante e retoma exatamente de onde parou entre web e app.
Posso escrever e executar código nesta aula de AI Agents?
Sim. Cada aula de AI Agents inclui um editor de código integrado, então você escreve e executa código real direto no navegador e recebe feedback de IA instantaneamente — nenhuma configuração local necessária.
Todas as aulas deste curso
- Python assíncrono para desenvolvedores de agentes
- Filas de eventos e intermediários de mensagens
- Execução paralela não bloqueante de ferramentas
- Estruturas assíncronas para agentes: LangChain e além