에이전트 UI 및 상태 동기화 심층 튜토리얼
순수 파이썬만을 사용해 외부 프레임워크에 의존하지 않고 전체 에이전트 UI(Agentic UI) 스택을 밑바닥부터 구축하는 심층 튜토리얼입니다. AG-UI 이벤트 스트림과 A2UI 선언형 레이어를 도입하여 자연어 기반 UI 생성, JSON Patch를 활용한 상태 동기화, 그리고 안전한 인간 개입(Human-in-the-loop) 승인 흐름을 구현하는 과정을 다룹니다.
에디터 추천 | 에이전트 AI 기술 | AI 쇼츠 | 인공지능 애플리케이션 | 소프트웨어 엔지니어링 | 스태프 튜토리얼
이 튜토리얼에서는 핵심 개념을 추상화해 주는 외부 프레임워크에 의존하지 않고, 순수 Python만을 사용하여 전체 에이전트 UI(Agentic UI) 스택을 밑바닥부터 구축합니다. 에이전트의 동작을 실시간으로 관찰할 수 있도록 AG-UI 이벤트 스트림을 구현하고, 실행 가능한 코드 대신 구조화된 JSON으로 인터페이스를 정의할 수 있는 선언형 레이어인 A2UI를 도입합니다.
이어지 과정에서 대형 언어 모델(LLM)이 자연어로부터 완전한 형태의 사용자 인터페이스를 생성하도록 하고, JSON Patch 업데이트를 통해 에이전트와 UI의 상태를 동기화하며, 중요한 작업에 대해 인간의 개입(Human-in-the-loop)을 통한 안전장치를 적용합니다. 또한, 에이전트의 추론 결과가 어떻게 프로토콜을 준수하는 인터랙티브 UI로 변환되는지 엔드투엔드(End-to-End) 관점을 명확하게 이해할 수 있습니다.
[코드] 복사 완료 | 다른 브라우저 사용
import subprocess, sys for pkg in ["openai", "rich", "pydantic"]: subprocess.check_call([sys.executable, "-m", "pip", "install", "-q", pkg])
import os, getpass if os.environ.get("OPENAI_API_KEY"): API_KEY = os.environ["OPENAI_API_KEY"] print("✅ Using OPENAI_API_KEY from environment.") else: try: from google.colab import userdata API_KEY = userdata.get("OPENAI_API_KEY") print("✅ Using OPENAI_API_KEY from Colab Secrets.") except Exception: API_KEY = getpass.getpass("🔑 Enter your OpenAI API key (hidden): ") print("✅ API key received.")
BASE_URL = os.environ.get("OPENAI_BASE_URL", "https://api.openai.com/v1") MODEL = os.environ.get("OPENAI_MODEL", "gpt-4o-mini")
import json, re, time, uuid, copy, textwrap from enum import Enum from dataclasses import dataclass, field, asdict from typing import Any, Optional, Generator from pydantic import BaseModel, Field from openai import OpenAI from rich.console import Console from rich.panel import Panel from rich.table import Table from rich.tree import Tree from rich.text import Text from rich.markdown import Markdown from rich import box
console = Console(width=105) client = OpenAI(api_key=API_KEY, base_url=BASE_URL)
def llm(messages, **kw): try: return client.chat.completions.create(model=MODEL, messages=messages, temperature=0.2, **kw) except Exception as e: console.print(f"[red]LLM error: {e}[/]") return None
def hdr(n, title, sub=""): console.print() console.rule(f"[bold cyan]SECTION {n}", style="cyan") body = f"[bold white]{title}[/]\n[dim]{sub}[/]" if sub else f"[bold white]{title}[/]" console.print(Panel(body, border_style="cyan", padding=(1, 2)))
hdr(1, "AG-UI Protocol — Event System", "실제 AG-UI 프로토콜은 SSE를 통해 스트리밍되는 약 16개의 이벤트 유형을 사용합니다.\n여기서는 모든 핵심 이벤트 유형과 스트리밍 이미터를 순수 Python으로 구현합니다.")
class AGUIEventType(str, Enum): RUN_STARTED = "RUN_STARTED" RUN_FINISHED = "RUN_FINISHED" RUN_ERROR = "RUN_ERROR" TEXT_MESSAGE_START = "TEXT_MESSAGE_START" TEXT_MESSAGE_CONTENT = "TEXT_MESSAGE_CONTENT" TEXT_MESSAGE_END = "TEXT_MESSAGE_END" TOOL_CALL_START = "TOOL_CALL_START" TOOL_CALL_ARGS = "TOOL_CALL_ARGS" TOOL_CALL_RESULT = "TOOL_CALL_RESULT" TOOL_CALL_END = "TOOL_CALL_END" STATE_SNAPSHOT = "STATE_SNAPSHOT" STATE_DELTA = "STATE_DELTA" INTERRUPT = "INTERRUPT" CUSTOM = "CUSTOM" STEP_STARTED = "STEP_STARTED" STEP_FINISHED = "STEP_FINISHED"
@dataclass class AGUIEvent: type: AGUIEventType data: dict = field(default_factory=dict) event_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8]) timestamp: float = field(default_factory=time.time)
def to_sse(self) -> str:
payload = {"type": self.type.value, "id": self.event_id, **self.data}
return f"event: ag-ui\ndata: {json.dumps(payload)}\n\n"
def to_json(self) -> dict:
return {"type": self.type.value, "id": self.event_id, "ts": self.timestamp, **self.data}
class AGUIEventStream: def init(self): self.events: list[AGUIEvent] = [] self.listeners: list = []
def emit(self, event: AGUIEvent):
self.events.append(event)
for listener in self.listeners:
listener(event)
def on(self, callback):
self.listeners.append(callback)
def replay(self) -> list[