LangChain 으로 HyperClovaX 적용하는 방법

오랜만에 포스팅하게 되는데 정말 그동안 많은 일들이 있었네요. 오늘은 Stockelper를 개발하는 과정에서 랭체인과 하이퍼크로버를 결합하기 위해 겪었던 과정들을 포스팅해봅니다.

 

LangChain 을 활용한 Custom LLM 사용하기 (with HyperClova X)

 

예전에 CustomLLM 기능을 참고해서 클래스를 상속받아서 구현하는 형태로 구현을 한 적이 있는데요. 이 때도 이상하게 Stream 기능이. 잘 동작하지 않았습니다. 아마 지금 돌이켜보니깐 Metadata json 파싱과정에서 뭔가 원활하게 코드가 구성이 안된 것 같습니다. 랭체인이 0.3 버전이 되면서 Pydantic 도 v2로 이미 배포를 해두었다는 걸 알게 되었습니다.

Chat models | 🦜️🔗 LangChain

그땐 알지 못했던 Chat models라고 해서, 랭체인을 통해서 다양한 LLM을 커스터마이징 해서 쓸 수 있는 Class 타입의 코드 docs 가 있었습니다.

 

Chat models 

구현이 가능한 주요 컴포넌트 : 

_generate: 챗 응답을 생성하는 핵심 메서드

_llm_type: 모델 타입을 위한 고유 식별자

_stream: (선택사항) 스트리밍 응답 지원

_identifying_params: 모델 추적과 모니터링 지원

 

하이퍼 크로버의 코드를 결합해야 하기 때문에 여기서 중요한 건 completion_excutor의 값과 request_data를 래핑을 해주어야 합니다.

import requests


class CompletionExecutor:
    def __init__(self, host, api_key, api_key_primary_val, request_id):
        self._host = host
        self._api_key = api_key
        self._api_key_primary_val = api_key_primary_val
        self._request_id = request_id

    def execute(self, completion_request):
        headers = {
            'X-NCP-CLOVASTUDIO-API-KEY': self._api_key,
            'X-NCP-APIGW-API-KEY': self._api_key_primary_val,
            'X-NCP-CLOVASTUDIO-REQUEST-ID': self._request_id,
            'Content-Type': 'application/json; charset=utf-8',
            'Accept': 'text/event-stream'
        }

        with requests.post(self._host + '/testapp/v1/chat-completions/HCX-003',
                           headers=headers, json=completion_request, stream=True) as r:
            for line in r.iter_lines():
                if line:
                    print(line.decode("utf-8"))


if __name__ == '__main__':
    completion_executor = CompletionExecutor(
        host='https://clovastudio.stream.ntruss.com',
        api_key='api_key',
        api_key_primary_val='api_key_primary_val',
        request_id='request_id'
    )

    preset_text = [{"role":"system","content":"당신은 AI 전문가입니다"},
    				{"role":"user","content":"안녕하세요"}]

    request_data = {
        'messages': preset_text,
        'topP': 0.8,
        'topK': 0,
        'maxTokens': 3450,
        'temperature': 0.1,
        'repeatPenalty': 0.09,
        'stopBefore': [],
        'includeAiFilters': True,
        'seed': 0
    }

    print(preset_text)
    completion_executor.execute(request_data)

 

 

 

HyperCLOVA X를 LangChain과 통합하기 위해 BaseChatModel을 상속받아 구현했습니다. 주요 구성요소는 다음과 같습니다:

class ClovaX(BaseChatModel):
    # test앱 정보 
    api_url: str = Field(default="https://clovastudio.stream.ntruss.com/testapp/v1/chat-completions/HCX-003")
    api_key: str = Field(...)
    api_key_primary_val: str = Field(...)
    request_id: str = Field(...)

 

pydantic Field를 사용해 모델의 기본 설정 관리를 할 수 있도록 합니다. 이전에는 그냥 넣었던 거 같은데 pydantic을 써야 합니다. 자꾸 에러가 났습니다.

 

def _generate(
    self,
    messages: List[BaseMessage],
    stop: Optional[List[str]] = None,
    run_manager: Optional[CallbackManagerForLLMRun] = None,
    **kwargs: Any,
) -> ChatResult:

