에이전트플로우 멀티턴 채팅하기

1. 서론

• 한계 Flowsie 상에서 제공하는 Buffer Window Memory Conversation History Enable Memory 등을 활용하면 기본적으로 멀티턴 기능이 작동합니다. 하지만, 위와 같은 기능을 사용하게 되었을 때 오래된 채팅 내역일 수록 중복되어 기록되어 불필요한 토큰을 소모하게 하고 최종 LLM 출력에도 노이즈로 작용하여 성능을 저하시킵니다. 위와 같은 문제 외에도, 배포된 채팅 애플리케이션에서 업로드한 파일이 Chat History 안에 남지 않아 어느 턴에 무슨 파일을 업로드 했는지에 관한 정보를 알 수 없어 기존 기능에서는 한계가 존재합니다. \

• 해결 따라서 본 심화 워크플로우에서는 Flowise 내부 멀티턴 기능에 의존하지 않고, 자체적인 GenOS 상의 파이썬 워크플로우를 활용해 멀티턴에 사용할 채팅 내역을 커스텀하고 업로드된 파일 내역 또한 기록되는 대화 쌍 내에 포함시키는 작업을 진행하도록 하겠습니다.

2. 전체 구조

• 전체 구조는 크게 이하 두개 워크플로우로 나뉩니다.

  • Python 워크플로우: $genosUploaded 초기화, chat history 기록

  • Flowise 워크플로우: 채팅 전반의 흐름 구조 제어

  자세한 사항은 후술하도록 하겠습니다.

3. 구현 방법

3.1 파이썬 워크플로우 생성

GenOS 메뉴 >> 에이전트 >> 워크플로우 >> 워크플로우 생성 이후 필요한 정보를 기입한 후 생성한 다음 아래와 같은 Python 코드를 작성하여 넣어준 뒤 저장하고 배포합니다.

