LLM to Agentic 2) - Adapter부터 Tool Registry까지

OpenAI Function Calling을 Agent 구조로 감싸기

OpenAI API에 tools를 전달하면 모델은 필요한 도구를 선택할 수 있다.

하지만 모델이 도구를 선택한다고 해서 함수가 자동으로 실행되는 것은 아니다. 애플리케이션이 모델의 응답을 해석하고, 함수를 실행하고, 실행 결과를 다시 모델에게 전달해야 한다.

출처: Function calling | OpenAI API

이번 글에서는 다음 흐름을 직접 구현한다.

사용자 요청
→ OpenAI가 도구 사용 여부 판단
→ 도구를 선택했다면 애플리케이션이 실행
→ 실행 결과를 messages에 추가
→ 누적된 context로 OpenAI 재호출
→ 최종 답변

범위가 너무 넓어지는 것을 막기 위해 다음 내용은 제외한다.

• 사용자와 여러 차례 대화하는 대화형 멀티턴
• SSE를 이용한 실시간 진행 상황 전달
• File, Shell, Browser 같은 실제 Agent 도구
• 도구 실행 오류에서 자동 복구하는 과정
• 장기 작업과 context 압축

대신 add_numbers라는 작은 도구 하나를 사용해 다음 구조에 집중한다.

• OpenAI Adapter
• Agent 내부 데이터 계약
• 모델의 자동 도구 선택
• 구조화된 도구 결과
• Tool Registry
• Toolset filtering
• ReAct 도구 호출 loop

---

1. OpenAI 예제 코드를 Agent loop에 바로 넣어도 될까?

가장 단순하게 구현하면 Agent loop에서 OpenAI API를 직접 호출할 수 있다.

response = client.responses.create(
    model=model_name,
    input=messages,
    tools=tools,
)

응답에 function_call이 있으면 직접 필드를 꺼낸다.

item = response.output[0]

tool_name = item.name
arguments = json.loads(item.arguments)
call_id = item.call_id

작은 예제에서는 충분하다.

하지만 이 코드가 Agent loop 안에 들어가면 loop가 OpenAI의 형식을 직접 알아야 한다.

Agent loop가 알아야 하는 것

- response.output
- response.output_text
- function_call
- function_call_output
- call_id
- arguments JSON
- OpenAI tool schema

결과적으로 Agent의 핵심 반복문과 OpenAI API 변환 코드가 한곳에 섞인다.

Agent가 해야 하는 일
- 모델에게 현재 context 전달
- tool call 여부 판단
- 도구 실행
- observation 추가
- 종료 판단

OpenAI 변환 코드가 해야 하는 일
- Message를 OpenAI input으로 변환
- ToolSpec을 OpenAI tool schema로 변환
- OpenAI function_call을 ToolCall로 변환

서로 다른 책임이다.

이를 분리하기 위해 중간에 Adapter를 둔다.

Agent 내부 계약
        ↓
OpenAI Adapter
        ↓
OpenAI Responses API

---

2. OpenAI만 사용할 건데도 Adapter가 필요할까?

현재 OpenAI만 사용한다고 해도 Adapter는 의미가 있다.

다른 모델로 교체할 가능성만을 위해 Adapter를 만드는 것은 아니다.

첫 번째 목적은 OpenAI 요청·응답 형식이 Agent 전체로 퍼지는 것을 막는 것이다.

우리 Message
→ Adapter
→ OpenAI input

OpenAI response
→ Adapter
→ 우리 AssistantOutput

두 번째 목적은 테스트다.

Agent loop가 OpenAI SDK를 직접 사용하면 loop를 테스트할 때마다 실제 API 응답 객체가 필요하다. 반면 Agent loop가 공통 Model 계약만 바라보면 테스트에서는 ScriptedModel을 넣을 수 있다.

model = ScriptedModel(
    scripted_outputs=(
        AssistantOutput(...),
        AssistantOutput(...),
    )
)

실제 API 호출과 비용 없이 다음 동작을 검증할 수 있다.

