데이터 분석 워크플로우 만들기

"데이터 분석" 애플리케이션 규격에 맞는 워크플로우를 구현합니다. 일반적인 워크플로우와 달리 데이터 분석 워크플로우에는 데이터 분석을 위해 구현해야 하는 필수 기능이 정해져 있습니다. 예시 섹션에서 기능 구현에 관해 좀 더 자세한 설명을 볼 수 있습니다.

목차

  1. 데이터 소스와 워크플로우 연동

  2. 워크플로우 구현 시 지켜야 하는 백엔드 규격

  3. 워크플로우 구현 예시

  4. 데이터 분석 채팅 실행 예시

데이터 소스와 워크플로우 연동

"애플리케이션" > "데이터 분석" > "데이터 소스"에서 업로드한 데이터는 RDB에 내부적으로 삽입됩니다. "데이터 분석" > "데이터 분석 채팅" 생성 시 데이터 소스와 데이터 분석을 위해 구현한 워크플로우가 함께 구성됩니다. "데이터 분석 채팅" 생성하기 도큐먼트를 참고해주세요.

채팅에 의해 함께 구성된 데이터와 워크플로우를 연동하기 위해서는 워크플로우를 구현할 때 이 데이터가 저장된 DB에 접근해야 합니다. DB의 정보(host, port, user 등)가 API를 통해 워크플로우로 전달되고, 워크플로우 내부 코드창에서 $vars 변수로 해당 정보를 얻을 수 있습니다. 다음과 같습니다.

const mysql = require('mysql2/promise');
// Connect to the DB
const connection = await mysql.createConnection({
    host: $vars.host,
    port: Number($vars.port),
    user: $vars.user,
    password: $vars.password,
    database: $vars.database
});

const tableNames = $vars.tables;  // Array of table names in the database

  • 워크플로우의 'CustomFuction' 노드를 사용해 Javascript 코드에서 $vars 변수를 불러옵니다.

  • 위 코드와 같은 방식으로 데이터 소스가 저장된 DB에 연결할 수 있습니다.

  • DB에 연결하여 데이터 스키마 추출, SQL 쿼리 실행 등을 워크플로우 'CustomFuction' 노드에서 코드 작성을 통해 구현할 수 있습니다.

워크플로우 구현 시 지켜야 하는 백엔드 규격

데이터 분석 에이전트로서 갖춰야 하는 기능을 중심으로 백엔드 API 파싱을 위한 규격을 정의했습니다. 규격은 다음과 같습니다:

Role Name
Description
Output State Name

Schema

데이터의 스키마를 추출하는 노드

{schema, db_schema, rdb_schema, db_table}

Query

알맞은 SQL Query를 생성하는 LLM 노드

{statement, query, sql_query}

Data

SQL Query를 실행하여 얻은 데이터를 반환하는 노드

{data, sql_result}

Chart

사용자의 질의와 데이터가 차트 생성에 적합한 경우 차트 생성을 위한 데이터를 추출/리턴하는 노드

{chart_data, chart}

Report

최종 데이터 분석 리포트를 작성하는 LLM 노드

{report}

Chat

일상적인 채팅을 진행하는 LLM 노드

[비고] 해당 노드 인식 시 파싱 종료

  • 해당 Description의 기능을 수행하는 워크플로우 노드 이름에 "DAA" + Role Name이 포함되도록 합니다 (예: "DAA-Schema", "DAA Generate Query", "DAA_Get_Data").

    • Role Name이 노드명에 포함되어 있으면 해당 노드는 해당 역할로 인식됩니다: 노드명을 띄어쓰기 및 특수문자로 tokenize 하여 Role - Output State 규격을 인식합니다.

    • Role Name 외 다른 텍스트를 이름에 포함시킬 수 있습니다.

  • 해당 Role Name이 이름에 포함된 노드는 위 정의된 Output State Name 중에서 하나의 이름으로 된 state를 반드시 리턴해야 합니다. 리턴되는 state명을 기준으로 백엔드에서 파싱이 이루어집니다.

  • 노드의 이름(Role Name) - 기능(Description) - 출력값(Output State) 세 가지가 반드시 매칭되어야 합니다.

