병렬 함수 호출

병렬 함수 호출이란 무엇인가요

병렬 도구 호출 인식하기

from openai import OpenAI
import json

client = OpenAI()

# A question that naturally requires two independent lookups
response = client.chat.completions.create(
    model='gpt-4o',
    messages=[{'role': 'user', 'content': 'Compare the weather in London and Tokyo right now.'}],
    tools=tools
)

message = response.choices[0].message
print('Number of tool calls:', len(message.tool_calls))
# Might print: Number of tool calls: 2

for tc in message.tool_calls:
    print(f'  {tc.function.name}({tc.function.arguments})')
# get_current_weather({"location": "London"})
# get_current_weather({"location": "Tokyo"})

asyncio로 도구 호출 실행하기

import asyncio
import json

async def execute_tool_call_async(tool_call) -> tuple:
    '''Execute a single tool call and return (tool_call_id, result).'''
    name = tool_call.function.name
    args = json.loads(tool_call.function.arguments)

    # Async version of your tool (uses httpx, aiohttp, etc.)
    if name == 'get_current_weather':
        result = await async_get_weather(**args)
    elif name == 'get_stock_price':
        result = await async_get_stock(**args)
    else:
        result = f'Unknown tool: {name}'

    return tool_call.id, str(result)

async def execute_all_parallel(tool_calls) -> list:
    '''Execute all tool calls concurrently.'''
    tasks = [execute_tool_call_async(tc) for tc in tool_calls]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

모든 결과를 함께 다시 보내기

async def run_parallel_tool_calls(user_message: str) -> str:
    messages = [{'role': 'user', 'content': user_message}]

    response = client.chat.completions.create(
        model='gpt-4o', messages=messages, tools=tools
    )
    assistant_message = response.choices[0].message
    messages.append(assistant_message)  # Add assistant's tool_calls

    if response.choices[0].finish_reason == 'tool_calls':
        # Execute all tool calls in parallel
        results = await execute_all_parallel(assistant_message.tool_calls)

        # Add all results to conversation
        for tool_call_id, result in results:
            messages.append({
                'role': 'tool',
                'tool_call_id': tool_call_id,
                'content': result
            })

        # One more API call to synthesize results
        final = client.chat.completions.create(model='gpt-4o', messages=messages)
        return final.choices[0].message.content

    return assistant_message.content

동기식 도구에 ThreadPoolExecutor 사용하기

from concurrent.futures import ThreadPoolExecutor, as_completed
import json

def execute_all_with_threads(tool_calls) -> list:
    results = []
    with ThreadPoolExecutor(max_workers=len(tool_calls)) as executor:
        future_to_id = {
            executor.submit(execute_tool_call, tc): tc.id
            for tc in tool_calls
        }
        for future in as_completed(future_to_id):
            tool_call_id = future_to_id[future]
            try:
                result = future.result(timeout=15)
            except Exception as e:
                result = f'Tool failed: {str(e)}'
            results.append((tool_call_id, str(result)))
    return results

개별 도구 실패 처리하기

async def safe_execute_all(tool_calls) -> list:
    tasks = [execute_tool_call_async(tc) for tc in tool_calls]
    raw_results = await asyncio.gather(*tasks, return_exceptions=True)

    results = []
    for tc, result in zip(tool_calls, raw_results):
        if isinstance(result, Exception):
            results.append((tc.id, f'Tool error: {str(result)}'))
        else:
            tool_call_id, output = result
            results.append((tool_call_id, output))
    return results

모델이 병렬 호출을 사용하는 경우

병렬 호출 속도 제한하기

import asyncio

async def rate_limited_execute_all(tool_calls, max_concurrent: int = 5) -> list:
    semaphore = asyncio.Semaphore(max_concurrent)

    async def limited_call(tc):
        async with semaphore:
            return await execute_tool_call_async(tc)

    tasks = [limited_call(tc) for tc in tool_calls]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    return [
        (tc.id, str(r) if not isinstance(r, Exception) else f'Error: {r}')
        for tc, r in zip(tool_calls, results)
    ]

순차 호출과 병렬 호출 연결하기

async def multi_round_agent(user_message: str) -> str:
    messages = [{'role': 'user', 'content': user_message}]
    MAX_ROUNDS = 5

    for _ in range(MAX_ROUNDS):
        response = client.chat.completions.create(
            model='gpt-4o', messages=messages, tools=tools
        )
        choice = response.choices[0]
        messages.append(choice.message)

        if choice.finish_reason == 'stop':
            return choice.message.content  # Done

        if choice.finish_reason == 'tool_calls':
            # Execute all tool calls in parallel (may be 1 or many)
            results = await safe_execute_all(choice.message.tool_calls)
            for tc_id, result in results:
                messages.append({'role': 'tool', 'tool_call_id': tc_id, 'content': result})
            # Loop continues for potentially sequential next call

    return 'Max rounds reached.'

지연 시간 개선 측정하기

import time
import asyncio

async def benchmark_parallel_vs_sequential():
    tool_calls = [...]  # 5 independent tool calls

    # Sequential
    start = time.time()
    for tc in tool_calls:
        await execute_tool_call_async(tc)
    sequential_time = time.time() - start

    # Parallel
    start = time.time()
    await asyncio.gather(*[execute_tool_call_async(tc) for tc in tool_calls])
    parallel_time = time.time() - start

    print(f'Sequential: {sequential_time:.2f}s')
    print(f'Parallel:   {parallel_time:.2f}s')
    print(f'Speedup:    {sequential_time/parallel_time:.1f}x')

병렬 도구 호출 디버깅하기

빠른 확인

레슨 요약