• 모델이 tool call을 반환했을 때
• tool result가 context에 추가되는지
• 여러 번 도구를 호출하는지
• 최종 답변에서 멈추는지
• 최대 반복 횟수에서 종료되는지

세 번째 목적은 변환 코드의 중복을 줄이는 것이다.

Adapter가 없으면 여러 파일에서 다음 코드를 각각 작성할 수 있다.

json.loads(item.arguments)
item.call_id
response.output_text

변환 지점이 많아지면 call_id를 누락하거나, 잘못된 필드를 연결하거나, arguments JSON 변환을 빼먹을 가능성이 커진다.

Adapter가 휴먼에러를 완전히 제거하는 것은 아니다. Adapter 안의 코드도 잘못 작성할 수 있다.

다만 외부 API 변환 지점을 하나로 제한해 실수가 발생할 수 있는 면적을 줄인다.

---

3. Agent 내부 데이터 계약 정의

Agent loop가 OpenAI 응답 객체를 직접 사용하지 않으려면 프로젝트 내부에서 사용할 데이터 형태가 필요하다.

이번 구현에서는 Pydantic BaseModel을 사용했다.

from typing import Any, Literal

from pydantic import BaseModel, ConfigDict


class FrozenModel(BaseModel):
    model_config = ConfigDict(frozen=True)

frozen=True는 객체가 만들어진 뒤 필드가 임의로 변경되는 것을 막는다.

ToolCall

모델이 어떤 도구를 어떤 인자로 호출할지 나타낸다.

class ToolCall(FrozenModel):
    id: str
    name: str
    arguments: dict[str, Any]

예를 들면 다음과 같다.

ToolCall(
    id="call-1",
    name="add_numbers",
    arguments={
        "left": 2,
        "right": 3,
    },
)

id는 이후 도구 실행 결과와 연결할 때 사용한다.

AssistantOutput

모델의 한 번의 응답을 표현한다.

class AssistantOutput(FrozenModel):
    content: str
    tool_call: ToolCall | None = None

도구가 필요하지 않다면 일반 답변만 들어 있다.

AssistantOutput(
    content="안녕하세요.",
    tool_call=None,
)

도구가 필요하다면 tool_call이 들어 있다.

AssistantOutput(
    content="",
    tool_call=ToolCall(
        id="call-1",
        name="add_numbers",
        arguments={
            "left": 2,
            "right": 3,
        },
    ),
)

Message

Agent가 모델에게 전달할 context 한 항목이다.

class Message(FrozenModel):
    role: Literal["user", "assistant", "tool"]
    content: str
    tool_call: ToolCall | None = None
    tool_call_id: str | None = None

역할은 세 종류다.

user
→ 사용자의 요청

assistant
→ 모델의 답변 또는 tool call

tool
→ 도구 실행 결과

ToolSpec

모델에게 도구의 사용법을 설명한다.

class ToolSpec(FrozenModel):
    name: str
    description: str
    parameters: dict[str, Any]

add_numbers 도구라면 다음과 같은 정보가 들어간다.

ToolSpec(
    name="add_numbers",
    description="두 정수를 더한다.",
    parameters={
        "type": "object",
        "properties": {
            "left": {"type": "integer"},
            "right": {"type": "integer"},
        },
        "required": ["left", "right"],
    },
)

이 ToolSpec은 함수를 실행하는 코드가 아니다.

모델에게 다음을 알려주는 설명서다.

add_numbers라는 도구가 있다.
left와 right라는 정수가 필요하다.

---

4. Model 계약과 OpenAI Adapter

Agent loop는 구체적인 OpenAI SDK가 아니라 추상적인 Model을 사용한다.

from abc import ABC, abstractmethod


class Model(ABC):
    @abstractmethod
    def complete(
        self,
        messages: tuple[Message, ...],
        tools: tuple[ToolSpec, ...],
    ) -> AssistantOutput:
        raise NotImplementedError

Agent loop가 기대하는 것은 단순하다.

Message와 ToolSpec을 전달한다.
→ AssistantOutput을 받는다.

OpenAI Adapter는 이 계약을 구현한다.

