참고: 리서치 에이전트의 UI 컴포넌트 관리

본 문서에서는 실전: 예시 프로젝트와 함께 이해하기 섹션에서 사용된 리서치 에이전트에서 GenOS 채팅 상에서 표현되는 UI 컴포넌트(inline citation, chart) 등을 어떻게 관리하고 있는지 설명합니다.

---

UI 컴포넌트가 필요한 Tools

search

검색어, 기간, 사이트 도메인을 지정하여 수행한 웹 검색 결과를 가져오는 도구입니다.

• 툴 호출 결과

- 【id†title†source】: date — snippet
...

• 예시 툴 호출 결과 (query: "python")

- 【0:0†Welcome to Python.org†Python.org】: 2023. 4. 23. — Python is a programming language that lets you work quickly and integrate systems more effectively. ...
- 【0:1†Python - 파이썬†나무위키】: 파이썬은 멀티 패러다임 언어로, 절차적 프로그래밍, 함수형 프로그래밍, 객체 지향 등 다양한 패러다임을 모두 지원하는 언어이다. 속도에서 약간의 희생되는 부분이 ...
- 【0:2†파이썬 - 위키백과, 우리 모두의 백과사전†Wikipedia】: 파이썬은 비영리 파이썬 소프트웨어 재단이 관리하는 개방형, 공동체 기반 개발 모델이 있다. 목차. 1 개요; 2 역사. 2.1 파이썬 2; 2.2 파이썬 3; 2.3 ...

• search 함수의 ID는 turn:idx와 같이 정의됩니다. 이때 turn은 에이전트가 이 도구를 호출한 순서를, idx는 검색 결과의 인덱스를 의미합니다. 가령 위의 예시의 경우에는 가장 처음 사용한 search의 결과임을 알 수 있습니다.

open

웹 페이지 열람과 스크롤을 담당하는 도구입니다.

• 툴 호출 결과

# 【turn:0†title†domain】
**viewing lines [{start} - {end-1}] of {len(lines)}**

```contents
{body}

- 예시 툴 호출 결과 (url: "https://ko.wikipedia.org/wiki/%ED%8C%8C%EC%9D%B4%EC%8D%AC")

```markdown
# 【1:0†파이썬 - 위키백과, 우리 모두의 백과사전†ko.wikipedia.org】
**viewing lines [0 - 99] of 425**

```contents
위키백과, 우리 모두의 백과사전.

[Image 0] 이 문서는 프로그래밍 언어에 관한 것입니다. 그리스 신화의 괴물에 대해서는 【0:1†피톤】 문서를, 미사일에 대해서는 【0:2†파이톤 5 미사일】 문서를 참고하십시오.

파이썬   
Python[Image 1]  
【0:3†패러다임】【0:3†프로그래밍 패러다임】: 【0:4†객체 지향 프로그래밍】, 【0:5†명령형 프로그래밍】, 【0:6†함수형 프로그래밍】, 【0:7†절차적 프로그래밍】, 【0:8†반영】  
【0:9†설계자】【0:10†귀도 반 로섬】  
【0:11†개발자】【0:12†파이썬 소프트웨어 재단】  
발표일 1991년 2월 20일   
최근 버전 3.13.7[ 1 ] [Image 2: 위키데이터에서 편집하기]
...

- ID/링크 매핑 규칙
  - `open`으로 새 페이지를 열면, 해당 페이지 자체가 `【turn:0†title†domain】` ID로 다시 매핑됩니다.
  - 본문 내 하이퍼링크는 페이지 처리 시 `【turn:linkId†text】` 또는 외부 도메인의 경우 `【turn:linkId†text†domain】` 형식으로 재작성되며, 각 `linkId`는 해당 `turn` 내에서 1부터 증가합니다.

### `generate_chart`

차트를 생성하여 UI에 임베드하는 도구입니다(MCP 기반). 성공 시, UI가 차트를 치환 렌더링할 수 있도록 전용 ID를 반환합니다.

- 툴 호출 결과

```markdown
Chart '{data_json['title']}' has been successfully generated. You can display it to the user by using the following ID: `【idx†chart】`

• 동작 및 반환

  • 차트가 생성되면 내부적으로 새로운 차트 ID가 할당됩니다: 【N†chart】 (세션 내 0부터 증가).

  • 어시스턴트 메시지에는 다음과 같은 안내가 포함됩니다: Chart '<title>' has been successfully generated ... ID: 【N†chart】.

  • UI는 본문 내 【N†chart】 토큰을 실제 차트(iframe 등)로 치환하여 렌더링합니다.

---

어떻게 ID를 UI로 바꾸는가?

이전 섹션의 GenOS 워크플로우 연동(Python Step)의 파이썬 스텝에서는 스트리밍된 토큰이 "【"로 시작한다면 GenOS 채팅에 토큰 정보를 보내는 것을 중단하고, "】"이 나왔을 때, 특수 토큰 사이의 ID를 추출해 적절한 값으로 바꾸어서 출력하는 방식으로 동작합니다.

mocking-flowise 프로젝트의 app/api/chat.py에는 다음과 같이 tool_state를 emit합니다.

...(중략)...
while True:
    if client_disconnected.is_set():
        break
    await emit("tool_state", states.tool_state.model_dump())
...(중략)...

이렇게 전달된 tool_state는 다음과 같이 파이썬 스텝에서 tool_state라는 변수로 저장되고, 특수 토큰 사이의 ID를 적절한 값으로 바꾸는 데 사용됩니다.

from main_socketio import sio_server
import aiohttp, json, re
from urllib.parse import urlparse

async def run(data: dict) -> dict:
        ...(중략)...
    def replace_citation_segment(segment: str) -> str:
        # segment: "【...】" 형태. 내부의 turn:id 만 URL로 치환하고 나머지는 제거
        try:
            if not (segment.startswith("【") and segment.endswith("】")):
                return segment
            body = segment[1:-1]
            ids = re.findall(r"(\d+:\d+)", body)
            if not ids:
                return ""
            id_to_url = {}
            if isinstance(tool_state, dict):
                id_to_url = tool_state.get("id_to_url", {}) or {}
            mapped = [id_to_url[idv] for idv in ids if idv in id_to_url]
        ...(중략)...
    async def process_token(ev_text: str):
        ...(중략)...
        try:
            if re.fullmatch(r"【\d+†chart】", segment):
                key = segment[1:-1]
                id_to_iframe = {}
                if isinstance(tool_state, dict):
                    id_to_iframe = tool_state.get("id_to_iframe", {}) or {}
                replaced = id_to_iframe.get(key, "")
            else:
                replaced = replace_citation_segment(segment)
        ...(중략)...
    
    async with aiohttp.ClientSession(
        timeout=aiohttp.ClientTimeout(total=None),
        read_bufsize=BIG, max_line_size=BIG
    ) as session:
        async with session.post(endpoint, json=data, headers={"x-request-from":"internal"}) as resp:
            reasoning = ""
            tool_state = None
            citation_buffer = ""
            inside_citation = False
            async for line in resp.content:
                if not line:
                    continue
                decoded = line.decode("utf-8").strip()
                if decoded.startswith("data:"):
                    decoded = decoded.removeprefix("data:")
                try:
                    payload = json.loads(decoded)
                except json.JSONDecodeError:
                    continue

                event = payload.get("event")
                ev_data = payload.get("data")

                if event == "tool_state":
                    tool_state = ev_data
        ...(중략)...