Multi-Turn Tool Loop

12 줄의 agentic loop

"Agent" 는 tool loop. 모델 호출, tool 호출 체크, 실행, 결과 append, 다시 호출. 모델이 더 호출 대신 text 반환할 때 멈춤. 그게 다.

한 iteration 의 모양

모델에 contents 보냄.
Response 에 function_calls 없으면 response.text 반환 — 끝.
그렇지 않으면: 모델 turn 을 contents 에 append (호출 기억하도록).
모든 function call 실행 (independent 하면 병렬).
function-response part 들 모두 담은 single user turn append.
Loop.

병렬 호출이 정상

현대 Gemini 모델이 한 turn 에 multiple function_call part 자주 emit. 동시에 실행하고 한 user turn 에 모든 결과 반환 — API 가 id field 로 결과를 호출에 매칭.

Budget 필요해

Tool loop 가 모델이 계속 re-call 하면 영원히 돌 수 있어. 항상 iteration 카운트와 총 경과 시간 bound.

Code

Minimal loop·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')

Dispatch 테이블 — tool 이름을 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

3-tool agent 빌드: get_current_time, add_numbers, concat_strings. 첫 코드 블록의 agent_loop 에 wire. Flash 한테 물어: "47 + 53 은? 결과를 ' is the answer' 와 concat. 그 다음 시간 알려줘." 여러 turn 가로질러 세 tool 다 호출 관찰. tool 하나가 항상 에러 반환하게 sabotage 하면 max_iterations 발동 확인.