Electron과 FastAPI로 스트리밍 Agent 대화 구현

지난 글에서는 모델이 "판단—도구 호출—결과 관찰—계속 판단"이라는 최소 루프를 완성하도록 했다. 이번 글에서는 Agent의 사고방식을 바꾸지 않고, 여기에 실제로 쓸 수 있는 데스크톱 진입점을 하나 연결한다. Electron은 사용자 인터페이스를 담당하고, FastAPI는 Agent를 실행하며, HTTP는 명령을 담당하고, SSE는 실행 과정을 데스크톱으로 지속적으로 밀어 넣는 역할을 맡는다.

채팅 애플리케이션은 겉보기에는 입력창과 메시지 목록에 불과하지만, Agent 대화는 일반적인 질의응답보다 도구 호출, 실행 상태, 취소, 오류, 재연결이 더 많다. 만약 백엔드가 모든 실행이 끝난 뒤에야 텍스트 한 단락을 반환한다면, 사용자는 Agent가 지금 무엇을 하고 있는지 알 수 없을 뿐 아니라 잘못된 방향으로 갈 때 제때 멈출 수도 없다.

따라서 DevMind의 M2는 단순히 "글자를 한 글자씩 표시하는 것"이 아니라, 관찰 가능하고 취소 가능하며 재연결 가능한 Run 이벤트 채널을 구축하는 것이다.

1. 이번에 완성할 것

지난 글에서는 이미 devmind-server에서 실제 모델과 도구를 연결하는 최소 Agent Loop를 완성했다. 이번 글은 한 걸음 더 나아간다. 예측 가능한 AG-UI Mock을 유지하는 동시에, 지난 글의 Agent Loop를 정식 AG-UI 엔드포인트에 연결하고, 이어서 pnpm으로 Electron + React + TypeScript 기반의 devmind-desktop을 생성한다. 사용자는 Mock으로 안정적으로 인터페이스를 연동·디버깅할 수 있고, 데스크톱에서 실제로 모델 판단, 도구 실행, 최종 응답을 트리거할 수도 있다.

RunEvent를 자체 개발하지 않는다. Renderer는 @ag-ui/client를 사용해 표준 RunAgentInput을 전송하고, FastAPI는 ag-ui-protocol을 사용해 표준 AG-UI 이벤트 스트림을 반환한다.

Electron Renderer │ ├── POST / api / agent（RunAgentInput）────────→ Python Agent Server │ │ │←── RUN_STARTED ──────────────────────────────────┤ │←── TEXT_MESSAGE_START / CONTENT / END ───────────┤ │←── TOOL_CALL_START / ARGS / END / RESULT ────────┤ │←── CUSTOM（소량의 DevMind 도메인 이벤트）───────────────┤ │←── RUN_FINISHED / RUN_ERROR ─────────────────────┘

1.1 이번 단계의 완료 기준

- apps/devmind-desktop을 처음부터 생성하고 실행한다.

- 기존 apps/devmind-server에 AG-UI Python SDK를 설치하고, /api/agent/mock을 유지하며, 정식 /api/agent가 지난 글의 실제 Agent Loop를 호출하도록 한다.

- Renderer가 HttpAgent를 통해 스트리밍 텍스트, 도구 이벤트, Run 생명주기를 수신한다.

- 범용 이벤트에는 AG-UI를 사용하고, DevMind는 네임스페이스와 버전 번호가 붙은 소량의 CUSTOM 이벤트만 추가한다.

- Zod는 DevMind 도메인 이벤트와 이후의 로컬 RPC만 검증하며, AG-UI 범용 이벤트를 중복 검증하지 않는다.

- abortRun()은 현재 클라이언트의 관찰을 중단하는 것일 뿐, 영속 Agent를 안정적으로 취소하는 것과 동일하지 않다는 점을 명확히 한다.

1.2 이번 단계에서 여전히 하지 않는 것

- 이번 글에서 LangGraph Checkpointer와 진정한 재시작 간 복구를 완성하지 않는다.

- 이번 글에서 완전한 Interrupt/Resume를 구현하지 않고, 우선 표준 프로토콜과 페이지 경계만 확정한다.

- 이번 글에서 Jira, GitLab, Knowledge, Workflow, Permission의 실제 환경을 연동하지 않는다.

- 이번 글에서 Electron 로컬 도구 WSS/RPC를 구현하지 않는다. 이는 이후에 Main 프로세스가 담당한다.

- 데모 코드를 이미 완성된 프로덕션 역량인 것처럼 서술하지 않는다.

2. 먼저 AG-UI, HTTP/MCP, WSS/RPC의 역할을 구분하자

DevMind는 하나의 연결로 모든 것을 담당하게 하지 않고, 경계에 따라 프로토콜을 선택한다. 최종 구조는 다음과 같다.

React Renderer │ │ AG-UI: 메시지, 실행 상태, 도구 표시, Interrupt / Resume ▼ Python Agent Server + LangGraph │ ├── HTTP / MCP: Jira, GitLab, Knowledge, Workflow, Permission │ └── 독립 WSS / RPC: Electron 로컬 Git, 파일, 터미널, Playwright

경계 　 프로토콜 　 담당하는 것 　 담당하지 않는 것 React Renderer ↔ Agent Server 　 AG-UI over HTTP/SSE 　 메시지, Run 생명주기, 도구 표시, 상태 동기화, Interrupt/Resume 　 로컬 명령을 직접 실행하지 않고, 영속화를 대신하지 않음 Agent Server ↔ 기업 서비스 　 HTTP / MCP 　 Jira, GitLab, 지식 베이스, 워크플로, 권한 역량 　 UI 이벤트 형식을 담당하지 않음 Agent Server ↔ Electron Main 　 독립 WSS/RPC 　 로컬 Git, 파일, 터미널, Playwright의 양방향 호출 　 Renderer에 임의의 시스템 권한을 노출하지 않음

AG-UI는 애플리케이션 계층 프로토콜이고, SSE는 현재 그것이 사용하는 스트리밍 전송 방식이다. Renderer는 @ag-ui/client를 통해 POST 요청을 보내고 이벤트 스트림을 읽으며, Python Server는 ag-ui-protocol을 통해 표준 이벤트를 출력한다. 우리는 AG-UI를 채택하고, 더 이상 UI 프로토콜 전체를 자체 개발하지 않는다.

LangGraph는 Agent의 영속 실행, Checkpoint, Interrupt/Resume를 담당한다. 페이지가 닫히거나 HTTP 스트림이 끊기는 것은 클라이언트가 잠시 더 이상 관찰하지 않는다는 의미일 뿐이며, Server의 영속 작업을 자동으로 소멸시켜서는 안 된다.

Electron 로컬 도구는 여전히 메인 Agent 시스템 설계 안에 두되, 연결과 실행은 Electron Main이 관리한다. Renderer는 안전한 Preload + IPC를 통해서만 상태를 표시하거나 사용자 확인을 제출하며, 로컬 도구 WSS를 직접 보유할 수 없고 Server가 내려보낸 임의의 명령을 실행할 수도 없다.

3. 먼저 Electron을 알아보고, 그다음 실행 경계를 이해하자

이전에 주로 React 웹 애플리케이션을 개발했다면, Electron을 우선 "웹 인터페이스를 데스크톱 애플리케이션에 담는" 실행 플랫폼으로 이해하면 된다. Electron은 Chromium과 Node.js를 결합한다. Chromium은 페이지 렌더링을 담당하고, Node.js와 Electron API는 창, 파일, 터미널, 메뉴, 시스템 알림 등의 데스크톱 기능을 담당한다. 따라서 동일한 프런트엔드 기술을 Windows, macOS, Linux에서 실행할 수 있다.

하지만 Electron이 "React 페이지가 Node.js를 마음대로 호출할 수 있다"는 뜻은 아니다. 페이지는 모델 출력, 저장소 내용, 도구 결과를 표시하는데, 이러한 내용은 모두 신뢰할 수 없는 입력일 수 있다. 만약 Renderer가 파일을 직접 읽고 쓰거나 명령을 실행할 수 있다면, 잘못 렌더링된 내용 한 조각이 인터페이스 경계를 넘어 이 컴퓨터에 영향을 미칠 수 있다. 따라서 Electron을 이해하는 첫걸음은 API를 외우는 것이 아니라 서로 다른 실행 경계를 먼저 구분하는 것이다.

3.1 React Web에서 Electron으로, 먼저 네 가지 개념을 알아야 한다

개념 　 우선 어떻게 이해할 수 있는가 　 DevMind에서의 역할 Main 　 데스크톱 애플리케이션의 백그라운드 총괄 　 창을 생성하고, 애플리케이션 생명주기를 관리하며, 파일·터미널 등 로컬 시스템 역량을 담당한다. Renderer 　 React를 실행하는 페이지 　 입력, 메시지, 실행 상태, 상호작용 표시를 담당하며, 본질적으로는 여전히 브라우저 페이지의 보안 경계를 따른다. Preload 　 페이지 로딩 전에 주입되는 통제된 브리지 스크립트 　 contextBridge를 통해 Renderer에 소량이고 명확하며 검증 가능한 데스크톱 역량을 노출한다. Preload는 세 번째 독립 프로세스가 아니다. IPC 　 Main과 Renderer 사이의 통신 메커니즘 　 Renderer가 Preload를 거쳐 Main에 로컬 작업 실행을 요청하고, 다시 구조화된 결과를 받도록 한다. IPC도 프로세스가 아니다.

따라서 더 정확한 표현은 이렇다. Electron은 주로 Main과 하나 이상의 Renderer 프로세스로 구성되며, Preload는 안전한 브리지 스크립트이고, IPC는 이들 사이의 통신 방식이다. 이어서 각각 이 경계들이 현재 프로젝트에서 어떤 역할을 담당하는지 살펴본다.

프런트엔드 엔지니어에게 Renderer는 일반적인 React 애플리케이션과 매우 비슷하다. 실제로 바뀌어야 하는 것은, Electron이 "Node.js가 딸린 브라우저 페이지"가 아니라 Main, Preload, Renderer라는 세 가지 보안 경계라는 점이다.

3.2 Main: 애플리케이션 생명주기와 시스템 역량