워크플로우 구현 예시

워크플로우 3.0.0 버전을 예시로 데이터 분석 워크플로우로서 필수적으로 구현해야 할 노드를 살펴봅니다. 위 백엔드 규격에 맞추어 필수 노드와 해당 노드의 아웃풋을 맞추어 구현하면 됩니다.

State 정의

  • Start 노드에서 파싱을 위해 필수적으로 사용될 키 값(rdb_schema, sql_query, sql_result, chart_data, report 등)을 정의합니다. 필수키 값 외 구현 시 필요한 키를 추가적으로 정의할 수 있습니다.

스키마 추출 - Schema

  • Schema 노드는 데이터의 스키마를 추출해 {schema, db_schema, rdb_schema, db_table} 중 하나의 키로 리턴합니다.

  • 다음 예시에서는 'CustomFuction' 노드로 구현되었습니다. 예시 코드는 아래와 같습니다:

    const mysql = require('mysql2/promise');

  const tableNames = $vars.tables;
  const database = $vars.database;

  try {
      const connection = await mysql.createConnection({
          host: $vars.host,
          port: Number($vars.port),
          user: $vars.user,
          password: $vars.password,
          database: database
      });

      if (!Array.isArray(tableNames) || tableNames.length === 0) {
          throw new Error("rdb_table must be a non-empty array.");
      }

      const resultList = [];
      for (const tableName of tableNames) {
          const [columnsData] = await connection.execute(`
              SELECT COLUMN_NAME, DATA_TYPE, COLUMN_KEY
              FROM INFORMATION_SCHEMA.COLUMNS
              WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ?
              ORDER BY ORDINAL_POSITION
          `, [database, tableName]);

          const columnDetails = [];
          for (const col of columnsData) {
              const colInfo = {
                  name: col.COLUMN_NAME,
                  type: col.DATA_TYPE
              };
          
              if (col.COLUMN_KEY === 'PRI') colInfo.primary = true;
              else if (col.COLUMN_KEY === 'UNI') colInfo.unique = true;

              // Get unique values of a column
              const [res] = await connection.execute(
                  `SELECT COUNT(DISTINCT \`${col.COLUMN_NAME}\`) as cnt FROM \`${tableName}\``
              );

              // Add 'unique_values' k-v if the column has less or same as 30 unique values
              if (res[0].cnt > 0 && res[0].cnt <= 30) {  
                  const [values] = await connection.execute(
                      `SELECT DISTINCT \`${col.COLUMN_NAME}\` as val FROM \`${tableName}\` WHERE \`${col.COLUMN_NAME}\` IS NOT NULL LIMIT 30`
                  );
                  colInfo.unique_values = values.map(r => r.val);
              // Add 'sample_values' k-v for the column including more than 30 unique values
              } else {
                  const [sampleValues] = await connection.execute(
                      `SELECT \`${col.COLUMN_NAME}\` as val FROM \`${tableName}\` ORDER BY RAND() LIMIT 15`
                  );  // Extract sample values randomly from the column
                  colInfo.sample_values = sampleValues.map(r => r.val);
              }

              columnDetails.push(colInfo);
          }

          resultList.push({
              table_name: tableName,
              columns: columnDetails,
          });
      }

      await connection.end();

      return {
          "notice": "If a column has less or same as 30 unique values, the \"unique_values\" are shown. Or for more than 30 unique values, the \"sample_values\" are shown.", 
          "tables": resultList
      };

  } catch (error) {
      return {
          error: error.message,
          debug_info: {
              host: $vars.host,
              port: Number($vars.port),
              user: $vars.user,
              password: $vars.password,
              database: database,
              tables: tableNames
          }
      };
  }

