랭체인v0.3 하이퍼클로바x 연동하기

랭체인v0.3 하이퍼클로바x 연동하기

 

AI 업계에서 일하고 계신 분들이라면 랭체인(LangChain)이라는 단어가 낯설지 않을 텐데요. 요즘은 개발자부터 MLOps 엔지니어까지 여러 다양한 직군의 채용공고만 보아도 랭체인이라는 단어가 심심치 않게 등장하고 있습니다. 그만큼 랭체인은 AI 서비스를 개발하는 데 있어 중요한 프레임워크로 빠르게 자리 잡아가고 있는 것 같습니다.

 

랭체인의 장점은 모듈화 된 구조로 인해 모델과 프롬프트를 자유자재로 교체해서 사용할 수 있다는 점과 배치 처리, 비동기호출 같은 다양한 함수들을 활용하여 답변을 생성할 수 있기 때문에 백엔드 개발에 필요한 여러 많은 작업들을 최소화시켜준다는 점입니다.

 

특히, 많은 사람들이 사용하고 있는 클로드나 GPT 같은 저명한 오픈소스 모델 연동 모듈은 랭체인 커뮤니티에서 써드파티 라이브러리로 제공하고 있어 별도의 API 호출 로직을 구현할 필요 없이 API키만 있으면 랭체인을 통해 편리하게 결괏값을 받아와서 사용할 수 있습니다.

 

단, 모든 AI 모델에 대한 연동 모듈이 제공되고 있는 것은 아닌데요. 특히 한국 모델들은 임베딩 모듈을 제외하면 답변 생성 모듈로는 아직 공식적으로 공식적으로 지원하는 사례가 없기 때문에 이 경우에는 직접 랭체인 내 코어 모듈을 상속받아 커스텀 클래스로 만들어주어야 합니다.

 

따라서 이번 포스팅에서는 가장 최근에 나온 랭체인 v0.3 버전에서 하이퍼클로바 X 연동 커스텀 모델 클래스를 만들어보고 랭체인에서 제공하고 있는 배치함수를 통해 이전에 만들어놓은 캐릭터챗봇 학습용 데이터셋을 좀 더 빠르고 효율적으로 구축할 수 있도록 기존 코드를 수정해 보도록 하겠습니다.

 

---

 

🤔 랭체인이란?

랭체인은 LLM으로 구동되는 애플리케이션과 파이프라인을 신속하게 구축할 수 있도록 도와주는 프레임워크로 현재 많은 개발자들의 주목을 받고 있습니다. 2022년 10월 해리슨 체이스(Harrison Chase)에 의해 오픈 소스 프로젝트로 시작했으며, 2024년 1월 초에 v0.1.0 안정(stable) 버전이 출시되었습니다.

 

다만, 마이너 업데이트임에도 불구하고 주요 코드들이 너무 자주 바뀌어 개발자들이 사용하면서 피로를 많이 느낀다는 점이 단점으로 부각되고 있는데요. 올해 초 stable버전인 v0.1이 나온 지 얼마 안돼서 5월에는 v0.2, 불과 한 달 전인 9월 16일에는 v0.3이 출시되었습니다.

 

v0.2에서 v0.3 업데이트에는 큰 차이점은 없지만 만약 아직 v0.2 보다 낮은 버전을 사용 중이시라면 이후 버전으로의 마이그레이션 작업량이 꽤 될 수 있기 때문에 주의하셔야 합니다. 다만, 앞으로 출시될 v1.0.0에서 상당수 deprecated 될 예정이라고 공지하고 있어 미리미리 조금씩 레거시 코드들을 업데이트하여 새로 출시될 버전에 대해 준비해 놓는 것도 좋을 것 같습니다.

 

v0.3 릴리즈 노트 보러 가기

LangChain v0.3 | 🦜️🔗 LangChain

 

---

 

👨🏻‍🔧 랭체인 커스텀 모델 클래스 만들기

1. LLM vs chat model 어떤 커스텀 LLM 클래스를 사용해야 할까?

랭체인에서 하이퍼클로바 X API를 사용하기 위해서는 직접 커스텀 모델 클래스로 만들어야 한다고 말씀드렸었는데요. 이때 고려해봐야 할 사항은 단일 메시지를 사용하느냐 아니면 멀티턴 메시지(AI와 대화를 여러 번 주고받은 대화내역)를 사용하느냐에 따라 구현 방법이 달라집니다.

 

간단한 명령어를 전달하여 출력결과를 얻고 싶으시다면 langchain_core.language_models 모듈 안에 있는 LLM 클래스를 상속받아서 구현하시면 되고 챗봇 처럼 대화내용을 전달하여 답변 결과를 얻고 싶으시다면 BaseChatModel 클래스를 상속받아서 구현하시면 됩니다.

 