Main은 창 생성, 애플리케이션 종료, 메뉴, 이후의 로컬 도구 프로세스를 담당한다. Node.js와 운영체제 역량을 보유하므로, 임의의 IPC나 파일 인터페이스를 페이지에 직접 노출해서는 안 된다.

3.3 Preload: 통제된 역량 브리지

Preload는 격리된 컨텍스트에서 실행되며, contextBridge를 통해서만 명확하고 검증 가능한 역량을 노출한다. M2의 HTTP와 SSE는 Renderer가 Web API로 직접 완료할 수 있으므로, Preload가 당장 모든 네트워크 요청을 프록시할 필요는 없다.

3.4 Renderer: React 인터페이스를 담당

Renderer는 입력, 메시지 표시, 실행 상태, 중지 버튼을 담당하며, Node.js, 파일 시스템, 터미널에 직접 접근하지 않는다. 이후 메시지에 악성 HTML이 포함되더라도, 그것 때문에 로컬 시스템 권한을 얻을 수 없다.

3.5 IPC: 서로 다른 실행 경계 간의 통신 방식

IPC는 Inter-Process Communication의 약자, 즉 "프로세스 간 통신"이다. 새로운 프로세스가 아니라 Main과 Renderer가 메시지를 교환하는 메커니즘이다. Renderer는 파일 시스템, 터미널, 시스템 대화상자를 직접 호출할 수 없다. 이러한 역량이 필요할 때는 Preload가 노출한 최소 인터페이스를 호출하고, Preload가 IPC를 통해 요청을 Main에 넘긴 뒤 결과를 다시 페이지로 반환해야 한다.

한 번의 전형적인 호출: React 페이지가 window.desktop.openFile()를 호출한다 → Preload 내부에서 ipcRenderer.invoke("file:open")를 실행한다 → Main이 ipcMain.handle()로 요청을 받아 시스템 역량을 호출한다 → 결과가 원래 경로를 따라 Renderer로 반환된다.

IPC 채널은 백엔드 API처럼 설계해야 한다. 채널 이름이 명확하고, 파라미터는 Zod로 검증되며, 반환값 구조가 고정되어 있고, 비즈니스에 필요한 역량만 개방한다. ipcRenderer 전체나 fs 또는 임의의 명령 실행 역량을 window에 직접 붙여서는 안 된다.

이번 단계의 HTTP 요청과 SSE 구독은 Renderer가 FastAPI를 직접 호출할 수 있으므로, 당장 IPC로 네트워크 요청을 프록시할 필요는 없다. 나중에 node-pty, 로컬 파일, 시스템 알림, 보안 자격 증명을 연동할 때 Preload + IPC를 통해 Main의 로컬 역량에 접근하면 된다.

이 장은 우선 개념을 세우는 것이며, 아직 생성하지 않은 파일을 미리 붙여 넣지 않는다. 제14장에서는 지난 글에서 이미 완성한 devmind-server부터 시작해, devmind-desktop을 단계적으로 생성하고, 의존성을 설치한 뒤, Main, Preload, Renderer를 작성한다. 구현할 때 크로스 오리진을 피하려고 webSecurity: false를 설정하지 말고, Renderer에서 nodeIntegration도 켜지 말아야 한다.

4. AG-UI로 Agent와 UI의 데이터 형식을 통일하기

이 단계에서는 여전히 SSE로 스트리밍 데이터를 전송하지만, RunEvent 전체를 자체 설계하지는 않는다. SSE는 전송 방식일 뿐이고, AG-UI가 바로 Agent Server와 Renderer가 공동으로 이해하는 이벤트 프로토콜이다.

DevMind의 목표 구조는 다음과 같다.

이 그림에는 세 가지 서로 다른 프로토콜 경계가 있다.

- Renderer와 Agent Server 사이에는 AG-UI를 사용하며, 메시지, 실행 생명주기, 도구 표시, 상태 동기화, 중단 복구를 담당한다.

- Agent Server와 Jira, GitLab, Knowledge, Workflow, Permission 등 기업 서비스 사이에는 HTTP 또는 MCP를 사용한다.

- Agent Server와 Electron Main의 로컬 도구 사이에는 독립적이고 범위가 매우 작은 WSS/RPC를 사용한다. Renderer는 이 연결을 직접 보유하지 않는다.

4.1 왜 자체 개발한 RunEvent를 계속 쓰지 않는가

assistant.delta, tool.call.started, run.succeeded 같은 사설 이벤트를 계속 사용하면 단기적으로는 코드가 많지 않아 보이지만, 나중에는 버전 호환, 도구 호출, 상태 동기화, Interrupt/Resume, 다중 프런트엔드 접속 문제를 스스로 해결해야 한다. AG-UI는 이미 이러한 범용 문제를 위해 이벤트와 클라이언트 SDK를 정의해 두었으므로, DevMind는 비즈니스 차이만 남기고 범용 프로토콜을 중복해서 발명하지 않는다.

페이지가 표현해야 하는 내용과 AG-UI 표준 이벤트 Run 시작 RUN_STARTED, Assistant 메시지 시작·증분·종료 TEXT_MESSAGE_START, TEXT_MESSAGE_CONTENT, TEXT_MESSAGE_END, 도구 이름·인자·종료·결과 TOOL_CALL_START, TOOL_CALL_ARGS, TOOL_CALL_END, TOOL_CALL_RESULT, Agent 상태 스냅샷과 증분 STATE_SNAPSHOT, STATE_DELTA, Run 정상 종료 RUN_FINISHED, Run 실패 RUN_ERROR, 사람의 입력 대기 RUN_FINISHED(이때 outcome.type을 interrupt로 지정)

AG-UI의 Python 필드는 snake_case를 사용하며, 네트워크로 인코딩된 후에는 TypeScript가 더 익숙한 camelCase로 변환된다. 예를 들어 Python의 message_id는 Renderer가 받는 이벤트에서 messageId가 된다. 프런트엔드와 백엔드가 두 벌의 필드 매핑을 다시 손으로 작성해서는 안 된다.

4.2 표준 이벤트 우선, 소수의 도메인 이벤트만 추가

표준 이벤트가 이미 의미를 표현할 수 있을 때는 반드시 표준 이벤트를 직접 사용해야 한다. UI가 실제로 알아야 하지만 AG-UI에 대응하는 의미가 없는 기업 정보에 대해서만 CUSTOM을 사용한다.

첫 버전에서는 버전 번호가 붙은 세 개의 도메인 이벤트만 미리 마련한다.

- devmind.workflow.status.v1 : 요구사항 흐름의 현재 노드와 상태를 표시한다.

- devmind.permission.required.v1 : 특정 고위험 동작에 승인 또는 권한 부여가 필요함을 알린다.

- devmind.artifact.created.v1 : 요구사항 문서, 테스트 보고서 등의 산출물이 생성되었음을 알린다.

이벤트 이름에는 반드시 devmind. 네임스페이스와 버전 번호가 붙어야 한다. CUSTOM은 소량의 UI 알림만 담을 수 있으며, 모든 내부 이벤트를 그대로 Renderer에 전달해서는 안 된다.

4.3 내부 도메인 이벤트와 AG-UI 이벤트는 같은 것이 아니다

Agent Server 내부에는 여전히 자체적인 도메인 이벤트 모델이 남아 있다. 예를 들어 WorkflowNodeChanged, PermissionRequested, LocalToolExecutionStarted 등이다. 이러한 이벤트는 비즈니스 디커플링, 감사, 서비스 간 협업에 사용되며, UI 경계에 도달하면 Adapter가 다시 AG-UI 표준 이벤트 또는 소량의 CUSTOM 이벤트로 매핑한다.

즉, 내부 도메인 모델은 기업 비즈니스를 안정적으로 지원할 수 있고, AG-UI는 인터페이스를 안정적으로 지원한다. 둘은 매핑을 통해 연결되며 서로를 오염시키지 않는다.

5. AG-UI의 HTTP + SSE 상호작용 방식

AG-UI의 HttpAgent는 한 번의 POST 요청으로 RunAgentInput을 제출하고, 같은 응답에서 text/event-stream을 계속 읽는다. 이는 브라우저 네이티브 EventSource의 GET 구독 모델이 아니다.

Renderer Agent Server │ │ ├── POST /api/agent ──────────────→│ RunAgentInput │ │ │←──────── RUN_STARTED ────────────┤ │←──── TEXT_MESSAGE_CONTENT ───────┤ │←──────── TOOL_CALL_* ────────────┤ │←──────── CUSTOM(소량)──────────┤ │←──── RUN_FINISHED / RUN_ERROR ───┤ │ │

RunAgentInput에서 가장 중요한 필드는 다음과 같다.

- threadId : 하나의 연속된 세션을 나타내는 안정적인 식별자.

- runId : 이번 실행의 고유 식별자.

- messages : 현재 세션 메시지.

- state : 프런트엔드와 백엔드에서 동기화해야 하는 Agent 상태.

- tools : 프런트엔드가 선언하고 Agent가 호출하도록 허용한 도구 정의.

- context : 현재 요청에서 사용하는 컨텍스트 정보.

- resume : LangGraph Interrupt를 복구할 때 제출하는 답변.

정식 프로젝트에서는 로그인 상태와 권한 컨텍스트로 사용자 신원을 검증해야 하며, 요청 본문에 threadId가 포함되었다는 이유만으로 임의의 세션을 읽도록 허용해서는 안 된다.

6. LangGraph는 지속 실행을 담당하고, AG-UI는 영속화를 담당하지 않는다

AG-UI는 '프런트엔드와 백엔드가 Agent 이벤트를 어떻게 교환하는가'를 규정하지만, 작업을 저장하지는 않는다. DevMind는 이후 LangGraph Checkpointer를 사용해 스레드 상태, 노드 진행 상황, Interrupt를 저장하여 Agent Server가 재시작되거나 페이지가 새로 고쳐지거나 사용자가 나중에 돌아와도 복구할 수 있게 한다.

책임은 엄격히 분리해야 한다.

- AG-UI: 범용 UI 프로토콜과 이벤트 스트림.

- LangGraph: Agent 그래프 실행, Checkpoint, Interrupt/Resume.

- PostgreSQL: 비즈니스 데이터, Thread/Run 메타데이터, 감사 기록.

- Redis: 단기 상태, 분산 조정 및 필요한 이벤트 포워딩.

