LLM 커스터마이징
현재 AI 모델 도메인에서는 다양한 모델이 존재하며, LangChain의 공식 통합은 모든 모델을 커버하지 않습니다. 때로는 사용자 고유의 모델을 만들어 LangChain 프레임워크에 통합해야 할 수도 있습니다.
이 장에서는 사용자 고유의 LLM 래퍼를 만드는 방법을 소개하며, 여기에는 LangChain에서 지원하지 않는 모델 또는 사용자 고유의 모델을 편리하게 사용할 수 있게 됩니다.
LangChain에서는 사용자 고유의 LLM 또는 LangChain에서 지원하는 래퍼와는 다른 래퍼를 사용하려면 사용자 고유의 LLM 래퍼를 만들 수 있습니다. 사용자 고유의 LLM은 오직 두 가지 필수 메소드를 구현하는 것만 필요합니다:
-
_call메소드는 입력으로 문자열과 선택적으로 스탑 단어 목록을 받고, 모델을 호출하는 문자열을 반환합니다. -
_llm_type속성은 모델 이름을 나타내는 문자열을 반환하며, 이는 오로지 로깅 목적으로 사용됩니다.
필수 메소드 외에도 사용자 고유의 LLM은 선택적으로 다음 메소드를 구현할 수 있습니다:
-
_identifying_params속성은 클래스를 출력하는 데 도움을 주는데 사용되며, 딕셔너리를 반환해야 합니다.
간단한 사용자 고유의 LLM 구현
이제 입력된 문자열의 처음 n개의 문자만 반환하는 매우 간단한 사용자 고유의 LLM을 구현해보겠습니다.
from typing import Any, List, Mapping, Optional
from langchain_core.callbacks.manager import CallbackManagerForLLMRun
from langchain_core.language_models.llms import LLM
class CustomLLM(LLM):
n: int
@property
def _llm_type(self) -> str:
return "custom"
def _call(
self,
prompt: str,
stop: Optional[List[str]] = None,
run_manager: Optional[CallbackManagerForLLMRun] = None,
**kwargs: Any,
) -> str:
if stop is not None:
raise ValueError("stop kwargs are not permitted.")
return prompt[: self.n]
@property
def _identifying_params(self) -> Mapping[str, Any]:
"""출처"""
return {"n": self.n}
이제 이 사용자 고유의 LLM을 다른 LLM과 마찬가지로 사용할 수 있습니다.
사용자 고유의 LLM 사용
사용자 고유의 LLM 객체를 인스턴스화하고 사용하여 사용자 고유의 LLM을 호출하고 출력을 커스터마이즈하는 방법을 시연할 수 있습니다.
llm = CustomLLM(n=10)
llm.invoke("This is a foobar thing")
'This is a '
또한 사용자 고유의 LLM을 출력하여 커스터마이즈된 출력을 확인할 수 있습니다.
print(llm)
CustomLLM
Params: {'n': 10}
채팅 모델 커스터마이징
이제 LangChain의 채팅 모델을 커스터마이즈하는 방법에 대해 설명하겠습니다.
메시지 입력과 출력
채팅 모델에서는 메시지가 입력과 출력의 중심입니다. 메시지는 사용자가 입력한 내용과 모델이 생성한 응답을 말합니다.
메시지
채팅 모델은 메시지를 입력으로 받고, 하나 이상의 메시지를 출력으로 생성합니다. LangChain에서는 다음과 같은 내장 메시지 유형이 있습니다.
-
SystemMessage: AI 동작을 초기화하기 위해 사용되며, 일반적으로 일련의 입력 메시지 중 첫 번째 메시지로 사용됩니다. -
HumanMessage: 채팅 모델과의 상호작용을 나타냅니다. -
AIMessage: 채팅 모델에서의 메시지를 나타내며, 텍스트 또는 도구 호출 요청일 수 있습니다. -
FunctionMessage/ToolMessage: 도구 호출의 결과를 모델로 반환하기 위해 사용됩니다.
이러한 메시지 유형은 OpenAI의 function 및 tool 매개변수를 기반으로 조정하는 등 특정 요구에 따라 확장 및 커스터마이즈할 수 있습니다.
from langchain_core.messages import (
AIMessage,
BaseMessage,
FunctionMessage,
HumanMessage,
SystemMessage,
)
부드러운 변형
모든 채팅 메시지에는 이름에 Chunk가 포함된 부드러운 변형이 있습니다.
from langchain_core.messages import (
AIMessageChunk,
FunctionMessageChunk,
HumanMessageChunk,
SystemMessageChunk,
ToolMessageChunk,
)
이러한 Chunk는 채팅 모델을 스트리밍할 때 사용되며, 각각 누적 속성을 정의합니다!
예시
AIMessageChunk(content="Hello") + AIMessageChunk(content=" world!")
반환
AIMessageChunk(content='Hello world!')
간단한 채팅 모델
SimpleChatModel을 상속받으면 빠르게 간단한 채팅 모델을 구현할 수 있습니다.
채팅 모델에 필요한 모든 기능을 포함하고는 있지 않을 수 있지만, 빠른 구현이 가능합니다. 더 많은 기능이 필요한 경우 아래에 설명된 BaseChatModel로 전환하는 것이 가능합니다.
SimpleChatModel을 상속받기 위해서는 다음 인터페이스의 구현이 필요합니다:
-
_call메서드 - 외부 모델 API 호출의 구현.
또한, 다음을 지정할 수 있습니다:
-
_identifying_params속성 - 모델 매개변수화 정보를 기록하는 데 사용됨.
선택 사항:
-
_stream메서드 - 스트리밍 출력을 구현하는 데 사용됨.
기본 채팅 모델
BaseChatModel을 상속받기 위해서는 _generate 메서드와 _llm_type 속성의 구현이 필요합니다. 선택적으로 _stream, _agenerate, _astream, _identifying_params의 구현이 가능합니다.
사용자 정의 채팅 모델 예시
이 섹션에서는 CustomChatModelAdvanced라는 사용자 정의 채팅 모델의 코드 구현을 포함하여 채팅 결과 생성, 스트리밍 출력 및 비동기 스트림 구현을 보여줍니다.
from typing import Any, AsyncIterator, Dict, Iterator, List, Optional
from langchain_core.callbacks import (
AsyncCallbackManagerForLLMRun,
CallbackManagerForLLMRun,
)
from langchain_core.language_models import BaseChatModel, SimpleChatModel
from langchain_core.messages import AIMessageChunk, BaseMessage, HumanMessage
from langchain_core.outputs import ChatGeneration, ChatGenerationChunk, ChatResult
from langchain_core.runnables import run_in_executor
class CustomChatModelAdvanced(BaseChatModel):
"""마지막 메시지의 첫 `n`개 문자를 반환하는 사용자 정의 채팅 모델을 구현합니다."""
n: int
"""사용자 정의 모델 매개변수"""
def _generate(
self,
messages: List[BaseMessage],
stop: Optional[List[str]] = None,
run_manager: Optional[CallbackManagerForLLMRun] = None,
**kwargs: Any,
) -> ChatResult:
"""여기서 모델 호출 로직을 구현하며, 일반적으로는 서드파티 모델의 API를 호출한 후, API에서 반환한 결과를 langchain이 인식할 수 있는 형식으로 캡슐화합니다.
주요 매개변수 설명:
messages: 메시지로 구성된 프롬프트 목록
"""
last_message = messages[-1]
tokens = last_message.content[: self.n]
message = AIMessage(content=tokens)
generation = ChatGeneration(message=message)
return ChatResult(generations=[generation])
def _stream(
self,
messages: List[BaseMessage],
stop: Optional[List[str]] = None,
run_manager: Optional[CallbackManagerForLLMRun] = None,
**kwargs: Any,
) -> Iterator[ChatGenerationChunk]:
"""모델 스트리밍 출력의 구현, '_generate' 메서드와 유사하지만 스트리밍 출력 처리가 포함됨."""
last_message = messages[-1]
tokens = last_message.content[: self.n]
for token in tokens:
chunk = ChatGenerationChunk(message=AIMessageChunk(content=token))
if run_manager:
run_manager.on_llm_new_token(token, chunk=chunk)
yield chunk
async def _astream(
self,
messages: List[BaseMessage],
stop: Optional[List[str]] = None,
run_manager: Optional[AsyncCallbackManagerForLLMRun] = None,
**kwargs: Any,
) -> AsyncIterator[ChatGenerationChunk]:
"""`stream` 메서드의 비동기 버전 구현"""
result = await run_in_executor(
None,
self._stream,
messages,
stop=stop,
run_manager=run_manager.get_sync() if run_manager else None,
**kwargs,
)
for chunk in result:
yield chunk
@property
def _llm_type(self) -> str:
"""사용자 정의 모델의 태그를 반환합니다"""
return "echoing-chat-model-advanced"
@property
def _identifying_params(self) -> Dict[str, Any]:
"""사용자 정의 디버그 정보를 반환합니다"""
return {"n": self.n}
사용자 정의 채팅 모델 테스트
채팅 모델을 테스트해 봅시다. invoke, batch, stream 메서드 및 비동기 스트림 구현을 사용합니다.
model = CustomChatModelAdvanced(n=3)
model.invoke([HumanMessage(content="안녕!")])
model.invoke("안녕")
model.batch(["안녕", "안녕히 가세요"])
for chunk in model.stream("고양이"):
print(chunk.content, end="|")
async for chunk in model.astream("고양이"):
print(chunk.content, end="|")
async for event in model.astream_events("고양이", version="v1"):
print(event)