저는 과거 글로벌 제조 SaaS 스타트업에서 AI 통합 파이프라인을 구축했던 엔지니어입니다. 이번 튜토리얼에서는 3D 프린팅 공장의 생산 스케줄링을 자동화하는 멀티 모델 AI Agent 시스템을 HolySheep AI 게이트웨이를 활용해 구축하는 방법을 단계별로 설명드리겠습니다. 슬라이싱 최적화에는 GPT-4.1을, 고객 커뮤니케이션에는 Claude Sonnet 4.5를, 그리고 비용 관리에는 Gemini 2.5 Flash와 DeepSeek V3.2를 활용하는 하이브리드 아키텍처를 다룹니다.
1. 3D 프린팅 공장 스케줄링 자동화의 필요성
3D 프린팅 공장에서 가장 큰 운영 부담은 바로 생산 스케줄링입니다. 수십 개의 프린터, 수백 개의 주문, 다양한 재료별 소모량, 납기 일정, 그리고 고객별 우선순위를 동시에 고려해야 합니다. 수동 스케줄링은 인간 오류와 자원 낭비를 초래하며, 특히 금형 제작이나 의료기기 분야에서 치명적인 딜레이를 유발할 수 있습니다.
저는 이전 프로젝트에서 일간 500건 이상의 주문과 30대 이상의 프린터를 관리하는 공장에서 이 문제를 직접 경험했습니다. 기존 ERP 시스템의 스케줄링 모듈은 지나치게 경직되어 실시간 변경에 대응하지 못했고, 결국 생산 관리자가 매일 2시간 이상을 수동 조정에 소비해야 했습니다. 이 문제를 해결하기 위해 HolySheep AI 게이트웨이 기반의 멀티 모델 AI Agent 시스템을 도입했고, 결과적으로 스케줄링 시간을 85% 절감했습니다.
2. 월 1,000만 토큰 기준 비용 비교표
HolySheep AI를 통해 API를 통합하는 개발자분들께 가장 먼저 확인해야 할 부분이 비용입니다. 아래 표는 2026년 5월 기준 주요 AI 모델의 출력 토큰 가격과 월 1,000만 토큰 사용 시 총 비용을 비교한 것입니다.
| 모델 | 출력 토큰 가격 | 월 1,000만 토큰 비용 | 주요 용도 | 특징 |
|---|---|---|---|---|
| GPT-4.1 | $8.00/MTok | $80 | 슬라이싱 최적화 | 코드 생성 최고, 구조화된 출력 강점 |
| Claude Sonnet 4.5 | $15.00/MTok | $150 | 고객 커뮤니케이션 | 프롬프트 정확도 최고, 장문 작성 전문 |
| Gemini 2.5 Flash | $2.50/MTok | $25 | 配额监控 대시보드 | 가성비 최고, 빠른 응답 속도 |
| DeepSeek V3.2 | $0.42/MTok | $4.20 | 로그 분석, 기본 라우팅 | 비용 효율성 최고, 가벼운 작업 적합 |
3. HolySheep AI 선택 시 월 1,000만 토큰 비용 최적화 시나리오
| 사용 시나리오 | GPT-4.1 | Claude Sonnet 4.5 | Gemini 2.5 Flash | DeepSeek V3.2 | 총 월 비용 |
|---|---|---|---|---|---|
| 기본 플랜 (각 250만 토큰) |
$20 | $37.50 | $6.25 | $1.05 | $64.80 |
| 고성능 플랜 (슬라이싱 500만, Claude 300만, Gemini 150만, DeepSeek 50만) |
$40 | $45 | $3.75 | $0.21 | $88.96 |
| 비용 최적화 플랜 (Claude only 500만, 나머지 DeepSeek) |
$40 → DeepSeek $2.10 |
$75 | → DeepSeek $1.05 | $3.36 | $79.41 |
4. 시스템 아키텍처 개요
저가 설계한 3D 프린팅 스케줄링 시스템은 네 개의 핵심 Agent로 구성됩니다. 각 Agent는 특정 역할에 최적화된 모델을 사용하며, HolySheep AI의 단일 API 키로 모두 연결됩니다.
- 슬라이싱 최적화 Agent: GPT-4.1 — STL 파일 분석, 적층 방향 결정, 서포트 구조 생성
- 고객 커뮤니케이션 Agent: Claude Sonnet 4.5 — 주문 확인, 납기 통보, 변경 사항 안내
- 스케줄링 로직 Agent: DeepSeek V3.2 — 우선순위 계산, 자원 배분, 충돌 감지
- 모니터링 Agent: Gemini 2.5 Flash — API 사용량 대시보드, 비용 알림, QoS 모니터링
5. 핵심 기능 구현 코드
5.1 슬라이싱 최적화 Agent — GPT-4.1
슬라이싱 최적화는 3D 프린팅에서 인쇄 품질과 소요 시간을 좌우하는 핵심 과정입니다. 저는 수백 개의 STL 파일을 배치 처리하면서 적층 방향, 서포트 사용 여부,fills density 등을 자동으로 최적화하는 파이프라인을 구축했습니다.
"""
HolySheep AI 게이트웨이 — 슬라이싱 최적화 Agent
GPT-4.1을 활용한 3D 모델 최적화 및 슬라이서 파라미터 생성
"""
import requests
import json
from typing import Dict, List, Optional
class SlicingOptimizerAgent:
"""3D 프린팅 슬라이싱 파라미터 자동 최적화 Agent"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.model = "gpt-4.1"
def optimize_slicing_params(
self,
stl_file_path: str,
material: str,
printer_type: str,
priority: str = "normal"
) -> Dict:
"""
STL 파일 분석 후 최적 슬라이싱 파라미터 생성
Args:
stl_file_path: STL 파일 경로
material: 재료 유형 (pla, abs, petg, resin, nylon)
printer_type: 프린터 유형 (fdm, sla, sls)
priority: 작업 우선순위
Returns:
최적화된 슬라이서 설정 딕셔너리
"""
# STL 파일 메타데이터 분석 (실제 구현 시 Trimesh 라이브러리 활용)
model_analysis = self._analyze_stl_geometry(stl_file_path)
# GPT-4.1에 최적화 요청
prompt = self._build_optimization_prompt(
model_analysis, material, printer_type, priority
)
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": self.model,
"messages": [
{
"role": "system",
"content": "당신은 3D 프린팅 슬라이싱 최적화 전문가입니다. \
입력된 STL 모델의 지오메트리를 분석하고, \
재료와 프린터 유형에 맞는 최적 슬라이싱 파라미터를 \
JSON 형태로 반환합니다."
},
{
"role": "user",
"content": prompt
}
],
"temperature": 0.3,
"response_format": {"type": "json_object"}
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
if response.status_code != 200:
raise RuntimeError(
f"GPT-4.1 API 오류: {response.status_code} - {response.text}"
)
result = response.json()
return json.loads(result["choices"][0]["message"]["content"])
def _analyze_stl_geometry(self, file_path: str) -> Dict:
"""STL 파일 지오메트리 분석 (기본값 반환)"""
return {
"vertices": 15000,
"faces": 28000,
"bounding_box": {"x": 200, "y": 150, "z": 100},
"surface_area": 45000,
"volume": 2500000,
"has_overhangs": True,
"overhang_angle": 45
}
def _build_optimization_prompt(
self,
analysis: Dict,
material: str,
printer: str,
priority: str
) -> str:
return f"""3D 모델 분석 결과:
- 정점 수: {analysis['vertices']}
- 면 수: {analysis['faces']}
- 바운딩 박스: {analysis['bounding_box']}mm
- 함몰 구조 있음: {analysis['has_overhangs']}
- 함몰 각도: {analysis['overhang_angle']}도
재료: {material}
프린터: {printer}
우선순위: {priority}
아래 JSON 형식으로 최적 슬라이싱 파라미터를 생성해주세요:
{{
"layer_height": 마이크론 단위,
"infill_percentage": 퍼센트,
"infill_pattern": 패턴_유형,
"wall_thickness": mm,
"support_type": support_유형,
"support_angle": 도,
"print_speed": mm/s,
"nozzle_temperature": 도,
"bed_temperature": 도,
"fan_speed": 퍼센트,
"estimated_time_minutes": 예상_인쇄_시간_분,
"material_usage_grams": 예상_소모량_g,
"quality_notes": 품질_설명
}}"""
사용 예시
if __name__ == "__main__":
optimizer = SlicingOptimizerAgent(
api_key="YOUR_HOLYSHEEP_API_KEY"
)
result = optimizer.optimize_slicing_params(
stl_file_path="/models/gear assembly v3.stl",
material="petg",
printer_type="fdm",
priority="urgent"
)
print(f"최적화 완료: {result['estimated_time_minutes']}분")
print(f"예상 소모량: {result['material_usage_grams']}g")
print(f"노즐 온도: {result['nozzle_temperature']}°C")
5.2 고객 커뮤니케이션 Agent — Claude Sonnet 4.5
저는 고객 커뮤니케이션에서Claude Sonnet 4.5의 장문 처리 능력과 프롬프트 정확도를 특히 중요하게 생각합니다. 3D 프린팅 고객은 설계 변경 요청, 납기 문의, 품질 이슈 등 다양한 커뮤니케이션 채널을 통해 공장과 소통하며, 각각의 상황에 맞는 자연스럽고 전문적인 응답을 생성해야 합니다.
"""
HolySheep AI 게이트웨이 — 고객 커뮤니케이션 Agent
Claude Sonnet 4.5를 활용한 주문 관리 및 고객 안내 시스템
"""
import requests
import json
from datetime import datetime, timedelta
from typing import Dict, List
class CustomerCommunicationAgent:
"""3D 프린팅 고객 커뮤니케이션 자동화 Agent"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.model = "claude-sonnet-4-5"
def generate_quote_response(
self,
customer_name: str,
order_details: Dict,
is_accepted: bool,
custom_message: str = None
) -> str:
"""
주문 확정/거절 응답 메시지 생성
Args:
customer_name: 고객명
order_details: 주문 상세 정보
is_accepted: 견적 수락 여부
custom_message: 커스텀 메시지
Returns:
Claude가 생성한 고객 응답 메시지
"""
headers = {
"x-api-key": self.api_key,
"Content-Type": "application/json",
"anthropic-version": "2023-06-01"
}
if is_accepted:
delivery_date = datetime.now() + timedelta(
days=order_details.get("lead_time_days", 5)
)
system_prompt = """당신은 전문적이고 친절한 3D 프린팅 서비스의 \
고객 관리 담당자입니다. 한국 고객에게 자연스럽고 정중한 \
주문 확인 메시지를 작성합니다. 반드시 아래 정보를 포함하세요:
- 주문 번호
- 인쇄 물품명
- 예상 납기일
- 결제 안내
-后续联系人 정보"""
user_prompt = f"""
고객명: {customer_name}
주문 번호: {order_details['order_id']}
주문 물품: {order_details['item_name']}
수량: {order_details['quantity']}개
재료: {order_details['material']}
예상 납기: {delivery_date.strftime('%Y년 %m월 %d일')}
결제 기한: {(datetime.now() + timedelta(days=2)).strftime('%Y년 %m월 %d일')}
총 금액: {order_details['total_price']:,}원
위 주문의 결제 확인을 요청하는 \
친근하고 전문적인 확인 메시지를 작성해주세요.
{'(추가 메시지: ' + custom_message + ')' if custom_message else ''}"""
else:
system_prompt = """당신은 3D 프린팅 서비스의 \
고객 관리 담당자입니다. 견적 거부 시 \
친절하게 대안을 제시하고 관계를 유지하는 \
응답 메시지를 작성합니다."""
user_prompt = f"""
고객명: {customer_name}
거부 사유: {order_details.get('rejection_reason', '일정 문제')}
원래 요청 물품: {order_details['item_name']}
기존 관계를 유지하면서 대안이나 \
차기 협력 기회를 제시하는 \
전문적인 거절 메시지를 작성해주세요.
{'(추가 메시지: ' + custom_message + ')' if custom_message else ''}"""
payload = {
"model": self.model,
"max_tokens": 1024,
"system": system_prompt,
"messages": [
{"role": "user", "content": user_prompt}
]
}
response = requests.post(
f"{self.base_url}/messages",
headers=headers,
json=payload,
timeout=30
)
if response.status_code != 200:
raise RuntimeError(
f"Claude API 오류: {response.status_code} - {response.text}"
)
result = response.json()
return result["content"][0]["text"]
def generate_progress_notification(
self,
customer_email: str,
order_id: str,
current_stage: str,
completion_percentage: int
) -> Dict:
"""생산 진행 상황 알림 메시지 생성"""
stage_messages = {
" slicing": "슬라이싱 완료",
"printing": "인쇄 중",
"post_processing": "후처리 진행 중",
"quality_check": "품질 검사 중",
"packaging": "포장 준비 중",
"shipped": "배송 출발"
}
headers = {
"x-api-key": self.api_key,
"Content-Type": "application/json",
"anthropic-version": "2023-06-01"
}
payload = {
"model": self.model,
"max_tokens": 512,
"system": "3D 프린팅 주문 진행 상황을 \
고객에게 알리는 친근한 알림 메시지를 작성합니다.",
"messages": [
{
"role": "user",
"content": f"주문 ID: {order_id}\n\
현재 단계: {stage_messages.get(current_stage, current_stage)}\n\
완료율: {completion_percentage}%\n\n\
위 내용을 기반으로 진행 상황을 \
알려주는 간결하고 명확한 \
이메일 알림 메시지를 작성해주세요."
}
]
}
response = requests.post(
f"{self.base_url}/messages",
headers=headers,
json=payload,
timeout=30
)
result = response.json()
return {
"email": customer_email,
"subject": f"[3D프린트 주문 {order_id}] 생산 진행 안내",
"body": result["content"][0]["text"],
"stage": current_stage,
"completion": completion_percentage
}
사용 예시
if __name__ == "__main__":
agent = CustomerCommunicationAgent(
api_key="YOUR_HOLYSHEEP_API_KEY"
)
# 주문 확정 응답 생성
quote_response = agent.generate_quote_response(
customer_name="홍길동",
order_details={
"order_id": "ORD-2026-0524-001",
"item_name": "자동차 드래프트 미러 브래킷",
"quantity": 3,
"material": "PA12 (나일론)",
"lead_time_days": 7,
"total_price": 450000
},
is_accepted=True,
custom_message="빠른 납기 부탁드립니다."
)
print("주문 확인 메시지:")
print(quote_response)
# 생산 진행 알림 생성
notification = agent.generate_progress_notification(
customer_email="[email protected]",
order_id="ORD-2026-0524-001",
current_stage="printing",
completion_percentage=65
)
print(f"\n이메일 발송: {notification['subject']}")
5.3 스케줄링 로직 Agent — DeepSeek V3.2
자원 배분과 충돌 감지 같은 반복적인 로직에는 DeepSeek V3.2가 제격입니다. $0.42/MTok라는 놀라운 비용 효율성 덕분에 저는 수시로 스케줄링 시뮬레이션을 돌려보면서 최적안을 탐색할 수 있었습니다.
"""
HolySheep AI 게이트웨이 — 스케줄링 및 자원 배분 Agent
DeepSeek V3.2를 활용한 프린터 스케줄링 및 충돌 감지
"""
import requests
import json
from typing import Dict, List
from dataclasses import dataclass
@dataclass
class PrintJob:
"""인쇄 작업 정의"""
job_id: str
customer_id: str
item_name: str
printer_id: str
estimated_minutes: int
priority: int # 1= highest, 5= lowest
material: str
deadline_hours: int
@dataclass
class Printer:
"""프린터 상태"""
printer_id: str
printer_type: str
available: bool
current_job_end: int = 0 # 분 단위
class SchedulingAgent:
"""DeepSeek V3.2 기반 스케줄링 최적화 Agent"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.model = "deepseek-v3.2"
def optimize_schedule(
self,
pending_jobs: List[Dict],
available_printers: List[Dict]
) -> Dict:
"""
대기 중인 작업과 가용 프린터 기반 최적 스케줄 생성
Args:
pending_jobs: 대기 작업 목록
available_printers: 가용 프린터 목록
Returns:
최적화된 스케줄 및 충돌 감지 결과
"""
# DeepSeek V3.2에 스케줄링 최적화 요청
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": self.model,
"messages": [
{
"role": "system",
"content": "당신은 제조 공정 스케줄링 최적화 전문가입니다. \
입력된 작업 목록과 프린터 상태를 분석하여 \
-deadline 지키기, 우선순위 순서, 자원 효율성\
을 고려한 최적 스케줄을 JSON 형태로 제시합니다."
},
{
"role": "user",
"content": self._build_schedule_prompt(
pending_jobs, available_printers
)
}
],
"temperature": 0.2
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=20
)
if response.status_code != 200:
raise RuntimeError(
f"DeepSeek API 오류: {response.status_code}"
)
result = response.json()
schedule_text = result["choices"][0]["message"]["content"]
# 충돌 감지 및 검증
try:
# JSON 파싱 시도
schedule = json.loads(schedule_text)
except json.JSONDecodeError:
# JSON이 아닌 경우 텍스트 파싱
schedule = self._parse_text_schedule(schedule_text)
conflicts = self._detect_conflicts(schedule, available_printers)
return {
"schedule": schedule,
"conflicts": conflicts,
"efficiency_score": self._calculate_efficiency(
schedule, pending_jobs
)
}
def _build_schedule_prompt(
self,
jobs: List[Dict],
printers: List[Dict]
) -> str:
jobs_text = "\n".join([
f"- Job {j['job_id']}: {j['item_name']}, "
f"소요시간 {j['estimated_minutes']}분, "
f"우선순위 {j['priority']}, "
f"마감 {j['deadline_hours']}시간 후, "
f"재료 {j['material']}"
for j in jobs
])
printers_text = "\n".join([
f"- Printer {p['printer_id']}: {p['printer_type']}, "
f"가용 여부 {p['available']}"
for p in printers
])
return f"""대기 작업 목록:
{jobs_text}
가용 프린터:
{printers_text}
요구사항:
1. 모든 작업이 deadline 내에 완료되도록 스케줄링
2. 우선순위가 높은 작업 먼저 배정
3. 동일 프린터에서 연속 작업 시 재료 전환 최소화
4. 결과는 아래 JSON 형식으로 반환:
{{
"assignments": [
{{
"job_id": "...",
"printer_id": "...",
"start_time_minutes": ...,
"end_time_minutes": ...,
"meets_deadline": true/false
}}
],
"utilization_rate": 전체 프린터利用率 (%),
"unmet_deadlines": deadline 미달성 작업 목록
}}"""
def _detect_conflicts(
self,
schedule: Dict,
printers: List[Dict]
) -> List[Dict]:
"""스케줄 충돌 감지"""
conflicts = []
# 프린터별 할당량 추적
printer_slots = {p["printer_id"]: [] for p in printers}
for assignment in schedule.get("assignments", []):
printer_id = assignment["printer_id"]
start = assignment["start_time_minutes"]
end = assignment["end_time_minutes"]
# 시간 겹침 확인
for existing_start, existing_end in printer_slots[printer_id]:
if not (end <= existing_start or start >= existing_end):
conflicts.append({
"type": "time_overlap",
"printer_id": printer_id,
"conflicting_jobs": [
assignment["job_id"]
],
"overlap_minutes": min(end, existing_end) - max(start, existing_start)
})
printer_slots[printer_id].append((start, end))
return conflicts
def _calculate_efficiency(
self,
schedule: Dict,
jobs: List[Dict]
) -> float:
"""스케줄 효율성 점수 계산"""
if not schedule.get("assignments"):
return 0.0
total_jobs = len(jobs)
deadline_met = sum(
1 for a in schedule["assignments"]
if a.get("meets_deadline", False)
)
return round((deadline_met / total_jobs) * 100, 2)
def _parse_text_schedule(self, text: str) -> Dict:
"""텍스트 스케줄을 딕셔너리로 파싱"""
return {
"assignments": [],
"utilization_rate": 0,
"unmet_deadlines": []
}
사용 예시
if __name__ == "__main__":
agent = SchedulingAgent(api_key="YOUR_HOLYSHEEP_API_KEY")
pending_jobs = [
{
"job_id": "J001",
"item_name": "프로토타입 하우징",
"estimated_minutes": 180,
"priority": 1,
"deadline_hours": 24,
"material": "abs"
},
{
"job_id": "J002",
"item_name": "데코레이션 피겨",
"estimated_minutes": 90,
"priority": 3,
"deadline_hours": 72,
"material": "pla"
}
]
printers = [
{"printer_id": "FDM-01", "printer_type": "fdm", "available": True},
{"printer_id": "FDM-02", "printer_type": "fdm", "available": True}
]
result = agent.optimize_schedule(pending_jobs, printers)
print(f"효율성 점수: {result['efficiency_score']}%")
print(f"충돌 감지: {len(result['conflicts'])}건")
print(f"할당 작업 수: {len(result['schedule']['assignments'])}")
6. API Quota 모니터링 시스템 — Gemini 2.5 Flash
Enterprise 환경에서 여러 AI API를 동시에 사용하면 비용 관리와 Quota 모니터링이 중요합니다. 저는 Gemini 2.5 Flash의 빠른 응답 속도와 저렴한 가격($2.50/MTok)을 활용하여 실시간 대시보드와 비용 알림 시스템을 구축했습니다.
"""
HolySheep AI 게이트웨이 — API Quota 모니터링 시스템
Gemini 2.5 Flash를 활용한 비용 추적 및 알림
"""
import requests
import json
from datetime import datetime, timedelta
from typing import Dict, List
from collections import defaultdict
class APIMonitorAgent:
"""AI API 사용량 및 Quota 모니터링 Agent"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.model = "gemini-2.5-flash"
# 모델별 가격 (2026년 5월 기준)
self.pricing = {
"gpt-4.1": 8.00,
"claude-sonnet-4-5": 15.00,
"gemini-2.5-flash": 2.50,
"deepseek-v3.2": 0.42
}
# 사용량 추적
self.usage_log = defaultdict(list)
def track_usage(
self,
model: str,
input_tokens: int,
output_tokens: int,
request_type: str
) -> Dict:
"""
API 사용량 기록 및 비용 계산
Args:
model: 사용 모델명
input_tokens: 입력 토큰 수
output_tokens: 출력 토큰 수
request_type: 요청 유형 (slicing, communication, scheduling)
Returns:
사용량 및 비용 정보
"""
cost_per_mtok = self.pricing.get(model, 0)
# 출력 토큰 기준 과금
output_cost = (output_tokens / 1_000_000) * cost_per_mtok
usage_record = {
"timestamp": datetime.now().isoformat(),
"model": model,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"cost_usd": round(output_cost, 4),
"request_type": request_type
}
self.usage_log[model].append(usage_record)
return usage_record
def generate_cost_report(
self,
period_days: int = 7
) -> str:
"""Gemini 2.5 Flash로 비용 보고서 생성"""
# 사용량 집계
summary = self._aggregate_usage(period_days)
# Gemini Flash로 보고서 생성
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": self.model,
"messages": [
{
"role": "system",
"content": "당신은 DevOps 비용 분석 전문가입니다. \
AI API 사용량 데이터를 분석하여 \
비용 최적화 방안을 제시하는 \
명확하고 실행 가능한 보고서를 작성합니다."
},
{
"role": "user",
"content": f"""최근 {period_days}일간 AI API 사용량:
{json.dumps(summary, indent=2, ensure_ascii=False)}
위 데이터를 바탕으로:
1. 총 비용 및 모델별 비용 비중
2. 사용 패턴 분석 (요일별, 시간대별)
3. 비용 최적화 추천 (모델 전환, 캐싱, 배치 처리)
4. 월간 예상 비용
를 포함하여 한국어로 보고서를 작성해주세요."""
}
],
"temperature": 0.3
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=25
)
if response.status_code != 200:
raise RuntimeError(f"Gemini API 오류: {response.status_code}")
result = response.json()
return result["choices"][0]["message"]["content"]
def _aggregate_usage(self, days: int) -> Dict:
"""사용량 집계"""
cutoff = datetime.now() - timedelta(days=days)
total_by_model = defaultdict(lambda: {
"requests": 0,
"input_tokens": 0,
"output_tokens": 0,
"cost_usd": 0
})
for model, records in self.usage_log.items():
for record in records:
record_time = datetime.fromisoformat(record["timestamp"])
if record_time >= cutoff:
total_by_model[model]["requests"] += 1
total_by_model[model]["input_tokens"] += record["input_tokens"]
total_by_model[model]["output_tokens"] += record["output_tokens"]
total_by_model[model]["cost