여러 차례 이어지는 도구 호출 반복

12줄로 만드는 에이전트형 반복

"에이전트"의 핵심은 도구 호출 반복이야. 모델을 호출하고, 도구 호출이 있는지 확인하고, 실행하고, 결과를 덧붙인 뒤 다시 모델을 호출해. 모델이 또 다른 호출 대신 텍스트를 반환하면 멈춰. 그게 전부야.

한 번의 반복이 흘러가는 순서

모델에 contents를 보내.
응답에 function_calls가 없으면 response.text를 반환하고 끝내.
호출이 있다면 모델의 차례를 contents에 덧붙여 호출을 기억하게 해.
모든 함수 호출을 실행해. 서로 독립이라면 병렬로 돌려.
모든 function-response part를 사용자 차례 하나에 담아 덧붙여.
처음으로 돌아가.

병렬 호출은 정상적인 동작이야

요즘 Gemini 모델은 한 차례에 여러 function_call part를 자주 내놓아. 독립적인 호출은 동시에 실행하고 모든 결과를 한 사용자 차례로 돌려줘. API가 id 필드로 결과와 호출을 맞춰.

반드시 예산을 둬

모델이 계속 도구를 다시 호출하면 반복은 영원히 돌 수 있어. 반복 횟수와 전체 경과 시간에 항상 상한을 둬.

Code

최소 호출 반복·python

from google import genai
from google.genai import types

client = genai.Client()

def agent_loop(initial_prompt, tools, dispatch, max_iterations=10):
    contents = [types.Content(
        role='user',
        parts=[types.Part(text=initial_prompt)],
    )]
    config = types.GenerateContentConfig(tools=[tools])

    for iteration in range(max_iterations):
        response = client.models.generate_content(
            model='gemini-2.5-flash',
            contents=contents,
            config=config,
        )
        calls = response.function_calls or []
        if not calls:
            return response.text

        # Append the model's call turn (so it remembers)
        contents.append(response.candidates[0].content)

        # Execute calls and build a user-turn of results
        result_parts = []
        for fc in calls:
            result = dispatch(fc.name, dict(fc.args))
            result_parts.append(types.Part.from_function_response(
                name=fc.name,
                response={'result': result},
                id=fc.id,
            ))
        contents.append(types.Content(role='user', parts=result_parts))

    raise RuntimeError(f'Loop did not terminate within {max_iterations} iterations')

asyncio로 병렬 실행·python

import asyncio

async def agent_loop_async(prompt, tools, dispatch_async, max_iter=10):
    contents = [types.Content(role='user', parts=[types.Part(text=prompt)])]
    config = types.GenerateContentConfig(tools=[tools])

    for _ in range(max_iter):
        response = await client.aio.models.generate_content(
            model='gemini-2.5-flash', contents=contents, config=config,
        )
        calls = response.function_calls or []
        if not calls:
            return response.text

        contents.append(response.candidates[0].content)

        # Run all calls concurrently
        results = await asyncio.gather(*[
            dispatch_async(fc.name, dict(fc.args)) for fc in calls
        ])

        contents.append(types.Content(role='user', parts=[
            types.Part.from_function_response(
                name=fc.name, response={'result': r}, id=fc.id,
            )
            for fc, r in zip(calls, results)
        ]))

    raise RuntimeError('Loop budget exhausted')

디스패치 표 — 도구 이름과 Python 함수 연결·python

TOOLS_REGISTRY = {
    'set_light_values': lambda args: actually_set_lights(**args),
    'get_weather':      lambda args: weather_api(args['location']),
    'lookup_order':     lambda args: db.get_order(args['order_id']),
}

def dispatch(name, args):
    fn = TOOLS_REGISTRY.get(name)
    if not fn:
        return {'error': f'Unknown tool: {name}'}
    try:
        return fn(args)
    except Exception as e:
        # Return errors to the model — let it decide whether to retry
        return {'error': str(e)}

# Now call it
final_text = agent_loop(
    'Set the lights warm and 30%, then check the weather in Seoul.',
    tools=Tool(function_declarations=[set_lights, get_weather]),
    dispatch=dispatch,
)

Exercise

get_current_time, add_numbers, concat_strings 도구 3개를 가진 에이전트를 만들어 첫 코드 블록의 agent_loop에 연결해. Flash에게 "47 + 53은? 결과를 ' is the answer'와 이어 붙이고, 그다음 시간을 알려줘."라고 물어. 여러 차례에 걸쳐 세 도구를 모두 부르는지 관찰하고, 도구 하나가 항상 오류를 반환하게 망가뜨려 max_iterations가 작동하는지도 확인해.