Daddy Makers: LLM 모델 통합과 다양한 데이터소스를 지원하는 LangChain 아키텍처 및 동작 메커니즘 분석하기

이 글은 해리슨 체이스가 개발한 오픈소스인 랭체인(LangChain) 아키텍처와 동작 방법을 분석한다. 현재 개발자들 사이에 대중적으로 사용되고 있는 랭체인은 LLM 모델 통합과 다양한 데이터소스를 지원하여, LLM모델의 활용성을 극대화한다. 이 글을 통해, LLM 서비스 개발에 필요한 랭체인의 아키텍처와 동작 원리를 이해할 수 있다.

LLM 용어와 상세 개념은 다음 링크를 참고한다.

개요

Langchain(랭체인)은 LLM에 원하는 결과를 얻을 수 있도록, 다양한 프롬프트 입력 및 구조화된 출력, RAG, 튜닝과 같은 기능을 제공하는 라이브러리다. 랭체인 설치는 다음과 같다(참고).

pip install langchain

기본 사용법

랭체인은 모델 입출력, 데이터 검색, 에이전트 지원, 체인, 컨텍스트 메모리 기능을 제공하며, LCEL(LangChain Expression Language)를 이용해 각 구성요소를 유기적으로 연결시킬 수 있다. LCEL은 유닉스 파이프라인 개념을 차용했다. 다음은 LCEL 예시를 보여준다.

from langchain.chat_models import ChatOpenAI

from langchain.prompts import ChatPromptTemplate

from langchain.schema import BaseOutputParser

# LCEL 예시

chain = ChatPromptTemplate() | ChatOpenAI() | CustomOutputParser()

이와 더불어, 목적에 맞는 다양한 프롬프트 템플릿, 구조화된 출력을 제공한다.

from langchain.output_parsers.json import SimpleJsonOutputParser

json_prompt = PromptTemplate.from_template(

"Return a JSON object with `birthdate` and `birthplace` key that answers the following question: {question}"

)

json_parser = SimpleJsonOutputParser() # JSON 파서

# 프롬프트, 모델, 파서 체인 생성

json_chain = json_prompt | model | json_parser # 유닉스 파이프라인 개념 차용함.

result_list = list(json_chain.stream({"question": "When and where was Elon Musk born?"}))

print(result_list)

구조 분석

패키지 구조

langchain 구조를 분석하기 위해, github clone 후 UML로 모델링해 본다. 주요 패키지는 다음과 같다.

cli는 langchain의 command line interface, core는 langchain의 핵심 구현 코드가 정의된다. 이 부분은 다음 패키지로 구성된다.

참고로, 이 패키지들은 다음 그림의 일부이다.

Langchain v.0.2.0 패키지

LCEL 언어 동작 구조

이 중에 핵심적인 것만 분석해 본다. 우선, LCEL의 동작 방식을 위해 어떤 디자인패턴을 구현하였는 지 확인한다. 이 부분은 runnables 패키지가 담당한다. 이 언어는 유닉스의 파이프라인 처리를 다음과 같이 흉내낸다.

z = a | b | c

z.stream('abc')

이를 위해, 파이썬 문법을 적극 사용하고 있다. 우선 '|' 연산자를 오버로딩(overloading)하기 위해, 파이썬 Runnable 클래스를 정의해 __or__ 연산자를 구현한다. 이 연산자는 self object와 right object 두 객체를 입력받아 list를 만든 후 리턴하는 역할을 한다. 앞의 예시에서 보면, 'a | b'를 실행가능한 객체 리스트로 만들어 리턴한다. 결론적으로 a, b, c객체를 리스트로 만들고, 이 리스트를 z에 할당한다.

z의 stream()이 호출되면, a, b, c를 각각 invoke() 하여 실행하는 방식이다. 그러므로, 다음 그림과 같이, LCEL를 지원하는 객체는 Runnable을 파생받아야 하며, stream, invoke와 같은 주요 함수를 구현해야 한다.

각 주요 클래스 함수의 동작방식을 좀 더 자세히 살펴본다. LCEL 파이프라인을 구현하는 __or__()함수는 다음과 같이 입력된 객체를 Sequence 형식의 steps 리스트로 담아두는 역할을 한다.

class Runnable():

...

def __or__(self, other: Union[Runnable[Any, Other], Callable[[Any], Other], ...]):

return RunnableSequence(self, coerce_to_runnable(other))

def stream(self, input: Input, config, **kwargs: Optional[Any]) -> Iterator[Output]:

yield self.invoke(input, config, **kwargs) # steam 호출 시 invoke 함수 호출

def invoke(self, input: Input, config: Optional[RunnableConfig] = None):

