LangChain

LangChain의 아키텍처 및 개요

모듈형 구조

langchain-core

• LangChain의 추상화와 기본 구성 요소를 가지고 있는 라이브러리로, 모든 LangChain 프로젝트의 기반이 됨
• 기본 데이터 구조: Prompt, Document, Message등 LLM 어플리케이션에서 공통적으로 사용되는 데이터 형식 정의
• 핵심 인터페이스: Runnable 인터페이스를 통해 모든 구성요소를 일관된 방식으로 연결, 실행할 수 있도록 함
• 기본 구성 요소 추상화: LLM, ChatModel, Retriever, Tool과 같은 구성 요소에 대한 추상 클래스 제공.
  • 개발자가 특정 구현에 얽매이지 않고 어플리케이션 로직에 집중 가능.

langchain

• langchain-core의 추상화를 실제로 구현하고, 여러 구성 요소를 조합하여 복잡한 어플리케이션을 만들 수 있는 기능 제공
• Chains: 여러 단계를 순차적으로 실행하는 로직의 묶음. 기본적인 LLMChain부터 특정 목적을 위해 여러 구성 요소를 결합한 복잡한 체인들도 포함.
• Agents: LLM이 어떤 Tool을 사용해야 할지 스스로 판단, 그 결과를 바탕으로 다음 행동을 결정하게 하는 로직. ReAct 프롬프팅 기법 등 활용.
• Memory: 대화의 맥락을 기억하고 관리하는 기능. 사용자와 LLM 간의 이전 상호작용을 저장하여, 보다 자연스러운 대화를 이어갈 수 있도록 함.

통합 패키지

• OpenAI, Google 등 주요 외부 서비스와의 연동을 위한 패키지.
• LangChain 팀 또는 파트너사가 직접 관리하며, 독립적인 버전 관리가 가능.
• 메인 패키지를 가볍게 유지하고, 사용자가 필요로 하는 외부 서비스 연동 기능만 선택적으로 추가할 수 있어, 의존성 관리가 용이
• ex) langchain-openai, langchain-anthropic, langchain-google-vertexai

langchain-community

• 커뮤니티가 유지보수하는 모든 서드파티(third-party) 통합 기능을 가지고 있음.
• 주요 통합 기능들은 별도의 패키지로 분리되었으며, 이 패키지에 포함된 의존성은 모두 선택 사항으로 설치되도록 하여 패키지 경량화를 유지

langgraph

• 에이전트(agent)를 그래프 형태로 모델링하여 구축하기 위한 강력한 확장 라이브러리
• 복잡하고 프로덕션 수준의 에이전트를 만들기 위한 AgentExecutor의 후속 기술

디자인 패턴

Template Method Pattern

• BaseLLM 추상 클래스는 LLM을 호출하는 흐름을 정의.
• 실제 특정 LLM과 통신하는 세부 로직(_call method )은 OpenAI 서브클래스에서 구체적으로 구현
• 이를 통해 개발자는 LLM 을 쉽게 교체하면서도 일관된 방식으로 모델 호출 가능.

Strategy Pattern

• 알고리즘군을 정의하고 각각을 캡슐화하여 필요에 따라 동적으로 교체
• BaseLLM, BaseChatModel 등으로 추상화 하여 구체적인 구현체(OpenAI, Ollama 등)가 런타임에 교체 가능
• ex) Retriever 의 경우, VectorStoreRetriever, EnsembleRetriever 등으로 선택하여 사용 가능

Command Pattern & Chain of Responsibility Pattern

• Command Pattern
  • 각 구성 요소(PromptTemplate, LLM, OutputParser 등)는 Runnable 인터페이스를 구현하며, 이는 실행 가능한 '커맨드(명령)'로 볼 수 있음.
  • 각 커맨드는 입력을 받아 특정 작업을 수행하고 결과를 반환
• Chain of Responsibility Pattern
  • LCEL의 파이프(|) 연산자는 이러한 커맨드들을 연결하여 처리 과정의 Chain을 생성.
  • 첫 번째 구성 요소의 출력이 다음 구성 요소의 입력으로 자동으로 전달되는 구조는, 요청이 연쇄적으로 처리되는 책임 연쇄 패턴과 유사
  • 이를 통해 개발자는 복잡한 데이터 처리 흐름을 간결하고 직관적인 코드로 구성 가능

Runnable 프로토콜과 LCEL(LangChain Expression Language)

Runnable 프로토콜

개념

• LangChain의 모든 계산 단위를 표현하는 인터페이스
• invoke, batch, stream 및 이들의 비동기 버전인 ainvoke, abattch, astream과 같은 표준 method를 구현
• Chain, Model, PromptTemplate, Retriever, OutputParser 등 LangChain의 거의 모든 핵심 구성 요소는 Runnable이라는 공통의 규칙(인터페이스)을 따르는 클래스
• Runnable 프로토콜을 따르는 것만으로도 병렬 처리 기능 활용 가능

