2026 - 오픈소스 에이전트 프레임워크 비교하기 (2) - Control Flow: OpenAI Agents SDK

About this offer

0. 들어가며 지난 글에서는 여러 오픈소스 Agent Framework의 실제 실행 코드를 따라가며 Agent Loop가 어떤 단위로 반복되고, 각 프레임워크가 그 실행 결과를 어떻게 표현하는지 살펴봤습니다. 이번 글에서는 여기서 한 단계 더 들어가 누가 Agent의 Control Flow를 소유하는지 살펴보겠습니다. Model이 Tool을 호출하거나 다른 Agent를 선택하면 그 결정이 그대로 다음 실행으로 이어지는지, Framework의 Runner가 별도의 상태 전이를 결정하는지, 또는 LangGraph처럼 처음부터 Graph Runtime이 실행 가능한 Node와 경로를 관리하는지 실제 코드를 따라가 보겠습니다. 특히 OpenAI Agents SDK의 Agent Loop와 LangGraph의 Graph Runtime을 중심으로 비교 하고, Google ADK와 PydanticAI처럼 두 접근이 겹쳐 있는 구조도 함께 살펴보겠습니다. 이번 글에서는 OpenAI Agents SDK만 먼저 보겠습니다. 전체 구조 요약 Model + Runtime + Runner 구조: Control Flow의 각 단계를 역할별로 분담 Control Flow 단계 주체 무엇을 결정하는가 제품에서의 의미 Action Decision Model 현재 Context를 바탕으로 Tool 호출, Handoff 요청, 응답 생성 중 다음 행동을 선택 Agent가 현재 상황에 맞는 다음 행동을 선택 Output Interpretation Runtime Model Output을 현재 실행 환경에서 어떤 작업으로 처리할지 해석 Model Output을 바로 실행하지 않고 Approval, Guardrail, 현재 실행 상태 등의 정책을 적용 Transition Resolution Runtime Tool과 Side Effect 처리 결과를 바탕으로 이번 Turn을 어떤 상태로 끝낼지 결정 승인 대기, 실행 중단, Final Output, Agent 전환 등의 상태를 제품에서 명시적으로 관리 Loop Advancement AgentRunner 확정된 Control State를 실제 다음 실행에 반영 같은 Agent로 다음 Turn을 진행하거나, Agent를 교체하거나, 실행을 중단·재개·종료 1. Control Flow가 무엇인가요? 먼저 이 글에서 말하는 Control Flow 가 무엇인지부터 정리해보겠습니다. Control Flow는 Agent 분야에서 새로 등장한 개념이 아니라 프로그래밍 언어와 컴퓨터과학에서 오래전부터 사용해 온 개념이라고 하는데요. NIST의 Dictionary for Information Systems는 ISO의 정의를 인용해 Control Flow를 다음과 같이 설명합니다. “an abstraction of all possible paths that an execution sequence may take through a program.” 프로그램의 실행 시퀀스가 따를 수 있는 모든 가능한 경로의 추상화 가장 단순한 프로그램은 위에서 아래로 순서대로 실행되지만, 실제 프로그램에는 조건문과 반복문, 함수 호출, 예외 처리 같은 구조가 들어갑니다. 예를 들어 다음 코드에서는 is_valid 값에 따라 실행할 코드가 달라집니다. if is_valid: process() else: reject() 여기서 중요한 것은 process() 와 reject() 가 무엇을 하는지가 아니라, 둘 중 무엇을 다음에 실행할지를 결정하는 구조 입니다. 이것이 Control Flow입니다. Agent에서도 같은 질문을 할 수 있습니다. Model이 응답을 생성한 뒤 Agent는 다음에 무엇을 해야 할까요? Tool 실행 같은 Model을 다시 호출 다른 Agent에게 실행을 넘기기 (Handoff) Human Approval을 기다리며 멈추기 (HITL) 충분한 결과가 만들어졌다면 실행 종료 즉 Agent의 Control Flow는 현재 실행 상태에서 다음에 어떤 작업을 실행할지 결정하고, 그 실행을 다음 단계로 연결하는 방식 입니다. 예를 들어 전형적인 Agent Loop에서는 Model이 Tool Call을 생성하면 Tool을 실행하고, 그 결과를 다시 Model에 전달합니다. 반면 Graph 기반 Runtime에서는 미리 정의된 Node와 Edge, 현재 State를 바탕으로 다음에 실행할 Node를 결정할 수 있습니다. 이 두 구조는 모두 Model과 Tool을 사용할 수 있지만, 다음 실행 경로를 결정하는 방식이 다릅니다. 그래서 이번 글에서 Control Flow를 살펴볼 때에는 크게 세 가지 축에서 나눠서 살펴보고자 합니다. Action Decision 현재 상황에서 Tool을 호출할지, 답변할지, 다른 Agent에게 넘길지 누가 선택하는가 Transition Resolution Model이나 Tool의 실행 결과를 보고 다음 상태를 누가 확정하는가 Execution Topology 애초에 어떤 실행 경로가 존재할 수 있는지를 어디에서 정의하는가 1. Agent Loop의 Control Flow — OpenAI Agents SDK 이제 실제 코드 수준에서 Control Flow를 살펴보겠습니다. 지난 글에서는 OpenAI Agents SDK의 한 Turn이 끝나면 결과가 NextStepRunAgain, NextStepHandoff, NextStepInterruption, NextStepFinalOutput 중 하나로 정리된다는 것까지 살펴봤습니다. OpenAI Agents SDK는 Control Flow를 어떻게 구현하고 있을까요? AgentRunner : Loop를 돌리는 주체 먼저 가장 바깥쪽인 src/agents/run.py 부터 보겠습니다. Runner.run() 의 설명에는 Agent를 Final Output이 만들어질 때까지 Loop로 실행한다고 되어 있습니다. """ Run a workflow starting at the given agent. The agent will run in a loop until a final output is generated. The loop runs like so: 1. The agent is invoked with the given input. 2. If there is a final output, the loop terminates. 3. If there's a handoff, we run the loop again, with the new agent. 4. Else, we run tool calls (if any), and re-run the loop. """ Source: OpenAI Agents SDK · src/agents/run.py · Runner.run() · https://raw.githubusercontent.com/openai/openai-agents-python/main/src/agents/run.py Handoff가 발생하면 새 Agent로 다시 Loop를 실행하고, 그 외 Tool Call이 있다면 Tool을 실행한 뒤 다시 Loop를 돈다는 구조입니다. 한 번의 Turn은 하나의 AI invocation과 그 과정에서 발생하는 Tool Call을 묶은 단위 입니다. 실제 실행부를 보면, AgentRunner 가 Turn을 직접 세면서 run_single_turn() 을 호출합니다. current_turn += 1 ... turn_result = await run_single_turn(...) Source: OpenAI Agents SDK · src/agents/run.py · AgentRunner.run() 간단히 도식화해볼까요? 하지만 AgentRunner 가 직접 Model을 호출하고 결과를 보고 Tool 실행 여부까지 판단하는 것은 아닙니다. Runner는 말 그대로 '실행시키는 역할'이죠. 실제 처리는 run_single_turn() 아래에서 여러 단계로 나뉩니다. run_single_turn() - 호출된 결과 처리 src/agents/run_internal/run_loop.py 의 run_single_turn() 으로 내려가 보겠습니다. 이 함수는 먼저 현재 Agent를 기준으로 System Prompt, Tool, Handoff, Output Schema 등을 준비합니다. 우선 현재 execution_agent에서 System Prompt와 Prompt 설정을 가져오고, 사용할 Handoff와 Output Schema를 결정합니다. system_prompt, prompt_config = await gather_with_cancel( execution_agent.get_system_prompt(context_wrapper), execution_agent.get_prompt(context_wrapper), ) handoffs = await get_handoffs(execution_agent, context_wrapper) output_schema = get_output_schema(execution_agent) Source: OpenAI Agents SDK · src/agents/run_internal/run_loop.py · run_single_turn() · 2026-10-02 · line 1941-1958 중간에 resolve_tool_name_collisions() 도 호출합니다. 현재 Turn에서 사용할 Tool과 Handoff가 Model 입장에서 같은 이름으로 노출될 수 있기 때문에, run_config.tool_name_collision_policy에 따라 이름 충돌을 먼저 정리하는 역할입니다. 이 단계가 끝나면 이번 Model Call에 전달할 Tool / Handoff 목록이 확정됩니다. 이후 현재까지의 입력과 생성된 Item을 Provider Input으로 만든 다음 get_new_response() 를 호출합니다. new_response = await get_new_response(...) ... return await get_single_step_result_from_response(...) Source: OpenAI Agents SDK · src/agents/run_internal/run_loop.py · run_single_turn() get_new_response() 안에서는 실제 model.get_response() 가 호출됩니다. 이 단계에서 다음 값들이 전달됩니다. system_instructions input model_settings tools output_schema handoffs Model은 이 정보를 보고 Text Response를 만들거나 Function Tool을 호출하고, Handoff에 해당하는 Tool을 호출하는 등의 선택을 할 수 있습니다. 여기까지 오면 Model이 ModelResponse 를 반환합니다. ModelResponse 는 먼저 ProcessedResponse 로 변환된다 그런데 Model 호출이 끝나면 run_single_turn() 은 ModelResponse 를 그대로 실행하지 않고, ProcessedResponse 라는 최종 결과물을 만듭니다. 다음 단계인 get_single_step_result_from_response() 에 넘기고, 여기서 가장 먼저 process_model_response() 를 호출합니다. processed_response = process_model_response( agent=item_agent, all_tools=all_tools, response=new_response, output_schema=output_schema, handoffs=handoffs, existing_items=pre_step_items, run_config=run_config, server_manages_conversation=server_manages_conversation, ... ) ... return await execute_tools_and_side_effects( ... new_response=new_response, processed_response=processed_response, ... ) Source: OpenAI Agents SDK · src/agents/run_internal/turn_resolution.py · get_single_step_result_from_response() · 2026-10-02 · line 2852-2909 이 함수가 하는 일은 Model Provider가 반환한 출력을 Agent Runtime이 이해하고 실행할 수 있는 의미 단위로 해석하는 것 이고 그 결과물이 바로 ProcessedResponse 라고 이해할 수 있습니다. Model Response + 추가 입력 실제 함수 signature부터 살펴보겠습니다. def process_model_response( *, agent: Agent[Any], all_tools: list[Tool], response: ModelResponse, output_schema: AgentOutputSchemaBase | None, handoffs: list[Handoff], existing_items: Sequence[RunItem] | None = None, run_config: RunConfig | None = None, server_manages_conversation: bool = False, server_managed_input_items: Sequence[Any] | None = None, allow_apply_patch_function_fallback: bool = True, ) -> ProcessedResponse: Source: OpenAI Agents SDK · src/agents/run_internal/turn_resolution.py · process_model_response() · 2026-10-02 · line 2138-2150 response 하나만 받는 것이 아니라 현재 Agent, 사용할 수 있는 Tool 전체, Handoff 목록, Output Schema, 이전에 생성한 Item, RunConfig까지 함께 받습니다. 그래야 Model이 출력한 ModelResponse 항목이 Runtime에서 무엇을 의미하는지 결정할 수 있기 때문 입니다. 예를 들어 Model이 다음 Function Call을 반환했다고 해보겠습니다. transfer_to_support(...) 얼핏 보면 그냥 함수 호출이지만, 실제로는 일반 Function Tool이 아닐 수 있습니다. 예를 들어 현재 Agent에 transfer_to_support 라는 Handoff가 등록되어 있다면 이 호출은 Tool 실행이 아니라 Agent를 교체하기 위한 Handoff 요청 으로 해석해야 합니다. 그래서 process_model_response() 는 먼저 현재 실행 환경을 바탕으로 lookup map을 만듭니다 (이외에도 Shell, Apply Patch, Code Interpreter, Programmatic Tool, Hosted MCP 등을 각각 찾아둡니다). handoff_map = {handoff.tool_name: handoff for handoff in handoffs} function_map = build_function_tool_lookup_map( [tool for tool in all_tools if isinstance(tool, FunctionTool)] ) custom_tool_map = { tool.name: tool for tool in all_tools if isinstance(tool, CustomTool) } computer_tool = next( (tool for tool in all_tools if isinstance(tool, ComputerTool)), None, ) Source: OpenAI Agents SDK · src/agents/run_internal/turn_resolution.py · process_model_response() · 2026-10-02 · line 2151-2184 그리고 response.output 을 하나씩 순회하면서 각 Output Item이 실제로 무엇인지 판별합니다. for output in response.output: output_type = get_mapping_or_attr(output, "type") ... Source: OpenAI Agents SDK · src/agents/run_internal/turn_resolution.py · process_model_response() · 2026-10-02 · line 2202-2210 특히 Function Call 처리 부분을 보면 이 함수의 역할이 명확하게 드러납니다. if is_handoff_tool_call(output, handoff_map): ... items.append(HandoffCallItem(raw_item=output, agent=agent)) handoff = ToolRunHandoff( tool_call=output, handoff=handoff_map[output.name], ) run_handoffs.append(handoff) else: lookup_key = get_function_tool_lookup_key_for_call(output) func_tool = function_map.get(lookup_key) if lookup_key is not None else None ... items.append( ToolCallItem( raw_item=output, agent=agent, ... ) ) functions.append( ToolRunFunction( tool_call=output, function_tool=func_tool, ) ) Source: OpenAI Agents SDK · src/agents/run_internal/turn_resolution.py · process_model_response() · 2026-10-02 · line 2755-2836 같은 ResponseFunctionToolCall 이라도 현재 Runtime에 어떤 Handoff와 Tool이 등록되어 있는지를 확인한 뒤 서로 다른 의미로 바꾸고 있습니다. Handoff 이름과 일치하면 ToolRunHandoff Function Tool과 일치하면 ToolRunFunction Tool을 찾지 못하면 설정에 따라 Error 처리 또는 ToolRunFunctionNotFound Apply Patch와 같은 특수 Tool이면 별도의 실행 타입 즉 ModelResponse 는 Model이 무엇을 출력했는지 를 나타내는 객체이고, ProcessedResponse 는 그 출력이 현재 Agent Runtime에서 무엇을 의미하는지 해석한 결과 라고 볼 수 있습니다. ProcessedResponse 정의 더 뜯어보기 ProcessedResponse 의 class 정의를 조금 더 보겠습니다. @dataclass class ProcessedResponse: new_items: list[RunItem] handoffs: list[ToolRunHandoff] functions: list[ToolRunFunction] computer_actions: list[ToolRunComputerAction] local_shell_calls: list[ToolRunLocalShellCall] shell_calls: list[ToolRunShellCall] apply_patch_calls: list[ToolRunApplyPatchCall] tools_used: list[str] mcp_approval_requests: list[ToolRunMCPApprovalRequest] interruptions: list[ToolApprovalItem] function_tools_not_found: list[ToolRunFunctionNotFound] = ... custom_tool_calls: list[ToolRunCustom] = ... Source: OpenAI Agents SDK · src/agents/run_internal/run_steps.py · ProcessedResponse · 2026-10-02 · line 110-124 보면 크게 중요한 두 부류가 있는데요. 첫 번째는 new_items 입니다. MessageOutp