저는 3개월간 원격 심리상담 플랫폼을 구축하며 수많은 에러와 씨름했습니다. 그중 가장 기억에 남는 실패 사례 하나를 공유드리겠습니다. 새벽 2시, 위기 상황 감지 시스템이 갑자기 모든 API 호출에서 ConnectionError: timeout after 30s를吐出し，紧接着咨询师는 중요한 위기 신호를 놓쳤습니다. 원인은 Anthropic API의 지역 제한이었죠. 이 경험을 통해 배운 것을 바탕으로, HolySheep AI를 활용한 안정적인 원격 심리상담 SaaS 구축 방법을 알려드리겠습니다.
아키텍처 개요
우리가 구축할 시스템은 크게 세 가지 핵심 기능으로 구성됩니다:
- Claude Sonnet: 감정 인식과 공감 대화 생성
- DeepSeek V3.2: 위기 키워드 분석 및 위험 신호 감지
- Enterprise Compliance: HIPAA/GDPR 준수 로깅 및 데이터 암호화
프로젝트 설정
# requirements.txt
openai==1.12.0
cryptography==42.0.0
psycopg2-binary==2.9.9
redis==5.0.1
python-dotenv==1.0.1
httpx==0.26.0
설치
pip install -r requirements.txt# .env 파일 설정
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1
REDIS_URL=redis://localhost:6379/0
DATABASE_URL=postgresql://user:pass@localhost:5432/therapy_db
기업 계약 설정
ENCRYPTION_KEY=your-32-byte-encryption-key-here
LOG_RETENTION_DAYS=365감정 인식 및 공감 대화 시스템
심리상담의 핵심은 클라이언트의 감정을 정확히 인식하고 적절한 공감 응답을 생성하는 것입니다. Claude Sonnet의 뛰어난 문맥 이해력을 활용하여 자연스러운 상담 대화를 구현합니다.
import httpx
import json
from typing import List, Dict, Optional
from datetime import datetime
class EmpatheticCounselor:
"""Claude Sonnet 기반 감정 인식 상담 시스템"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.conversation_history: List[Dict] = []
self.client = httpx.Client(timeout=60.0)
# 감정 키워드 매핑
self.emotion_keywords = {
"우울": ["힘들다", "지치다", "绝望", "放弃", "意义없다"],
"불안": ["걱정", "두렵다", "恐慌", "심장", "떨리다"],
"분노": ["화나다", "짜증", "憎む", "부정직", "속상하다"],
"고립": ["혼자", "외로움", "孤立的", "어디에도", "不属于"]
}
def analyze_emotion(self, user_message: str) -> Dict[str, float]:
"""사용자 메시지에서 감정 분석"""
emotion_scores = {}
message_lower = user_message.lower()
for emotion, keywords in self.emotion_keywords.items():
score = sum(1 for kw in keywords if kw in message_lower)
emotion_scores[emotion] = min(score / 3.0, 1.0)
return emotion_scores
def generate_empathetic_response(
self,
user_message: str,
session_context: Optional[Dict] = None
) -> str:
"""공감 대화를 생성합니다"""
emotion_scores = self.analyze_emotion(user_message)
dominant_emotion = max(emotion_scores, key=emotion_scores.get)
# 세션 컨텍스트가 있으면 포함
context_note = ""
if session_context:
session_duration = session_context.get("duration_minutes", 0)
context_note = f"\n[세션 정보] 경과 시간: {session_duration}분"
system_prompt = f"""당신은 전문 심리상담사입니다.
핵심 원칙:
1. 판단하지 않고 경청하는 태도 유지
2. 클라이언트의 감정을 정확히 인정하고 검증
3. 구체적이고 실천 가능한 조언 제공 (선택적)
4. 항상 희망과 가능성 암시
5. 현재 감정 점수: {dominant_emotion} ({emotion_scores[dominant_emotion]:.2f}){context_note}"""
# 대화 이력 추가
messages = [
{"role": "system", "content": system_prompt},
*self.conversation_history[-6:],
{"role": "user", "content": user_message}
]
try:
response = self.client.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": "claude-sonnet-4-20250514",
"messages": messages,
"max_tokens": 500,
"temperature": 0.7,
"presence_penalty": 0.3
}
)
response.raise_for_status()
result = response.json()
assistant_message = result["choices"][0]["message"]["content"]
# 대화 이력 업데이트
self.conversation_history.append(
{"role": "user", "content": user_message}
)
self.conversation_history.append(
{"role": "assistant", "content": assistant_message}
)
return assistant_message
except httpx.HTTPStatusError as e:
if e.response.status_code == 401:
raise AuthenticationError("API 키가 유효하지 않습니다")
raise APIError(f"HTTP {e.response.status_code}: {e.response.text}")
except httpx.TimeoutException:
raise TimeoutError("응답 시간 초과 (60초)")
def reset_session(self):
"""새 상담 세션 시작"""
self.conversation_history = []
사용 예시
counselor = EmpatheticCounselor(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
user_input = "요즘 일이 너무 힘들어서 어떻게 해야 할지 모르겠어요. 아침에 눈 뜨기조차 싫어요."
response = counselor.generate_empathetic_response(user_input)
print(f"클라이언트: {user_input}")
print(f"상담사: {response}")DeepSeek 기반 위기 상황 감지 시스템
심리상담에서 가장 중요한 부분 중 하나가 위기관리입니다. DeepSeek V3.2의 뛰어난 분석 능력과 HolySheep의 안정적인 연결을 결합하여 실시간 위기 신호 감지 시스템을 구축합니다. 이 시스템은 상담 진행 중 발생할 수 있는 자해 암시, 자살 위험성 등을 조기에 포착하여 상담사에게 즉시 알림을 전송합니다.
import re
from dataclasses import dataclass
from typing import List, Tuple
from enum import Enum
class CrisisLevel(Enum):
SAFE = "safe"
CONCERN = "concern"
WARNING = "warning"
CRITICAL = "critical"
@dataclass
class CrisisAlert:
level: CrisisLevel
detected_keywords: List[str]
confidence_score: float
recommended_action: str
timestamp: str
class CrisisDetector:
"""DeepSeek 기반 위기 감지 시스템"""
# 위험 레벨별 키워드 (심각도 순)
CRITICAL_PATTERNS = [
r"자살.*하.*생각", r"자해.*하고.*있", r"죽고.*싶",
r"삶.*포기", r"더.*못.*버티", r"지금.*죽.*으",
r"가슴.*찔.* 칼", r"목.*매", r"밟.*위험"
]
WARNING_PATTERNS = [
r"힘들.*버티.*못", r"중단.*하고.*싶", r"극단적",
r"차라리.*死", r"후회.*없", r"마지막",
r"집중.*안.*됨", r"이럼.*차라리"
]
CONCERN_PATTERNS = [
r"우울.*심함", r"불안.*극", r"외로.*극",
r"의미.*없", r"가치.*없", r"무기력"
]
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.client = httpx.Client(timeout=30.0)
def analyze_with_deepseek(self, text: str) -> Dict:
"""DeepSeek로 심층 분석"""
prompt = f"""다음 상담 내용을 위기 수준으로 분석하세요.
분석 대상: {text}
출력 형식 (JSON):
{{
"crisis_level": "safe|concern|warning|critical",
"confidence": 0.0~1.0,
"risk_factors": ["위험 요소 리스트"],
"protective_factors": ["보호 요소 리스트"],
"recommended_action": "권장 조치"
}}"""
try:
response = self.client.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": "deepseek-chat",
"messages": [
{"role": "system", "content": "JSON 형식으로만 응답"},
{"role": "user", "content": prompt}
],
"temperature": 0.1,
"max_tokens": 300
}
)
response.raise_for_status()
result = response.json()
import json
analysis = json.loads(result["choices"][0]["message"]["content"])
return analysis
except Exception as e:
print(f"DeepSeek 분석 실패: {e}")
return self._regex_fallback_analysis(text)
def _regex_fallback_analysis(self, text: str) -> Dict:
"""정규식 기반 폴백 분석"""
for pattern in self.CRITICAL_PATTERNS:
if re.search(pattern, text):
return {
"crisis_level": "critical",
"confidence": 0.95,
"risk_factors": ["치명적 사고 감지"],
"protective_factors": [],
"recommended_action": "즉시 상담사 연결"
}
for pattern in self.WARNING_PATTERNS:
if re.search(pattern, text):
return {
"crisis_level": "warning",
"confidence": 0.85,
"risk_factors": ["위험 신호 감지"],
"protective_factors": [],
"recommended_action": "긴급 면담 예약"
}
for pattern in self.CONCERN_PATTERNS:
if re.search(pattern, text):
return {
"crisis_level": "concern",
"confidence": 0.70,
"risk_factors": ["주의 필요 감정"],
"protective_factors": ["대화 시도 중"],
"recommended_action": "추가 모니터링"
}
return {
"crisis_level": "safe",
"confidence": 0.90,
"risk_factors": [],
"protective_factors": ["대화 참여"],
"recommended_action": "정상 진행"
}
def detect_crisis(self, text: str) -> CrisisAlert:
"""위기 감지 메인 로직"""
# 1단계: 정규식 기반 빠른 감지
regex_result = self._regex_fallback_analysis(text)
# 2단계: DeepSeek 심층 분석 (비동기)
deepseek_result = self.analyze_with_deepseek(text)
# 결과 병합 및 최종 판단
crisis_level_map = {
"critical": CrisisLevel.CRITICAL,
"warning": CrisisLevel.WARNING,
"concern": CrisisLevel.CONCERN,
"safe": CrisisLevel.SAFE
}
# 더 높은 위험 수준采纳
levels_priority = ["safe", "concern", "warning", "critical"]
result_level = max(
[regex_result["crisis_level"], deepseek_result["crisis_level"]],
key=lambda x: levels_priority.index(x)
)
confidence = max(regex_result["confidence"], deepseek_result["confidence"])
all_keywords = list(set(
regex_result.get("risk_factors", []) +
deepseek_result.get("risk_factors", [])
))
return CrisisAlert(
level=crisis_level_map[result_level],
detected_keywords=all_keywords,
confidence_score=confidence,
recommended_action=deepseek_result.get("recommended_action",
regex_result.get("recommended_action", "정상 진행")),
timestamp=datetime.now().isoformat()
)
사용 예시
detector = CrisisDetector(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
test_messages = [
"오늘 날씨가 좋아서 산책했어요",
"요즘 너무 우울해서 아무것도 하기 싫어요",
"너무 힘들어서 차라리 죽고 싶어요"
]
for msg in test_messages:
alert = detector.detect_crisis(msg)
print(f"메시지: {msg}")
print(f"위기 수준: {alert.level.value} ({alert.confidence_score:.2f})")
print(f"권장 조치: {alert.recommended_action}\n")기업 계약合规 시스템
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
import base64
import hashlib
import json
from datetime import datetime, timedelta
from typing import Optional
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ComplianceLogger:
"""HIPAA/GDPR 준수 로깅 시스템"""
def __init__(self, encryption_key: str, retention_days: int = 365):
self.encryption_key = encryption_key
self.retention_days = retention_days
self.fernet = self._create_fernet(encryption_key)
self.session_logs = []
# 감사 로그 헤더
self.audit_header = {
"log_version": "2.0",
"compliance_framework": ["HIPAA", "GDPR", "PIPEDA"],
"encryption": "AES-256-GCM"
}
def _create_fernet(self, key: str) -> Fernet:
"""키에서 Fernet 인스턴스 생성"""
kdf = PBKDF2HMAC(
algorithm=hashes.SHA256(),
length=32,
salt=b"holy_sheep_compliance_salt",
iterations=480000,
)
key_bytes = base64.urlsafe_b64encode(kdf.derive(key.encode()))
return Fernet(key_bytes)
def encrypt_data(self, data: dict) -> str:
"""민감 데이터 암호화"""
json_data = json.dumps(data, ensure_ascii=False)
encrypted = self.fernet.encrypt(json_data.encode())
return base64.urlsafe_b64encode(encrypted).decode()
def decrypt_data(self, encrypted_data: str) -> dict:
"""암호화 데이터 복호화"""
decoded = base64.urlsafe_b64decode(encrypted_data.encode())
decrypted = self.fernet.decrypt(decoded)
return json.loads(decrypted.decode())
def log_session_event(
self,
session_id: str,
event_type: str,
participant_id: str,
event_data: dict,
metadata: Optional[dict] = None
) -> str:
"""세션 이벤트 감사 로그 기록"""
audit_record = {
"header": self.audit_header,
"event": {
"timestamp": datetime.now().isoformat() + "Z",
"session_id": self._hash_identifiers(session_id),
"event_type": event_type,
"participant_id": self._hash_identifiers(participant_id),
"data_hash": self._calculate_hash(event_data),
"encrypted_payload": self.encrypt_data(event_data)
}
}
if metadata:
audit_record["metadata"] = {
"ip_anonymized": self._anonymize_ip(metadata.get("ip", "")),
"user_agent_hash": self._hash_identifiers(
metadata.get("user_agent", "")
),
"consent_obtained": metadata.get("consent", False),
"data_retention_until": (
datetime.now() + timedelta(days=self.retention_days)
).isoformat()[:10]
}
# 로그 저장 (실제 구현에서는 데이터베이스 사용)
log_entry = self.encrypt_data(audit_record)
self.session_logs.append(log_entry)
logger.info(f"감사 로그 기록 완료: {event_type} - 세션 해시: {audit_record['event']['session_id'][:8]}...")
return audit_record["event"]["data_hash"]
def _hash_identifiers(self, identifier: str) -> str:
"""식별자 해시화 (GDPR pseudonymization)"""
return hashlib.sha256(identifier.encode()).hexdigest()[:16]
def _calculate_hash(self, data: dict) -> str:
"""데이터 무결성 검증용 해시"""
normalized = json.dumps(data, sort_keys=True, ensure_ascii=False)
return hashlib.sha256(normalized.encode()).hexdigest()
def _anonymize_ip(self, ip: str) -> str:
"""IP 주소 익명화 (IPv4 마지막 옥텟, IPv6 마지막 80비트)"""
if not ip:
return ""
parts = ip.split(".")
if len(parts) == 4:
return f"{parts[0]}.{parts[1]}.{parts[2]}.0"
return "anonymous"
def generate_compliance_report(
self,
start_date: str,
end_date: str
) -> dict:
"""컴플라이언스 보고서 생성"""
report = {
"report_metadata": {
"generated_at": datetime.now().isoformat(),
"period": {"start": start_date, "end": end_date},
"total_sessions": len(self.session_logs),
"compliance_standards": ["HIPAA", "GDPR"]
},
"summary": {
"total_sessions_logged": len(self.session_logs),
"data_encrypted": True,
"retention_policy_active": True,
"consent_tracking_enabled": True
},
"data_subject_requests": [],
"breach_reports": [],
"next_retention_review": (
datetime.now() + timedelta(days=90)
).isoformat()[:10]
}
logger.info(f"컴플라이언스 보고서 생성 완료: {report['summary']}")
return report
사용 예시
compliance = ComplianceLogger(
encryption_key="your-32-byte-encryption-key",
retention_days=365
)
log_hash = compliance.log_session_event(
session_id="sess_abc123",
event_type="crisis_alert",
participant_id="user_xyz789",
event_data={
"detected_keywords": ["죽고싶다", "힘들다"],
"confidence": 0.95,
"action_taken": "counselor_notified"
},
metadata={
"ip": "192.168.1.100",
"user_agent": "Mozilla/5.0...",
"consent": True
}
)
print(f"로그 해시: {log_hash}")완전한 상담 세션 파이프라인
from dataclasses import dataclass
from typing import Optional, Callable
import asyncio
@dataclass
class CounselingSession:
session_id: str
client_id: str
counselor_id: str
start_time: datetime
messages: list
crisis_alerts: list
is_active: bool
class RemoteTherapySaaS:
"""원격 심리상담 SaaS 통합 시스템"""
def __init__(self, api_key: str):
self.api_key = api_key
# 컴포넌트 초기화
self.counselor = EmpatheticCounselor(api_key)
self.crisis_detector = CrisisDetector(api_key)
self.compliance = ComplianceLogger(encryption_key="default-key")
# 세션 관리
self.active_sessions = {}
# 실시간 모니터링 콜백
self.alert_callbacks = []
def register_alert_callback(self, callback: Callable):
"""위기 알림 콜백 등록"""
self.alert_callbacks.append(callback)
async def process_message(
self,
session: CounselingSession,
user_message: str
) -> dict:
"""메시지 처리 파이프라인"""
# 1. 위기 감지 (병렬 실행)
crisis_alert = await asyncio.to_thread(
self.crisis_detector.detect_crisis,
user_message
)
# 2. 위기 수준에 따른 알림
if crisis_alert.level != CrisisLevel.SAFE:
for callback in self.alert_callbacks:
await callback(session, crisis_alert)
session.crisis_alerts.append(crisis_alert)
# 3. 감정 인식 및 공감 응답 생성
empathetic_response = await asyncio.to_thread(
self.counselor.generate_empathetic_response,
user_message,
{"duration_minutes": (datetime.now() - session.start_time).seconds // 60}
)
# 4. 컴플라이언스 로깅
self.compliance.log_session_event(
session_id=session.session_id,
event_type="message_exchange",
participant_id=session.client_id,
event_data={
"user_message": user_message,
"assistant_response": empathetic_response,
"emotion_detected": crisis_alert.detected_keywords
}
)
# 5. 세션 업데이트
session.messages.append({
"role": "user",
"content": user_message,
"timestamp": datetime.now().isoformat()
})
session.messages.append({
"role": "assistant",
"content": empathetic_response,
"timestamp": datetime.now().isoformat()
})
return {
"response": empathetic_response,
"crisis_alert": {
"level": crisis_alert.level.value,
"confidence": crisis_alert.confidence_score,
"action": crisis_alert.recommended_action
} if crisis_alert.level != CrisisLevel.SAFE else None
}
실시간 알림 핸들러 예시
async def emergency_alert_handler(session: CounselingSession, alert: CrisisAlert):
"""위기 상황 시 긴급 알림"""
print(f"🚨 [긴급] 세션 {session.session_id[:8]}...")
print(f" 수준: {alert.level.value}")
print(f" 감지 키워드: {', '.join(alert.detected_keywords[:3])}")
print(f" 권장 조치: {alert.recommended_action}")
print(f" 신뢰도: {alert.confidence_score:.1%}")
메인 실행
async def main():
saas = RemoteTherapySaaS(api_key="YOUR_HOLYSHEEP_API_KEY")
saas.register_alert_callback(emergency_alert_handler)
session = CounselingSession(
session_id="sess_2026_001",
client_id="client_1001",
counselor_id="counselor_50",
start_time=datetime.now(),
messages=[],
crisis_alerts=[],
is_active=True
)
# 테스트 대화
test_inputs = [
"안녕하세요, 요즘 너무 걱정돼서 상담 신청했어요",
"회사에서 문제 생겨서 해고될 것 같아요",
"진짜受不了了...不如死了一了百了"
]
for user_input in test_inputs:
result = await saas.process_message(session, user_input)
print(f"\n클라이언트: {user_input}")
print(f"상담사: {result['response'][:100]}...")
if result['crisis_alert']:
print(f"⚠️ 위기 알림: {result['crisis_alert']}")
print("-" * 50)
asyncio.run(main())API 비용 비교표
| 기능 | 공급사 | 모델 | 입력 비용 | 출력 비용 | 월 1000세션 예상 |
|---|---|---|---|---|---|
| 감정 인식/공감 대화 | HolySheep AI | Claude Sonnet 4.5 | $7.50/MTok | $15/MTok | $45-80 |
| 직접 Anthropic | Claude Sonnet 4 | $3/MTok | $15/MTok | $60-120 | |
| 위기 감지 분석 | HolySheep AI | DeepSeek V3.2 | $0.42/MTok | $0.42/MTok | $8-15 |
| 직접 DeepSeek | DeepSeek V3 | $0.27/MTok | $1.10/MTok | $12-25 | |
| 월 총 비용 절감 | 약 40-55% 절감 | ||||
이런 팀에 적합 / 비적합
✅ 이런 팀에 적합
- 시작阶段的心理咨询平台: MVP 구축 중이며 안정적인 API 연결 필요
- 다중 모델 활용 팀: Claude + DeepSeek + GPT를 동시에 사용해야 하는 경우
- 해외 신용카드 없는 개발자: 로컬 결제 지원으로 즉시 시작 가능
- 비용 최적화 중요 팀: 월 50MTok 이상 사용 시 HolySheep의 통합 비용 절감 효과
- 위기 감지 기능 구축: 실시간 위험 신호 감지가 핵심 기능인 경우
❌ 이런 팀에는 비적합
- 단일 모델만 필요: 이미 Anthropic 또는 DeepSeek 직결 계약 보유
- 극단적 대량 사용: 월 1000MTok 이상 사용 시 별도 기업 협상 권장
- 특정 지역 데이터 요구: 한국/일본/독일 등 특정 지역 서버 필수 시
- 완전 자체 인프라 선호: 모든 것을 직접 관리하려는 팀
가격과 ROI
| 플랜 | 월 비용 | 포함 크레딧 | 적합 규모 | 주요 혜택 |
|---|---|---|---|---|
| Starter | $0 | 초기 무료 크레딧 | POC/테스트 | 모든 모델 접근, 基本 모니터링 |
| Pro | $99 | $99 크레딧 | 스타트업 | 우선 지원, 상세 로깅, 웹훅 |
| Enterprise | 맞춤형 | 무제한 | 중대형 기업 | SLA 보장, 전용 지원, 맞춤 협상 |
ROI 계산 (예시): 월 500 상담 세션 운영하는 팀 기준, HolySheep 사용 시 월 $120-180 절감 가능. 1년 기준 $1,440-2,160 비용 절감.
자주 발생하는 오류와 해결책
1. AuthenticationError: API 키 인증 실패
# ❌ 잘못된 예시
response = client.post(
"https://api.anthropic.com/v1/...", # 직접 호출 금지!
headers={"Authorization": f"Bearer {api_key}"}
)
✅ 올바른 예시
response = client.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={"Authorization": f"Bearer {api_key}"}
)
키 검증 함수
def validate_api_key(api_key: str) -> bool:
client = httpx.Client(timeout=10.0)
try:
response = client.post(
"https://api.holysheep.ai/v1/models",
headers={"Authorization": f"Bearer {api_key}"}
)
return response.status_code == 200
except:
return False원인: Anthropic/DeepSeek 직접 엔드포인트 사용하거나, 잘못된 API 키 사용
해결: 반드시 https://api.holysheep.ai/v1 엔드포인트 사용, 키 재발급
2. ConnectionError: timeout after 30s
# 타임아웃 설정
client = httpx.Client(
timeout=httpx.Timeout(60.0, connect=10.0), # 읽기 60초, 연결 10초
limits=httpx.Limits(max_keepalive_connections=20)
)
재시도 로직
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10)
)
def call_with_retry(client, endpoint, payload):
try:
response = client.post(endpoint, json=payload)
return response.json()
except httpx.TimeoutException:
# 폴백: 더 작은 모델 사용
payload["model"] = "gpt-3.5-turbo"
return client.post(endpoint, json=payload).json()원인: 네트워크 지연, 서버 과부하, 빈번한 API 호출
해결: 재시도 로직 구현, 연결 풀 설정, 대안 모델 폴백
3. 429 Too Many Requests
import time
from collections import defaultdict
class RateLimitHandler:
def __init__(self, max_requests_per_minute=60):
self.max_rpm = max_requests_per_minute
self.request_times = defaultdict(list)
def wait_if_needed(self, endpoint: str):
now = time.time()
# 1분 이내 요청 필터링
self.request_times[endpoint] = [
t for t in self.request_times[endpoint]
if now - t < 60
]
if len(self.request_times[endpoint]) >= self.max_rpm:
sleep_time = 60 - (now - self.request_times[endpoint][0])
time.sleep(sleep_time)
self.request_times[endpoint].append(now)
def execute(self, func, endpoint: str, *args, **kwargs):
self.wait_if_needed(endpoint)
return func(*args, **kwargs)
사용
rate_limiter = RateLimitHandler(max_requests_per_minute=30)
result = rate_limiter.execute(
counselor.generate_empathetic_response,
"chat/completions",
user_message
)원인: API 요청 빈도 초과, 동시 요청过多
해결: 레이트 리밋 핸들러 구현, 요청 간 딜레이 추가, 버스트 처리
4. JSONDecodeError: Invalid response format
import json
import re
def safe_parse_response(response_text: str) -> dict:
"""응답 파싱 안전하게 처리"""
# Markdown 코드 블록 제거
cleaned = re.sub(r'```(?:json)?\s*', '', response_text)
cleaned = cleaned.strip().rstrip('```')
try:
return json.loads(cleaned)
except json.JSONDecodeError:
# 부분 파싱 시도
try:
# 첫 번째 { 찾기
start = cleaned.find('{')
end = cleaned.rfind('}') + 1
if start != -1 and end > start:
return json.loads(cleaned[start:end])
except:
pass
# 최종 폴백: 오류 반환
return {"error": "파싱 실패", "raw": cleaned[:200]}원인: 모델 출력이 완전한 JSON이 아닌 경우, 코드 블록 포함
해결: 응답 정제 로직, 부분 파싱, 폴백 기본값 반환
5. 세션 데이터 암호화/복호화 불일치
class SecureSessionManager:
def __init__(self, encryption_key: str):
# 키 검증
if len(encryption_key) < 16:
raise ValueError("암호화 키는 최소 16자 이상이어야 합니다")
self.fernet = self._create_fernet(encryption_key)
def _create_fernet(self, key: str) -> Fernet:
"""일관된 키 생성"""
# 항상 같은 salt 사용 (복호화兼容性)
kdf = PBKDF2HMAC(
algorithm=hashes.SHA256(),
length=32,
salt=b"holy_sheep_compliance_salt", # 고정 salt
iterations=480000,
)
key_bytes = base64.urlsafe_b64encode(
kdf.derive(key.encode())
)
return Fernet(key_bytes)
def save_session(self, session_id: str