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" # 일반 정보
@dataclass
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
@dataclass
class RecommendationItem:
"""추천 항목"""
name: str
category: str # 식이, 운동, 약물 등
description: str
constraints_satisfied: bool
embedding: Optional[np.ndarray] = None
@dataclass
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. 구체적인 권장사항은 - 태그
로 표시
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'- (.*?)
'
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("""
""", 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()