AI 모델이 함수를 호출할 때 가장 흔하게遭遇하는 문제가 바로 파라미터 추출 실패입니다. 저도 처음 Function Calling을 구현했을 때, 모델이 {"name": "get_weather", "arguments": "{"location":"}처럼 불완전한 JSON을 반환해서 하루 종일 디버깅한 경험이 있습니다.
이번 튜토리얼에서는 HolySheep AI를 사용하여 Function Calling 오류를 체계적으로 처리하는 재시도 전략을 단계별로 알아보겠습니다. HolySheep AI는 단일 API 키로 다양한 AI 모델을 통합할 수 있어 여러 모델의 Function Calling 동작을 비교 테스트하기 정말 좋습니다.
1. Function Calling이란?
Function Calling(함수 호출)은 AI 모델이 개발자가 정의한 함수를 실행할 수 있게 하는 기술입니다. 예를 들어 사용자가 "서울 날씨 알려줘"라고 하면, 모델은 자동으로 get_weather 함수를 호출하여 실제 날씨 데이터를 가져올 수 있습니다.
하지만 문제는 여기서 발생합니다. 모델이 항상 정확한 파라미터를 생성하는 것은 아니며, 네트워크 오류나Rate Limit 등의 이유로 호출이 실패할 수 있습니다. 이러한 상황에서 적절한 재시도 전략이 필수적입니다.
2. HolySheep AI에서 Function Calling 기본 설정
먼저 HolySheep AI에서 Function Calling을 사용하는 기본 코드를 살펴보겠습니다. HolySheep AI는 지금 가입하면 무료 크레딧을 제공하며, GPT-4.1, Claude Sonnet, Gemini 등 주요 모델을 단일 API 키로 테스트할 수 있습니다.
import openai
import json
import time
HolySheep AI API 설정
client = openai.OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
Function 정의
functions = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "특정 지역의 날씨 정보를 가져옵니다",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "도시 이름 (예: 서울, 도쿄, 뉴욕)"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "온도 단위"
}
},
"required": ["location"]
}
}
}
]
def get_weather(location: str, unit: str = "celsius") -> dict:
"""날씨 정보 반환 (시뮬레이션)"""
return {
"location": location,
"temperature": 22,
"unit": unit,
"condition": "맑음"
}
기본 Function Calling 요청
response = client.chat.completions.create(
model="gpt-4.1",
messages=[
{"role": "user", "content": "서울 날씨 어때?"}
],
tools=functions,
tool_choice="auto"
)
print("응답:", response.choices[0].message)3. 재시도 전략의 핵심: Exponential Backoff
파라미터 추출 실패가 발생했을 때 가장 효과적인 재시도 전략은 Exponential Backoff(지수적 백오프)입니다. 이 방식은 실패할수록 대기 시간을 2배, 4배, 8배로 늘려가며 재시도합니다.
HolySheep AI의 평균 응답 지연 시간은 약 800~1200ms 정도이며, Rate Limit 발생 시 적절한 재시도 대기 시간을 두면 불필요한 실패를 줄일 수 있습니다.
import openai
import json
import time
import random
from typing import Any, Optional
from dataclasses import dataclass
client = openai.OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
@dataclass
class RetryConfig:
"""재시도 설정"""
max_retries: int = 5
base_delay: float = 1.0 # 기본 대기 시간 (초)
max_delay: float = 60.0 # 최대 대기 시간 (초)
exponential_base: float = 2.0
class FunctionCallingError(Exception):
"""Function Calling 관련 오류"""
def __init__(self, message: str, error_type: str, retry_count: int):
super().__init__(message)
self.error_type = error_type
self.retry_count = retry_count
def extract_json_from_arguments(arguments_str: str) -> dict:
"""arguments 문자열에서 JSON 파싱 시도"""
try:
return json.loads(arguments_str)
except json.JSONDecodeError:
# 불완전한 JSON 복구 시도
# 1. 마지막 쉼표 제거
fixed = arguments_str.rstrip(',')
# 2. 괄호 쌍 맞추기
open_braces = fixed.count('{') - fixed.count('}')
if open_braces > 0:
fixed += '}' * open_braces
try:
return json.loads(fixed)
except json.JSONDecodeError:
raise FunctionCallingError(
f"JSON 파싱 실패: {arguments_str[:100]}...",
"JSON_PARSE_ERROR",
retry_count=0
)
def call_with_retry(
client,
messages: list,
functions: list,
model: str = "gpt-4.1",
config: RetryConfig = None
) -> dict:
"""재시도 로직이 포함된 Function Calling"""
if config is None:
config = RetryConfig()
last_error = None
for attempt in range(config.max_retries + 1):
try:
response = client.chat.completions.create(
model=model,
messages=messages,
tools=functions,
tool_choice="auto"
)
message = response.choices[0].message
# 도구 호출이 없는 경우
if not message.tool_calls:
return {"type": "text", "content": message.content}
# 각 도구 호출 처리
results = []
for tool_call in message.tool_calls:
function_name = tool_call.function.name
arguments_str = tool_call.function.arguments
# 파라미터 추출 및 검증
try:
arguments = extract_json_from_arguments(arguments_str)
except FunctionCallingError as e:
# 파라미터 추출 실패 시 재시도
raise FunctionCallingError(
f"파라미터 추출 실패: {str(e)}",
"PARAMETER_EXTRACTION_ERROR",
retry_count=attempt
)
results.append({
"name": function_name,
"arguments": arguments
})
return {"type": "function_calls", "calls": results}
except FunctionCallingError as e:
last_error = e
if attempt < config.max_retries:
# 지수적 대기 시간 계산
delay = min(
config.base_delay * (config.exponential_base ** attempt),
config.max_delay
)
# 지터(Jitter) 추가 - 동시 요청 충돌 방지
delay += random.uniform(0, 0.5)
print(f"시도 {attempt + 1} 실패, {delay:.2f}초 후 재시도...")
time.sleep(delay)
else:
raise last_error
except Exception as e:
# 기타 오류 처리 (Rate Limit, Network 등)
last_error = e
if attempt < config.max_retries:
delay = config.base_delay * (config.exponential_base ** attempt)
print(f"오류 발생: {str(e)}, {delay:.2f}초 후 재시도...")
time.sleep(delay)
else:
raise FunctionCallingError(
f"최대 재시도 횟수 초과: {str(last_error)}",
"MAX_RETRIES_EXCEEDED",
retry_count=attempt
)
사용 예시
functions = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "특정 지역의 날씨 정보를 가져옵니다",
"parameters": {
"type": "object",
"properties": {
"location": {"type": "string"},
"unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
},
"required": ["location"]
}
}
}
]
messages = [{"role": "user", "content": "서울 날씨 어때?"}]
try:
result = call_with_retry(client, messages, functions)
print("결과:", result)
except FunctionCallingError as e:
print(f"최종 오류: {e.error_type}, 재시도 횟수: {e.retry_count}")4. 고급 재시도 전략: Circuit Breaker 패턴
단순 재시도만으로는 충분하지 않은 경우가 있습니다. 예를 들어 API가 완전히 다운되거나 Rate Limit가 지속적으로 발생하면 무한 재시도는 시스템 전체를 마비시킬 수 있습니다.
Circuit Breaker(서킷 브레이커) 패턴은 이러한 상황을 방지합니다. 연속 실패 횟수가 임계치를 넘으면ircuit를 "열어" 요청을 차단하고, 일정 시간 후 다시 시도합니다.
import time
from enum import Enum
from datetime import datetime, timedelta
from dataclasses import dataclass, field
class CircuitState(Enum):
CLOSED = "closed" # 정상 - 요청 통과
OPEN = "open" # 차단 - 요청 거부
HALF_OPEN = "half_open" #_half-open 상태 - 테스트 요청 허용
@dataclass
class CircuitBreaker:
"""서킷 브레이커 구현"""
failure_threshold: int = 5 # 서킷을 열失败的 연속 요청 수
success_threshold: int = 3 # 서킷을 닫을 연속 성공 요청 수
timeout: float = 30.0 # 서킷이 다시 열릴 때까지 대기 시간 (초)
# 내부 상태
failures: int = 0
successes: int = 0
state: CircuitState = CircuitState.CLOSED
last_failure_time: float = field(default_factory=time.time)
next_attempt_time: float = field(default_factory=time.time)
def record_success(self):
"""성공 기록"""
self.successes += 1
self.failures = 0
if self.state == CircuitState.HALF_OPEN:
if self.successes >= self.success_threshold:
self.state = CircuitState.CLOSED
print("서킷 브레이커: CLOSED 상태로 전환 (복구 완료)")
def record_failure(self):
"""실패 기록"""
self.failures += 1
self.successes = 0
self.last_failure_time = time.time()
if self.state == CircuitState.CLOSED:
if self.failures >= self.failure_threshold:
self.state = CircuitState.OPEN
self.next_attempt_time = time.time() + self.timeout
print(f"서킷 브레이커: OPEN 상태로 전환 ({self.timeout}초 후 재시도)")
elif self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.OPEN
self.next_attempt_time = time.time() + self.timeout
print("서킷 브레이커: HALF_OPEN에서 OPEN으로 전환")
def can_execute(self) -> bool:
"""요청 실행 가능 여부 확인"""
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
if time.time() >= self.next_attempt_time:
self.state = CircuitState.HALF_OPEN
self.successes = 0
print("서킷 브레이커: HALF_OPEN 상태로 전환 (테스트 요청 허용)")
return True
return False
# HALF_OPEN 상태에서는 항상 요청 허용
return True
class RobustFunctionCaller:
"""재시도 + 서킷 브레이커가 적용된 Function Caller"""
def __init__(self, client, max_retries: int = 3):
self.client = client
self.max_retries = max_retries
self.circuit_breaker = CircuitBreaker(
failure_threshold=5,
success_threshold=2,
timeout=30.0
)
def call_function(
self,
messages: list,
functions: list,
model: str = "gpt-4.1"
) -> dict:
"""안전하게 Function Calling 실행"""
# 서킷 브레이커 확인
if not self.circuit_breaker.can_execute():
remaining_time = self.circuit_breaker.next_attempt_time - time.time()
raise FunctionCallingError(
f"서킷 브레이커가 OPEN 상태입니다. {remaining_time:.1f}초 후 재시도하세요.",
"CIRCUIT_BREAKER_OPEN",
retry_count=0
)
try:
result = self._execute_with_retry(messages, functions, model)
self.circuit_breaker.record_success()
return result
except Exception as e:
self.circuit_breaker.record_failure()
raise
def _execute_with_retry(
self,
messages: list,
functions: list,
model: str,
attempt: int = 0
) -> dict:
"""재시도 로직 포함 실행"""
try:
response = self.client.chat.completions.create(
model=model,
messages=messages,
tools=functions,
tool_choice="auto"
)
message = response.choices[0].message
if not message.tool_calls:
return {"type": "text", "content": message.content}
# 도구 호출 파싱
calls = []
for tool_call in message.tool_calls:
try:
arguments = json.loads(tool_call.function.arguments)
except json.JSONDecodeError as e:
# 파라미터 파싱 실패 시 재시도
if attempt < self.max_retries:
print(f"파라미터 파싱 실패, {attempt + 1}번째 재시도...")
return self._execute_with_retry(
messages, functions, model, attempt + 1
)
raise FunctionCallingError(
f"파라미터 추출 실패: {str(e)}",
"PARAMETER_EXTRACTION_ERROR",
retry_count=attempt
)
calls.append({
"name": tool_call.function.name,
"arguments": arguments
})
return {"type": "function_calls", "calls": calls}
except Exception as e:
if attempt < self.max_retries:
delay = 2 ** attempt
print(f"실행 실패, {delay}초 후 재시도... ({attempt + 1}/{self.max_retries})")
time.sleep(delay)
return self._execute_with_retry(messages, functions, model, attempt + 1)
raise
사용 예시
caller = RobustFunctionCaller(client)
functions = [
{
"type": "function",
"function": {
"name": "search_products",
"description": "제품 검색",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string"},
"category": {"type": "string"}
},
"required": ["query"]
}
}
}
]
messages = [{"role": "user", "content": "노트북 추천해줘"}]
try:
result = caller.call_function(messages, functions)
print("결과:", result)
except FunctionCallingError as e:
print(f"오류: {e.error_type}")
print(f"메시지: {str(e)}")자주 발생하는 오류와 해결책
오류 1: Invalid JSON 형식으로 인한 파라미터 추출 실패
오류 메시지:
{
"error": {
"code": "INVALID_JSON",
"message": "Function arguments parsing failed: Unexpected end of JSON input"
}
}원인: 모델이 불완전한 JSON을 반환하거나, 중첩된 구조에서 괄호가 어긋난 경우 발생합니다. 특히 GPT-4.1 모델에서 긴 컨텍스트 사용 시 발생 빈도가 높아집니다.
해결 코드:
import json
import re
def safe_parse_arguments(arguments: str) -> dict:
"""안전한 JSON 파싱 및 복구"""
# 1단계: 기본 파싱 시도
try:
return json.loads(arguments)
except json.JSONDecodeError:
pass
# 2단계: 불완전한 괄호 복구
def balance_braces(s):
"""중괄호 쌍 맞추기"""
open_count = s.count('{')
close_count = s.count('}')
# 부족한 닫는 괄호 추가
while close_count < open_count:
s += '}'
close_count += 1
# 여분의 닫는 괄호 제거 (문자열 앞쪽에서)
while close_count > open_count:
# 마지막 닫는 괄호 찾기
last_close = s.rfind('}')
if last_close > 0:
s = s[:last_close]
else:
break
close_count = s.count('}')
return s
# 3단계: 따옴표 문제 해결
fixed = arguments
# 불필요한 이스케이프 제거
fixed = fixed.replace('\\"', '"')
fixed = fixed.replace('\\\\', '\\')
# 따옴표 균형 맞추기
fixed = balance_braces(fixed)
# 4단계: 파싱 재시도
try:
return json.loads(fixed)
except json.JSONDecodeError as e:
# 5단계: 마지막逗号(쉼표) 제거 후 재시도
fixed = re.sub(r',(\s*[}\]])', r'\1', fixed)
try:
return json.loads(fixed)
except json.JSONDecodeError:
raise ValueError(f"JSON 복구 실패: 원본={arguments[:100]}")오류 2: Rate Limit 초과 (429 Too Many Requests)
오류 메시지:
{
"error": {
"code": "rate_limit_exceeded",
"message": "Rate limit exceeded for model gpt-4.1. Retry after 5 seconds.",
"retry_after": 5
}
}원인: HolySheep AI의 Rate Limit에 도달했거나, 모델별 동시 요청 제한을 초과했습니다. HolySheep AI의 Rate Limit는 요청 빈도에 따라 동적으로 조정됩니다.
해결 코드:
import time
import asyncio
from typing import Callable, Any
class RateLimitHandler:
"""Rate Limit 전용 처리기"""
def __init__(self):
self.request_times = []
self.max_requests_per_minute = 60
self.min_interval = 60 / self.max_requests_per_minute
def wait_if_needed(self):
"""Rate Limit 도달 시 대기"""
current_time = time.time()
# 1분 이내 요청 기록 필터링
self.request_times = [
t for t in self.request_times
if current_time - t < 60
]
# Rate Limit 확인
if len(self.request_times) >= self.max_requests_per_minute:
oldest = min(self.request_times)
wait_time = 60 - (current_time - oldest) + 1
print(f"Rate Limit 도달, {wait_time:.1f}초 대기...")
time.sleep(wait_time)
self.request_times.append(time.time())
async def async_call_with_rate_limit(
handler: RateLimitHandler,
func: Callable,
*args,
**kwargs
) -> Any:
"""비동기 Rate Limit 처리"""
handler.wait_if_needed()
return await func(*args, **kwargs)
사용 예시
async def call_api_async():
handler = RateLimitHandler()
async def make_request():
return await async_call_with_rate_limit(
handler,
# 실제 API 호출 함수를 여기에 넣습니다
lambda: client.chat.completions.create(
model="gpt-4.1",
messages=[{"role": "user", "content": "테스트"}],
tools=[]
)
)
# 동시 요청 시뮬레이션
tasks = [make_request() for _ in range(10)]
results = await asyncio.gather(*tasks, return_exceptions=True)
return results오류 3: 필수 파라미터 누락
오류 메시지:
{
"error": {
"code": "MISSING_REQUIRED_PARAMETER",