async def run(chat_data: dict) -> dict:
    import os, re, json
    from datetime import datetime

    # ===== 설정 상수 =====
    BASE_DIR = 'chat_history'                 # 채팅 혹은 업로드 파일을 저장할 기본 디렉터리
    MAX_UPLOAD_SIZE = 10 * 1024 * 1024        # 업로드 파일 최대 허용 크기 (10MB)
    UPLOAD_FIELD = 'genosUploaded'            # 업로드 필드 키 이름

    # ===== 유틸리티 함수들 =====
    def _today() -> str:
        """
        현재 날짜를 'YYYY-MM-DD' 형식의 문자열로 반환
        """
        return datetime.now().strftime('%Y-%m-%d')

    def _ensure_dir(path: str):
        """
        주어진 경로의 디렉터리가 존재하지 않으면 생성
        """
        os.makedirs(path, exist_ok=True)

    def _get_path(session: str, filename: str) -> str:
        """
        세션별 디렉터리 경로와 파일 경로를 생성 및 반환
        - BASE_DIR/오늘날짜/session 디렉터리 생성
        - 해당 디렉터리 내 filename 경로 리턴
        """
        session_dir = os.path.join(BASE_DIR, _today(), session)
        _ensure_dir(session_dir)
        return os.path.join(session_dir, filename)

    def _read_file(path: str, max_size: int = None) -> str | None:
        """
        파일을 읽어 문자열로 반환
        - 파일이 없으면 빈 문자열
        - max_size 지정 시, 크기 초과 시 None 반환
        """
        if not os.path.exists(path):
            return ''
        size = os.path.getsize(path)
        if max_size and size > max_size:
            return None
        with open(path, 'r', encoding='utf-8') as f:
            return f.read()

    def _write_file(path: str, content: str) -> bool:
        """
        문자열 콘텐츠를 파일에 쓰기
        - 성공 시 True, 실패 시 False
        """
        try:
            with open(path, 'w', encoding='utf-8') as f:
                f.write(content)
            return True
        except Exception:
            return False

    def _read_json(path: str) -> dict | None:
        """
        JSON 파일을 읽어 dict로 반환
        - 파일 없으면 빈 dict
        - 파싱 에러 시 None
        """
        if not os.path.exists(path):
            return {}
        try:
            with open(path, 'r', encoding='utf-8') as f:
                return json.load(f)
        except Exception:
            return None

    def _write_json(path: str, data: dict) -> bool:
        """
        dict를 JSON 파일로 저장
        - 성공 시 True, 실패 시 False
        """
        try:
            with open(path, 'w', encoding='utf-8') as f:
                json.dump(data, f, ensure_ascii=False, indent=2)
            return True
        except Exception:
            return False

    def _extract_file_name(text: str) -> str:
        """
        텍스트에서 처음 등장하는 file_name 값을 추출해 반환
        - 정규식 사용
        """
        m = re.search(r'file_name="(.*?)"', text)
        return m.group(1) if m else ''

    # ===== 메인 로직 =====
    order = chat_data.get('order', '')
    session = chat_data.get('session_id', '')

    # 1) 파일 업로드 중복 처리
    if order == 'check_genosUploaded':
        content = chat_data.get(UPLOAD_FIELD, '')
        path = _get_path(session, f"{UPLOAD_FIELD}.txt")

        # 이전 업로드 내용 읽기 (크기 초과 시 None)
        prev = _read_file(path, MAX_UPLOAD_SIZE)
        if prev is None:
            return 'FILE_TOO_LARGE'

        # 파일 이름 비교
        old_name = _extract_file_name(prev or '')
        new_name = _extract_file_name(content)
        if old_name and old_name == new_name:
            return 'NO_FILE_UPLOADED'

        # 새 콘텐츠 저장
        return content if _write_file(path, content) else 'FILE_ERROR'

    # 2) 채팅 기록 처리: chat.json 읽기/쓰기
    chat_path = _get_path(session, 'chat.json')

    # 2-1) write: 대화 추가 저장
    if order == 'write':
        history = _read_json(chat_path)
        if history is None:
            return {'status': 'error', 'message': 'Invalid chat history file'}

        ts = datetime.now().isoformat()
        history[ts] = {
            'question': chat_data.get('question'),
            'file_attachment': chat_data.get('file_content'),
            'answer': chat_data.get('answer')
        }
        success = _write_json(chat_path, history)
        return (
            {'status': 'success', 'message': 'Chat history saved', 'timestamp': ts}
            if success
            else {'status': 'error', 'message': 'Failed to write history'}
        )

    # 2-2) read: 최근 채팅 반환 (윈도우 & 글자 수 제한)
    if order == 'read':
        history = _read_json(chat_path)
        if history is None:
            return {'status': 'error', 'message': 'Invalid chat history file'}

        # 요청 파라미터 처리
        read_window = int(chat_data.get('read_window_size', 3))
        char_limit = int(chat_data.get('chat_length', float('inf')))

        # 최신순 정렬 후 최대 윈도우 크기 적용
        items = sorted(history.items(), key=lambda x: x[0], reverse=True)
        for win in range(min(read_window, len(items)), 0, -1):
            slice_ = items[:win]
            # 총 문자 수 계산
            total_len = sum(
                len(str(v))
                for ts, item in slice_
                for v in [ts, *item.values()]
            )
            # 문자 수 제한 안 맞으면 윈도우 줄여 재시도
            if total_len <= char_limit:
                # 오래된 순으로 정렬하여 반환 리스트 생성
                recent = [dict(timestamp=ts, **item)
                          for ts, item in sorted(slice_, key=lambda x: x[0])]
                return {
                    'status': 'success',
                    'message': f'Retrieved {win} chats ({total_len}/{char_limit})',
                    'recent_chats': recent,
                    'final_window_size': win,
                    'total_length': total_len,
                    'requested_window_size': read_window,
                    'character_limit': char_limit
                }
        # 제한에 맞는 채팅이 없을 때
        return {
            'status': 'success',
            'message': 'No chats within limit',
            'recent_chats': [],
            'final_window_size': 0,
            'total_length': 0
        }

    # 3) 잘못된 order 값 처리
    return {'status': 'error', 'message': f"Invalid order: {order}. Must be 'write', 'read', or 'check_genosUploaded'"}