SQL 쿼리 생성 - Query

  • Query노드는 사용자 질의와 데이터 스키마에 적합한 쿼리를 생성하고, 생성한 쿼리를 {statement, query, sql_query} 중 하나의 키로 리턴합니다.

  • 다음 예시에서는 'LLM' 노드로 구현되었습니다. 프롬프트 엔지니어링을 통해 쿼리 생성 성능을 최적화합니다.

SQL 쿼리 실행 - Data

  • Data 노드는 앞서 생성한 SQL 쿼리를 연동된 DB에서 실행해 검색된 데이터를 스트링 포맷으로 {data, sql_result}중 하나의 키로 리턴합니다.

  • 다음 예시에서는 'CustomFuction' 노드로 구현되었습니다. 예시 코드는 아래와 같습니다:

    const mysql = require('mysql2/promise');

let cleanQuery = '';
let connection;

try {
    // SQL 입력 검증
    if (!$SQL_QUERY || typeof $SQL_QUERY !== 'string') {
        throw new Error('SQL_QUERY is required and must be a string');
    }

    cleanQuery = $SQL_QUERY.trim();
    if (!cleanQuery) {
        throw new Error('SQL_QUERY cannot be empty');
    }

    // DB 연결 설정값 검증
    const requiredVars = ['host', 'port', 'user', 'password', 'database'];
    for (const key of requiredVars) {
        if (!$vars[key]) {
            throw new Error(`Missing required DB config: ${key}`);
        }
    }

    // DB 연결
    connection = await mysql.createConnection({
        host:     $vars.host,
        port:     Number($vars.port),
        user:     $vars.user,
        password: $vars.password,
        database: $vars.database
    });

    // 쿼리 실행
    const [rows, fields] = await connection.execute(cleanQuery);
    const columnOrientedData = {};  // 컬럼 단위로 변환
    if (Array.isArray(rows) && rows.length > 0) {
        for (const field of fields) {
            columnOrientedData[field.name] = rows.map(row => row[field.name]);
        }
    }

    const columnJson = JSON.stringify(columnOrientedData);
    const MAX_LENGTH = 3000;  // 결과 길이 제한
    if (columnJson.length > MAX_LENGTH) {
        return {
            success: true,
            columns: {
                message: `[!] 쿼리는 정상적으로 실행되었지만, 결과가 약 ${columnJson.length.toLocaleString()}자에 달하여 ${String(MAX_LENGTH)}자를 초과하였습니다.\n\n너무 많은 실행 결과를 불러일으키지 않도록 쿼리 조건을 제한하여 다시 시도해주세요.`
            }
        };
    }

    // 정상 결과 반환
    return {
        success: true,
        columns: columnOrientedData
    };

} catch (error) {
    // SQL 실행 오류
    const errorMap = {
        ER_NO_SUCH_TABLE:        { type: 'TABLE_NOT_FOUND',  msg: `테이블을 찾을 수 없습니다: ${error.message}` },
        ER_BAD_FIELD_ERROR:      { type: 'COLUMN_NOT_FOUND', msg: `컬럼을 찾을 수 없습니다: ${error.message}` },
        ER_PARSE_ERROR:          { type: 'SYNTAX_ERROR',     msg: `SQL 문법 오류: ${error.message}` },
        ECONNREFUSED:            { type: 'CONNECTION_ERROR', msg: '데이터베이스 연결에 실패했습니다' },
        ER_ACCESS_DENIED_ERROR:  { type: 'ACCESS_DENIED',    msg: '데이터베이스 접근 권한이 없습니다' },
        PROTOCOL_CONNECTION_LOST:{ type: 'CONNECTION_LOST',  msg: '데이터베이스 연결이 끊어졌습니다' }
    };

    const mapped = errorMap[error.code] ?? {
        type: error.code ?? 'UNKNOWN_ERROR',
        msg:  error.message
    };

    return {
        success: false,
        error: {
            type: mapped.type,
            message: mapped.msg,
            original_message: error.message,
            code: error.code ?? null,
            errno: error.errno ?? null,
            sql_state: error.sqlState ?? null
        },
        query: cleanQuery || $SQL_QUERY,
        timestamp: new Date().toISOString()
    };

} finally {
    // 연결 종료
    if (connection) {
        try {
            await connection.end();
        } catch (closeError) {
            console.warn('[!] DB 연결 종료 중 에러:', closeError.message);
        }
    }
}