생성하는 부분에서 그대로 docs 거에 가져오고 하나씩 보면

 

messages: List [BaseMessage] 이 랭체인이 지원하는 SystemMessage, HumanMessage, AIMessage를 통해 쌓아서 그걸 _generate 메서드로 전달을 해주는 방식입니다. 즉 LLM과 대화한 부분을 넘겨주는 메서드입니다.

 

stop: Optional [List [str]] = None 은. 특정 단어가 나오면 생성이 끝이 나게끔 해주는 파라미터로 none이고 기본적으로 쓸 일은 없어 보입니다.

 

run_manager: Optional [CallbackManagerForLLMRun] = None 은 생성과정을 추적하고 관리하는 역할입니다. 다양한 콜백매니저를 통해 상황 모니터링이 가능합니다.

 

 **kwargs:any 사용자가 커스터마이징을 해서 추가적인 키워드를 인자로 받을 수도 있습니다.

 

-> ChatResult: ChatResult 객체를 반환해서 사용을 한다.

preset_text = [{"role":"system","content":"당신은 AI 전문가입니다"},
    				{"role":"user","content":"안녕하세요"}]

 

하이퍼크로버가 위 방식처럼 데이터를 전달하는 점이 있습니다. 우리는 랭체인을 통해서 프롬프트 관리를 하지만, 위와 같은 형태로 바꿔주어야 하기 때문에 convert 함수를 만들어야 합니다.

def _convert_message_to_naver_chat_message(
    message: BaseMessage,
) -> Dict:
    if isinstance(message, ChatMessage):
        return dict(role=message.role, content=message.content)
    elif isinstance(message, HumanMessage):
        return dict(role="user", content=message.content)
    elif isinstance(message, SystemMessage):
        return dict(role="system", content=message.content)
    elif isinstance(message, AIMessage):
        return dict(role="assistant", content=message.content)
    else:
        logger.warning(
            "FunctionMessage, ToolMessage not yet supported "
            "(https://api.ncloud-docs.com/docs/clovastudio-chatcompletions)"
        )
        raise ValueError(f"Got unknown type {message}")

 

그 후에 이제 data 파라미터를 넣고 랭체인 양식대로 작성해 주면 생성코드 부분은 끝이 납니다.

data = {
            "messages": messages_list,
            "maxTokens": 1024,
            "temperature": 0.1,
            "topP": 0.8,
            "repeatPenalty": 1.2
        }
        response = requests.post(self.api_url, json=data, headers=headers)
        response_json = response.json()

        message = AIMessage(
            content=response_json["result"]["message"]["content"]
        )

        generation = ChatGeneration(message=message)
        return ChatResult(generations=[generation])

 

그렇게 리스폰해주고 사용하면 되는데.. Stream 코드를 작성하고 있다가. 파싱오류가 자꾸 떠서 고민을 하던 중 hyperclovax langchain을 쳤는데 두둥.. 랭체인이 드디어 하이퍼크로버를 지원하고 있었습니다.

Chat models | 🦜️🔗 LangChain

from langchain_community.chat_models import ChatClovaX

chat = ChatClovaX(
    model="HCX-003",
    max_tokens=100,
    temperature=0.5,
    # clovastudio_api_key="..."    # set if you prefer to pass api key directly instead of using environment variables
    # task_id="..."    # set if you want to use fine-tuned model
    # service_app=False    # set True if using Service App. Default value is False (means using Test App)
    # include_ai_filters=False     # set True if you want to detect inappropriate content. Default value is False
    # other params...
)

 

위에서 어렵게 어렵게 클래스 상속받아 코드를 구현해 놨더니 이런! 저런! 심지어 파인튜닝 모델까지도.. 정말 어이가 없었지만 바로 langchain-community를 설치하고 0.3.3 버전을 확인한 후 코드를 실행했습니다.

 

계속 지우고 깔아봐도 0.3.3 버전이 맞는데 ChatClovaX에 import를  할 수가 없었습니다. 그래서 랭체인 github를 들가서 확인해 보니 소스코드는 있었고 어 왜 안되지? 만 반복하다가 pypi 가 일정 업데이트가 안된 거 일 수도 있다고 GPT가 말을 해주었습니다.

 

