{ "cells": [ { "cell_type": "markdown", "id": "cell-0", "metadata": {}, "source": [ "# OneClickAI 통합 텍스트 분류\n", "이 노트북은 Google Colab 환경에서 실행하도록 작성되었습니다.\n", "웹앱의 **통합 텍스트 분류**와 같은 방법으로 학습합니다.\n", "\n", "글(텍스트 열)을 입력으로 받고, 출력 열마다 **분류**(카테고리)나 **회귀**(수치)를 정해 한 번에 학습합니다.\n", "모델은 웹앱과 같은 **작은 트랜스포머**입니다. 토큰마다 임베딩 벡터가 되고, 어텐션이 낱말 순서와 문맥을 봅니다.\n", "\n", "**순서**\n", "1. 패키지 설치 및 파일 업로드\n", "2. 열 역할과 종류 정하기\n", "3. 글을 숫자로 바꾸기 (토큰 사전과 시퀀스)\n", "4. 모델 학습 (임베딩 + 어텐션 블록)\n", "5. 결과 확인 및 예측" ] }, { "cell_type": "markdown", "id": "cell-1", "metadata": {}, "source": [ "## 1. 패키지 설치 및 파일 업로드\n", "\n", "웹앱과 같은 토크나이저(`tiktoken o200k_base`)를 쓰기 위해 패키지를 설치합니다.\n", "엑셀 파일(`.xlsx`)을 업로드해 주세요. 두 가지 모양을 읽습니다.\n", "\n", "- **시트 하나**: 웹앱의 `데이터 저장(.xlsx)` 버튼으로 받은 파일입니다. 첫 행은 열 이름, 첫 열은 항목(이름)입니다.\n", "- **원인/결과 두 시트**: 텍스트 분류 페이지의 엑셀입니다. 두 시트를 행 순서대로 옆으로 붙입니다.\n", " 결과 시트의 열이 원핫(행마다 1이 하나)이면, 1인 열 이름을 값으로 쓰는 `결과` 열 하나로 합칩니다.\n", "\n", "웹앱은 행을 999개까지 쓰지만 이 노트북에는 제한이 없습니다. 영화 리뷰 2000개 엑셀도 그대로 올릴 수 있습니다." ] }, { "cell_type": "code", "id": "cell-2", "metadata": {}, "source": [ "# ==========================================\n", "# 패키지 설치\n", "# ==========================================\n", "!pip install -q tiktoken\n", "\n", "# ==========================================\n", "# 불러오기\n", "# ==========================================\n", "import math\n", "import unicodedata\n", "from collections import Counter\n", "\n", "import numpy as np\n", "import pandas as pd\n", "import matplotlib.pyplot as plt\n", "import keras\n", "import tiktoken\n", "from google.colab import files\n", "\n", "# ==========================================\n", "# 파일 업로드\n", "# ==========================================\n", "print('데이터 파일(.xlsx)을 업로드 해주세요.\\n')\n", "uploaded = files.upload()\n", "\n", "if len(uploaded.keys()) == 0:\n", " raise ValueError('파일이 업로드되지 않았습니다.')\n", "\n", "DATA_FILE_PATH = list(uploaded.keys())[0]\n", "\n", "if not DATA_FILE_PATH.endswith(('.xls', '.xlsx')):\n", " raise ValueError('지원하지 않는 파일 형식입니다. .xlsx 파일을 업로드 해주세요.')\n", "\n", "\n", "# ==========================================\n", "# 칸 값 다루기 (웹앱과 같은 규칙)\n", "# ==========================================\n", "def is_blank(value):\n", " if value is None:\n", " return True\n", " if isinstance(value, float) and math.isnan(value):\n", " return True\n", " return isinstance(value, str) and value.strip() == ''\n", "\n", "\n", "def js_number(n):\n", " \"\"\"숫자를 웹앱(자바스크립트)과 같은 글자로 쓴다. 1.0 은 '1'\"\"\"\n", " n = float(n)\n", " if n.is_integer() and abs(n) < 1e21:\n", " return str(int(n))\n", " return repr(n)\n", "\n", "\n", "def js_string(value):\n", " \"\"\"칸 값을 자바스크립트의 String(값) 처럼 글자로 바꾼다\"\"\"\n", " if value is None:\n", " return ''\n", " if isinstance(value, (bool, np.bool_)):\n", " return 'true' if value else 'false'\n", " if isinstance(value, (int, float, np.integer, np.floating)):\n", " return js_number(value) if math.isfinite(value) else ''\n", " return str(value)\n", "\n", "\n", "def to_cell_string(value):\n", " \"\"\"칸 값을 비교용 글자로: NFC, 대시 통일, 앞뒤 공백 제거\"\"\"\n", " text = unicodedata.normalize('NFC', js_string(value))\n", " for dash in '\\u2013\\u2014\\u2212':\n", " text = text.replace(dash, '-')\n", " return text.strip()\n", "\n", "\n", "def parse_number(value):\n", " \"\"\"숫자로 읽히면 float, 아니면 None\"\"\"\n", " text = to_cell_string(value)\n", " if text == '' or '_' in text:\n", " return None\n", " try:\n", " n = float(text)\n", " except ValueError:\n", " return None\n", " return n if math.isfinite(n) else None\n", "\n", "\n", "def read_table(frame):\n", " \"\"\"시트를 [행][칸] 목록으로. 빈 행은 빼고, 열은 머리글이 있는 곳까지만 둔다\"\"\"\n", " table = [list(row) for row in frame.values.tolist()]\n", " table = [row for row in table if not all(is_blank(v) for v in row)]\n", " if not table:\n", " raise ValueError('시트에 데이터가 없습니다.')\n", " filled = [i for i, v in enumerate(table[0]) if not is_blank(v)]\n", " if not filled:\n", " raise ValueError('첫 행(머리글)에 열 이름이 없습니다.')\n", " width = filled[-1] + 1\n", " return [row[:width] + [None] * (width - len(row)) for row in table]\n", "\n", "\n", "# ==========================================\n", "# 엑셀 읽기\n", "# ==========================================\n", "sheets = pd.read_excel(DATA_FILE_PATH, sheet_name=None, header=None, dtype=object)\n", "merged_output_count = 0 # 원인/결과를 붙였을 때 결과 시트에서 온 열 개수\n", "collapsed_index = None # 원핫을 합쳐 만든 열의 위치 (분류로 고정)\n", "\n", "if '원인' in sheets and '결과' in sheets:\n", " cause = read_table(sheets['원인'])\n", " result = read_table(sheets['결과'])\n", " if len(cause) != len(result):\n", " raise ValueError(f\"'원인' 시트({len(cause) - 1}행)와 '결과' 시트({len(result) - 1}행)의 행 수가 다릅니다.\")\n", " cols = cause[0] + result[0][1:]\n", " rows = [c + r[1:] for c, r in zip(cause[1:], result[1:])]\n", " merged_output_count = len(result[0]) - 1\n", " print(\"'원인' 시트와 '결과' 시트를 행 순서대로 붙였습니다.\")\n", "else:\n", " sheet_name = '데이터' if '데이터' in sheets else list(sheets.keys())[0]\n", " table = read_table(sheets[sheet_name])\n", " cols, rows = table[0], table[1:]\n", " print(f\"'{sheet_name}' 시트를 읽었습니다.\")\n", "\n", "if is_blank(cols[0]):\n", " cols[0] = '항목'\n", "\n", "\n", "# 결과 시트의 열들이 원핫이면 '결과' 열 하나로 합친다 (예: positive, negative -> 결과)\n", "def is_one_hot(cols, rows, count):\n", " if count < 2 or count > len(cols) - 2 or not rows:\n", " return False\n", " start = len(cols) - count\n", " for row in rows:\n", " values = [parse_number(v) for v in row[start:]]\n", " if any(v not in (0, 1) for v in values) or sum(values) != 1:\n", " return False\n", " return True\n", "\n", "\n", "if merged_output_count and is_one_hot(cols, rows, merged_output_count):\n", " start = len(cols) - merged_output_count\n", " names = [to_cell_string(c) or f'열 {start + i}' for i, c in enumerate(cols[start:])]\n", " folded = []\n", " for row in rows:\n", " label = ''\n", " for i, v in enumerate(row[start:]):\n", " if parse_number(v) == 1:\n", " label = names[i]\n", " folded.append(row[:start] + [label])\n", " cols = cols[:start] + ['결과']\n", " rows = folded\n", " merged_output_count = 1\n", " collapsed_index = len(cols) - 1\n", " print(f\"결과 시트의 {', '.join(names)} 열은 행마다 1이 하나뿐이라 '결과' 열 하나로 합쳤습니다.\")\n", "\n", "column_names = [to_cell_string(c) or f'열 {i}' for i, c in enumerate(cols)]\n", "print(f'\\n행 {len(rows)}개, 열 {len(cols)}개: {column_names}')\n", "display(pd.DataFrame(rows[:5], columns=column_names))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "id": "cell-3", "metadata": {}, "source": [ "## 2. 열 역할과 종류 정하기\n", "\n", "열마다 **역할**(입력/출력/제외)과 **종류**(텍스트/카테고리/수치)를 정합니다.\n", "아래 설정을 비워 두면 웹앱과 같은 규칙으로 자동으로 정합니다.\n", "\n", "- 숫자가 아닌 열은 글이 길거나(평균 8자 이상) 서로 다른 값이 많으면 **텍스트**, 아니면 **카테고리**\n", "- 숫자 열은 정수이면서 종류가 5개 이하면 **카테고리**, 아니면 **수치**\n", "- 텍스트 열은 입력, 나머지 열은 출력 (원인/결과 파일은 결과 시트에서 온 열이 출력)\n", "- 출력 열은 카테고리면 **분류**(Softmax), 수치면 **회귀**(Linear)로 배웁니다\n", "\n", "웹앱에서 받은 파일에는 열 설정이 들어 있지 않습니다.\n", "웹앱에서 바꾼 설정(예: 긴급도 1~5를 회귀로)은 아래 칸에 다시 적어 주세요." ] }, { "cell_type": "code", "id": "cell-4", "metadata": {}, "source": [ "# ==========================================\n", "# 열 설정 (비워 두면 웹앱과 같은 규칙으로 자동으로 정합니다)\n", "# ==========================================\n", "INPUT_COLUMNS = [] # 입력으로 쓸 텍스트 열 이름. 예: ['리뷰']\n", "OUTPUT_COLUMNS = [] # 출력으로 쓸 열 이름. 예: ['부서', '긴급도']\n", "KIND_OVERRIDES = {} # 출력 종류 바꾸기. 'categorical' 은 분류, 'numeric' 은 회귀. 예: {'긴급도': 'numeric'}\n", "VOCAB_SIZE = 500 # 텍스트 열 하나의 사전 크기 (웹앱 선택지: 500, 1000, 2000)\n", "NORMALIZATION = 'minMax11' # 회귀 출력 정규화: 'none', 'minMax01'(0~1), 'minMax11'(-1~1, 기본), 'zScore'\n", "\n", "# ==========================================\n", "# 웹앱과 같은 기준값\n", "# ==========================================\n", "MAX_TEXT_CHARS = 200 # 칸 하나에서 읽는 최대 글자 수 (넘는 부분은 자름)\n", "MAX_CATEGORIES = 50 # 카테고리 열 하나의 종류 최대 개수\n", "CATEGORICAL_UNIQUE_MAX = 5 # 정수이면서 종류가 이 개수 이하이면 카테고리로 추정\n", "TEXT_MIN_AVG_CHARS = 8 # 숫자가 아닌 열의 평균 글자 수가 이 값 이상이면 텍스트로 추정\n", "TEXT_UNIQUE_RATIO = 0.5 # 서로 다른 값의 비율이 이 값 이상이면 텍스트로 추정 (행이 20개 이상일 때)\n", "TEXT_RATIO_MIN_ROWS = 20\n", "LABEL_REPEAT_MIN = 3 # 값 하나가 평균 이만큼 되풀이되면 글이 길어도 라벨로 본다\n", "\n", "KIND_LABELS = {'text': '텍스트', 'categorical': '카테고리', 'numeric': '수치'}\n", "ROLE_LABELS = {'input': '입력', 'output': '출력', 'exclude': '제외'}\n", "\n", "\n", "def normalize_text(value):\n", " \"\"\"앞뒤 공백을 떼고 NFC 로 맞춘 글자 (자르지 않음)\"\"\"\n", " return unicodedata.normalize('NFC', js_string(value)).strip()\n", "\n", "\n", "def text_length(value):\n", " \"\"\"글자 수. 웹앱(자바스크립트)처럼 이모지 같은 글자는 2로 센다\"\"\"\n", " return len(normalize_text(value).encode('utf-16-le')) // 2\n", "\n", "\n", "def suggest_kind(is_numeric, is_integer, unique_count, filled, avg_length):\n", " if is_numeric:\n", " return 'categorical' if is_integer and unique_count <= CATEGORICAL_UNIQUE_MAX else 'numeric'\n", " if filled == 0:\n", " return 'categorical'\n", " if unique_count > MAX_CATEGORIES:\n", " return 'text'\n", " if filled >= TEXT_RATIO_MIN_ROWS and unique_count / filled >= TEXT_UNIQUE_RATIO:\n", " return 'text'\n", " label_like = filled >= unique_count * LABEL_REPEAT_MIN\n", " if avg_length >= TEXT_MIN_AVG_CHARS and not label_like:\n", " return 'text'\n", " return 'categorical'\n", "\n", "\n", "def analyze_columns(cols, rows):\n", " \"\"\"열마다 값을 훑어 숫자인지, 서로 다른 값, 글 길이 등을 센다 (항목 열은 뺀다)\"\"\"\n", " stats = []\n", " for c in range(1, len(cols)):\n", " cells, numbers = [], []\n", " is_numeric = is_integer = True\n", " empty = length_sum = max_length = 0\n", " sample = ''\n", " for row in rows:\n", " raw = row[c] if c < len(row) else None\n", " s = to_cell_string(raw)\n", " if s == '':\n", " empty += 1\n", " continue\n", " cells.append(s)\n", " length = text_length(raw)\n", " length_sum += length\n", " max_length = max(max_length, length)\n", " if not sample:\n", " sample = normalize_text(raw)\n", " n = parse_number(s)\n", " if n is None:\n", " is_numeric = is_integer = False\n", " else:\n", " if not n.is_integer():\n", " is_integer = False\n", " numbers.append(n)\n", " if not numbers:\n", " is_numeric = is_integer = False\n", "\n", " # 서로 다른 값 (분류의 종류). 숫자 열은 1 과 1.0 을 같은 값으로 보고 크기 순서로 둔다\n", " unique, seen = [], set()\n", " for s in cells:\n", " key = js_number(parse_number(s)) if is_numeric else s\n", " if key not in seen:\n", " seen.add(key)\n", " unique.append(key)\n", " if is_numeric:\n", " unique.sort(key=float)\n", "\n", " filled = len(cells)\n", " mean = sum(numbers) / len(numbers) if numbers else None\n", " std = math.sqrt(max(sum(n * n for n in numbers) / len(numbers) - mean ** 2, 0)) if numbers else None\n", " avg_length = length_sum / filled if filled else 0\n", " stats.append({\n", " 'index': c,\n", " 'name': column_names[c],\n", " 'is_numeric': is_numeric,\n", " 'unique': unique,\n", " 'empty': empty,\n", " 'min': min(numbers) if numbers else None,\n", " 'max': max(numbers) if numbers else None,\n", " 'mean': mean,\n", " 'std': std,\n", " 'max_length': max_length,\n", " 'sample': sample,\n", " 'suggested': suggest_kind(is_numeric, is_integer, len(unique), filled, avg_length),\n", " })\n", " return stats\n", "\n", "\n", "stats = analyze_columns(cols, rows)\n", "by_name = {s['name']: s for s in stats}\n", "\n", "# ==========================================\n", "# 자동 역할과 종류\n", "# ==========================================\n", "first_output = len(cols) - merged_output_count if merged_output_count else None\n", "for s in stats:\n", " kind = s['suggested']\n", " if first_output is not None:\n", " # 원인/결과 파일: 결과 시트에서 온 열은 출력, 원인 시트의 열은 텍스트면 입력 아니면 제외\n", " if s['index'] >= first_output:\n", " role = 'output'\n", " # 원핫을 합친 열과 텍스트로 추정된 출력 열은 분류로 고정한다 (텍스트는 출력이 될 수 없다)\n", " if s['index'] == collapsed_index or kind == 'text':\n", " kind = 'categorical'\n", " else:\n", " role = 'input' if kind == 'text' else 'exclude'\n", " else:\n", " role = 'input' if kind == 'text' else 'output'\n", " s['role'], s['kind'] = role, kind\n", "\n", "# ==========================================\n", "# 직접 적은 설정 반영\n", "# ==========================================\n", "for name in list(INPUT_COLUMNS) + list(OUTPUT_COLUMNS) + list(KIND_OVERRIDES):\n", " if name not in by_name:\n", " raise ValueError(f\"'{name}' 열이 없습니다. 열 이름: {column_names[1:]}\")\n", "if INPUT_COLUMNS:\n", " for s in stats:\n", " if s['name'] in INPUT_COLUMNS:\n", " s['role'] = 'input'\n", " elif s['role'] == 'input':\n", " s['role'] = 'exclude'\n", "if OUTPUT_COLUMNS:\n", " for s in stats:\n", " if s['name'] in OUTPUT_COLUMNS:\n", " s['role'] = 'output'\n", " elif s['role'] == 'output':\n", " s['role'] = 'exclude'\n", "for name, kind in KIND_OVERRIDES.items():\n", " if kind not in ('categorical', 'numeric'):\n", " raise ValueError(f\"'{name}' 의 종류는 'categorical' 이나 'numeric' 으로 적어 주세요.\")\n", " by_name[name]['kind'] = kind\n", "for s in stats:\n", " if s['role'] == 'input' and not s['is_numeric']:\n", " s['kind'] = 'text' # 입력 열은 늘 글로 읽는다\n", "\n", "# ==========================================\n", "# 검사 (웹앱과 같은 조건)\n", "# ==========================================\n", "input_specs = [s for s in stats if s['role'] == 'input']\n", "output_specs = [s for s in stats if s['role'] == 'output']\n", "if not input_specs:\n", " raise ValueError('입력으로 쓸 텍스트 열이 없습니다. INPUT_COLUMNS 에 적어 주세요.')\n", "if not output_specs:\n", " raise ValueError('출력으로 쓸 열이 없습니다. OUTPUT_COLUMNS 에 적어 주세요.')\n", "for s in input_specs:\n", " if s['is_numeric']:\n", " raise ValueError(f\"'{s['name']}' 열은 숫자라 입력으로 쓸 수 없습니다. 이 노트북은 텍스트 열만 입력으로 씁니다.\")\n", " if len(s['unique']) <= 1:\n", " raise ValueError(f\"'{s['name']}' 열은 글이 한 가지뿐이라 학습에 도움이 되지 않습니다.\")\n", " if s['max_length'] > MAX_TEXT_CHARS:\n", " print(f\"'{s['name']}' 열에 {MAX_TEXT_CHARS}자가 넘는 글이 있습니다. 앞 {MAX_TEXT_CHARS}자까지만 읽습니다.\")\n", "for s in input_specs + output_specs:\n", " if s['empty']:\n", " raise ValueError(f\"'{s['name']}' 열에 빈칸이 {s['empty']}개 있습니다. 채우거나 빈 행을 지워 주세요.\")\n", "for s in output_specs:\n", " if s['kind'] == 'text':\n", " raise ValueError(f\"'{s['name']}' 열은 텍스트라 출력으로 쓸 수 없습니다. KIND_OVERRIDES 에 'categorical' 로 적거나 빼 주세요.\")\n", " if s['kind'] == 'numeric':\n", " if not s['is_numeric']:\n", " raise ValueError(f\"'{s['name']}' 열에 숫자가 아닌 값이 있어 회귀로 쓸 수 없습니다.\")\n", " if s['min'] == s['max']:\n", " raise ValueError(f\"'{s['name']}' 열의 값이 모두 같아 회귀 학습을 할 수 없습니다.\")\n", " else:\n", " if len(s['unique']) < 2:\n", " raise ValueError(f\"'{s['name']}' 열에 종류가 두 개 이상 있어야 분류할 수 있습니다.\")\n", " if len(s['unique']) > MAX_CATEGORIES:\n", " raise ValueError(f\"'{s['name']}' 열의 종류가 {len(s['unique'])}개로 너무 많습니다. (최대 {MAX_CATEGORIES}개)\")\n", "\n", "\n", "def learning_of(s):\n", " if s['role'] == 'input':\n", " return '토큰 시퀀스 (임베딩 + 어텐션)'\n", " if s['role'] == 'output':\n", " return '분류 (Softmax)' if s['kind'] == 'categorical' else '회귀 (Linear)'\n", " return '-'\n", "\n", "\n", "display(pd.DataFrame([{\n", " '열': s['name'],\n", " '역할': ROLE_LABELS[s['role']],\n", " '종류': KIND_LABELS[s['kind']],\n", " '학습': learning_of(s),\n", " '서로 다른 값': len(s['unique']),\n", " '예시': s['sample'][:30],\n", "} for s in stats]))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "id": "cell-5", "metadata": {}, "source": [ "## 3. 글을 숫자로 바꾸기 (토큰 사전과 시퀀스)\n", "\n", "- 글을 `o200k_base` 토큰으로 쪼갭니다. 한 칸에서 200자까지 읽고 나머지는 자릅니다.\n", "- 텍스트 열마다 **사전**을 만듭니다. 학습 행에 나온 토큰을 나온 글 수가 많은 순서로 세우고, 앞에서 `VOCAB_SIZE`개를 담습니다.\n", " 글 하나에만 나온 토큰도 담습니다. 나온 글 수가 같으면 번호가 작은 토큰이 먼저입니다.\n", "- 글을 **사전 자리 번호의 줄(시퀀스)** 로 바꿉니다. 0은 빈칸, 1은 사전에 없는 토큰이고, 사전 토큰은 2번부터입니다.\n", " `SEQ_LEN`개까지만 넣고, 넘는 뒷부분은 버리고 모자라면 0으로 채웁니다. 순서와 반복이 그대로 남습니다.\n", "- 사전은 **학습 행으로만** 만듭니다. 검증 행의 글이 사전을 고르면 검증 loss가 실제보다 좋아 보이기 때문입니다.\n", "- 행 순서는 웹앱과 같은 시드(13)로 섞고, 뒤쪽 20%를 검증에 씁니다. 20행 미만이면 검증 없이 모두 학습합니다.\n", "- 회귀 출력은 설정한 방식으로 정규화하고, 예측할 때 원래 단위로 되돌립니다." ] }, { "cell_type": "code", "id": "cell-6", "metadata": {}, "source": [ "# ==========================================\n", "# 문장 길이 (웹앱 톱니바퀴 기본값 64. 선택지 32, 64, 100, 200)\n", "# ==========================================\n", "SEQ_LEN = 64\n", "\n", "# 시퀀스의 예약 자리 (웹앱과 같다)\n", "PAD_ID = 0 # 빈칸\n", "UNK_ID = 1 # 사전에 없는 토큰\n", "RESERVED = 2 # 사전 토큰은 2번부터\n", "\n", "# ==========================================\n", "# 토크나이저 (웹앱과 같은 o200k_base)\n", "# ==========================================\n", "encoder = tiktoken.get_encoding('o200k_base')\n", "\n", "\n", "def prepare_text(value, max_chars=MAX_TEXT_CHARS):\n", " \"\"\"토큰으로 바꾸기 직전의 글: NFC, 앞뒤 공백 제거, max_chars 자까지 (웹앱처럼 UTF-16 단위로 센다)\"\"\"\n", " text = normalize_text(value)\n", " units = text.encode('utf-16-le')\n", " if len(units) // 2 <= max_chars:\n", " return text\n", " end = max_chars\n", " last = int.from_bytes(units[(end - 1) * 2:end * 2], 'little')\n", " if 0xD800 <= last <= 0xDBFF: # 이모지처럼 두 칸짜리 글자를 반으로 자르지 않는다\n", " end -= 1\n", " return units[:end * 2].decode('utf-16-le')\n", "\n", "\n", "def tokenize(value):\n", " text = prepare_text(value)\n", " if text == '':\n", " return []\n", " # '<|endoftext|>' 같은 특수 토큰 글자도 막지 않고 일반 글자로 쪼갠다 (웹앱과 같다)\n", " return encoder.encode(text, allowed_special=set(), disallowed_special=())\n", "\n", "\n", "def token_label(token_id):\n", " \"\"\"토큰 하나를 읽을 수 있는 이름으로. 한글이 바이트 조각으로 나뉜 토큰은 '조각#번호'\"\"\"\n", " text = encoder.decode([token_id])\n", " if not text or '\\ufffd' in text:\n", " return f'조각#{token_id}'\n", " if text.strip() == '':\n", " if '\\n' in text:\n", " return '(줄바꿈)'\n", " if '\\t' in text:\n", " return '(탭)'\n", " return f'(공백 {len(text)}칸)' if len(text) > 1 else '(공백)'\n", " return text.strip()\n", "\n", "\n", "def distinct_labels(vocab):\n", " \"\"\"사전 안에서 이름이 겹치면 (' 안' 과 '안') 공백이 붙은 쪽에만 표시를 단다 (웹앱과 같다)\"\"\"\n", " base = {t: token_label(t) for t in vocab}\n", " counts = Counter(base.values())\n", " labels, used = {}, set()\n", " for t in vocab:\n", " name = base[t]\n", " if counts[name] > 1:\n", " raw = encoder.decode([t])\n", " if raw[:1].isspace():\n", " name = f'(앞에 공백){name}'\n", " elif raw[-1:].isspace():\n", " name = f'(뒤에 공백){name}'\n", " if name in used: # 그래도 겹치면 (공백 한 칸과 두 칸처럼) 번호를 붙인다\n", " name = f'{name}#{t}'\n", " used.add(name)\n", " labels[t] = name\n", " return labels\n", "\n", "\n", "def shuffle_order(n, seed=13):\n", " \"\"\"웹앱과 같은 시드 고정 셔플 (mulberry32 + Fisher-Yates). 32비트 정수 연산을 그대로 옮겼다\"\"\"\n", " order = list(range(n))\n", " a = seed & 0xFFFFFFFF\n", "\n", " def rand():\n", " nonlocal a\n", " a = (a + 0x6D2B79F5) & 0xFFFFFFFF\n", " t = ((a ^ (a >> 15)) * (a | 1)) & 0xFFFFFFFF\n", " t ^= (t + (((t ^ (t >> 7)) * (t | 61)) & 0xFFFFFFFF)) & 0xFFFFFFFF\n", " return ((t ^ (t >> 14)) & 0xFFFFFFFF) / 4294967296\n", "\n", " for i in range(n - 1, 0, -1):\n", " j = int(rand() * (i + 1))\n", " order[i], order[j] = order[j], order[i]\n", " return order\n", "\n", "\n", "# ==========================================\n", "# 학습 행과 검증 행 나누기\n", "# ==========================================\n", "n_rows = len(rows)\n", "VALIDATION_SPLIT = 0.2 if n_rows >= 20 else 0.0\n", "order = shuffle_order(n_rows)\n", "train_count = math.floor(n_rows * (1 - VALIDATION_SPLIT)) # 케라스도 이 수만큼 앞에서 학습에 쓴다\n", "vocab_rows = [rows[i] for i in order[:train_count]]\n", "\n", "\n", "# ==========================================\n", "# 입력: 텍스트 열마다 사전 만들기\n", "# ==========================================\n", "def build_text_encoder(spec):\n", " doc_freq = Counter()\n", " for row in vocab_rows:\n", " doc_freq.update(set(tokenize(row[spec['index']]))) # 글 하나에 여러 번 나와도 1로 센다\n", " ranked = sorted(doc_freq.items(), key=lambda item: (-item[1], item[0]))\n", " vocab = [token_id for token_id, _ in ranked[:VOCAB_SIZE]]\n", " if not vocab:\n", " raise ValueError(f\"'{spec['name']}' 열에서 토큰을 찾지 못했습니다.\")\n", " sample = next((prepare_text(row[spec['index']]) for row in rows if prepare_text(row[spec['index']])), '')\n", " return {\n", " 'index': spec['index'],\n", " 'name': spec['name'],\n", " 'vocab': vocab,\n", " 'position': {token_id: i + RESERVED for i, token_id in enumerate(vocab)}, # 토큰 -> 시퀀스 자리 번호\n", " 'labels': distinct_labels(vocab),\n", " 'doc_freq': doc_freq,\n", " 'sample': sample,\n", " }\n", "\n", "\n", "def encode_sequence(enc, value, seq_len=SEQ_LEN):\n", " \"\"\"글 하나를 사전 자리 번호의 줄로. 사전에 없는 토큰은 UNK, 넘는 뒷부분은 버리고 모자라면 PAD 로 채운다\"\"\"\n", " token_ids = tokenize(value)\n", " kept = token_ids[:seq_len]\n", " ids = [enc['position'].get(t, UNK_ID) for t in kept]\n", " ids += [PAD_ID] * (seq_len - len(ids))\n", " known = len({t for t in kept if t in enc['position']})\n", " return np.array(ids, dtype='int32'), known, len(token_ids) - len(kept)\n", "\n", "\n", "# ==========================================\n", "# 출력: 분류는 원핫, 회귀는 정규화한 값 하나\n", "# ==========================================\n", "def build_output_encoder(spec):\n", " if spec['kind'] == 'categorical':\n", " return {'index': spec['index'], 'name': spec['name'], 'kind': 'categorical', 'categories': list(spec['unique'])}\n", " return {\n", " 'index': spec['index'], 'name': spec['name'], 'kind': 'numeric', 'normalization': NORMALIZATION,\n", " 'min': spec['min'], 'max': spec['max'], 'mean': spec['mean'], 'std': spec['std'],\n", " }\n", "\n", "\n", "def normalize(value, enc):\n", " lo, hi, method = enc['min'], enc['max'], enc['normalization']\n", " if method == 'none':\n", " return value\n", " if method == 'minMax01':\n", " return 0 if hi == lo else (value - lo) / (hi - lo)\n", " if method == 'zScore':\n", " return 0 if enc['std'] == 0 else (value - enc['mean']) / enc['std']\n", " return 0 if hi == lo else (value - lo) / (hi - lo) * 2 - 1\n", "\n", "\n", "def denormalize(value, enc):\n", " lo, hi, method = enc['min'], enc['max'], enc['normalization']\n", " if method == 'none':\n", " return value\n", " if method == 'minMax01':\n", " return lo if hi == lo else value * (hi - lo) + lo\n", " if method == 'zScore':\n", " return enc['mean'] if enc['std'] == 0 else value * enc['std'] + enc['mean']\n", " return lo if hi == lo else (value + 1) / 2 * (hi - lo) + lo\n", "\n", "\n", "def encode_output(enc, raw):\n", " text = to_cell_string(raw)\n", " if enc['kind'] == 'numeric':\n", " n = parse_number(text)\n", " return None if n is None else [normalize(n, enc)]\n", " categories = enc['categories']\n", " index = categories.index(text) if text in categories else -1\n", " if index == -1 and parse_number(text) is not None:\n", " key = js_number(parse_number(text)) # 숫자 카테고리는 1 과 1.0 을 같은 칸으로 본다\n", " index = next((k for k, c in enumerate(categories) if parse_number(c) is not None and js_number(parse_number(c)) == key), -1)\n", " if index == -1:\n", " return None\n", " vector = [0.0] * len(categories)\n", " vector[index] = 1.0\n", " return vector\n", "\n", "\n", "input_encoders = [build_text_encoder(s) for s in input_specs]\n", "output_encoders = [build_output_encoder(s) for s in output_specs]\n", "\n", "# 텍스트 열마다 [행, SEQ_LEN] 의 정수 행렬 (모델 입력 순서), 그리고 자리 번호 [행, SEQ_LEN]\n", "X_parts = [np.stack([encode_sequence(e, row[e['index']])[0] for row in rows]) for e in input_encoders]\n", "positions = np.tile(np.arange(SEQ_LEN, dtype='int32'), (n_rows, 1))\n", "Y_heads = []\n", "for enc in output_encoders:\n", " values = []\n", " for r, row in enumerate(rows):\n", " vector = encode_output(enc, row[enc['index']])\n", " if vector is None:\n", " raise ValueError(f\"{r + 1}번째 행의 '{enc['name']}' 값을 읽을 수 없습니다.\")\n", " values.append(vector)\n", " Y_heads.append(np.array(values, dtype='float32'))\n", "\n", "# 사전을 고를 때와 같은 순서로 섞는다. 케라스의 validation_split 은 뒤쪽 행을 그대로 떼어 검증에 쓴다\n", "X_parts = [x[order] for x in X_parts]\n", "Y_heads = [y[order] for y in Y_heads]\n", "\n", "for enc, x in zip(input_encoders, X_parts):\n", " cut = sum(1 for row in rows if len(tokenize(row[enc['index']])) > SEQ_LEN)\n", " print(f\"[{enc['name']}] 사전 {len(enc['vocab'])}개 (설정 {VOCAB_SIZE}개, 학습 행에 나온 토큰 {len(enc['doc_freq'])}개), \"\n", " f\"문장 길이 {SEQ_LEN}토큰을 넘는 글 {cut}개\")\n", " top = ', '.join(f\"{enc['labels'][t]}({enc['doc_freq'][t]})\" for t in enc['vocab'][:20])\n", " print(f' 많이 나온 토큰 20개 (나온 글 수): {top}')\n", "for enc in output_encoders:\n", " if enc['kind'] == 'categorical':\n", " print(f\"[{enc['name']}] 분류 {len(enc['categories'])}종류: {enc['categories']}\")\n", " else:\n", " print(f\"[{enc['name']}] 회귀, 값 {enc['min']:g} ~ {enc['max']:g}, 정규화 {enc['normalization']}\")\n", "print(f'\\n입력 {len(X_parts)}열 x {SEQ_LEN}토큰, 출력 차원 {sum(y.shape[1] for y in Y_heads)}, 학습 {train_count}행, 검증 {n_rows - train_count}행')" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "id": "cell-7", "metadata": {}, "source": [ "## 4. 모델 학습 (임베딩 + 어텐션 블록)\n", "\n", "웹앱과 같은 구조의 작은 트랜스포머입니다.\n", "\n", "- 텍스트 열마다: 토큰 임베딩 + 자리 임베딩 (어텐션은 순서를 모르므로 자리 벡터로 알려 줍니다)\n", " → Transformer 인코더 블록 `NUM_LAYERS`개 (멀티헤드 어텐션 → 잔차 + LayerNorm → FFN(2배) → 잔차 + LayerNorm)\n", " → 자리 평균(풀링)\n", "- 열이 둘이면 두 벡터를 이어 붙입니다 → Dropout(0.2) → Dense(32, ReLU) → 출력 열마다 헤드 하나.\n", " 분류는 Softmax + categorical crossentropy, 회귀는 Linear + MSE이고, 전체 loss는 헤드 loss의 합입니다.\n", "- 웹앱이 쓰는 tfjs에는 `MultiHeadAttention` 레이어가 없어서 dense와 dot, softmax로 직접 조립하지만, 케라스에는 있으니 그대로 씁니다. 계산하는 내용은 같습니다.\n", "- Adam, 학습률 0.001, 배치 16, 40번 학습에 조기 종료(patience 10)를 켭니다 (웹앱 기본값).\n", " 글이 많은 표는 4~9번째 에포크쯤에서 검증 loss가 가장 낮고 그 뒤 학습 데이터를 외우기 시작하며, 100행쯤의 작은 표는 30~40번은 돌아야 배웁니다.\n", " patience를 5로 두면 작은 표에서 검증 loss가 초반에 흔들리다 배우기 전에 멈추는 일이 있어 10으로 둡니다.\n", "- 조기 종료는 검증 loss가 `PATIENCE`번 나아지지 않을 때 멈추고, 가장 좋았던 에포크의 가중치로 되돌립니다.\n", " 케라스는 끝까지 돌아도 가장 좋았던 에포크로 되돌리고, 웹앱은 일찍 멈췄을 때만 되돌립니다. 20행 미만이면 검증 데이터가 없어 끝까지 돕니다." ] }, { "cell_type": "code", "id": "cell-8", "metadata": {}, "source": [ "# ==========================================\n", "# 학습 설정 (웹앱 톱니바퀴 기본값)\n", "# ==========================================\n", "EMBED_DIM = 32 # 임베딩 차원 (선택지 16, 32, 64)\n", "NUM_HEADS = 2 # 어텐션 헤드 수 (선택지 1, 2, 4)\n", "NUM_LAYERS = 1 # 어텐션 블록 수 (선택지 1, 2, 3)\n", "HIDDEN_UNITS = 32 # 풀링 뒤 은닉층 노드 수 (웹앱은 고정)\n", "EPOCHS = 40\n", "BATCH_SIZE = 16\n", "LEARNING_RATE = 0.001\n", "EARLY_STOPPING = True # 검증 데이터가 있을 때만 걸린다 (웹앱 기본값도 켬)\n", "PATIENCE = 10\n", "\n", "head_dim = max(1, EMBED_DIM // NUM_HEADS)\n", "\n", "\n", "def transformer_block(x, tag):\n", " \"\"\"Transformer 인코더 블록 (웹앱의 transformerEncoderBlock 과 같다)\"\"\"\n", " attn = keras.layers.MultiHeadAttention(num_heads=NUM_HEADS, key_dim=head_dim, name=f'{tag}_attn')(x, x)\n", " x = keras.layers.LayerNormalization(name=f'{tag}_norm1')(keras.layers.Add(name=f'{tag}_add1')([x, attn]))\n", " ff = keras.layers.Dense(EMBED_DIM * 2, activation='relu', name=f'{tag}_ff1')(x)\n", " ff = keras.layers.Dense(EMBED_DIM, name=f'{tag}_ff2')(ff)\n", " return keras.layers.LayerNormalization(name=f'{tag}_norm2')(keras.layers.Add(name=f'{tag}_add2')([x, ff]))\n", "\n", "\n", "# ==========================================\n", "# 모델: 텍스트 열마다 임베딩 + 블록 + 풀링, 이어 붙인 뒤 은닉층과 헤드\n", "# ==========================================\n", "position_input = keras.Input(shape=(SEQ_LEN,), dtype='int32', name='positions')\n", "token_inputs, pooled = [], []\n", "for i, enc in enumerate(input_encoders):\n", " tokens = keras.Input(shape=(SEQ_LEN,), dtype='int32', name=f'tokens_{i}')\n", " token_inputs.append(tokens)\n", " token_embedding = keras.layers.Embedding(len(enc['vocab']) + RESERVED, EMBED_DIM, name=f't{i}_token_embedding')(tokens)\n", " position_embedding = keras.layers.Embedding(SEQ_LEN, EMBED_DIM, name=f't{i}_position_embedding')(position_input)\n", " x = keras.layers.Add(name=f't{i}_embed_sum')([token_embedding, position_embedding])\n", " for k in range(NUM_LAYERS):\n", " x = transformer_block(x, f't{i}_block{k}')\n", " pooled.append(keras.layers.GlobalAveragePooling1D(name=f't{i}_pool')(x))\n", "\n", "joined = pooled[0] if len(pooled) == 1 else keras.layers.Concatenate(name='pooled')(pooled)\n", "hidden = keras.layers.Dense(HIDDEN_UNITS, activation='relu', name='hidden')(keras.layers.Dropout(0.2, name='dropout')(joined))\n", "\n", "outputs, losses = [], []\n", "for k, enc in enumerate(output_encoders):\n", " categorical = enc['kind'] == 'categorical'\n", " outputs.append(keras.layers.Dense(\n", " len(enc['categories']) if categorical else 1,\n", " activation='softmax' if categorical else 'linear',\n", " name=f'head_{k}',\n", " )(hidden))\n", " losses.append('categorical_crossentropy' if categorical else 'mean_squared_error')\n", "\n", "single_head = len(outputs) == 1\n", "model = keras.Model(token_inputs + [position_input], outputs[0] if single_head else outputs)\n", "model.compile(\n", " optimizer=keras.optimizers.Adam(learning_rate=LEARNING_RATE),\n", " loss=losses[0] if single_head else losses,\n", ")\n", "model.summary()\n", "for k, enc in enumerate(output_encoders):\n", " print(f\"head_{k}: {enc['name']} ({'분류' if enc['kind'] == 'categorical' else '회귀'})\")\n", "\n", "callbacks = []\n", "if EARLY_STOPPING and VALIDATION_SPLIT > 0:\n", " callbacks.append(keras.callbacks.EarlyStopping(\n", " monitor='val_loss', patience=PATIENCE, restore_best_weights=True, verbose=1,\n", " ))\n", "\n", "# ==========================================\n", "# 학습\n", "# ==========================================\n", "history = model.fit(\n", " X_parts + [positions],\n", " Y_heads[0] if single_head else Y_heads,\n", " epochs=EPOCHS,\n", " batch_size=BATCH_SIZE,\n", " validation_split=VALIDATION_SPLIT,\n", " shuffle=True,\n", " callbacks=callbacks,\n", " verbose=2,\n", ")" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "id": "cell-9", "metadata": {}, "source": [ "## 5. 결과 확인 및 예측\n", "\n", "학습 손실 그래프를 그리고, 검증 데이터로 출력 열마다 정확도(분류)나 평균 오차(회귀)를 봅니다.\n", "마지막 칸에서 직접 쓴 글로 예측해 보세요. 대괄호로 묶인 토큰이 사전에 있는 토큰이고, 문장 길이를 넘은 토큰은 `(잘림)` 뒤에 씁니다." ] }, { "cell_type": "code", "id": "cell-10", "metadata": {}, "source": [ "# ==========================================\n", "# 모델 결과 확인 (그래프)\n", "# ==========================================\n", "plt.plot(history.history['loss'])\n", "legend = ['train']\n", "if 'val_loss' in history.history:\n", " plt.plot(history.history['val_loss'])\n", " legend.append('val')\n", "plt.title('model loss')\n", "plt.ylabel('loss')\n", "plt.xlabel('epoch')\n", "plt.legend(legend, loc='upper right')\n", "plt.show()\n", "\n", "# ==========================================\n", "# 출력 열마다 성능 보기\n", "# ==========================================\n", "if VALIDATION_SPLIT > 0:\n", " X_check = [x[train_count:] for x in X_parts] + [positions[train_count:]]\n", " Y_check = [y[train_count:] for y in Y_heads]\n", " print(f'검증 데이터 {n_rows - train_count}행 (학습에 쓰지 않은 행)\\n')\n", "else:\n", " X_check, Y_check = X_parts + [positions], Y_heads\n", " print(f'학습 데이터 {n_rows}행 (행이 20개 미만이라 검증 데이터가 없습니다)\\n')\n", "\n", "pred = model.predict(X_check, verbose=0)\n", "if single_head:\n", " pred = [pred]\n", "for enc, p, y in zip(output_encoders, pred, Y_check):\n", " if enc['kind'] == 'categorical':\n", " accuracy = float(np.mean(np.argmax(p, axis=1) == np.argmax(y, axis=1)))\n", " print(f\"[{enc['name']}] 분류 정확도: {accuracy * 100:.1f}%\")\n", " else:\n", " predicted = np.array([denormalize(float(v), enc) for v in p[:, 0]])\n", " actual = np.array([denormalize(float(v), enc) for v in y[:, 0]])\n", " print(f\"[{enc['name']}] 회귀 평균 오차(MAE): {np.mean(np.abs(predicted - actual)):.3f}\")" ], "execution_count": null, "outputs": [] }, { "cell_type": "code", "id": "cell-11", "metadata": {}, "source": [ "# ==========================================\n", "# 직접 쓴 글로 예측하기\n", "# ==========================================\n", "def predict(texts):\n", " \"\"\"texts: {입력 열 이름: 글}. 웹앱 예측 화면처럼 출력 열마다 결과를 보여 준다\"\"\"\n", " parts = []\n", " for enc in input_encoders:\n", " text = texts.get(enc['name'], '')\n", " ids, known, cut = encode_sequence(enc, text)\n", " parts.append(ids[None, :])\n", " token_ids = tokenize(text)\n", " shown = [f'[{token_label(t)}]' if t in enc['position'] else token_label(t) for t in token_ids[:SEQ_LEN]]\n", " if cut:\n", " shown.append('(잘림) ' + ' '.join(token_label(t) for t in token_ids[SEQ_LEN:]))\n", " print(f\"[{enc['name']}] {prepare_text(text)}\")\n", " print(f' 토큰 {len(token_ids)}개, 사전에 있는 토큰 {known}개: {\" \".join(shown)}')\n", " if prepare_text(text) and known == 0:\n", " print(' 주의: 사전에 있는 토큰이 없어 모델이 이 글에서 아는 단어가 없습니다.')\n", "\n", " result = model.predict(parts + [positions[:1]], verbose=0)\n", " if single_head:\n", " result = [result]\n", " print()\n", " for enc, p in zip(output_encoders, result):\n", " p = p[0]\n", " if enc['kind'] == 'categorical':\n", " ranked = sorted(zip(enc['categories'], p), key=lambda item: -item[1])\n", " print(f\"[{enc['name']}] \" + ', '.join(f'{label} {prob * 100:.1f}%' for label, prob in ranked))\n", " else:\n", " print(f\"[{enc['name']}] {denormalize(float(p[0]), enc):.2f}\")\n", "\n", "\n", "# 기본값은 표의 첫 글입니다. 원하는 글로 바꿔 보세요.\n", "my_texts = {enc['name']: enc['sample'] for enc in input_encoders}\n", "# my_texts = {'리뷰': '이 영화 정말 재미있어요!'}\n", "predict(my_texts)" ], "execution_count": null, "outputs": [] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "name": "python", "version": "3.10.0" } }, "nbformat": 4, "nbformat_minor": 5 }