방법 1. 커스텀 LLM 클래스 사용방법 (LLM 클래스 상속)

How to create a custom LLM class | 🦜️🔗 LangChain

 

방법 2. 커스텀 chat model 클래스 사용방법 (BaseChatModel 클래스 상속)

How to create a custom chat model class | 🦜️🔗 LangChain

 

각 커스텀 모델 클래스의 사용 용도에 대해서는 아래 코드를 통해 조금 더 직관적으로 살펴보실 수 있을 것 같습니다. 우선은 커스텀 LLM 클래스를 만들었다고 가정하고 다음과 같이 간단한 코드를 넣어보겠습니다.

 

커스텀 LLM 클래스 호출방법

llm = CustomLLM()
llm.invoke("안녕하세요")  # 답변 출력 예시) 안녕하세요. 무엇을 도와드릴까요?

 

이 경우에는 프롬프트 적용 시 다음과 같이 체이닝을 연결할 수 있습니다.

 

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages(
    [("system", "당신은 챗봇입니다."), ("human", "{input}")]
)

llm = CustomLLM()
chain = prompt | llm
chain.invoke("안녕하세요")           # 답변 출력 예시) 안녕하세요. 무엇을 도와드릴까요?

 

어떻게 보면 답변이 문자열(string) 형태로 온다는 점에서 편리할 수도 있을 것 같지만 AI와 계속해서 대화를 이어나가야 하는 상황이라면 생성된 답변 결과를 다시 프롬프트에 넣고 체이닝을 해야 되는 번거로움이 수반될 수 있습니다. 따라서 이 경우에는 커스텀 chat model을 사용하는 것이 더 적합할 수 있습니다.

 

커스텀 chat model 클래스 호출방법

from langchain_core.messages import AIMessage, HumanMessage

model = CustomChatModel()

model.invoke(
    [
        HumanMessage(content="안녕하세요"),
        AIMessage(content="안녕하세요. 무엇을 도와드릴까요?"),
        HumanMessage(content="너는 누구야?"),
    ]
)

# 답변 출력 예시)
# AIMessage(content='저는 AI 입니다.', response_metadata={}, id='run-ddb42bd6-4fdd-4bd2-8be5-e11b67d3ac29-0')

 

chat model을 사용하신다면 다음과 같이 프롬프트를 별도로 체이닝 하지 않고도 대화 내용을 배열 형태로 전달하여 결괏값을 얻을 수 있습니다. 실제로 공식 가이드 문서에서도 커스텀 llm 방식은 옛날 모델 호출 방식이므로 요즘 트렌드와는 거리가 멀어 llm보다는 chat models 커스텀 방식을 사용할 것을 권장하고 있습니다.

 