batch : multi-threading

• 내부적으로 ThreadPoolExecutor 생성.
• executor.map()을 통해 여러 입력값을 각각의 thread에 할당하여 self.invoke()를 동시에 호출
• 여러 thread를 만들어 작업을 병렬로 처리하는게 핵심
• I/O-bound 작업에서 효과적.
  • 하나의 thread가 API 응답 대기상태가 되어도, 다른 thread가 다른 작업 처리 가능

abatch : 비동기 I/O 활용

• asyncio.gather(혹은 gather_with_concurrency)를 사용하여 여러개의 ainvoke() 를 하나의 thread 내에서 동시에 실행
• event loop를 사용하여 I/O 대기 시간에 다른 작업으로 전환(context switching)하며 처리.
• 많은 I/O 작업을 처리할 때 thread 생성 비용이 없어 ThreadPoolExecutor 방식보다 가볍고 효율적.

bind

• 특정 함수나 모델을 나중에 호출할 떄, 항상 똑같이 들어가는 인자가 있으면 미리 바인딩

# 'calculator'라는 도구를 LLM 모델에 바인딩
model_with_calculator = model.bind(tools=[calculator])

# 응답 format 설정을 모델에 미리 바인딩
json_model = model.bind(response_format={"type": "json_object"})

with_config

• Chain이나 Model이 실행될 때, 실행에 대한 메타데이터를 추가하고 싶을 때 사용
• 실행 기록을 추적, 분석하는데 유용

# 체인을 실행할 때, "user_123의 요청"이라는 태그를 붙임
chain.with_config(
    {"metadata": {"user_id": "user_123"}}
).invoke("오늘 날씨 어때?")

LCEL(LangChain Expression Language)

• 기존 Runnable들을 조합하여 새로운 Runnable을 만드는 선언적인 방법
• 파이프 연산자(|)로 RunnableSequence 생성

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_template("Tell me a joke about {topic}")
model = ChatOpenAI()
output_parser = StrOutputParser()

chain = prompt | model | output_parser

• Runnable 클래스에 오버로딩된 |연산자가 RunnableSequence 인스턴스를 생성.
• chain.invoke({"topic":"bears"})가 호출되면 아래 단계를 실행
  1. prompt.invoke({"topic":"bears"})가 호출되어 PromptValue 객체 반환
  2. PromptValue가 model.invoke()의 입력으로 들어가 AIMessage 객체를 반환
  3. AIMessage가 output_parser.invoke()의 입력으로 들어가 str 객체 반환

추가 기능

RunnableParallel

• 동일한 입력을 여러 Runnable에 동시에 제공하여 병렬로 실행하는 데 사용
• LCEL에서는 Runnable을 값으로 가지는 딕셔너리가 자동으로 RunnableParallel로 강제 변환(coercion)
• 검색 증강 생성(RAG, Retrieval-Augmented Generation) 파이프라인에서 매우 유용

from langchain_core.runnables import RunnablePassthrough, RunnableParallel
# retriever, prompt, model, StrOutputParser는 이미 정의되었다고 가정합니다.
retriever =...
prompt =...
model =...

chain = RunnableParallel(
    context=retriever,
    question=RunnablePassthrough()
) | prompt | model | StrOutputParser()

• chain.invoke("What is LCEL?")을 호출하면, RunnableParallel 단계가 먼저 실행
• retriever.invoke(...)와 RunnablePassthrough.invoke(...)를 병렬로 호출
  • RunnablePassthrough는 단순히 원본 입력을 그대로 통과시키는 역할
• {"context":, "question": "What is LCEL?"} 형태의 딕셔너리가 생성되고, 이 딕셔너리가 다음 단계인 prompt의 입력으로 전달

핵심 컴포넌트

모델

• BaseLLM, BaseChatModel 의 추상화 모델이 있음.
• 예를들어 BaseLLM 추상화 클래스를 OpenAI, Anthropic, Ollama 등의 LLM이 구현

데이터 연결

• Document Loaders : 파일, 웹 등 다양한 소스로부터 텍스트 콘텐츠와 소스 메타데이터를 읽어와 표준화된 Document 객체로 패키징
• 텍스트 분할기
  • CharacterTextSplitter: 가장 기본적인 분할기로, 단순히 텍스트를 순회하며 chunk_size를 초과할 때마다 지정된 separator를 기준으로 분할
  • RecursiveCharacterTextSplitter
    • ["\n\n", "\n", " ", ""]와 같이 의미론적 경계가 될 가능성이 높은 순서대로 구분자 리스트를 받음.
    • 첫 번째 구분자(단락을 의미하는 \n\n)로 텍스트를 분할. 만약 분할된 조각 중 여전히 chunk_size보다 큰 것이 있다면, 해당 조각에 대해서만 나머지 구분자 리스트(["\n", " ", ""])를 사용하여 재귀적으로 분할.
