[RAG] LCEL(랭체인 표현 언어)과 Runnable을 이용한 RAG 파이프라인

1. 도입부 (Introduction)

과거에는 RAG 파이프라인을 구축하기 위해 텍스트 추출, 임베딩 변환, 벡터 저장소 쿼리, 프롬프트 조립, LLM 호출, 응답 파싱 등의 과정을 절차지향적인 스파게티 코드로 일일이 구현해야 했다. 이 과정에서 유지보수성은 최악으로 치달았고, 비동기 처리나 스트리밍을 지원하기 위해 수십 줄의 동기화 코드를 추가해야 하는 지옥이 펼쳐졌다.

이러한 고질적인 문제를 우아하게 해결하기 위해 등장한 혁신적인 도구가 바로 LCEL(LangChain Expression Language, 랭체인 표현 언어)이다. 리눅스의 파이프 연산자 기호(|)를 차용하여 선언적으로 컴포넌트들을 연결하는 LCEL은 가독성을 극대화할 뿐만 아니라, 비동기·스트리밍·병렬 처리를 엔진 수준에서 완벽하게 보장한다.

이번 글에서는 랭체인의 핵심 인터페이스인 Runnable의 동작 원리를 파헤치고, 실제 문서를 임베딩하여 로컬 벡터 저장소에 올린 뒤 Gradio를 통해 실시간 스트리밍 답변을 제공하는 고품질 RAG 파이프라인을 밑바닥부터 구현해 보겠다.

[RAG] LLM과 LangChain 프레임워크 기초 및 RAG 흐름 파악하기

 

 

2. 주요 특징 및 구조도 (Main Features & Architecture)

LCEL로 설계된 RAG 시스템은 사용자 질문 수신부터 답변 출력까지의 데이터 전이 과정이 매우 선형적이고 명확하게 통제된다. 이 흐름을 한눈에 조감할 수 있는 아키텍처 구조는 다음과 같다.

 

 

3. 상세 가이드 및 심층 분석 (Detailed Guide)

3.1 LCEL과 Runnable의 핵심 메커니즘

LCEL 파이프라인을 구성하는 모든 컴포넌트는 Runnable이라는 기본 인터페이스를 공유한다. 이 공유 인터페이스 덕분에 프롬프트 템플릿, LLM, 벡터 검색기, 출력 파서가 서로의 내부 규격을 알 필요 없이 마치 레고 블록처럼 유기적으로 결합될 수 있다.

Runnable 인터페이스가 보장하는 3대 핵심 실행 메서드는 다음과 같다:

1. invoke: 단일 입력에 대해 파이프라인을 동기적으로 실행하고 최종 결과를 일시 반환한다.
2. stream: 입력에 대한 응답을 생성되는 즉시 실시간 토큰 조각(chunk) 단위로 끊어서 제너레이터 형태로 전달한다.
3. batch: 대량의 입력 리스트를 수신하여 멀티스레드 기반으로 병렬 처리하고 각 응답의 리스트를 한 번에 반환한다.

 

3.2 RAG의 수문장: RunnablePassthrough와 RunnableParallel

체인을 연결할 때 가장 까다로운 부분은 프롬프트가 요구하는 입력 규격을 맞춰주는 작업이다. 예컨대 프롬프트 템플릿이 다음과 같이 선언되어 있다고 가정하자.

template = "이 맥락({context})을 참조해서 질문({question})에 답해줘."

이 템플릿에 데이터를 안전하게 밀어 넣으려면 최종 형태가 반드시 {"context": "...", "question": "..."} 형태의 딕셔너리 구조여야 한다. 하지만 사용자가 던지는 최초의 입력값은 단순 문자열 "가입 혜택이 뭐야?"에 불과하다.

이 간극을 메우는 장치가 바로 RunnableParallel과 RunnablePassthrough이다.

• RunnableParallel: 여러 연산을 병렬로 처리하여 하나의 딕셔너리로 결합해 주는 허브 역할을 한다.
• RunnablePassthrough: 들어온 입력값(여기서는 사용자의 질문 문자열)을 한 글자도 건드리지 않고 그대로 다음 컴포넌트로 흘려보내는 파이프 역할을 수행한다.