- Workflow Server: 요구사항 분석부터 테스트 완료까지의 기업 연구개발 프로세스 흐름을 담당하며, Agent 세션 복구를 맡지 않는다.

HttpAgent.abortRun()은 현재 클라이언트의 HTTP 스트림과 관찰 과정만 중단할 뿐, 영속화된 Agent를 신뢰성 있게 취소하는 것과 동일하지 않다. 정식 비즈니스 취소는 반드시 Agent Server가 Run 상태를 검증하고 취소 의도를 기록하며, LangGraph가 안전 지점에서 멈추도록 해야 한다.

7. Interrupt / Resume의 통일된 의미

사용자가 고위험 작업을 확인해야 할 때 permission.waiting 같은 사설 종료 상태를 새로 추가하지 않는다. LangGraph가 Interrupt를 생성하면, AG-UI는 RUN_FINISHED를 통해 outcome.type = interrupt와 중단 목록을 반환한다. 사용자가 확인한 후 Renderer는 동일한 threadId와 RunAgentInput.resume을 사용해 답변을 제출한다.

전형적인 흐름은 다음과 같다.

- Agent가 Push 실행, MR 생성 또는 테스트 환경 배포를 준비한다.

- Permission Service가 해당 동작에 확인이 필요하다고 판단한다.

- LangGraph가 해당 노드에서 Interrupt하고 Checkpoint를 저장한다.

- Renderer가 AG-UI Interrupt를 받아 승인 카드를 표시한다.

- 사용자가 승인하거나 거부한다.

- Renderer가 resume을 통해 결과를 제출한다.

- LangGraph가 새로운 서로 무관한 작업을 만드는 것이 아니라 원래 노드에서 계속한다.

현재 M2에서는 먼저 표준 메시지, 도구, Run 이벤트를 완성하고, M3에서 ag-ui-langgraph를 연동한 후 진정한 영속화 Interrupt/Resume를 구현한다. 이번 글에서는 먼저 프로토콜과 UI 경계를 정해 두어 이후에 프런트엔드를 다시 리팩터링하지 않도록 한다.

8. Electron 로컬 도구가 왜 독립적인 WSS/RPC를 사용하는가

Git, 파일, 터미널, Playwright 조작은 사용자 컴퓨터에서 발생하며, Python Server가 있는 컴퓨터에 있지 않다. 이들은 일반 UI 이벤트도 아니고 Jira, GitLab 같은 원격 기업 서비스도 아니므로, AG-UI의 전송 계층에 밀어 넣지 않는다.

정식 링크는 다음과 같다.

LangGraph Tool Node │ │ WSS/RPC: requestId, tool, args, timeout, result, error ▼ Electron Main ├── 로컬 Git ├── 파일 시스템 ├── node-pty / xterm 백엔드 └── Playwright

Electron Main은 기기 신원, 연결, 도구 화이트리스트, 인자 검증, 타임아웃, 취소, 결과 반환을 담당하고, Preload는 Renderer에 필요한 안전 능력만 노출한다. Renderer는 AG-UI의 TOOL_CALL_*을 통해 실행 과정을 표시하지만, Server가 보낸 임의의 명령을 직접 실행하지는 않는다.

9. TanStack Query와 Zod는 어디에 두는가

AG-UI SDK는 이미 범용 이벤트의 파싱과 타입을 처리하므로, Renderer는 RUN_STARTED, TEXT_MESSAGE_CONTENT 등의 이벤트를 위해 Zod Schema를 다시 작성하지 않는다.

두 기술은 여전히 유지하되 책임을 다음과 같이 조정한다.

- TanStack Query: Jira, Workflow, Permission, 지식베이스 등 일반 HTTP 리소스를 읽고, 비스트리밍 Mutation을 실행한다.

- Zod: DevMind의 CUSTOM 이벤트 페이로드, Electron IPC 인자, 로컬 WSS/RPC 요청을 검증한다.

고빈도 Token 증분은 대화 상태로 직접 들어가며, 모든 Chunk를 TanStack Query 캐시에 기록해서는 안 된다.

10. M2의 구현 경계

이번 글에서는 AG-UI Python SDK로 결정론적 데모 Agent를 구성하고, @ag-ui/client로 이벤트를 수신한다. 이렇게 하면 모델 키, 네트워크 변동 또는 LangGraph 설정에 방해받지 않고 프로토콜, SSE, Electron Renderer, 도구 카드를 먼저 검증할 수 있다.

M2 완료 후 달성해야 할 것:

- Renderer가 표준 RunAgentInput을发起할 수 있다.

- Server가 합법적인 AG-UI SSE 이벤트를 출력할 수 있다.

- 텍스트가 Delta에 따라 단계별로 표시된다.

- 도구 호출이 동일한 toolCallId에 따라 하나의 카드로 집계된다.

- CUSTOM 이벤트가 Zod 검증을 거친다.

- UI가 실행, 완료, 실패, 클라이언트 관찰 중단을 구분할 수 있다.

M2는 영속화 복구를 이미 완료했다고 주장하지 않지만, 결정론적 Mock과 실제 Agent Loop 두 가지 AG-UI 링크를 동시에 갖춘다. M3은 정식 실제 링크의 내부 Runtime만 LangGraph + Checkpointer + ag-ui-langgraph로 마이그레이션하고, Mock은 계속 프런트엔드 연동과 프로토콜 회귀 테스트에 사용한다.

11. 개발 환경의 CORS와 보안 경계

Electron 개발 시기의 Renderer는 보통 로컬 Vite 주소에서 실행되므로, FastAPI는 실제로 사용하는 개발 Origin만 허용한다. 손쉽게 하려고 allow_origins=["*"]를 설정하거나 Electron의 webSecurity를 꺼서는 안 된다.

프로덕션 환경에서는 추가로 처리해야 한다.

- 단기 액세스 토큰 또는 보안 쿠키로 Agent Server를 인증한다.

- 사용자의 threadId, Jira, 프로젝트에 대한 접근 권한을 검증한다.

- 로그에 전체 Prompt, 자격 증명, 도구의 민감한 인자를 기록해서는 안 된다.

- 로컬 RPC는 기기 수준의 단기 자격 증명을 사용하고, 도구, 디렉터리, 명령 범위를 제한한다.

12. UI가 AG-UI 이벤트를 어떻게 소비하는가

페이지는 모든 이벤트를 하나의 '생각 중'으로 렌더링해서는 안 되고, 의미에 따라 집계해야 한다.

- RUN_STARTED : 상태를 실행 중으로 전환하고 runId를 기록한다.

- TEXT_MESSAGE_START/CONTENT/END : Assistant 메시지를 생성하고, Delta를 덧붙이고, 스트리밍 커서를 종료한다.

- TOOL_CALL_START/ARGS/END/RESULT : toolCallId에 따라 도구 카드를 생성하고 갱신한다.

- CUSTOM : name에 따라 해당 도메인 카드로 넘긴다.

- RUN_FINISHED : success와 interrupt를 구분한다.

- RUN_ERROR: 보안 오류 메시지와 문제를 추적할 수 있는 Run ID를 표시한다.

AG-UI 클라이언트는 메시지와 이벤트 처리 흐름을 유지하고, React 컴포넌트는 비즈니스에 필요한 파생 상태만 로컬 State에 저장한다. 나중에 더 완전한 Agent UI를 연결할 때도 Server의 이벤트 형식을 바꿀 필요가 없다.

13. 먼저 최종 아키텍처를 세우고, 단계별로 구현한다

이 글의 코드는 "최종 아키텍처와 호환되는 최소 구현"이며, 최종 런타임이 아니다:

단계 Agent Server Renderer 로컬 도구 M2 ag-ui-protocol + 결정적 Mock + 실제 Agent Loop 어댑터 @ag-ui/client가 메시지와 도구 이벤트를 소비 아직 연결하지 않음 M3 LangGraph + Checkpointer + ag-ui-langgraph Interrupt/Resume 추가 아직 연결하지 않음 M4 LangGraph Tool Node가 로컬 도구를 스케줄링 도구 권한과 진행 상황 표시 Electron Main + WSS/RPC 이후 HTTP/MCP로 엔터프라이즈 서비스 연결 흐름, 권한, 산출물 표시 엄격한 화이트리스트와 감사

이렇게 하면 학습 순서는 여전히 최소 폐쇄 루프에서 점진적으로 확장되지만, 프로토콜과 아키텍처 방향은 처음부터 올바르다.

14. UI 설계와 전체 실습

아래에서는 기존 apps/devmind-server에서 이어서, 먼저 Mock과 실제 Agent Loop가 동시에 AG-UI를 출력하도록 하고, 그다음 apps/devmind-desktop을 처음부터 생성한다. Electron 프로젝트는 일괄적으로 pnpm을 사용하며, 각 단계마다 먼저 디렉터리와 파일을 만들고, 그다음 의존성을 설치하고 코드를 작성하고 실행 및 검증한다.

14.1 시작점 확인: M1의 devmind-server 계속 사용

이 글에서는 Python 프로젝트를 다시 초기화하지 않는다. 먼저 이전 글의 apps/devmind-server를 완성하고, 이미 src/devmind_server/main.py, src/devmind_server/api/health.py, agent/loop.py, agent/model_gateway.py, 도구 레지스트리, 세 가지 데모 도구와 테스트가 있는지 확인한다. 동시에 로컬 .env에 사용 가능한 실제 모델 설정이 채워져 있는지 확인한다.

pwd # 현재 디렉터리를 표시하고, apps를 포함한 리포지터리 루트에 있는지 확인 ls # 리포지터리 루트의 파일과 폴더 확인 ls apps # apps 아래에 devmind-server가 이미 있는지 확인 cd apps/devmind-server # 이전 글에서 이미 완성한 Python 백엔드 프로젝트로 이동 uv sync # pyproject.toml과 uv.lock에 따라 프로젝트 가상 환경 복원 uv run pytest -q # 먼저 이전 글의 테스트를 실행해 현재 시작점에 문제가 없는지 확인 cd ../.. # 리포지터리 루트로 돌아가 이후 증분 개발 계속

14.2 백엔드에 AG-UI 설치 및 파일 생성

14.2.1 의존성 설치

cd apps/devmind-server # 리포지터리 루트에서 기존 백엔드 프로젝트로 진입 uv add ag-ui-protocol sse-starlette # AG-UI Python 타입/인코더와 SSE 응답 라이브러리 설치 uv sync # 의존성 동기화 및 현재 프로젝트의 .venv 갱신