class OpenAIModel(Model):
    def __init__(
        self,
        client,
        model_name: str,
        instructions: str,
    ) -> None:
        self.client = client
        self.model_name = model_name
        self.instructions = instructions

OpenAI client는 OpenAIModel 내부에 들어 있다.

따라서 Agent loop에서 매번 client를 전달하지 않는다.

model = OpenAIModel(
    client=OpenAI(),
    model_name="사용할 모델",
    instructions="You are a helpful assistant.",
)

사용할 때는 공통 계약만 호출한다.

output = model.complete(
    messages,
    tool_specs,
)

---

5. 내부 Message를 OpenAI input으로 변환

우리 프로젝트의 Message와 OpenAI Responses API의 input 형식은 서로 다르다.

일반 user 메시지는 비교적 단순하다.

Message(
    role="user",
    content="2와 3을 더해줘",
)

OpenAI input으로는 다음처럼 전달한다.

{
    "role": "user",
    "content": "2와 3을 더해줘",
}

하지만 assistant의 tool call은 다른 형태로 변환해야 한다.

Message(
    role="assistant",
    content="",
    tool_call=ToolCall(
        id="call-1",
        name="add_numbers",
        arguments={
            "left": 2,
            "right": 3,
        },
    ),
)

OpenAI에는 다음 형태로 보낸다.

{
    "type": "function_call",
    "call_id": "call-1",
    "name": "add_numbers",
    "arguments": '{"left": 2, "right": 3}',
}

도구 실행 결과도 별도 형식이 필요하다.

Message(
    role="tool",
    content="5",
    tool_call_id="call-1",
)

OpenAI input:

{
    "type": "function_call_output",
    "call_id": "call-1",
    "output": "5",
}

이를 Adapter 함수 하나에서 처리한다.

def to_openai_input(
    messages: tuple[Message, ...],
) -> list[dict[str, Any]]:
    items = []

    for message in messages:
        if (
            message.role == "assistant"
            and message.tool_call is not None
        ):
            items.append(
                {
                    "type": "function_call",
                    "call_id": message.tool_call.id,
                    "name": message.tool_call.name,
                    "arguments": json.dumps(
                        message.tool_call.arguments,
                        ensure_ascii=False,
                    ),
                }
            )

        elif message.role == "tool":
            items.append(
                {
                    "type": "function_call_output",
                    "call_id": message.tool_call_id,
                    "output": message.content,
                }
            )

        else:
            items.append(
                {
                    "role": message.role,
                    "content": message.content,
                }
            )

    return items

call_id가 중요한 이유는 모델이 요청한 도구 호출과 실행 결과를 연결해야 하기 때문이다.

function_call
call_id="call-1"

function_call_output
call_id="call-1"

두 ID가 같아야 모델이 다음을 이해할 수 있다.

이 5라는 결과는 내가 조금 전에 요청한 add_numbers의 결과다.

---

6. ToolSpec을 OpenAI tool 형식으로 변환

내부 ToolSpec도 OpenAI tool 형식으로 바꾼다.

def to_openai_tools(
    tools: tuple[ToolSpec, ...],
) -> list[dict[str, Any]]:
    return [
        {
            "type": "function",
            "name": tool.name,
            "description": tool.description,
            "parameters": tool.parameters,
            "strict": True,
        }
        for tool in tools
    ]

이제 OpenAIModel.complete()에서 두 변환 함수를 사용할 수 있다.

class OpenAIModel(Model):
    def complete(
        self,
        messages: tuple[Message, ...],
        tools: tuple[ToolSpec, ...],
    ) -> AssistantOutput:
        response = self.client.responses.create(
            model=self.model_name,
            instructions=self.instructions,
            input=to_openai_input(messages),
            tools=to_openai_tools(tools),
            tool_choice="auto",
            parallel_tool_calls=False,
        )

        for item in response.output:
            if item.type == "function_call":
                return AssistantOutput(
                    content=response.output_text,
                    tool_call=ToolCall(
                        id=item.call_id,
                        name=item.name,
                        arguments=json.loads(item.arguments),
                    ),
                )

        return AssistantOutput(
            content=response.output_text,
        )