차트 생성에 필요한 데이터 추출 - Chart

  • Chart 노드는 차트 생성을 위해 필요한 데이터를 추출해 {chart_data, chart} 중 하나의 키로 리턴합니다. 해당 값은 프론트엔드로 전달되어 차트 생성 시 사용됩니다.

  • 이 예시에서 정의한 차트 타입은 다음과 같습니다:

    • single: 하나의 차트 생성

    • mixed: 동일 y축에 두 개의 차트를 겹쳐 생성

    • dual-axis: y축이 다른 두 개의 차트를 겹쳐 생성

    • 차트 타입 별 설정:

      'single': {
  "chart_type": "bar|line|pie",
  "title": "<차트 제목>",
  "x_values": ["<x축1>", "<x축2>"],
  "y_values": [<값1>, <값2>],
  "y_label": "<y축 라벨>"
}

'mixed': {
  "chart_type": "mixed",
  "title": "<차트 제목>",
  "x_values": ["<x축1>", "<x축2>"],
  "datasets": [
    { "type": "bar|line|pie", "label": "<라벨>", "data": [<값1>, <값2>] }
  ],
  "y_label": "<y축 라벨>"
}

'dual_axis': {
  "chart_type": "dual_axis",
  "title": "<차트 제목>",
  "x_values": ["<x축1>", "<x축2>"],
  "datasets": [
    { "type": "bar|line", "label": "<라벨1>", "data": [<값1>, <값2>], "yAxisID": "y1" },
    { "type": "line", "label": "<라벨2>", "data": [<값1>, <값2>], "yAxisID": "y2" }
  ],
  "y_axes": [
    { "id": "y1", "label": "<좌측축 라벨>" },
    { "id": "y2", "label": "<우측축 라벨>" }
  ]
}

  • 이 예시에서는 사용자 질의 및 데이터 스키마에 따라 차트 타입을 결정한 뒤, 위 차트 타입 설정을 프롬프트에 포함시켜 LLM이 적절한 차트 데이터를 추출하도록 구현했습니다.

최종 리포트 작성 - Report

  • Report 노드는 사용자 질의에 대한 답변으로 최종 출력될 텍스트를 생성하여 {report} 키로 리턴합니다.

  • 이 예시에서는 'LLM' 노드를 사용해 사용자 질의와 검색된 데이터(Data 노드에서 리턴된 state)를 참조해 최종 텍스트를 작성하도록 프롬프트 엔지니어링하였습니다.

  • 차트가 생성되지 않는 경우의 데이터 분석 리포트는 Chat 노드로 구현합니다.: 대시보드에 저장할 차트가 없는 경우 해당 대화 내용을 별도 저장하지 않습니다.

일상 채팅 - Chat

  • Chat 노드는 사용자 질의나 데이터가 데이터 분석 프로세스를 진행하기 적합하지 않은 경우 등 데이터 분석 애플리케이션 내 로그에 해당 대화 정보를 기록할 필요가 없는 경우의 대화를 처리하도록 구현할 수 있는 모듈입니다.

  • 백엔드 파싱을 위해 리턴할 값이 존재하지 않습니다.

  • 조건에 따라 필요 시 'LLM' Chat 노드를 구현할 수 있습니다.

데이터 분석 채팅 실행 예시

데이터 분석 워크플로우 구성/배포 후 "데이터 분석 채팅"을 생성하여 사용하고자 하는 데이터 소스 및 해당 워크플로우를 지정, 배포하면 채팅을 사용할 수 있습니다.

Previous보고서 생성 워크플로우 만들기NextSearch Filter을 적용한 RAG 워크플로우 만들기

Last updated

Was this helpful?