이 시점에는 ag-ui-langgraph를 설치하지 않는다. 이 글에서는 결정적 Mock을 유지해 프로토콜 이해를 돕는 동시에, 가벼운 어댑터 계층으로 이전 글에서 직접 작성한 Agent Loop를 연결한다. M3에서 다시 정식 실제 경로의 내부 Runtime을 LangGraph로 이전한다.

14.2.2 디렉터리와 파일 생성

mkdir -p src/devmind_server/agent # Agent 구현 디렉터리 존재 확인 mkdir -p src/devmind_server/api # FastAPI 라우트 디렉터리 존재 확인 mkdir -p tests # 백엔드 테스트 디렉터리 존재 확인 touch src/devmind_server/agent/ag_ui_demo.py # 결정적 AG-UI Mock 생성기를 생성 및 보존 touch src/devmind_server/api/ag_ui.py # 먼저 Mock을 제공하고 이후 실제 엔드포인트를 추가할 FastAPI 라우트 생성 touch tests/test_ag_ui_stream.py # AG-UI 및 Runtime 이벤트 스트림 테스트 파일 생성

이 단계에서는 먼저 Mock, Router, 테스트 파일만 생성한다. 실제 Runtime 계약, 이벤트 타입, 팩터리와 Adapter는 14.4에서 필요할 때 하나씩 생성하여, 독자가 아직 실제 경로에 들어서기도 전에 설명되지 않은 디렉터리를 마주하지 않도록 한다.

apps/devmind-server/ ├── pyproject.toml ├── src/devmind_server/ │ ├── main.py │ ├── agent/ │ │ ├── loop.py # 직접 작성한 AgentRuntime으로 astream과 호환 run 구현 │ │ ├── events.py # 프로토콜과 무관한 Runtime 이벤트 │ │ ├── runtime.py # RunCommand와 AgentRuntime 인터페이스 │ │ ├── factory.py # 모델, 도구, Runtime 조립 진입점 │ │ ├── model_gateway.py # astream을 지원하는 실제 모델 게이트웨이 │ │ ├── tool_registry.py # 도구 등록과 통합 실행 진입점 │ │ ├── ag_ui_demo.py # 보존된 결정적 프로토콜 레벨 Mock │ │ └── adapters/ │ │ ├── __init__.py │ │ └── ag_ui.py # Runtime에서 AG-UI로의 Adapter │ └── api/ │ ├── health.py │ └── ag_ui.py # /api/agent와 /api/agent/mock └── tests/ └── test_ag_ui_stream.py

14.3 결정적인 AG-UI Mock 이벤트 생성기 보존

아래 코드를 src/devmind_server/agent/ag_ui_demo.py에 작성한다. 이것은 실제 모델을 호출하지 않고 출력이 완전히 예측 가능한 Mock이며, 어떤 사설 RunEvent도 정의하지 않고 AG-UI SDK의 이벤트 객체를 직접 생성한다. 나중에 이 파일을 삭제하지 말 것, 이 파일은 계속 프런트엔드 연동과 자동화 테스트에 사용된다.

import asyncio # 비동기 대기를 제공하며, 모델과 도구의 스트리밍 실행 과정을 시뮬레이션하는 데 사용 import json # 도구 파라미터와 도구 결과를 안정적인 JSON 문자열로 인코딩 from collections.abc import AsyncIterator # 비동기 이벤트 생성기의 반환 타입 설명 from uuid import uuid4 # 메시지, 도구 호출, 도구 결과에 고유 식별자 생성 from ag_ui.core import BaseEvent # 모든 AG-UI 이벤트가 공통으로 상속하는 기반 타입 임포트 from ag_ui.core import CustomEvent # 소수 비즈니스 확장에서 사용하는 사용자 정의 이벤트 임포트 from ag_ui.core import RunAgentInput # Renderer가 제출한 표준 실행 입력 임포트 from ag_ui.core import RunErrorEvent # 실행 실패 이벤트 임포트 from ag_ui.core import RunFinishedEvent # 실행 성공 또는 중단 완료 이벤트 임포트 from ag_ui.core import RunStartedEvent # 실행 시작 이벤트 임포트 from ag_ui.core import TextMessageContentEvent # Assistant 텍스트 증분 이벤트 임포트 from ag_ui.core import TextMessageEndEvent # Assistant 텍스트 종료 이벤트 임포트 from ag_ui.core import TextMessageStartEvent # Assistant 텍스트 시작 이벤트 임포트 from ag_ui.core import ToolCallArgsEvent # 도구 파라미터 증분 이벤트 임포트 from ag_ui.core import ToolCallEndEvent # 도구 호출 파라미터 종료 이벤트 임포트 from ag_ui.core import ToolCallResultEvent # 도구 실행 결과 이벤트 임포트 from ag_ui.core import ToolCallStartEvent # 도구 호출 시작 이벤트 임포트 def last_user_text ( input_data: RunAgentInput ) -> str : # 표준 입력에서 마지막 사용자 텍스트 읽기 if not input_data.messages: # 메시지 목록이 비어 있으면 시연용 기본 질문 사용 return "DevMind를 소개해 주세요" # 시연에 편한 기본 질문 반환 content = getattr (input_data.messages[- 1 ], "content" , "" ) # 마지막 메시지의 content를 안전하게 읽기 return content if isinstance (content, str ) else str (content) # 문자열이 아닌 내용을 표시 가능한 텍스트로 변환 async def run_mock ( input_data: RunAgentInput ) -> AsyncIterator[BaseEvent]: # 데모 Agent의 표준 이벤트 스트림 정의 message_id = str (uuid4()) # 이번 Assistant 메시지 ID 생성 tool_call_id = str (uuid4()) # 이번 도구 호출 ID 생성 tool_result_message_id = str (uuid4()) # 도구 결과 메시지 ID 생성 prompt = last_user_text(input_data) # 사용자가 방금 제출한 질문 가져오기 try : # 예기치 않은 예외를 포착해 AG-UI 오류 이벤트로 변환 yield RunStartedEvent(thread_id=input_data.thread_id, run_id=input_data.run_id) # 이번 Run이 시작되었음을 UI에 알림 yield TextMessageStartEvent(message_id=message_id, role= "assistant" ) # Assistant 메시지 하나를 생성하도록 UI에 알림 for delta in ( "질문을 분석하는 중입니다:" , prompt, "。\n" ): # 모델이 텍스트를 반환하는 것을 구간별로 시뮬레이션 await asyncio.sleep( 0.25 ) # 관찰 가능한 스트리밍 간격을 둠 yield TextMessageContentEvent(message_id=message_id, delta=delta) # 현재 텍스트 증분 전송 yield ToolCallStartEvent(tool_call_id=tool_call_id, tool_call_name= "demo_text" , parent_message_id=message_id) # 도구 카드 생성을 UI에 알림 yield ToolCallArgsEvent(tool_call_id=tool_call_id, delta=json.dumps({ "text" : prompt}, ensure_ascii= False )) # 도구 파라미터 JSON 전송 yield ToolCallEndEvent(tool_call_id=tool_call_id) # 도구 파라미터가 모두 전송되었음을 표시 await asyncio.sleep( 0.4 ) # 도구 실행 소요 시간 시뮬레이션 result = { "ok" : True , "summary" : "데모 도구 실행 완료" } # 구조화된 도구 결과 구성 yield ToolCallResultEvent(message_id=tool_result_message_id, tool_call_id=tool_call_id, content=json.dumps(result, ensure_ascii= False )) # 동일한 호출 ID의 도구 결과 반환 yield CustomEvent(name= "devmind.workflow.status.v1" , value={ "node" : "requirement_analysis" , "status" : "completed" }) # 네임스페이스로 제약된 도메인 이벤트 하나를 시연 yield TextMessageContentEvent(message_id=message_id, delta= "데모 작업이 완료되었습니다." ) # 마지막 Assistant 텍스트 구간 추가 yield TextMessageEndEvent(message_id=message_id) # 이 Assistant 메시지가 끝났음을 UI에 알림 yield RunFinishedEvent(thread_id=input_data.thread_id, run_id=input_data.run_id, result={ "ok" : True }) # 이번 Run의 유일한 성공 종료 상태 전송 except Exception: # 사용자에게 직접 노출되어서는 안 되는 내부 예외 포착 yield RunErrorEvent(message= "실행 실패, Run ID로 로그를 조회하세요" , code= "MOCK_RUN_FAILED" ) # 안전하고 안정적인 오류 코드와 안내 반환

도구 결과의 content가 JSON 문자열을 사용하는 것은 표준 TOOL_CALL_RESULT 계약을 유지하기 위해서다. 실제 도구 실행 결과는 먼저 비식별화와 크기 제한을 거친 뒤 UI 이벤트로 들어간다.

14.3.1 먼저 Mock만 포함한 AG-UI 라우트 생성

이때는 아직 실제 모델과 Agent Loop를 임포트하지 않는다. 아래 코드를 src/devmind_server/api/ag_ui.py 에 작성하고, /api/agent/mock 만 개방하여 AG-UI 타입, 이벤트 인코딩, SSE 응답 경로를 먼저 검증한다.

