ノンブロッキングなツールの並列実行
複数のツールを同時に実行するための asyncio.gather() を学びます。
「ノンブロッキングなツールの並列実行」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン3/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応の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を指定すると、1つのツールの失敗によってすべての並列呼び出しが中断されることを防げます。各結果は値または例外のいずれかになるため、個別に確認してください。
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())並列結果の統合
並列実行の後、結果を1つのコンテキストにまとめて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)セマフォによる並列性の制限
ツール呼び出しを並列で実行しすぎると、APIのレート制限に達したり、サービスに過大な負荷をかけたりする可能性があります。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が1回のレスポンスで複数のツール呼び出しを返した場合、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に渡すコンテキストを整理するため、統合した結果からは必ず重複を排除してください。
よくある質問
「ノンブロッキングなツールの並列実行」レッスンは無料ですか？
はい。「ノンブロッキングなツールの並列実行」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。
「ノンブロッキングなツールの並列実行」で何を学びますか？
複数のツールを同時に実行するための asyncio.gather() を学びます。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Agentsを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン3/4です。
「ノンブロッキングなツールの並列実行」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Agentsレッスンでコードを書いて実行できますか？
はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。