callback_manager = get_callback_manager_for_config(config)

run_manager = callback_manager.on_chain_start(dumpd(self), input)

for i, step in enumerate(self.steps):

input = step.invoke(input) # steps 리스트의 객체 invoke() 호출 후 input 갱신

invoke()가 호출되면, 담아둔 steps 리스트의 객체를 각각 invoke()한다. 이런 방식으로 파이프라인을 구현하고 있다.

프롬프트 템플릿 처리 방식

JSON과 같이 구조화된 출력을 생성할 때는 ...Template 이름을 가진 클래스가 create_template_from_message_type() 함수를 구현한다. 그 과정을 확인해 보자.

def create_template_from_message_type(message_type, template, template_format):

if message_type in ("human", "user"): # human, user 일 경우, human template 생성

message =HumanTemplate.from_template(template, template_format)

elif message_type in ("ai", "assistant"): # ai 일 경우, AI template 생성

message = AIMessagePromptTemplate.from_template(template, template_format)

elif message_type == "system":

message = SystemMessagePromptTemplate.from_template(...)

elif message_type == "placeholder": # 변수 형태로 출력할 경우, placeholder 생성

if isinstance(template, str):

var_name = template[1:-1]

message = MessagesPlaceholder(variable_name=var_name, optional=True)

elif len(template) == 2 and isinstance(template[1], bool):

var_name = var_name_wrapped[1:-1]

message = MessagesPlaceholder(variable_name=var_name, optional=is_optional)

프롬프트 템플릿에 따라 적절한 템플릿 객체를 생성해 처리한다. invoke() 했을 때, input에 대한 각각 적합한 프롬프트를 출력한다.

변수가 담긴 출력을 위해서는 메시지에서 변수를 예측해야 한다. 이 부분은 from_messages_and_schema()가 담당한다.

def from_messages_and_schema(messages, schema):

# 자동적으로 메시지로 부터 변수를 추정함

input_vars: Set[str] = set()

partial_vars: Dict[str, Any] = {}

for _message in _messages:

if isinstance(_message, MessagesPlaceholder) and _message.optional:

partial_vars[_message.variable_name] = []