랭체인에서는 llm 보다는 chat models 사용을 권장 ( https://python.langchain.com/docs/concepts/#llms )

 

시간적 여유가 되신다면 둘 다 구현해서 적재적소에 맞게 사용하시면 더 좋습니다. 예를 들어, OpenAI 모델들과 연결할 수 있는 langchain_openai 써드파티 라이브러리의 내부 구조를 보시면 레거시 연동까지 고려하여 다음과 같이 각각 전부 구현이 되어있음을 확인하실 수 있습니다.

 

# langchain_openai-0.2.2 내부 패키지구조
langchain_openai
|- chat_models
    |- __init__.py
    |- azure.py  # 애저 chat model
    |- base.py   # openai chat model
|- llms
    |- __init__.py
    |- azure.py  # 애저 LLM
    |- base.py   # openai LLM

 

2. 랭체인 커스텀 chat model 클래스 만들기

저는 공식가이드 문서에서 권장하는 대로 chat model을 가지고 커스텀 클래스를 만들어보도록 하겠습니다. 커스텀 클래스를 만드는 방법은 굉장히 쉽습니다. langchain_core.language_models 모듈에 있는 BaseChatModel을 상속받아 _generate() 함수와 _stream() 함수의 내용만 채워주시면 바로 사용하실 수 있는데요.

 

저는 스트림 응답 방식은 사용하지 않을 예정이라서 _generate() 함수만 조금 수정을 해서 사용해 보도록 하겠습니다. 공식 가이드는 아래 링크를 통해서 확인하실 수 있습니다.

 

How to create a custom chat model class | 🦜️🔗 LangChain

 

먼저 간단하게 구글의 코랩을 이용해서 랭체인 설치부터 커스텀 클래스 만들기까지 한번 진행해 보도록 하겠습니다. 코랩에서는 랭체인을 다음과 같이 설치할 수 있습니다.

 

!pip install langchain
!pip install load_dotenv

 

구글의 코랩에서 랭체인 및 load_dotenv 모듈 설치

 

load_dotenv는 API키를. env 파일을 통해서 불러오기 위해 사용하는 모듈입니다. HyperCLOVA X API를 사용하기 위해서는 API 키가 두 종류 필요하기 때문에 해당 모듈을 랭체인과 같이 설치해 줍니다. API키는. env파일을 만들어서 클로바스튜디오에서 발급받은 API키를 아래와 같이 입력한 후 코랩에 업로드합니다.

 

X-NCP-CLOVASTUDIO-API-KEY=클로바스튜디오에서 테스트앱 발행 후 API 키값 입력
X-NCP-APIGW-API-KEY=클로바스튜디오에서 테스트앱 발행 후 API 키값 입력

 

HyperCLOVA X API가 담긴 환경변수 파일(.env)을 코랩에 업로드

 

만약. env파일이 보이지 않는다면 숨겨진 파일 표시 아이콘을 클릭하여 잘 업로드되었는지 확인합니다. .env파일이 잘 업로드되어있다면 다음과 같이 메모리에서도 로드할 수 있도록 코드를 실행합니다.

 

from dotenv import load_dotenv
load_dotenv('.env')

 

다음은 랭체인 버전도 같이 확인해 보시면 좋습니다. 저는 현재 기준으로 가장 최신 버전인 0.3.3 버전을 사용하였습니다.

 

랭체인 버전 확인하기

 

그리고 이제 클래스를 만들어주면 되는데요. 공식 가이드 문서를 참조해서 다음과 같이 간단하게 호출할 수 있을 정도로만 커스텀해보았습니다. 환경변수에 API 키값만 잘 설정해 놓으면 아래 클래스를 활용하여 HyperCLOVA X를 호출하는데 사용하실 수 있습니다.

 

import os
import requests
from pydantic import Field
from typing import Any, List, Optional

from langchain_core.callbacks import CallbackManagerForLLMRun
from langchain_core.language_models import BaseChatModel
from langchain_core.messages import AIMessage, BaseMessage
from langchain_core.outputs import ChatGeneration, ChatResult

class HyperCLOVAXChatModel(BaseChatModel):
    api_url: str = Field(default="https://clovastudio.stream.ntruss.com/testapp/v1/chat-completions/HCX-003")

    def _generate(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> ChatResult:
        headers = {
            "Content-Type": "application/json",
            "X-NCP-CLOVASTUDIO-API-KEY" : os.getenv("X-NCP-CLOVASTUDIO-API-KEY"),
            "X-NCP-APIGW-API-KEY": os.getenv("X-NCP-APIGW-API-KEY")
        }

        # 역할 매핑 함수
        def map_role(role: str) -> str:
            if role == "human":
                return "user"
            elif role == "ai":
                return "assistant"
            else:
                return role

        # API 형식으로 메시지 변환
        api_messages = [
            {"role": map_role(msg.type), "content": msg.content}
            for msg in messages
        ]

        data = {
            "messages": api_messages,
            "maxTokens": 100,
            "temperature": 0.5,
            "topP": 0.8,
            "repeatPenalty": 5.0
        }
        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])

    @property
    def _llm_type(self) -> str:
        return "HyperCLOVA X"

 

위와 같이 generate() 함수에 API 호출함수만 넣어주면 간단하게 커스텀 챗 모델 구현이 가능합니다. 더욱 더 다양한 기능들을 사용하고 싶다면 다른 모델들은 어떻게 구현하였는지 확인해 보시면 좋습니다. 예를 들어, 위에서 예시로 든 langchain_openai 모듈을 다운로드하여 chat model 클래스를 어떻게 구현하였는지 참고해 보시는 것을 추천드립니다.

 

---

 

🏭 랭체인으로 배치 돌려보기

위에서 커스텀 chat model을 만들 때 BaseChatModel을 상속해서 구현하였기 때문에 채팅 모델 호출 시 필요한 여러 다양한 함수들을 커스텀 모델 클래스에서도 동일하게 사용할 수 있게 되었습니다. 예를 들어, 기본적으로 제공하고 있는 LLM 호출 함수인 invoke와 비동기 호출인 ainvoke 그리고 Thread Pool 기반으로 동작하는 batch함수 등을 사용할 수 있습니다.

 

model = HyperCLOVAXChatModel()
model.batch(["안녕", "뭐해?"])
model.invoke("좋은 아침이야")

 

랭체인으로 커스텀모델 (HyperCLOVA X) 클래스를 batch와 invoke 함수로 각각 호출한 결과

 

