비동기 에이전트 프레임워크: LangChain과 그 너머
LangChain과 LangGraph에서 ainvoke(), astream(), 비동기 연결을 다룹니다.
비동기 에이전트 프레임워크: LangChain과 그 너머은(는) CoddyKit의 무료 AI Agents 강의입니다. 이것은 4개 중 4번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 AI 튜터와 함께 브라우저에서 직접 실습할 수 있습니다. 이 강의는 AI Agents 학습 경로의 일부이며, 진행 상황이 웹과 CoddyKit 앱에 동기화됩니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.
LangChain의 비동기 실행
LangChain은 모든 인터페이스의 비동기 버전을 제공합니다. invoke()가 있는 모든 구성 요소에는 ainvoke()도 있고, stream()이 있는 모든 구성 요소에는 astream()도 있습니다. 비동기 방식은 운영 에이전트에 권장되는 접근법입니다.
비동기 LLM 호출을 위한 ainvoke()
ainvoke()는 invoke()의 비동기 버전입니다. 비동기 함수 안에서 사용하여 블로킹되지 않는 LLM 호출을 수행하십시오. 이를 통해 여러 에이전트나 요청이 이벤트 루프를 공유할 수 있습니다.
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()
astream()은 LLM에서 토큰이 도착하는 대로 생성합니다. 따라서 전체 응답이 완성될 때까지 기다리지 않고 사용자에게 실시간으로 응답을 스트리밍할 수 있습니다.
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
LangChain 콜백은 LLM 시작, LLM 종료, 도구 시작, 체인 오류와 같은 특정 이벤트에서 실행됩니다. AsyncCallbackHandler는 에이전트 루프를 차단하지 않고 이러한 이벤트를 비동기적으로 처리합니다.
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]})LangGraph 비동기 노드 함수
LangGraph 노드는 비동기 함수일 수 있습니다. 노드를 async def로 정의하면 LangGraph가 그래프를 실행하는 동안 해당 노드를 기다립니다. 운영 그래프에는 이 패턴을 사용하는 것이 좋습니다.
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')LangGraph의 비동기 스트리밍
LangGraph는 그래프가 실행되는 동안 중간 상태를 비동기적으로 스트리밍할 수 있습니다. 전체 실행이 끝날 때까지 기다리지 말고 각 노드의 출력이 완료되는 대로 확인하려면 astream()을 사용하십시오.
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))세마포어를 사용한 사용량 제한
OpenAI와 다른 LLM 응용 프로그래밍 인터페이스는 분당 요청 수에 대한 사용량 제한을 둡니다. 많은 에이전트 작업을 동시에 실행하더라도 사용량 제한을 초과하지 않도록 비동기 세마포어를 사용하십시오.
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))비동기 도구 정의
LangChain에서는 도구 함수가 비동기 함수일 수 있습니다. 비동기 도구는 에이전트 실행 중에 기다려지므로 도구 내부에서 블로킹되지 않는 응용 프로그래밍 인터페이스 호출을 수행할 수 있습니다.
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"})')OpenAI를 직접 사용하는 비동기 에이전트
LangChain 없이 OpenAI SDK만 사용하여 완전한 비동기 에이전트 루프를 직접 구축할 수 있습니다. 이를 통해 최대한의 제어권을 확보하고 오버헤드를 최소화할 수 있습니다.
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)취소와 정리
비동기 작업은 취소할 수 있습니다. 에이전트 실행이 취소될 때(예: 사용자 요청이나 제한 시간 초과로 취소될 때) 리소스가 정리되도록 asyncio.CancelledError를 올바르게 처리하십시오.
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)비동기 에이전트 코드 테스트
pytest-asyncio를 사용하여 비동기 에이전트 함수를 테스트하십시오. 이벤트 루프에서 실행하려면 @pytest.mark.asyncio로 테스트 함수에 표시를 추가하십시오.
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')지식 확인: 비동기 프레임워크
비동기 에이전트 프레임워크에 대한 이해도를 확인해 보십시오.
비동기 프레임워크 요약
비동기 LangChain은 운영에 적합한 비동기 에이전트를 위해 ainvoke(), astream(), AsyncCallbackHandler를 제공합니다. LangGraph는 비동기 노드 함수를 기본적으로 지원합니다. 사용량 제한에는 세마포어를, 직접 호출에는 AsyncOpenAI 클라이언트를, 테스트에는 비동기 파이썬 테스트 도구를 사용하십시오. 적절한 취소 처리를 통해 에이전트가 중단될 때 리소스가 깔끔하게 정리되도록 할 수 있습니다.
자주 묻는 질문
“비동기 에이전트 프레임워크: LangChain과 그 너머” 강의는 무료인가요?
네 — “비동기 에이전트 프레임워크: LangChain과 그 너머” 전체 내용을 이 웹사이트에서 무료로 읽을 수 있습니다. 인터랙티브하게 실습하려면(내장 코드 에디터와 24/7 AI 튜터), CoddyKit PRO로 업그레이드하면 AI Agents 강의 전체를 잠금 해제할 수 있습니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.
“비동기 에이전트 프레임워크: LangChain과 그 너머”에서 뭘 배우나요?
LangChain과 LangGraph에서 ainvoke(), astream(), 비동기 연결을 다룹니다. 브라우저에서 직접 실행하는 실습 코드로 AI Agents을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.
AI Agents을(를) 시작하는 데 경험이 필요한가요?
사전 경험은 필요하지 않습니다. CoddyKit의 AI Agents은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 4번째 강의입니다.
“비동기 에이전트 프레임워크: LangChain과 그 너머” 강의는 얼마나 걸리나요?
대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.
이 AI Agents 강의에서 코드를 작성하고 실행할 수 있나요?
네. 모든 AI Agents 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.
이 강의의 모든 강의
- 에이전트 개발자를 위한 비동기 Python
- 이벤트 큐와 메시지 브로커
- 비차단 병렬 도구 실행
- 비동기 에이전트 프레임워크: LangChain과 그 너머