from collections.abc import AsyncIterator # 통합 이벤트 스트림 함수가 받는 비동기 이터레이터 설명 from fastapi import APIRouter # FastAPI 라우터 임포트 from fastapi import Request # Accept 요청 헤더를 읽기 위한 요청 객체 임포트 from starlette.responses import StreamingResponse # 비동기 생성기를 지원하는 스트리밍 응답 임포트 from ag_ui.core import BaseEvent # 모든 AG-UI 이벤트가 공통으로 상속하는 기반 타입 임포트 from ag_ui.core import RunAgentInput # 표준 AG-UI 요청 본문 임포트 및 검증 from ag_ui.encoder import EventEncoder # SSE 형식을 직접 작성하지 않도록 공식 이벤트 인코더 임포트 from devmind_server.agent.ag_ui_demo import run_mock # 이전 절에서 완성한 결정적 Mock 임포트 router = APIRouter(prefix= "/api" , tags=[ "agent" ]) # 통일된 /api 접두사를 가진 Agent 라우트 생성 def stream_response ( events: AsyncIterator[BaseEvent], request: Request ) -> StreamingResponse: # 표준 이벤트 스트림을 HTTP 응답으로 인코딩 encoder = EventEncoder(accept=request.headers.get( "accept" )) # 클라이언트 Accept 헤더에 따라 공식 인코더 생성 async def event_stream (): # StreamingResponse가 소비하는 비동기 생성기 정의 async for event in events: # Mock이 생성한 표준 AG-UI 이벤트를 하나씩 읽기 yield encoder.encode(event) # 공식 인코더를 사용해 합법적인 SSE 데이터 프레임 생성 return StreamingResponse( # 이벤트를 지속적으로 출력하는 HTTP 응답 반환 event_stream(), # 비동기 이벤트 생성기를 Starlette에 전달 media_type=encoder.get_content_type(), # 인코더가 선언한 AG-UI 콘텐츠 타입 사용 headers={ # 캐시 및 리버스 프록시 관련 응답 헤더 설정 "Cache-Control" : "no-store" , # 중간 계층의 Agent 이벤트 캐시 또는 재생 금지 "X-Accel-Buffering" : "no" , # Nginx가 스트리밍 응답을 버퍼링하지 않도록 알림 }, # 응답 헤더 구성 완료 ) # StreamingResponse 생성 완료 @router.post( "/agent/mock" ) # 1단계에서 유일하게 개방한 Mock 엔드포인트 등록 async def run_mock_agent ( input_data: RunAgentInput, request: Request ) -> StreamingResponse: # 표준 입력을 받아 Mock 이벤트 반환 return stream_response(run_mock(input_data), request) # 공식 인코더로 결정적인 AG-UI SSE 출력

14.3.2 main.py 에 Mock 라우트와 개발 기간 CORS 등록

이전 편에서 이미 src/devmind_server/main.py 를 생성했으니 두 번째 진입점을 새로 만들지 않는다. 기존 헬스 체크를 유지한 채 AG-UI Router와 개발 기간 CORS를 추가한다:

from fastapi import FastAPI # FastAPI 애플리케이션 타입 임포트 from fastapi.middleware.cors import CORSMiddleware # 개발 기간용 CORS 미들웨어 임포트 from devmind_server.api.ag_ui import router as ag_ui_router # 방금 생성한 AG-UI Mock 라우터 임포트 from devmind_server.api.health import router as health_router # 이전 글에서 이미 있던 헬스 체크 라우터 임포트

app = FastAPI(title="DevMind Agent Server") # 유일한 FastAPI 애플리케이션 인스턴스 생성