OpenAI가 function_call을 반환하면 Adapter가 ToolCall로 바꾼다.

일반 답변이라면 content만 들어 있는 AssistantOutput으로 바꾼다.

Agent loop는 OpenAI 응답 객체를 직접 볼 필요가 없다.

---

7. 모델이 도구 사용 여부를 직접 선택한다

tool_choice="auto"로 설정하면 모델은 도구 사용 여부를 직접 판단한다.

response = self.client.responses.create(
    ...,
    tool_choice="auto",
)

도구가 필요 없다고 판단하면:

AssistantOutput(
    content="최종 답변",
    tool_call=None,
)

도구가 필요하다고 판단하면:

AssistantOutput(
    content="",
    tool_call=ToolCall(...),
)

Agent loop는 tool_call의 존재 여부만 확인하면 된다.

if output.tool_call is None:
    return output.content

tool_call 없음
→ 최종 답변

tool_call 있음
→ 도구 실행

Adapter가 OpenAI의 여러 응답 형식을 내부의 단순한 분기로 바꿔준 것이다.

---

8. 문자열 대신 구조화된 ToolResult 사용

처음에는 도구가 문자열만 반환할 수 있다.

def add_numbers(arguments):
    return "5"

하지만 문자열만으로는 성공과 실패를 구분하기 어렵다.

"5"
→ 성공 결과

"file not found"
→ 실패 결과

둘 다 그냥 문자열

이를 해결하기 위해 ToolResult를 만든다.

class ToolResult(FrozenModel):
    ok: bool
    output: str
    error_code: str | None = None

성공:

ToolResult(
    ok=True,
    output="5",
    error_code=None,
)

실패:

ToolResult(
    ok=False,
    output="tool failed",
    error_code="tool_error",
)

생성을 쉽게 하기 위해 factory method를 둔다.

@classmethod
def success(cls, output: str) -> "ToolResult":
    return cls(
        ok=True,
        output=output,
    )

@classmethod
def failure(
    cls,
    error_code: str,
    output: str,
) -> "ToolResult":
    return cls(
        ok=False,
        output=output,
        error_code=error_code,
    )

성공과 실패 상태가 모순되지 않도록 검증한다.

@model_validator(mode="after")
def validate_error_code(self) -> "ToolResult":
    if self.ok and self.error_code is not None:
        raise ValueError(
            "successful tool result cannot have error_code"
        )

    if not self.ok and self.error_code is None:
        raise ValueError(
            "failed tool result requires error_code"
        )

    return self

허용되는 상태:

ok=True,  error_code=None
ok=False, error_code가 존재

거부되는 상태:

ok=True,  error_code가 존재
ok=False, error_code=None

이 단계에서는 실패 결과의 데이터 형태를 만든 것이다.

모든 예외를 실패 observation으로 바꾸고 모델이 다음 round에서 복구하게 만드는 기능은 이후 단계에서 추가한다.

---

9. ToolSpec과 handler를 하나의 ToolEntry로 묶기

도구에는 서로 연결되어야 하는 두 정보가 있다.

모델에게 보여주는 설명
→ ToolSpec

실제로 실행하는 함수
→ handler

두 목록을 따로 관리하면 문제가 생길 수 있다.

tool_specs = [
    ADD_NUMBERS_SPEC,
]

handlers = {
    "plus_numbers": add_numbers,
}

모델은 add_numbers를 호출했는데 dispatch에는 plus_numbers만 있을 수 있다.

이를 막기 위해 ToolEntry 하나에 spec과 handler를 함께 넣는다.

class ToolEntry:
    def __init__(
        self,
        spec: ToolSpec,
        handler: ToolHandler,
        toolsets: frozenset[str],
    ) -> None:
        self.spec = spec
        self.handler = handler
        self.toolsets = toolsets

add_numbers 도구를 정의해보자.

ADD_NUMBERS_TOOL = ToolEntry(
    spec=ToolSpec(
        name="add_numbers",
        description="두 정수를 더한다.",
        parameters=AddNumbersArguments.model_json_schema(),
    ),
    handler=add_numbers,
    toolsets=frozenset({"math"}),
)

