Dify로 구축한 AI 워크플로우를 프로덕션 환경에서 운영할 때, 예상치 못한 에러는 반드시 발생합니다. 저도 처음 Dify API를 사용할 때 무수한 에러 메시지와 씨름했죠. 이 튜토리얼에서는 Dify 워크플로우 API를 안정적으로 호출하고, 다양한 에러 상황을 체계적으로 처리하는 방법을 알려드리겠습니다.
Dify 워크플로우 API란?
Dify는 오픈소스 AI 앱 개발 플랫폼으로, 코딩 없이 LLM 기반 애플리케이션을 만들 수 있습니다. 워크플로우는 여러 AI 단계를 순차 또는 병렬로 연결한 자동화 파이프라인입니다. 예를 들어 "사용자 질문 → 관련 문서 검색 → GPT-4로 답변 생성 → 답변 포맷팅" 같은流程을 만들 수 있죠.
기본 API 호출 구조
먼저 Dify 워크플로우 API의 기본 호출 방식을 이해해보겠습니다. HolySheep AI를 사용하면 단일 API 키로 다양한 모델을 Dify와 연동할 수 있습니다.
Python으로 Dify 워크플로우 호출하기
import requests
import json
import time
from typing import Dict, Any, Optional
class DifyWorkflowClient:
"""
Dify 워크플로우 API 호출 클라이언트
HolySheep AI 게이트웨이 사용
"""
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.dify_endpoint = f"{base_url}/dify/workflow"
def run_workflow(
self,
workflow_id: str,
inputs: Dict[str, Any],
timeout: int = 60
) -> Dict[str, Any]:
"""
Dify 워크플로우 실행
Args:
workflow_id: Dify 워크플로우 ID
inputs: 워크플로우 입력 파라미터
timeout: 요청 타임아웃(초)
Returns:
워크플로우 실행 결과
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"workflow_id": workflow_id,
"inputs": inputs,
"response_mode": "blocking" # blocking 또는 streaming
}
try:
response = requests.post(
self.dify_endpoint,
headers=headers,
json=payload,
timeout=timeout
)
# HTTP 상태 코드 확인
response.raise_for_status()
return response.json()
except requests.exceptions.Timeout:
raise DifyTimeoutError(f"워크플로우 실행 시간 초과 ({timeout}초)")
except requests.exceptions.ConnectionError:
raise DifyConnectionError("서버 연결 실패")
except requests.exceptions.HTTPError as e:
raise DifyAPIError(f"HTTP 에러: {e.response.status_code}")
사용 예시
client = DifyWorkflowClient(api_key="YOUR_HOLYSHEEP_API_KEY")
result = client.run_workflow(
workflow_id="your-workflow-id",
inputs={"user_query": "서울 날씨 알려줘"}
)
print(result)커스텀 예외 클래스 정의
효과적인 에러 처리를 위해서는 구체적인 예외 클래스를 정의하는 것이 중요합니다. 이는 에러의 종류를 명확히 구분하고, 적절한 복구策略을 세울 수 있게 해줍니다.
class DifyAPIError(Exception):
"""Dify API 관련 기본 에러"""
def __init__(self, message: str, status_code: int = None, error_code: str = None):
self.message = message
self.status_code = status_code
self.error_code = error_code
super().__init__(self.message)
class DifyTimeoutError(DifyAPIError):
"""워크플로우 실행 시간 초과"""
pass
class DifyConnectionError(DifyAPIError):
"""서버 연결 실패"""
pass
class DifyValidationError(DifyAPIError):
"""입력 파라미터 유효성 검사 실패"""
pass
class DifyRateLimitError(DifyAPIError):
"""API 호출rate limit 초과"""
def __init__(self, message: str, retry_after: int = None):
super().__init__(message)
self.retry_after = retry_after
class DifyQuotaExceededError(DifyAPIError):
"""API 할당량 초과"""
pass
class DifyWorkflowError(DifyAPIError):
"""워크플로우 실행 중 내부 에러"""
def __init__(self, message: str, workflow_error: dict = None):
super().__init__(message)
self.workflow_error = workflow_error
def parse_dify_error(response: requests.Response) -> DifyAPIError:
"""
Dify API 에러 응답 파싱
Args:
response: requests.Response 객체
Returns:
적절한 DifyAPIError 서브클래스 인스턴스
"""
try:
error_data = response.json()
except json.JSONDecodeError:
return DifyAPIError(
message=f"알 수 없는 에러: {response.text[:200]}",
status_code=response.status_code
)
error_code = error_data.get("code", "unknown")
error_message = error_data.get("message", "에러 메시지 없음")
if error_code == "validation_error":
return DifyValidationError(
message=error_message,
status_code=response.status_code,
error_code=error_code
)
elif error_code == "rate_limit_exceeded":
return DifyRateLimitError(
message=error_message,
retry_after=error_data.get("retry_after")
)
elif error_code == "quota_exceeded":
return DifyQuotaExceededError(
message=error_message,
status_code=response.status_code,
error_code=error_code
)
else:
return DifyWorkflowError(
message=error_message,
workflow_error=error_data
)재시도 로직 구현
네트워크 일시적 장애나 서버 과부하로 인한 실패는 다시 시도하면 성공하는 경우가 많습니다. 지수 백오프(Exponential Backoff)를 활용한 재시도 로직을 구현해보겠습니다.
import time
import random
from functools import wraps
from typing import Callable, Any
def retry_with_backoff(
max_retries: int = 3,
base_delay: float = 1.0,
max_delay: float = 60.0,
exponential_base: float = 2.0,
jitter: bool = True
):
"""
지수 백오프를 활용한 재시자 장식자
Args:
max_retries: 최대 재시도 횟수
base_delay: 기본 지연 시간(초)
max_delay: 최대 지연 시간(초)
exponential_base: 지수 증가 기준값
jitter: 랜덤 지터 적용 여부
"""
def decorator(func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs) -> Any:
last_exception = None
for attempt in range(max_retries + 1):
try:
return func(*args, **kwargs)
except DifyTimeoutError:
# 타임아웃은 재시도
last_exception = DifyTimeoutError("재시도 중 타임아웃")
except DifyConnectionError:
# 연결 에러도 재시도
last_exception = DifyConnectionError("재시도 중 연결 실패")
except DifyRateLimitError as e:
# Rate limit 에러는 retry_after만큼 대기
wait_time = e.retry_after if e.retry_after else base_delay
print(f"Rate limit 초과. {wait_time}초 후 재시도...")
time.sleep(wait_time)
last_exception = e
except DifyValidationError:
# 유효성 검사 에러는 재시도 불가
raise
except DifyQuotaExceededError:
# 할당량 초과도 재시도 불가
raise
except DifyAPIError as e:
# 기타 API 에러는 재시도
last_exception = e
# 다음 재시도까지 대기
if attempt < max_retries:
delay = min(base_delay * (exponential_base ** attempt), max_delay)
if jitter:
# 랜덤 지터 추가 (0.5 ~ 1.5배)
delay = delay * (0.5 + random.random())
print(f"재시도 {attempt + 1}/{max_retries} ({delay:.2f}초 후)")
time.sleep(delay)
# 모든 재시도 실패
raise last_exception
return wrapper
return decorator
실제 사용 예시
class RobustDifyClient(DifyWorkflowClient):
@retry_with_backoff(max_retries=3, base_delay=2.0, jitter=True)
def run_workflow_safe(self, workflow_id: str, inputs: Dict[str, Any]) -> Dict[str, Any]:
"""
재시자 로직이 적용된 워크플로우 실행
"""
return self.run_workflow(workflow_id, inputs)
사용
client = RobustDifyClient(api_key="YOUR_HOLYSHEEP_API_KEY")
try:
result = client.run_workflow_safe(
workflow_id="document-qa-workflow",
inputs={
"question": "이 문서의 주요 내용은 무엇인가요?",
"document_id": "doc-12345"
}
)
print(f"성공: {result['data']['outputs']}")
except DifyTimeoutError:
print("워크플로우 실행 시간 초과 - 나중에 다시 시도해주세요")
except DifyValidationError as e:
print(f"입력값 오류: {e.message}")
except DifyQuotaExceededError:
print("API 할당량 초과 - HolySheep AI 대시보드에서 확인하세요")
except DifyAPIError as e:
print(f"API 에러 발생: {e.message}")응답 구조와 에러 파싱
Dify 워크플로우 API의 응답 구조를 이해하는 것은 에러 처리의 핵심입니다. 정상 응답과 에러 응답의 구조를 확인해보겠습니다.
# 정상 응답 구조 예시
success_response = {
"code": "success",
"message": "워크플로우 실행 완료",
"data": {
"id": "run-abc123", # 실행 ID
"workflow_id": "workflow-xyz",
"status": "succeeded", # succeeded, failed, running
"outputs": {
"answer": "서울의 날씨는 맑습니다...",
"confidence": 0.95
},
"elapsed_time": 3.42, # 실행 소요 시간(초)
"total_tokens": 1250,
"created_at": "2024-01-15T10:30:00Z"
}
}
에러 응답 구조 예시
error_response = {
"code": "failed", # 또는 validation_error, rate_limit_exceeded, quota_exceeded
"message": "워크플로우 실행 중 오류가 발생했습니다",
"data": {
"error": {
"type": "node_execution_error",
"node_id": "llm-node-1",
"message": "LLM 응답 생성 실패",
"details": "Model rate limit exceeded"
}
}
}
def handle_workflow_response(response_data: Dict) -> Dict[str, Any]:
"""
Dify 워크플로우 응답 처리
Returns:
.outputs 또는 에러 정보
"""
code = response_data.get("code", "")
if code == "success":
return {
"status": "success",
"result": response_data["data"]["outputs"],
"elapsed_time": response_data["data"].get("elapsed_time"),
"execution_id": response_data["data"]["id"]
}
elif code == "failed":
error_info = response_data.get("data", {}).get("error", {})
raise DifyWorkflowError(
message=error_info.get("message", "워크플로우 실행 실패"),
workflow_error=error_info
)
else:
raise DifyAPIError(
message=response_data.get("message", "알 수 없는 응답"),
error_code=code
)실전 에러 처리 패턴
이제 실제 프로덕션 환경에서 자주 사용되는 에러 처리 패턴들을 정리해보겠습니다. HolySheep AI의 안정적인 연결을 활용하면 대부분의 네트워크 에러를 줄일 수 있습니다.
1. 입력 검증과 사전 체크
from typing import List, Optional
import re
class DifyInputValidator:
"""Dify 워크플로우 입력값 검증"""
MAX_TEXT_LENGTH = 10000
MAX_ARRAY_ITEMS = 100
@staticmethod
def validate_text_input(value: Any, field_name: str, required: bool = True) -> str:
"""텍스트 입력 검증"""
if value is None or value == "":
if required:
raise DifyValidationError(
message=f"'{field_name}' 필드는 필수입니다",
error_code="validation_error"
)
return ""
if not isinstance(value, str):
raise DifyValidationError(
message=f"'{field_name}' 필드는 문자열이어야 합니다",
error_code="validation_error"
)
if len(value) > DifyInputValidator.MAX_TEXT_LENGTH:
raise DifyValidationError(
message=f"'{field_name}' 필드가 너무 깁니다 (최대 {DifyInputValidator.MAX_TEXT_LENGTH}자)",
error_code="validation_error"
)
# 위험한 패턴 검사 (프롬프트 인젝션 방지)
dangerous_patterns = [
r"ignore\s+previous\s+instructions",
r"#{3,}", # Markdown 헤더 과다
]
for pattern in dangerous_patterns:
if re.search(pattern, value, re.IGNORECASE):
raise DifyValidationError(
message=f"'{field_name}'에 부적절한 패턴이 포함되어 있습니다",
error_code="validation_error"
)
return value.strip()
@staticmethod
def validate_workflow_inputs(inputs: Dict[str, Any], required_fields: List[str]) -> Dict[str, Any]:
"""워크플로우 전체 입력값 검증"""
validated = {}
for field in required_fields:
if field not in inputs:
raise DifyValidationError(
message=f"필수 필드 '{field}'가 누락되었습니다",
error_code="validation_error"
)
validated[field] = DifyInputValidator.validate_text_input(
inputs[field], field
)
# 선택적 필드 처리
for field, value in inputs.items():
if field not in required_fields:
if isinstance(value, str):
validated[field] = DifyInputValidator.validate_text_input(
value, field, required=False
)
else:
validated[field] = value
return validated
실전 사용
try:
validated_inputs = DifyInputValidator.validate_workflow_inputs(
inputs={
"question": "서울 날씨 알려줘",
"user_id": "user-12345"
},
required_fields=["question"] # question만 필수
)
result = client.run_workflow(
workflow_id="weather-qa",
inputs=validated_inputs
)
except DifyValidationError as e:
print(f"입력 검증 실패: {e.message}")
# 사용자에게 친숙한 에러 메시지 전달
print("입력값을 확인하고 다시 시도해주세요")2. 폴백 워크플로우 전략
class DifyFallbackStrategy:
"""
워크플로우 실패 시 폴백 전략
메인 워크플로우가 실패하면 단순한 대체 워크플로우로 전환
"""
def __init__(self, client: DifyWorkflowClient):
self.client = client
def run_with_fallback(
self,
primary_workflow_id: str,
fallback_workflow_id: str,
inputs: Dict[str, Any],
enable_simple_fallback: bool = True
) -> Dict[str, Any]:
"""
폴백策略이 적용된 워크플로우 실행
Args:
primary_workflow_id: 주요 워크플로우 ID
fallback_workflow_id: 폴백 워크플로우 ID
inputs: 입력 데이터
enable_simple_fallback: 단순 응답 폴백 활성화
"""
try:
# 먼저 주요 워크플로우 시도
print(f"주요 워크플로우 실행: {primary_workflow_id}")
result = self.client.run_workflow(primary_workflow_id, inputs)
return {
"status": "primary_success",
"result": result["data"]["outputs"],
"workflow": primary_workflow_id
}
except DifyTimeoutError:
# 타임아웃 시 폴백
print("타임아웃 발생 - 폴백 워크플로우 시도")
if enable_simple_fallback:
return self._execute_simple_fallback(
fallback_workflow_id, inputs
)
raise
except DifyWorkflowError as e:
# 워크플로우 내부 에러 시 폴백
print(f"워크플로우 에러: {e.message}")
if enable_simple_fallback:
return self._execute_simple_fallback(
fallback_workflow_id, inputs
)
raise
except DifyRateLimitError:
# Rate limit 도 폴백
print("Rate limit 초과 - 폴백 워크플로우 시도")
if enable_simple_fallback:
return self._execute_simple_fallback(
fallback_workflow_id, inputs
)
raise
def _execute_simple_fallback(
self,
fallback_workflow_id: str,
inputs: Dict[str, Any]
) -> Dict[str, Any]:
"""단순 폴백 워크플로우 실행"""
# 폴백 워크플로우에 최적화된 입력 변환
fallback_inputs = {
"query": inputs.get("question", inputs.get("query", "")),
"mode": "fast" # 빠른 응답 모드
}
result = self.client.run_workflow(fallback_workflow_id, fallback_inputs)
return {
"status": "fallback_used",
"result": result["data"]["outputs"],
"workflow": fallback_workflow_id,
"warning": "빠른 응답 모드로 실행됨"
}
사용 예시
fallback_client = DifyFallbackStrategy(client)
result = fallback_client.run_with_fallback(
primary_workflow_id="advanced-qa-workflow",
fallback_workflow_id="simple-qa-workflow",
inputs={"question": "인공지능의 미래는?", "user_id": "user-123"}
)
if result["status"] == "fallback_used":
print(f"⚠️ {result['warning']}")
print(f"결과: {result['result']}")모니터링과 로깅
에러 처리의 마지막 단계는 모니터링입니다. 발생한 에러를 기록하고 패턴을 분석하면 시스템의 약점을 파악할 수 있습니다.
import logging
from datetime import datetime
from collections import defaultdict
로깅 설정
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class DifyErrorTracker:
"""Dify API 에러 추적 및 모니터링"""
def __init__(self):
self.error_counts = defaultdict(int)
self.error_log = []
self.total_requests = 0
self.successful_requests = 0
def track_request(self, success: bool, error: Exception = None):
"""요청 결과 추적"""
self.total_requests += 1
if success:
self.successful_requests += 1
else:
self.error_counts[type(error).__name__] += 1
self.error_log.append({
"timestamp": datetime.now().isoformat(),
"error_type": type(error).__name__,
"error_message": str(error),
"success_rate": self.get_success_rate()
})
def get_success_rate(self) -> float:
"""성공률 계산"""
if self.total_requests == 0:
return 0.0
return (self.successful_requests / self.total_requests) * 100
def get_error_summary(self) -> dict:
"""에러 요약 반환"""
return {
"total_requests": self.total_requests,
"successful_requests": self.successful_requests,
"failed_requests": self.total_requests - self.successful_requests,
"success_rate": f"{self.get_success_rate():.2f}%",
"error_breakdown": dict(self.error_counts),
"recent_errors": self.error_log[-10:] # 최근 10개 에러
}
def should_alert(self) -> bool:
"""알림 조건 체크"""
# 성공률이 90% 이하이면 알림
return self.get_success_rate() < 90
모니터링이 적용된 클라이언트
class MonitoredDifyClient(DifyWorkflowClient):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.tracker = DifyErrorTracker()
def run_workflow_with_tracking(
self,
workflow_id: str,
inputs: Dict[str, Any]
) -> Dict[str, Any]:
"""모니터링이 적용된 워크플로우 실행"""
try:
result = self.run_workflow(workflow_id, inputs)
self.tracker.track_request(success=True)
logger.info(f"워크플로우 성공: {workflow_id}")
return result
except DifyAPIError as e:
self.tracker.track_request(success=False, error=e)
logger.error(f"Dify API 에러: {e.error_code} - {e.message}")
raise
except Exception as e:
self.tracker.track_request(success=False, error=e)
logger.error(f"예상치 못한 에러: {str(e)}")
raise
주기적 상태 확인
monitored_client = MonitoredDifyClient(api_key="YOUR_HOLYSHEEP_API_KEY")
코드 실행 후...
summary = monitored_client.tracker.get_error_summary()
print(f"총 요청: {summary['total_requests']}")
print(f"성공률: {summary['success_rate']}")
print(f"에러 유형: {summary['error_breakdown']}")
if monitored_client.tracker.should_alert():
print("🚨 알림: 성공률이 임계값 이하입니다!")자주 발생하는 오류와 해결책
1. 타임아웃 에러 (Timeout Error)
# 증상: "DifyTimeoutError: 워크플로우 실행 시간 초과 (60초)"
원인: 워크플로우 실행 시간이 설정된 타임아웃을 초과
해결책 1: 타임아웃 시간 증가
result = client.run_workflow(
workflow_id="complex-workflow",
inputs=inputs,
timeout=120 # 60초에서 120초로 증가
)
해결책 2: 비동기 실행模式下 전환
payload = {
"workflow_id": "complex-workflow",
"inputs": inputs,
"response_mode": "blocking" # → "async"로 변경
}
response = requests.post(endpoint, headers=headers, json=payload)
task_id = response.json()["data"]["id"]
나중에 결과 조회
import time
for _ in range(30): # 최대 5분 대기
result_response = requests.get(f"{endpoint}/runs/{task_id}", headers=headers)
status = result_response.json()["data"]["status"]
if status == "succeeded":
print(result_response.json()["data"]["outputs"])
break
elif status == "failed":
print("실패:", result_response.json()["data"]["error"])
break
time.sleep(10)2. Rate Limit 초과 에러
# 증상: "DifyRateLimitError: Rate limit exceeded"
원인: API 호출 빈도가 허용 범위를 초과
해결책 1: 재시자 로직 활용
@retry_with_backoff(max_retries=5, base_delay=5.0)
def call_with_rate_limit_handling():
return client.run_workflow("workflow-id", inputs)
해결책 2: 호출 간격 조정 (Rate Limiter 적용)
import time
from threading import Lock
class RateLimiter:
"""단순 Rate Limiter"""
def __init__(self, calls_per_second: float):
self.min_interval = 1.0 / calls_per_second
self.last_call = 0
self.lock = Lock()
def wait(self):
with self.lock:
elapsed = time.time() - self.last_call
if elapsed < self.min_interval:
time.sleep(self.min_interval - elapsed)
self.last_call = time.time()
분당 60회로 제한
limiter = RateLimiter(calls_per_second=1.0) # 1초에 1회
for query in queries:
limiter.wait() # 호출 전 대기
result = client.run_workflow("workflow-id", {"query": query})
print(result)
해결책 3: HolySheep AI 대시보드에서 rate limit 확인 및 업그레이드
https://www.holysheep.ai/dashboard 에서 API limits 설정 확인
3. 입력 유효성 검사 실패
# 증상: "DifyValidationError: 'question' 필드는 필수입니다"
원인: 워크플로우에 필요한 필수 필드가 누락되거나 형식이 잘못됨
해결책 1: 입력값 사전 검증
def prepare_workflow_inputs(raw_inputs: dict, schema: dict) -> dict:
"""워크플로우 입력 스키마 기반 검증"""
validated = {}
errors = []
for field, rules in schema.items():
value = raw_inputs.get(field)
# 필수 필드 체크
if rules.get("required") and not value:
errors.append(f"'{field}'는 필수 입력값입니다")
continue
# 타입 체크
if value and rules.get("type"):
expected_type = rules["type"]
if expected_type == "string" and not isinstance(value, str):
errors.append(f"'{field}'는 문자열이어야 합니다")
elif expected_type == "number" and not isinstance(value, (int, float)):
errors.append(f"'{field}'는 숫자여야 합니다")
# 길이 체크
if value and rules.get("max_length"):
if len(str(value)) > rules["max_length"]:
errors.append(f"'{field}'가 너무 깁니다 (최대 {rules['max_length']}자)")
if value:
validated[field] = value
if errors:
raise DifyValidationError(
message="; ".join(errors),
error_code="validation_error"
)
return validated
스키마 정의
input_schema = {
"question": {"required": True, "type": "string", "max_length": 1000},
"user_id": {"required": True, "type": "string"},
"max_results": {"required": False, "type": "number"}
}
사용
try:
validated = prepare_workflow_inputs(
raw_inputs={"question": "서울 날씨", "user_id": "user-123"},
schema=input_schema
)
result = client.run_workflow("qa-workflow", validated)
except DifyValidationError as e:
print(f"입력 오류: {e.message}")
# 사용자에게 표시할 메시지4. 할당량 초과 (Quota Exceeded)
# 증상: "DifyQuotaExceededError: Monthly quota exceeded"
원인: 월간 API 사용량 할당량 소진
해결책 1: 현재 사용량 확인
import requests
def check_quota_usage(api_key: str) -> dict:
"""HolySheep AI 할당량 확인"""
response = requests.get(
"https://api.holysheep.ai/v1/quota",
headers={"Authorization": f"Bearer {api_key}"}
)
data = response.json()
return {
"used": data.get("usage", 0),
"limit": data.get("limit", 0),
"remaining": data.get("remaining", 0),
"reset_date": data.get("reset_date")
}
사용량 확인
quota = check_quota_usage("YOUR_HOLYSHEEP_API_KEY")
print(f"사용량: ${quota['used']:.2f} / ${quota['limit']:.2f}")
print(f"잔여: ${quota['remaining']:.2f}")
해결책 2: 비용 최적화 - 더 저렴한 모델로 전환
HolySheep AI 가격표 참고:
- Gemini 2.5 Flash: $2.50/MTok (초저렴)
- DeepSeek V3.2: $0.42/MTok (가장 저렴)
- GPT-4.1: $8/MTok (고가)
WORKFLOW_MODEL_COSTS = {
"complex-reasoning": "claude-sonnet-4", # 복잡한 추론
"fast-response": "gemini-2.5-flash", # 빠른 응답
"simple-task": "deepseek-v3.2" # 단순 작업
}
def select_cost_effective_model(task_type: str) -> str:
"""작업 유형에 따른 비용 최적화 모델 선택"""
return WORKFLOW_MODEL_COSTS.get(task_type, "gemini-2.5-flash")
해결책 3: HolySheep AI에서 할당량 증가 요청
https://www.holysheep.ai/dashboard -> Billing -> Plan Upgrade
5. 워크플로우 내부 노드 에러
# 증상: "DifyWorkflowError: LLM 노드 응답 생성 실패 - Model rate limit"
원인: 워크플로우 내部的 LLM 노드에서 에러 발생
해결책 1: 에러 정보 상세 분석
def analyze_workflow_error(error_response: dict) -> dict:
"""워크플로우 에러 상세 분석"""
error_data = error_response.get("data", {}).get("error", {})
return {
"error_type": error_data.get("type"),
"node_id": error_data.get("node_id"),
"node_type": error_data.get("node_type"),
"message": error_data.get("message"),
"details": error_data.get("details"),
"suggestion": _get_error_suggestion(error_data)
}
def _get_error_suggestion(error_data: dict) -> str:
"""에러 유형별 권장 해결책"""
error_type = error_data.get("type", "")
message = error_data.get("message", "").lower()
if "rate limit" in message:
return "LLM 모델 rate limit 초과 - 잠시 후 재시도하거나 HolySheep AI에서 모델 quota 확인"
elif "api key" in message or "authentication" in message:
return "API 키 오류 - HolySheep AI에서 API 키 확인"
elif "timeout" in message:
return "LLM 응답 시간 초과 - 워크플로우 타임아웃 증가"
elif "context length" in message or "token limit" in message:
return "입력 토큰 초과 - 입력 텍스트 축소 또는 청킹策略 변경"
else:
return "Dify 워크플로우 설정 확인 필요"
사용
try:
result = client.run_workflow("workflow-id", inputs)
except DifyWorkflowError as e:
if e.workflow_error:
analysis = analyze_workflow_error(
{"data": {"error": e.workflow_error}}
)
print(f"에러 유형: {analysis['error_type']}")
print(f"발생 노드: {analysis['node_id']}")
print(f"권장 해결책: {analysis['suggestion']}")
해결책 2: 워크플로우 재설계 (에러 발생 노드 분리)
Dify 에서 에러 발생 노드에 "실패 시 폴백" 브랜치 추가
https://docs.dify.org/guides/workflow-error-handling
실전 통합 예시: 완전한 에러 처리 파이프라인
"""
완전한 Dify 워크플로우 API 에러 처리 파이프라인
HolySheep AI 게이트웨이 활용
"""
import requests
import time
from datetime import datetime
class ProductionDifyPipeline:
"""
프로덕션 환경용 Dify 워크플로우 파이프라인
- 재시자 로직
- 폴백 워크플로우
- 상세 로깅
- 모니터링
"""
def __init__(self, api_key: str):
self.client = DifyWorkflowClient(api_key)
self.fallback_client = DifyFallbackStrategy(self.client)
self.monitor = MonitoredDifyClient(api_key)
def execute(
self,
primary_workflow: str,
fallback_workflow: str,
inputs: dict,
require_validation: bool = True
) -> dict:
"""
프로덕션 워크플로우 실행
Returns:
{"status": "success/fallback/error", "data": {...}, "logs": [...]}
"""
logs = []
start_time = time.time()
try:
# 1단계: 입력 검증
if require_validation:
validated = DifyInputValidator.validate_workflow_inputs(
inputs,
required_fields=["question"]
)
logs.append({
"time": datetime.now().isoformat(),
"step": "validation",
"status": "passed"
})
else:
validated = inputs
# 2단계: 주요 워크플로우 실행 (모니터링 포함)
result = self.monitor.run_workflow_with_tracking(
primary_workflow,
validated
)
logs.append({
"time": datetime.now().isoformat(),
"step": "primary_workflow",
"status": "success",
"workflow_id": primary_workflow
})
return {
"status": "success",
"data": result["data"]["outputs"],
"execution_time": time.time() - start_time,
"logs": logs
}
except DifyTimeoutError as