Spaces:
Sleeping
Sleeping
| import os | |
| import re | |
| import glob | |
| import tempfile | |
| from typing import Dict, List, TypedDict, Optional, Tuple, Set, Any, Union | |
| from dataclasses import dataclass | |
| from enum import Enum | |
| import numpy as np | |
| import pandas as pd | |
| from langchain_openai import ChatOpenAI, OpenAIEmbeddings | |
| from langchain_community.vectorstores import FAISS | |
| from langchain.text_splitter import RecursiveCharacterTextSplitter | |
| from langchain.schema import Document | |
| from langgraph.graph import StateGraph, END | |
| import json | |
| from datetime import datetime | |
| import logging | |
| import streamlit as st | |
| from streamlit_lottie import st_lottie | |
| import requests | |
| # 로깅 설정 | |
| logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') | |
| logger = logging.getLogger(__name__) | |
| # ========== 데이터 모델 정의 ========== | |
| class DiseaseStage(Enum): | |
| """신장 질환 단계""" | |
| CKD_1 = "CKD Stage 1" | |
| CKD_2 = "CKD Stage 2" | |
| CKD_3 = "CKD Stage 3" | |
| CKD_4 = "CKD Stage 4" | |
| CKD_5 = "CKD Stage 5" | |
| DIALYSIS = "Dialysis" | |
| TRANSPLANT = "Transplant" | |
| class TaskType(Enum): | |
| """질문 유형 분류""" | |
| DIET_RECOMMENDATION = "diet_recommendation" # 식단 추천 | |
| DIET_ANALYSIS = "diet_analysis" # 특정 식품 분석 | |
| MEDICATION = "medication" # 복약 관련 | |
| LIFESTYLE = "lifestyle" # 생활 관리 | |
| DIAGNOSIS = "diagnosis" # 진단/검사 | |
| EXERCISE = "exercise" # 운동 | |
| GENERAL = "general" # 일반 정보 | |
| class PatientConstraints: | |
| """환자 개별 제약조건""" | |
| egfr: float # 사구체여과율 | |
| disease_stage: DiseaseStage | |
| on_dialysis: bool | |
| comorbidities: List[str] # 동반질환 목록 | |
| medications: List[str] # 복용 약물 목록 | |
| age: int | |
| gender: str | |
| # 영양 제한사항 | |
| protein_restriction: Optional[float] = None # g/day | |
| sodium_restriction: Optional[float] = None # mg/day | |
| potassium_restriction: Optional[float] = None # mg/day | |
| phosphorus_restriction: Optional[float] = None # mg/day | |
| fluid_restriction: Optional[float] = None # ml/day | |
| calorie_target: Optional[float] = None # kcal/day | |
| class RecommendationItem: | |
| """추천 항목""" | |
| name: str | |
| category: str # 식이, 운동, 약물 등 | |
| description: str | |
| constraints_satisfied: bool | |
| embedding: Optional[np.ndarray] = None | |
| class FoodItem: | |
| """식품 정보 (실제 CSV 구조 반영)""" | |
| food_code: str # 식품코드 | |
| name: str # 식품명 | |
| food_category_major: str # 식품대분류명 | |
| food_category_minor: str # 식품중분류명 | |
| serving_size: float # 영양성분함량기준량 (보통 100g) | |
| calories: float # 에너지(kcal) | |
| water: float # 수분(g) | |
| protein: float # 단백질(g) | |
| fat: float # 지방(g) | |
| carbohydrate: float # 탄수화물(g) | |
| sugar: float # 당류(g) | |
| dietary_fiber: float # 식이섬유(g) | |
| calcium: float # 칼슘(mg) | |
| iron: float # 철(mg) | |
| phosphorus: float # 인(mg) | |
| potassium: float # 칼륨(mg) | |
| sodium: float # 나트륨(mg) | |
| cholesterol: float # 콜레스테롤(mg) | |
| saturated_fat: float # 포화지방산(g) | |
| def get_nutrients_per_serving(self, serving_g: float = 100) -> Dict[str, float]: | |
| """지정된 양(g)에 대한 영양소 함량 계산""" | |
| ratio = serving_g / self.serving_size | |
| return { | |
| 'calories': self.calories * ratio, | |
| 'protein': self.protein * ratio, | |
| 'fat': self.fat * ratio, | |
| 'carbohydrate': self.carbohydrate * ratio, | |
| 'sodium': self.sodium * ratio, | |
| 'potassium': self.potassium * ratio, | |
| 'phosphorus': self.phosphorus * ratio | |
| } | |
| def is_suitable_for_patient(self, constraints: PatientConstraints, | |
| serving_g: float = 100) -> Tuple[bool, List[str]]: | |
| """환자 제약조건에 적합한지 확인""" | |
| issues = [] | |
| nutrients = self.get_nutrients_per_serving(serving_g) | |
| # 일일 제한량의 30%를 한 끼 기준으로 설정 | |
| meal_ratio = 0.3 | |
| # 단백질 체크 | |
| if constraints.protein_restriction: | |
| if nutrients['protein'] > constraints.protein_restriction * meal_ratio: | |
| issues.append(f"단백질 함량이 높음 ({nutrients['protein']:.1f}g)") | |
| # 나트륨 체크 | |
| if constraints.sodium_restriction: | |
| if nutrients['sodium'] > constraints.sodium_restriction * meal_ratio: | |
| issues.append(f"나트륨 함량이 높음 ({nutrients['sodium']:.0f}mg)") | |
| # 칼륨 체크 | |
| if constraints.potassium_restriction: | |
| if nutrients['potassium'] > constraints.potassium_restriction * meal_ratio: | |
| issues.append(f"칼륨 함량이 높음 ({nutrients['potassium']:.0f}mg)") | |
| # 인 체크 | |
| if constraints.phosphorus_restriction: | |
| if nutrients['phosphorus'] > constraints.phosphorus_restriction * meal_ratio: | |
| issues.append(f"인 함량이 높음 ({nutrients['phosphorus']:.0f}mg)") | |
| return len(issues) == 0, issues | |
| # ========== State 정의 ========== | |
| class GraphState(TypedDict): | |
| """LangGraph State""" | |
| user_query: str | |
| patient_constraints: PatientConstraints | |
| task_type: TaskType | |
| draft_response: str | |
| draft_items: List[RecommendationItem] | |
| corrected_items: List[RecommendationItem] | |
| final_response: str | |
| catalog_results: List[Document] | |
| iteration_count: int | |
| error: Optional[str] | |
| food_analysis_results: Optional[Dict[str, Any]] | |
| recommended_foods: Optional[List[FoodItem]] | |
| meal_plan: Optional[Dict[str, List[FoodItem]]] | |
| current_node: str # 현재 처리 중인 노드 | |
| processing_log: List[str] # 처리 로그 | |
| # ========== Catalog 관리 ========== | |
| class KidneyDiseaseCatalog: | |
| """신장질환 정보 카탈로그 - 싱글톤 패턴 적용""" | |
| _instance = None | |
| _initialized = False | |
| def __new__(cls, *args, **kwargs): | |
| if cls._instance is None: | |
| cls._instance = super(KidneyDiseaseCatalog, cls).__new__(cls) | |
| return cls._instance | |
| def __init__(self, documents_path: str = "./data"): | |
| if KidneyDiseaseCatalog._initialized: | |
| return | |
| self.embeddings = OpenAIEmbeddings() | |
| self.vectorstore = None | |
| self.documents_path = documents_path | |
| self.metadata_index = {} # 문서 메타데이터 인덱스 | |
| # 태그 매핑 정의 | |
| self.field_mapping = { | |
| "식이": "diet", "운동": "exercise", "진단": "diagnosis", | |
| "복약": "medication", "치료": "treatment", "교육": "education", | |
| "생활": "lifestyle" | |
| } | |
| self.status_mapping = { | |
| "CKD": "chronic_kidney_disease", "HD": "hemodialysis", | |
| "PD": "peritoneal_dialysis", "DIA": "dialysis", | |
| "TX": "transplant", "ALL": "all" | |
| } | |
| self.level_mapping = { | |
| "COM": "common", "STD": "standard", "DM": "diabetes", | |
| "HTN": "hypertension", "OLD": "elderly", "PREG": "pregnancy", | |
| "OBES": "obesity", "SYM": "symptom" | |
| } | |
| self.priority_mapping = { | |
| "S1": "emergency", "S2": "caution", "S3": "general", "S4": "reference" | |
| } | |
| # 초기화 시 문서 로드 | |
| self.load_documents() | |
| KidneyDiseaseCatalog._initialized = True | |
| def parse_filename_tags(self, filename: str) -> Dict[str, str]: | |
| """파일명에서 태그 파싱""" | |
| pattern = r'\[([^-]+)-([^-]+)-([^-]+)-([^\]]+)\]' | |
| match = re.search(pattern, filename) | |
| if match: | |
| field, status, level, priority = match.groups() | |
| return { | |
| "field": self.field_mapping.get(field, field), | |
| "patient_status": self.status_mapping.get(status, status), | |
| "personalization_level": self.level_mapping.get(level, level), | |
| "safety_priority": self.priority_mapping.get(priority, priority), | |
| "raw_tags": f"{field}-{status}-{level}-{priority}" | |
| } | |
| return {} | |
| def load_documents(self): | |
| """권위있는 기관의 문서들을 로드""" | |
| if self.vectorstore is not None: | |
| logger.info("Documents already loaded") | |
| return | |
| documents = [] | |
| # data 폴더의 모든 txt 파일 로드 | |
| file_pattern = os.path.join(self.documents_path, "*.txt") | |
| file_paths = glob.glob(file_pattern) | |
| if not file_paths: | |
| logger.warning(f"No documents found in {self.documents_path}. Creating sample files...") | |
| file_paths = self._create_comprehensive_sample_files() | |
| for file_path in file_paths: | |
| try: | |
| filename = os.path.basename(file_path) | |
| tags = self.parse_filename_tags(filename) | |
| with open(file_path, 'r', encoding='utf-8') as f: | |
| content = f.read() | |
| title_pattern = r'^#\s*(.+)$' | |
| title_match = re.search(title_pattern, content, re.MULTILINE) | |
| if title_match: | |
| title = title_match.group(1) | |
| else: | |
| title = filename.split(']')[-1].replace('.txt', '').strip() | |
| if not title: | |
| title = filename.replace('.txt', '') | |
| source = self._extract_source(content, filename) | |
| doc = Document( | |
| page_content=content, | |
| metadata={ | |
| "filename": filename, | |
| "title": title, | |
| "source": source, | |
| "timestamp": datetime.now().isoformat(), | |
| **tags | |
| } | |
| ) | |
| documents.append(doc) | |
| self.metadata_index[filename] = doc.metadata | |
| logger.info(f"Loaded document: {filename}") | |
| except Exception as e: | |
| logger.error(f"Error loading file {file_path}: {e}") | |
| continue | |
| # 텍스트 분할 | |
| text_splitter = RecursiveCharacterTextSplitter( | |
| chunk_size=2000, | |
| chunk_overlap=100, | |
| separators=["\n\n", "\n", ".", "!", "?", ",", " ", ""] | |
| ) | |
| split_docs = text_splitter.split_documents(documents) | |
| for doc in split_docs: | |
| doc.metadata["chunk_id"] = f"{doc.metadata['filename']}_{hash(doc.page_content)}" | |
| self.vectorstore = FAISS.from_documents(split_docs, self.embeddings) | |
| logger.info(f"Loaded {len(documents)} documents ({len(split_docs)} chunks) into vectorstore") | |
| def _extract_source(self, content: str, filename: str) -> str: | |
| """문서 내용에서 출처 기관 추출""" | |
| source_patterns = [ | |
| "보건복지부", "질병관리청", "대한의학회", "대한신장학회", | |
| "대한당뇨병학회", "대한의료사회복지사협회" | |
| ] | |
| for pattern in source_patterns: | |
| if pattern in content: | |
| return pattern | |
| return "관련 기관" | |
| def _create_comprehensive_sample_files(self) -> List[str]: | |
| """포괄적인 샘플 파일 생성""" | |
| sample_files = [] | |
| samples = [ | |
| { | |
| "filename": "[식이-CKD-STD-S3] 만성콩팥병 환자의 단백질 섭취 가이드.txt", | |
| "content": """# 만성콩팥병 환자의 단백질 섭취 가이드 | |
| ## 개요 | |
| 만성콩팥병(CKD) 환자의 적절한 단백질 섭취는 질병 진행을 늦추고 영양 상태를 유지하는 데 중요합니다. | |
| ## 단계별 단백질 섭취 권장량 | |
| - CKD 1-2단계: 정상 섭취 (체중 kg당 0.8-1.0g) | |
| - CKD 3-4단계: 제한 필요 (체중 kg당 0.6-0.8g) | |
| - CKD 5단계(투석 전): 엄격한 제한 (체중 kg당 0.6g) | |
| - 혈액투석 환자: 증가 필요 (체중 kg당 1.2g) | |
| - 복막투석 환자: 더 증가 필요 (체중 kg당 1.2-1.3g) | |
| ## 양질의 단백질 선택 | |
| 1. 동물성 단백질: 달걀, 생선, 닭가슴살 | |
| 2. 식물성 단백질: 두부, 콩류 (인 함량 주의) | |
| ## 주의사항 | |
| - 과도한 단백질 섭취는 신장에 부담을 줍니다 | |
| - 개인별 상태에 따라 섭취량 조절이 필요합니다 | |
| - 정기적인 영양 상담을 받으세요 | |
| 출처: 대한신장학회""" | |
| }, | |
| { | |
| "filename": "[복약-HD-HTN-S2] 혈액투석 환자의 고혈압 약물 관리.txt", | |
| "content": """# 혈액투석 환자의 고혈압 약물 관리 | |
| ## 주요 원칙 | |
| 혈액투석 환자의 약 70-80%가 고혈압을 동반하며, 적절한 약물 관리가 필수적입니다. | |
| ## 복약 시간 조절 | |
| 1. 투석 후 복용 권장 약물 | |
| - ACE 억제제, ARB: 투석으로 제거될 수 있음 | |
| - 베타차단제: 투석 중 저혈압 위험 | |
| 2. 투석과 무관하게 복용 가능한 약물 | |
| - 칼슘채널차단제: 투석으로 제거되지 않음 | |
| ## 약물 상호작용 주의 | |
| - 인결합제와 다른 약물은 최소 2시간 간격 | |
| - 철분제와 일부 항생제는 동시 복용 금지 | |
| ## 혈압 목표 | |
| - 투석 전: 140/90 mmHg 미만 | |
| - 투석 후: 130/80 mmHg 미만 | |
| 출처: 대한신장학회""" | |
| }, | |
| { | |
| "filename": "[식이-HD-STD-S2] 혈액투석 환자의 칼륨 제한 식이요법.txt", | |
| "content": """# 혈액투석 환자의 칼륨 제한 식이요법 | |
| ## 칼륨 제한의 중요성 | |
| 혈액투석 환자는 소변량 감소로 칼륨 배설이 어려워 고칼륨혈증 위험이 높습니다. | |
| ## 일일 칼륨 섭취 권장량 | |
| - 혈액투석 환자: 2000-2500mg/일 | |
| - 잔여 신기능에 따라 조절 필요 | |
| ## 고칼륨 식품 (제한 필요) | |
| - 과일: 바나나, 참외, 토마토, 오렌지 | |
| - 채소: 시금치, 감자, 고구마, 버섯 | |
| - 기타: 초콜릿, 견과류, 우유 | |
| ## 칼륨 감소 조리법 | |
| 1. 채소는 잘게 썰어 물에 2시간 담근 후 헹구기 | |
| 2. 끓는 물에 데친 후 국물은 버리기 | |
| 3. 과일은 통조림 사용 (시럽 제거) | |
| 출처: 보건복지부""" | |
| }, | |
| { | |
| "filename": "[생활-CKD-STD-S3] 만성콩팥병 환자의 수분 섭취 관리.txt", | |
| "content": """# 만성콩팥병 환자의 수분 섭취 관리 | |
| ## 수분 제한이 필요한 경우 | |
| - 소변량이 하루 500ml 이하로 감소 | |
| - 부종이 있는 경우 | |
| - 심부전을 동반한 경우 | |
| ## 일일 수분 섭취량 계산 | |
| - 기본 공식: 전날 소변량 + 500ml | |
| - 투석 환자: 투석 간 체중 증가 1kg 이내 | |
| ## 수분 섭취 관리 요령 | |
| 1. 모든 액체류 포함 (국, 우유, 아이스크림 등) | |
| 2. 작은 컵 사용하기 | |
| 3. 얼음 조각으로 갈증 해소 | |
| 4. 무설탕 껌이나 신 사탕 활용 | |
| ## 주의사항 | |
| - 과도한 수분 제한도 위험 | |
| - 개인별 상태에 따라 조절 | |
| - 정기적인 체중 측정 필요 | |
| 출처: 대한의학회""" | |
| }, | |
| { | |
| "filename": "[운동-CKD-STD-S3] 만성콩팥병 환자의 운동 가이드.txt", | |
| "content": """# 만성콩팥병 환자의 운동 가이드 | |
| ## 운동의 이점 | |
| - 심혈관 기능 개선 | |
| - 혈압 조절 | |
| - 근력 유지 | |
| - 우울감 감소 | |
| ## 권장 운동 | |
| 1. 유산소 운동 | |
| - 걷기: 주 5회, 30분 | |
| - 자전거: 저강도로 시작 | |
| - 수영: 관절에 무리 없음 | |
| 2. 근력 운동 | |
| - 가벼운 덤벨 운동 | |
| - 저항 밴드 운동 | |
| - 주 2-3회, 15-20분 | |
| ## 운동 시 주의사항 | |
| - 투석 직후는 피하기 | |
| - 탈수 주의 | |
| - 가슴 통증, 호흡곤란 시 즉시 중단 | |
| - 운동 전후 혈압 체크 | |
| 출처: 대한의료사회복지사협회""" | |
| }, | |
| { | |
| "filename": "[진단-CKD-STD-S3] 만성콩팥병의 진단과 검사.txt", | |
| "content": """# 만성콩팥병의 진단과 검사 | |
| ## 진단 기준 | |
| 3개월 이상 다음 중 하나 이상 존재 시: | |
| - eGFR < 60 ml/min/1.73m² | |
| - 알부민뇨 (ACR ≥ 30mg/g) | |
| - 신장 손상의 증거 | |
| ## 주요 검사 | |
| 1. 혈액검사 | |
| - 크레아티닌, eGFR | |
| - 전해질 (Na, K, Ca, P) | |
| - 빈혈 지표 (Hb, ferritin) | |
| 2. 소변검사 | |
| - 단백뇨/알부민뇨 | |
| - 현미경 검사 | |
| 3. 영상검사 | |
| - 신장 초음파 | |
| - 필요시 CT, MRI | |
| ## 정기 검진 주기 | |
| - CKD 1-2단계: 연 1회 | |
| - CKD 3단계: 6개월마다 | |
| - CKD 4-5단계: 3개월마다 | |
| 출처: 질병관리청""" | |
| }, | |
| { | |
| "filename": "[식이-CKD-DM-S2] 당뇨병성 신증 환자의 식사 관리.txt", | |
| "content": """# 당뇨병성 신증 환자의 식사 관리 | |
| ## 특별 고려사항 | |
| 당뇨병과 신장병을 함께 관리해야 하는 복잡한 상황입니다. | |
| ## 영양 관리 원칙 | |
| 1. 혈당 조절 | |
| - 규칙적인 식사 시간 | |
| - 당지수가 낮은 식품 선택 | |
| - 단순당 제한 | |
| 2. 단백질 조절 | |
| - CKD 3-4단계: 0.6-0.8g/kg/일 | |
| - 양질의 단백질 위주 | |
| 3. 나트륨 제한 | |
| - 2000mg/일 이하 | |
| - 가공식품 피하기 | |
| ## 주의 식품 | |
| - 과일: 당도 높은 과일 제한 | |
| - 곡류: 현미, 잡곡 (인 함량 주의) | |
| - 음료: 과일주스, 스포츠음료 금지 | |
| 출처: 대한당뇨병학회""" | |
| }, | |
| { | |
| "filename": "[식이-ALL-COM-S4] 식사가 건강에 미치는 영향.txt", | |
| "content": """# 식사가 건강에 미치는 영향 | |
| # 이대서울병원 신장내과 류동열 | |
| Key Message: 우리 몸에 나쁜 음식은 결코 없습니다. 골고루 적당량을 섭취하면 식사가 보약입니다. | |
| 건강을 유지하기 위해서는 식사를 통해 우리 몸과 마음이 필요한 것을 적절한 시간에 적당한 양만큼 얻을 수 있어야만 합니다. 건강한 식사행동이 질병과 사망에 미치는 영향을 조사한 연구에 따르면 건강을 해치는 15가지 나쁜 식사행동과 일반인에게 권장되는 적절한 양은 다음과 같습니다: | |
| | 번호 | 나쁜 식사행동 | 일일 권장량 (권장 범위) | | |
| | -- | --------------------------- | -------------------- | | |
| | 1 | 과일을 적게 먹는 것 | 250 g (200–300) | | |
| | 2 | 채소를 적게 먹는 것 | 360 g (290–430) | | |
| | 3 | 콩류를 적게 먹는 것 | 60 g (50–70) | | |
| | 4 | 정백하지 않은 통알곡을 적게 먹는 것 | 125 g (100–150) | | |
| | 5 | 견과류를 적게 먹는 것 | 21 g (16–25) | | |
| | 6 | 우유를 적게 먹는 것 | 435 g (350–520) | | |
| | 7 | 붉은 살코기를 많이 먹는 것 | 23 g (18–27) | | |
| | 8 | 가공육류를 많이 먹는 것 | 2 g (0–4) | | |
| | 9 | 설탕이 첨가된 음료수를 많이 먹는 것 | 3 g (0–5) | | |
| | 10 | 식이섬유를 적게 먹는 것 | 24 g (19–28) | | |
| | 11 | 칼슘을 적게 먹는 것 | 1.25 g (1.00–1.50) | | |
| | 12 | 해산물에 포함된 오메가-3 지방산을 적게 먹는 것 | 250 mg (200–300) | | |
| | 13 | 다가불포화지방산을 적게 먹는 것 | 총 열량의 11% (9–13) | | |
| | 14 | 트랜스지방산을 많이 먹는 것 | 총 열량의 0.5% (0.0–1.0) | | |
| | 15 | 염분을 많이 먹는 것 | 3 g (1–5) | | |
| 전 세계 사람들이 사망하는 원인 중 22%가 이처럼 나쁜 식사행동과 관련이 있으며, 특히 우리나라에서는 과일과 통알곡을 적게 먹고 염분 섭취가 많은 것이 사망과 관련된 나쁜 식사행동 중 가장 주요한 것들이었습니다. | |
| 만성콩팥병 환자라고 하더라도 과일과 채소 섭취량을 적절하게 섭취하는 균형 잡힌 식사를 하면 사망 위험을 줄여준다는 연구결과도 있습니다. | |
| 출처: 이대서울병원 신장내과""" | |
| }, | |
| { | |
| "filename": "[식이-ALL-STD-S2] 만성콩팥병 환자의 칼륨 제한 주의사항.txt", | |
| "content": """# 만성콩팥병 환자는 칼륨이 많이 들어있는 과일이나 채소를 지나치게 섭취하지 않도록 주의합니다 | |
| 콩팥 기능이 저하된 만성콩팥병 환자는 칼륨 배설기능이 저하되어 있으므로 과일류, 채소류, 콩류, 견과류의 섭취량을 줄여야 합니다. | |
| 칼륨이 적게 들어 있는 음식을 골라 먹는 법과 칼륨을 낮추는 조리법을 배워 실천해야 합니다. | |
| 고칼륨혈증이 발견된 경우 칼륨을 올릴 수 있는 약물에 대해 확인이 필요하기도 합니다. | |
| ## 관련 상식 | |
| 칼륨은 식품에 널리 들어 있지만 주요 공급원은 과일류, 채소류, 콩류, 견과류입니다. | |
| 콩팥 기능이 정상인 경우 이러한 식품은 섬유질, 비타민, 미네랄, 기타 중요한 영양소의 주요 공급원이므로 과일류, 채소류, 콩류, 견과류 섭취를 제한하지 않습니다. | |
| 하지만 만성콩팥병 환자가 남아 있는 콩팥 기능에 비해 많은 양의 칼륨을 섭취하면 혈액 속의 칼륨 수치가 지나치게 높아지는 고칼륨혈증이 유발되어 근육쇠약감, 부정맥 등이 발생할 수 있으며, 심하면 심장마비 같은 치명적이고 위급한 합병증을 유발할 수 있습니다. | |
| ## 실천 방법 | |
| 만성콩팥병 환자는 콩팥 기능 감소에 따라 담당의와 상의하여 과일류, 채소류, 콩류, 견과류의 섭취량을 줄여야 합니다. | |
| 채소는 따뜻한 물에 2시간 이상 담가 놓았다가 새 물에 몇 번 헹군 후 섭취합니다. | |
| 채소는 물에 삶거나 데친 후 물은 버리고 채소만 섭취합니다. | |
| 저나트륨 소금은 소금의 주성분인 나트륨 일부를 칼륨으로 대체한 소금이라 콩팥 기능이 나쁘면 오히려 고칼륨혈증을 일으킬 수 있어 주의해야 합니다. | |
| 출처: 나와 가족을 위한 만성콩팥병 예방과 관리 정보""" | |
| }, | |
| { | |
| "filename": "[식이-HD-STD-S3] 외식하고 싶은데 무엇을 먹을 수 있나요.txt", | |
| "content": """# 외식하고 싶은데 무엇을 먹을 수 있나요? | |
| 여의도성모병원 영양팀, 한국임상영양학회, 영양사 박주연 | |
| Key Message: 미리 계획하고 염분이 적으면서 균형된 메뉴를 선택합니다. | |
| 최근 1인 가구 증가 및 서구화된 식습관 등 식생활에 많은 변화가 이뤄지고 있으며, 식생활에 있어서 외식이 차지하는 비율 또한 높아지고 있는 추세입니다. 사회생활을 병행하는 혈액투석 환자에게서도 회식, 약속 등으로 인한 외식은 피할 수 없는 부분입니다. | |
| ## 1. 외식 시 식사 원칙 | |
| ### 1) 미리 계획 합니다. | |
| ① 외식이 필요한 날은 외식 전, 후 집에서 먹는 식사의 양이나 종류를 평소보다 더욱 주의 깊게 조절합니다. | |
| ② 신선한 재료를 사용하고, 염분 조절 등 개별 주문이 가능한 식당을 선택하는 것이 좋습니다. | |
| ### 2) 적절한 단백질이 포함된 균형 잡힌 메뉴를 선택합니다. | |
| ① 빈혈 예방과 투석시 손실되는 아미노산 보충을 위해 적절한 단백질 섭취가 필요합니다. | |
| ② 과량의 단백질 섭취는 투석 간 노폐물을 축적시킬 수 있으므로, 한 끼에 몰아서 섭취하지 않도록 합니다. | |
| ### 3) 염분, 칼륨, 인 함량이 높은 식품은 피합니다. | |
| ① 집에서 직접 준비한 식사에 비해 외식 메뉴는 염분 함량이 높은 경우가 많습니다. | |
| ② 식사 주문 시 염분을 넣지 않도록 주문하고, 소금 등은 별도로 요청하여 적당량 첨가합니다. | |
| ## 2. 외식 메뉴 선택 및 섭취 요령 | |
| ### 1) 비빔밥, 회덮밥 등 | |
| - 칼륨 함량이 높은 채소는 제외하고, 생채소는 제공량의 절반만 섭취합니다. | |
| - 염분조절을 위해 고추장, 간장 등 양념을 최소한으로 사용합니다. | |
| ### 2) 갈비탕, 설렁탕 등 탕류 | |
| - 소금을 추가로 넣지 않고, 건더기 위주로 섭취합니다. | |
| ### 3) 칼국수, 비빔국수, 냉면 등 면류 | |
| - 염분 함량이 높아 주의가 필요합니다. | |
| - 국물이 있는 면류는 국물은 먹지 않고, 비빔양념은 최소량만 사용합니다. | |
| ### 4) 스테이크, 돈까스 | |
| - 제공되는 고기양이 많은 편으로, 한 번에 섭취하는 고기 양을 조절해야 합니다. | |
| - 염분제한을 위해 소스는 가급적이면 뿌리지 않습니다. | |
| ### 5) 파스타, 리조또 | |
| - 오일로 조리된 메뉴를 선택합니다. | |
| - 인 함량이 높은 크림소스나, 칼륨/염분이 높은 토마토소스는 절반만 섭취합니다. | |
| 출처: 여의도성모병원 영양팀, 한국임상영양학회""" | |
| }, | |
| { | |
| "filename": "[운동-ALL-STD-S2] 운동을 시작하기 전 주의사항이 있나요.txt", | |
| "content": """# 운동을 시작하기 전 주의사항이 있나요? | |
| 구미차병원 신장내과, 대한신장학회 근육감소증 및 여림 연구회 김준철 | |
| ## 1. 유산소 운동을 해야 해요? 근력 운동을 해야 해요? | |
| 한 마디로 얘기하자면 두 가지 운동 모두가 필요하고 또 중요합니다. 만성 콩팥병 환자들에게 있어 만성콩팥병의 원인이 되는 당뇨병이나 고혈압 그리고 합병증으로 동반되는 여러 심혈관계 질환의 위험 인자들을 조절하거나 예방 혹은 치료하는 데 유산소운동은 큰 도움이 됩니다. | |
| 그리고 만성콩팥병 환자들에게서 흔히 볼 수 있는 단백질-에너지 소모(Protein Energy Wasting), 근감소증(sarcopenia), 그리고 노쇠(Frailty)로 인한 일상 생활의 장애 및 그로 인한 부작용들을 예방 혹은 치료하는 데 있어 지속적인 근력 운동은 특별히 더 큰 도움이 되므로 두 가지 형태의 운동을 함께 유지하는 것이 가장 좋습니다. | |
| ## 2. 유산소 운동과 근력 운동 모두 공통적으로 준비운동과 정리운동이 필요합니다. | |
| 본격적인 운동 시작 전에는 근육과 인대 그리고 심장에 갑작스런 부담으로 인한 부상이나 부작용을 피하기 위해 준비운동을 시행하는 것이 안전합니다. 일반적으로 5분에서 10분 정도의 시간을 할애하여 가벼운 몸 풀기를 하시면 됩니다. | |
| 본격적인 운동을 마친 후에도 준비 운동과 같은 형태의 가벼운 몸 풀기나, 앞서 실행하였던 같은 종류의 유산소 운동을 "중등도 강도"에서 "가벼운 강도"로 낮춰서 5분에서 10분 정도의 시간을 들여서 정리운동을 해 주는 것이 좋습니다. | |
| ## 3. 운동을 해서는 안 되는 상황 | |
| 다음과 같은 경우는 운동을 피하고 담당의사와 상의하는 것이 좋습니다. | |
| ### 절대적 운동 금기 상황 | |
| - 2일 이내의 급성 심근 경색증 혹은 협심증 | |
| - 불안정성 협심증을 진단받고 치료 중인 경우 | |
| - 조절되지 않는 심각한 종류의 부정맥을 가지고 있는 경우 | |
| - 증상을 동반하는 심한 대동맥 협착증이 있는 경우 | |
| - 조절되지 않는 호흡 곤란의 증상을 동반하는 심부전 | |
| - 급성 폐경색 혹은 색전증 | |
| - 급성 심근염이나 급성 심막염 | |
| - 이미 진단되었거나 의심되는 대동맥 박리증 | |
| - 발열, 전신근육통 혹은 림프염 등을 동반한 급성 전신 감염 상태 | |
| ### 상대적 운동 금기 상황 | |
| - 좌측 주관상동맥 협착증 | |
| - 중등도의 협착성 판막 심장 질환 | |
| - 저칼륨혈증이나 고칼륨혈증과 같은 전해질 이상 | |
| - 조절되지 않은 고혈압(안정시 수축기 혈압 200 mmHg 혹은 이완기 혈압 110 mmHg 이상) | |
| - 증상을 동반하는 빈맥이나 서맥 | |
| - 비후성 심근병증이나 폐쇄성 심장 질환을 진단받은 경우 | |
| - 운동으로 인해 악화 가능성이 있는 신경운동계 혹은 근골격근계 질환을 동반한 경우 | |
| - 조절되지 않은 대사성 질환(예: 당뇨병, 갑상선 기능 항진증) | |
| 출처: 구미차병원 신장내과, 대한신장학회""" | |
| }, | |
| { | |
| "filename": "[운동-DIA-STD-S3] 적절한 근력 운동 방법에 대해 알려주세요.txt", | |
| "content": """# 적절한 근력 운동 방법에 대해 알려주세요. | |
| 구미차병원 신장내과, 대한신장학회 근육감소증 및 여림 연구회 김준철 | |
| ## 1. 운동 횟수(Frequency) | |
| 1. 같은 부위의 근육에 대한 근력 운동은 최소 48시간의 간격을 두는 것이 부상을 최소화 할 수 있습니다. | |
| 2. 매일 근육 운동을 하고자 한다면 운동하고자 하는 근육 부위를 달리하여 이틀에 한 번씩 해당 근육 운동이 차례가 돌아올 수 있도록 하면 됩니다. | |
| 3. 일반적으로 5일 이상 근력 운동을 쉬게 되면 이전 운동의 효과가 없어지기 시작하기 때문에 해당 근육 부위의 운동을 최소 주 2회를 시행하는 것이 근력의 유지 및 향상을 도모할 수 있습니다. | |
| ## 2. 운동 강도(Intensity) | |
| 1. 운동 기구를 이용하여 근력운동을 하는 경우는 특정 무게나 저항을 정하여 해당 부위 근육 운동을 할 때 1회 운동(1 set)을 할 때, 12회-14회를 반복하였을 때 해당 근육이 뻐근함을 느낄 정도로 무게와 저항 정도를 정하는 것이 안전합니다. | |
| 2. 이 때 뻐근함을 넘어서 통증을 느낄 정도의 무게나 저항은 운동 강도가 지나치게 높게 정한 것을 의미하므로 그 정도를 더 낮게 정하여 부상에 유의하셔야 합니다. | |
| 3. 이미 어느 정도의 좋은 근력을 가지고 있는 경우는 더 적은 횟수, 예를 들면 8회-10회를 시행하면 해당 근육의 뻐근함을 느낄 정도로 무게와 저항을 정하여 근력 운동을 시행하기도 하지만 이 경우 부상 위험도는 더 증가할 수 있어 조심스러운 운동 시작이 필요합니다. | |
| ## 3. 운동 시간(Time) | |
| 근력 운동에서의 운동 시간에 해당되는 것은 "운동 강도" 부분에서 설명드린 1회 운동을 총 몇 차례 반복하여 시행하는지에 따라 정해집니다. 근력 운동에서는 1회 운동을 "한 세트(1 set)"라고 표현합니다. | |
| ## 4. 운동량(Volume)과 증량 속도(Progression) | |
| 운동량은 해당 근육의 근력 운동을 한 주간 동안 시행하는 횟수와 시행할 때 적용하는 무게 혹은 저항, 그리고 각각 몇 "세트"를 시행하는 지를 곱한 값으로 결정됩니다. | |
| ### 운동량과 운동 속도는 어떻게 증가시켜야 하나요? | |
| 1. 평균적인 체력을 가지고 있는 환우께서 처음 근력 운동을 시작하는 경우 우선 욕심내지 않고 12회-14회를 반복하였을 때 해당 근육이 뻐근함을 느낄 정도로 무게와 저항 정도를 정하여 가능한 부상을 피하는 것이 가장 중요합니다. | |
| 2. 우선 12회-14회를 무리 없이 반복할 수 있는 무게와 저항을 유지한 채 3분-5분 간격으로 같은 무게 혹은 저항으로 12회-14회를 처음 세트와 마찬가지로 반복하게 합니다. | |
| 3. 이렇게 보통 2-4세트까지 무리 없이 시행할 수 있는 근력을 확보하게 되고, 현재 시행하고 있는 무게나 저항을 한 번에 16회-18회 정도를 쉽게 반복하여 운동할 수 있는 단계에 도달하면 무게나 저항을 현재보다 10% 전후를 기준으로 증가하여 시행합니다. | |
| 4. 일반적으로 2주-4주 전후의 간격이 필요하지만 개인차가 있을 수 있어 근력 운동의 증량 속도는 다양할 수 있습니다. | |
| 5. 무엇보다 부상을 피하는 것이 가장 중요하게 유념해야 할 부분입니다. | |
| 출처: 구미차병원 신장내과, 대한신장학회""" | |
| }, | |
| { | |
| "filename": "[진단-ALL-COM-S4] 건강한 사람에게서 콩팥병을 의심할 수 있는 증상은 무엇이 있나요.txt", | |
| "content": """# 건강한 사람에게서 콩팥병을 의심할 수 있는 증상은 무엇이 있나요? | |
| ## 일반인을 위한 만성콩팥병 바로알기 | |
| 1. 소변에서 거품이 보이면 단백뇨를 의심해야 합니다. | |
| 2. 붉은 소변의 원인은 다양하므로 빠른 시간 내에 진료가 필요합니다. | |
| 3. 소변을 자주 보면 여성의 경우 방광염을, 중년 이후의 남성인 경우 전립선 질환을 먼저 의심해야 합니다. | |
| 4. 옆구리 통증의 원인은 콩팥 질환도 가능하지만 다른 질환일 가능성도 있으므로 검사가 필요합니다. | |
| 5. 아침에 일어났을 때 얼굴이 붓는다면 소변 검사와 혈액 검사를 통하여 콩팥병을 확인해야 합니다. | |
| 6. 임신 중의 부종은 흔한 일이지만 임신과 연관된 합병증인 임신 중독증 혹은 콩팥병을 의심해야 하므로 주기적 산전 진찰이 필요합니다. | |
| 출처: 일반인을 위한 만성콩팥병 바로알기""" | |
| }, | |
| { | |
| "filename": "[진단-ALL-HTN-S4] 고혈압이 콩팥병에 의한 것인지 의심해야 할 경우는 무엇인지요.txt", | |
| "content": """# 고혈압이 콩팥병에 의한 것인지 의심해야 할 경우는 무엇인지요? | |
| ## 일반인을 위한 만성콩팥병 바로알기 | |
| 다음과 같은 경우에 고혈압이 콩팥병에 의한 것인지 의심해 보아야 합니다: | |
| 1. 소변 검사에서 혈뇨나 단백뇨가 동반되는 경우 | |
| 2. 몸이 붓는 증상(부종)이 같이 동반되는 경우 | |
| 3. 염분 섭취량에 따라 혈압이 크게 영향 받을 때 | |
| 4. 35세 이전에 발생한 고혈압 또는 60세 이후에 발생한 고혈압인 경우 | |
| 5. 고혈압이 갑자기 발생할 때 | |
| 6. 혈압이 약물 치료에도 불구하고 잘 조절되지 않을 때 | |
| 7. 잘 조절되던 혈압이 뚜렷한 이유 없이 상승할 때 | |
| 출처: 일반인을 위한 만성콩팥병 바로알기""" | |
| }, | |
| { | |
| "filename": "[진단-ALL-SYM-S2] 혈액 검사에서 나트륨 농도가 낮다고 합니다. 무슨 이야기인가요.txt", | |
| "content": """# 혈액 검사에서 나트륨 농도가 낮다고 합니다. 무슨 이야기인가요? | |
| ## 일반인을 위한 만성콩팥병 바로알기 | |
| 혈액 나트륨 농도가 정상보다 낮아지는 '저나트륨혈증'을 말하며, 이는 노인에게 가장 흔하게 발생하는 전해질 이상입니다. | |
| ## 원인 | |
| 원인은 매우 다양한데, 이뇨제나 정신 질환 치료 약제 사용, 체액량 감소, 심부전, 간경화, 각종 폐 또는 뇌질환 등이 있습니다. | |
| 특히, 최근에는 혈압약에 이뇨제가 포함되어 있는 경우가 많으며, 이러한 약제를 복용하는 상태에서 설사, 구토, 식사량 저하 등이 갑자기 발생하는 경우 저나트륨혈증 발병의 위험도가 증가합니다. | |
| ## 치료의 중요성 | |
| 저나트륨혈증의 원인과 발생 속도에 따라서 위중도가 달라질 수 있으며, 급격하게 낮아지는 경우 전신 경련이나 의식 저하가 발생될 수 있기 때문에 원인에 대한 철저한 조사와 더불어 적극적인 치료가 필요합니다. | |
| 출처: 일반인을 위한 만성콩팥병 바로알기""" | |
| }, | |
| { | |
| "filename": "[진단-DIA-SYM-S2] 빈혈이 심해요.txt", | |
| "content": """# 빈혈이 심해요. | |
| 서울대학교병원 신장내과 이하정 | |
| Key Message: 투석 환자의 빈혈은 심혈관합병증 및 사망의 위험을 높일 수 있어 경구 혹은 주사 철분제 및 합성조혈호르몬을 이용한 적극적인 치료가 필요합니다. | |
| ## 빈혈의 원인 | |
| 빈혈은 투석 환자의 거의 대부분에서 나타날 수 있는 흔한 현상입니다. 신장은 에리스로포이에틴(erythropoietin)이라는 혈액을 만드는 것을 돕는 조혈호르몬을 분비합니다. 신장 기능이 나빠지면 조혈호르몬의 분비가 감소하여 빈혈이 생깁니다. | |
| 조혈호르몬 이외에도 다음과 같은 원인으로 빈혈이 생길 수 있습니다: | |
| - 요독으로 인한 적혈구 수명 단축 | |
| - 철분의 결핍 | |
| - 출혈성 질환 | |
| - 심한 부갑상선 항진증으로 인한 골수의 섬유화 | |
| - 급성 혹은 만성 염증성 질환 | |
| - 엽산 결핍 | |
| ## 빈혈의 증상과 합병증 | |
| 빈혈이 생기면 쉽게 피로하고 전신 쇠약감을 느끼며, 추위를 잘 견디지 못하고 심한 경우 호흡곤란을 호소하는 경우도 있으며 이로 인해 삶의 질이 저하됩니다. | |
| 장기간 빈혈에 적응하여 특별한 증상을 느끼지 못하는 경우도 많지만, 증상이 없다고 하더라도 빈혈이 적절히 치료되지 못하고 장기간 지속되는 경우 심장에 부담을 주어 심비대 및 이로 인한 심부전을 유발하게 됩니다. 심비대와 심부전은 모두 투석 환자의 중요한 심혈관계 합병증으로 주요 사망의 원인이 될 수 있습니다. | |
| ## 빈혈의 치료 | |
| 빈혈을 치료하기 위해서는 다음과 같은 치료가 필요합니다: | |
| 1. **철분 보충**: 경구 철분제 혹은 주사 철분제로 보충이 가능하며, 정기적으로 체내 저장량을 모니터링 하면서 충분히 보충해야 합니다. | |
| 2. **엽산 보충**: 경구 약제로 보충이 가능합니다. | |
| 3. **조혈호르몬 보충**: 피하 혹은 정맥 주사 제제로 개발된 합성에리스로포이에틴을 정기적으로 맞아 보충할 수 있습니다. | |
| 4. **적절한 투석**: 요독을 최소화하기 위해 적절한 효율의 투석 치료를 유지하는 것이 중요합니다. | |
| ## 치료 저항성 빈혈 | |
| 철분, 엽산, 조혈호르몬을 충분히 보충하여 주는 경우에도 빈혈이 호전되지 않는다면, 출혈성 질환이 동반되어 있지 않는지 확인이 필요합니다. 또한 조혈호르몬에 대한 저항성이 생기지 않았는지 확인할 필요가 있습니다. | |
| 급성 염증성 질환, 인 조절이 잘 되지 않아 발생하는 심한 부갑상선 기능 항진증, 악성 종양과 같은 질환 등은 합성 조혈호르몬의 저항성을 유도할 수 있으므로 빈혈 교정을 위해 치료가 필요합니다. | |
| 출처: 서울대학교병원 신장내과""" | |
| } | |
| ] | |
| for filename, content in samples: | |
| filepath = os.path.join(self.documents_path, filename) | |
| with open(filepath, 'w', encoding='utf-8') as f: | |
| f.write(content) | |
| sample_files.append(filepath) | |
| logger.info(f"Created sample file: {filename}") | |
| return sample_files | |
| def search(self, query: str, k: int = 5, | |
| filters: Optional[Dict[str, Any]] = None) -> List[Document]: | |
| """관련 문서 검색""" | |
| if not self.vectorstore: | |
| logger.warning("Vectorstore not loaded, loading now...") | |
| self.load_documents() | |
| logger.info(f"Searching for: '{query}' with k={k}, filters={filters}") | |
| results = self.vectorstore.similarity_search(query, k=k*2) | |
| if filters: | |
| filtered_results = [] | |
| for doc in results: | |
| match = True | |
| for key, value in filters.items(): | |
| if key in doc.metadata and doc.metadata[key] != value: | |
| match = False | |
| break | |
| if match: | |
| filtered_results.append(doc) | |
| results = filtered_results[:k] | |
| else: | |
| results = results[:k] | |
| logger.info(f"Found {len(results)} documents") | |
| return results | |
| def search_by_patient_context(self, query: str, | |
| constraints: PatientConstraints, | |
| task_type: TaskType, | |
| k: int = 5) -> List[Document]: | |
| """환자 상태와 작업 유형을 고려한 맞춤형 검색""" | |
| filters = {} | |
| # 작업 유형에 따른 필터 | |
| task_field_mapping = { | |
| TaskType.DIET_RECOMMENDATION: "diet", | |
| TaskType.DIET_ANALYSIS: "diet", | |
| TaskType.MEDICATION: "medication", | |
| TaskType.LIFESTYLE: "lifestyle", | |
| TaskType.DIAGNOSIS: "diagnosis", | |
| TaskType.EXERCISE: "exercise" | |
| } | |
| if task_type in task_field_mapping: | |
| filters["field"] = task_field_mapping[task_type] | |
| # 환자 상태에 따른 필터 | |
| if constraints.on_dialysis: | |
| filters["patient_status"] = "hemodialysis" | |
| elif constraints.disease_stage in [DiseaseStage.CKD_3, DiseaseStage.CKD_4]: | |
| filters["patient_status"] = "chronic_kidney_disease" | |
| # 동반질환에 따른 검색 | |
| additional_results = [] | |
| if "당뇨" in constraints.comorbidities: | |
| additional_results.extend( | |
| self.search(query, k=k//3, filters={"personalization_level": "diabetes"}) | |
| ) | |
| if "고혈압" in constraints.comorbidities: | |
| additional_results.extend( | |
| self.search(query, k=k//3, filters={"personalization_level": "hypertension"}) | |
| ) | |
| main_results = self.search(query, k=k-len(additional_results), filters=filters) | |
| all_results = main_results + additional_results | |
| logger.info(f"Patient context search found {len(all_results)} total documents") | |
| return all_results | |
| # ========== 식품 영양 분석 ========== | |
| class FoodNutritionDatabase: | |
| """식품 영양 성분 데이터베이스 - 싱글톤 패턴 적용""" | |
| _instance = None | |
| _initialized = False | |
| def __new__(cls, *args, **kwargs): | |
| if cls._instance is None: | |
| cls._instance = super(FoodNutritionDatabase, cls).__new__(cls) | |
| return cls._instance | |
| def __init__(self, csv_path: str = "통합식품영양성분정보(음식)_20241224.csv"): | |
| if FoodNutritionDatabase._initialized: | |
| return | |
| self.csv_path = csv_path | |
| self.food_data = None | |
| self.load_food_data() | |
| FoodNutritionDatabase._initialized = True | |
| def load_food_data(self): | |
| """CSV 파일에서 식품 데이터 로드""" | |
| try: | |
| # CSV 파일 로드 시도 | |
| if os.path.exists(self.csv_path): | |
| self.food_data = pd.read_csv(self.csv_path, encoding='utf-8') | |
| logger.info(f"Loaded food data from {self.csv_path}") | |
| else: | |
| raise FileNotFoundError(f"CSV file not found: {self.csv_path}") | |
| # 컬럼명 정리 (실제 CSV 구조에 맞게) | |
| column_mapping = { | |
| '식품코드': 'food_code', | |
| '식품명': 'name', | |
| '식품대분류명': 'category_major', | |
| '식품중분류명': 'category_minor', | |
| '영양성분함량기준량': 'serving_size', | |
| '에너지(kcal)': 'calories', | |
| '수분(g)': 'water', | |
| '단백질(g)': 'protein', | |
| '지방(g)': 'fat', | |
| '탄수화물(g)': 'carbohydrate', | |
| '당류(g)': 'sugar', | |
| '식이섬유(g)': 'dietary_fiber', | |
| '칼슘(mg)': 'calcium', | |
| '철(mg)': 'iron', | |
| '인(mg)': 'phosphorus', | |
| '칼륨(mg)': 'potassium', | |
| '나트륨(mg)': 'sodium', | |
| '콜레스테롤(mg)': 'cholesterol', | |
| '포화지방산(g)': 'saturated_fat' | |
| } | |
| self.food_data = self.food_data.rename(columns=column_mapping) | |
| # 숫자형 컬럼 변환 | |
| numeric_columns = ['calories', 'protein', 'fat', 'carbohydrate', | |
| 'sodium', 'potassium', 'phosphorus', 'calcium', | |
| 'water', 'sugar', 'dietary_fiber', 'iron', | |
| 'cholesterol', 'saturated_fat'] | |
| for col in numeric_columns: | |
| if col in self.food_data.columns: | |
| self.food_data[col] = pd.to_numeric(self.food_data[col], errors='coerce') | |
| # serving_size를 숫자로 변환 (예: "100g" -> 100) | |
| if 'serving_size' in self.food_data.columns: | |
| if self.food_data['serving_size'].dtype == 'object': | |
| self.food_data['serving_size'] = self.food_data['serving_size'].str.extract('(\d+)').astype(float) | |
| else: | |
| self.food_data['serving_size'] = pd.to_numeric(self.food_data['serving_size'], errors='coerce') | |
| # NaN 값을 0으로 채우기 | |
| self.food_data = self.food_data.fillna(0) | |
| logger.info(f"Loaded {len(self.food_data)} food items from database") | |
| except Exception as e: | |
| logger.error(f"Error loading food database: {e}") | |
| logger.info("Creating sample food data...") | |
| self.food_data = self._create_sample_data() | |
| def _create_sample_data(self): | |
| """샘플 식품 데이터 생성""" | |
| sample_data = { | |
| 'food_code': ['D101-001', 'D101-002', 'D101-003', 'D101-004', 'D101-005', 'D101-006', | |
| 'D101-007', 'D101-008', 'D101-009', 'D101-010'], | |
| 'name': ['쌀밥', '닭가슴살', '브로콜리', '사과', '두부', '달걀', '감자', '우유', '연어', '시금치'], | |
| 'category_major': ['곡류', '육류', '채소류', '과일류', '콩류', '난류', '서류', '유제품류', '어패류', '채소류'], | |
| 'category_minor': ['밥류', '가금류', '녹황색채소', '과일', '두부', '계란', '감자류', '우유류', '생선류', '엽채류'], | |
| 'serving_size': [100, 100, 100, 100, 100, 100, 100, 100, 100, 100], | |
| 'calories': [130, 165, 34, 52, 76, 155, 77, 61, 208, 23], | |
| 'water': [68.5, 65.3, 89.3, 85.6, 84.6, 76.2, 79.3, 87.7, 68.5, 91.4], | |
| 'protein': [2.7, 31.0, 2.8, 0.3, 8.1, 13.0, 2.0, 3.3, 20.4, 2.9], | |
| 'fat': [0.3, 3.6, 0.4, 0.2, 4.8, 11.0, 0.1, 3.3, 13.4, 0.4], | |
| 'carbohydrate': [28.2, 0, 6.6, 13.8, 1.9, 1.1, 17.6, 4.8, 0, 3.6], | |
| 'sugar': [0.1, 0, 1.7, 10.4, 0.7, 0.4, 0.8, 5.0, 0, 0.4], | |
| 'dietary_fiber': [0.4, 0, 2.6, 2.4, 0.3, 0, 1.8, 0, 0, 2.2], | |
| 'calcium': [10, 11, 47, 6, 350, 56, 10, 113, 12, 99], | |
| 'iron': [0.5, 0.9, 0.7, 0.1, 5.4, 1.8, 0.8, 0.1, 0.8, 2.7], | |
| 'phosphorus': [43, 210, 66, 11, 110, 198, 57, 93, 252, 49], | |
| 'potassium': [35, 256, 316, 107, 121, 138, 421, 150, 490, 558], | |
| 'sodium': [1, 74, 30, 1, 7, 142, 6, 50, 44, 79], | |
| 'cholesterol': [0, 85, 0, 0, 0, 373, 0, 12, 55, 0], | |
| 'saturated_fat': [0.1, 1.0, 0.1, 0, 0.7, 3.3, 0, 1.9, 3.1, 0.1] | |
| } | |
| return pd.DataFrame(sample_data) | |
| def search_foods(self, query: str, limit: int = 10) -> List[FoodItem]: | |
| """식품 검색""" | |
| logger.info(f"Searching for food: '{query}'") | |
| # 검색어가 포함된 식품 찾기 | |
| mask = self.food_data['name'].str.contains(query, case=False, na=False) | |
| results = self.food_data[mask].head(limit) | |
| food_items = [] | |
| for _, row in results.iterrows(): | |
| food_item = FoodItem( | |
| food_code=str(row.get('food_code', '')), | |
| name=row['name'], | |
| food_category_major=row.get('category_major', ''), | |
| food_category_minor=row.get('category_minor', ''), | |
| serving_size=float(row.get('serving_size', 100)), | |
| calories=float(row['calories']), | |
| water=float(row.get('water', 0)), | |
| protein=float(row['protein']), | |
| fat=float(row['fat']), | |
| carbohydrate=float(row['carbohydrate']), | |
| sugar=float(row.get('sugar', 0)), | |
| dietary_fiber=float(row.get('dietary_fiber', 0)), | |
| calcium=float(row.get('calcium', 0)), | |
| iron=float(row.get('iron', 0)), | |
| phosphorus=float(row['phosphorus']), | |
| potassium=float(row['potassium']), | |
| sodium=float(row['sodium']), | |
| cholesterol=float(row.get('cholesterol', 0)), | |
| saturated_fat=float(row.get('saturated_fat', 0)) | |
| ) | |
| food_items.append(food_item) | |
| logger.info(f"Found {len(food_items)} food items for '{query}'") | |
| return food_items | |
| def recommend_foods_for_patient(self, constraints: PatientConstraints, | |
| meal_type: str = "all", | |
| limit: int = 20) -> List[FoodItem]: | |
| """환자 제약조건에 맞는 식품 추천""" | |
| logger.info(f"Recommending foods for patient with constraints, meal_type={meal_type}") | |
| # 필터링 조건 설정 | |
| filtered_data = self.food_data.copy() | |
| # 단백질 제한 (한 끼 기준 = 일일 제한량의 30%) | |
| if constraints.protein_restriction: | |
| max_protein = constraints.protein_restriction * 0.3 | |
| filtered_data = filtered_data[filtered_data['protein'] <= max_protein] | |
| # 나트륨 제한 | |
| if constraints.sodium_restriction: | |
| max_sodium = constraints.sodium_restriction * 0.3 | |
| filtered_data = filtered_data[filtered_data['sodium'] <= max_sodium] | |
| # 칼륨 제한 | |
| if constraints.potassium_restriction: | |
| max_potassium = constraints.potassium_restriction * 0.3 | |
| filtered_data = filtered_data[filtered_data['potassium'] <= max_potassium] | |
| # 인 제한 | |
| if constraints.phosphorus_restriction: | |
| max_phosphorus = constraints.phosphorus_restriction * 0.3 | |
| filtered_data = filtered_data[filtered_data['phosphorus'] <= max_phosphorus] | |
| # 식사 유형에 따른 필터링 | |
| if meal_type == "breakfast": | |
| # 아침식사에 적합한 카테고리 | |
| breakfast_categories = ['곡류', '유제품류', '과일류', '난류'] | |
| mask = filtered_data['category_major'].isin(breakfast_categories) | |
| if mask.any(): | |
| filtered_data = filtered_data[mask] | |
| elif meal_type == "lunch" or meal_type == "dinner": | |
| # 점심/저녁에 적합한 카테고리 | |
| main_categories = ['곡류', '육류', '어패류', '채소류', '콩류'] | |
| mask = filtered_data['category_major'].isin(main_categories) | |
| if mask.any(): | |
| filtered_data = filtered_data[mask] | |
| # 칼로리 기준으로 정렬 (적절한 칼로리 범위 우선) | |
| if constraints.calorie_target: | |
| target_cal_per_meal = constraints.calorie_target / 3 | |
| filtered_data['cal_diff'] = abs(filtered_data['calories'] - target_cal_per_meal * 0.5) | |
| filtered_data = filtered_data.sort_values('cal_diff') | |
| # 상위 N개 선택 | |
| top_foods = filtered_data.head(limit) | |
| # FoodItem 객체로 변환 | |
| recommended_foods = [] | |
| for _, row in top_foods.iterrows(): | |
| food_item = FoodItem( | |
| food_code=str(row.get('food_code', '')), | |
| name=row['name'], | |
| food_category_major=row.get('category_major', ''), | |
| food_category_minor=row.get('category_minor', ''), | |
| serving_size=float(row.get('serving_size', 100)), | |
| calories=float(row['calories']), | |
| water=float(row.get('water', 0)), | |
| protein=float(row['protein']), | |
| fat=float(row['fat']), | |
| carbohydrate=float(row['carbohydrate']), | |
| sugar=float(row.get('sugar', 0)), | |
| dietary_fiber=float(row.get('dietary_fiber', 0)), | |
| calcium=float(row.get('calcium', 0)), | |
| iron=float(row.get('iron', 0)), | |
| phosphorus=float(row['phosphorus']), | |
| potassium=float(row['potassium']), | |
| sodium=float(row['sodium']), | |
| cholesterol=float(row.get('cholesterol', 0)), | |
| saturated_fat=float(row.get('saturated_fat', 0)) | |
| ) | |
| recommended_foods.append(food_item) | |
| logger.info(f"Recommended {len(recommended_foods)} foods for {meal_type}") | |
| return recommended_foods | |
| def create_daily_meal_plan(self, constraints: PatientConstraints) -> Dict[str, List[FoodItem]]: | |
| """하루 식단 계획 생성""" | |
| logger.info("Creating daily meal plan") | |
| meal_plan = { | |
| 'breakfast': [], | |
| 'lunch': [], | |
| 'dinner': [], | |
| 'snack': [] | |
| } | |
| # 각 식사별 추천 식품 | |
| meal_plan['breakfast'] = self.recommend_foods_for_patient( | |
| constraints, meal_type='breakfast', limit=5 | |
| ) | |
| meal_plan['lunch'] = self.recommend_foods_for_patient( | |
| constraints, meal_type='lunch', limit=5 | |
| ) | |
| meal_plan['dinner'] = self.recommend_foods_for_patient( | |
| constraints, meal_type='dinner', limit=5 | |
| ) | |
| # 간식 추천 (칼로리가 낮은 식품) | |
| snack_data = self.food_data[self.food_data['calories'] < 100] | |
| if constraints.protein_restriction: | |
| snack_data = snack_data[snack_data['protein'] < constraints.protein_restriction * 0.1] | |
| snack_foods = [] | |
| for _, row in snack_data.head(3).iterrows(): | |
| food_item = FoodItem( | |
| food_code=str(row.get('food_code', '')), | |
| name=row['name'], | |
| food_category_major=row.get('category_major', ''), | |
| food_category_minor=row.get('category_minor', ''), | |
| serving_size=float(row.get('serving_size', 100)), | |
| calories=float(row['calories']), | |
| water=float(row.get('water', 0)), | |
| protein=float(row['protein']), | |
| fat=float(row['fat']), | |
| carbohydrate=float(row['carbohydrate']), | |
| sugar=float(row.get('sugar', 0)), | |
| dietary_fiber=float(row.get('dietary_fiber', 0)), | |
| calcium=float(row.get('calcium', 0)), | |
| iron=float(row.get('iron', 0)), | |
| phosphorus=float(row['phosphorus']), | |
| potassium=float(row['potassium']), | |
| sodium=float(row['sodium']), | |
| cholesterol=float(row.get('cholesterol', 0)), | |
| saturated_fat=float(row.get('saturated_fat', 0)) | |
| ) | |
| snack_foods.append(food_item) | |
| meal_plan['snack'] = snack_foods | |
| logger.info("Daily meal plan created successfully") | |
| return meal_plan | |
| # ========== LLM 응답 생성 ========== | |
| class DraftGenerator: | |
| """초안 응답 생성기""" | |
| def __init__(self): | |
| self.llm = ChatOpenAI(temperature=0.7, model="gpt-4o") | |
| def generate_draft(self, query: str, constraints: PatientConstraints, | |
| context_docs: List[Document]) -> Tuple[str, List[RecommendationItem]]: | |
| """제약조건을 고려한 초안 생성""" | |
| logger.info("Generating draft response") | |
| context = "\n".join([doc.page_content for doc in context_docs]) | |
| constraints_text = f""" | |
| 환자 정보: | |
| - eGFR: {constraints.egfr} ml/min | |
| - 질병 단계: {constraints.disease_stage.value} | |
| - 투석 여부: {'예' if constraints.on_dialysis else '아니오'} | |
| - 동반질환: {', '.join(constraints.comorbidities) if constraints.comorbidities else '없음'} | |
| - 복용 약물: {', '.join(constraints.medications) if constraints.medications else '없음'} | |
| - 연령: {constraints.age}세 | |
| - 성별: {constraints.gender} | |
| 영양 제한사항: | |
| - 단백질: {constraints.protein_restriction}g/일 | |
| - 나트륨: {constraints.sodium_restriction}mg/일 | |
| - 칼륨: {constraints.potassium_restriction}mg/일 | |
| - 인: {constraints.phosphorus_restriction}mg/일 | |
| - 수분: {constraints.fluid_restriction}ml/일 | |
| """ | |
| prompt = f""" | |
| 다음 신장질환 환자의 질문에 대해 답변하세요. | |
| 질문: {query} | |
| 참고 문서: | |
| {context} | |
| {constraints_text} | |
| 답변 시 다음 사항을 준수하세요: | |
| 1. 환자의 개별 상태를 반영한 맞춤형 답변 제공 | |
| 2. 구체적인 권장사항은 <item>태그</item>로 표시 | |
| 3. 의학적으로 정확하고 이해하기 쉬운 설명 제공 | |
| 4. 참고 문서의 내용을 활용하여 근거 있는 답변 작성 | |
| """ | |
| response = self.llm.predict(prompt) | |
| items = self._extract_items(response) | |
| logger.info(f"Generated draft with {len(items)} recommendation items") | |
| return response, items | |
| def _extract_items(self, response: str) -> List[RecommendationItem]: | |
| """응답에서 추천 항목 추출""" | |
| items = [] | |
| pattern = r'<item>(.*?)</item>' | |
| matches = re.findall(pattern, response, re.DOTALL) | |
| for match in matches: | |
| category = "식이" if any(word in match for word in ["섭취", "식사", "음식"]) else "기타" | |
| item = RecommendationItem( | |
| name=match.strip(), | |
| category=category, | |
| description=match.strip(), | |
| constraints_satisfied=False | |
| ) | |
| items.append(item) | |
| return items | |
| # ========== Correction Algorithm ========== | |
| class CorrectionAlgorithm: | |
| """제약조건 만족을 위한 보정 알고리즘""" | |
| def __init__(self, catalog: KidneyDiseaseCatalog): | |
| self.catalog = catalog | |
| self.embeddings = OpenAIEmbeddings() | |
| def correct_items(self, draft_items: List[RecommendationItem], | |
| constraints: PatientConstraints) -> List[RecommendationItem]: | |
| """초안 항목들을 제약조건에 맞게 보정""" | |
| logger.info(f"Correcting {len(draft_items)} draft items") | |
| corrected_items = [] | |
| for item in draft_items: | |
| item.embedding = self._get_embedding(item.name) | |
| similar_docs = self.catalog.search(item.name, k=10) | |
| best_replacement = self._find_best_replacement( | |
| item, similar_docs, constraints | |
| ) | |
| if best_replacement: | |
| corrected_items.append(best_replacement) | |
| else: | |
| item.constraints_satisfied = False | |
| corrected_items.append(item) | |
| logger.info(f"Corrected to {len(corrected_items)} items") | |
| return corrected_items | |
| def _get_embedding(self, text: str) -> np.ndarray: | |
| """텍스트 임베딩 생성""" | |
| return np.array(self.embeddings.embed_query(text)) | |
| def _find_best_replacement(self, original_item: RecommendationItem, | |
| candidates: List[Document], | |
| constraints: PatientConstraints) -> Optional[RecommendationItem]: | |
| """제약조건을 만족하는 최적 대체 항목 찾기""" | |
| best_item = None | |
| best_score = float('inf') | |
| for doc in candidates: | |
| if self._check_constraints(doc, constraints): | |
| doc_embedding = self._get_embedding(doc.page_content) | |
| distance = np.linalg.norm(original_item.embedding - doc_embedding) | |
| if distance < best_score: | |
| best_score = distance | |
| best_item = RecommendationItem( | |
| name=doc.metadata.get('title', doc.page_content[:50]), | |
| category=doc.metadata.get('field', original_item.category), | |
| description=doc.page_content, | |
| constraints_satisfied=True, | |
| embedding=doc_embedding | |
| ) | |
| return best_item | |
| def _check_constraints(self, doc: Document, constraints: PatientConstraints) -> bool: | |
| """문서가 환자 제약조건을 만족하는지 검증""" | |
| content = doc.page_content.lower() | |
| if constraints.on_dialysis: | |
| if "투석 금지" in content or "투석 환자 제외" in content: | |
| return False | |
| if constraints.disease_stage in [DiseaseStage.CKD_4, DiseaseStage.CKD_5]: | |
| if "진행성 신부전 주의" in content: | |
| return False | |
| for comorbidity in constraints.comorbidities: | |
| if comorbidity == "당뇨" and "당뇨 금기" in content: | |
| return False | |
| if comorbidity == "고혈압" and "혈압 상승 주의" in content: | |
| return False | |
| return True | |
| # ========== LangGraph Nodes ========== | |
| def classify_task(state: GraphState) -> GraphState: | |
| """질문 유형 분류 - LLM 사용""" | |
| logger.info("=== CLASSIFY TASK NODE ===") | |
| logger.info(f"User query: {state['user_query']}") | |
| state["current_node"] = "분류" | |
| state["processing_log"].append("질문 유형 분석 중...") | |
| llm = ChatOpenAI(temperature=0.3, model="gpt-4o") | |
| prompt = f""" | |
| 다음 질문을 분석하여 적절한 카테고리로 분류하세요. | |
| 질문: {state['user_query']} | |
| 카테고리: | |
| - diet_recommendation: 식단 추천, 하루 식사 계획, 무엇을 먹어야 할지 묻는 질문 | |
| - diet_analysis: 특정 음식의 영양 성분, 섭취 가능 여부, 영양소 함량 분석 | |
| - medication: 약물 복용 방법, 시간, 부작용, 상호작용 | |
| - lifestyle: 일상생활 관리, 수면, 스트레스, 수분 섭취 | |
| - diagnosis: 검사 결과 해석, 질병 단계, 수치 의미 | |
| - exercise: 운동 방법, 종류, 강도, 주의사항 | |
| - general: 위 카테고리에 속하지 않는 일반적인 질문 | |
| 카테고리 이름만 반환하세요. | |
| """ | |
| response = llm.predict(prompt).strip().lower() | |
| # 카테고리 매핑 | |
| category_mapping = { | |
| 'diet_recommendation': TaskType.DIET_RECOMMENDATION, | |
| 'diet_analysis': TaskType.DIET_ANALYSIS, | |
| 'medication': TaskType.MEDICATION, | |
| 'lifestyle': TaskType.LIFESTYLE, | |
| 'diagnosis': TaskType.DIAGNOSIS, | |
| 'exercise': TaskType.EXERCISE, | |
| 'general': TaskType.GENERAL | |
| } | |
| selected_task = category_mapping.get(response, TaskType.GENERAL) | |
| state["task_type"] = selected_task | |
| logger.info(f"Task classified as: {selected_task.value}") | |
| state["processing_log"].append(f"질문 유형: {selected_task.value}") | |
| logger.info("=== END CLASSIFY TASK ===\n") | |
| return state | |
| def retrieve_context(state: GraphState) -> GraphState: | |
| """관련 문서 검색""" | |
| logger.info("=== RETRIEVE CONTEXT NODE ===") | |
| logger.info(f"Query: {state['user_query']}") | |
| logger.info(f"Task type: {state['task_type'].value}") | |
| state["current_node"] = "검색" | |
| state["processing_log"].append("관련 문서 검색 중...") | |
| catalog = KidneyDiseaseCatalog() | |
| results = catalog.search_by_patient_context( | |
| state["user_query"], | |
| state["patient_constraints"], | |
| state["task_type"] | |
| ) | |
| state["catalog_results"] = results | |
| for i, doc in enumerate(results[:3]): | |
| logger.info(f"Document {i+1}: {doc.metadata.get('title', 'Unknown')} " | |
| f"[{doc.metadata.get('raw_tags', 'No tags')}]") | |
| state["processing_log"].append(f"{len(results)}개 관련 문서 검색 완료") | |
| logger.info("=== END RETRIEVE CONTEXT ===\n") | |
| return state | |
| def analyze_diet_request(state: GraphState) -> GraphState: | |
| """식이 관련 요청 분석 및 식품 데이터베이스 검색""" | |
| logger.info("=== ANALYZE DIET REQUEST NODE ===") | |
| state["current_node"] = "식품 분석" | |
| state["processing_log"].append("식품 정보 분석 중...") | |
| food_db = FoodNutritionDatabase() | |
| query = state["user_query"] | |
| constraints = state["patient_constraints"] | |
| # LLM을 사용하여 질문에서 언급된 식품 추출 | |
| llm = ChatOpenAI(temperature=0.3, model="gpt-4o") | |
| prompt = f""" | |
| 다음 질문에서 언급된 모든 식품명을 추출하세요. | |
| 질문: {query} | |
| 식품명만 쉼표로 구분하여 나열하세요. 없으면 "없음"이라고 답하세요. | |
| """ | |
| food_names_response = llm.predict(prompt).strip() | |
| logger.info(f"Extracted food names: {food_names_response}") | |
| mentioned_foods = [] | |
| if food_names_response != "없음": | |
| food_names = [name.strip() for name in food_names_response.split(',')] | |
| for food_name in food_names: | |
| found_foods = food_db.search_foods(food_name, limit=3) | |
| mentioned_foods.extend(found_foods) | |
| # 식품 분석 결과 생성 | |
| analysis_results = { | |
| 'mentioned_foods': [], | |
| 'suitable_foods': [], | |
| 'unsuitable_foods': [], | |
| 'nutritional_summary': {} | |
| } | |
| # 언급된 식품 분석 | |
| for food in mentioned_foods: | |
| is_suitable, issues = food.is_suitable_for_patient(constraints) | |
| food_info = { | |
| 'name': food.name, | |
| 'nutrients': food.get_nutrients_per_serving(100), | |
| 'suitable': is_suitable, | |
| 'issues': issues | |
| } | |
| analysis_results['mentioned_foods'].append(food_info) | |
| if is_suitable: | |
| analysis_results['suitable_foods'].append(food) | |
| else: | |
| analysis_results['unsuitable_foods'].append((food, issues)) | |
| state["food_analysis_results"] = analysis_results | |
| logger.info(f"Analyzed {len(mentioned_foods)} foods") | |
| state["processing_log"].append(f"{len(mentioned_foods)}개 식품 분석 완료") | |
| logger.info("=== END ANALYZE DIET REQUEST ===\n") | |
| return state | |
| def generate_meal_plan(state: GraphState) -> GraphState: | |
| """일일 식단 계획 생성""" | |
| logger.info("=== GENERATE MEAL PLAN NODE ===") | |
| state["current_node"] = "식단 생성" | |
| state["processing_log"].append("일일 식단 계획 생성 중...") | |
| food_db = FoodNutritionDatabase() | |
| constraints = state["patient_constraints"] | |
| # 하루 식단 생성 | |
| meal_plan = food_db.create_daily_meal_plan(constraints) | |
| # 영양소 총량 계산 | |
| daily_totals = { | |
| 'calories': 0, | |
| 'protein': 0, | |
| 'sodium': 0, | |
| 'potassium': 0, | |
| 'phosphorus': 0 | |
| } | |
| for meal_type, foods in meal_plan.items(): | |
| for food in foods: | |
| nutrients = food.get_nutrients_per_serving(100) | |
| for nutrient, value in nutrients.items(): | |
| if nutrient in daily_totals: | |
| daily_totals[nutrient] += value | |
| state["meal_plan"] = meal_plan | |
| # 기존 food_analysis_results가 있으면 업데이트, 없으면 생성 | |
| if state.get("food_analysis_results") is None: | |
| state["food_analysis_results"] = {} | |
| state["food_analysis_results"].update({ | |
| 'meal_plan': meal_plan, | |
| 'daily_totals': daily_totals, | |
| 'recommendations': [] | |
| }) | |
| # 제약조건 대비 검증 | |
| if daily_totals['protein'] > constraints.protein_restriction: | |
| state["food_analysis_results"]['recommendations'].append( | |
| f"주의: 추천 식단의 단백질 총량({daily_totals['protein']:.1f}g)이 " | |
| f"일일 제한량({constraints.protein_restriction}g)을 초과합니다." | |
| ) | |
| logger.info("Meal plan generated successfully") | |
| state["processing_log"].append("식단 계획 생성 완료") | |
| logger.info("=== END GENERATE MEAL PLAN ===\n") | |
| return state | |
| def generate_diet_response(state: GraphState) -> GraphState: | |
| """식이 관련 최종 응답 생성""" | |
| logger.info("=== GENERATE DIET RESPONSE NODE ===") | |
| state["current_node"] = "응답 생성" | |
| state["processing_log"].append("식이 관련 답변 생성 중...") | |
| llm = ChatOpenAI(temperature=0.5, model="gpt-4o") | |
| task_type = state["task_type"] | |
| constraints = state["patient_constraints"] | |
| context_docs = state.get("catalog_results", []) | |
| # 참고 문서 내용 추출 | |
| context = "\n\n".join([ | |
| f"[{doc.metadata.get('title', 'Document')}]\n{doc.page_content[:500]}..." | |
| for doc in context_docs[:3] | |
| ]) | |
| if task_type == TaskType.DIET_RECOMMENDATION and state.get("meal_plan"): | |
| # 식단 추천 응답 | |
| meal_plan = state["meal_plan"] | |
| daily_totals = state["food_analysis_results"]["daily_totals"] | |
| prompt = f""" | |
| 신장질환 환자를 위한 일일 식단을 추천합니다. | |
| 환자 정보: | |
| - 질병 단계: {constraints.disease_stage.value} | |
| - 단백질 제한: {constraints.protein_restriction}g/일 | |
| - 나트륨 제한: {constraints.sodium_restriction}mg/일 | |
| - 칼륨 제한: {constraints.potassium_restriction}mg/일 | |
| - 인 제한: {constraints.phosphorus_restriction}mg/일 | |
| 추천 식단: | |
| 아침: {', '.join([f.name for f in meal_plan['breakfast'][:3]])} | |
| 점심: {', '.join([f.name for f in meal_plan['lunch'][:3]])} | |
| 저녁: {', '.join([f.name for f in meal_plan['dinner'][:3]])} | |
| 간식: {', '.join([f.name for f in meal_plan['snack'][:2]])} | |
| 영양소 총량: | |
| - 칼로리: {daily_totals['calories']:.0f} kcal | |
| - 단백질: {daily_totals['protein']:.1f} g | |
| - 나트륨: {daily_totals['sodium']:.0f} mg | |
| - 칼륨: {daily_totals['potassium']:.0f} mg | |
| - 인: {daily_totals['phosphorus']:.0f} mg | |
| 참고 자료: | |
| {context} | |
| 위 정보를 바탕으로 환자가 이해하기 쉽게 설명하고, | |
| 각 식사의 영양학적 장점과 주의사항을 포함해주세요. | |
| 의료진과의 상담 필요성도 언급하세요. | |
| """ | |
| elif task_type == TaskType.DIET_ANALYSIS and state.get("food_analysis_results"): | |
| # 특정 식품 분석 응답 | |
| analysis = state["food_analysis_results"] | |
| foods_summary = [] | |
| for food_info in analysis.get('mentioned_foods', []): | |
| summary = f"{food_info['name']}: " | |
| if food_info['suitable']: | |
| summary += "섭취 가능" | |
| else: | |
| summary += f"주의 필요 ({', '.join(food_info['issues'])})" | |
| foods_summary.append(summary) | |
| prompt = f""" | |
| 환자가 질문한 식품들의 영양 분석 결과입니다. | |
| 질문: {state['user_query']} | |
| 분석 결과: | |
| {chr(10).join(foods_summary) if foods_summary else "분석된 식품이 없습니다."} | |
| 환자의 제한사항: | |
| - 단백질: {constraints.protein_restriction}g/일 | |
| - 나트륨: {constraints.sodium_restriction}mg/일 | |
| - 칼륨: {constraints.potassium_restriction}mg/일 | |
| - 인: {constraints.phosphorus_restriction}mg/일 | |
| 참고 자료: | |
| {context} | |
| 위 분석을 바탕으로 각 식품의 섭취 가능 여부와 | |
| 적절한 섭취량을 구체적으로 설명해주세요. | |
| """ | |
| else: | |
| # 일반 식이 관련 응답 | |
| prompt = f""" | |
| 신장질환 환자의 식이 관련 질문에 답변하세요. | |
| 질문: {state['user_query']} | |
| 환자 정보: | |
| - 질병 단계: {constraints.disease_stage.value} | |
| - 영양 제한사항이 있습니다. | |
| 참고 자료: | |
| {context} | |
| 환자 상태를 고려한 구체적이고 실용적인 답변을 제공하세요. | |
| 의료진과의 상담 필요성도 언급하세요. | |
| """ | |
| response = llm.predict(prompt) | |
| state["final_response"] = response | |
| logger.info("Diet response generated") | |
| state["processing_log"].append("답변 생성 완료") | |
| logger.info("=== END GENERATE DIET RESPONSE ===\n") | |
| return state | |
| def generate_general_response(state: GraphState) -> GraphState: | |
| """일반 질문에 대한 응답 생성""" | |
| logger.info("=== GENERATE GENERAL RESPONSE NODE ===") | |
| state["current_node"] = "응답 생성" | |
| state["processing_log"].append("일반 답변 생성 중...") | |
| generator = DraftGenerator() | |
| draft_response, draft_items = generator.generate_draft( | |
| state["user_query"], | |
| state["patient_constraints"], | |
| state.get("catalog_results", []) | |
| ) | |
| state["draft_response"] = draft_response | |
| state["draft_items"] = draft_items | |
| # 보정이 필요한 경우 | |
| if draft_items: | |
| catalog = KidneyDiseaseCatalog() | |
| corrector = CorrectionAlgorithm(catalog) | |
| corrected_items = corrector.correct_items(draft_items, state["patient_constraints"]) | |
| state["corrected_items"] = corrected_items | |
| # 최종 응답 생성 | |
| llm = ChatOpenAI(temperature=0.3, model="gpt-4o") | |
| context_docs = state.get("catalog_results", []) | |
| context = "\n\n".join([ | |
| f"[{doc.metadata.get('title', 'Document')}]\n{doc.page_content[:500]}..." | |
| for doc in context_docs[:3] | |
| ]) | |
| if state.get("corrected_items"): | |
| corrected_names = [item.name for item in state["corrected_items"]] | |
| prompt = f""" | |
| 다음 정보를 포함하여 환자 질문에 답변하세요: | |
| 질문: {state["user_query"]} | |
| 환자 정보: | |
| - 질병 단계: {state["patient_constraints"].disease_stage.value} | |
| - 투석 여부: {'예' if state["patient_constraints"].on_dialysis else '아니오'} | |
| 참고 자료: | |
| {context} | |
| 초안 답변: {draft_response} | |
| 검증된 권장사항: {json.dumps(corrected_names, ensure_ascii=False)} | |
| 위 정보를 종합하여 환자에게 도움이 되는 답변을 작성하세요. | |
| 의료진과의 상담 필요성을 반드시 언급하세요. | |
| """ | |
| else: | |
| prompt = f""" | |
| 다음 질문에 대해 정확하고 이해하기 쉽게 답변하세요: | |
| 질문: {state["user_query"]} | |
| 환자 정보: | |
| - 질병 단계: {state["patient_constraints"].disease_stage.value} | |
| - 투석 여부: {'예' if state["patient_constraints"].on_dialysis else '아니오'} | |
| 참고 자료: | |
| {context} | |
| 초안: {draft_response} | |
| 위 정보를 바탕으로 환자에게 도움이 되는 답변을 작성하세요. | |
| 의료진과의 상담 필요성을 반드시 언급하세요. | |
| """ | |
| final_response = llm.predict(prompt) | |
| state["final_response"] = final_response | |
| logger.info("General response generated") | |
| state["processing_log"].append("답변 생성 완료") | |
| logger.info("=== END GENERATE GENERAL RESPONSE ===\n") | |
| return state | |
| def route_after_classification(state: GraphState) -> str: | |
| """태스크 분류 후 라우팅""" | |
| task_type = state["task_type"] | |
| if task_type in [TaskType.DIET_RECOMMENDATION, TaskType.DIET_ANALYSIS]: | |
| logger.info(f"Routing to diet_path for task type: {task_type.value}") | |
| return "diet_path" | |
| else: | |
| logger.info(f"Routing to general_path for task type: {task_type.value}") | |
| return "general_path" | |
| def route_diet_subtask(state: GraphState) -> str: | |
| """식이 관련 세부 태스크 라우팅""" | |
| if state["task_type"] == TaskType.DIET_RECOMMENDATION: | |
| logger.info("Routing to meal_plan for diet recommendation") | |
| return "meal_plan" | |
| else: | |
| logger.info("Routing to food_analysis for diet analysis") | |
| return "food_analysis" | |
| def route_after_retrieve(state: GraphState) -> str: | |
| """문서 검색 후 라우팅""" | |
| if state["task_type"] in [TaskType.DIET_RECOMMENDATION, TaskType.DIET_ANALYSIS]: | |
| logger.info("Routing to diet_response") | |
| return "diet_response" | |
| else: | |
| logger.info("Routing to general_response") | |
| return "general_response" | |
| # ========== Workflow 구성 ========== | |
| def create_kidney_disease_rag_workflow(): | |
| """신장질환 RAG 워크플로우 생성""" | |
| logger.info("Creating kidney disease RAG workflow") | |
| workflow = StateGraph(GraphState) | |
| # 노드 추가 | |
| workflow.add_node("classify", classify_task) | |
| workflow.add_node("retrieve", retrieve_context) | |
| workflow.add_node("analyze_diet", analyze_diet_request) | |
| workflow.add_node("generate_meal_plan", generate_meal_plan) | |
| workflow.add_node("generate_diet_response", generate_diet_response) | |
| workflow.add_node("generate_general_response", generate_general_response) | |
| # 시작점 | |
| workflow.set_entry_point("classify") | |
| # 분류 후 라우팅 | |
| workflow.add_conditional_edges( | |
| "classify", | |
| route_after_classification, | |
| { | |
| "diet_path": "analyze_diet", | |
| "general_path": "retrieve" | |
| } | |
| ) | |
| # 식이 경로 - 세부 분기 | |
| workflow.add_conditional_edges( | |
| "analyze_diet", | |
| route_diet_subtask, | |
| { | |
| "meal_plan": "generate_meal_plan", | |
| "food_analysis": "retrieve" | |
| } | |
| ) | |
| # 식단 생성 후 문서 검색 | |
| workflow.add_edge("generate_meal_plan", "retrieve") | |
| # 문서 검색 후 응답 생성으로 라우팅 | |
| workflow.add_conditional_edges( | |
| "retrieve", | |
| route_after_retrieve, | |
| { | |
| "diet_response": "generate_diet_response", | |
| "general_response": "generate_general_response" | |
| } | |
| ) | |
| # 최종 노드들은 END로 | |
| workflow.add_edge("generate_diet_response", END) | |
| workflow.add_edge("generate_general_response", END) | |
| compiled_workflow = workflow.compile() | |
| logger.info("Workflow compiled successfully") | |
| return compiled_workflow | |
| # ========== Streamlit UI ========== | |
| def main(): | |
| # 페이지 설정 | |
| st.set_page_config( | |
| page_title="신장질환 AI 상담 시스템", | |
| page_icon="🏥", | |
| layout="wide", | |
| initial_sidebar_state="expanded" | |
| ) | |
| # 사용자 정의 CSS | |
| st.markdown(""" | |
| <style> | |
| .main { | |
| padding: 2rem; | |
| } | |
| .stButton>button { | |
| background-color: #10b981; | |
| color: white; | |
| border-radius: 10px; | |
| border: none; | |
| padding: 0.5rem 1rem; | |
| font-weight: bold; | |
| transition: background-color 0.3s; | |
| } | |
| .stButton>button:hover { | |
| background-color: #059669; | |
| } | |
| .chat-message { | |
| padding: 1.5rem; | |
| border-radius: 1rem; | |
| margin-bottom: 1rem; | |
| background-color: #f3f4f6; | |
| } | |
| .user-message { | |
| background-color: #e0f2fe; | |
| } | |
| .assistant-message { | |
| background-color: #f0fdf4; | |
| } | |
| </style> | |
| """, unsafe_allow_html=True) | |
| # Lottie 애니메이션 로드 | |
| def load_lottie_url(url: str): | |
| try: | |
| r = requests.get(url) | |
| if r.status_code == 200: | |
| return r.json() | |
| except: | |
| pass | |
| return None | |
| # 헤더 | |
| col1, col2, col3 = st.columns([1, 2, 1]) | |
| with col2: | |
| st.title("🏥 신장질환 AI 상담 시스템") | |
| st.caption("맞춤형 의료 정보를 제공하는 AI 시스템 - OpenAI & LangGraph 기반") | |
| # 사이드바 - 환자 정보 입력 | |
| with st.sidebar: | |
| st.header("⚙️ 설정") | |
| # API 키 입력 | |
| api_key = st.text_input( | |
| "OpenAI API Key", | |
| type="password", | |
| placeholder="sk-...", | |
| help="OpenAI API 키를 입력하세요" | |
| ) | |
| if api_key: | |
| os.environ["OPENAI_API_KEY"] = api_key | |
| st.divider() | |
| st.header("👤 환자 정보") | |
| # 기본 정보 | |
| col1, col2 = st.columns(2) | |
| with col1: | |
| age = st.number_input("나이", min_value=0, max_value=150, value=65) | |
| gender = st.selectbox("성별", ["남성", "여성"]) | |
| with col2: | |
| egfr = st.number_input("eGFR (ml/min)", min_value=0.0, max_value=150.0, value=25.0) | |
| disease_stage = st.selectbox( | |
| "신장 질환 단계", | |
| options=[stage.value for stage in DiseaseStage], | |
| index=3 # CKD Stage 4 | |
| ) | |
| on_dialysis = st.checkbox("투석 중", value=False) | |
| # 동반질환 및 약물 | |
| st.subheader("🏥 동반질환") | |
| comorbidities = st.multiselect( | |
| "동반질환 선택", | |
| ["당뇨", "고혈압", "심부전", "간질환", "통풍"], | |
| default=["당뇨", "고혈압"] | |
| ) | |
| st.subheader("💊 복용 약물") | |
| medications = st.text_area( | |
| "복용 중인 약물 (쉼표로 구분)", | |
| value="ARB, 인결합제", | |
| help="예: ARB, 인결합제, 베타차단제" | |
| ).split(",") | |
| medications = [med.strip() for med in medications if med.strip()] | |
| st.divider() | |
| # 영양 제한사항 | |
| st.header("🥗 영양 제한사항 (일일)") | |
| protein = st.number_input("단백질 (g)", min_value=0.0, value=40.0) | |
| sodium = st.number_input("나트륨 (mg)", min_value=0.0, value=2000.0) | |
| potassium = st.number_input("칼륨 (mg)", min_value=0.0, value=2000.0) | |
| phosphorus = st.number_input("인 (mg)", min_value=0.0, value=800.0) | |
| fluid = st.number_input("수분 (ml)", min_value=0.0, value=1500.0) | |
| calorie = st.number_input("칼로리 (kcal)", min_value=0.0, value=1800.0) | |
| # 메인 영역 | |
| # 세션 상태 초기화 | |
| if "messages" not in st.session_state: | |
| st.session_state.messages = [] | |
| if "workflow" not in st.session_state: | |
| st.session_state.workflow = None | |
| # 워크플로우 초기화 | |
| if api_key and st.session_state.workflow is None: | |
| with st.spinner("시스템 초기화 중..."): | |
| try: | |
| st.session_state.workflow = create_kidney_disease_rag_workflow() | |
| st.success("✅ 시스템이 준비되었습니다!") | |
| except Exception as e: | |
| st.error(f"초기화 실패: {e}") | |
| # 채팅 기록 표시 | |
| for message in st.session_state.messages: | |
| with st.chat_message(message["role"]): | |
| st.markdown(message["content"]) | |
| # 처리 로그가 있으면 표시 | |
| if "processing_log" in message: | |
| with st.expander("🔍 처리 과정 보기"): | |
| for log in message["processing_log"]: | |
| st.caption(log) | |
| # 사용자 입력 | |
| if prompt := st.chat_input("신장질환에 대해 무엇이든 물어보세요..."): | |
| if not api_key: | |
| st.error("⚠️ OpenAI API 키를 입력해주세요!") | |
| return | |
| if not st.session_state.workflow: | |
| st.error("⚠️ 시스템이 초기화되지 않았습니다!") | |
| return | |
| # 사용자 메시지 추가 | |
| st.session_state.messages.append({"role": "user", "content": prompt}) | |
| with st.chat_message("user"): | |
| st.markdown(prompt) | |
| # AI 응답 생성 | |
| with st.chat_message("assistant"): | |
| with st.spinner("생각 중..."): | |
| try: | |
| # 환자 제약조건 생성 | |
| patient_constraints = PatientConstraints( | |
| egfr=egfr, | |
| disease_stage=next(s for s in DiseaseStage if s.value == disease_stage), | |
| on_dialysis=on_dialysis, | |
| comorbidities=comorbidities, | |
| medications=medications, | |
| age=age, | |
| gender=gender, | |
| protein_restriction=protein, | |
| sodium_restriction=sodium, | |
| potassium_restriction=potassium, | |
| phosphorus_restriction=phosphorus, | |
| fluid_restriction=fluid, | |
| calorie_target=calorie | |
| ) | |
| # 초기 상태 생성 | |
| initial_state = GraphState( | |
| user_query=prompt, | |
| patient_constraints=patient_constraints, | |
| task_type=TaskType.GENERAL, | |
| draft_response="", | |
| draft_items=[], | |
| corrected_items=[], | |
| final_response="", | |
| catalog_results=[], | |
| iteration_count=0, | |
| error=None, | |
| food_analysis_results=None, | |
| recommended_foods=None, | |
| meal_plan=None, | |
| current_node="", | |
| processing_log=[] | |
| ) | |
| # 워크플로우 실행 | |
| result = st.session_state.workflow.invoke(initial_state) | |
| # 응답 표시 | |
| response = result["final_response"] | |
| st.markdown(response) | |
| # 식단 계획이 있으면 표시 | |
| if result.get("meal_plan"): | |
| st.divider() | |
| st.subheader("📋 추천 식단") | |
| meal_plan = result["meal_plan"] | |
| cols = st.columns(4) | |
| for idx, (meal_type, foods) in enumerate(meal_plan.items()): | |
| with cols[idx % 4]: | |
| st.markdown(f"**{meal_type.upper()}**") | |
| for food in foods[:3]: | |
| nutrients = food.get_nutrients_per_serving(100) | |
| st.caption(f"• {food.name}") | |
| st.caption(f" 칼로리: {nutrients['calories']:.0f}kcal") | |
| st.caption(f" 단백질: {nutrients['protein']:.1f}g") | |
| # 식품 분석 결과가 있으면 표시 | |
| if result.get("food_analysis_results") and result["food_analysis_results"].get("mentioned_foods"): | |
| st.divider() | |
| st.subheader("🔍 식품 영양 분석") | |
| for food_info in result["food_analysis_results"]["mentioned_foods"]: | |
| col1, col2 = st.columns([1, 3]) | |
| with col1: | |
| if food_info['suitable']: | |
| st.success("✅ 적합") | |
| else: | |
| st.warning("⚠️ 주의") | |
| with col2: | |
| st.markdown(f"**{food_info['name']}**") | |
| if not food_info['suitable']: | |
| for issue in food_info['issues']: | |
| st.caption(f"• {issue}") | |
| nutrients = food_info['nutrients'] | |
| st.caption( | |
| f"100g당: 단백질 {nutrients['protein']:.1f}g, " | |
| f"나트륨 {nutrients['sodium']:.0f}mg, " | |
| f"칼륨 {nutrients['potassium']:.0f}mg" | |
| ) | |
| # 응답 저장 | |
| message_data = { | |
| "role": "assistant", | |
| "content": response, | |
| "processing_log": result.get("processing_log", []) | |
| } | |
| st.session_state.messages.append(message_data) | |
| except Exception as e: | |
| st.error(f"오류 발생: {str(e)}") | |
| logger.error(f"Error: {e}", exc_info=True) | |
| # 하단 정보 | |
| st.divider() | |
| col1, col2, col3 = st.columns(3) | |
| with col1: | |
| st.caption("⚠️ 이 시스템은 의료 정보 제공 목적이며, 실제 진료를 대체할 수 없습니다.") | |
| with col2: | |
| if st.button("💬 새 대화 시작"): | |
| st.session_state.messages = [] | |
| st.rerun() | |
| with col3: | |
| if st.button("📥 대화 내용 다운로드"): | |
| conversation = "\n\n".join([ | |
| f"{'사용자' if msg['role'] == 'user' else 'AI'}: {msg['content']}" | |
| for msg in st.session_state.messages | |
| ]) | |
| st.download_button( | |
| label="다운로드", | |
| data=conversation, | |
| file_name=f"kidney_consultation_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt", | |
| mime="text/plain" | |
| ) | |
| if __name__ == "__main__": | |
| main() |