배포가 완료 되었다면, 아래 절차를 통해 워크플로우 상세 >> 인증 키 탭 클릭 >> 인증키 생성 인증키를 생성하고, 생성된 인증키 및 해당 워크플로우의 ID를 기록해둡니다. 워크플로우의 ID의 경우 https://genos.genon.ai:3443/agent/workflow/detail/2491/에서 2491가 바로 해당 워크플로우의 ID 입니다.

\

3.2 Flowise 워크플로우 생성

새롭게 GenOS 메뉴 >> 에이전트 >> 워크플로우 >> 워크플로우 생성 을 통해 워크플로우를 생성한 뒤 리비전 상세에서 Flowise 단계 추가 를 눌러 Agentflow 3.0.0을 클릭합니다. 이후 사진 같은 Flowise 워크플로우를 작성합니다.

\

각 노드별로 아래 사항을 확인합니다.

1. Start

• Flow state에 file_content chat_history 가 생성되어 있는지 여부

• Ephemeral Memory 토글 버튼이 off 되어 있는지 여부 \

1. Custom Function: genosUploaded 초기화

• Input Variables 와 Update Flow State가 올바르게 채워 졌는지 여부

• Javascript Function 이 아래와 같은지 여부

const fetch = require('node-fetch');

const WORKFLOW_URL = 'https://genos.mnc.ai:3443/api/gateway/workflow/이전에 작성한 파이썬 워크플로우 ID 넘버/run/v2';
const BEARER_TOKEN = '이전에 생성한 파이썬 워크플로우 인증 키';

// 업로드된 파일이 문자열 형태로 들어감
const file_str = $vars.genosUploaded;
const session = $session;

// JSON payload 구조는 그대로
const payload = {
  order: 'check_genosUploaded',
  genosUploaded: file_str,
  session_id: session
};

// 워크플로우 서버로 POST
const options = {
    method: 'POST',
    headers: {
        'Content-Type': 'application/json',
        'Authorization': `Bearer ${BEARER_TOKEN}`
    },
    body: JSON.stringify(payload)
};

try {
    const response = await fetch(WORKFLOW_URL, options);
    if (!response.ok) {
        const errorText = await response.text();
        return `워크플로우 서버 응답 실패 (HTTP ${response.status}): ${errorText}`;
    }
    const json = await response.json();
    // data 필드만 추출
    const result = json?.data || 'NO_FILE_UPLOADED';
    return result;
} catch (e) {
    return `서버 또는 네트워크 시스템 오류: ${String(e)}`;
}

\

1. Custom Function: chat history 읽기

• Input Variables 와 Update Flow State가 올바르게 채워 졌는지 여부

• Javascript Function 이 아래와 같은지 여부

const fetch = require('node-fetch');

const WORKFLOW_URL = 'https://genos.mnc.ai:3443/api/gateway/workflow/이전에 작성한 파이썬 워크플로우 ID 넘버/run/v2';
const BEARER_TOKEN = '이전에 생성한 파이썬 워크플로우 인증 키';

// 채팅 데이터 구성 (read용, read_window_size, chat_length 통해 커스텀 가능)
const chat_data = {
    session_id: $session,
    order: "read",
    read_window_size: "4",
    chat_length: "15000"
};

// 워크플로우 서버로 POST
const options = {
    method: 'POST',
    headers: {
        'Content-Type': 'application/json',
        'Authorization': `Bearer ${BEARER_TOKEN}`
    },
    body: JSON.stringify(chat_data)
};

try {
    const response = await fetch(WORKFLOW_URL, options);
    if (!response.ok) {
        const errorText = await response.text();
        return `❌ 채팅 읽기 실패 (HTTP ${response.status}): ${errorText}`;
    }
    
    const json = await response.json();
    const result = json?.data || 'NO_CHAT_HISTORY';
    
    // Python 워크플로우 응답 형식에 맞게 처리
    if (result === 'NO_CHAT_HISTORY') {
        return 'NO_CHAT_HISTORY';
    }
    
    // 성공적으로 읽어왔는지 확인
    if (result.status === "success") {
        // recent_chats 배열 반환
        return result.recent_chats || [];
    } else {
        return `❌ 채팅 읽기 오류: ${result.message}`;
    }
    
} catch (e) {
    return `❌ 서버 또는 네트워크 시스템 오류: ${String(e)}`;
}