이제 모델용 schema와 handler는 같은 원본에서 나온다.

ADD_NUMBERS_TOOL
├─ spec.name = "add_numbers"
├─ spec.description
├─ spec.parameters
├─ handler = add_numbers
└─ toolsets = {"math"}

---

10. ToolRegistry를 단일 원본으로 사용

도구가 많아지면 ToolEntry를 Registry에 등록한다.

registry = ToolRegistry()
registry.register(ADD_NUMBERS_TOOL)

_entries는 내부 상태이므로 직접 수정하지 않는다.

registry._entries = {...}  # 사용하지 않음

외부에서는 register()를 사용한다.

class ToolRegistry:
    def __init__(self) -> None:
        self._entries: dict[str, ToolEntry] = {}

    def register(
        self,
        entry: ToolEntry,
    ) -> None:
        name = entry.spec.name

        if name in self._entries:
            raise ValueError(
                f"duplicate tool name: {name}"
            )

        self._entries[name] = entry

Registry는 ToolView를 만든다.

@dataclass(frozen=True)
class ToolView:
    specs: tuple[ToolSpec, ...]
    dispatch: Mapping[str, ToolHandler]

ToolView.specs
→ 모델에게 전달

ToolView.dispatch
→ 실제 도구 실행

두 값 모두 같은 ToolEntry 목록에서 만들어진다.

ToolView(
    specs=(
        ADD_NUMBERS_TOOL.spec,
    ),
    dispatch={
        "add_numbers": ADD_NUMBERS_TOOL.handler,
    },
)

---

11. Toolset으로 필요한 도구만 노출

도구가 많아지면 모든 실행에서 모든 도구를 모델에게 보여줄 필요가 없다.

예를 들어 도구를 다음처럼 분류할 수 있다.

math
- add_numbers
- multiply_numbers

workspace
- search_files
- read_file

execution
- run_shell

현재 작업이 수학 계산이라면 math 도구만 노출한다.

tool_view = registry.view(
    frozenset({"math"})
)

모델에게 전달되는 도구:

tool_view.specs

실제로 실행 가능한 도구:

tool_view.dispatch

두 목록에는 math toolset에 속한 도구만 들어간다.

registry 전체
├─ add_numbers       {"math"}
├─ read_file         {"workspace"}
└─ run_shell         {"execution"}

registry.view({"math"})
└─ add_numbers

Toolset은 아직 완전한 권한 시스템은 아니다.

하지만 실행 목적에 필요한 도구만 모델에게 보여주는 시작점이 된다.

• 모델의 도구 선택 부담 감소
• 불필요한 도구 노출 방지
• profile별 도구 구성
• 이후 권한 정책의 기반

---

12. 최종 Agent loop

이제 앞에서 만든 구조를 조립한다.

def run_agent(
    prompt: str,
    model: Model,
    tool_view: ToolView,
    max_rounds: int = 3,
) -> RunResult:
    messages = [
        Message(
            role="user",
            content=prompt,
        )
    ]

    traces = []

    for _ in range(max_rounds):
        output = model.complete(
            tuple(messages),
            tool_view.specs,
        )

        if output.tool_call is None:
            return RunResult(
                answer=output.content,
                tool_trace=tuple(traces),
            )

        call = output.tool_call

        messages.append(
            Message(
                role="assistant",
                content=output.content,
                tool_call=call,
            )
        )

        handler = tool_view.dispatch[call.name]
        tool_result = handler(call.arguments)

        traces.append(
            ToolTrace(
                name=call.name,
                arguments=call.arguments,
                result=tool_result,
            )
        )

        messages.append(
            Message(
                role="tool",
                content=tool_result.output,
                tool_call_id=call.id,
            )
        )

    raise RuntimeError(
        f"agent exceeded max_rounds={max_rounds}"
    )

각 round에서 하는 일은 네 가지다.

1. 누적 messages를 모델에 전달
2. tool call이 없으면 최종 종료
3. tool call이 있으면 handler 실행
4. 실행 결과를 messages에 추가

---

13. 도구가 필요 없는 경우

