Spaces:
Sleeping
Sleeping
| # -*- coding: utf-8 -*- | |
| """ | |
| 자기인식 지원 앱 — 지인 데이터 수집판 (P-25-0322) v0.3-collect | |
| ================================================================================ | |
| 데이터 플라이휠 + 영구 저장 + 안전 고지. | |
| · 측정: KoSimCSE 임베딩으로 가치·인지·감정 4벡터 독립 측정 | |
| · 라벨: 측정값을 '보여주기 전' 독립 자기보고 → 깨끗한 검증쌍 | |
| · 저장: 데이터를 두 스트림으로 분리·구조화하여 HF Dataset에 영구 저장 | |
| - sessions.jsonl : 세션·온보딩 기준선 (1행/세션) | |
| - labels.jsonl : 라벨 검증쌍 (1행/라벨 턴) | |
| · 안전: 위기 안내 + 동의 고지(동의 시에만 발화 저장) | |
| 운영(영구 저장 켜기): | |
| Space Secret 에 HF_TOKEN(쓰기 권한) 과 HF_DATASET_REPO("아이디/데이터셋명") 설정. | |
| 미설정 시 로컬 파일로만 저장(Colab은 세션 보존, Spaces는 재시작 시 초기화). | |
| ⚠️ 실험 도구이며 상담·치료가 아님. 발화는 민감정보 — 동의·익명화 전제. 정신건강 위기는 전문기관으로. | |
| ================================================================================ | |
| """ | |
| import numpy as np | |
| from numpy.linalg import norm | |
| import os, json, time, uuid, datetime | |
| from contextlib import nullcontext | |
| VERSION = "0.4.15-consent-save" # 데이터 저장 문제 수정: 동의 체크박스 기본 ON(value=True, 깜빡 방지)+안내문 '원치 않으면 해제'로 변경. do_chat이 미동의 시 채팅 아래 저장 경고 표시. 입력창 아래 '💾 내 데이터 저장 상태 확인' 버튼 추가(저장 ON/OFF·참가자코드·턴수 안내). # (4)Gemini thinking 끔(thinkingBudget=0)+출력상한500 → 토큰·비용·지연 감소. (2)EVA/EAR 자기보고 발화앵커 구체화. (3)단일발화: 짧은발화 길이가중 억제(6자미만 제외)+EVA/EAR 평활(나침반 안흔들림), reveal은 원시값 유지. # 온보딩 Q1~Q4 강도별 4지선다(2→4, 기준선 정밀화). EAR 고각성 라벨에 분노 추가(매우 들뜸·긴장·분노). # 측정 나침반(심플 3축 막대 EVA·EAR·VAL, REI제외; EVA강도신뢰·EAR/VAL방향위주; profile_md→gr.HTML). 성찰질문형 넛지(주제당1회·재청시재제공·조언금지·안전우선). # 자기보고 4지선다(중간 제거, EVA/EAR/VAL/REI+PILOT). VAL 질문 직관화. build_prompt: 약간 긍정적 존댓말+닉네임 호칭+관심 말투, 3턴부터 사용자 유형별 대화 전략. # 체크박스 첫 선택 지연 수정: CheckboxGroup을 빈 choices 대신 공통항목으로 초기화(프라이밍) → 첫 선택부터 즉시 렌더. on_domain은 도메인 선택 시 도메인 항목 추가. # 관심사 체크박스: on_domain이 gr.update 대신 새 CheckboxGroup 인스턴스 반환(6.0에서 빈 choices 갱신 확실). + 진단 로그(이벤트 발동·반환 개수). # 6.0 대응(트레이스백 기반): content가 리스트로 와 .strip() 크래시(=원래 2턴 오류) → content 정규화 헬퍼. Chatbot type 버전조건부(6.0은 생략). README는 버전 핀 제거(4.44.1은 Py3.13 audioop 비호환 → 작동하던 6.x 사용). | |
| # ============================== 설정 ============================== | |
| ENCODER = "BM-K/KoSimCSE-roberta-multitask" # 또는 "BAAI/bge-m3" | |
| USE_SENTENCE_TRANSFORMERS = ENCODER.startswith("BAAI") | |
| LLM_MODE = "gemini" # "template" | "gemini" | "local" | |
| W_BASE = 6.0 | |
| GATE = 0.5 # (구) 하드 신뢰 게이트 — 누적에선 소프트 가중(|coord|)으로 대체됨 | |
| LEN_REF = 150 # 길이 정규화 기준(자): 초과 발화는 가중치를 완만히 감소(긴 글의 분석 편향 완화) | |
| MIN_MEASURE_LEN = 6 # 이 미만(자)은 측정 신뢰 불가 → 누적·평활에서 제외(가중 0). 데이터: 짧은 발화 어휘노이즈 | |
| RELIABLE_LEN = 18 # 이 이상이면 측정 신뢰(full 가중). MIN~RELIABLE은 약하게 합성(짧을수록 약하게) | |
| FREE_WEIGHT = 0.5 | |
| DATA_DIR = "data" # 수집 데이터 폴더(분리 저장) | |
| HF_DATASET_REPO = os.environ.get("HF_DATASET_REPO", "") # 예: "myid/selfaware-data" | |
| # Gemini 토큰 단가(USD per 1M tokens): (입력, 출력). 출력엔 thinking 토큰 포함해 합산. | |
| GEMINI_PRICES = { | |
| "gemini-2.5-flash": (0.30, 2.50), | |
| "gemini-2.5-flash-lite": (0.10, 0.40), | |
| "gemini-flash-latest": (0.30, 2.50), | |
| } | |
| def _gemini_cost(model, tin, tout): | |
| pin, pout = GEMINI_PRICES.get(model, (0.30, 2.50)) | |
| return tin / 1e6 * pin + tout / 1e6 * pout | |
| # ============================== 축 정의 문항(한국어 1인칭) ============================== | |
| CONSTRUCTS = { | |
| "VAL": {"label": ("자율 지향", "순응 지향"), "pos": [ | |
| "나는 내 삶을 어떻게 살지 스스로 결정하는 것을 좋아한다.", | |
| "새롭고 독창적인 생각을 떠올리는 것이 나에게 중요하다.", | |
| "나는 무엇이든 내 방식대로 해결하는 편이다.", | |
| "내 목표를 스스로 자유롭게 선택하는 것이 중요하다.", | |
| "남에게 묻기보다 내 판단을 믿고 따르는 편이다.", | |
| "나는 호기심을 따라 새로운 것을 탐험하는 것을 가치 있게 여긴다.", | |
| "누가 알려주기보다 세상을 스스로 이해하고 싶다.", | |
| "나는 독립적으로 일을 해내는 것에 자부심을 느낀다.", | |
| "남을 따라 하기보다 내 방식을 새로 만드는 편이 좋다.", | |
| "내 인생의 방향을 스스로 정하는 것이 무척 중요하다.", | |
| "나는 창의적으로, 내 식대로 시도하는 것을 좋아한다.", | |
| "나에게 무엇이 옳은지 스스로 판단할 수 있다고 믿는다.", | |
| ], "neg": [ | |
| "나는 사람은 규칙을 잘 지켜야 한다고 생각한다.", | |
| "시키는 일을 잘 따르는 것이 나에게 중요하다.", | |
| "나는 예의 바르게 행동하고 잘못된 일을 하지 않으려 애쓴다.", | |
| "물려받은 전통과 관습을 지키는 것이 나에게 중요하다.", | |
| "나는 권위와 윗사람을 존중해야 한다고 믿는다.", | |
| "전통을 존중하는 것을 중요하게 여긴다.", | |
| "나는 튀기보다 집단에 어울리는 편을 택한다.", | |
| "정해진 방식을 함부로 의심하지 않는 편이 낫다고 생각한다.", | |
| "나에게 기대되는 바를 지키는 것이 옳게 느껴진다.", | |
| "나는 순종적이고 믿음직한 사람이 되는 것을 가치 있게 여긴다.", | |
| "사회 규범에는 이유가 있으니 지켜야 한다고 믿는다.", | |
| "나에게는 새로움보다 안정과 질서를 지키는 것이 더 중요하다.", | |
| # 선택-동사 순응 보강 — 진단 결과 '능동 선택형 순응'(남들 따라 고르기)이 | |
| # 자율로 오측정되던 문제 완화. 독립 데이터 검증 AUC 0.90→0.95. | |
| "남들이 많이 사는 물건을 골라서 산다.", | |
| "유행하는 쪽을 보고 그대로 선택한다.", | |
| "다수가 정한 방향에 내 결정을 맞춘다.", | |
| "주변에서 고르는 것을 보고 똑같이 고른다.", | |
| "나도 남들 하는 대로 무난한 쪽을 택한다.", | |
| "사람들이 좋다고 하는 것을 골라 산다.", | |
| "내 취향보다 대세를 따라 고르는 편이다.", | |
| "남들 눈을 의식해서 선택을 정한다.", | |
| ]}, | |
| "REI": {"label": ("분석적", "직관적"), "pos": [ | |
| "나는 깊이 생각해야 하는 문제를 즐긴다.", | |
| "행동하기 전에 상황을 논리적으로 분석하는 것을 좋아한다.", | |
| "나는 단계를 밟아 차근차근 따져보는 편이다.", | |
| "복잡한 문제를 깊이 고민하는 것이 만족스럽다.", | |
| "나는 결론을 내릴 때 논리와 근거에 의존한다.", | |
| "문제를 부분으로 나누어 분석하는 것을 좋아한다.", | |
| "어려운 지적 과제를 풀어내는 것을 즐긴다.", | |
| "결정할 때 감정보다 이성을 더 믿는다.", | |
| "나는 결정 전에 장단점을 신중히 따져본다.", | |
| "추상적이고 분석적인 사고가 즐겁다.", | |
| "분명한 근거로 뒷받침된 결론을 선호한다.", | |
| "무언가의 논리를 풀어내면 만족스럽다.", | |
| "느낌이 좋았지만 가격을 비교하고 구매했다.", | |
| "느낌보다 데이터를 우선해 전공을 정했다.", | |
| "감으로 판단하지 않고 사실을 확인했다.", | |
| "직감이 들어도 먼저 수치를 확인한다.", | |
| "막연한 느낌을 숫자로 바꿔 확인했다.", | |
| "직감이 나빴지만 안전 데이터를 검증했다.", | |
| "감보다 근거를 보고 이직을 결정했다.", | |
| "느낌에 기대지 않고 논문을 분석했다.", | |
| "느낌이 와도 계약 조건을 따져본다.", | |
| "직감을 가설로 세워 실험을 설계했다.", | |
| ], "neg": [ | |
| "나는 보통 직감에 따라 행동한다.", | |
| "나는 종종 무엇이 옳은지 그냥 느낌으로 안다.", | |
| "분석보다 첫인상에 더 의존하는 편이다.", | |
| "나는 많은 결정을 느낌에 따라 내린다.", | |
| "설명할 수 없어도 내 직관을 믿는다.", | |
| "내 예감은 대체로 들어맞는다.", | |
| "나는 그 순간 옳게 느껴지는 대로 하는 편이다.", | |
| "나는 상황을 본능으로 읽는다.", | |
| "나는 결정을 느낌으로 더듬어 가는 편이 좋다.", | |
| "나는 감정과 인상에 따라 선택하곤 한다.", | |
| "따져보지 않아도 옳다는 걸 아는 경우가 많다.", | |
| "나는 신중한 분석보다 직관에 더 의존한다.", | |
| ]}, | |
| "EVA": {"label": ("긍정", "부정"), "pos": [ | |
| "지금 나는 즐겁고 기분이 좋다.", "따뜻한 만족감이 느껴진다.", | |
| "오늘 나는 희망차고 밝다.", "나는 흐뭇하고 만족스럽다.", | |
| "나는 고맙고 뿌듯하다.", "좋고 기분 좋은 느낌이 함께한다.", | |
| "나는 행복하고 마음이 가볍다.", "나는 흡족하고 긍정적이다.", | |
| "지금 내 안에 기쁨이 있다.", "나는 즐겁고 편안하다.", | |
| "나는 감사하고 마음이 따뜻하다.", "밝은 안녕감이 느껴진다.", | |
| ], "neg": [ | |
| "지금 나는 시무룩하고 우울하다.", "무거운 슬픔이 느껴진다.", | |
| "오늘 나는 낙담하고 서럽다.", "나는 언짢고 불만스럽다.", | |
| "나는 비참하고 의기소침하다.", "나쁘고 불쾌한 느낌이 함께한다.", | |
| "나는 침울하고 풀이 죽었다.", "나는 속상하고 부정적이다.", | |
| "지금 내 안에 슬픔이 있다.", "나는 슬프고 마음이 불편하다.", | |
| "나는 씁쓸하고 마음이 차갑다.", "어두운 괴로움이 느껴진다.", | |
| ]}, | |
| "EAR": {"label": ("고각성", "저각성"), "pos": [ | |
| "나는 활력이 넘치고 또렷이 깨어 있다.", "나는 들뜨고 잔뜩 흥분돼 있다.", | |
| "내 몸이 활성화되고 긴장돼 있다.", "나는 격렬하고 잔뜩 달아올라 있다.", | |
| "나는 안절부절못하며 에너지가 넘친다.", "심장이 빠르게 뛰는 것 같다.", | |
| "나는 긴장되고 크게 각성돼 있다.", "나는 안달이 나고 자극받아 있다.", | |
| "나는 한껏 충전돼 들떠 있다.", "나는 곤두서고 신경이 팽팽하다.", | |
| "몸에 각성이 솟구치는 느낌이다.", "나는 또렷하고 활기차게 깨어 있다.", | |
| ], "neg": [ | |
| "나는 차분하고 고요하다.", "나는 졸리고 느긋하다.", | |
| "나는 조용하고 가라앉아 있다.", "나는 나른하고 멍하다.", | |
| "내 에너지가 낮고 느리게 느껴진다.", "나는 누그러지고 서두르지 않는다.", | |
| "나는 평온하고 쉬고 있다.", "나는 축 늘어지고 무겁다.", | |
| "나는 잔잔하고 조용하다.", "나는 졸음이 오고 잦아든다.", | |
| "깊은 고요함이 느껴진다.", "나는 느긋하고 반쯤 잠든 듯하다.", | |
| ]}, | |
| } | |
| # ============================== 온보딩(선택지 = 구성-순수 문장) ============================== | |
| ONBOARD = [ | |
| {"construct": "VAL", "q": "Q1. 중요한 결정을 앞두고 가까운 사람이 당신과 '다른' 의견을 강하게 말합니다. 당신은?", "choices": [ | |
| ("끝까지 내 판단대로 한다", "나는 주변이 강하게 반대해도 끝까지 내 판단대로 결정한다."), | |
| ("대체로 내 생각을 따른다", "나는 주변 의견을 듣더라도 대체로 내 생각을 따라 결정하는 편이다."), | |
| ("주변 의견을 꽤 반영한다", "나는 결정할 때 주변의 의견을 꽤 반영해 조정하는 편이다."), | |
| ("주변 뜻에 맞춰 따른다", "나는 중요한 결정에서 주변의 뜻에 맞춰 따르는 편이다."), | |
| ]}, | |
| {"construct": "VAL", "q": "Q2. 익숙한 안정을 포기해야 새로운 기회를 잡을 수 있습니다. 당신은?", "choices": [ | |
| ("과감히 새로 도전한다", "나는 안정을 포기하더라도 과감히 내 방식대로 새롭게 도전한다."), | |
| ("어느 정도 시도해본다", "나는 안정을 조금 양보하더라도 새로운 기회를 어느 정도 시도하는 편이다."), | |
| ("대체로 안정을 지킨다", "나는 새로운 기회보다 대체로 안정과 익숙함을 지키는 편이다."), | |
| ("안정과 조화를 지킨다", "나는 새로운 기회보다 안정과 사람들과의 조화를 확실히 지킨다."), | |
| ]}, | |
| {"construct": "REI", "q": "Q3. 처음 보는 복잡한 문제를 풀어야 합니다. 당신의 첫 반응은?", "choices": [ | |
| ("철저히 분석부터 한다", "나는 복잡한 문제를 만나면 자료를 모아 철저히 논리적으로 분석한다."), | |
| ("우선 근거를 따져본다", "나는 복잡한 문제를 만나면 우선 근거를 따져보며 차근차근 접근하는 편이다."), | |
| ("대체로 직감을 따른다", "나는 복잡한 문제를 만나면 대체로 전체 느낌과 직감을 따르는 편이다."), | |
| ("바로 직감으로 간다", "나는 복잡한 문제를 만나면 바로 전체 느낌을 잡아 직감으로 판단한다."), | |
| ]}, | |
| {"construct": "REI", "q": "Q4. 큰 선택에서 '근거'와 '직감'이 서로 다른 방향을 가리킵니다. 당신은?", "choices": [ | |
| ("확실히 근거를 따른다", "나는 근거와 데이터가 가리키는 방향을 확실히 따라 결정한다."), | |
| ("대체로 근거를 따른다", "나는 근거와 직감이 부딪치면 대체로 근거 쪽을 따르는 편이다."), | |
| ("대체로 직감을 따른다", "나는 근거와 직감이 부딪치면 대체로 직감 쪽을 따르는 편이다."), | |
| ("확실히 직감을 따른다", "나는 직감과 느낌이 가리키는 방향을 확실히 따라 결정한다."), | |
| ]}, | |
| ] | |
| # ============================== 라벨 수집 설계 ============================== | |
| LABEL_SCHEMES = { | |
| "EVA": {"q": "잠깐 — 방금 그 말을 할 때, 기분은 어느 쪽에 더 가까웠나요?", | |
| "opts": [("매우 슬픔", -2), ("약간 슬픔", -1), ("약간 즐거움", 1), ("매우 즐거움", 2)], | |
| "poles": ("즐거운", "슬픈"), "reveal_lead": "기분을 ‘{}’ 쪽으로"}, | |
| "EAR": {"q": "잠깐 — 방금 그 말을 할 때, 마음의 에너지는 어느 쪽이었나요?", | |
| "opts": [("매우 차분·처짐", -2), ("약간 가라앉음", -1), ("약간 들뜸", 1), ("매우 들뜸·긴장·분노", 2)], | |
| "poles": ("들뜬·긴장된", "차분한"), "reveal_lead": "에너지를 ‘{}’ 쪽으로"}, | |
| # VAL 직관화: '어땠나요'(추상) → '방금 한 그 일이 누구의 뜻에 가까웠나'(구체). | |
| "VAL": {"q": "잠깐 — 방금 말씀하신 그 일은 누구의 뜻에 더 가까웠나요? (내 뜻대로 ↔ 남·상황에 맞춰)", | |
| "opts": [("전적으로 남·상황에 맞춤", -2), ("주로 맞춤", -1), ("주로 내 뜻대로", 1), ("전적으로 내 뜻대로", 2)], | |
| "poles": ("주도적인", "맞춰주는"), "reveal_lead": "태도를 ‘{}’ 쪽으로"}, | |
| # REI: '말투'(문체) 기준. 측정이 분석↔직관 어휘/문체를 보므로 자기보고도 문체를 묻도록 정렬. | |
| "REI": {"q": "잠깐 — 방금 ‘말투’는 어느 쪽에 더 가깝나요?", | |
| "opts": [("매우 직감적 말투", -2), ("약간 직감적", -1), ("약간 분석적", 1), ("매우 분석적 말투", 2)], | |
| "poles": ("분석적인", "직감적인"), "reveal_lead": "말투를 ‘{}’ 쪽으로"}, | |
| } | |
| _AXIS_CYCLE = ["EVA", "VAL", "EAR", "REI"] | |
| def pick_axis(n): return _AXIS_CYCLE[(n - 1) % len(_AXIS_CYCLE)] | |
| def _now(): return datetime.datetime.now().isoformat(timespec="seconds") | |
| # ============================== 관심사(2단계 계층형) — 측정 아님: 자기보고 + 대화 맥락 ============================== | |
| # 1단계 가치영역(우선순위) → 2단계 행복/방해(공통4 + 1순위 분류별3). VAL/REI 측정 축은 건드리지 않음. | |
| VALUE_DOMAINS = [ | |
| ("🏡 가정·관계", "family"), ("🏆 성공·성취", "achievement"), ("🕊️ 자유·자기실현", "freedom"), | |
| ("🌿 안정·평온", "stability"), ("📚 배움·성장", "growth"), ("💪 건강", "health"), ("🎉 재미·즐거움", "fun"), | |
| ] | |
| COMMON_HAPPY = [("🤝 가까운 사람과 함께할 때", "relation"), ("💗 몸과 마음이 건강할 때", "health"), | |
| ("😄 즐겁고 재미있는 경험을 할 때", "fun"), ("🌿 안정되고 여유로울 때", "stability")] | |
| DOMAIN_HAPPY = { | |
| "family": [("👨👩👧 가족과 따뜻한 시간을 보낼 때", "family"), ("💞 소중한 사람에게 힘이 되어줄 때", "relation"), ("🫂 사람들과 깊이 연결됐다고 느낄 때", "relation")], | |
| "achievement": [("🎯 스스로 정한 목표를 이뤄낼 때", "achievement"), ("📈 실력이 늘고 성장하는 게 느껴질 때", "growth"), ("🏅 노력을 인정받을 때", "recognition")], | |
| "freedom": [("🧭 스스로 결정하고 선택할 때", "autonomy"), ("✨ 새로운 걸 시도하고 도전할 때", "openness"), ("🪶 무엇에도 얽매이지 않을 때", "autonomy")], | |
| "stability": [("🛏️ 마음이 편안하고 걱정 없을 때", "stability"), ("🍵 느긋하게 쉴 여유가 있을 때", "stability"), ("🏠 일상이 안정적으로 돌아갈 때", "stability")], | |
| "growth": [("💡 새로운 걸 배우고 깨달을 때", "growth"), ("🔍 호기심이 채워질 때", "growth"), ("🌱 어제보다 나아진 나를 느낄 때", "growth")], | |
| "health": [("🏃 몸이 가볍고 활력 있을 때", "health"), ("🧘 마음이 건강하고 단단할 때", "health"), ("😴 잘 쉬고 회복됐을 때", "health")], | |
| "fun": [("🎮 좋아하는 걸 즐길 때", "fun"), ("🎨 무언가에 몰입하고 빠져들 때", "flow"), ("🤣 마음껏 웃을 때", "fun")], | |
| } | |
| COMMON_BARRIER = [("💸 경제적 압박·돈 걱정", "money"), ("⏰ 시간이 부족하고 쫓길 때", "time"), | |
| ("😔 몸이 지치고 아플 때", "health"), ("🪞 나도 모르게 남과 비교하게 될 때", "comparison")] | |
| DOMAIN_BARRIER = { | |
| "family": [("💔 가까운 사람과 갈등·서운함이 있을 때", "relation"), ("🫥 외롭거나 단절된 느낌일 때", "relation"), ("🤐 마음을 나눌 사람이 없을 때", "relation")], | |
| "achievement": [("📉 노력만큼 성과가 안 날 때", "achievement"), ("🪫 실력이 정체된 느낌일 때", "growth"), ("🥀 인정받지 못한다고 느낄 때", "recognition")], | |
| "freedom": [("⛓️ 내 뜻대로 못 하고 얽매일 때", "autonomy"), ("🚧 선택의 여지가 없을 때", "autonomy"), ("📋 정해진 틀에 갇힌 느낌일 때", "autonomy")], | |
| "stability": [("🌪️ 일상이 흔들리고 불안정할 때", "stability"), ("😰 마음이 쉴 틈 없이 불안할 때", "anxiety"), ("🌫️ 미래가 불확실하게 느껴질 때", "uncertainty")], | |
| "growth": [("🧱 배우고 싶은데 여건이 안 될 때", "growth"), ("😶🌫️ 제자리걸음 같다고 느낄 때", "growth"), ("❓ 뭘 해야 할지 막막할 때", "uncertainty")], | |
| "health": [("🤒 몸이 자주 아프거나 무거울 때", "health"), ("😣 잠을 못 자고 회복이 안 될 때", "health"), ("🥵 체력이 달릴 때", "health")], | |
| "fun": [("🫠 즐길 여유나 흥미가 없을 때", "fun"), ("😑 모든 게 지루하게 느껴질 때", "fun"), ("🪫 좋아하던 것도 시큰둥할 때", "fun")], | |
| } | |
| _DOMAIN_TAG = {lbl: tag for lbl, tag in VALUE_DOMAINS} | |
| _HAPPY_TAG = {lbl: tag for lbl, tag in COMMON_HAPPY} | |
| _BARRIER_TAG = {lbl: tag for lbl, tag in COMMON_BARRIER} | |
| for _v in DOMAIN_HAPPY.values(): _HAPPY_TAG.update({l: t for l, t in _v}) | |
| for _v in DOMAIN_BARRIER.values(): _BARRIER_TAG.update({l: t for l, t in _v}) | |
| _COMMON_HAPPY_SET = {l for l, _ in COMMON_HAPPY} | |
| _COMMON_BARRIER_SET = {l for l, _ in COMMON_BARRIER} | |
| def happy_choices(tag): return [l for l, _ in COMMON_HAPPY + DOMAIN_HAPPY.get(tag, [])] | |
| def barrier_choices(tag): return [l for l, _ in COMMON_BARRIER + DOMAIN_BARRIER.get(tag, [])] | |
| def _txt(s): return s.split(" ", 1)[1] if " " in s else s # 이모지 제거(프롬프트용) | |
| # ============================== 데이터 저장소(분리 스트림 + HF Dataset) ============================== | |
| class DataStore: | |
| """수집 데이터를 sessions/labels 두 스트림으로 분리·구조화. 동의 시에만 발화 저장. HF Dataset 영구화.""" | |
| def __init__(self, data_dir=DATA_DIR, repo=HF_DATASET_REPO, version=VERSION, encoder=ENCODER): | |
| self.dir = data_dir; os.makedirs(data_dir, exist_ok=True) | |
| self.labels_path = os.path.join(data_dir, "labels.jsonl") | |
| self.sessions_path = os.path.join(data_dir, "sessions.jsonl") | |
| self.interests_path = os.path.join(data_dir, "interests.jsonl") | |
| self.responses_path = os.path.join(data_dir, "responses.jsonl") | |
| self.reveals_path = os.path.join(data_dir, "reveals.jsonl") | |
| self.pilot_path = os.path.join(data_dir, "pilot_labels.jsonl") # 연구 파일럿: 4축 자기라벨 + 측정 + LLM 점수 (원문 미저장) | |
| self.pilot_saved = 0 | |
| self.version, self.encoder = version, encoder | |
| self.stats = [] # 실시간 통계용(발화 없음, PII 아님) | |
| self.resp_stats = [] # 응답 모델/폴백 집계(PII 아님) | |
| self.tok_in = 0; self.tok_out = 0; self.cost_total = 0.0 # 실측 토큰·비용 누적 | |
| self.saved = 0 | |
| self.scheduler = None | |
| if repo and os.environ.get("HF_TOKEN"): | |
| try: | |
| from huggingface_hub import CommitScheduler | |
| self.scheduler = CommitScheduler(repo_id=repo, repo_type="dataset", | |
| folder_path=data_dir, path_in_repo="data", | |
| every=5, private=True) | |
| print("HF Dataset 동기화 ON:", repo) | |
| except Exception as e: | |
| print("HF Dataset 동기화 OFF:", e) | |
| if os.path.exists(self.labels_path): | |
| try: | |
| for ln in open(self.labels_path, encoding="utf-8"): | |
| r = json.loads(ln); ax = r["labeled_axis"] | |
| self.stats.append({"axis": ax, "coord": float(r["coord_" + ax]), | |
| "label_value": int(r["label_value"])}) | |
| self.saved += 1 | |
| except Exception: | |
| pass | |
| if os.path.exists(self.responses_path): | |
| try: | |
| for ln in open(self.responses_path, encoding="utf-8"): | |
| r = json.loads(ln) | |
| self.resp_stats.append({"model": r.get("response_model"), | |
| "depth": int(r.get("fallback_depth", 0))}) | |
| self.tok_in += int(r.get("prompt_tokens", 0) or 0) | |
| self.tok_out += int(r.get("output_tokens", 0) or 0) | |
| self.cost_total += float(r.get("cost_usd", 0) or 0) | |
| except Exception: | |
| pass | |
| def _write(self, path, rec): | |
| lock = self.scheduler.lock if self.scheduler else nullcontext() | |
| with lock: | |
| with open(path, "a", encoding="utf-8") as f: | |
| f.write(json.dumps(rec, ensure_ascii=False) + "\n") | |
| def save_session(self, session, participant, consent, baseline, onboard_rows): | |
| if not consent: | |
| return | |
| self._write(self.sessions_path, { | |
| "type": "session", "ts": int(time.time()), "datetime": _now(), | |
| "session": session, "participant": participant or None, | |
| "encoder": self.encoder, "app_version": self.version, "consent": True, | |
| "baseline_VAL": round(float(baseline[0]), 4), "baseline_REI": round(float(baseline[1]), 4), | |
| "onboard": onboard_rows}) | |
| def save_interests(self, session, participant, domains, happy, barriers, consent): | |
| if not consent: | |
| return | |
| self._write(self.interests_path, { | |
| "type": "interests", "ts": int(time.time()), "datetime": _now(), | |
| "session": session, "participant": participant or None, | |
| "domains": domains, "happy": happy, "barriers": barriers, | |
| "app_version": self.version}) | |
| def add_label(self, session, participant, turn, utt, coords, coords_cum, axis, label, value, consent): | |
| self.stats.append({"axis": axis, "coord": float(coords[axis]), | |
| "coord_cum": float(coords_cum[axis]), "label_value": int(value)}) | |
| if not consent: | |
| return | |
| self._write(self.labels_path, { | |
| "type": "label", "ts": int(time.time()), "datetime": _now(), | |
| "session": session, "participant": participant or None, "turn": int(turn), | |
| "char_len": len(utt), | |
| "coord_VAL": round(float(coords["VAL"]), 4), "coord_REI": round(float(coords["REI"]), 4), | |
| "coord_EVA": round(float(coords["EVA"]), 4), "coord_EAR": round(float(coords["EAR"]), 4), | |
| "coord_VAL_cum": round(float(coords_cum["VAL"]), 4), "coord_REI_cum": round(float(coords_cum["REI"]), 4), | |
| "coord_EVA_cum": round(float(coords_cum["EVA"]), 4), "coord_EAR_cum": round(float(coords_cum["EAR"]), 4), | |
| "labeled_axis": axis, "label_text": label, "label_value": int(value), | |
| "agree": (None if value == 0 else bool((coords[axis] >= 0) == (value > 0))), | |
| "agree_cum": (None if value == 0 else bool((coords_cum[axis] >= 0) == (value > 0))), | |
| "encoder": self.encoder, "app_version": self.version}) | |
| self.saved += 1 | |
| def add_reveal(self, session, participant, turn, axis, measured_sign, measured_label, self_value, fit_value, consent): | |
| # (ii) 측정 공개 후 "이게 맞나요?" 재질문. 편향없는 자기보고(labels.jsonl)와 분리된 2차 검증 신호. | |
| # fit_value: +1 맞음 / 0 모름 / -1 틀림 (사용자가 측정을 본 뒤의 직접 판단) | |
| if not consent: | |
| return | |
| self._write(self.reveals_path, { | |
| "type": "reveal", "ts": int(time.time()), "datetime": _now(), | |
| "session": session, "participant": participant or None, "turn": int(turn), | |
| "labeled_axis": axis, "measured_sign": int(measured_sign), "measured_label": measured_label, | |
| "self_report_value": int(self_value), "fit_value": int(fit_value), | |
| "encoder": self.encoder, "app_version": self.version}) | |
| def add_pilot_label(self, session, participant, turn, char_len, ko, ko_cum, llm, self_labels, consent, act_self=None): | |
| # 연구 파일럿 — 4축 자기라벨(-2..+2) + KoSimCSE 측정 + LLM(Gemini) 채점(-100..+100). | |
| # act_self: ACT(신체 활성도) 5번째 축 자기보고(-2..+2) 또는 None. 검증 중 — 측정/LLM은 4축 유지. | |
| # 프로토콜 5.6: 수집 시점에 점수 계산, 발화 원문은 저장하지 않음(길이만). | |
| self.pilot_saved += 1 | |
| if not consent: | |
| return | |
| rec = { | |
| "type": "pilot_label", "ts": int(time.time()), "datetime": _now(), | |
| "session": session, "participant": participant or None, "turn": int(turn), | |
| "char_len": int(char_len), | |
| "ko_VAL": round(float(ko["VAL"]), 4), "ko_REI": round(float(ko["REI"]), 4), | |
| "ko_EVA": round(float(ko["EVA"]), 4), "ko_EAR": round(float(ko["EAR"]), 4), | |
| "ko_VAL_cum": (round(float(ko_cum["VAL"]), 4) if ko_cum.get("VAL") is not None else None), | |
| "ko_REI_cum": (round(float(ko_cum["REI"]), 4) if ko_cum.get("REI") is not None else None), | |
| "ko_EVA_cum": round(float(ko_cum["EVA"]), 4), "ko_EAR_cum": round(float(ko_cum["EAR"]), 4), | |
| "self_VAL": int(self_labels["VAL"]), "self_REI": int(self_labels["REI"]), | |
| "self_EVA": int(self_labels["EVA"]), "self_EAR": int(self_labels["EAR"]), | |
| "self_ACT": (int(act_self) if act_self is not None else None), | |
| "llm_VAL": (round(float(llm["VAL"]), 1) if llm else None), | |
| "llm_REI": (round(float(llm["REI"]), 1) if llm else None), | |
| "llm_EVA": (round(float(llm["EVA"]), 1) if llm else None), | |
| "llm_EAR": (round(float(llm["EAR"]), 1) if llm else None), | |
| "llm_model": ("gemini" if llm else None), | |
| "encoder": self.encoder, "app_version": self.version} | |
| self._write(self.pilot_path, rec) | |
| def add_response(self, session, participant, turn, model, depth, char_len, consent, usage=None): | |
| # 응답 LLM 모델/폴백/실측 토큰·비용 기록 — 발화 원문 없음(PII 아님) | |
| self.resp_stats.append({"model": model, "depth": int(depth)}) | |
| tin = int((usage or {}).get("in", 0) or 0) | |
| tout = int((usage or {}).get("out", 0) or 0) | |
| cost = _gemini_cost(model, tin, tout) if model else 0.0 | |
| self.tok_in += tin; self.tok_out += tout; self.cost_total += cost | |
| if not consent: | |
| return | |
| self._write(self.responses_path, { | |
| "type": "response", "ts": int(time.time()), "datetime": _now(), | |
| "session": session, "participant": participant or None, "turn": int(turn), | |
| "char_len": int(char_len), "response_model": model, "fallback_depth": int(depth), | |
| "prompt_tokens": tin, "output_tokens": tout, "cost_usd": round(cost, 6), | |
| "ok": bool(model is not None), "app_version": self.version}) | |
| def _resp_section(self): | |
| if not self.resp_stats: | |
| return "" | |
| from collections import Counter | |
| mc = Counter((r["model"] or "실패") for r in self.resp_stats) | |
| fb = sum(1 for r in self.resp_stats if r["depth"] >= 1 and r["model"] is not None) | |
| fail = sum(1 for r in self.resp_stats if r["model"] is None) | |
| out = [f"\n**응답 모델 사용** (총 {len(self.resp_stats)}턴):"] | |
| for mdl, ct in mc.most_common(): | |
| out.append(f"- {mdl}: {ct}회") | |
| out.append(f"- 폴백 발생 {fb}회 · 응답 실패 {fail}회") | |
| if self.tok_in or self.tok_out: | |
| out.append(f"- 실측 토큰 누적: 입력 {self.tok_in:,} · 출력 {self.tok_out:,}") | |
| out.append(f"- 실측 API 비용 누적: ${self.cost_total:.4f} (≈{self.cost_total*1380:,.0f}원)") | |
| return "\n".join(out) | |
| def summary_md(self): | |
| head = ("🟢 HF Dataset 영구 저장 ON" if self.scheduler | |
| else "🟡 로컬 저장만 — Spaces는 재시작 시 초기화 (HF_TOKEN·HF_DATASET_REPO 설정 시 영구화)") | |
| if not self.stats: | |
| return (f"### 측정 검증 데이터\n{head}\n\n아직 라벨이 없습니다. 대화 후 뜨는 자기보고를 눌러 보세요." | |
| + self._resp_section()) | |
| from collections import defaultdict | |
| by = defaultdict(list) | |
| for r in self.stats: by[r["axis"]].append(r) | |
| lines = [f"### 측정 검증 데이터 (라벨 {len(self.stats)}개)", "실사용 발화 측정-자기보고 일치율:"] | |
| allnn = [] | |
| for ax in ["EVA", "EAR", "VAL", "REI"]: | |
| nn = [r for r in by.get(ax, []) if r["label_value"] != 0] | |
| if not nn: continue | |
| agr = np.mean([(r["coord"] >= 0) == (r["label_value"] > 0) for r in nn]) | |
| allnn += [(r["coord"] >= 0) == (r["label_value"] > 0) for r in nn] | |
| lines.append(f"- {ax}: **{agr*100:.0f}%** (n={len(nn)})") | |
| if allnn: | |
| lines.append(f"\n**전체 일치율 {np.mean(allnn)*100:.0f}%** (우연 기대 50%)") | |
| rs = self._resp_section() | |
| if rs: | |
| lines.append(rs) | |
| lines.append(f"\n_저장 레코드 {self.saved}개 · {head}_") | |
| return "\n".join(lines) | |
| # ============================== 측정 엔진 ============================== | |
| class MeasurementEngine: | |
| def __init__(self, encoder_name=ENCODER): | |
| self.encoder_name = encoder_name | |
| self._load_encoder() | |
| self.axes, self.refs = {}, {} | |
| for c, d in CONSTRUCTS.items(): | |
| ep, en = self._embed(d["pos"]), self._embed(d["neg"]) | |
| v = ep.mean(0) - en.mean(0); v = v / (norm(v) + 1e-8) | |
| self.axes[c] = v | |
| ref = np.vstack([ep, en]) @ v | |
| self.refs[c] = (ref.mean(), ref.std() + 1e-8) | |
| def _load_encoder(self): | |
| if USE_SENTENCE_TRANSFORMERS: | |
| from sentence_transformers import SentenceTransformer | |
| self._st = SentenceTransformer(self.encoder_name) | |
| else: | |
| from transformers import AutoModel, AutoTokenizer | |
| import torch | |
| self._torch = torch | |
| self._tok = AutoTokenizer.from_pretrained(self.encoder_name) | |
| self._mdl = AutoModel.from_pretrained(self.encoder_name).eval() | |
| def _embed(self, sents): | |
| if isinstance(sents, str): sents = [sents] | |
| if USE_SENTENCE_TRANSFORMERS: | |
| return np.asarray(self._st.encode(sents, normalize_embeddings=True), dtype=np.float64) | |
| out = [] | |
| for i in range(0, len(sents), 16): | |
| b = sents[i:i + 16] | |
| inp = self._tok(b, padding=True, truncation=True, max_length=64, return_tensors="pt") | |
| with self._torch.no_grad(): | |
| e = self._mdl(**inp).last_hidden_state[:, 0] | |
| e = e / (e.norm(dim=1, keepdim=True) + 1e-8) | |
| out.append(e.cpu().double().numpy()) | |
| return np.vstack(out) | |
| def measure(self, text): | |
| emb = self._embed(text)[0] | |
| return {c: float((emb @ v - self.refs[c][0]) / self.refs[c][1]) for c, v in self.axes.items()} | |
| # ============================== 프로파일 ============================== | |
| def new_profile(): | |
| return { | |
| "baseline": {"VAL": None, "REI": None}, | |
| "cum": {"VAL": None, "REI": None}, | |
| "wsum": {"VAL": 0.0, "REI": 0.0}, # 세션 누적 가중치 합 (소프트 게이트용) | |
| "wcsum": {"VAL": 0.0, "REI": 0.0}, # 세션 누적 (가중치×좌표) 합 | |
| "n": 0, "emotion_traj": [], "last": None, | |
| "smooth": {"EVA": None, "EAR": None}, # EVA/EAR 약한 합성(평활)값 — 나침반 표시용(단일 발화 노이즈 완화) | |
| "last_utt": None, "label_axis": None, "participant": None, "interests": None, | |
| "pending_reveal": None, | |
| "end_suggested": False, "ended": False, | |
| "phase": "build", "clarify_queue": [], "clarify_asked": False, "explore_offered": False, | |
| } | |
| def set_baseline(profile, val_coord, rei_coord): | |
| profile["baseline"]["VAL"] = val_coord | |
| profile["baseline"]["REI"] = rei_coord | |
| profile["cum"]["VAL"] = val_coord | |
| profile["cum"]["REI"] = rei_coord | |
| return profile | |
| def _len_weight(n): | |
| # 측정 신뢰도 ∝ 발화 길이. 너무 짧으면 제외, 짧으면 약하게(약한 합성), 충분하면 full, 너무 길면 완만 감소. | |
| if n < MIN_MEASURE_LEN: | |
| return 0.0 # 제외(어휘 노이즈로 측정 불가) | |
| if n < RELIABLE_LEN: | |
| return 0.3 + 0.7 * (n - MIN_MEASURE_LEN) / (RELIABLE_LEN - MIN_MEASURE_LEN) # 0.3→1.0 선형 | |
| if n <= LEN_REF: | |
| return 1.0 | |
| return (LEN_REF / max(n, 1)) ** 0.5 # 긴 글의 분석 편향 완화(기존) | |
| def update_cumulative(profile, m, utt_len=0): | |
| lw = _len_weight(utt_len) | |
| for c in ("VAL", "REI"): | |
| cn = m[c] | |
| w = abs(cn) * lw # 소프트 게이트(신뢰=|coord|) × 길이 가중(짧으면 0~약하게) | |
| profile["wsum"][c] += w | |
| profile["wcsum"][c] += w * cn | |
| base = profile["baseline"][c] if profile["baseline"][c] is not None else 0.0 | |
| # 기준선 앵커(가중치 W_BASE) + 세션 전체 신뢰가중 누적 | |
| profile["cum"][c] = (W_BASE * base + profile["wcsum"][c]) / (W_BASE + profile["wsum"][c]) | |
| # EVA/EAR 약한 합성(평활): 단일 발화 노이즈 완화. 짧은 발화(lw 작음)는 거의 반영 안 함, 너무 짧으면(lw=0) 이전값 유지. | |
| for c in ("EVA", "EAR"): | |
| prev = profile["smooth"][c] | |
| if prev is None: | |
| profile["smooth"][c] = float(m[c]) | |
| else: | |
| alpha = 0.6 * lw | |
| profile["smooth"][c] = float((1 - alpha) * prev + alpha * m[c]) | |
| profile["emotion_traj"].append(profile["smooth"]["EVA"]) # 궤적은 평활값으로(노이즈 완화) | |
| profile["n"] += 1 | |
| profile["last"] = m # 원시값 — 재질문(reveal)은 '방금 그 발화'를 비교하므로 원시 유지 | |
| return profile | |
| def emotion_trend(profile): | |
| t = profile["emotion_traj"] | |
| if len(t) < 2: return 0.0 | |
| k = min(4, len(t)) | |
| return float(np.mean(t[-k:]) - np.mean(t[:k])) | |
| # ============================== 응답 LLM(언어 출력만) ============================== | |
| def build_prompt(profile): | |
| cum, last = profile["cum"], profile["last"] or {"EVA": 0, "EAR": 0} | |
| val = "자율 지향(스스로 탐색하도록 도움)" if (cum["VAL"] or 0) >= 0 else "순응 지향(안정감과 명확한 방향 제시)" | |
| rei = "분석적(논리·근거 중심)" if (cum["REI"] or 0) >= 0 else "직관적(비유·느낌 중심)" | |
| eva = "즐거운 편" if last["EVA"] >= 0 else "가라앉은 편" | |
| ear = "높음" if last["EAR"] >= 0 else "낮음" | |
| nick = (profile.get("participant") or "").strip() | |
| honorific = f"{nick}님" if nick else "사용자님" | |
| base = ( | |
| f"당신은 '{honorific}'와 편하게 이야기 나누는 따뜻하고 다정한 대화 상대입니다. " | |
| "항상 존댓말을 쓰고, 약간 밝고 긍정적인 말투를 쓰되 과하게 들뜨지는 마세요(느낌표·감탄사·이모지 남발 금지). " | |
| f"사용자에게 진심으로 관심을 보이고, 가끔 자연스럽게 '{honorific}'이라고 불러 주세요(매 문장 호명은 어색하니 가끔만). " | |
| "가장 중요한 규칙: 사용자가 방금 꺼낸 주제·감정에 먼저 반응하고 그 안에 머무르세요. " | |
| "다른 고민을 캐묻거나 화제를 돌리지 말고, 사용자가 말한 것에 진짜로 응답하세요. " | |
| "사용자가 무겁거나 진지한 이야기를 하면 즉시 톤을 낮추고 그 이야기에 충분히 머무르세요. " | |
| "당신은 AI입니다 — 쉬고 있다거나 감정이 있다는 식으로 자신을 꾸며내지 말고, 사용자에게 집중하세요. " | |
| "아래 측정 결과를 은근히 반영하되(진단·분석은 직접 언급 금지):\n" | |
| f"- 가치관: {val}\n- 인지 방식: {rei}\n- 지금 기분: {eva} · 에너지: {ear}\n" | |
| "사용자의 가치관·인지 방식에 맞춰 말투를 조절하고, 현재 기분에 톤을 맞추세요. " | |
| "한국어로 2~3문장, 일상 대화체로 짧게. " | |
| "매 턴 질문하지 말고 공감으로 자연스럽게 마무리하세요(아래 '지금 할 일'이 있으면 그건 따르세요)." | |
| ) | |
| phase = profile.get("phase", "build") | |
| q = profile.get("clarify_queue") or [] | |
| if phase == "build" and profile.get("n", 0) >= 3: | |
| base += ( | |
| "\n[대화 전략] 대화가 어느 정도 쌓였습니다. 사용자가 지금 무엇을 원하는지 살펴 그에 맞게 이끌어 주세요: " | |
| "· 감정을 알아주길 원하는 듯하면 → 먼저 그 감정에 정서적으로 공감하기. " | |
| "· 인정·이해를 원하는 듯하면 → 그 마음과 노력을 충분히 인정하고 이해해 주기. " | |
| "· 지식·생각을 나누고 싶어 하면 → 흥미를 보이며 그 주제로 함께 이야기 나누기. " | |
| "· 뿌듯함·자랑을 드러내면 → 진심으로 함께 기뻐하고 축하해 주기. " | |
| "· 고민을 털어놓고 싶어 하면 → 판단 없이 들어주고 곁에 있어 주기. " | |
| "사용자가 스스로 더 이야기하고 싶도록 여지를 열어두되(주도권은 사용자에게), 억지로 깊이 캐묻거나 원치 않는 방향으로 몰지 마세요. " | |
| "어떤 유형인지 확실하지 않으면 일단 공감하며 가볍게 따라가세요." | |
| ) | |
| base += ( | |
| "\n[넛지] 사용자가 불편하거나 힘든 상황을 이야기하면, 그 주제에서 '한 번만', 스스로 작은 변화를 떠올리도록 돕는 부드러운 질문을 건네세요. " | |
| "예: \"혹시 지금 마음을 가볍게 해줄 수 있는 게 어떤 게 있을까요?\" " | |
| "조언이나 해결책을 직접 제시하지 말고(사용자가 스스로 답을 찾도록), 같은 주제에서 반복하지 마세요. " | |
| "사용자가 새 주제로 넘어가거나 다시 청하면 그때 다시 건네도 됩니다. " | |
| "사용자가 정말 힘들어 보이면 넛지 대신 그냥 충분히 들어주세요(안녕이 최우선)." | |
| ) | |
| if phase == "clarify" and q: | |
| target = q[0] | |
| if target == "VAL": | |
| base += ("\n[지금 할 일] 사용자의 '가치관'(스스로 정함 ↔ 주변에 맞춤)이 온보딩에서 불명확했습니다. " | |
| "구체적인 실제 생활 상황을 하나 들어, 어느 쪽에 가까운지 자연스럽게 확인하는 질문을 '하나만' 하세요. 캐묻지 마세요.") | |
| else: | |
| base += ("\n[지금 할 일] 사용자의 '인지 방식'(분석 ↔ 직감)이 온보딩에서 불명확했습니다. " | |
| "구체적인 실제 생활 상황을 하나 들어, 어느 쪽에 가까운지 자연스럽게 확인하는 질문을 '하나만' 하세요. 캐묻지 마세요.") | |
| elif phase == "explore": | |
| bar = "" | |
| it0 = profile.get("interests") | |
| if it0 and it0.get("barriers"): | |
| bar = f"사용자가 힘든 점으로 '{_txt(it0['barriers'][0]['item'])}'을(를) 꼽았습니다. " | |
| base += ("\n[지금 할 일] 라포가 쌓였습니다. " + bar + | |
| "사용자가 지금 이야기하는 흐름을 끊지 말고, 그 안에서 한 번만 부드럽게 더 깊이 들어가는 질문을 던져 자기이해를 도와보세요. " | |
| "사용자가 원치 않거나 부담스러워하면 즉시 물러나 공감으로 돌아가세요. " | |
| "절대 부정적 감정을 억지로 끌어내거나 캐묻지 마세요. 사용자의 안녕이 최우선입니다.") | |
| it = profile.get("interests") | |
| if it: | |
| ctx = [] | |
| dom = ", ".join(_txt(d["item"]) for d in it.get("domains", [])) | |
| hap = ", ".join(_txt(h["item"]) for h in it.get("happy", [])[:3]) | |
| bar = ", ".join(_txt(b["item"]) for b in it.get("barriers", [])[:3]) | |
| if dom: ctx.append(f"중요시하는 것: {dom}") | |
| if hap: ctx.append(f"행복요인: {hap}") | |
| if bar: ctx.append(f"방해요인: {bar}") | |
| if ctx: | |
| base += "\n참고 맥락(직접 나열하지 말고 자연스럽게만 반영): " + " · ".join(ctx) | |
| return base | |
| def generate_response(profile, history, user_msg): | |
| sys_prompt = build_prompt(profile) | |
| if LLM_MODE == "gemini": | |
| return _gemini(sys_prompt, history, user_msg) | |
| if LLM_MODE == "local": | |
| return (_local(sys_prompt, history, user_msg), "local", 0, None) | |
| return (_template(profile, user_msg), "template", 0, None) | |
| def detect_ambiguity(onboard_rows, baseline_val, baseline_rei): | |
| """온보딩에서 명료화가 필요한 구성(VAL/REI)을 반환. 부호충돌 또는 약신호(|기준선|<0.3).""" | |
| queue = [] | |
| for ax, base in [("VAL", baseline_val), ("REI", baseline_rei)]: | |
| cs = [r["coord"] for r in onboard_rows if r["axis"] == ax and r.get("coord") is not None] | |
| conflict = (len(cs) == 2 and (cs[0] >= 0) != (cs[1] >= 0)) | |
| weak = (base is not None and abs(base) < 0.3) | |
| if conflict or weak: | |
| queue.append(ax) | |
| return queue | |
| def generate_opening(profile): | |
| """앱이 먼저 대화를 여는 첫 메시지(LLM 생성). 관심사 있으면 그걸로, 없으면 직업 질문.""" | |
| it = profile.get("interests") | |
| if it and it.get("domains"): | |
| dom = _txt(it["domains"][0]["item"]) | |
| op = (f"사용자가 요즘 중요하게 여기는 것으로 '{dom}'을(를) 꼽았습니다. " | |
| "이걸 자연스럽게 언급하며 따뜻하게 대화를 여세요. 한두 문장으로 관심을 보이고, " | |
| "그 주제로 부담 없는 질문을 하나만 하세요.") | |
| else: | |
| op = ("사용자가 관심사를 밝히지 않았습니다. 따뜻하게 인사하고, 요즘 어떤 일을 하며 지내는지" | |
| "(직업이나 하루 일과) 가볍게 물어 대화를 여세요. 한두 문장과 질문 하나로.") | |
| sys = "당신은 사용자와 편하게 수다 떠는 친근한 대화 상대입니다. 친구처럼 가볍고 자연스럽게, 한국어로 2~3문장. " + op | |
| return _gemini(sys, [], "(따뜻하게 먼저 대화를 시작해 주세요.)") | |
| def _template(profile, user_msg): | |
| last = profile["last"] or {"EVA": 0, "EAR": 0} | |
| cum = profile["cum"] | |
| feel = "마음이 무거우신 듯해요" if last["EVA"] < 0 else "기분이 괜찮아 보이세요" | |
| nudge = "이 일을 스스로는 어떻게 바라보고 싶으세요?" if (cum["VAL"] or 0) >= 0 \ | |
| else "지금 가장 마음이 놓이는 방향은 어떤 쪽일까요?" | |
| return f"{feel}. {nudge}" | |
| def _msg_text(content): | |
| # Gradio 버전별 메시지 content 정규화: 4.x는 문자열, 6.0은 리스트([{'type':'text','text':...}])일 수 있음. | |
| if content is None: | |
| return "" | |
| if isinstance(content, str): | |
| return content | |
| if isinstance(content, dict): | |
| return str(content.get("text", "") or "") | |
| if isinstance(content, (list, tuple)): | |
| parts = [] | |
| for it in content: | |
| if isinstance(it, str): | |
| parts.append(it) | |
| elif isinstance(it, dict): | |
| parts.append(str(it.get("text", "") or "")) | |
| return " ".join(p for p in parts if p) | |
| return str(content) | |
| def _gemini(sys_prompt, history, user_msg): | |
| import os, requests | |
| key = os.environ.get("GEMINI_API_KEY", "").strip() | |
| if not key: | |
| return ("[GEMINI_API_KEY 가 설정되지 않았습니다. Colab Secrets 또는 Space Secret에 추가하세요.]", None, 0, None) | |
| models = ["gemini-2.5-flash", "gemini-2.5-flash-lite", "gemini-flash-latest"] | |
| # 최근 대화 이력 포함 — 맥락 유지. 비용 절제 위해 마지막 6개(=약 3턴)만. (이력 없으면 AI가 직전 흐름을 못 봄) | |
| hist = (history or [])[-6:] | |
| while hist and hist[0].get("role") == "assistant": | |
| hist = hist[1:] # Gemini contents는 user 턴으로 시작해야 함 — 선두 assistant 제거 | |
| contents = [] | |
| for m in hist: | |
| role = "model" if m.get("role") == "assistant" else "user" | |
| c = _msg_text(m.get("content")).strip() | |
| if c: | |
| contents.append({"role": role, "parts": [{"text": c}]}) | |
| contents.append({"role": "user", "parts": [{"text": user_msg}]}) | |
| payload = {"system_instruction": {"parts": [{"text": sys_prompt}]}, | |
| "contents": contents, | |
| "generationConfig": { | |
| "thinkingConfig": {"thinkingBudget": 0}, # thinking 끔 → 출력 토큰·비용·지연 대폭 감소(2.5-flash는 기본 on) | |
| "maxOutputTokens": 500, # 2~3문장이면 충분 — 폭주 방지 안전망 | |
| "temperature": 0.9, | |
| }} | |
| last_err = "" | |
| for i, mdl in enumerate(models): | |
| url = (f"https://generativelanguage.googleapis.com/v1beta/models/" | |
| f"{mdl}:generateContent?key={key}") | |
| try: | |
| data = requests.post(url, json=payload, timeout=30).json() | |
| except Exception as e: | |
| last_err = f"요청 실패: {e}"; continue | |
| if "error" in data: | |
| msg = data["error"].get("message", str(data["error"])) | |
| last_err = f"{mdl}: {msg}" | |
| if any(w in msg.lower() for w in ["not found", "not supported", "permission", "model", "quota", "rate", "exhaust"]): | |
| continue | |
| return (f"[Gemini 오류] {last_err}", None, i, None) | |
| cands = data.get("candidates") | |
| if not cands: | |
| last_err = f"{mdl}: candidates 없음 ({data.get('promptFeedback', {})})"; continue | |
| parts = cands[0].get("content", {}).get("parts", []) | |
| text = "".join(p.get("text", "") for p in parts).strip() | |
| if text: | |
| um = data.get("usageMetadata", {}) or {} | |
| usage = {"in": int(um.get("promptTokenCount", 0) or 0), | |
| "out": int(um.get("candidatesTokenCount", 0) or 0) + int(um.get("thoughtsTokenCount", 0) or 0)} | |
| return (text, mdl, i, usage) | |
| last_err = f"{mdl}: 빈 응답 (finishReason: {cands[0].get('finishReason')})" | |
| return (f"[Gemini 응답 실패] 마지막 원인 — {last_err}", None, len(models), None) | |
| def _gemini_score_4axis(text): | |
| # 연구 파일럿 — 발화를 4축으로 LLM 채점(-100..+100). 실패 시 None. | |
| sys = ("다음 한국어 문장을 네 독립 축으로 채점하라. 각 축 -100~+100 정수. " | |
| "오직 JSON만 출력하고 설명은 절대 쓰지 마라: {\"VAL\":n,\"REI\":n,\"EVA\":n,\"EAR\":n}\n" | |
| "VAL 자율(+)↔순응(-): 내 기준으로 정하면 +, 주변·규칙에 맞추면 -.\n" | |
| "REI 분석(+)↔직관(-): 근거·논리로 정하면 +, 직감·느낌으로 정하면 -.\n" | |
| "EVA 긍정(+)↔부정(-): 감정이 긍정적이면 +, 부정적이면 -.\n" | |
| "EAR 고각성(+)↔저각성(-): 에너지·흥분·긴장이 높으면 +, 차분·처짐이면 - (감정 좋고나쁨과 무관).\n" | |
| "해당 축이 무관하면 0.") | |
| try: | |
| txt, mdl, depth, usage = _gemini(sys, [], (text or "").strip()) | |
| except Exception: | |
| return None | |
| if mdl is None or not txt: | |
| return None | |
| import json as _json, re as _re | |
| s = txt.strip().replace("```json", "").replace("```", "").strip() | |
| m = _re.search(r"\{.*\}", s, _re.DOTALL) | |
| if not m: | |
| return None | |
| try: | |
| d = _json.loads(m.group(0)) | |
| return {ax: float(d.get(ax, 0)) for ax in ("VAL", "REI", "EVA", "EAR")} | |
| except Exception: | |
| return None | |
| # 연구 파일럿 자기라벨 도구 (프로토콜 부록 A) — 4축 5단계 | |
| PILOT_AXES = { | |
| "EVA": ("기분(감정가)", ["-2 매우 부정적", "-1 약간 부정적", "+1 약간 긍정적", "+2 매우 긍정적"]), | |
| "EAR": ("에너지·각성", ["-2 매우 차분·처짐", "-1 약간 가라앉음", "+1 약간 들뜸·또렷", "+2 매우 흥분·곤두섬"]), | |
| "REI": ("사고방식", ["-2 순전히 직감·느낌", "-1 주로 느낌", "+1 주로 근거·논리", "+2 순전히 근거·논리"]), | |
| "VAL": ("자율성", ["-2 전적으로 주변·규칙", "-1 주로 주변", "+1 주로 내 기준", "+2 전적으로 내 기준"]), | |
| } | |
| # ACT(신체 활성도) — 검증 중인 5번째 축 후보. 자기보고만 수집(측정·LLM은 4축 유지). | |
| # 각성(에너지 높낮이)과 별개로, '몸이 격렬하게 반응/소진된 정도'. 차분(저활성) vs 탈진(고활성) 구분 검증용. | |
| PILOT_ACT = ("신체 활성도(실험)", ["-2 완전히 이완·고요", "-1 약간 느슨", "+1 약간 격렬·긴장", "+2 매우 격렬·소진"]) | |
| def _parse_pilot_val(opt): | |
| return int(opt.split()[0]) # "-2 매우..." → -2, "+1 약간..." → 1 (중간 없는 4지선다 대응) | |
| _PILOT_MAP = {ax: {opt: _parse_pilot_val(opt) for opt in opts} for ax, (_lbl, opts) in PILOT_AXES.items()} | |
| _PILOT_MAP["ACT"] = {opt: _parse_pilot_val(opt) for opt in PILOT_ACT[1]} | |
| PILOT_PASSWORD = "qwer" | |
| def _local(sys_prompt, history, user_msg): | |
| global _LOCAL_PIPE | |
| try: | |
| _LOCAL_PIPE | |
| except NameError: | |
| from transformers import pipeline | |
| _LOCAL_PIPE = pipeline("text-generation", model="kakaocorp/kanana-nano-2.1b-instruct", | |
| device_map="auto", max_new_tokens=180) | |
| msgs = [{"role": "system", "content": sys_prompt}, {"role": "user", "content": user_msg}] | |
| try: | |
| out = _LOCAL_PIPE(msgs)[0]["generated_text"] | |
| return out[-1]["content"] if isinstance(out, list) else str(out) | |
| except Exception as e: | |
| return f"[로컬 모델 오류: {e}]" | |
| # ============================== Gradio UI ============================== | |
| def _wmean(pairs): | |
| sw = sum(w for _, w in pairs) | |
| return (sum(v * w for v, w in pairs) / sw) if sw else 0.0 | |
| def build_app(): | |
| import gradio as gr | |
| print("측정 엔진 로딩 중 (KoSimCSE)...") | |
| engine = MeasurementEngine() | |
| store = DataStore() | |
| print("준비 완료.") | |
| label2stmt = [{lab: stmt for lab, stmt in item["choices"]} for item in ONBOARD] | |
| def label_update(axis, visible): | |
| if not visible: | |
| return gr.update(choices=[], value=None, visible=False) | |
| sc = LABEL_SCHEMES[axis] | |
| return gr.update(choices=[o[0] for o in sc["opts"]], label=sc["q"], value=None, visible=True) | |
| def do_onboard(*args): | |
| picks = args[:-4]; consent = args[-4]; participant = args[-3]; session = args[-2]; old_profile = args[-1] | |
| coords = {"VAL": [], "REI": []}; onboard_rows = [] | |
| for i, item in enumerate(ONBOARD): | |
| c = item["construct"] | |
| radio, free = picks[2 * i], picks[2 * i + 1] | |
| coord = None | |
| if radio: | |
| coord = engine.measure(label2stmt[i][radio])[c] | |
| coords[c].append((coord, 1.0)) | |
| if free and free.strip(): | |
| fc = engine.measure(free.strip())[c] | |
| coords[c].append((fc, FREE_WEIGHT)) | |
| coord = fc if coord is None else coord | |
| onboard_rows.append({"axis": c, "choice": radio or None, | |
| "coord": (round(float(coord), 4) if coord is not None else None)}) | |
| if not coords["VAL"] or not coords["REI"]: | |
| return old_profile, "가치(Q1·Q2)와 인지(Q3·Q4) 각각에서 최소 한 가지는 선택하거나 입력해 주세요.", gr.update(), gr.update(), gr.update(), store.summary_md() | |
| profile = set_baseline(new_profile(), _wmean(coords["VAL"]), _wmean(coords["REI"])) | |
| profile["participant"] = (participant.strip() if participant and participant.strip() else None) | |
| profile["interests"] = old_profile.get("interests") # 관심사 보존(흐름 A: 관심사 → 온보딩) | |
| store.save_session(session, profile["participant"], bool(consent), | |
| (profile["baseline"]["VAL"], profile["baseline"]["REI"]), onboard_rows) | |
| # 애매한 온보딩 항목 감지 → 명료화 큐 | |
| profile["clarify_queue"] = detect_ambiguity(onboard_rows, profile["baseline"]["VAL"], profile["baseline"]["REI"]) | |
| profile["phase"] = "open" | |
| # 앱이 먼저 대화를 연다 — 템플릿(LLM 호출 없음, 무료 한도 절약) | |
| it = profile.get("interests") | |
| if it and it.get("domains"): | |
| op_text = (f"안녕! 🙂 아까 '{_txt(it['domains'][0]['item'])}' 얘기 했었죠. " | |
| "거기서부터 가볍게 시작해도 좋고, 아래 버튼에서 골라도 돼요. 요즘 어때요?") | |
| else: | |
| op_text = ("안녕! 🙂 편하게 수다 떨어요. 무슨 얘기부터 할지 고민되면 " | |
| "아래 버튼에서 하나 골라봐요 — 아니면 그냥 떠오르는 대로 적어도 좋아요.") | |
| opening_chat = [{"role": "assistant", "content": op_text}] | |
| return (profile, render_profile(profile), gr.update(value=opening_chat, visible=True), | |
| gr.update(visible=True), gr.update(visible=True), store.summary_md()) | |
| END_SUGGEST_TURN = 12 | |
| EXPLORE_TURN = 8 | |
| def do_chat(msg, chat, profile, session, consent): | |
| if not msg or not msg.strip(): | |
| return chat, profile, render_profile(profile), "", label_update(None, False), "", store.summary_md(), gr.update(visible=False) | |
| # 동의 안내: 미동의 상태면 측정·자기보고가 저장되지 않음을 채팅 아래에 명확히 안내 | |
| consent_note = ("" if bool(consent) else | |
| "⚠️ 현재 **익명 저장 동의가 해제**되어 있어요. 대화는 그대로 되지만 " | |
| "측정 결과가 **저장되지 않습니다**. 연구에 도움을 주시려면 위 '연구 참여 안내·동의'의 " | |
| "체크박스를 켜 주세요. (대화 원문은 어떤 경우에도 저장되지 않습니다.)") | |
| if profile.get("ended"): | |
| return (chat, profile, render_profile(profile), "", label_update(None, False), | |
| "대화를 이미 마쳤어요. 더 하려면 페이지를 새로고침해 주세요.", store.summary_md(), gr.update(visible=False)) | |
| if profile["baseline"]["VAL"] is None: | |
| chat = chat + [{"role": "assistant", "content": "먼저 위 질문 몇 개만 완료해 주세요 🙂"}] | |
| return chat, profile, render_profile(profile), "", label_update(None, False), consent_note, store.summary_md(), gr.update(visible=False) | |
| m = engine.measure(msg) | |
| profile = update_cumulative(profile, m, len(msg)) | |
| profile["last_utt"] = msg | |
| profile["pending_reveal"] = None # 새 메시지 → 직전 재질문 상태 정리 | |
| profile["label_axis"] = pick_axis(profile["n"]) | |
| # --- 대화 단계 전환 (응답 생성 전) --- | |
| phase = profile.get("phase", "build") | |
| if phase == "open": | |
| # 오프닝에 대한 첫 답변을 받음 → 명료화 필요하면 clarify, 아니면 build | |
| if profile.get("clarify_queue"): | |
| profile["phase"] = "clarify"; profile["clarify_asked"] = False | |
| else: | |
| profile["phase"] = "build" | |
| elif phase == "clarify": | |
| if profile.get("clarify_asked") and profile.get("clarify_queue"): | |
| profile["clarify_queue"].pop(0) # 직전에 물어본 구성에 답함 → 제거 | |
| if profile.get("clarify_queue"): | |
| profile["clarify_asked"] = True # 이번 턴에 queue[0]을 확인 | |
| else: | |
| profile["phase"] = "build"; profile["clarify_asked"] = False | |
| elif phase == "build": | |
| if profile["n"] >= EXPLORE_TURN and not profile.get("explore_offered"): | |
| profile["phase"] = "explore"; profile["explore_offered"] = True | |
| elif phase == "explore": | |
| profile["phase"] = "build" # explore는 '한 번'만 — 다음 턴부터 일반 대화로 복귀(캐묻기 방지) | |
| reply, rmodel, rdepth, rusage = generate_response(profile, chat, msg) | |
| store.add_response(session, profile.get("participant"), profile["n"], | |
| rmodel, rdepth, len(msg), bool(consent), rusage) | |
| if profile["n"] >= END_SUGGEST_TURN and not profile["end_suggested"]: | |
| profile["end_suggested"] = True | |
| reply = reply + "\n\n(오늘 충분히 이야기 나눴어요 🙂 더 하셔도 좋고, 마치려면 아래 '✋ 이제 그만할래'를 눌러주세요.)" | |
| chat = chat + [{"role": "user", "content": msg}, {"role": "assistant", "content": reply}] | |
| return chat, profile, render_profile(profile), "", label_update(profile["label_axis"], True), consent_note, store.summary_md(), gr.update(visible=False) | |
| def do_end(chat, profile): | |
| profile["ended"] = True | |
| profile["pending_reveal"] = None | |
| chat = chat + [{"role": "assistant", | |
| "content": "오늘 대화는 여기까지 할게요. 솔직하게 이야기 나눠줘서 고마워요 🙂 " | |
| "측정 개선에 큰 도움이 됩니다. 더 하고 싶으면 페이지를 새로고침해 주세요."}] | |
| return (chat, gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), | |
| gr.update(value=None, visible=False), "대화를 마쳤어요. 고마워요 🙂", store.summary_md(), | |
| gr.update(value=None, visible=False)) | |
| def check_save_status(profile, consent, session): | |
| n = profile.get("n", 0) if profile else 0 | |
| if not bool(consent): | |
| return ("⚠️ **저장 꺼짐** — 현재 익명 저장 동의가 해제되어 있어, 지금까지의 측정이 " | |
| "저장되지 **않습니다**. 위쪽 '연구 참여 안내·동의'의 체크박스를 켜시면, " | |
| "이후 대화부터 측정 좌표·자기보고가 익명으로 저장됩니다. " | |
| "(대화 원문은 어떤 경우에도 저장되지 않아요.)") | |
| pcode = (profile.get("participant") or "").strip() if profile else "" | |
| pcode_txt = f"참가자 코드 **{pcode}**로 " if pcode else "익명으로 " | |
| return (f"✅ **저장 켜짐** — {pcode_txt}측정 좌표와 자기보고가 익명으로 저장되고 있어요. " | |
| f"지금까지 대화 {n}턴. 솔직하게 이야기해 주셔서 측정 개선에 큰 도움이 됩니다. " | |
| f"(대화 원문은 저장되지 않고, 길이 등 숫자만 남습니다.)") | |
| def do_label(choice, profile, consent, session): | |
| if not choice or profile.get("last") is None or profile.get("label_axis") is None: | |
| return "", store.summary_md(), gr.update(), gr.update(visible=False), profile | |
| axis = profile["label_axis"] | |
| value = dict(LABEL_SCHEMES[axis]["opts"])[choice] | |
| coord = profile["last"][axis] | |
| # 누적 좌표: VAL/REI는 세션 전체 누적, EVA/EAR(상태축)는 턴별값 그대로 | |
| cum = profile["cum"] | |
| coords_cum = {"VAL": cum["VAL"], "REI": cum["REI"], | |
| "EVA": profile["last"]["EVA"], "EAR": profile["last"]["EAR"]} | |
| # 1) 편향없는 자기보고를 '먼저' 저장 (1차 검증 신호 — 측정 공개 전, 불변) | |
| store.add_label(session, profile.get("participant"), profile["n"], | |
| profile.get("last_utt", ""), profile["last"], coords_cum, axis, choice, value, bool(consent)) | |
| # 2) (ii) 측정 공개 + "맞나요?" 재질문 — 매 턴이 아니라 3턴마다만(피로 감소). 그 외 턴은 공개 없이 가볍게. | |
| if profile["n"] % 3 != 0: | |
| return ("기록했어요. 고마워요 🙂", store.summary_md(), | |
| gr.update(value=None, visible=False), gr.update(visible=False), profile) | |
| sch = LABEL_SCHEMES[axis] | |
| measured_sign = 1 if coord >= 0 else -1 | |
| measured_label = sch["poles"][0] if coord >= 0 else sch["poles"][1] | |
| profile["pending_reveal"] = {"axis": axis, "turn": int(profile["n"]), | |
| "measured_sign": measured_sign, "measured_label": measured_label, | |
| "self_value": int(value)} | |
| note = ("기록했어요, 고마워요 🙂 하나만 더 — 제가 방금 " + sch["reveal_lead"].format(measured_label) | |
| + " 느꼈는데, 실제로도 그랬어요? 맞아도 틀려도 다 도움이 되니 편하게 알려줘요.") | |
| fit_choices = ["응, 맞아요", "잘 모르겠어요", "아니, 달라요"] | |
| return (note, store.summary_md(), gr.update(value=None, visible=False), | |
| gr.update(choices=fit_choices, value=None, visible=True, label="제 느낌이 맞았나요?"), profile) | |
| def do_fit(choice, profile, consent, session): | |
| pr = profile.get("pending_reveal") | |
| if not choice or not pr: | |
| return "", gr.update(visible=False), profile | |
| fit_value = {"응, 맞아요": 1, "잘 모르겠어요": 0, "아니, 달라요": -1}.get(choice, 0) | |
| store.add_reveal(session, profile.get("participant"), pr["turn"], pr["axis"], | |
| pr["measured_sign"], pr["measured_label"], pr["self_value"], fit_value, bool(consent)) | |
| profile["pending_reveal"] = None | |
| if fit_value == 1: | |
| note = "오, 맞았네요! 알려줘서 고마워요 🙂" | |
| elif fit_value == -1: | |
| note = "알려줘서 정말 고마워요 — ‘틀렸다’는 이 한 번이 측정을 더 정확하게 만들어요 🙏" | |
| else: | |
| note = "괜찮아요, 그것도 좋은 답이에요. 고마워요 🙂" | |
| return note, gr.update(value=None, visible=False), profile | |
| def on_domain(d1): | |
| # 빈 상태가 아니라 공통 항목으로 시작 → 첫 선택부터 즉시 렌더(6.0 한 박자 지연 방지). | |
| print(f"[on_domain] 호출됨 · d1={d1!r} · tag={_DOMAIN_TAG.get(d1)!r}", flush=True) | |
| tag = _DOMAIN_TAG.get(d1) | |
| h = happy_choices(tag) if tag else [l for l, _ in COMMON_HAPPY] | |
| b = barrier_choices(tag) if tag else [l for l, _ in COMMON_BARRIER] | |
| print(f"[on_domain] happy {len(h)}개 · barrier {len(b)}개 반환", flush=True) | |
| return (gr.CheckboxGroup(choices=h, value=[], label="요즘 나를 행복하게 하는 것 (최대 3개 · 1순위 선택 시 더 추가됨)"), | |
| gr.CheckboxGroup(choices=b, value=[], label="행복을 방해하는 것 (최대 3개)")) | |
| def save_interests(d1, d2, happy_sel, barrier_sel, consent, session, profile): | |
| if not d1: | |
| return "1순위(가장 중요한 것)를 먼저 선택해 주세요.", profile | |
| if len(happy_sel) > 3 or len(barrier_sel) > 3: | |
| return "행복·방해 요인은 각각 최대 3개까지만 골라주세요.", profile | |
| doms = [d1] + ([d2] if (d2 and d2 != "— 없음 —" and d2 != d1) else []) | |
| domains = [{"item": d, "tag": _DOMAIN_TAG.get(d, ""), "rank": i + 1} for i, d in enumerate(doms)] | |
| happy = [{"item": h, "tag": _HAPPY_TAG.get(h, ""), "common": h in _COMMON_HAPPY_SET} for h in happy_sel] | |
| barriers = [{"item": b, "tag": _BARRIER_TAG.get(b, ""), "common": b in _COMMON_BARRIER_SET} for b in barrier_sel] | |
| profile["interests"] = {"domains": domains, "happy": happy, "barriers": barriers} | |
| store.save_interests(session, profile.get("participant"), domains, happy, barriers, bool(consent)) | |
| return "관심사를 반영했어요. 이제 편하게 대화를 시작해 보세요.", profile | |
| def pilot_unlock(pw): | |
| if (pw or "").strip() == PILOT_PASSWORD: | |
| return (True, "✓ 연구 파일럿 입장됨. 대화에서 메시지를 보낸 뒤, 아래에서 그 발화의 네 축을 평가해 저장하세요.", | |
| gr.update(visible=True)) | |
| return (False, "✗ 비밀번호가 올바르지 않습니다.", gr.update(visible=False)) | |
| def pilot_save_fn(v_eva, v_ear, v_rei, v_val, v_act, profile, session, consent, unlocked): | |
| blank = (gr.update(), gr.update(), gr.update(), gr.update(), gr.update()) | |
| if not unlocked: | |
| return ("먼저 비밀번호로 입장해 주세요.", *blank) | |
| if not consent: | |
| return ("저장하려면 상단의 동의에 체크해 주세요 (발화 원문은 저장되지 않습니다).", *blank) | |
| if profile.get("last_utt") is None or profile.get("last") is None: | |
| return ("먼저 대화에서 메시지를 한 번 보내 주세요 — 직전에 보낸 발화를 평가합니다.", *blank) | |
| vals = {"EVA": v_eva, "EAR": v_ear, "REI": v_rei, "VAL": v_val} | |
| if any(vals[a] is None for a in vals): | |
| return ("네 축(기분·에너지·사고방식·자율성)을 모두 평가해 주세요. (활성도는 선택)", *blank) | |
| try: | |
| self_labels = {a: _PILOT_MAP[a][vals[a]] for a in vals} | |
| act_self = _PILOT_MAP["ACT"][v_act] if v_act is not None else None # ACT는 선택(실험 항목) | |
| except KeyError: | |
| return ("선택지를 다시 골라 주세요.", *blank) | |
| llm = _gemini_score_4axis(profile["last_utt"]) # 발화 텍스트는 메모리에서만 사용, 저장 안 함 | |
| ko = profile["last"] | |
| cum = profile.get("cum", {}) | |
| ko_cum = {"VAL": cum.get("VAL"), "REI": cum.get("REI"), | |
| "EVA": profile["last"]["EVA"], "EAR": profile["last"]["EAR"]} | |
| store.add_pilot_label(session, profile.get("participant"), profile.get("n", 0), | |
| len(profile["last_utt"]), ko, ko_cum, llm, self_labels, bool(consent), act_self) | |
| note = "측정·자기라벨·LLM 채점 저장" if llm else "측정·자기라벨 저장 (LLM 채점 실패)" | |
| if act_self is not None: | |
| note += " (+활성도)" | |
| total = getattr(store, "pilot_saved", 0) | |
| return (f"✓ 저장됐어요 — {note}. (누적 {total}건) 다음 발화를 보낸 뒤 또 평가할 수 있어요.", | |
| gr.update(value=None), gr.update(value=None), gr.update(value=None), gr.update(value=None), gr.update(value=None)) | |
| SELFREPORT_CSS = """ | |
| #selfreport_box {background:#FFF7E6; border:1px solid #E6C674; border-radius:12px; | |
| padding:10px 12px; margin:8px 0 6px 0; box-shadow:0 2px 10px rgba(180,140,40,0.12);} | |
| #selfreport_box span, #selfreport_box label {font-weight:600 !important;} | |
| #fit_box {background:#E7F4F4; border:1px solid #7FBFC4; border-radius:12px; | |
| padding:10px 12px; margin:2px 0 6px 0; box-shadow:0 2px 10px rgba(14,124,134,0.12);} | |
| #fit_box span, #fit_box label {font-weight:600 !important;} | |
| #selfreport_box, #fit_box {position:sticky; bottom:8px; z-index:30;} | |
| """ | |
| with gr.Blocks(title="자기인식 지원 (0322 데이터 수집판)", css=SELFREPORT_CSS) as app: | |
| gr.Markdown("## 자기인식 지원 대화 — 데이터 수집판") | |
| session_state = gr.State("") | |
| prof_state = gr.State(new_profile()) | |
| with gr.Accordion("연구 참여 안내 · 동의 (먼저 읽어주세요)", open=True): | |
| gr.Markdown( | |
| "이 앱은 **측정 도구 개선을 위한 실험**입니다. 대화하면 가치·인지·감정이 자동 측정되고, 가끔 자기보고를 여쭤봅니다. " | |
| "동의하시면 **측정 좌표(숫자)와 자기보고 라벨만 익명으로 저장**되고, **대화 원문은 저장하지 않습니다**(길이만 숫자로 남음). " | |
| "응답 생성을 위해 입력은 Google Gemini API로 전송됩니다(저장 안 함). " | |
| "**아래 체크는 기본으로 켜져 있습니다 — 익명 저장에 참여하시려면 그대로 두시고, 원치 않으면 해제하세요.** 언제든 중단할 수 있습니다.") | |
| consent = gr.Checkbox(value=True, label="✅ 익명 저장(측정 좌표·자기보고 라벨만, 발화 원문 제외)에 동의합니다. — 연구에 큰 도움이 됩니다") | |
| participant = gr.Textbox(label="참가자 코드 (선택 · 별명 권장 — 예: 라일락-01. 다시 올 땐 같은 코드를 쓰면 변화를 볼 수 있어요)", lines=1) | |
| with gr.Accordion("1단계 · 관심사 (먼저 알려주세요 · 건너뛰어도 됩니다)", open=True): | |
| gr.Markdown("대화를 당신 맥락에 맞추기 위한 선택입니다. 건너뛰어도 됩니다.") | |
| with gr.Row(): | |
| d1 = gr.Dropdown(choices=[l for l, _ in VALUE_DOMAINS], label="1순위 — 요즘 가장 중요한 것") | |
| d2 = gr.Dropdown(choices=["— 없음 —"] + [l for l, _ in VALUE_DOMAINS], value="— 없음 —", | |
| label="2순위 (선택)") | |
| happy_cg = gr.CheckboxGroup(choices=[l for l, _ in COMMON_HAPPY], label="요즘 나를 행복하게 하는 것 (최대 3개 · 1순위 선택 시 더 추가됨)") | |
| barrier_cg = gr.CheckboxGroup(choices=[l for l, _ in COMMON_BARRIER], label="행복을 방해하는 것 (최대 3개)") | |
| interests_btn = gr.Button("관심사 반영") | |
| interests_status = gr.Markdown("") | |
| with gr.Accordion("2단계 · 짧은 질문 (가치·인지)", open=True): | |
| gr.Markdown("가까운 보기를 고르거나 아래 칸에 직접 한 문장 적어 주세요. 완료하면 대화가 시작됩니다.") | |
| onboard_inputs = [] | |
| for item in ONBOARD: | |
| gr.Markdown(f"**{item['q']}**") | |
| r = gr.Radio(choices=[c[0] for c in item["choices"]], label="선택지") | |
| t = gr.Textbox(label="또는 직접 적어주세요 (한 문장 이상, 선택)", lines=2, | |
| placeholder="예: 나는 보통 ~한 편이에요. 왜냐하면 ~") | |
| onboard_inputs += [r, t] | |
| onboard_btn = gr.Button("대화 시작하기", variant="primary") | |
| with gr.Row(): | |
| with gr.Column(scale=3): | |
| _cb_kwargs = dict(label="대화", height=360, visible=False) | |
| try: | |
| if int(gr.__version__.split(".")[0]) < 6: | |
| _cb_kwargs["type"] = "messages" # 4.x/5.x: 메시지 딕셔너리 쓰려면 필요. 6.0: 인자 자체가 없음(기본 messages) | |
| except Exception: | |
| pass | |
| chatbot = gr.Chatbot(**_cb_kwargs) | |
| # 주제 선택 칩 — 시작 시 표시(주제 고민 줄이기), 첫 메시지 후 숨김 | |
| with gr.Row(visible=False) as topic_row: | |
| tb1 = gr.Button("오늘 있었던 일", size="sm") | |
| tb2 = gr.Button("요즘 신경 쓰이는 거", size="sm") | |
| tb3 = gr.Button("그냥 사는 얘기", size="sm") | |
| tb4 = gr.Button("관계 얘기", size="sm") | |
| # 자기보고(라벨) — 입력창 '바로 위'에 강조 박스로 표시 (스크롤 없이 즉시 눈에 띄도록) | |
| label_radio = gr.Radio(choices=[], label="자기보고", visible=False, elem_id="selfreport_box") | |
| fit_radio = gr.Radio(choices=[], label="시스템 측정이 맞나요?", visible=False, elem_id="fit_box") | |
| label_status = gr.Markdown("") | |
| with gr.Row(): | |
| msg = gr.Textbox(show_label=False, placeholder="메시지 입력", scale=8) | |
| send_btn = gr.Button("입력", variant="primary", scale=1, min_width=64) | |
| with gr.Row(): | |
| save_check_btn = gr.Button("💾 내 데이터 저장 상태 확인", size="sm", scale=1) | |
| save_check_status = gr.Markdown("") | |
| with gr.Row(visible=False) as helper_row: | |
| qb_other = gr.Button("🔄 다른 얘기", size="sm") | |
| end_btn = gr.Button("✋ 이제 그만할래", size="sm", variant="secondary") | |
| with gr.Column(scale=2): | |
| profile_md = gr.HTML("<div style='color:#999;padding:6px;font-family:system-ui;'>대화를 시작하면 여기에 측정 나침반이 보여요.</div>") | |
| stats_md = gr.Markdown(store.summary_md()) | |
| with gr.Accordion("도움이 필요하면 — 상담 연락처", open=False): | |
| gr.Markdown("이 앱은 자기이해를 돕는 실험 도구이며 상담·치료가 아닙니다. " | |
| "힘들거나 위기라고 느껴지면 연락하세요 — **자살예방 109**(24시간) · **정신건강 1577-0199** · 긴급 **119·112**.") | |
| with gr.Accordion("🔬 연구 파일럿 (비밀번호 필요)", open=False): | |
| gr.Markdown( | |
| "연구 참가자 전용입니다. 입장하면 **방금 보낸 발화**에 대해 네 축(기분·에너지·사고방식·자율성)을 " | |
| "직접 5단계로 평가해 저장합니다. 측정값(KoSimCSE)·자기 라벨·LLM(Gemini) 채점이 함께 기록되며, " | |
| "**발화 원문은 저장되지 않습니다**(길이만). 저장은 상단 동의 체크가 있어야 합니다.") | |
| pilot_state = gr.State(False) | |
| with gr.Row(): | |
| pilot_pw = gr.Textbox(label="비밀번호", type="password", scale=3) | |
| pilot_enter = gr.Button("입장", scale=1, min_width=80) | |
| pilot_status = gr.Markdown("") | |
| with gr.Group(visible=False) as pilot_panel: | |
| gr.Markdown("**직전에 보낸 발화**를 떠올리며, 그때의 상태를 골라 주세요.") | |
| sr_eva = gr.Radio(choices=PILOT_AXES["EVA"][1], label="기분 — 그 말을 할 때 내 기분은? (감정의 좋고/나쁨)") | |
| sr_ear = gr.Radio(choices=PILOT_AXES["EAR"][1], label="에너지 — 그때 내 에너지·긴장 수준은? (기분 좋고나쁨과 무관, 에너지만)") | |
| sr_rei = gr.Radio(choices=PILOT_AXES["REI"][1], label="사고방식 — 그 결정/생각은 무엇으로? (직감 ↔ 근거)") | |
| sr_val = gr.Radio(choices=PILOT_AXES["VAL"][1], label="자율성 — 그건 누구의 기준으로? (주변 ↔ 나)") | |
| sr_act = gr.Radio(choices=PILOT_ACT[1], label="🧪 신체 활성도 (선택·실험) — 그때 몸이 얼마나 격렬했나/소진됐나? (에너지 높낮이와 별개: 차분한 평온 ↔ 격렬·탈진)") | |
| pilot_save = gr.Button("이번 발화 자기라벨 저장", variant="primary") | |
| pilot_save_status = gr.Markdown("") | |
| chat_outputs = [chatbot, prof_state, profile_md, msg, label_radio, label_status, stats_md, fit_radio] | |
| def _hide_topics(): return gr.update(visible=False) | |
| onboard_btn.click(do_onboard, onboard_inputs + [consent, participant, session_state, prof_state], | |
| [prof_state, profile_md, chatbot, topic_row, helper_row, stats_md]) | |
| msg.submit(do_chat, [msg, chatbot, prof_state, session_state, consent], chat_outputs).then(_hide_topics, None, topic_row) | |
| send_btn.click(do_chat, [msg, chatbot, prof_state, session_state, consent], chat_outputs).then(_hide_topics, None, topic_row) | |
| save_check_btn.click(check_save_status, [prof_state, consent, session_state], save_check_status) | |
| def quick_send(text): | |
| def _fn(chat, profile, session, consent): | |
| return do_chat(text, chat, profile, session, consent) | |
| return _fn | |
| # 주제 칩 → 그 주제로 가볍게 대화 시작 (클릭 후 칩 숨김) | |
| ti = [chatbot, prof_state, session_state, consent] | |
| tb1.click(quick_send("오늘 있었던 일 얘기해볼까"), ti, chat_outputs).then(_hide_topics, None, topic_row) | |
| tb2.click(quick_send("요즘 좀 신경 쓰이는 일이 있어"), ti, chat_outputs).then(_hide_topics, None, topic_row) | |
| tb3.click(quick_send("그냥 사는 얘기나 해볼까"), ti, chat_outputs).then(_hide_topics, None, topic_row) | |
| tb4.click(quick_send("사람들이랑 지내는 얘기 좀 해볼까"), ti, chat_outputs).then(_hide_topics, None, topic_row) | |
| qb_other.click(quick_send("다른 얘기 하자"), ti, chat_outputs).then(_hide_topics, None, topic_row) | |
| end_btn.click(do_end, [chatbot, prof_state], | |
| [chatbot, msg, topic_row, helper_row, label_radio, label_status, stats_md, fit_radio]) | |
| label_radio.input(do_label, [label_radio, prof_state, consent, session_state], | |
| [label_status, stats_md, label_radio, fit_radio, prof_state]) | |
| fit_radio.input(do_fit, [fit_radio, prof_state, consent, session_state], | |
| [label_status, fit_radio, prof_state]) | |
| d1.change(on_domain, [d1], [happy_cg, barrier_cg]) | |
| interests_btn.click(save_interests, [d1, d2, happy_cg, barrier_cg, consent, session_state, prof_state], | |
| [interests_status, prof_state]) | |
| pilot_enter.click(pilot_unlock, [pilot_pw], [pilot_state, pilot_status, pilot_panel]) | |
| pilot_save.click(pilot_save_fn, | |
| [sr_eva, sr_ear, sr_rei, sr_val, sr_act, prof_state, session_state, consent, pilot_state], | |
| [pilot_save_status, sr_eva, sr_ear, sr_rei, sr_val, sr_act]) | |
| # 각 사용자 접속(페이지 로드)마다 고유 세션 ID 생성 — 사용자 간 세션 분리 | |
| app.load(lambda: str(uuid.uuid4())[:8], outputs=[session_state]) | |
| return app | |
| def render_profile(profile): | |
| last = profile.get("last") | |
| cum = profile.get("cum", {}) | |
| sm = profile.get("smooth", {}) | |
| if not last: | |
| return "<div style='color:#999;padding:6px;font-family:system-ui;'>대화를 시작하면 여기에 측정 나침반이 보여요.</div>" | |
| def axis_bar(x, name, neg_pole, pos_pole, trusted): | |
| x = 0.0 if x is None else x | |
| c = max(-2.5, min(2.5, x)) | |
| width = abs(c) / 2.5 * 50.0 # 트랙의 절반(50%)이 한쪽 최대 | |
| color = "#3b82f6" if trusted else "#94a3b8" | |
| opacity = "1" if trusted else "0.5" | |
| fill = (f"left:50%;width:{width:.0f}%;" if c >= 0 else f"right:50%;width:{width:.0f}%;") | |
| tag = "" if trusted else " <span style='font-size:10px;color:#aaa;'>(방향 위주)</span>" | |
| return ( | |
| "<div style='margin:7px 0;'>" | |
| "<div style='display:flex;justify-content:space-between;font-size:12px;color:#777;'>" | |
| f"<span>{neg_pole}</span><span style='font-weight:600;color:#333;'>{name}{tag}</span><span>{pos_pole}</span></div>" | |
| "<div style='position:relative;height:10px;background:#eee;border-radius:5px;margin-top:3px;'>" | |
| "<div style='position:absolute;left:50%;top:0;bottom:0;width:1px;background:#bbb;'></div>" | |
| f"<div style='position:absolute;top:0;bottom:0;{fill}background:{color};border-radius:5px;opacity:{opacity};'></div>" | |
| "</div></div>" | |
| ) | |
| bars = (axis_bar(sm.get("EVA"), "기분", "슬픔", "즐거움", True) | |
| + axis_bar(sm.get("EAR"), "에너지", "차분", "들뜸", False) | |
| + axis_bar(cum.get("VAL"), "자율성", "맞춤", "내 뜻", False)) | |
| tr = emotion_trend(profile) | |
| trend = " · ↗ 긍정 흐름" if tr > 0.1 else (" · ↘ 가라앉는 흐름" if tr < -0.1 else " · → 안정") | |
| return ( | |
| "<div style='font-family:system-ui;max-width:430px;'>" | |
| "<div style='font-weight:600;margin-bottom:6px;color:#444;'>🧭 측정 나침반 (실시간)</div>" | |
| f"{bars}" | |
| f"<div style='font-size:11px;color:#aaa;margin-top:6px;'>측정값일 뿐 진단이 아니에요{trend}</div>" | |
| "</div>" | |
| ) | |
| if __name__ == "__main__": | |
| build_app().launch() |