그래서 그럼 그냥 코드 저거 가져다 쓰면 되잖아 생각이 들었고 그냥 가져와서 utils로 만들어 쓰기로 하였습니다.

https://github.com/langchain-ai/langchain/blob/master/libs/community/langchain_community/chat_models/naver.py

langchain/libs/community/langchain_community/chat_models/naver.py at master · langchain-ai/langchain

코드를 한번 쭉 보았는데 비슷하지만 httpx 관련과 조금. 더짜임새 있게 짠 모습을 볼 수 있었습니다. 진짜 개발자는 아무나 하는 게 아니다.. 란 생각과 stream 코드도 쭉 보았는데, 토큰처리 통신 관련이기 때문에 확실히 아 이거 원래도 못 짰겠다 생각은 들었습니다.

 

def _stream(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> Iterator[ChatGenerationChunk]:
        message_dicts, params = self._create_message_dicts(messages, stop)
        params = {**params, **kwargs, "stream": True}

        default_chunk_class: Type[BaseMessageChunk] = AIMessageChunk
        for sse in self._completion_with_retry(
            messages=message_dicts, run_manager=run_manager, **params
        ):
            new_chunk = _convert_chunk_to_message_chunk(sse, default_chunk_class)
            default_chunk_class = new_chunk.__class__
            gen_chunk = ChatGenerationChunk(message=new_chunk)

            if run_manager:
                run_manager.on_llm_new_token(
                    token=cast(str, new_chunk.content), chunk=gen_chunk
                )

            yield gen_chunk

 

 

SSE(Server-Sent Events)를 사용하여 실시간 스트리밍 처리 한 후 각 청크를 적절한 메시지 타입으로 변환하고 콜백 매니저를 통한 토큰 처리한 후 제너레이터를 사용한 메모리 효율적인 처리까지 지원을 해서 streaming 처리를 하는 모습입니다. 차라리 진작 랭체인 독스를 들가서 다른 model 들을 참조해 볼걸 그랬나 라는 생각도 들었습니다.

결론

 

from naver import ChatClovaX  # 로컬에 만든 naver.py 파일에서 import
from langchain_core.messages import SystemMessage, HumanMessage

# ChatClovaX 인스턴스 생성
chat = ChatClovaX(
    ncp_clovastudio_api_key="api_key",
    ncp_apigw_api_key="api_key_primary",
    model_name="HCX-003",
    temperature=0.1,
    top_p=0.8,
    top_k=0,
    max_tokens=3450,
    repeat_penalty=0.09,
    include_ai_filters=True,
    seed=0
)

# 메시지 설정
messages = [
    SystemMessage(content="""#역할
당신의 역할은 나스닥 상장 레버리지 ETF 트레이딩을 위한 고급 가상 어시스턴트입니다..."""),
    HumanMessage(content="안녕하세요")
]

# 동기 방식으로 실행
response = chat.invoke(messages)
print(response.content)

# 또는 스트리밍 방식으로 실행
for chunk in chat.stream(messages):
    print(chunk.content, end="")

# 비동기 방식으로 실행하려면
async def async_chat():
    response = await chat.ainvoke(messages)
    print(response.content)

# 비동기 스트리밍
async def async_stream_chat():
    async for chunk in chat.astream(messages):
        print(chunk.content, end="")

 

결론

 

naver.py로 랭체인 코드를 긁어와서 utils로 만들어주고 import 해서 쓰시면 됩니다. 이 과정을 개발하기 위해 왜 이렇게 삽질을 했나 싶기도 했지만 어쨌든 잘 돌아가니 저처럼 고생하지 말고 가져다 쓰시길 추천드리겠습니다.

 

이제 LangChain 이 공식으로 지원해 주니 clovaX로 개발이 조금 더 수월해지지 않을까 기대를 하고 있습니다. 현재 나스닥 ETF레버리지 매수 매도를 위한 봇을 만드는 중인데 개발 계획에 탄력이 될 것 같습니다.