위와 같이 호출했을 경우 응답 결과가 잘 떨어지는 것을 확인하실 수 있습니다. 원래는 타입을 명시해주어야 하지만 일반적으로 단순한 문자열이 주어지는 경우에는 자동으로 HumanMessage로 변환되어 처리되기 때문에 문자열만으로도 편리하게 사용할 수 있습니다.

 

제공되는 함수들 중에 배치함수를 활용해서 기존에 반복문을 통해서 호출하는 코드에 비해 얼마나 속도가 빨라졌는지 확인해 보도록 하겠습니다. 소스코드는 이전에 만들어놓았던 캐릭터챗봇 데이터셋 자동생성기를 이용하도록 하겠습니다.

 

그러면 준비된 125개의 질문에 대해 먼저 기존 소스코드부터 얼마나 걸리는지 확인해 보도록 하겠습니다. 로그를 확인해 보면 랭체인을 안 쓰고 순수 반복문을 통해 모델 API를 호출한 경우 17분 14초 정도 걸리는 것을 확인해 볼 수 있었습니다.

 

랭체인 없이 반복문을 통해서 HyperCLOVA X를 호출한 결과 17분 소요

 

그리고 다음은 랭체인의 batch 함수를 사용해서 모델 API를 호출한 경우입니다.

 

랭체인의 batch 함수를 통해서 HyperCLOVA X를 호출한 결과 5분 소요

 

5분 31초로 랭체인의 batch함수를 쓰지 않았을 때보다 평균적으로 3배 이상 차이가 나는 것을 확인해 볼 수 있었습니다. 단, 주의해야 할 점은 HyperCLOVA X에 랭체인 batch 함수를 돌리는 경우 아래와 같은 오류가 발생할 수 있는데요.

 

{'status': {'code': '42901', 'message': 'Too many requests - rate exceeded'}, 'result': None}

 

HyperCLOVA X의 경우 다음과 같이 분당 처리할 수 있는 최대 이용량이 정해져 있기 때문에 이를 고려해서 오류가 나지 않도록 설정해 주는 것이 중요합니다. 테스트 앱의 경우 분당처리량은 다음과 같습니다.

 

테스트앱에서의 분당 호출 최대 이용량 가이드문서( https://guide.ncloud-docs.com/docs/clovastudio-ratelimiting )

 

• QPM (Queries per Minute) : 1분 동안 모델 및 도구에 작업을 요청한 횟수
• TPM (Tokens per Minute) : 1분 동안 처리할 토큰 수 (입력된 토큰 수+결괏값 생성 시 사용할 최대 토큰 수(maxTokens))

 

다행히도 랭체인에서는 max_concurrency라는 옵션을 사용하면 오류가 발생하지 않도록 동시 호출 개수를 미리 제한을 할 수가 있습니다. 따라서 해당 옵션으로 최대 쓰레드 생성 개수를 조절하여 최적의 상황을 찾는 것이 중요합니다. 저의 경우에는 max_concurrency를 2로 조정했을 때 125개의 쿼리에 대해 5분 내로 모두 성공적으로 처리해 낼 수 있었습니다.

 

if __name__ == '__main__':
    with open('datasets.csv', 'r') as file:
        reader = csv.reader(file)
        rows = list(reader)

    messages = [[SystemMessage(content=system_prompt), HumanMessage(content=user_prompt.format(question=row[2]))] for row in rows]
    responses = model.batch(messages, config={"max_concurrency":2})

 

전체적인 코드베이스가 궁금하시다면 아래 링크를 통해 참고하시면 좋을것 같습니다.

 

GitHub - chucoding/hcx-datasets-auto-generator: 하이퍼클로바X 튜닝용 데이터셋 자동생성기

 

 

---

 

🌠 마무리하며

지금까지 랭체인 v0.3 버전을 이용하여 HyperCLOVA X 모델을 연동하고 배치 함수까지 써보는 과정을 진행해 보았는데요. 직접 구현을 해보니까 랭체인이 왜 각광을 받는 프레임워크인지 깨달을 수 있었던 시간이었습니다.

 

저는 랭체인에서 배치함수만 간단히 써보았지만, 이외에도 비동기방식 등 다양한 호출 함수들이 있고 프롬프트로 사용할 수 있는 RAG나 벡터 DB, 임베딩 같은 모듈들도 선택해서 사용할 수 있기 때문에 한번 배워두면 생성형 AI로 서비스를 만드실 때 다양한 방면에서 활용이 가능하겠다는 생각이 듭니다.

 

비록 지금은 커스텀 클래스로 만들어보았지만 언젠가는 HyperCLOVA X 같은 국내 모델들도 랭체인에서 공식적으로 지원이 되어 조금 더 간편하게 써 볼 수 있으면 좋을 것 같다는 생각이 듭니다.