\

1. LLM

• Enable Memory가 꺼져있는지 여부

• Add Messages 에서의 시스템 프롬프트가 아래와 같이 작성되어 있는지 여부

Your Job:

Chat History, Attatched File, User's Question 을 참고하여 답하시오.



Chat History:

{{$flow.state.chat_history}}



Attatched File:

{{$flow.state.file_content}}



User's Question:

{{question}}

\

1. Custom Function: chat history 쓰기

• Input Variables 와 Update Flow State가 올바르게 채워 졌는지 여부

• Javascript Function 이 아래와 같은지 여부

const fetch = require('node-fetch');

const WORKFLOW_URL = 'https://genos.mnc.ai:3443/api/gateway/workflow/이전에 작성한 파이썬 워크플로우 ID 넘버/run/v2';
const BEARER_TOKEN = '이전에 생성한 파이썬 워크플로우 인증 키';

// 채팅 데이터 구성
const chat_data = {
    session_id: $session,
    question: $question,
    file_content: $file_content || null,
    answer: $answer,
    order: "write",
    read_window_size: null,
    chat_length: null,
};

// 워크플로우 서버로 POST
const options = {
    method: 'POST',
    headers: {
        'Content-Type': 'application/json',
        'Authorization': `Bearer ${BEARER_TOKEN}`
    },
    body: JSON.stringify(chat_data)
};

try {
    const response = await fetch(WORKFLOW_URL, options);
    if (!response.ok) {
        const errorText = await response.text();
        console.log(`❌ 채팅 저장 실패 (HTTP ${response.status}): ${errorText}`);
    } else {
        const json = await response.json();
        const result = json?.data || { status: "error", message: "No response data" };
        
        // 성공 여부 확인
        if (result.status === "success") {
            console.log(`✅ 채팅 저장 완료: ${result.message}`);
        } else {
            console.log(`❌ 채팅 저장 오류: ${result.message}`);
        }
    }
} catch (e) {
    console.log(`❌ 서버 또는 네트워크 시스템 오류: ${String(e)}`);
}

// 무조건 answer를 반환
return $answer;

\

1. 위 단계를 모두 끝냈다면 워크플로우를 저장한 뒤 배포하고, 튜토리얼: 기본편의 채팅 애플리케이션 만들기 를 참고해 채팅을 배포한 뒤 사용합니다.

\

4. 마치며: 주의사항

위 구현 방법을 따라 재현하면 이제 Custom 하게 제작된 chat history를 통해 보다 효율적이게 멀티턴을 구현할 수 있습니다. 또한, 단순히 이전 채팅을 기록함을 넘어 업로드된 파일 내용 또한 chat history에 남기에 파일 기반으로 채팅을 진행하는 경우 보다 정확한 답변이 가능합니다. 다만, 현재 예시 워크플로우 내에서는 Custom Function: chat history 읽기 부분에서 최대 4턴을 기억하고 또한 그 4턴 이내에 15000자가 넘는 경우 다시 3턴, 그럼에도 15000자가 넘으면 2턴까지 chat history를 불러오도록 설정 되어 있으므로, 해당 부분은 사용자의 용도와 워크플로우에 사용된 LLM의 능력에 따라 맞춤설정이 필요합니다. 또한, 대화내역 자체를 chat_history 라는 state로 관리하고 있기에, 앞선 설명처럼 LLM Node Agent Node 등의 LLM을 활용하는 부분에서 Flowise 내부 자체적인 멀티턴 기능(Enable Memory)은 Off한 뒤 별도로 시스템 프롬프트에 chat_history라는 state를 입력하여 대화에 참조되도록 하여야 합니다.