elif isinstance(

_message, (BaseChatPromptTemplate, BaseMessagePromptTemplate)

input_vars.update(_message.input_variables) # 변수값 업데이트

출력 텍스트에서 구조화된 출력을 위해, 변수와 변수 유형을 검증해야 한다. 이 기능은 PyDantic을 이용해 처리한다.

앞서 정의된 클래스는 LLM에 입력할 프롬프트를 작성하고, 출력된 결과를 파싱하는 것에 초점을 맞추고 있다.

LLM 모델 호출과 임베딩 데이터베이스 처리

LLM 처리 부분은 llm.py, llms.py 소스에 구현되어 있다. Chain은 전문가가 질문에 답변하며 정보를 업데이트하는 연속된 질의를 담당한다. BaseLLM은 LLM 이 호출될 수 있도록 일반화한 클래스이다. invoke()가 호출되면 프롬프트, 입력과 함께 _agenerate()이 호출된다. 이 함수는 실제 설치된 라마와 같은 LLM혹은 OpenAI의 ChatGPT API를 호출해 출력을 생성하는 역할을 한다.

class LLMChain(Chain):

async def _agenerate(self, prompts: List[str], ..., **kwargs: Any) -> LLMResult:

class BaseLLM():

async def _agenerate(self, prompts: List[str], ..., **kwargs: Any) -> LLMResult:

return await run_in_executor(None, self._generate, prompts, stop, ..., **kwargs)

LangChain은 임베딩 벡터 데이터베이스를 지원함으로써 LlamaIndex와 함께 RAG를 처리한다. 이를 통해, 환각 현상을 줄여주고, 전문적인 응답이 가능하도록 한다. 이 부분은 VectorStore 기본 클래스에서 담당한다. LangChain은 Chroma 등 다양한 데이터베이스와 임베딩 벡터 간 유사도 계산을 지원한다. 벡터 유사도 검색을 통해, 질문의 배경이 되는 데이터를 미리 LLM에게 전달해준다.

다음은 임베딩 벡터 데이터베이스의 유사도 검색 함수 구현 일부를 보여준다.

def similarity_search_with_score_by_vector(self, embedding: List[float], k: int = 4, filter: Optional[dict] = None) -> List[Tuple[Document, float]]:

filter_condition = ""

if filter is not None: # 필터의 key, value, conditions을 얻어 SQL 문을 정의함

conditions = [

f"metadata->>{key!r} = {value!r}" for key, value in filter.items()

]

filter_condition = f"WHERE {' AND '.join(conditions)}"

# 쿼리문 생성

sql_query = f"""

SELECT *, l2_distance(embedding, :embedding) as distance

FROM {self.collection_name}

{filter_condition}

ORDER BY embedding <-> :embedding

LIMIT :k

"""

# 임베딩 검색 파라메터 설정

params = {"embedding": embedding, "k": k}

# 쿼리 질의 및 fetch 후 결과 획득

with self.engine.connect() as conn:

results: Sequence[Row] = conn.execute(text(sql_query), params).fetchall()

# 출력 형태로 반환

documents_with_scores = [

(

Document(

page_content=result.document,

metadata=result.metadata,

result.distance if self.embedding_function is not None else None,

)

for result in results

]

return documents_with_scores

임베딩 벡터 간 유사도 검색은 cosine 함수, kNN 등 다양한 것이 사용될 수 있다.

LLamaIndex 구조 분석

LLamaIndex(라마 인덱스)는 LLM 클라우드에서 원하는 LLM을 검색하여, 모델을 다운로드받고, 사용하기 전까지 필요한 단계를 자동화한 라이브러리이다. 사용자 데이터 소스를 LLM에 연결하는 방법을 손쉽게 할 수 있도록 한다.

LLamaIndex는 RAG 애플리케이션에 대한 벡터 임베딩을 로드, 준비하는 데 도움이 되는 데이터 커넥터, 인덱스 및 쿼리 엔진과 같은 여러 도구를 제공한다. LLamaIndex는 다음 그림과 같이 RAG에 대한 지식 베이스를 생성하고, 이를 LLM에게 전해주는 기능을 구현하고 있다.

LLM과 지식베이스 관계

예를 들어, LLamaIndex는 사용자 데이터인 pdf documents를 load, parse, index, query하는 API를 제공한다. 여기서 index는 임베딩 벡터로 데이터 변환을 지원함을 의미한다.

LlamaIndex와 RAG 처리 방식

다음은 langchain과 함께 사용한 llama_index의 동작 방식을 보여준다.

from langchain import Chain

from llama_index import LLaMAIndex

class ProcessingChain(Chain): # LLM 체인 생성

def __init__(self):

super().__init__()

def _call(self, inputs):

return {"output": f"Processed: {inputs['input']}"} # 이해를 위해, 간단한 LLM 출력 가정

processing_chain = ProcessingChain() # LLM 체인 생성

index = LLaMAIndex() # 라마 인덱스 생성

# 문서 정의. 예를 들어, PDF, Image등 임베딩 처리될 수 있는 것이면 어떤 데이터도 가능

documents = [

{"id": 1, "content": "LangChain is a framework for building applications with large language models."},

{"id": 2, "content": "LLaMAIndex is used for indexing and querying text data."}

]

# 문서들 인덱싱 처리

for doc in documents:

processed_content = processing_chain.run({"input": doc["content"]})["output"]

index.add_document(doc["id"], processed_content) # 인덱스로 추가

query = "What is LangChain?" # 질의함

results = index.query(query) # 결과 리턴

print("Query results:") # 결과 출력

for result in results:

print(result)

마무리

LangChain을 개발한 해리슨 체이스(Harrison Chase)은 머신러닝 프로그래머로 일하며, LLM을 사용할 때 불편한 점을 재빨리 개선한 사람이다. 그는 2022년 말 랭체인(LangChain)을 공동 설립했다. 그는 누구보다도 파이썬 언어를 잘 알고 있었던 것 같다. 앞서 분석한 대로, 파이썬의 일반화 구문을 적극 활용해 누구나 GPT-4와 같은 대규모 언어 모델로 구동되는 앱을 단 20줄의 코드로 개발할 수 있는 LangChain을 개발했다. 2023년 3월 1000만 달러의 시드 투자를 주도했고, 불과 몇 주 뒤 2000만 달러 규모의 투자를 진행하였다. 이때 이미 회사는 93k 팔로워를 보유하고 있었다. 대중 속의 그는 앤드류 응과 같은 AI 석학과 잘 소통하는 모습을 보인다.

이렇게 빠르게 성장한 LangChain에 대한 첫 번째 커밋은 프롬프트 템플릿을 위한 Python의 가벼운 래퍼 클래스로써 시작된 것이다. 그는 2022년 초에 Notion 채팅 구현 시 LLM만으로는 쉽지 않다는 것을 발견했다. 이 문제를 가만두지 않고, LangChain 라이브러리를 재빨리 개선했다.

현재 LangChain은 LLM과 더불어 전문적인 채팅, 에이전트 서비스, 정보 생성 서비스 개발에 필수적인 기술이 되었다.

레퍼런스