비차단 병렬 도구 실행
여러 도구를 동시에 실행하는 asyncio.gather()를 학습합니다.
비차단 병렬 도구 실행은(는) CoddyKit의 무료 AI Agents 강의입니다. 이것은 4개 중 3번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 AI 튜터와 함께 브라우저에서 직접 실습할 수 있습니다. 이 강의는 AI Agents 학습 경로의 일부이며, 진행 상황이 웹과 CoddyKit 앱에 동기화됩니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.
도구를 병렬로 실행하는 이유
에이전트가 여러 독립적인 도구의 결과를 필요로 할 때 도구를 순차적으로 실행하면 시간이 낭비됩니다. 각 도구에 500밀리초가 걸린다면 도구 3개를 순차적으로 실행할 때는 1.5초가 걸리지만, 병렬로 실행하면 500밀리초만 걸립니다. 즉, 3배 빨라집니다.
병렬 호출을 위한 asyncio.gather()
asyncio.gather()는 여러 코루틴을 동시에 실행하고 모든 결과를 순서대로 반환합니다. 에이전트 도구를 병렬로 실행할 때 사용하는 기본 도구입니다.
import asyncio
import time
async def search_web(query: str) -> list:
await asyncio.sleep(0.5) # Simulate 500ms web search
return [f'Web result for: {query}']
async def search_database(query: str) -> list:
await asyncio.sleep(0.3) # Simulate 300ms DB query
return [f'DB result for: {query}']
async def get_weather(location: str) -> dict:
await asyncio.sleep(0.4) # Simulate 400ms API call
return {'location': location, 'temp': '22C'}
async def run_parallel():
start = time.perf_counter()
# Sequential: 0.5 + 0.3 + 0.4 = 1.2s
# Parallel: max(0.5, 0.3, 0.4) = 0.5s
web_results, db_results, weather = await asyncio.gather(
search_web('Python async'),
search_database('Python async'),
get_weather('New York')
)
elapsed = (time.perf_counter() - start) * 1000
print(f'Completed in {elapsed:.0f}ms (parallel)')
return web_results, db_results, weather
asyncio.run(run_parallel())개별 도구 실패 처리
return_exceptions=True를 사용하면 하나의 도구가 실패해도 모든 병렬 호출이 중단되지 않습니다. 각 결과는 값 또는 예외이므로 하나씩 확인하십시오.
import asyncio
async def tool_that_fails(name: str):
await asyncio.sleep(0.2)
if name == 'flaky_api':
raise ConnectionError(f'{name}: service unavailable')
return f'{name}: success'
async def parallel_with_fault_tolerance():
tool_names = ['web_search', 'flaky_api', 'database', 'weather_api']
coros = [tool_that_fails(name) for name in tool_names]
results = await asyncio.gather(*coros, return_exceptions=True)
tool_results = {}
errors = {}
for name, result in zip(tool_names, results):
if isinstance(result, Exception):
errors[name] = str(result)
print(f'Tool {name} FAILED: {result}')
else:
tool_results[name] = result
print(f'Tool {name} OK: {result}')
print(f'\nSucceeded: {len(tool_results)}/{len(tool_names)}')
print('Errors:', errors)
return tool_results, errors
asyncio.run(parallel_with_fault_tolerance())병렬 결과 결합
병렬 실행이 끝나면 결과를 하나의 컨텍스트로 결합하여 LLM에 전달하십시오. 각 정보가 어디에서 왔는지 명확하게 표시하십시오.
import asyncio
import json
async def parallel_research(query: str) -> dict:
web_task = search_web(query)
db_task = search_database(query)
weather_task = get_weather('New York')
results = await asyncio.gather(
web_task, db_task, weather_task,
return_exceptions=True
)
context_parts = []
sources_used = []
tool_names = ['web_search', 'database', 'weather']
for name, result in zip(tool_names, results):
if isinstance(result, Exception):
context_parts.append(f'[{name}]: unavailable ({result})')
else:
context_parts.append(f'[{name}]: {json.dumps(result)}')
sources_used.append(name)
combined_context = '\n'.join(context_parts)
return {
'query': query,
'context': combined_context,
'sources': sources_used
}
result = asyncio.run(parallel_research('Python performance tips'))
print('Sources used:', result['sources'])개별 도구의 제한 시간
asyncio.wait_for()를 사용하여 개별 도구 호출에 제한 시간을 추가하십시오. 느린 도구 하나 때문에 전체 병렬 일괄 작업이 무기한 차단되어서는 안 됩니다.
import asyncio
async def slow_tool(name: str) -> str:
await asyncio.sleep(10) # Very slow
return f'{name} result'
async def tool_with_timeout(coro, tool_name: str, timeout_seconds: float):
try:
result = await asyncio.wait_for(coro, timeout=timeout_seconds)
return result
except asyncio.TimeoutError:
return f'TIMEOUT: {tool_name} exceeded {timeout_seconds}s'
except Exception as e:
return f'ERROR: {tool_name}: {str(e)}'
async def parallel_with_timeouts():
results = await asyncio.gather(
tool_with_timeout(search_web('query'), 'web_search', 2.0),
tool_with_timeout(slow_tool('slow_api'), 'slow_api', 1.0),
tool_with_timeout(search_database('query'), 'database', 2.0)
)
for result in results:
print(result)
asyncio.run(parallel_with_timeouts())동적 병렬 도구 선택
에이전트는 질의에 따라 어떤 도구를 병렬로 실행할지 동적으로 결정할 수 있습니다. 도구 이름을 비동기 함수에 매핑하고 선택된 도구를 동시에 실행하는 분배기를 구성하십시오.
import asyncio
TOOL_REGISTRY = {
'web_search': search_web,
'database': search_database,
'weather': get_weather
}
async def execute_tools_parallel(tool_calls: list) -> dict:
'''
tool_calls: list of {'name': str, 'args': dict}
'''
tasks = {}
for call in tool_calls:
tool_name = call['name']
args = call.get('args', {})
fn = TOOL_REGISTRY.get(tool_name)
if fn:
# Get first positional arg (simplified)
first_arg = next(iter(args.values()), '') if args else ''
tasks[tool_name] = fn(first_arg)
else:
print(f'Unknown tool: {tool_name}')
if not tasks:
return {}
results = await asyncio.gather(*tasks.values(), return_exceptions=True)
return {
name: result
for name, result in zip(tasks.keys(), results)
}
tool_calls = [
{'name': 'web_search', 'args': {'query': 'async Python'}},
{'name': 'weather', 'args': {'location': 'London'}}
]
results = asyncio.run(execute_tools_parallel(tool_calls))
print('Results:', results)세마포어를 사용한 병렬성 제한
너무 많은 도구 호출을 병렬로 실행하면 응용 프로그래밍 인터페이스 사용량 제한에 걸리거나 서비스에 과부하가 발생할 수 있습니다. 동시에 실행되는 도구 호출 수를 제한하려면 asyncio.Semaphore를 사용하십시오.
import asyncio
MAX_CONCURRENT = 3
semaphore = asyncio.Semaphore(MAX_CONCURRENT)
async def rate_limited_tool(tool_fn, *args):
async with semaphore: # Blocks if MAX_CONCURRENT calls already running
return await tool_fn(*args)
async def process_many_queries(queries: list) -> list:
print(f'Processing {len(queries)} queries with max {MAX_CONCURRENT} concurrent')
tasks = [rate_limited_tool(search_web, q) for q in queries]
results = await asyncio.gather(*tasks, return_exceptions=True)
successes = [r for r in results if not isinstance(r, Exception)]
print(f'Completed: {len(successes)}/{len(queries)}')
return results
queries = [f'query-{i}' for i in range(10)]
results = asyncio.run(process_many_queries(queries))
print('Done')부분 결과 스트리밍
asyncio.as_completed()를 사용하면 모든 결과가 끝날 때까지 기다리지 않고 도구 결과가 도착하는 대로 처리할 수 있습니다. 사용자에게 부분 결과를 즉시 표시하십시오.
import asyncio
async def search_slow(query: str) -> dict:
await asyncio.sleep(1.0)
return {'source': 'slow_db', 'results': [f'Slow result for: {query}']}
async def search_fast(query: str) -> dict:
await asyncio.sleep(0.2)
return {'source': 'fast_cache', 'results': [f'Fast result for: {query}']}
async def process_as_available(query: str):
coros = [
search_fast(query),
search_slow(query),
search_web(query),
search_database(query)
]
tasks = [asyncio.create_task(c) for c in coros]
partial_results = []
print('Processing results as they arrive:')
for future in asyncio.as_completed(tasks):
result = await future
partial_results.append(result)
print(f' Got result {len(partial_results)}: {result}')
# In a real agent: stream this to the user interface
return partial_results
asyncio.run(process_as_available('machine learning'))LangChain의 병렬 도구 호출
LLM이 하나의 응답에서 여러 도구 호출을 반환하면 LangChain은 기본적으로 병렬 도구 호출을 지원합니다. 효율적으로 처리하려면 asyncio.gather()를 사용하십시오.
import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
llm = ChatOpenAI(model='gpt-4o-mini', api_key='sk-...')
TOOL_EXECUTORS = {
'search_web': lambda args: search_web(args.get('query', '')),
'search_database': lambda args: search_database(args.get('query', '')),
'get_weather': lambda args: get_weather(args.get('location', 'New York'))
}
async def handle_parallel_tool_calls(response_message) -> list:
if not response_message.tool_calls:
return []
tasks = []
tool_call_ids = []
for tool_call in response_message.tool_calls:
executor = TOOL_EXECUTORS.get(tool_call['name'])
if executor:
tasks.append(executor(tool_call['args']))
tool_call_ids.append(tool_call['id'])
results = await asyncio.gather(*tasks, return_exceptions=True)
return [
{'tool_call_id': tc_id, 'result': r}
for tc_id, r in zip(tool_call_ids, results)
]
print('Parallel LangChain tool execution defined')결과 중복 제거
검색을 병렬로 실행하면 서로 다른 도구가 겹치는 결과를 반환할 수 있습니다. 같은 정보가 컨텍스트에 여러 번 나타나지 않도록 LLM에 전달하기 전에 중복을 제거하십시오.
import hashlib
def deduplicate_results(all_results: list) -> list:
seen_hashes = set()
unique_results = []
for result in all_results:
content = str(result)
content_hash = hashlib.md5(content.encode()).hexdigest()
if content_hash not in seen_hashes:
seen_hashes.add(content_hash)
unique_results.append(result)
return unique_results
def merge_parallel_results(tool_results: dict) -> list:
all_items = []
for tool_name, results in tool_results.items():
if isinstance(results, Exception):
continue
if isinstance(results, list):
for item in results:
if isinstance(item, dict):
item['source'] = tool_name
all_items.append(item)
else:
all_items.append({'source': tool_name, 'data': results})
return deduplicate_results(all_items)
sample_results = {
'web': ['Result A', 'Result B'],
'db': ['Result B', 'Result C'] # Result B is duplicate
}
merged = merge_parallel_results(sample_results)
print(f'Before: {sum(len(v) for v in sample_results.values())} items')
print(f'After dedup: {len(merged)} items')병렬 실행 속도 향상 측정
병렬화로 실제 속도가 얼마나 향상되는지 측정하십시오. 순차 실행 시간과 병렬 실행 시간을 비교하여 이점을 수치화하고 추가된 복잡성을 정당화하십시오.
import asyncio
import time
async def measure_speedup(tools_and_args: list):
# Sequential timing
seq_start = time.perf_counter()
seq_results = []
for fn, args in tools_and_args:
result = await fn(*args)
seq_results.append(result)
seq_time = (time.perf_counter() - seq_start) * 1000
# Parallel timing
par_start = time.perf_counter()
par_results = await asyncio.gather(*[fn(*args) for fn, args in tools_and_args])
par_time = (time.perf_counter() - par_start) * 1000
speedup = seq_time / par_time if par_time > 0 else 0
print(f'Sequential: {seq_time:.0f}ms')
print(f'Parallel: {par_time:.0f}ms')
print(f'Speedup: {speedup:.1f}x')
return speedup
tools = [
(search_web, ('query',)),
(search_database, ('query',)),
(get_weather, ('London',))
]
asyncio.run(measure_speedup(tools))지식 확인: 병렬 실행
비차단 병렬 도구 실행에 대한 이해도를 확인해 보십시오.
병렬 도구 실행 요약
asyncio를 사용한 병렬 도구 실행은 에이전트 지연 시간을 크게 줄입니다. 동시 호출에는 asyncio.gather()를, 오류 허용에는 return_exceptions=True를, 도구별 제한 시간에는 asyncio.wait_for()를, 사용량 제한에는 Semaphore를, 부분 결과 스트리밍에는 asyncio.as_completed()를 사용하십시오. LLM에 전달할 컨텍스트를 깔끔하게 유지하려면 결합한 결과의 중복을 항상 제거하십시오.
AI 튜터와 함께 AI Agents을(를) 배우세요 — 무료
브라우저에서 실제 코드를 작성하고 실행하며, 24/7 AI 튜터로부터 즉각적인 도움을 받고, 웹이나 앱에서 중단한 부분부터 계속 학습하세요.
- 코스
- 60
- 레슨
- 239
자주 묻는 질문
“비차단 병렬 도구 실행” 강의는 무료인가요?
네 — “비차단 병렬 도구 실행” 전체 내용을 이 웹사이트에서 무료로 읽을 수 있습니다. 인터랙티브하게 실습하려면(내장 코드 에디터와 24/7 AI 튜터), CoddyKit PRO로 업그레이드하면 AI Agents 강의 전체를 잠금 해제할 수 있습니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.
“비차단 병렬 도구 실행”에서 뭘 배우나요?
여러 도구를 동시에 실행하는 asyncio.gather()를 학습합니다. 브라우저에서 직접 실행하는 실습 코드로 AI Agents을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.
AI Agents을(를) 시작하는 데 경험이 필요한가요?
사전 경험은 필요하지 않습니다. CoddyKit의 AI Agents은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 3번째 강의입니다.
“비차단 병렬 도구 실행” 강의는 얼마나 걸리나요?
대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.
이 AI Agents 강의에서 코드를 작성하고 실행할 수 있나요?
네. 모든 AI Agents 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.