Blokkeringsfri parallell kjøring av verktøy
asyncio.gather() for å kjøre flere verktøy samtidig.
Blokkeringsfri parallell kjøring av verktøy er en gratis leksjon i AI-agenter på CoddyKit. Dette er leksjon 3 av 4. Du kan lese hele leksjonen gratis nedenfor – og deretter øve praktisk i nettleseren med en innebygd kodeeditor og en AI-veileder som er tilgjengelig døgnet rundt. Den er en del av læringsløpet i AI-agenter, og fremdriften din synkroniseres mellom nettet og CoddyKit-appen. Kurset i AI-agenter inneholder totalt 4 leksjoner.
Hvorfor parallell kjøring av verktøy?
Når en agent trenger resultater fra flere uavhengige verktøy, er sekvensiell kjøring tidkrevende. Hvis hvert verktøy bruker 500 ms, tar tre verktøy 1,5 sekunder sekvensielt, men bare 500 ms parallelt — en tredobling av hastigheten.
asyncio.gather() for parallelle kall
asyncio.gather() kjører flere coroutiner samtidig og returnerer alle resultatene i riktig rekkefølge. Dette er det viktigste verktøyet for parallell kjøring av agentverktøy.
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())Håndtering av feil i enkeltverktøy
Med return_exceptions=True avbryter ikke en feil i ett verktøy alle parallelle kall. Hvert resultat er enten en verdi eller et unntak — kontroller hvert resultat individuelt.
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())Kombinere parallelle resultater
Etter parallell kjøring kombinerer De resultatene i én samlet kontekst for LLM-en. Merk tydelig hvor hvert informasjonsstykke kommer fra.
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'])Tidsavbrudd for enkeltverktøy
Bruk asyncio.wait_for() for å legge til et tidsavbrudd for individuelle verktøykall. Et tregt verktøy bør ikke blokkere hele den parallelle gruppen på ubestemt tid.
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())Dynamisk valg av parallelle verktøy
En agent kan dynamisk avgjøre hvilke verktøy som skal kjøres parallelt, basert på spørringen. Bygg en dispatcher som kobler verktøynavn til asynkrone funksjoner og kjører de valgte funksjonene samtidig.
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)Begrense parallellitet med semaforer
Hvis for mange verktøykall kjøres parallelt, kan API-ets hastighetsgrenser nås, eller en tjeneste kan bli overbelastet. Bruk asyncio.Semaphore for å begrense hvor mange verktøykall som kjører samtidig.
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')Strømme delvise resultater
Med asyncio.as_completed() kan De behandle verktøyresultater etter hvert som de kommer, i stedet for å vente på at alle skal bli ferdige. Vis delvise resultater til brukerne umiddelbart.
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'))Parallelle verktøykall i LangChain
LangChain støtter parallelle verktøykall innebygd når LLM-en returnerer flere verktøykall i ett svar. Håndter dem med asyncio.gather() for effektivitet.
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')Fjerning av dupliserte resultater
Når De kjører parallelle søk, kan ulike verktøy returnere overlappende resultater. Fjern duplikater før resultatene presenteres for LLM-en, slik at den samme informasjonen ikke vises flere ganger i konteksten.
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')Måle hastighetsøkningen ved parallell kjøring
Mål den faktiske hastighetsøkningen ved parallellisering. Sammenlign kjøretiden for sekvensiell og parallell kjøring for å tallfeste gevinsten og begrunne den økte kompleksiteten.
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))Kunnskapssjekk: Parallell kjøring
Test Deres forståelse av ikke-blokkerende parallell kjøring av verktøy.
Oppsummering av parallell kjøring av verktøy
Parallell kjøring av verktøy med asyncio reduserer agentens ventetid betydelig. Bruk asyncio.gather() for samtidige kall, return_exceptions=True for feiltoleranse, asyncio.wait_for() for tidsavbrudd per verktøy, Semaphore for hastighetsbegrensning og asyncio.as_completed() for strømming av delvise resultater. Fjern alltid duplikater fra sammenslåtte resultater for å holde konteksten ryddig for LLM-en.
Lær deg AI-agenter med en AI-veileder – gratis
Skriv og kjør ekte kode i nettleseren, få umiddelbar hjelp fra en AI-veileder som er tilgjengelig døgnet rundt, og fortsett der du slapp – på nettet eller i appen.
- Kurs
- 60
- Leksjoner
- 239
Ofte stilte spørsmål
Er leksjonen «Blokkeringsfri parallell kjøring av verktøy» gratis?
Ja – hele teksten i «Blokkeringsfri parallell kjøring av verktøy» er gratis å lese her på nettet. For å øve interaktivt med en innebygd kodeeditor og en AI-veileder som er tilgjengelig døgnet rundt, og for å låse opp resten av AI-agenter-kurset, kan du oppgradere til CoddyKit PRO. Kurset i AI-agenter inneholder totalt 4 leksjoner.
Hva lærer jeg i «Blokkeringsfri parallell kjøring av verktøy»?
asyncio.gather() for å kjøre flere verktøy samtidig. Du øver på AI-agenter med praktisk kode som du kjører direkte i nettleseren, mens en AI-veileder som er tilgjengelig døgnet rundt, svarer på spørsmålene dine mens du jobber deg gjennom leksjonen.
Trenger jeg erfaring for å begynne med AI-agenter?
Ingen tidligere erfaring er nødvendig. AI-agenter på CoddyKit er lagt opp for både nybegynnere og viderekomne, så De kan begynne her eller helt fra start og lære i Deres eget tempo. Dette er leksjon 3 av 4.
Hvor lang tid tar leksjonen «Blokkeringsfri parallell kjøring av verktøy»?
De fleste CoddyKit-leksjoner tar omtrent 5–10 minutter. Hver leksjon er kort og interaktiv, slik at De gjør jevne fremskritt og kan fortsette akkurat der De slapp – både på nettet og i appen.
Kan jeg skrive og kjøre kode i denne AI-agenter-leksjonen?
Ja. Alle AI-agenter-leksjoner har en innebygd kodeeditor, slik at De kan skrive og kjøre ekte kode direkte i nettleseren og få umiddelbar tilbakemelding fra AI – uten lokal konfigurering.
Alle leksjonene i dette kurset
- Async Python for agentutviklere
- Hendelseskøer og meldingsmeglere
- Blokkeringsfri parallell kjøring av verktøy
- Asynkrone agentrammeverk: LangChain og videre