• TextSplitter의 선택과 그 파라미터(chunk_size, chunk_overlap)는 RAG 시스템의 품질에 직접적인 영향을 미치는 중요한 하이퍼파라미터
• chunk가 너무 크거나 너무 작거나 문장 중간에서 잘리면 의미가 크게 없음.
  • chunk_overlap으로 조각 경계에서 발생하는 문맥 손실 문제를 완화.

레거시 Chains

BaseChain

• LCEL 등장 이전 체인은 BaseChain을 상속받아 제작.
  1. _validate_inputs가 호출되어 self.input_keys에 명시된 모든 키가 입력값에 존재하는지 확인
  2. prep_inputs가 호출되어 self.memory로부터 데이터를 로드하는 등의 전처리 작업을 수행
  3. 추상 메소드인 self._call(inputs)가 실행
  4. _validate_outputs가 _call의 결과에 self.output_keys의 모든 키가 포함되어 있는지 확인
  5. prep_outputs가 호출되어 self.memory에 데이터를 저장하는 등의 후처리 작업을 수행

상태 관리 : 메모리

메모리

• 체인이나 에이전트의 호출 간에 상태를 유지하는 개념
• ConversationBufferMemory가 가장 일반적인 구현체 중 하나

BaseMemory 인터페이스

핵심 메소드

• load_memory_variables(self, inputs) -> dict
  • 체인 실행이 시작될 때 호출.
  • 내부 저장소에서 대화 기록을 검색하여 프롬프트에 주입될 딕셔너리(ex. {'history': 'Human:...\nAI:...'}) 형태로 포맷하여 반환
• save_context(self, inputs, outputs) -> None
  • 체인 실행이 끝날 때 호출
  • 사용자의 최신 입력과 AI의 최신 출력을 받아 내부 저장소에 저장
  • ex) self.chat_memory.add_user_message(...)

ConversationBufferMemory

대화 기록을 버퍼에 저장하고 이를 단일 문자열 또는 메시지 리스트로 노출

ConversationChain과의 상호작용

1. conversation.predict(input="Hi, my name is Andrew") 호출.
2. 체인의 __call__ 메소드가 시작되고, memory.load_memory_variables({})를 호출. 버퍼가 비어 있으므로 {'history': ''}를 반환.
3. 프롬프트는 비어있는 기록과 새로운 입력을 사용하여 포맷.
4. LLM이 "Hello Andrew..."를 생성.
5. __call__이 끝나면서 memory.save_context({"input": "Hi..."}, {"output": "Hello..."})를 호출. 메모리는 이 대화 턴을 저장.
6. conversation.predict(input="What is my name?") 호출.
7. __call__ 메소드가 다시 memory.load_memory_variables({})를 호출. 메모리는 {'history': 'Human: Hi, my name is Andrew\nAI: Hello Andrew...'}를 반환.
8. 프롬프트는 이 기록과 함께 포맷되어, LLM이 질문에 올바르게 답변할 수 있도록 문맥을 제공.

RunnableWithMessageHistory

• 현대적 메모리 관리
• LCEL에서는 RunnableWithMessageHistory로 체인을 감싸서 메모리를 관리
• 애플리케이션 로직과 상태 관리를 완벽하게 분리하는 우수한 추상화를 제공
• runnable, get_session_history 함수, 그리고 input_messages_key, history_messages_key와 같은 키들을 인자로 받음.

동작 (invoke 메소드)

• 런타임 config에서 session_id를 추출.
• get_session_history(session_id) 함수를 호출하여 해당 세션의 BaseChatMessageHistory 객체를 가져옴. 이 함수는 개발자가 정의하며, 메모리 저장소(인메모리 딕셔너리, Redis, 데이터베이스 등)와의 연결을 담당.
• 가져온 히스토리 객체에서 메시지를 로드.
• 이 메시지들을 history_messages_key 아래에 입력 딕셔너리로 주입.
• 이렇게 증강된 입력을 가지고 래핑된 runnable을 호출.
• 실행이 끝나면 새로운 입력과 최종 출력을 히스토리 객체에 추가하여 저장.

특징

• 레거시 체인에서는 체인 객체 자체가 메모리 객체에 대한 참조(self.memory)를 가지고 있어 둘이 강하게 결합
• 반면, RunnableWithMessageHistory를 사용하면 핵심 체인(prompt | model)은 메모리의 존재를 전혀 알지 못하는 순수한 상태 비저장(stateless) 함수가 됨
• RunnableWithMessageHistory는 이 상태 비저장 체인을 감싸는 "상태 저장 데코레이터"처럼 작동하여, 외부 소스로부터 상태(대화 기록)를 주입
• 개발자는 핵심 체인을 전혀 변경하지 않고 get_session_history 함수만 교체하여 인메모리 저장소에서 Redis나 데이터베이스 저장소로 쉽게 전환