Tool Loop Orchestration — 두 layer, 두 책임 · GPT Wire Quest

Adapter는 provider별 message 형식, event parsing, tool result 구조처럼 wire를 알아. Orchestrator는 handler를 언제 부르고, JSONL을 언제 쓰며, loop를 언제 끝내고, hook을 언제 적용할지 알아.

loop를 Adapter 안에 숨기지 마

Adapter가 모든 것을 처리하면 간단해 보이지만 호출별 측정과 개입이 어려워져. 특정 tool call에 logging hook을 넣거나 handler 실행 전에 arguments를 검사하거나 wire와 독립적으로 loop를 replay하기 힘들어.

Orchestrator가 결정을 맡아

Adapter는 TextDelta, ToolCallRequested 같은 typed event를 yield해. Orchestrator는 ToolCallRequested를 받으면 handler를 실행하고, 결과를 다음 호출의 input으로 넣으며, on_tool_call, on_text_delta, on_done hook을 적용해.

분리된 loop는 CI에서 replay할 수 있어

Orchestrator가 loop를 맡으면 capture한 event sequence를 mock adapter로 재생해 같은 동작에 도달하는지 확인할 수 있어. 실제 provider wire를 다시 호출하지 않고 logic regression을 잡아.

Code

Adapter 가 아니라 Orchestrator 가 loop own·python

import asyncio, json, logging
from dataclasses import dataclass

logger = logging.getLogger(__name__)

@dataclass
class AgentConfig:
    model: str = "gpt-4.1"
    max_iterations: int = 10        # Safety limit on tool loops
    tool_timeout_seconds: float = 30.0

class ProductionAgent:
    def __init__(self, adapter, tools, tool_schemas, config=None):
        self.adapter = adapter
        self.tools = tools            # name → callable
        self.tool_schemas = tool_schemas
        self.config = config or AgentConfig()

    async def run(self, user_message, conversation_history=None):
        messages = conversation_history or [
            {"role": "system", "content": "You are a helpful assistant."}
        ]
        messages.append({"role": "user", "content": user_message})

        for iteration in range(self.config.max_iterations):
            text, tool_calls = "", []
            async for chunk in self.adapter.stream(
                messages=messages, tools=self.tool_schemas
            ):
                if isinstance(chunk, TextDelta): text += chunk.content
                elif isinstance(chunk, ToolCallComplete):
                    tool_calls.append(chunk)

            if not tool_calls:
                messages.append({"role": "assistant", "content": text})
                return text, messages

            # Execute tools in parallel with per-tool timeout
            results = await asyncio.gather(*[
                self._execute_tool(tc) for tc in tool_calls
            ])
            # Add results to messages...

        return "Max iterations reached.", messages

    async def _execute_tool(self, tc):
        try:
            fn = self.tools[tc.name]
            args = json.loads(tc.arguments)
            return await asyncio.wait_for(fn(**args), timeout=self.config.tool_timeout_seconds)
        except asyncio.TimeoutError:
            return {"error": f"Tool {tc.name} timed out"}
        except Exception as e:
            return {"error": str(e)}

Tool Loop Orchestration — 두 layer, 두 책임

loop를 Adapter 안에 숨기지 마

Orchestrator가 결정을 맡아

분리된 loop는 CI에서 replay할 수 있어

Code

External links

Exercise

Progress

댓글 0