사용자 입력:

messages = [
    Message(
        role="user",
        content="한 문장으로 인사해줘",
    )
]

모델 응답:

AssistantOutput(
    content="안녕하세요. 만나서 반갑습니다.",
    tool_call=None,
)

tool_call이 없으므로 바로 종료한다.

if output.tool_call is None:
    return RunResult(
        answer=output.content,
        tool_trace=(),
    )

흐름:

user
→ model
→ final answer

---

14. 도구가 필요한 경우

사용자 입력:

add_numbers 도구를 사용해 2와 3을 더하고,
그 결과에 4를 다시 더해줘.

첫 번째 messages:

[
    Message(
        role="user",
        content="2와 3을 더한 뒤 4를 더해줘",
    )
]

첫 번째 모델 응답:

ToolCall(
    id="call-1",
    name="add_numbers",
    arguments={
        "left": 2,
        "right": 3,
    },
)

도구를 실행한다.

add_numbers({
    "left": 2,
    "right": 3,
})

결과:

ToolResult.success("5")

messages에 tool call과 결과가 추가된다.

user
2와 3을 더한 뒤 4를 더해줘

assistant
add_numbers(left=2, right=3) 호출

tool
5

모델은 누적된 context를 보고 두 번째 도구를 호출한다.

ToolCall(
    id="call-2",
    name="add_numbers",
    arguments={
        "left": 5,
        "right": 4,
    },
)

두 번째 도구 실행 결과:

ToolResult.success("9")

messages:

user
2와 3을 더한 뒤 4를 더해줘

assistant
add_numbers(2, 3)

tool
5

assistant
add_numbers(5, 4)

tool
9

마지막으로 모델은 도구 호출 없이 답변한다.

AssistantOutput(
    content="2와 3을 더한 뒤 4를 더하면 9입니다.",
    tool_call=None,
)

최종 흐름:

사용자 요청
→ model: add_numbers(2, 3)
→ tool: 5
→ model: add_numbers(5, 4)
→ tool: 9
→ model: 최종 답변

이것이 가장 작은 ReAct 도구 사용 구조다.

---

15. 06~10에서 추가된 것

단계	추가된 핵심
06	실제 OpenAI Responses API Adapter
07	tool_choice="auto"와 사용자 입력
08	구조화된 ToolResult와 Model ABC
09	ToolSpec과 handler의 단일 원본 Registry
10	필요한 toolset만 포함하는 ToolView

전체 구조를 한 번에 보면 다음과 같다.

사용자 요청
        ↓
AgentLoop
        ↓
Model.complete(
    messages,
    tool_view.specs,
)
        ↓
OpenAI Adapter
        ↓
OpenAI Responses API
        ↓
AssistantOutput
        ↓
tool_call이 있는가?
    ├─ 없다 → final
    └─ 있다
         ↓
    tool_view.dispatch
         ↓
      ToolResult
         ↓
  role="tool" Message
         ↓
    다음 model round

---

마무리

OpenAI Function Calling은 모델이 사용할 함수를 선택하고 arguments를 생성할 수 있게 해준다.

하지만 실제 Agent를 만들기 위해서는 그 주변 구조가 필요하다.

• OpenAI 형식을 내부 계약으로 바꾸는 Adapter
• 요청·응답의 형태를 고정하는 Pydantic 모델
• 성공과 실패를 표현하는 ToolResult
• 모델용 schema와 실제 handler를 연결하는 ToolRegistry
• 필요한 도구만 선택하는 Toolset
• tool result를 context에 추가하고 다시 판단하는 Agent loop

여기까지는 도구의 정상 실행 흐름에 집중했다.

아직 다음 문제들이 남아 있다.

모델이 존재하지 않는 도구를 호출하면?
arguments가 잘못되면?
도구 내부에서 예외가 발생하면?
깨진 JSON이 전달되면?
모델이 같은 도구를 무한히 반복하면?

다음 단계에서는 이러한 실패를 Python 예외로 Agent를 종료시키지 않고, 모델이 읽을 수 있는 observation으로 변환해 스스로 다음 행동을 수정하도록 만든다.