이 둘을 조합하면 다음과 같이 물 흐르듯 유연한 RAG 샌드박스가 완성된다.

retriever = vectorstore.as_retriever()
setup_and_retrieval = RunnableParallel(
    {"context": retriever, "question": RunnablePassthrough()}
)

이 설정 덕분에 사용자가 던진 하나의 질문 텍스트가 검색기의 데이터 인출용 쿼리로 활용됨과 동시에, 프롬프트의 {question} 슬롯으로 안전하게 동시 조달되는 이중 수송 체계가 구축된다.

 

3.3 실전 RAG 및 Gradio 스트리밍 파이프라인 전체 구현 코드

다음은 실무 환경과 주피터 노트북 스펙을 모두 만족하는 완전하고 실행 가능한 단일 파이썬 스크립트다. 임베딩 가동부터 로컬 Chroma DB 구축, LCEL 파이프라인 빌드, 그리고 Gradio 웹 인터페이스 구동까지 올인원으로 작성되었다.

import os
import time
from dotenv import load_dotenv

# 랭체인 코어 및 환경 컴포넌트 임포트
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableParallel
import gradio as gr

# [1] 환경 설정: .env 파일에 적힌 OPENAI_API_KEY 로드
load_dotenv()

# 실습을 위한 가상 문서 파일 생성 (프로젝트 정책 가이드라인)
sample_doc_path = "company_policy.txt"
with open(sample_doc_path, "w", encoding="utf-8") as f:
    f.write("""
    [인천 지사 신규 가입 혜택 정책]
    일시: 2026년 6월 중 가입한 신규 회원 대상
    혜택 내용: 가입 즉시 사용 가능한 10,000 마일리지 적립 포인트 지급.
    유효기간: 지급일로부터 90일간 유효하며, 미사용 시 자동 소멸됨.
    제한 사항: 다른 쿠폰 및 할인 혜택과 중복 사용 불가함.
    """)

try:
    # [2] 문서 로더 가동: 텍스트 파일을 메모리로 로드
    loader = TextLoader(sample_doc_path, encoding="utf-8")
    raw_documents = loader.load()

    # [3] 텍스트 분할기(Splitter) 적용: 대용량 문서를 토큰 및 의미 단위로 쪼갬
    # chunk_size는 분할할 최대 글자 수이며, chunk_overlap은 분할 경계면의 문맥 소실을 막기 위해 겹치는 글자 수다.
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=100, 
        chunk_overlap=20
    )
    splitted_docs = text_splitter.split_documents(raw_documents)

    # [4] 임베딩 모델 인스턴스화: 텍스트를 고차원 밀집 벡터로 변환하는 엔진
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

    # [5] 벡터 데이터베이스(Vector Store) 구축: 쪼개진 문서 조각들을 벡터로 변환하여 Chroma DB 메모리 영역에 영속화
    # 실무 시나리오에 맞춰 영속성 데이터 저장 및 리트리버 변환 처리
    vectorstore = Chroma.from_documents(
        documents=splitted_docs, 
        embedding=embeddings
    )
    
    # 벡터 스토어를 검색기(Retriever) 모드로 전환
    # search_kwargs={"k": 2}는 가장 연관성이 높은 문서 상위 2개를 가져오겠다는 옵션이다.
    retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

    # [6] 프롬프트 템플릿(Prompt Template) 디자인: 시스템과 사용자의 변수 매핑 레이아웃 확정
    prompt_template = ChatPromptTemplate.from_messages([
        ("system", "너는 철저히 주어진 [Context] 정보만을 기반으로 답변하는 전문 지식 도우미다. 관련 없는 주관적인 생각이나 뇌피셜은 답변에서 원천 배제하라."),
        ("user", "[Context]\n{context}\n\n[Question]\n{question}")
    ])

    # [7] LLM 모델 설정 및 출력 파서 인스턴스 구축
    # temperature=0은 일관되고 정확한 사실 위주의 답변을 얻기 위한 필수 설정이다.
    model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    output_parser = StrOutputParser()

    # [8] LCEL(LangChain Expression Language)을 활용한 최종 RAG 체인 오케스트레이션
    # 파이프 연산자(|)를 타고 데이터가 수평 방향으로 완벽히 전이된다.
    rag_chain = (
        RunnableParallel({
            "context": retriever, 
            "question": RunnablePassthrough()
        })
        | prompt_template
        | model
        | output_parser
    )

    # [9] Gradio용 실시간 토큰 스트리밍 핵심 인터페이스 함수 설계
    def answer_invoke(message, history):
        """
        사용자 메시지를 받아 RAG 체인을 가동하고 한 글자씩 실시간으로 전송하는 제너레이터 함수
        Gradio의 ChatInterface 규약에 맞게 history 매개변수를 열어둔다.
        """
        partial_message = ""
        
        # rag_chain.stream을 호출하여 청크 단위로 토큰을 비동기식으로 받아옴
        for chunk in rag_chain.stream(message):
            if chunk is not None:
                partial_message = partial_message + chunk
                # 사용자에게 타이핑 느낌을 주기 위해 미세 지연 부여
                time.sleep(0.05)
                # 현재까지 누적된 메시지 문자열을 점진적으로 반환(yield)
                yield partial_message

    # [10] Gradio 챗봇 UI 인터페이스 빌드 및 론치
    demo = gr.ChatInterface(
        fn=answer_invoke, 
        title="2026 사내 신규 가입 정책 QA Bot"
    )

    # 로컬 서버 실행 (포트 7860에서 챗봇 인터페이스 구동)
    if __name__ == "__main__":
        demo.launch()

