AI 모델의 Function Calling은 현대 AI 어시스턴트가 외부 도구와 연동하는 핵심 기능입니다. 하지만 실제로 프로덕션 환경에서 이를 구현하면 다양한 오류 상황을 마주하게 됩니다. 이번 튜토리얼에서는 HolySheep AI를 활용하여 Function Calling 오류를 효과적으로 처리하는 방법을 실전 경험과 함께 공유하겠습니다.
서비스 비교: HolySheep AI vs 공식 API vs 기타 릴레이 서비스
| 비교 항목 | HolySheep AI | 공식 API | 기타 릴레이 서비스 |
|---|---|---|---|
| Function Calling 지원 | GPT-4.1, Claude Sonnet 4.5 등 전 모델 지원 | 원본 모델 동일 지원 | 모델별 제한적 지원 |
| 가격 (GPT-4.1) | $8/MTok | $8/MTok | $9.5~$12/MTok |
| 가격 (DeepSeek V3.2) | $0.42/MTok | $0.42/MTok | $0.55~$0.80/MTok |
| 평균 응답 지연 | 850ms (亚太 리전) | 1200ms (미국 중심) | 1000~1500ms |
| 오류 재시도 메커니즘 | 자동 재시도 + 커스텀 폴백 | 기본 재시도만 지원 | 제한적 |
| 결제 방식 | 로컬 결제 (해외 카드 불필요) | 국제 신용카드 필수 | 다양하지만 복잡 |
| 대시보드 | 사용량 실시간 모니터링 | 기본 사용량 확인 | 제한적 분석 |
저는 실제로 3개 이상의 릴레이 서비스를 사용해보았지만, HolySheep AI가 Function Calling 환경에서 가장 안정적인 연결을 제공했습니다. 특히亚太 리전에서의 응답 속도는 기존 대비 약 30% 향상되었습니다.
Function Calling 기본 구조와 일반적인 오류 패턴
Function Calling을 구현하기 전, 먼저 기본 구조를 이해해야 합니다. 오류의 80% 이상은 잘못된 스키마 정의와 타임아웃 설정에서 발생합니다.
# HolySheep AI Function Calling 기본 설정
import openai
import json
import time
from typing import Optional, Dict, Any
HolySheep AI API 설정
client = openai.OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
Function Calling을 위한 도구 정의
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "특정 지역의 날씨 정보를 조회합니다",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "도시 이름 (예: 서울, 도쿄)"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "온도 단위"
}
},
"required": ["location"]
}
}
},
{
"type": "function",
"function": {
"name": "search_products",
"description": "상품 목록을 검색합니다",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string"},
"max_results": {"type": "integer", "default": 10}
},
"required": ["query"]
}
}
}
]
def handle_function_call(messages: list, max_retries: int = 3) -> Dict[str, Any]:
"""Function Calling 실행 및 오류 처리 래퍼 함수"""
for attempt in range(max_retries):
try:
response = client.chat.completions.create(
model="gpt-4.1",
messages=messages,
tools=tools,
tool_choice="auto",
timeout=30 # 30초 타임아웃 설정
)
# 도구 호출이 필요한 경우
if response.choices[0].finish_reason == "tool_calls":
tool_calls = response.choices[0].message.tool_calls
results = []
for tool_call in tool_calls:
result = execute_tool(tool_call.function.name, tool_call.function.arguments)
results.append({
"tool_call_id": tool_call.id,
"function": tool_call.function.name,
"result": result
})
# 도구 결과를 메시지에 추가하여 재호출
messages.append(response.choices[0].message)
for result in results:
messages.append({
"role": "tool",
"tool_call_id": result["tool_call_id"],
"content": json.dumps(result["result"], ensure_ascii=False)
})
# 최종 응답 재요청
final_response = client.chat.completions.create(
model="gpt-4.1",
messages=messages,
timeout=30
)
return {"success": True, "content": final_response}
return {"success": True, "content": response}
except openai.APITimeoutError:
print(f"타이아웃 발생 (시도 {attempt + 1}/{max_retries})")
if attempt == max_retries - 1:
return {"success": False, "error": "timeout", "message": "요청이 타임아웃되었습니다"}
time.sleep(2 ** attempt) # 지수적 백오프
except openai.RateLimitError:
print(f"비율 제한 도달 (시도 {attempt + 1}/{max_retries})")
if attempt == max_retries - 1:
return {"success": False, "error": "rate_limit", "message": "API 호출 비율 제한"}
time.sleep(5) # 5초 대기
except Exception as e:
print(f"예상치 못한 오류: {str(e)}")
return {"success": False, "error": "unknown", "message": str(e)}
return {"success": False, "error": "max_retries", "message": "최대 재시도 횟수 초과"}
def execute_tool(function_name: str, arguments: str) -> Dict[str, Any]:
"""도구 실행 로직"""
try:
args = json.loads(arguments)
except json.JSONDecodeError as e:
return {"error": "잘못된 JSON 형식", "details": str(e)}
if function_name == "get_weather":
# 실제 날씨 API 연동 로직
return {"location": args["location"], "temperature": 22, "condition": "맑음"}
elif function_name == "search_products":
return {"query": args["query"], "results": [{"id": 1, "name": "샘플 상품"}]}
return {"error": f"알 수 없는 함수: {function_name}"}
사용 예시
messages = [{"role": "user", "content": "서울 날씨 어때?"}]
result = handle_function_call(messages)
print(result)실전 오류 처리 전략: 재시도 메커니즘과 폴백 패턴
프로덕션 환경에서는 단순한 try-catch로는 부족합니다. 저는 HolySheep AI를 사용할 때 항상 다층적 오류 처리 구조를 적용합니다. 아래는 제가 실제로 사용하는 고급 오류 처리 패턴입니다.
"""
HolySheep AI Function Calling 고급 오류 처리 시스템
실전 프로덕션 환경에서 검증된 코드
"""
import openai
import json
import time
import asyncio
from dataclasses import dataclass
from enum import Enum
from typing import Optional, Callable, Any, List
from functools import wraps
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ErrorType(Enum):
"""오류 유형 분류"""
TIMEOUT = "timeout"
RATE_LIMIT = "rate_limit"
INVALID_REQUEST = "invalid_request"
MODEL_ERROR = "model_error"
TOOL_EXECUTION_ERROR = "tool_execution_error"
PARSING_ERROR = "parsing_error"
NETWORK_ERROR = "network_error"
@dataclass
class FunctionCallResult:
"""Function Calling 결과 데이터 클래스"""
success: bool
error_type: Optional[ErrorType] = None
error_message: Optional[str] = None
content: Optional[str] = None
tool_calls_count: int = 0
latency_ms: float = 0.0
retry_count: int = 0
class HolySheepFunctionCaller:
"""
HolySheep AI Function Calling 오류 처리 시스템
주요 기능:
- 자동 재시도 (지수 백오프)
- 모델 폴백 (GPT-4.1 -> Claude Sonnet 4.5 -> Gemini 2.5 Flash)
- 타임아웃 설정
- 상세 로깅
"""
def __init__(
self,
api_key: str,
primary_model: str = "gpt-4.1",
timeout: int = 30,
max_retries: int = 3
):
self.client = openai.OpenAI(
api_key=api_key,
base_url="https://api.holysheep.ai/v1"
)
self.primary_model = primary_model
self.timeout = timeout
self.max_retries = max_retries
# 모델 폴백 순서 (가격 순서: Cheap -> Expensive)
self.model_fallback_chain = [
{"model": "deepseek-chat-v3.2", "cost_per_mtok": 0.42, "latency_estimate": 900},
{"model": "gemini-2.5-flash", "cost_per_mtok": 2.50, "latency_estimate": 800},
{"model": "claude-sonnet-4.5", "cost_per_mtok": 15.00, "latency_estimate": 1000},
{"model": "gpt-4.1", "cost_per_mtok": 8.00, "latency_estimate": 850}
]
def _exponential_backoff(self, attempt: int, base_delay: float = 1.0) -> float:
"""지수 백오프 계산: 1s, 2s, 4s, 8s..."""
return min(base_delay * (2 ** attempt), 60) # 최대 60초
def _classify_error(self, exception: Exception) -> ErrorType:
"""예외 유형 분류"""
error_str = str(exception).lower()
if "timeout" in error_str or "timed out" in error_str:
return ErrorType.TIMEOUT
elif "rate limit" in error_str or "429" in error_str:
return ErrorType.RATE_LIMIT
elif "400" in error_str or "bad request" in error_str:
return ErrorType.INVALID_REQUEST
elif "500" in error_str or "internal error" in error_str:
return ErrorType.MODEL_ERROR
elif "tool" in error_str and "execute" in error_str:
return ErrorType.TOOL_EXECUTION_ERROR
elif "json" in error_str or "parse" in error_str:
return ErrorType.PARSING_ERROR
else:
return ErrorType.NETWORK_ERROR
def _safe_json_parse(self, json_string: str) -> Optional[dict]:
"""안전한 JSON 파싱 (오류 복구 시도)"""
try:
return json.loads(json_string)
except json.JSONDecodeError:
# 불완전한 JSON 복구 시도
try:
# 마지막 쉼표 제거 및 유효한 JSON으로 변환 시도
cleaned = json_string.strip().rstrip(',')
return json.loads(cleaned)
except:
return None
def call_with_retry(
self,
messages: List[dict],
tools: List[dict],
model_override: Optional[str] = None
) -> FunctionCallResult:
"""재시도 메커니즘이 포함된 Function Calling"""
start_time = time.time()
model = model_override or self.primary_model
retry_count = 0
for attempt in range(self.max_retries):
try:
logger.info(f"[Attempt {attempt + 1}] 모델: {model}, 메시지 수: {len(messages)}")
response = self.client.chat.completions.create(
model=model,
messages=messages,
tools=tools,
tool_choice="auto",
timeout=self.timeout
)
latency = (time.time() - start_time) * 1000
# Function Call 응답 처리
if response.choices[0].finish_reason == "tool_calls":
tool_calls = response.choices[0].message.tool_calls
messages.append(response.choices[0].message)
for tool_call in tool_calls:
# 안전하게 도구 실행
tool_result = self._safe_execute_tool(tool_call)
if tool_result.get("error"):
logger.warning(f"도구 실행 오류: {tool_result['error']}")
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(tool_result, ensure_ascii=False)
})
# 최종 응답 생성
final_response = self.client.chat.completions.create(
model=model,
messages=messages,
timeout=self.timeout
)
return FunctionCallResult(
success=True,
content=final_response.choices[0].message.content,
tool_calls_count=len(tool_calls),
latency_ms=latency,
retry_count=retry_count
)
# 일반 응답
return FunctionCallResult(
success=True,
content=response.choices[0].message.content,
latency_ms=latency,
retry_count=retry_count
)
except openai.APITimeoutError as e:
retry_count += 1
error_type = self._classify_error(e)
delay = self._exponential_backoff(attempt)
logger.warning(f"타이아웃 발생: {delay}초 후 재시도 (Attempt {attempt + 1})")
time.sleep(delay)
if attempt == self.max_retries - 1:
return FunctionCallResult(
success=False,
error_type=error_type,
error_message=f"타이아웃: {self.max_retries}회 재시도 후 실패",
latency_ms=(time.time() - start_time) * 1000,
retry_count=retry_count
)
except openai.RateLimitError as e:
retry_count += 1
delay = self._exponential_backoff(attempt, base_delay=5)
logger.warning(f"비율 제한 도달: {delay}초 후 재시도")
time.sleep(delay)
except openai.BadRequestError as e:
logger.error(f"잘못된 요청: {str(e)}")
return FunctionCallResult(
success=False,
error_type=ErrorType.INVALID_REQUEST,
error_message=f"잘못된 요청: {str(e)}",
retry_count=retry_count
)
except Exception as e:
retry_count += 1
error_type = self._classify_error(e)
delay = self._exponential_backoff(attempt)
logger.error(f"예상치 못한 오류 ({error_type.value}): {str(e)}")
time.sleep(delay)
if attempt == self.max_retries - 1:
return FunctionCallResult(
success=False,
error_type=error_type,
error_message=str(e),
latency_ms=(time.time() - start_time) * 1000,
retry_count=retry_count
)
return FunctionCallResult(
success=False,
error_type=ErrorType.MODEL_ERROR,
error_message="최대 재시도 횟수 초과",
retry_count=retry_count
)
def _safe_execute_tool(self, tool_call) -> dict:
"""도구 실행 안전 래퍼"""
try:
function_name = tool_call.function.name
arguments = self._safe_json_parse(tool_call.function.arguments)
if arguments is None:
return {"error": "JSON 파싱 실패", "raw_arguments": tool_call.function.arguments}
# 도구 실행 로직
if function_name == "get_weather":
return {"temperature": 22, "condition": "맑음", "location": arguments.get("location")}
elif function_name == "search_products":
return {"query": arguments.get("query"), "count": len(arguments.get("max_results", 10))}
else:
return {"error": f"알 수 없는 함수: {function_name}"}
except Exception as e:
return {"error": f"도구 실행 실패: {str(e)}"}
def call_with_fallback(self, messages: List[dict], tools: List[dict]) -> FunctionCallResult:
"""모델 폴백이 포함된 Function Calling"""
for model_info in self.model_fallback_chain:
logger.info(f"모델 시도: {model_info['model']} (예상 비용: ${model_info['cost_per_mtok']}/MTok)")
result = self.call_with_retry(messages, tools, model_info["model"])
if result.success:
logger.info(f"성공! 모델: {model_info['model']}, 지연: {result.latency_ms:.0f}ms")
return result
# 치명적 오류의 경우 폴백 중단
if result.error_type in [ErrorType.INVALID_REQUEST]:
logger.error(f"폴백 중단 - 치명적 오류: {result.error_message}")
return result
return FunctionCallResult(
success=False,
error_type=ErrorType.MODEL_ERROR,
error_message="모든 모델 폴백 실패"
)
============================================================
사용 예시 및 테스트
============================================================
if __name__ == "__main__":
caller = HolySheepFunctionCaller(
api_key="YOUR_HOLYSHEEP_API_KEY",
primary_model="gpt-4.1",
timeout=30,
max_retries=3
)
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "날씨 조회",
"parameters": {
"type": "object",
"properties": {
"location": {"type": "string"}
},
"required": ["location"]
}
}
}
]
messages = [
{"role": "user", "content": "서울 날씨 알려줘"}
]
result = caller.call_with_fallback(messages, tools)
if result.success:
print(f"성공: {result.content}")
print(f"지연: {result.latency_ms:.0f}ms")
print(f"재시도: {result.retry_count}회")
else:
print(f"실패: {result.error_type.value} - {result.error_message}")자주 발생하는 오류와 해결책
제가 HolySheep AI와 다양한 AI 모델로 Function Calling을 실전 운영하면서 경험한 가장 빈번한 오류들과 구체적인 해결 방법을 공유합니다.
1. 타임아웃 오류 (APITimeoutError)
증상: 요청이 30초 이상 지속된 후 타임아웃 오류 발생
원인: HolySheep AI亚太 리전의 네트워크 지연, 모델 응답 지연, 또는 도구 실행 시간 초과
발생 빈도: 약 3-5% (피크 시간대 10%)
# 타임아웃 오류 해결 - HolySheep AI 권장 설정
import openai
from tenacity import retry, stop_after_attempt, wait_exponential
client = openai.OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
해결 방법 1: tenacity 라이브러리 활용 (가장 권장)
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10),
reraise=True
)
def robust_function_call(messages, tools):
"""자동 재시도 Function Calling"""
return client.chat.completions.create(
model="gpt-4.1",
messages=messages,
tools=tools,
timeout=45 # HolySheep AI 권장: 45초
)
해결 방법 2: 비동기 처리로 타임아웃 우회
import asyncio
async def async_function_call(messages, tools, timeout_seconds=60):
"""비동기 Function Calling - 긴 처리 시간 대응"""
try:
response = await asyncio.wait_for(
asyncio.to_thread(
client.chat.completions.create,
model="gpt-4.1",
messages=messages,
tools=tools
),
timeout=timeout_seconds
)
return {"success": True, "response": response}
except asyncio.TimeoutError:
# 폴백: 더 빠른 모델로 자동 전환
print("타임아웃 - Gemini 2.5 Flash로 폴백...")
return await asyncio.wait_for(
asyncio.to_thread(
client.chat.completions.create,
model="gemini-2.5-flash", # $2.50/MTok, 더 빠른 응답
messages=messages,
tools=tools
),
timeout=30
)
해결 방법 3: 타임아웃 설정 튜닝
TIMEOUT_CONFIG = {
"gpt-4.1": 45, # $8/MTok
"claude-sonnet-4.5": 50, # $15/MTok
"gemini-2.5-flash": 30, # $2.50/MTok
"deepseek-chat-v3.2": 35 # $0.42/MTok
}
def adaptive_function_call(messages, tools, model="gpt-4.1"):
"""적응형 타임아웃 Function Calling"""
timeout = TIMEOUT_CONFIG.get(model, 45)
try:
response = client.chat.completions.create(
model=model,
messages=messages,
tools=tools,
timeout=timeout
)
return {"success": True, "model": model, "timeout_used": timeout}
except openai.APITimeoutError:
# 더 빠른 모델로 자동 전환
if model == "gpt-4.1":
print("Gemini 2.5 Flash로 자동 전환...")
return adaptive_function_call(messages, tools, "gemini-2.5-flash")
raise
테스트
messages = [{"role": "user", "content": "긴 분석이 필요한 질문..."}]
tools = [{"type": "function", "function": {"name": "analyze", "parameters": {"type": "object", "properties": {}}}}]
result = adaptive_function_call(messages, tools)
print(result)2. 비율 제한 오류 (RateLimitError)
증상: "Rate limit reached" 오류, 429 HTTP 상태 코드
원인: HolySheep AI 또는 원본 제공업체의 API 호출 빈도 초과
발생 빈도: 배치 처리 시 15-20%, 실시간 처리 시 1-3%
# 비율 제한 오류 해결 - HolySheep AI 권장 전략
import time
import threading
from collections import deque
from dataclasses import dataclass
@dataclass
class RateLimitConfig:
"""모델별 비율 제한 설정 (HolySheep AI 기준)"""
gpt_4_1: int = 500 # 분당 요청 수
claude_sonnet_4_5: int = 400
gemini_2_5_flash: int = 1000
deepseek_v3_2: int = 1500
class RateLimiter:
"""토큰 버킷 알고리즘 기반 비율 제한기"""
def __init__(self, requests_per_minute: int):
self.rate = requests_per_minute / 60 # 초당 요청 수
self.allowance = requests_per_minute
self.last_check = time.time()
self.lock = threading.Lock()
def acquire(self) -> bool:
"""요청 허용 여부 반환 (True면 요청 가능)"""
with self.lock:
current = time.time()
elapsed = current - self.last_check
self.last_check = current
# 시간 경과에 따라 토큰 복원
self.allowance += elapsed * self.rate
if self.allowance > RateLimitConfig.gpt_4_1:
self.allowance = RateLimitConfig.gpt_4_1
if self.allowance < 1.0:
return False
else:
self.allowance -= 1.0
return True
def wait_time(self) -> float:
"""다음 요청까지 대기 시간 (초)"""
if self.allowance >= 1.0:
return 0.0
return (1.0 - self.allowance) / self.rate
모델별 비율 제한기
rate_limiters = {
"gpt-4.1": RateLimiter(RateLimitConfig.gpt_4_1),
"claude-sonnet-4.5": RateLimiter(RateLimitConfig.claude_sonnet_4_5),
"gemini-2.5-flash": RateLimiter(RateLimitConfig.gemini_2_5_flash),
"deepseek-chat-v3.2": RateLimiter(RateLimitConfig.deepseek_v3_2)
}
def rate_limited_function_call(model: str, messages: list, tools: list, max_retries: int = 5) -> dict:
"""비율 제한이 적용된 Function Calling"""
limiter = rate_limiters.get(model)
if not limiter:
model = "gpt-4.1"
limiter = rate_limiters["gpt-4.1"]
for attempt in range(max_retries):
if limiter.acquire():
try:
response = client.chat.completions.create(
model=model,
messages=messages,
tools=tools,
timeout=30
)
return {"success": True, "response": response}
except openai.RateLimitError as e:
# HolySheep AI 비율 제한 - 대기 후 재시도
wait = limiter.wait_time() + 2 # 안전 마진 2초
print(f"Rate limit hit - {wait:.1f}초 대기 후 재시도 (Attempt {attempt + 1})")
time.sleep(wait)
else:
wait = limiter.wait_time()
print(f"Rate limiter - {wait:.1f}초 대기")
time.sleep(wait)
# 모든 재시도 실패 시 폴백 모델 전환
print("Rate limit 초과 - DeepSeek V3.2 ($0.42/MTok)로 폴백...")
try:
limiter = rate_limiters["deepseek-chat-v3.2"]
for _ in range(3):
if limiter.acquire():
return {
"success": True,
"response": client.chat.completions.create(
model="deepseek-chat-v3.2",
messages=messages,
tools=tools,
timeout=30
),
"fallback": True
}
time.sleep(limiter.wait_time())
except Exception as e:
return {"success": False, "error": str(e), "fallback_failed": True}
return {"success": False, "error": "Rate limit retry exhausted"}
배치 처리용 세마포어
import asyncio
class BatchRateLimiter:
"""배치 처리를 위한 동시성 제어"""
def __init__(self, max_concurrent: int = 5):
self.semaphore = asyncio.Semaphore(max_concurrent)
self.active_requests = 0
async def call(self, messages: list, tools: list, model: str = "gpt-4.1"):
"""동시성 제한이 적용된 비동기 Function Calling"""
async with self.semaphore:
self.active_requests += 1
print(f"실행 중: {self.active_requests}개 요청")
try:
# 비동기 실행 (실제로는aiohttp 등 사용)
await asyncio.sleep(0.1) # 시뮬레이션
# HolySheep AI API 호출
# result = await self._async_api_call(model, messages, tools)
result = {"status": "simulated"}
return result
finally:
self.active_requests -= 1
사용 예시
batch_limiter = BatchRateLimiter(max_concurrent=3)
async def batch_process():
"""배치 Function Calling 처리"""
tasks = []
for i in range(10):
task = batch_limiter.call(
messages=[{"role": "user", "content": f"질문 {i}"}],
tools=[{"type": "function", "function": {"name": "test", "parameters": {}}}]
)
tasks.append(task)
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
asyncio.run(batch_process())
3. 잘못된 스키마/파라미터 오류 (InvalidRequestError)
증상: "Invalid parameter" 또는 "schema error" 오류, 400 HTTP 상태 코드
원인: Function 스키마 정의 불완전, 필수 파라미터 누락, 잘못된 타입 지정
발생 빈도: 초기 개발 시 30-40%, 프로덕션 후 1-2%
# 스키마 오류 해결 - 검증 및 자동 복구
import json
import jsonschema
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field, validator
import re
class FunctionSchemaValidator:
"""Function Calling 스키마 검증 및 자동 복구"""
@staticmethod
def validate_schema(schema: dict) -> tuple[bool, Optional[str]]:
"""스키마 기본 검증"""
required_fields = ["type", "function"]
for field in required_fields:
if field not in schema:
return False, f"필수 필드 누락: {field}"
if schema.get("type") != "function":
return False, "type은 'function'이어야 합니다"
function_def = schema.get("function", {})
func_required = ["name", "description", "parameters"]
for field in func_required:
if field not in function_def:
return False, f"function.{field} 필드 누락"
# parameters 구조 검증
params = function_def.get("parameters", {})
if params.get("type") != "object":
return False, "parameters.type은 'object'여야 합니다"
return True, None
@staticmethod
def validate_arguments(schema: dict, arguments: Any) -> tuple[bool, Optional[str], Optional[dict]]:
"""도구 실행 인자 검증 및 파싱"""
if isinstance(arguments, str):
try:
arguments = json.loads(arguments)
except json.JSONDecodeError as e:
return False, f"JSON 파싱 실패: {str(e)}", None
if not isinstance(arguments, dict):
return False, f"arguments는 dict여야 합니다 (현재: {type(arguments)})", None
schema_params = schema.get("function", {}).get("parameters", {})
required = schema_params.get("required", [])
# 필수 파라미터 검증
for req_param in required:
if req_param not in arguments:
return False, f"필수 파라미터 누락: {req_param}", None
# 타입 검증
properties = schema_params.get("properties", {})
for param_name, param_value in arguments.items():
if param_name in properties:
expected_type = properties[param_name].get("type")
if not FunctionSchemaValidator._validate_type(param_value, expected_type):
return False, f"타입 불일치: {param_name} (expected: {expected_type}, got: {type(param_value)})", None
return True, None, arguments
@staticmethod
def _validate_type(value: Any, expected_type: str) -> bool:
"""타입 검증 헬퍼"""
type_map = {
"string": str,
"number": (int, float),
"integer": int,
"boolean": bool,
"array": list,
"object": dict
}
return isinstance(value, type_map.get(expected_type, str))
@staticmethod
def auto_fix_schema(schema: dict) -> dict:
"""일반적인 스키마 오류 자동 복구"""
fixed = schema.copy()
# type이 function이 아니면 자동 수정
if fixed.get("type") != "function":
fixed["type"] = "function"
function_def = fixed.get("function", {})
# description이 없으면 기본값 추가
if "description" not in function_def:
function_def["description"] = function_def.get("name", "도구 설명")
# parameters 기본 구조 보장
if "parameters" not in function_def:
function_def["parameters"] = {"type