https://github.com/eususu/lite-llm-client
GitHub - eususu/lite-llm-client: LLM client library. it communicate with LLM by only REST API
LLM client library. it communicate with LLM by only REST API - eususu/lite-llm-client
github.com
필요성
2024년 7월 21일 시작한 프로젝트
나름 AI관련 업무를 시작하게 되면서, 계속 번거로왔다.
1. 파이썬 패키지 의존성의 무게와 충돌 (심한욕..)
2. 점점 늘어가는 테스트할 AI 클라이언트 (openai, gemini, anthropic, ...)
3. 각 클라이언트별 의존성 패키지 이슈
99. 파이썬 초보라 겸사겸사 스터디
위의 상황을 깔끔하게 정리하고 각 API를 조사하게 되었음
방법
모든 LLM 클라이언트들은 REST API를 제공하고 있음
이름만 다르지 결국 LLM이 하는 작업은 고정되어 있어, 인터페이스 형태만 다른 것임
설치형 LLM 들도 결국에는 현재 시점(2024/10/31)에는 사실상 OPEN AI REST API로 통일됨
이제 각 서비스 별 REST API를 구현하고, 이를 사용할 수 있는 파이썬 인터페이스를 구현하면 됨
class LLMMessageRole(IntEnum):
USER=1
SYSTEM=2
ASSISTANT=3
class LLMMessage(BaseModel):
role: LLMMessageRole
content: str
class InferenceOptions(BaseModel):
top_p:Optional[float]=None
top_k:Optional[float]=None
max_tokens: Optional[int]=None
temperature:float=0.0
class LLMClient(ABC):
@abstractmethod
def chat_completions(self, messages:List[LLMMessage], options:InferenceOptions):
raise NotImplementedError
이런 형태로 정리됨.
※ 파이썬으로 개발을하면서 타입 추론이 안되는 타입 때문에 고생한지라, 나름 typehint를 열심히 박아놨음.
이제 LLMClient를 상속받을 각 서비스별 클래스를 생성하여 구현만하면되는 간단한 작업.
내부 구현 코드는 github에 공개되어 있으니, 그것을 참조
자동 완성을 위한 symbol 은닉
나름 기능을 완성하고 라이브러리로서 사용하려는데, 대부분의 symbol이 자동완성되어 불편했음.
파이썬은 기본적으로 symbol의 이름이 underscore(_)로 시작하면 내부용도임을 의미한다고 함
주요 파일 이름과 멤버 이름을 적절하게 변경하여 주요 제공 기능이 먼저 나타나도록 처리함.
비동기
LLM 의 중요한 매력 중에 하나는 타자기를 치는 것처럼 몇글자씩 나타나는 것이라고 생각한다.
사실 LLM의 느린 추론 속도를 감추는데에도 한몪을 하고..
하여 인터페이스는 아래처럼 변경되었다.
이 작업을하면서 파이썬의 generator, iterator에 대해서 조금 더 보게되었음.
class LLMClient(ABC):
@abstractmethod
def chat_completions(self, messages:List[LLMMessage], options:InferenceOptions):
raise NotImplementedError
@abstractmethod
def async_chat_completions(self, messages:List[LLMMessage], options:InferenceOptions=InferenceOptions())->Iterator[str]:
raise NotImplementedError
인터페이스는 쉽다. 하지만 구현은 어떨까?
SSE(Service Sent Event)
요청 전문에 "stream": true 만 추가하는것으로 비동기는 시작된다.
다만 응답은 일반적인 HTTP 데이터인데, Content-Type이 text/event-stream으로 전달된다.
event-stream으로 오는 응답은 규격이 정해진것이 아니므로, open ai, gemini 등 자체 규격을 확인하고 그에 맞게 해석해야한다.
데이터는 간단하게 보면 아래처럼 온다
event: blah blah
data: blah blah
data:
이 값을 잘 보고 해석하면 되고 그 코드는 역시 github로..
※ 각 데이터의 한줄을 구분짓는 EOL도 서비스 별로 달라서 귀찮은 부분이 제법 있음
※ 이때 당시 SSE 응답 해석중에 gemini는 서비스도 내려가고, 불안했음.. (즉 언제 어떻게 수정되고 바뀔지 모르겠다)
어찌되었든 이 과정을 통해서 몇 글자의 응답을 빠르게 받을 수 있다
하나씩 응답하기
generator 개념에 익숙하지 않아서 어려웠지. 코드는 매우 단순하다. 반환해야 할 값을 return 쓰듯이 쓰면 된다.
yield "방금 받은 몇글자"
github 코드를 보면, 기초적인 SSE를 해석하는 부분과 각 LLM 벤더별로 SSE Data 를 해석하는 두개의 generator 중첩되어 있다
for event in decode_sse(http_response, data_type=SSEDataType.JSON):
"""
value example:
{'id': 'chatcmpl-9qLv6AAbMZcZudyYUJ2SsSYGZs16y', 'object': 'chat.completion.chunk', 'created': 1722264344, 'model': 'gpt-4o-2024-05-13', 'system_fingerprint': 'fp_400f27fa1f', 'choices': [{'index': 0, 'delta': {'role': 'assistant', 'content': ''}, 'logprobs': None, 'finish_reason': None}]}
"""
delta = event.event_value['choices'][0]['delta']
if 'content' in delta:
char = delta['content']
logging.debug(char)
yield char
# CHECK: first token is empty. is useful??
else:
# maybe last data?
pass
sse를 해석하여 실제 데이터만 추출한 iterator를 반환
그 iterator를 다시 하나씩 풀어서 json 포맷을 해석해서 다시 generator로 반환
각 LLM 서비스 별로 구현되어 있으니, 해당 서비스 구현 파일을 참조하면 상세하게 볼 수 있다.
그 결과 실제 사용하는 코드는 아래 처럼 단순해진다
answer = client.async_chat_completions(messages=messages)
for a in answer:
print(a)
마치면서
역시 새 언어를 배울때는 뭔가 만들어보는 것이 최고 인듯.
멀티 스레딩을 이용한 비동기 처리나 promise 등을 사용했었다가 새로운 개념을 접해서 즐거움.
내 나름의 스트레스인 의존성을 해결해서 마음이 편해짐.
SSE를 좀더 제대로 다뤄봐서 기쁨.
얼마 안되는 내 오픈소스!
앞으로
사실 앞으로는 아니고, 지금 하고 있는데, open telemetry trace를 적용해보고 있음.
나중에 필요성이 느껴지면, Multi modal도 처리해볼까 생각만 하고 있음.