app.add_middleware( # Electron 개발 기간 Renderer를 위한 통제된 CORS 구성 CORSMiddleware, # FastAPI의 CORS 미들웨어 사용 지정 allow_origins=["http://localhost:5173"], # 실제 사용하는 Vite 개발 주소만 허용 allow_credentials=True, # 이후 보호된 인증 정보를 함께 전달할 수 있도록 허용 allow_methods=["GET", "POST", "OPTIONS"], # 현재 필요한 HTTP 메서드만 개방 allow_headers=["Authorization", "Content-Type", "Accept"], # 현재 필요한 요청 헤더만 허용 ) # CORS 미들웨어 등록 완료

app.include_router(health_router) # 이전 글의 헬스 체크 인터페이스 유지 app.include_router(ag_ui_router) # 현재 /api/agent/mock만 포함하는 AG-UI Router 등록

기존 main.py에 다른 Router가 더 있다면 계속 유지해야 하며, 새로 추가된 import, 미들웨어, include_router만 병합하고 파일 전체를 덮어쓰지 않는다.

14.3.3 Server 기동 및 curl로 Mock 전체 흐름 검증

이제는 Mock 단계에 머물러 있고, 실제 모델 어댑터는 작성하지 않는다. 터미널 A를 열고 프로젝트 가상 환경에서 FastAPI를 기동한다:

저장소 루트 디렉터리에서 Python 백엔드 프로젝트로 이동 cd apps/devmind-server # 프로젝트 .venv로 FastAPI를 단일 프로세스 기동(프로덕션 배포에는 uvicorn 권장) uv run uvicorn devmind_server.main:app --host 127.0.0.1 --port 8000 # 코드를 자주 수정 → fastapi dev 사용, 저장하면 자동 반영되어 수동 재시작 불필요(개발 모드, 권장) uv run fastapi dev src/devmind_server/main.py --host 127.0.0.1 --port 8000

다시 터미널 B를 열고, 파라미터 배열로 curl 요청을 구성한다. 이렇게 하면 파라미터를 줄 단위로 설명할 수 있고, 줄 이음 백슬래시 뒤에 주석을 달아 명령이 깨지는 일도 없다:

curl -N -v 'http://127.0.0.1:8000/api/agent/mock' \ -H 'Content-Type: application/json' \ -H 'Accept: text/event-stream' \ --data '{"threadId":"thread-demo","runId":"run-demo","state":{},"messages":[{"id":"message-user","role":"user","content":"이 요구사항을 분석해줘"}],"tools":[],"context":[],"forwardedProps":{}}'

터미널에서는 RUN_STARTED, TEXT_MESSAGE_START / CONTENT / END, TOOL_CALL_START / ARGS / END / RESULT, CUSTOM, RUN_FINISHED를 안정적으로 확인할 수 있어야 한다. 여기까지 오면 Mock의 "생성기 작성—라우트 생성—애플리케이션 등록—서비스 기동—curl 검증"이 완전히 하나의 순환을 이룬다. 확인에 성공한 뒤에 실제 모델 단계로 넘어간다.

14.4 실제 스트리밍 모델과 Agent Loop 연결

이전 글에서 이미 ModelGateway + ToolRegistry + AgentLoop를 완성했다: Loop는 메시지 히스토리를 유지하고, 모델을 호출하고, Tool Call을 판단하고, 도구를 실행하고, 정지 조건을 제어하며, 마지막으로 run()을 통해 RunResult를 반환한다. 이 핵심 로직은 이번 글에서도 계속 재사용하며, Agent Loop를 새로 한 벌 구현하지 않는다.

하지만 기존 run()은 전체 실행이 끝난 뒤에야 결과를 반환할 수 있어, 실행 과정에서 UI에 "모델이 방금 한 단락의 텍스트를 생성했다", "이번 라운드에 어떤 도구를 호출하려는지", "도구 실행이 어느 단계까지 왔는지"를 지속적으로 알려줄 수 없다. 따라서 여기서는 이전 글의 Loop를 포기하는 것이 아니라 그 출력 방식을 업그레이드한다: 모델—도구 루프를 astream()에 집중시켜 프로토콜에 종속되지 않는 Runtime 이벤트를 단계적으로 생성하고, 기존 run()은 계속 유지하여 CLI와 일반 호출의 호환 진입점으로 삼는다. 외부에서는 다시 AgUiAdapter가 이 Runtime 이벤트를 AG-UI 표준 이벤트로 변환한다. 이렇게 하면 이전 글의 핵심 로직을 재사용하면서도, 이후 LangGraph Runtime으로 교체할 수 있는 안정적인 경계를 남길 수 있다.

React Renderer │ AG-UI ▼ FastAPI Router ▼ AgUiAdapter │ RunCommand / RuntimeEvent ▼ AgentRuntime └── 현재: AgentLoop 이후: LangGraphRuntime

여기의 Runtime 이벤트는 Server 내부에서 실제로 발생한 모델 라운드, 텍스트 증분, 도구 호출, 종료 상태만을 설명하며, 두 번째 프런트엔드 프로토콜이 아니고 Renderer에 직접 전송되지도 않는다. 범용 UI 상호작용은 여전히 통일적으로 AG-UI를 사용한다.

14.4.1 Runtime 계약과 Adapter 디렉터리 생성

먼저 이전 글에서 이미 완성한 Python Server로 들어가, 이번 절에서 새로 추가하는 파일을 생성한다. 각 파일에는 이후 소절에서 곧바로 코드를 작성하므로, 미리 최종 내용을 추측할 필요는 없다.

cd apps/devmind-server # 저장소 루트 디렉터리에서 기존 Python Agent Server로 이동 mkdir -p src/devmind_server/agent/adapters # 프로토콜 어댑터 디렉터리 생성, AG-UI와 Agent Runtime 격리 touch src/devmind_server/agent/adapters/__init__.py # adapters를 임포트 가능한 Python 패키지로 표시 touch src/devmind_server/agent/events.py # Server 내부 Runtime 이벤트 정의 파일 생성 touch src/devmind_server/agent/runtime.py # RunCommand와 AgentRuntime 인터페이스 파일 생성 touch src/devmind_server/agent/factory.py # 모델, 도구, Runtime의 통합 조립 진입점 생성 touch src/devmind_server/agent/adapters/ag_ui.py # Runtime에서 AG-UI로의 변환 계층 생성

완료 후, 이번 절에서 새로 추가된 구조는 다음과 같다:

src/devmind_server/agent/ ├── events.py ├── runtime.py ├── factory.py └── adapters/ ├── __init__.py └── ag_ui.py

14.4.2 프로토콜에 종속되지 않는 Runtime 이벤트 정의

다음 코드를 src/devmind_server/agent/events.py에 작성한다. 이 타입들은 Agent Runtime 내부 사실만을 표현하며, 어떤 AG-UI 타입도 임포트하지 않는다.

from dataclasses import dataclass # 불변 데이터 클래스로 구조가 명확한 Runtime 이벤트를 정의 from typing import Any # 모델이 이미 집계를 완료한 도구 파라미터를 표현 from langchain_core.messages import ToolMessage # 도구 레지스트리가 반환하는 표준 도구 메시지를 재사용

@dataclass( frozen= True ) # 실행이 시작된 후 이벤트 내용은 다운스트림에서 수정할 수 없다 class RuntimeStarted : # 하나의 Agent Run이 실행 단계에 진입했음을 나타낸다 thread_id: str # 소속 세션 ID를 저장 run_id: str # 이번 실행 ID를 저장

@dataclass( frozen= True ) # 매 라운드 모델 호출은 독립적인 Assistant 메시지를 가진다 class AssistantMessageStarted : # 새로운 한 라운드의 Assistant 메시지가 시작됨을 나타낸다 message_id: str # 이번 라운드 Assistant 메시지 ID를 저장 round_index: int # 현재 몇 번째 모델 호출 라운드인지 저장

@dataclass( frozen= True ) # 텍스트 델타는 UI가 실제로 필요로 하는 문자열만 저장한다 class AssistantTextDelta : # 모델이 방금 실제 텍스트 한 조각을 생성했음을 나타낸다 message_id: str # 현재 Assistant 메시지를 연결 delta: str # 모델 원본 텍스트 델타를 저장하며, 고정 길이로 자르지 않는다

@dataclass( frozen= True ) # Tool Call은 모델 Chunk 집계가 완료된 후에 생성되어야 한다 class ToolCallReady : # 도구 이름과 파라미터가 이미 완전하여 안전하게 표시하고 실행할 수 있음을 나타낸다 message_id: str # 해당 Tool Call을 생성한 Assistant 메시지를 연결 tool_call_id: str # 모델이 반환한 도구 호출 ID를 저장 tool_name: str # 모델이 선택한 도구 이름을 저장 tool_args: dict [ str , Any ] # 이미 완전히 파싱된 도구 파라미터를 저장

@dataclass( frozen= True ) # 매 라운드 모델 메시지는 반드시 명확히 끝나야 한다 class AssistantMessageFinished : # 이번 라운드 AIMessage의 텍스트와 Tool Call이 모두 완전함을 나타낸다 message_id: str # 닫아야 할 Assistant 메시지를 연결

@dataclass( frozen= True ) # 도구가 실제로 실행을 시작할 때 독립적인 내부 사실이 생성된다 class ToolExecutionStarted : # Runtime이 모든 사전 검사를 통과하여 도구 호출을 시작했음을 나타낸다 tool_call_id: str # 모델이 생성한 원본 Tool Call을 연결 tool_name: str # 현재 실제로 실행을 시작한 도구 이름을 저장

@dataclass( frozen= True ) # 도구 결과가 다음 라운드 모델 입력이 되기 전에 내부 이벤트를 먼저 형성한다 class ToolExecutionFinished : # 한 번의 도구 실행 시도가 끝났으며, 비즈니스 결과가 반드시 성공했음을 의미하지는 않는다 result_message_id: str # 도구 결과 메시지를 위해 독립적인 ID를 저장 tool_message: ToolMessage # tool_call_id와 통일된 성공 또는 실패 결과를 담은 표준 도구 메시지를 저장

@dataclass( frozen= True ) # 모든 통제된 종료 경로는 유일한 종료 상태 이벤트로 통일되어 수렴한다 class RuntimeStopped : # 이번 Agent Runtime이 정지되었음을 나타낸다 thread_id: str # 소속 세션 ID를 저장 run_id: str # 이번 실행 ID를 저장 stop_reason: str # completed, cancelled, timeout 등 안정적인 사유를 저장 answer: str | None # 정상 완료 시 최종 답변을 저장하며, 실패 시 비어 있을 수 있다 model_rounds: int # 실제로 완료된 모델 라운드 수를 저장 tool_calls: int # 실제로 실행된 도구 호출 수를 저장

RuntimeEvent = ( # 정상 실행 타임라인 순서대로 AgentRuntime이 출력을 허용하는 이벤트 유니온 타입을 나열 RuntimeStarted # 실행 시작 이벤트 | AssistantMessageStarted # Assistant 메시지 시작 이벤트 | AssistantTextDelta # 모델 텍스트 델타 이벤트 | ToolCallReady # 완전한 도구 호출 이벤트 | AssistantMessageFinished # Assistant 메시지 종료 이벤트 | ToolExecutionStarted # 도구 실행 시작 이벤트 | ToolExecutionFinished # 도구 실행 종료 이벤트 | RuntimeStopped # 실행의 유일한 종료 상태 이벤트 ) # RuntimeEvent 타입 정의 완료

여기서는 AG-UI의 전체 이벤트 모델을 복사하지 않았다. 예를 들어 SSE 인코딩, RUN_FINISHED, TOOL_CALL_ARGS는 모두 외부 Adapter에 속한다. 내부에는 손으로 작성한 Loop가 실제로 생성하는 실행 사실만 남긴다.

이벤트 타임라인: Runtime은 Adapter에 어떤 내부 이벤트를 생성하는가?

Runtime 이벤트는 Server 내부에서 이미 발생한 사실을 기술하며, 먼저 AgUiAdapter에 전달된 뒤 Renderer가 소비할 수 있는 AG-UI 표준 이벤트로 변환된다. 이것은 두 번째 프런트엔드 프로토콜이 아니며, UI에 그대로 전송되지도 않는다.

한 라운드의 Assistant 메시지는 먼저 모델이 텍스트와 Tool Call을 생성한다. Tool Call이 완전해지면 이번 라운드 AIMessage가 끝난다. Runtime이 취소와 수량 상한 검사를 통과한 후에야 비로소 도구 실행을 시작한다. 도구 결과는 메시지 히스토리에 들어가고, 모델은 이어서 다음 라운드 Assistant 메시지를 시작한다.

그림에 표시된 것은 성공 주경로다. 모델 타임아웃, 모델 실패, 사용자 취소 또는 안전 상한 도달 시에는 현재 단계에서 유일한 RuntimeStopped로 진입한다. 흐름을 맞추기 위해 실제로 발생하지 않은 도구 이벤트를 생성하지는 않는다.

RuntimeStarted AssistantMessageStarted AssistantTextDelta * ToolCallReady * AssistantMessageFinished ToolExecutionStarted ToolExecutionFinished * AssistantMessageStarted AssistantTextDelta * AssistantMessageFinished RuntimeStopped

AssistantMessageFinished는 이번 라운드 모델이 반환한 AIMessage가 텍스트와 Tool Call을 포함해 이미 완전함을 의미하며, 도구 실행 완료를 의미하지는 않는다. ToolExecutionFinished는 도구 실행 시도가 끝났음을 의미하며, 결과는 ok=true일 수도 있고 이전 글의 ToolRegistry가 통일적으로 래핑한 ok=false일 수도 있다. 오직 RuntimeStopped만이 전체 Run이 최종적으로 종료되었음을 나타낸다.

RuntimeEvent AG-UI 이벤트 의미 RuntimeStarted RUN_STARTED 하나의 Run이 시작됨. AssistantMessageStarted TEXT_MESSAGE_START 이번 라운드 Assistant 메시지 시작. AssistantTextDelta TEXT_MESSAGE_CONTENT 실제 모델 텍스트 델타를 지속적으로 전송. ToolCallReady TOOL_CALL_START → ARGS → END 도구 이름과 파라미터가 이미 완전하지만 도구 결과는 아직 생성되지 않음. AssistantMessageFinished TEXT_MESSAGE_END 이번 라운드 모델 메시지가 이미 완전함. ToolExecutionStarted 현재 별도로 매핑되지 않음 Server가 사전 검사를 통과하여 실제로 도구 실행을 시작함. ToolExecutionFinished TOOL_CALL_RESULT 도구가 통일된 성공 또는 실패 결과를 생성함. RuntimeStopped RUN_FINISHED / RUN_ERROR 전체 Run 종료.

현재 모델은 parallel_tool_calls=False로 설정되어 있어 도구가 순서대로 실행된다. 그림에서 여전히 ToolCallReady(1..N)를 사용하는 것은 프로토콜과 Runtime이 여러 개의 Tool Call을 연관 지을 수 있음을 표현하기 위함이다. 만약 이후 병렬 실행을 허용한다면, 반드시 계속 tool_call_id에 의존해 각각의 시작, 종료, 결과를 구분해야 한다.

각 id는 어떻게 생성되는가? 역할은 무엇인가?

왜 run_id가 여전히 필요한가?

순수 채팅 · 두 계층이면 충분 1회 질문 = 1회 모델 호출 = 1개 메시지 thread + message가 일대일로 대응되며 중간 계층이 필요 없다

Agent · 반드시 한 계층 더 필요 run 1회 질문 = 1회 run = N 라운드 호출 + M회 도구 run_id가 없으면 내부 메시지를 그룹화할 수 없고, 위치를 특정할 수 없고, 통계를 낼 수 없다

run_id는 도대체 무엇을 하는가(세 가지 실제 용도)

① 집계 렌더링: 한 번의 run에 대한 start → 각 라운드 메시지 → 도구 → stopped의 모든 이벤트를 run_id로 그룹화해야 프런트엔드가 이를 흩어진 메시지 더미가 아니라 하나의 완전한 "사고-실행-답변" 과정으로 렌더링할 수 있다 ② 생명주기 제어: 취소 / 타임아웃 / 재시도는 모두 run에 작용한다 — 사용자가 "중지"를 클릭할 때 stop_reason=cancelled는 전체 세션이 아니라 이번 run을 중지하는 것을 가리킨다 ③ 구분과 통계: 동일 세션 내의 여러 번 질문, 재생성, 동시 run은 run_id로 구분한다. RuntimeStopped 안의 model_rounds, tool_calls도 run을 통계 기준으로 삼는다

14.4.3 RunCommand와 AgentRuntime 인터페이스 정의

다음 코드를 src/devmind_server/agent/runtime.py 에 작성한다. Router는 더 이상 RunAgentInput을 Loop에 직접 전달하지 않고, 먼저 Adapter가 내부 명령으로 변환한다.

import asyncio # 현재 Run 전용 취소 신호를 제공 from collections.abc import AsyncIterator # 비동기 Runtime 이벤트 스트림을 설명 from dataclasses import dataclass, field # 불변 실행 명령을 정의하고 기본 취소 신호를 생성 from typing import Protocol # 구조적 인터페이스로 수기 Loop와 향후 LangGraph Runtime을 격리 from devmind_server.agent.events import RuntimeEvent # 프로토콜과 무관한 Runtime 이벤트 유니언 타입을 임포트

@dataclass( frozen= True ) # 명령은 생성 후 실행 도중 조용히 수정될 수 없음 class RunCommand : # Agent Run 한 번을 시작하는 데 필요한 최소 내부 입력을 정의 thread_id: str # 소속 세션 ID를 저장 run_id: str # AG-UI 요청이 제공한 Run ID를 저장 user_input: str # 이번에 Agent에 제출하는 사용자 텍스트를 저장 cancel_event: asyncio.Event = field(default_factory=asyncio.Event) # 현재 Run을 위한 독립 취소 신호를 생성

class AgentRuntime ( Protocol ): # Router와 Adapter가 의존하는 안정적인 Runtime 인터페이스를 정의 def astream ( self, command: RunCommand ) -> AsyncIterator[RuntimeEvent]: # 비동기 이벤트 생성기 인터페이스를 선언 ... # 구체 구현은 현재 AgentLoop가 제공하며, 추후 LangGraphRuntime으로 교체 가능

AgentRuntime은 "계약"이고, AgentLoop가 바로 "직원"이다. 이 코드는 인터페이스만 선언하며( ... 는 자리 표시자), 실제로 실행 가능한 구현은 AgentLoop에서 온다.

덕 타이핑(duck typing): 영어 속담 "걸음걸이가 오리 같고 울음소리가 오리 같다면, 그것은 오리다"에서 유래했다——Python은 당신이 "주장"하는 타입을 보지 않고, 호출에 필요한 메서드가 있는지만 본다. 그래서 AgentLoop는 class AgentLoop(AgentRuntime) 를 쓸 필요가 없고, 시그니처가 일치하는 astream() 메서드만 있으면 Python은 이를 합법적인 Runtime으로 취급한다. 이는 Python 곳곳에 존재한다: len() 은 __len__ 만 인정하고, for 루프는 반복 프로토콜만 인정하며, 파일 작업은 read()/write() 만 인정하고, 테스트에서 Fake 대역으로 실제 객체를 대체하는 것도 같은 사고방식이다.

Protocol이란: "덕 타이핑"을 구두 약속에서 서면 계약으로 바꾸는 것이다——인터페이스가 어떤 모습이어야 하는지(반드시 astream() 이 있어야 함)만 선언하고, 어떤 클래스든 구조가 일치하면 충족한 것으로 보며 이를 상속할 필요가 없다. 장점은 IDE와 타입 검사기가 코드를 작성할 때 바로 "객체에 메서드가 빠졌다"를 발견할 수 있어 런타임에서야 오류가 나지 않는다는 것이다.

연관은 factory가 객체를 생성하는 그 순간에 일어난다: create_agent_runtime() -> AgentRuntime 내부에서 return AgentLoop(gateway, registry) ——반환값은 계약으로 표기하고, 실제로 넣는 것은 AgentLoop 인스턴스이며, Router가 다시 이를 AgUiAdapter(runtime: AgentRuntime) 에 전달하고, 이후 runtime.astream() 을 호출할 때 Python은 실제 객체를 기준으로 메서드를 찾아 실행한다. 타입 표기는 IDE와 타입 검사기에게만 보여주며 런타임에서는 무시된다.

왜 굳이 AgentRuntime이어야 하고, AgentLoop에 직접 의존하면 안 되나? "사람을 바꿔도 자리는 바꾸지 않기" 위해서다. 만약 Router, Adapter가 AgentLoop라는 구체 클래스에 직접 의존하면, 나중에 LangGraphRuntime으로 전환할 때 모든 호출 지점을 함께 고쳐야 한다; AgentRuntime 인터페이스에 의존하면 새 구현에 시그니처가 일치하는 astream() 만 있으면 되고, factory에서 return 한 줄만 바꾸면 전환이 완료되어 Router, Adapter와 이벤트 타입은 한 줄도 건드릴 필요가 없으며 테스트도 대응하는 Fake 구현으로 바꾸기만 하면 된다.

14.4.4 ModelGateway에 진짜 astream() 추가

앞 편에서 만든 src/devmind_server/agent/model_gateway.py 를 열어라. 원래의 invoke() 를 유지하고, astream() 을 새로 추가한다. Gateway는 모델에 접근해 Chunk를 반환하는 것만 담당하고, AG-UI 이벤트는 생성하지 않는다.

from collections.abc import AsyncIterator # 비동기 모델 Chunk 스트림의 반환 타입을 설명 from langchain_core.language_models.chat_models import BaseChatModel # 채팅 모델 통합 인터페이스를 임포트 from langchain_core.messages import AIMessage # 비스트리밍 호출이 반환하는 완전한 Assistant 메시지를 임포트 from langchain_core.messages import AIMessageChunk # 스트리밍 호출이 구간별로 반환하는 Assistant 메시지 블록을 임포트 from langchain_core.messages import BaseMessage # 모델 입력 메시지의 공통 기반 클래스를 임포트 from langchain_core.tools import BaseTool # Tool 통합 기반 클래스를 임포트 from langchain_openai import ChatOpenAI # OpenAI 프로토콜 호환 채팅 모델 구현을 임포트 from devmind_server.core.config import get_settings # 앞 편에서 완성한 중앙 집중 설정 읽기 함수를 임포트

def create_chat_model_from_settings () -> BaseChatModel: # 환경 설정에 따라 구체 채팅 모델을 생성 settings = get_settings() # 모델명, 키, API 주소를 읽고 검증 return ChatOpenAI( # Tool Calling과 스트리밍 출력을 지원하는 모델 인스턴스를 구성 model=settings.model_name, # 설정의 모델 이름을 사용 api_key=settings.api_key.get_secret_value(), # SDK 클라이언트 생성 시에만 실제 키를 꺼냄 base_url=settings.base_url or None , # 사용자 정의 주소가 있으면 사용하고, 없으면 SDK 기본 주소를 사용 ) # 실제 채팅 모델 생성 완료

class ModelGateway : # 구체 모델 SDK를 격리하고 AgentLoop에 안정적인 인터페이스를 제공 def __init__ ( self, model: BaseChatModel, tools: list [BaseTool] ): # 모델 인스턴스와 노출을 허용하는 도구 목록을 받음 self._model = model.bind_tools( # 도구 Schema를 모델 인스턴스에 바인딩 tools, # 레지스트리가 제공하는 도구만 모델에 공개 parallel_tool_calls= False , # 현재 단계에서는 병렬 도구 호출을 금지해 병합 복잡도를 낮춤 ) # 도구가 있는 모델 생성 완료

async def invoke ( self, messages: list [BaseMessage] ) -> AIMessage: # 앞 편에 이미 있던 비스트리밍 호환 진입점을 유지 response = await self._model.ainvoke(messages) # 모델이 완전한 Assistant 메시지를 생성할 때까지 대기 if not isinstance (response, AIMessage): # 모델 반환 타입을 방어적으로 검사 raise TypeError( "모델이 AIMessage를 반환하지 않았습니다" ) # 타입 오류 시 즉시 이번 호출을 종료 return response # 완전한 메시지를 여전히 비스트리밍 진입점을 사용하는 호출자에게 전달

async def astream ( self, messages: list [BaseMessage] ) -> AsyncIterator[AIMessageChunk]: # 실제 비동기 스트리밍 진입점을 새로 추가 async for chunk in self._model.astream(messages): # 모델 서비스가 반환한 원시 Chunk를 계속 읽음 if not isinstance (chunk, AIMessageChunk): # 각 스트리밍 결과 타입을 방어적으로 검사 raise TypeError( "모델이 AIMessageChunk를 반환하지 않았습니다" ) # 안전하게 집계할 수 없는 반환값을 거부 yield chunk # 완전한 답변을 기다리지 않고 즉시 현재 Chunk를 AgentLoop에 전달

invoke() 와 astream() 은 이미 Tool Schema가 바인딩된 동일한 모델 인스턴스를 공유한다. 실제 HTTP 경로는 astream() 을 사용하고, 기존 CLI나 다른 일반 호출은 여전히 invoke() 를 계속 사용할 수 있다.

앞 편의 Fake Model도 동기화해 업데이트한다. run() 이 이제 유일한 astream() 메인 루프도 소비하므로, 앞 편에서 invoke() 만 구현한 테스트 대역도 스트리밍 인터페이스로 바꿔야 한다. tests/fakes.py 를 열고 FakeModelGateway 를 아래의 완전한 버전으로 업데이트한다:

from collections import deque # 테스트 사전 설정 응답을 순서대로 저장하고 꺼냄 from langchain_core.messages import AIMessage # 앞 편 테스트에서 전달한 완전한 모델 응답을 설명 from langchain_core.messages import AIMessageChunk # 완전한 Fake 응답을 하나의 결정적 스트리밍 Chunk로 변환

class FakeModelGateway : # 결정적 응답으로 실제 모델 네트워크 호출을 대체 def __init__ ( self, responses: list [AIMessage | Exception] ): # 모델 메시지 또는 예외로 구성된 사전 설정 시퀀스를 받음 self.responses = deque(responses) # 왼쪽부터 순서대로 꺼낼 수 있는 큐로 변환 self.received_messages = [] # 매 라운드 받은 완전한 메시지를 테스트 단언을 위해 저장

async def astream ( self, messages ): # AgentLoop가 현재 유일하게 사용하는 스트리밍 게이트웨이 인터페이스를 구현 self.received_messages.append( list (messages)) # 이번 라운드 메시지를 복사해 이후 추가가 단언에 영향을 주지 않게 함 response = self.responses.popleft() # 현재 라운드의 사전 설정 동작을 꺼냄 if isinstance (response, Exception): # 사전 설정이 예외일 때 모델 호출 실패를 시뮬레이션 raise response # 예외를 AgentLoop에 넘겨 안정적인 중지 사유로 변환하게 함 yield AIMessageChunk( # 하나의 Chunk로 현재 결정적 완전 응답을 표현 content=response.content, # 앞 편 테스트의 텍스트 내용을 유지 tool_calls=response.tool_calls, # 앞 편 테스트의 표준 Tool Call을 유지 ) # Fake Chunk 생성 완료

실제 모델은 보통 여러 Chunk를 생성하지만, Fake가 매 라운드 하나의 Chunk만 생성하는 것은 기존 테스트를 결정적으로 유지하기 위해서다. test_agent_loop.py 에서 "먼저 도구를 호출하고, 그다음 텍스트를 반환" 같은 테스트 데이터는 다시 작성할 필요가 없다.

14.4.5 AgentLoop를 generator 패스스루로 변경하기

이제 src/devmind_server/agent/loop.py 를 열고, 아래의 전체 버전으로 이전 글의 구현을 교체하세요. 핵심 변화는 AG-UI를 Loop에 밀어 넣는 것이 아니라, astream()을 유일한 모델—도구 루프로 만드는 것입니다. 호환되는 run()은 이 이벤트 스트림을 소비하고 RunResult 를 반환하기만 합니다.

import asyncio # 모델 타임아웃 제어와 취소 신호 제공 from collections.abc import AsyncIterator # Agent Runtime 비동기 이벤트 스트림 설명 from dataclasses import dataclass # 설정, 컨텍스트, 최종 결과 정의 from enum import StrEnum # 안정적인 문자열 값을 가진 중지 사유 정의 from uuid import uuid4 # 호환 호출과 매 라운드 Assistant 메시지를 위한 고유 ID 생성 from langchain_core.messages import AIMessage # 집계된 Assistant 메시지 타입 검증 from langchain_core.messages import AIMessageChunk # 모델 스트림의 텍스트와 Tool Call Chunk 집계 from langchain_core.messages import BaseMessage # 실행 컨텍스트의 표준 메시지 목록 설명 from langchain_core.messages import HumanMessage # 사용자 입력 메시지 생성 from langchain_core.messages import SystemMessage # 시스템 제약 메시지 생성 from langchain_core.messages.utils import message_chunk_to_message # 완전한 Chunk를 다시 AIMessage로 변환 from devmind_server.agent.events import AssistantMessageFinished # 메시지 종료 이벤트 임포트 from devmind_server.agent.events import AssistantMessageStarted # 메시지 시작 이벤트 임포트 from devmind_server.agent.events import AssistantTextDelta # 실제 텍스트 증분 이벤트 임포트 from devmind_server.agent.events import RuntimeEvent # Runtime 이벤트 유니온 타입 임포트 from devmind_server.agent.events import RuntimeStarted # 실행 시작 이벤트 임포트 from devmind_server.agent.events import RuntimeStopped # 실행 종료 상태 이벤트 임포트 from devmind_server.agent.events import ToolCallReady # 완전한 Tool Call 이벤트 임포트 from devmind_server.agent.events import ToolExecutionStarted # 도구 실행 시작 이벤트 임포트 from devmind_server.agent.events import ToolExecutionFinished # 도구 완료 이벤트 임포트 from devmind_server.agent.model_gateway import ModelGateway # 실제 모델 게이트웨이 임포트 from devmind_server.agent.runtime import RunCommand # 프로토콜 독립적인 실행 명령 임포트 from devmind_server.agent.tool_registry import ToolRegistry # 도구 통합 실행 진입점 임포트

class StopReason(StrEnum): # 한 번의 실행에서 허용되는 중지 사유 열거 COMPLETED = "completed" # 모델이 더 이상 도구를 요청하지 않고 최종 답변을 제시 CANCELLED = "cancelled" # 사용자 또는 상위 시스템이 취소를 요청 MAX_MODEL_ROUNDS = "max_model_rounds" # 모델 라운드 수가 안전 상한에 도달 MAX_TOOL_CALLS = "max_tool_calls" # 도구 호출 수가 안전 상한에 도달 MODEL_TIMEOUT = "model_timeout" # 단일 라운드 모델 호출 타임아웃 MODEL_FAILED = "model_failed" # 모델 호출 또는 메시지 집계 실패

@dataclass # 초기화 메서드 자동 생성 class LoopConfig: # Agent Loop의 안전 상한 저장 max_model_rounds: int = 8 # 한 번의 Run에서 최대 8라운드 모델 호출 max_tool_calls: int = 16 # 한 번의 Run에서 최대 16회 도구 실행 model_timeout_seconds: float = 60 # 단일 라운드 모델 호출은 최대 60초 대기

@dataclass # 한 번의 실행에서 변하는 상태를 한곳에서 관리 class RunContext: # 루프 과정에서 계속 변하는 데이터 저장 run_id: str # 현재 Run ID messages: list[BaseMessage] # 모델에 보낼 전체 메시지 이력 cancel_event: asyncio.Event # 현재 Run의 취소 신호 model_rounds: int = 0 # 완료된 모델 호출 라운드 수 tool_calls: int = 0 # 완료된 도구 호출 수

@dataclass # 앞선 글에서 이미 CLI와 테스트에 공개한 결과 타입 유지 class RunResult: # 한 번의 Run 최종 구조화 결과 설명 run_id: str # 이번 Run ID 반환 stop_reason: StopReason # 명확한 중지 사유 반환 answer: str | None # 정상 완료 시 최종 텍스트 반환 model_rounds: int # 실제 모델 라운드 수 반환 tool_calls: int # 실제 도구 호출 수 반환

class AgentLoop: # 현재 단계의 수기 작성 AgentRuntime 구현 def __init__(self, gateway: ModelGateway, registry: ToolRegistry, config: LoopConfig | None = None): # 모델, 도구, 설정 주입 self.gateway = gateway # 통합 모델 호출 진입점 저장 self.registry = registry # 통합 도구 실행 진입점 저장 self.config = config or LoopConfig() # 설정을 넘기지 않으면 기본 안전 상한 사용

async def astream(self, command: RunCommand) -> AsyncIterator[RuntimeEvent]: # Runtime 이벤트 스트림으로 전체 Agent Run 실행 context = RunContext( # 이번 Run에만 속한 컨텍스트 생성 run_id=command.run_id, # Adapter가 AG-UI 입력에서 얻은 Run ID 재사용 cancel_event=command.cancel_event, # 현재 Run 전용 취소 신호 재사용 messages=[ # 모델에 보낼 메시지 이력 초기화 SystemMessage(content=SYSTEM_PROMPT), # 도구 결과로 덮어쓸 수 없는 시스템 제약 삽입 HumanMessage(content=command.user_input), # 사용자의 이번 질문 삽입 ], # 초기 메시지 목록 완성 ) # 실행 컨텍스트 생성 완료 yield RuntimeStarted(thread_id=command.thread_id, run_id=command.run_id) # 먼저 실행 시작 사실 생성 while context.model_rounds < self.config.max_model_rounds: # 모델 라운드 상한 내에서 계속 루프 if context.cancel_event.is_set(): # 매 라운드 모델 호출 전 취소 신호 확인 yield self._stopped(command.thread_id, context, StopReason.CANCELLED) # 유일한 취소 종료 상태 생성 return # 모델 또는 도구 호출 중단 round_index = context.model_rounds + 1 # 곧 시작할 모델 라운드 계산 message_id = str(uuid4()) # 이번 라운드 Assistant 메시지에 독립 ID 생성 yield AssistantMessageStarted(message_id=message_id, round_index=round_index) # 다운스트림에 새 메시지 시작 알림 full_chunk: AIMessageChunk | None = None # 이번 라운드에서 이미 집계된 완전한 모델 Chunk 저장 try: # 타임아웃과 기타 모델 예외를 안정적인 종료 상태로 변환 async with asyncio.timeout(self.config.model_timeout_seconds): # 현재 모델 스트림 최장 시간 제한 async for chunk in self.gateway.astream(context.messages): # 실제 모델 스트림을 조각별로 소비 full_chunk = chunk if full_chunk is None else full_chunk + chunk # 텍스트와 Tool Call Chunk 집계 if isinstance(chunk.content, str) and chunk.content: # 비어 있지 않은 텍스트에만 표시 이벤트 생성 yield AssistantTextDelta(message_id=message_id, delta=chunk.content) # 실제 텍스트 증분 즉시 전달 except TimeoutError: # 현재 모델 라운드 타임아웃 포착 yield AssistantMessageFinished(message_id=message_id) # 이미 시작된 Assistant 메시지 닫기 yield self._stopped(command.thread_id, context, StopReason.MODEL_TIMEOUT) # 모델 타임아웃 종료 상태 생성 return # 현재 Run 종료 except Exception: # 모델 네트워크 오류, 타입 오류, 집계 예외 포착 yield AssistantMessageFinished(message_id=message_id) # 이미 시작된 Assistant 메시지 닫기 yield self._stopped(command.thread_id, context, StopReason.MODEL_FAILED) # 모델 실패 종료 상태 생성 return # 현재 Run 종료 if full_chunk is None: # 정상 모델 스트림은 최소한 하나의 Chunk를 반환해야 함 yield AssistantMessageFinished(message_id=message_id) # 이미 시작된 Assistant 메시지 닫기 yield self._stopped(command.thread_id, context, StopReason.MODEL_FAILED) # 빈 스트림을 모델 실패로 간주 return # 현재 Run 종료 response = message_chunk_to_message(full_chunk) # 모든 Chunk를 완전한 표준 메시지로 변환 if not isinstance(response, AIMessage): # 방어적 검

집계 결과 유형 조회 yield AssistantMessageFinished( message_id=message_id ) # 이미 시작된 Assistant 메시지를 닫음 yield self._stopped( command.thread_id, context, StopReason.MODEL_FAILED, ) # 타입 오류 종료 상태를 생성 return # 현재 Run을 종료

context.model_rounds += 1 # 모델 메시지를 완전히 획득한 뒤 라운드 수를 누적 context.messages.append(response) # 완전한 AIMessage를 메시지 이력에 넣음

for call in response.tool_calls: # 이미 완전히 집계된 Tool Call만 순회 yield ToolCallReady( # 안전하게 표시하고 실행할 수 있는 완전한 도구 호출 이벤트를 생성 message_id=message_id, # 호출을 생성한 Assistant 메시지를 연관시킴 tool_call_id=str(call["id"]), # 모델 도구 호출 ID를 저장 tool_name=str(call["name"]), # 모델이 선택한 도구 이름을 저장 tool_args=call.get("args", {}), # 이미 완전히 파싱된 인자를 저장 ) # ToolCallReady 이벤트를 완료

yield AssistantMessageFinished( message_id=message_id ) # 텍스트와 Tool Call이 모두 완전해진 뒤 이번 라운드 메시지를 닫음

if not response.tool_calls: # Tool Call이 없으면 모델이 이미 최종 답변을 제시했음을 의미 yield self._stopped( # 이번 Run의 유일한 성공 종료 상태를 생성 command.thread_id, # 소속 세션 ID를 전달 context, # 현재 실행 컨텍스트를 전달 StopReason.COMPLETED, # 정상 완료로 표시 answer=str(response.content), # 최종 Assistant 텍스트를 저장 ) # 성공 종료 상태 이벤트를 완료 return # 이벤트 생성기를 정상 종료

for call in response.tool_calls: # 모델이 반환한 순서대로 도구를 차례로 실행 if context.c