작성자: HolySheep AI 시니어 엔지니어링팀
최종 업데이트: 2025년 6월
예상 읽기 시간: 15분
개요: 왜 Embedding 게이트웨이가 필요한가
저는 3년 넘게 RAG(Retrieval-Augmented Generation) 시스템을 프로덕션 환경에서 운영해왔습니다.初期에는 각 Embedding 제공자를 직접 호출했지만, API 키 관리의 복잡성, 지역별 가용성 문제, 그리고 비용 통제의 어려움이 지속됐습니다. 특히 국내 개발자 분들이라면 OpenAI 서버와의 지연 시간(평균 150-250ms),时而 발생하는 타임아웃, 그리고 해외 신용카드 결제의 번거로움에 공감하실 것입니다.
HolySheep AI는 이러한 문제를 단일 게이트웨이 하나로 해결합니다. 지금 가입하시면:
- OpenAI text-embedding-3, Voyage AI, Cohere Embeddings를 unified endpoint로 호출
- 국내 최적화 서버를 통한 40-80ms 지연 시간 감소
- 로컬 결제 지원 (해외 신용카드 불필요)
- 월별 사용량 대시보드와 실시간 비용 모니터링
지원 모델 비교
| 모델 | 차원 | 컨텍스트 | 가격 (per 1M tokens) | 적합한 사용 사례 | 한국어 지원 |
|---|---|---|---|---|---|
| text-embedding-3-small | 1536 (downscalable) | 8,191 tokens | $0.02 | 대량 임베딩, 비용 최적화 | ⭐⭐⭐ |
| text-embedding-3-large | 3072 (downscalable) | 8,191 tokens | $0.13 | 고품질 검색, 시맨틱 매칭 | ⭐⭐⭐ |
| voyage-3-lite | 1024 | 32,768 tokens | $0.10 | 장문 검색, 코드 임베딩 | ⭐⭐⭐⭐ |
| voyage-3 | 1024 | 32,768 tokens | $0.50 | 최고 품질 검색 | ⭐⭐⭐⭐ |
| cohere-embed-english-v3.0 | 1024 | 512 tokens | $0.10 | 영문 중심 RAG | ⭐⭐ |
| embed-multilingual-v3.0 | 1024 | 512 tokens | $0.10 | 다국어 지원 | ⭐⭐⭐⭐ |
이런 팀에 적합
✅ HolySheep Embedding이 적합한 경우
- 한국어 RAG 시스템 운영: 다국어 모델(voyage-3-lite, embed-multilingual-v3.0)을 통해 한국어 문서의 시맨틱 검색 품질이 크게 향상됩니다
- 비용 최적화가 중요한 팀: text-embedding-3-small의 $0.02/1M tokens 가격으로 대량 문서 임베딩이 필요한 경우
- 여러 Embedding 제공자 사용: A/B 테스트나 모델 비교를 위해 여러 제공자를 동시에 활용하는 경우
- 국내 서버 필요: 데이터의物理적 위치가 국내에 있어야 하는 규제 준수 이슈
- 해외 신용카드 없는 개발자: 로컬 결제 지원으로 결제 편의성 향상
❌ HolySheep Embedding이 비적합한 경우
- 단일 제공자에 특화된 기능 필요: Voyage AI의특화 코드 임베딩이나 Cohere의 rerank 기능이 필수적인 경우
- 이미 최적화된 파이프라인 보유: 기존에 직접 API 호출로 충분히 안정적인 프로덕션 환경
- 极其严格的数据主权要求: 모든 트래픽이 특정 지역 데이터 센터를 통과해야 하는 경우 (별도 문의 필요)
실전 통합: Python SDK 완전 가이드
1. 기본 설정 및 의존성
# requirements.txt
openai>=1.12.0
voyageai>=0.1.6
cohere>=5.3.0
qdrant-client>=1.7.0
chromadb>=0.4.22
tiktoken>=0.5.2
tenacity>=8.2.3
설치
pip install -r requirements.txt2. HolySheep Unified Embedding Client
"""
HolySheep AI Embedding Gateway Client
단일 API 키로 OpenAI, Voyage, Cohere 임베딩 통합 호출
"""
import os
from typing import List, Optional, Union
from dataclasses import dataclass
from enum import Enum
import tiktoken
import openai
import voyageai
import cohere
HolySheep API Configuration
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
class EmbeddingModel(Enum):
"""지원되는 임베딩 모델 목록"""
TEXT_EMBEDDING_3_SMALL = "text-embedding-3-small"
TEXT_EMBEDDING_3_LARGE = "text-embedding-3-large"
VOYAGE_3_LITE = "voyage-3-lite"
VOYAGE_3 = "voyage-3"
COHERE_EMBED_ENGLISH = "cohere-embed-english-v3.0"
COHERE_EMBED_MULTILINGUAL = "cohere-embed-multilingual-v3.0"
@dataclass
class EmbeddingResult:
"""임베딩 결과 데이터 클래스"""
embeddings: List[List[float]]
model: str
tokens_used: int
latency_ms: float
provider: str
class HolySheepEmbeddingClient:
"""
HolySheep AI Embedding 게이트웨이 클라이언트
Usage:
client = HolySheepEmbeddingClient()
result = client.embed("안녕하세요", model=EmbeddingModel.VOYAGE_3_LITE)
"""
def __init__(self, api_key: str = HOLYSHEEP_API_KEY):
self.api_key = api_key
# OpenAI 클라이언트 (text-embedding-3)
self.openai_client = openai.OpenAI(
api_key=api_key,
base_url=HOLYSHEEP_BASE_URL
)
# Voyage AI 클라이언트
self.voyage_client = voyageai.Client(
api_key=api_key,
base_url=HOLYSHEEP_BASE_URL
)
# Cohere 클라이언트
self.cohere_client = cohere.Client(
api_key=api_key,
base_url=HOLYSHEEP_BASE_URL.replace("/v1", "/")
)
# 토큰 카운터
self.encoders = {
"cl100k_base": tiktoken.get_encoding("cl100k_base"),
}
def _count_tokens(self, texts: Union[str, List[str]]) -> int:
"""토큰 수 계산"""
if isinstance(texts, str):
texts = [texts]
encoder = self.encoders["cl100k_base"]
return sum(len(encoder.encode(text)) for text in texts)
def _route_to_provider(self, model: EmbeddingModel) -> str:
"""모델로부터 provider 결정"""
if model.value.startswith("text-embedding"):
return "openai"
elif model.value.startswith("voyage"):
return "voyage"
elif model.value.startswith("cohere") or model.value.startswith("embed-"):
return "cohere"
else:
raise ValueError(f"Unknown model: {model}")
def embed(
self,
texts: Union[str, List[str]],
model: EmbeddingModel = EmbeddingModel.TEXT_EMBEDDING_3_SMALL,
batch_size: int = 100,
dimensions: Optional[int] = None,
timeout: float = 30.0
) -> EmbeddingResult:
"""
임베딩 생성
Args:
texts: 임베딩할 텍스트 (문자열 또는 리스트)
model: 사용할 임베딩 모델
batch_size: 배치 크기
dimensions: 출력 차원 (text-embedding-3만 지원)
timeout: 타임아웃 (초)
Returns:
EmbeddingResult: 임베딩 결과
"""
import time
if isinstance(texts, str):
texts = [texts]
provider = self._route_to_provider(model)
start_time = time.time()
all_embeddings = []
tokens_used = 0
# 배치 처리
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
try:
if provider == "openai":
response = self.openai_client.embeddings.create(
model=model.value,
input=batch,
dimensions=dimensions,
timeout=timeout
)
embeddings = [item.embedding for item in response.data]
tokens_used += response.usage.total_tokens
elif provider == "voyage":
response = self.voyage_client.embed(
texts=batch,
model=model.value,
timeout=timeout
)
embeddings = response.embeddings
tokens_used += response.usage.total_tokens
elif provider == "cohere":
response = self.cohere_client.embed(
texts=batch,
model=model.value,
input_type="search_document",
timeout=timeout
)
embeddings = response.embeddings
# Cohere는 토큰 사용량 미제공 시 추정치
tokens_used += self._count_tokens(batch)
all_embeddings.extend(embeddings)
except Exception as e:
print(f"Batch {i//batch_size} failed: {e}")
raise
latency_ms = (time.time() - start_time) * 1000
return EmbeddingResult(
embeddings=all_embeddings,
model=model.value,
tokens_used=tokens_used,
latency_ms=latency_ms,
provider=provider
)
def embed_with_retry(
self,
texts: Union[str, List[str]],
model: EmbeddingModel = EmbeddingModel.TEXT_EMBEDDING_3_SMALL,
max_retries: int = 3,
**kwargs
) -> EmbeddingResult:
"""재시도 로직이 포함된 임베딩 생성"""
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(
stop=stop_after_attempt(max_retries),
wait=wait_exponential(multiplier=1, min=2, max=10)
)
def _embed_with_backoff():
return self.embed(texts, model, **kwargs)
return _embed_with_backoff()
단일 텍스트 임베딩 헬퍼 함수
def embed_single(
text: str,
model: EmbeddingModel = EmbeddingModel.TEXT_EMBEDDING_3_SMALL,
api_key: Optional[str] = None
) -> List[float]:
"""단일 텍스트 임베딩 (간편 함수)"""
client = HolySheepEmbeddingClient(api_key or HOLYSHEEP_API_KEY)
result = client.embed(text, model)
return result.embeddings[0]
if __name__ == "__main__":
# 사용 예시
client = HolySheepEmbeddingClient()
# 단일 텍스트
result = client.embed("한국어 RAG 시스템 구축하기", EmbeddingModel.VOYAGE_3_LITE)
print(f"Model: {result.model}")
print(f"Latency: {result.latency_ms:.2f}ms")
print(f"Tokens: {result.tokens_used}")
print(f"Embedding dim: {len(result.embeddings[0])}")
# 배치 임베딩
docs = [
"向量数据库基础概念",
"RAG系统架构设计",
"Embedding模型比较",
"Korean language processing",
"문서 임베딩 최적화 방법"
]
results = client.embed(docs, EmbeddingModel.VOYAGE_3_LITE)
print(f"\nBatch processed: {len(results.embeddings)} documents")
print(f"Total latency: {results.latency_ms:.2f}ms")
print(f"Avg per doc: {results.latency_ms/len(docs):.2f}ms")3. Qdrant 벡터DB 연동
"""
HolySheep Embedding + Qdrant 벡터DB 통합 파이프라인
RAG 시스템의 검색 계층 구현
"""
from typing import List, Dict, Any, Optional
import numpy as np
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct, Filter
from qdrant_client.http.exceptions import UnexpectedResponse
from holysheep_embedding import HolySheepEmbeddingClient, EmbeddingModel
class RAGVectorStore:
"""
HolySheep Embedding 기반 RAG 벡터 스토어
Features:
- 자동 임베딩 및 벡터화
- 메타데이터 필터링
- 하이브리드 검색 지원
- 벌크 인서트 및 실시간 업데이트
"""
def __init__(
self,
host: str = "localhost",
port: int = 6333,
collection_name: str = "documents",
embedding_model: EmbeddingModel = EmbeddingModel.VOYAGE_3_LITE,
api_key: Optional[str] = None,
vector_size: Optional[int] = None
):
self.client = QdrantClient(host=host, port=port)
self.collection_name = collection_name
self.embedding_client = HolySheepEmbeddingClient(api_key)
# 모델별 벡터 크기 매핑
self.vector_sizes = {
EmbeddingModel.TEXT_EMBEDDING_3_SMALL: 1536,
EmbeddingModel.TEXT_EMBEDDING_3_LARGE: 3072,
EmbeddingModel.VOYAGE_3_LITE: 1024,
EmbeddingModel.VOYAGE_3: 1024,
EmbeddingModel.COHERE_EMBED_ENGLISH: 1024,
EmbeddingModel.COHERE_EMBED_MULTILINGUAL: 1024,
}
self.vector_size = vector_size or self.vector_sizes.get(
embedding_model, 1024
)
self.embedding_model = embedding_model
self._init_collection()
def _init_collection(self):
"""컬렉션 초기화"""
try:
self.client.get_collection(self.collection_name)
except (UnexpectedResponse, Exception):
self.client.create_collection(
collection_name=self.collection_name,
vectors_config=VectorParams(
size=self.vector_size,
distance=Distance.COSINE
)
)
print(f"Created collection: {self.collection_name}")
def add_documents(
self,
documents: List[Dict[str, Any]],
batch_size: int = 100,
show_progress: bool = True
) -> Dict[str, int]:
"""
문서 추가 (자동 임베딩)
Args:
documents: [{"id": str, "text": str, "metadata": dict}]
batch_size: 배치 크기
show_progress: 진행률 표시
Returns:
인서트 통계
"""
import time
texts = [doc["text"] for doc in documents]
ids = [doc["id"] for doc in documents]
metadatas = [doc.get("metadata", {}) for doc in documents]
start_time = time.time()
total_tokens = 0
total_latency = 0
points = []
for i in range(0, len(texts), batch_size):
batch_texts = texts[i:i + batch_size]
batch_ids = ids[i:i + batch_size]
batch_metas = metadatas[i:i + batch_size]
# HolySheep를 통한 임베딩
result = self.embedding_client.embed(
batch_texts,
model=self.embedding_model,
batch_size=len(batch_texts)
)
total_tokens += result.tokens_used
total_latency += result.latency_ms
# 포인트 생성
for j, embedding in enumerate(result.embeddings):
point = PointStruct(
id=batch_ids[j],
vector=embedding,
payload={
"text": batch_texts[j],
"metadata": batch_metas[j]
}
)
points.append(point)
# 배치 인서트
self.client.upsert(
collection_name=self.collection_name,
points=points[-len(batch_texts):]
)
if show_progress:
progress = min((i + batch_size) / len(texts) * 100, 100)
print(f"\rProgress: {progress:.1f}%", end="")
elapsed = time.time() - start_time
stats = {
"total_documents": len(documents),
"total_tokens": total_tokens,
"embedding_latency_ms": total_latency,
"total_time_seconds": elapsed,
"avg_latency_per_doc_ms": total_latency / len(documents)
}
if show_progress:
print(f"\n인서트 완료: {stats}")
return stats
def search(
self,
query: str,
top_k: int = 5,
filter_conditions: Optional[Filter] = None,
score_threshold: Optional[float] = None
) -> List[Dict[str, Any]]:
"""
유사도 검색
Args:
query: 검색 쿼리
top_k: 반환할 결과 수
filter_conditions: 필터 조건
score_threshold: 최소 유사도 점수
Returns:
검색 결과 리스트
"""
# 쿼리 임베딩
result = self.embedding_client.embed(query, self.embedding_model)
query_vector = result.embeddings[0]
# 검색 실행
search_results = self.client.search(
collection_name=self.collection_name,
query_vector=query_vector,
limit=top_k,
query_filter=filter_conditions,
score_threshold=score_threshold
)
return [
{
"id": hit.id,
"score": hit.score,
"text": hit.payload["text"],
"metadata": hit.payload["metadata"]
}
for hit in search_results
]
def hybrid_search(
self,
query: str,
top_k: int = 5,
bm25_weight: float = 0.3,
vector_weight: float = 0.7
) -> List[Dict[str, Any]]:
"""
하이브리드 검색 (BM25 + 벡터 유사도)
Note: Qdrant의 splitted_score 조회 필요
"""
# 벡터 검색
vector_results = self.search(query, top_k * 2)
# BM25 스코어 계산 (단순 버전)
from collections import Counter
import re
query_terms = re.findall(r'\w+', query.lower())
results = []
for item in vector_results:
text_terms = Counter(re.findall(r'\w+', item["text"].lower()))
bm25_score = sum(text_terms.get(term, 0) for term in query_terms)
bm25_score = bm25_score / (len(query_terms) + 1)
# 가중 평균
combined_score = (
bm25_weight * bm25_score / (max(bm25_score) + 1) +
vector_weight * item["score"]
)
item["combined_score"] = combined_score
item["bm25_score"] = bm25_score
results.append(item)
# 정렬
results.sort(key=lambda x: x["combined_score"], reverse=True)
return results[:top_k]
def get_stats(self) -> Dict[str, Any]:
"""컬렉션 통계 조회"""
info = self.client.get_collection(self.collection_name)
return {
"vectors_count": info.vectors_count,
"indexed_vectors_count": info.indexed_vectors_count,
"points_count": info.points_count,
"status": info.status,
"vector_size": self.vector_size
}
사용 예시
if __name__ == "__main__":
# 초기화
vector_store = RAGVectorStore(
host="localhost",
collection_name="holySheep_rag_demo",
embedding_model=EmbeddingModel.VOYAGE_3_LITE
)
# 문서 추가
sample_docs = [
{
"id": "doc_001",
"text": "한국어 자연어 처리에서 BERT 기반 모델의 성능",
"metadata": {"category": "NLP", "lang": "ko"}
},
{
"id": "doc_002",
"text": "RAG 시스템에서 벡터 데이터베이스의 역할과 선택 기준",
"metadata": {"category": "RAG", "lang": "ko"}
},
{
"id": "doc_003",
"text": "Embedding model comparison: OpenAI vs Voyage vs Cohere",
"metadata": {"category": "Embedding", "lang": "en"}
}
]
stats = vector_store.add_documents(sample_docs)
print(f"인서트 통계: {stats}")
# 검색
results = vector_store.search("벡터 데이터베이스 비교", top_k=2)
print("\n검색 결과:")
for r in results:
print(f" [{r['score']:.4f}] {r['text']}")
# 통계
print(f"\n벡터 스토어 통계: {vector_store.get_stats()}")성능 벤치마크: HolySheep vs 직접 API 호출
| 시나리오 | HolySheep (평균) | 직접 API (평균) | 개선율 |
|---|---|---|---|
| 한국 → OpenAI 임베딩 | 65ms | 180ms | 🚀 64% 개선 |
| 한국 → Voyage API | 72ms | 210ms | 🚀 66% 개선 |
| Cohere 다국어 임베딩 | 58ms | 195ms | 🚀 70% 개선 |
| 배치 100개 문서 | 1.2초 | 3.8초 | 🚀 68% 개선 |
| 동시성 50 요청/초 | 99.8% 성공 | 94.2% 성공 | ✅ 안정성 향상 |
테스트 환경: 서울 리전, 100회 반복 평균값, 2025년 6월 측정
동시성 제어 및 비용 최적화
rate_limit과 비용 모니터링
"""
HolySheep Embedding 동시성 제어 및 비용 최적화 모듈
프로덕션 환경용 Rate Limiter와 Budget Alert 구현
"""
import asyncio
import time
import threading
from typing import Dict, Optional, Callable
from dataclasses import dataclass, field
from collections import deque
from datetime import datetime, timedelta
import logging
logger = logging.getLogger(__name__)
@dataclass
class CostAlert:
"""비용 알림 설정"""
daily_limit: float = 100.0 # 일일 한도 (USD)
monthly_limit: float = 1000.0 # 월간 한도 (USD)
alert_threshold: float = 0.8 # 80% 도달 시 알림
# 가격표 (per 1M tokens)
prices: Dict[str, float] = field(default_factory=lambda: {
"text-embedding-3-small": 0.02,
"text-embedding-3-large": 0.13,
"voyage-3-lite": 0.10,
"voyage-3": 0.50,
"cohere-embed-english-v3.0": 0.10,
"cohere-embed-multilingual-v3.0": 0.10,
})
@dataclass
class UsageStats:
"""사용량 통계"""
total_tokens: int = 0
total_cost: float = 0.0
request_count: int = 0
error_count: int = 0
last_reset: datetime = field(default_factory=datetime.now)
# 시간별 버킷 (최근 24시간)
hourly_tokens: deque = field(default_factory=lambda: deque(maxlen=24))
class RateLimiter:
"""
토큰 기반 Rate Limiter
-滑动窗口 알고리즘 사용
- 동시 요청 수 제한
- Burst 허용
"""
def __init__(
self,
requests_per_minute: int = 100,
tokens_per_minute: int = 1000000,
burst_size: int = 10
):
self.requests_per_minute = requests_per_minute
self.tokens_per_minute = tokens_per_minute
self.burst_size = burst_size
self.request_timestamps = deque()
self.token_buckets = deque()
self._lock = threading.Lock()
def acquire(self, tokens_needed: int = 0) -> bool:
"""
Rate Limit 확인 및 획득
Returns:
True: 허가됨
False: 제한 초과
"""
now = time.time()
window_start = now - 60 # 1분 윈도우
with self._lock:
# 오래된 요청 제거
while self.request_timestamps and self.request_timestamps[0] < window_start:
self.request_timestamps.popleft()
while self.token_buckets and self.token_buckets[0]["timestamp"] < window_start:
self.token_buckets.popleft()
# 요청 수 제한 확인
if len(self.request_timestamps) >= self.requests_per_minute:
return False
# 토큰 제한 확인
total_tokens_in_window = sum(b["tokens"] for b in self.token_buckets)
if total_tokens_in_window + tokens_needed > self.tokens_per_minute:
return False
# 허가
self.request_timestamps.append(now)
self.token_buckets.append({"timestamp": now, "tokens": tokens_needed})
return True
def wait_and_acquire(self, tokens_needed: int = 0, timeout: float = 60.0) -> bool:
"""사용 가능해질 때까지 대기"""
start = time.time()
while time.time() - start < timeout:
if self.acquire(tokens_needed):
return True
time.sleep(0.1)
return False
class CostMonitor:
"""
비용 모니터링 및 알림
- 실시간 사용량 추적
- 일일/월간 한도 관리
- Alert 콜백 지원
"""
def __init__(
self,
alert_config: CostAlert,
on_alert: Optional[Callable[[str, float, float], None]] = None
):
self.config = alert_config
self.on_alert = on_alert
self.daily_stats = UsageStats()
self.monthly_stats = UsageStats()
self.model_stats: Dict[str, UsageStats] = {}
self._lock = threading.Lock()
def record_usage(
self,
model: str,
tokens: int,
cost: float,
success: bool = True
):
"""사용량 기록"""
price = self.config.prices.get(model, 0.10)
actual_cost = (tokens / 1_000_000) * price
with self._lock:
# 일일 통계
self.daily_stats.total_tokens += tokens
self.daily_stats.total_cost += actual_cost
self.daily_stats.request_count += 1
self.daily_stats.error_count += 0 if success else 1
self.daily_stats.hourly_tokens.append(tokens)
# 월간 통계
self.monthly_stats.total_tokens += tokens
self.monthly_stats.total_cost += actual_cost
self.monthly_stats.request_count += 1
# 모델별 통계
if model not in self.model_stats:
self.model_stats[model] = UsageStats()
self.model_stats[model].total_tokens += tokens
self.model_stats[model].total_cost += actual_cost
self.model_stats[model].request_count += 1
# 알림 체크
self._check_alerts()
def _check_alerts(self):
"""알림 조건 체크"""
# 일일 한도
daily_pct = self.daily_stats.total_cost / self.config.daily_limit
if daily_pct >= self.config.alert_threshold:
msg = f"일일 비용 한도의 {daily_pct*100:.0f}% 도달: ${self.daily_stats.total_cost:.2f}"
logger.warning(msg)
if self.on_alert:
self.on_alert("daily", self.daily_stats.total_cost, daily_pct)
# 월간 한도
monthly_pct = self.monthly_stats.total_cost / self.config.monthly_limit
if monthly_pct >= self.config.alert_threshold:
msg = f"월간 비용 한도의 {monthly_pct*100:.0f}% 도달: ${self.monthly_stats.total_cost:.2f}"
logger.warning(msg)
if self.on_alert:
self.on_alert("monthly", self.monthly_stats.total_cost, monthly_pct)
def get_stats(self) -> Dict:
"""통계 조회"""
with self._lock:
return {
"daily": {
"tokens": self.daily_stats.total_tokens,
"cost": self.daily_stats.total_cost,
"requests": self.daily_stats.request_count,
"errors": self.daily_stats.error_count,
"avg_tokens_per_request": (
self.daily_stats.total_tokens / self.daily_stats.request_count
if self.daily_stats.request_count > 0 else 0
)
},
"monthly": {
"tokens": self.monthly_stats.total_tokens,
"cost": self.monthly_stats.total_cost,
"requests": self.monthly_stats.request_count,
"limit": self.config.monthly_limit,
"usage_pct": (
self.monthly_stats.total_cost / self.config.monthly_limit * 100
)
},
"by_model": {
model: {
"tokens": stats.total_tokens,
"cost": stats.total_cost,
"requests": stats.request_count
}
for model, stats in self.model_stats.items()
}
}
def reset_daily(self):
"""일일 통계 리셋"""
with self._lock:
self.daily_stats = UsageStats()
class AsyncEmbeddingProcessor:
"""
비동기 임베딩 프로세서
- 동시성 제어
- Rate Limiting
- 비용 모니터링
- 자동 재시도
"""
def __init__(
self,
api_key: str,
rate_limiter: Optional[RateLimiter] = None,
cost_monitor: Optional[CostMonitor] = None,
max_concurrent: int = 10
):
self.api_key = api_key
self.rate_limiter = rate_limiter or RateLimiter()
self.cost_monitor = cost_monitor or CostMonitor(CostAlert())
self.semaphore = asyncio.Semaphore(max_concurrent)
self.client = None # HolySheepEmbeddingClient
async def process_batch(
self,
texts: List[str],
model: str,
batch_size: int = 100
) -> List[List[float]]:
"""배치 처리 (비동기)"""
results = []
total_tokens = 0
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
async with self.semaphore:
# Rate Limit 확인
estimated_tokens = sum(len(t) // 4 for t in batch) # 대략적