except Exception as e:
    print(f"시스템 초기화 중 오류가 발생했습니다: {e}")

 

4. 실무 팁 및 주의사항 (Tips & Notes)

4.1 Chunk Size와 Overlap 설계의 황금비율

RAG 성능의 절반 이상은 문서 분할 단계에서 결정된다. Chunk Size를 너무 작게 잡으면 문장의 맥락이 조각조각 부서져 LLM이 문맥을 이해하기 어려워지고, 반대로 너무 크게 잡으면 쓸데없는 인접 텍스트까지 벡터 임베딩을 오염시켜 검색 신뢰도가 급격히 떨어진다.

• 실무 표준 권장치: 일반 규정이나 백서의 경우 Chunk Size 500~1000자, Chunk Overlap 50~150자 내외에서 시작하여 테스트 셋의 검색 성공률(Hit-rate)을 모니터링하며 점진적으로 튜닝하는 것이 가장 이상적이다.

 

4.2 Gradio 스트리밍 연동 시 흔히 저지르는 실수

주피터 노트북 환경이나 FastAPI 서빙 환경에서 스트리밍이 가동되지 않고 답변이 한꺼번에 툭 떨어지는 현상이 일어난다면 다음 두 가지를 검증해야 한다:

• Gradio 함수 내 yield 키워드 누락: Gradio의 gr.ChatInterface가 실시간 스트리밍을 구현하려면 바인딩된 타겟 함수가 반드시 return 대신 점진적 출력을 뱉어내는 yield 제너레이터 구조여야 한다.
• 출력 파서(StrOutputParser) 미장착: 체인 마지막 단계에 StrOutputParser()를 결합하지 않으면, 스트리밍 청크가 문자열이 아닌 복잡한 랭체인 코어 객체(예: AIMessageChunk) 형태로 수신되어 프론트엔드가 이를 문자열로 파싱하지 못하고 파이프가 정지할 수 있다.

 

5. 마무리 (Conclusion)

결국 고성능 AI 인프라 구축의 승부처는 "복잡하고 파편화되기 쉬운 데이터 처리 파이프라인을 얼마나 선언적이고 간결하게 제어하는가"에 달려 있다.

컴포넌트 간 결합도를 완전히 낮추고, 병렬 검색과 Passthrough 데이터를 아름답게 조율하여 단 한 줄의 | 파이프 체인으로 직렬화해 내는 LCEL 아키텍처는 현대 AI 소프트웨어 공학의 정수다. 여기에 Gradio 스트리밍 인터페이스까지 가볍게 도킹해 둘 때, 데이터 일관성이 보장되면서도 상용 수준의 민첩함을 선사하는 RAG 엔진이 비로소 완성된다.