출장 자동화 시스템
Agent-s engine.py
myeongjaechoi
2025. 3. 30. 15:16
기본 모듈 임포트
import os # 운영 체제 인터페이스를 제공하는 모듈 임포트
import backoff # API 호출 실패 시 재시도를 관리하는 라이브러리 임포트
import numpy as np # 수치 계산을 위한 NumPy 라이브러리 임포트
from anthropic import Anthropic # Anthropic AI API 클라이언트 임포트
from openai import APIConnectionError, APIError, AzureOpenAI, OpenAI, RateLimitError # OpenAI 관련 클래스와 에러 타입 임포트기본 LMM 엔진 클래스
class LMMEngine:
# LLM 엔진의 기본 클래스 - 모든 LLM 엔진 클래스의 부모 클래스
passOpenAI API 클래스 -> GPT 모델을 사용하여 텍스트 생성 ->이전 대화 내용을 바탕으로 다음 대화를 생성
- 예 : "안녕하세요?" -> 적절한 답변 생성
class LMMEngineOpenAI(LMMEngine):
# OpenAI API를 사용하는 LLM 엔진 클래스
def __init__(self, api_key=None, model=None, rate_limit=-1, **kwargs):
"""
LMMEngineOpenAI 클래스의 초기화 메서드
매개변수:
api_key (str, 선택적): OpenAI API 키. 없으면 환경 변수에서 가져옴
model (str, 필수): 사용할 OpenAI 모델명
rate_limit (int, 선택적): 분당 API 요청 제한 수. -1이면 제한 없음
**kwargs: 추가 매개변수
"""
assert model is not None, "model must be provided" # 모델명이 제공되지 않으면 오류 발생
self.model = model # 모델명 저장
api_key = api_key or os.getenv("OPENAI_API_KEY") # API 키가 제공되지 않으면 환경 변수에서 가져옴
if api_key is None: # API 키가 여전히 없으면 오류 발생
raise ValueError(
"An API Key needs to be provided in either the api_key parameter or as an environment variable named OPENAI_API_KEY"
)
self.api_key = api_key # API 키 저장
self.request_interval = 0 if rate_limit == -1 else 60.0 / rate_limit # API 요청 간격 계산 (초 단위)
self.llm_client = OpenAI(api_key=self.api_key) # OpenAI 클라이언트 초기화
@backoff.on_exception(
backoff.expo, (APIConnectionError, APIError, RateLimitError), max_time=60
) # API 오류 발생 시 지수 백오프로 60초까지 재시도
def generate(self, messages, temperature=0.0, max_new_tokens=None, **kwargs):
"""
이전 메시지를 기반으로 다음 메시지 생성
매개변수:
messages (list): 대화 메시지 목록
temperature (float, 선택적): 생성 결과의 무작위성 정도. 기본값 0.0 (결정적)
max_new_tokens (int, 선택적): 생성할 최대 토큰 수. 기본값 4096
**kwargs: 추가 매개변수
반환값:
str: 생성된 응답 텍스트
"""
return (
self.llm_client.chat.completions.create(
model=self.model, # 사용할 모델 지정
messages=messages, # 대화 메시지 목록 전달
max_tokens=max_new_tokens if max_new_tokens else 4096, # 최대 토큰 수 설정
temperature=temperature, # 온도(무작위성) 설정
**kwargs, # 추가 매개변수 전달
)
.choices[0] # 첫 번째 생성 결과 선택
.message.content # 생성된 메시지 내용 반환
)Anthropic API 클래스 -> Claude 모델을 사용하여 텍스트 생성 -> 답변을 생성
class LMMEngineAnthropic(LMMEngine):
# Anthropic API를 사용하는 LLM 엔진 클래스
def __init__(self, api_key=None, model=None, **kwargs):
"""
LMMEngineAnthropic 클래스의 초기화 메서드
매개변수:
api_key (str, 선택적): Anthropic API 키. 없으면 환경 변수에서 가져옴
model (str, 필수): 사용할 Anthropic 모델명
**kwargs: 추가 매개변수
"""
assert model is not None, "model must be provided" # 모델명이 제공되지 않으면 오류 발생
self.model = model # 모델명 저장
api_key = api_key or os.getenv("ANTHROPIC_API_KEY") # API 키가 제공되지 않으면 환경 변수에서 가져옴
if api_key is None: # API 키가 여전히 없으면 오류 발생
raise ValueError(
"An API Key needs to be provided in either the api_key parameter or as an environment variable named ANTHROPIC_API_KEY"
)
self.api_key = api_key # API 키 저장
self.llm_client = Anthropic(api_key=self.api_key) # Anthropic 클라이언트 초기화
@backoff.on_exception(
backoff.expo, (APIConnectionError, APIError, RateLimitError), max_time=60
) # API 오류 발생 시 지수 백오프로 60초까지 재시도
def generate(self, messages, temperature=0.0, max_new_tokens=None, **kwargs):
"""
이전 메시지를 기반으로 다음 메시지 생성
매개변수:
messages (list): 대화 메시지 목록 (첫 메시지는 시스템 프롬프트로 사용)
temperature (float, 선택적): 생성 결과의 무작위성 정도. 기본값 0.0 (결정적)
max_new_tokens (int, 선택적): 생성할 최대 토큰 수. 기본값 4096
**kwargs: 추가 매개변수
반환값:
str: 생성된 응답 텍스트
"""
return (
self.llm_client.messages.create(
system=messages[0]["content"][0]["text"], # 첫 번째 메시지를 시스템 프롬프트로 설정
model=self.model, # 사용할 모델 지정
messages=messages[1:], # 시스템 프롬프트 이후의 메시지들 전달
max_tokens=max_new_tokens if max_new_tokens else 4096, # 최대 토큰 수 설정
temperature=temperature, # 온도(무작위성) 설정
**kwargs, # 추가 매개변수 전달
)
.content[0] # 첫 번째 콘텐츠 항목 선택
.text # 생성된 텍스트 반환
)
@backoff.on_exception(
backoff.expo, (APIConnectionError, APIError, RateLimitError), max_time=60
) # API 오류 발생 시 지수 백오프로 60초까지 재시도
# Claude-3.7 Sonnet 사고 모드와 호환되는 메서드
def generate_with_thinking(
self, messages, temperature=0.0, max_new_tokens=None, **kwargs
):
"""
사고 과정이 포함된 응답 생성 (Claude-3.7 Sonnet 사고 모드 지원)
매개변수:
messages (list): 대화 메시지 목록 (첫 메시지는 시스템 프롬프트로 사용)
temperature (float, 선택적): 생성 결과의 무작위성 정도. 기본값 0.0 (결정적)
max_new_tokens (int, 선택적): 생성할 최대 토큰 수
**kwargs: 추가 매개변수
반환값:
str: 생성된 최종 응답 텍스트
"""
full_response = self.llm_client.messages.create(
system=messages[0]["content"][0]["text"], # 첫 번째 메시지를 시스템 프롬프트로 설정
model=self.model, # 사용할 모델 지정
messages=messages[1:], # 시스템 프롬프트 이후의 메시지들 전달
max_tokens=8192, # 최대 토큰 수 설정 (8192 고정)
thinking={"type": "enabled", "budget_tokens": 4096}, # 사고 모드 활성화 및 토큰 예산 설정
**kwargs, # 추가 매개변수 전달
)
thoughts = full_response.content[0].thinking # 사고 과정 추출
print("CLAUDE 3.7 THOUGHTS:", thoughts) # 사고 과정 출력
final_response = full_response.content[1].text # 최종 응답 추출
return final_response # 최종 응답 반환OpenAI 임베딩 엔진 클래스 -> 텍스트를 숫자로 된 벡터로 변환
class OpenAIEmbeddingEngine(LMMEngine):
# OpenAI 임베딩 API를 사용하는 엔진 클래스
def __init__(
self,
api_key=None,
rate_limit: int = -1,
display_cost: bool = True,
):
"""
OpenAI 임베딩 엔진 초기화
매개변수:
api_key (str, 선택적): OpenAI API 키. 없으면 환경 변수에서 가져옴
rate_limit (int, 선택적): 분당 API 요청 제한 수. -1이면 제한 없음. 기본값 -1
display_cost (bool, 선택적): API 호출 비용 표시 여부. 기본값 True
"""
self.model = "text-embedding-3-small" # 사용할 임베딩 모델 지정
self.cost_per_thousand_tokens = 0.00002 # 1000 토큰당 비용 설정
api_key = api_key or os.getenv("OPENAI_API_KEY") # API 키가 제공되지 않으면 환경 변수에서 가져옴
if api_key is None: # API 키가 여전히 없으면 오류 발생
raise ValueError(
"An API Key needs to be provided in either the api_key parameter or as an environment variable named OPENAI_API_KEY"
)
self.api_key = api_key # API 키 저장
self.display_cost = display_cost # 비용 표시 여부 저장
self.request_interval = 0 if rate_limit == -1 else 60.0 / rate_limit # API 요청 간격 계산 (초 단위)
@backoff.on_exception(
backoff.expo,
(
APIError,
RateLimitError,
APIConnectionError,
),
) # API 오류 발생 시 지수 백오프로 재시도
def get_embeddings(self, text: str) -> np.ndarray:
"""
텍스트의 임베딩 벡터 가져오기
매개변수:
text (str): 임베딩할 텍스트
반환값:
np.ndarray: 임베딩 벡터 배열
"""
client = OpenAI(api_key=self.api_key) # OpenAI 클라이언트 초기화
response = client.embeddings.create(model=self.model, input=text) # 임베딩 생성 요청
if self.display_cost: # 비용 표시 옵션이 활성화된 경우
total_tokens = response.usage.total_tokens # 사용된 총 토큰 수 추출
cost = self.cost_per_thousand_tokens * total_tokens / 1000 # 비용 계산
# print(f"Total cost for this embedding API call: {cost}") # 비용 출력 (주석 처리됨)
return np.array([data.embedding for data in response.data]) # 임베딩 벡터 배열 반환Azure OpenAI API 클래스 -> 비용 추적 기능
class LMMEngineAzureOpenAI(LMMEngine):
# Azure OpenAI API를 사용하는 LLM 엔진 클래스
def __init__(
self,
api_key=None,
azure_endpoint=None,
model=None,
api_version=None,
rate_limit=-1,
**kwargs
):
"""
LMMEngineAzureOpenAI 클래스의 초기화 메서드
매개변수:
api_key (str, 선택적): Azure OpenAI API 키. 없으면 환경 변수에서 가져옴
azure_endpoint (str, 선택적): Azure OpenAI 엔드포인트. 없으면 환경 변수에서 가져옴
model (str, 필수): 사용할 Azure OpenAI 모델명
api_version (str, 필수): 사용할 Azure OpenAI API 버전
rate_limit (int, 선택적): 분당 API 요청 제한 수. -1이면 제한 없음. 기본값 -1
**kwargs: 추가 매개변수
"""
assert model is not None, "model must be provided" # 모델명이 제공되지 않으면 오류 발생
self.model = model # 모델명 저장
assert api_version is not None, "api_version must be provided" # API 버전이 제공되지 않으면 오류 발생
self.api_version = api_version # API 버전 저장
api_key = api_key or os.getenv("AZURE_OPENAI_API_KEY") # API 키가 제공되지 않으면 환경 변수에서 가져옴
if api_key is None: # API 키가 여전히 없으면 오류 발생
raise ValueError(
"An API Key needs to be provided in either the api_key parameter or as an environment variable named AZURE_OPENAI_API_KEY"
)
self.api_key = api_key # API 키 저장
azure_endpoint = azure_endpoint or os.getenv("AZURE_OPENAI_API_BASE") # Azure 엔드포인트가 제공되지 않으면 환경 변수에서 가져옴
if azure_endpoint is None: # Azure 엔드포인트가 여전히 없으면 오류 발생
raise ValueError(
"An Azure API endpoint needs to be provided in either the azure_endpoint parameter or as an environment variable named AZURE_OPENAI_API_BASE"
)
self.azure_endpoint = azure_endpoint # Azure 엔드포인트 저장
self.request_interval = 0 if rate_limit == -1 else 60.0 / rate_limit # API 요청 간격 계산 (초 단위)
self.llm_client = AzureOpenAI(
azure_endpoint=self.azure_endpoint, # Azure 엔드포인트 설정
api_key=self.api_key, # API 키 설정
api_version=self.api_version, # API 버전 설정
) # Azure OpenAI 클라이언트 초기화
self.cost = 0.0 # 비용 초기화
# @backoff.on_exception(backoff.expo, (APIConnectionError, APIError, RateLimitError), max_tries=10) # 백오프 재시도 (주석 처리됨)
def generate(self, messages, temperature=0.0, max_new_tokens=None, **kwargs):
"""
이전 메시지를 기반으로 다음 메시지 생성
매개변수:
messages (list): 대화 메시지 목록
temperature (float, 선택적): 생성 결과의 무작위성 정도. 기본값 0.0 (결정적)
max_new_tokens (int, 선택적): 생성할 최대 토큰 수. 기본값 4096
**kwargs: 추가 매개변수
반환값:
str: 생성된 응답 텍스트
"""
completion = self.llm_client.chat.completions.create(
model=self.model, # 사용할 모델 지정
messages=messages, # 대화 메시지 목록 전달
max_tokens=max_new_tokens if max_new_tokens else 4096, # 최대 토큰 수 설정
temperature=temperature, # 온도(무작위성) 설정
**kwargs, # 추가 매개변수 전달
)
total_tokens = completion.usage.total_tokens # 사용된 총 토큰 수 추출
self.cost += 0.02 * ((total_tokens + 500) / 1000) # 비용 계산 및 누적 (토큰당 비용 * 토큰 수)
return completion.choices[0].message.content # 생성된 메시지 내용 반환vLLM API 클래스 -> 텍스트 생성(무료)
class LMMEnginevLLM(LMMEngine):
# vLLM API를 사용하는 LLM 엔진 클래스
def __init__(
self, base_url=None, api_key=None, model=None, rate_limit=-1, **kwargs
):
"""
LMMEnginevLLM 클래스의 초기화 메서드
매개변수:
base_url (str, 선택적): vLLM 엔드포인트 URL. 없으면 환경 변수에서 가져옴
api_key (str, 선택적): vLLM API 키
model (str, 필수): 사용할 모델명
rate_limit (int, 선택적): 분당 API 요청 제한 수. -1이면 제한 없음. 기본값 -1
**kwargs: 추가 매개변수
"""
assert model is not None, "model must be provided" # 모델명이 제공되지 않으면 오류 발생
self.model = model # 모델명 저장
self.api_key = api_key # API 키 저장 (필수 아님)
self.base_url = base_url or os.getenv("vLLM_ENDPOINT_URL") # 엔드포인트 URL이 제공되지 않으면 환경 변수에서 가져옴
if self.base_url is None: # 엔드포인트 URL이 여전히 없으면 오류 발생
raise ValueError(
"An endpoint URL needs to be provided in either the endpoint_url parameter or as an environment variable named vLLM_ENDPOINT_URL"
)
self.request_interval = 0 if rate_limit == -1 else 60.0 / rate_limit # API 요청 간격 계산 (초 단위)
self.llm_client = OpenAI(base_url=self.base_url, api_key=self.api_key) # vLLM 엔드포인트에 연결하는 OpenAI 호환 클라이언트 초기화
# @backoff.on_exception(backoff.expo, (APIConnectionError, APIError, RateLimitError), max_tries=10) # 백오프 재시도 (주석 처리됨)
# TODO: Qwen 모델용으로 선택된 기본 매개변수
def generate(
self,
messages,
temperature=0.0,
top_p=0.8,
repetition_penalty=1.05,
max_new_tokens=512,
**kwargs
):
"""
이전 메시지를 기반으로 다음 메시지 생성
매개변수:
messages (list): 대화 메시지 목록
temperature (float, 선택적): 생성 결과의 무작위성 정도. 기본값 0.0 (결정적)
top_p (float, 선택적): 상위 확률 샘플링 비율. 기본값 0.8
repetition_penalty (float, 선택적): 반복 패널티. 기본값 1.05
max_new_tokens (int, 선택적): 생성할 최대 토큰 수. 기본값 512
**kwargs: 추가 매개변수
반환값:
str: 생성된 응답 텍스트
"""
completion = self.llm_client.chat.completions.create(
model=self.model, # 사용할 모델 지정
messages=messages, # 대화 메시지 목록 전달
max_tokens=max_new_tokens if max_new_tokens else 4096, # 최대 토큰 수 설정
temperature=temperature, # 온도(무작위성) 설정
top_p=top_p, # 상위 확률 샘플링 비율 설정
extra_body={"repetition_penalty": repetition_penalty}, # 반복 패널티 설정 (추가 매개변수)
)
return completion.choices[0].message.content # 생성된 메시지 내용 반환HuggingFace API 클래스 -> 텍스트 생성(무료)
class LMMEngineHuggingFace(LMMEngine):
# HuggingFace API를 사용하는 LLM 엔진 클래스
def __init__(self, api_key=None, endpoint_url=None, rate_limit=-1, **kwargs):
"""
LMMEngineHuggingFace 클래스의 초기화 메서드
매개변수:
api_key (str, 선택적): HuggingFace 토큰. 없으면 환경 변수에서 가져옴
endpoint_url (str, 필수): HuggingFace 엔드포인트 URL
rate_limit (int, 선택적): 분당 API 요청 제한 수. -1이면 제한 없음. 기본값 -1
**kwargs: 추가 매개변수
"""
assert endpoint_url is not None, "HuggingFace endpoint must be provided" # 엔드포인트 URL이 제공되지 않으면 오류 발생
self.endpoint_url = endpoint_url # 엔드포인트 URL 저장
api_key = api_key or os.getenv("HF_TOKEN") # API 키가 제공되지 않으면 환경 변수에서 가져옴
if api_key is None: # API 키가 여전히 없으면 오류 발생
raise ValueError(
"A HuggingFace token needs to be provided in either the api_key parameter or as an environment variable named HF_TOKEN"
)
self.api_key = api_key # API 키 저장
self.request_interval = 0 if rate_limit == -1 else 60.0 / rate_limit # API 요청 간격 계산 (초 단위)
self.llm_client = OpenAI(base_url=self.endpoint_url, api_key=self.api_key) # HuggingFace 엔드포인트에 연결하는 OpenAI 호환 클라이언트 초기화
@backoff.on_exception(
backoff.expo, (APIConnectionError, APIError, RateLimitError), max_time=60
) # API 오류 발생 시 지수 백오프로 60초까지 재시도
def generate(self, messages, temperature=0.0, max_new_tokens=None, **kwargs):
"""
이전 메시지를 기반으로 다음 메시지 생성
매개변수:
messages (list): 대화 메시지 목록
temperature (float, 선택적): 생성 결과의 무작위성 정도. 기본값 0.0 (결정적)
max_new_tokens (int, 선택적): 생성할 최대 토큰 수. 기본값 4096
**kwargs: 추가 매개변수
반환값:
str: 생성된 응답 텍스트
"""
return (
self.llm_client.chat.completions.create(
model="tgi", # 모델로 "tgi" 고정 사용
messages=messages, # 대화 메시지 목록 전달
max_tokens=max_new_tokens if max_new_tokens else 4096, # 최대 토큰 수 설정
temperature=temperature, # 온도(무작위성) 설정
**kwargs, # 추가 매개변수 전달
)
.choices[0] # 첫 번째 생성 결과 선택
.message.content # 생성된 메시지 내용 반환
)
모든 클래스가 필요한 건 아니다. vLLM API랑 임베딩 엔진만으로도 가능