LangChain 실전 가이드 | LLM 애플리케이션 개발 프레임워크

작성 0xmote · 2026년 7월 4일 · 55분 읽기 · 수정 2026년 9월 27일 중급 가이드

이 글의 핵심

LLM API를 직접 호출하다 보면 프롬프트 조립, 대화 이력 관리, 외부 문서 검색을 매번 새로 짜게 됩니다. LangChain이 이를 Chain·Memory·Agent·Retriever라는 추상화로 어떻게 묶는지 보여 주고, 버퍼·윈도·요약 메모리 중 무엇을 고를지, 벡터 스토어를 로컬과 클라우드 중 어디에 둘지 판단하는 기준을 예제와 함께 정리합니다.

들어가며

LLM API를 직접 호출하는 코드는 처음엔 단순하지만, 프롬프트 조립, 대화 이력 보관, 외부 문서 검색, 도구 호출이 붙으면서 금방 비슷한 접착 코드가 반복됩니다. 이 글은 LangChain이 그 부분을 Chain·Memory·Agent·Retriever로 어떻게 나누는지 Python 예제로 따라가고, 메모리 종류와 벡터 스토어를 고르는 기준, 캐싱과 에러 처리, 그리고 LangChain 없이 직접 호출하는 편이 나은 경우까지 다룹니다. 이 글에서 다룰 내용:

LangChain 기본 개념과 설치
Chains: 여러 단계를 연결
Agents: 자율적으로 도구 사용
Memory: 대화 이력 관리
RAG: 문서 검색 기반 답변
Vector Stores: 임베딩과 유사도 검색
실전 예제

설치와 모델 초기화

설치

# Python
pip install langchain langchain-openai langchain-community
# 추가 패키지
pip install chromadb  # 벡터 스토어
pip install tiktoken  # 토큰 계산
pip install faiss-cpu  # 벡터 검색
# Node.js
npm install langchain @langchain/openai

기본 설정

개념: LangChain에서 LLM을 사용하려면 먼저 모델을 초기화해야 합니다. ChatOpenAI 클래스는 OpenAI의 GPT 모델을 LangChain에서 사용할 수 있게 래핑합니다. 주요 파라미터:

model: 사용할 모델 (gpt-4o, gpt-4o-mini 등)
temperature: 창의성 조절 (0=결정적, 1=창의적)
max_tokens: 최대 출력 토큰 수

import os
from langchain_openai import ChatOpenAI
# API 키 설정 (환경 변수 사용 권장)
os.environ["OPENAI_API_KEY"] = "sk-..."
# LLM 초기화
llm = ChatOpenAI(
    model="gpt-4o-mini",      # 모델 선택: gpt-4o-mini는 빠르고 저렴
    temperature=0.7,           # 0.0~1.0: 낮을수록 일관적, 높을수록 창의적
    max_tokens=1000            # 최대 출력 길이 제한
)
# 간단한 호출
response = llm.invoke("What is LangChain?")
print(response.content)
# 출력: "LangChain is a framework for developing applications powered by language models..."

실행 결과:

LangChain is a framework for developing applications powered by language models. 
It provides tools for prompt management, chains, agents, memory, and more.

주의사항:

API 키는 절대 코드에 하드코딩하지 말고 환경 변수나 .env 파일 사용
temperature=0은 일관된 답변이 필요할 때 (예: 데이터 추출)
temperature=0.7~1.0은 창의적 작업에 적합 (예: 글쓰기)

LCEL로 Chain 연결하기

기본 Chain

개념: Chain은 여러 컴포넌트를 파이프라인으로 연결하는 LangChain의 핵심 개념입니다. | 연산자(LCEL)를 사용하여 프롬프트 → LLM → 파서를 순차적으로 연결합니다. LCEL (LangChain Expression Language):

| 연산자로 컴포넌트를 체인처럼 연결
이전 단계의 출력이 자동으로 다음 단계의 입력이 됨
간결하고 읽기 쉬운 코드 동작 흐름:

프롬프트 템플릿: 입력 변수를 받아 프롬프트 생성
LLM: 프롬프트를 받아 응답 생성
출력 파서: LLM 응답을 원하는 형식으로 변환

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser
# 1. 프롬프트 템플릿 정의
prompt = ChatPromptTemplate.from_template(
    "Translate the following to {language}: {text}"
    # {language}, {text}는 실행 시 전달할 변수
)
# 2. LLM 초기화
llm = ChatOpenAI(model="gpt-4o-mini")
# 3. 출력 파서 (LLM 응답을 문자열로 변환)
output_parser = StrOutputParser()
# 4. Chain 구성 (LCEL - LangChain Expression Language)
# prompt → llm → output_parser 순서로 실행
chain = prompt | llm | output_parser
# 5. 실행
result = chain.invoke({
    "language": "Korean",      # 프롬프트의 {language}에 대입
    "text": "Hello, how are you?"  # 프롬프트의 {text}에 대입
})
print(result)  # "안녕하세요, 어떻게 지내세요?"

실행 결과:

안녕하세요, 어떻게 지내세요?

내부 동작:

prompt.invoke({"language": "Korean", "text": "Hello, how are you?"}) → "Translate the following to Korean: Hello, how are you?"
llm.invoke("Translate the following to Korean: Hello, how are you?") → AIMessage(content="안녕하세요, 어떻게 지내세요?")
output_parser.invoke(AIMessage(...)) → "안녕하세요, 어떻게 지내세요?"

실무 활용:

다국어 번역 API
텍스트 요약 파이프라인
데이터 변환 및 정제

Sequential Chain

개념: Sequential Chain은 여러 LLM 호출을 순차적으로 연결하는 파이프라인입니다. 각 Chain의 출력이 다음 Chain의 입력으로 자동 전달되어, 복잡한 작업을 단계별로 처리할 수 있습니다. 사용 시나리오:

콘텐츠 생성: 주제 → 개요 → 본문
데이터 분석: 수집 → 정제 → 분석 → 리포트
코드 리뷰: 분석 → 문제 발견 → 개선 제안 핵심 개념:
output_key: 각 Chain의 출력을 명명하여 다음 Chain에서 참조
input_variables: 최초 입력 변수
output_variables: 최종 출력 변수 (중간 결과 포함 가능)

from langchain.chains import LLMChain, SequentialChain
from langchain.prompts import PromptTemplate
# Chain 1: 주제 생성
topic_prompt = PromptTemplate(
    input_variables=[keyword],  # 입력: 키워드
    template="Generate a blog post topic about {keyword}"
)
topic_chain = LLMChain(
    llm=llm, 
    prompt=topic_prompt, 
    output_key="topic"  # 출력 키: 다음 Chain에서 {topic}으로 참조 가능
)
# Chain 2: 개요 작성
outline_prompt = PromptTemplate(
    input_variables=[topic],  # 입력: 이전 Chain의 "topic"
    template="Create an outline for: {topic}"
)
outline_chain = LLMChain(
    llm=llm, 
    prompt=outline_prompt, 
    output_key="outline"  # 출력 키: {outline}
)
# Chain 3: 본문 작성
content_prompt = PromptTemplate(
    input_variables=["topic", "outline"],  # 입력: 두 개의 이전 결과
    template="Write a blog post:\nTopic: {topic}\nOutline: {outline}"
)
content_chain = LLMChain(
    llm=llm, 
    prompt=content_prompt, 
    output_key="content"  # 출력 키: {content}
)
# Sequential Chain으로 연결
overall_chain = SequentialChain(
    chains=[topic_chain, outline_chain, content_chain],  # 실행 순서
    input_variables=[keyword],  # 최초 입력
    output_variables=["topic", "outline", "content"]  # 최종 출력 (중간 결과 포함)
)
# 실행
result = overall_chain.invoke({"keyword": "Python async programming"})
print(result["content"])

실행 결과 예시:

{
  "topic": "Mastering Python Async Programming: A Complete Guide",
  "outline": """
    1. Introduction to Asyncio
    2. Understanding Event Loop
    3. async/await Syntax
    4. Practical Examples
    5. Common Pitfalls
  """,
  "content": """
    # Mastering Python Async Programming
    
    Async programming in Python allows you to write concurrent code...
    [전체 블로그 글 내용]
  """
}

동작 흐름:

keyword="Python async programming" 입력
topic_chain 실행 → topic="Mastering Python Async Programming..."
outline_chain 실행 (topic 사용) → outline="1. Introduction..."
content_chain 실행 (topic + outline 사용) → content="# Mastering..." 비용 및 성능:

API 호출 횟수: 3회 (각 Chain마다 1회)
예상 토큰: 약 1,500~3,000 토큰 (모델에 따라 다름)
실행 시간: 약 5~15초 (네트워크 속도에 따라 다름) 주의사항:
중간 Chain 실패 시 전체 파이프라인 중단
output_key가 겹치면 덮어쓰기됨
각 Chain은 독립적으로 LLM 호출 (병렬 실행 불가) 최적화 팁:
temperature=0.7: 창의적 글쓰기에 적합
max_tokens 제한으로 비용 절감
캐싱으로 동일 입력 재사용 시 비용 절감

Prompt Template과 Few-Shot

기본 템플릿

from langchain.prompts import PromptTemplate
# 단순 템플릿
template = PromptTemplate(
    input_variables=[product],
    template="Write a product description for {product}"
)
prompt = template.format(product="iPhone 15")

Chat Prompt Template

from langchain.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate
system_template = "You are a {role} with {experience} years of experience."
system_message = SystemMessagePromptTemplate.from_template(system_template)
human_template = "{question}"
human_message = HumanMessagePromptTemplate.from_template(human_template)
chat_prompt = ChatPromptTemplate.from_messages([
    system_message,
    human_message
])
messages = chat_prompt.format_messages(
    role="Python developer",
    experience=10,
    question="How do I optimize database queries?"
)

Few-Shot Prompt Template

from langchain.prompts import FewShotPromptTemplate
examples = [
    {"input": "happy", "output": "sad"},
    {"input": "hot", "output": "cold"},
    {"input": "fast", "output": "slow"}
]
example_template = PromptTemplate(
    input_variables=["input", "output"],
    template="Input: {input}\nOutput: {output}"
)
few_shot_prompt = FewShotPromptTemplate(
    examples=examples,
    example_prompt=example_template,
    prefix="Find the antonym:",
    suffix="Input: {word}\nOutput:",
    input_variables=[word]
)
print(few_shot_prompt.format(word="big"))

Memory로 대화 이력 관리하기

ConversationBufferMemory

개념: ConversationBufferMemory는 대화 이력을 전체 저장하는 가장 기본적인 메모리입니다. 모든 메시지를 그대로 보관하여 LLM이 전체 대화 맥락을 파악할 수 있게 합니다. 장점:

전체 대화 맥락 유지
구현이 간단
정보 손실 없음 단점:
대화가 길어지면 토큰 비용 증가
컨텍스트 길이 제한에 걸릴 수 있음 사용 시나리오:
짧은 대화 (10~20턴 이내)
전체 맥락이 중요한 상담 챗봇
디버깅 및 테스트

from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
# 메모리 초기화
memory = ConversationBufferMemory()
# 대화 체인 생성
conversation = ConversationChain(
    llm=llm,
    memory=memory,      # 메모리 연결
    verbose=True        # 내부 동작 로그 출력
)
# 첫 번째 대화
print(conversation.predict(input="Hi, I'm John"))
# 출력: "Hello John! How can I help you today?"
# 두 번째 대화 (이전 대화 기억)
print(conversation.predict(input="What's my name?"))
# 출력: "Your name is John."
# 메모리 내용 확인
print(memory.load_memory_variables({}))
# 출력: {
#   "history": "Human: Hi, I'm John\nAI: Hello John! How can I help you today?\nHuman: What's my name?\nAI: Your name is John."
# }

실행 결과:

첫 번째 응답: Hello John! How can I help you today?
두 번째 응답: Your name is John.
메모리 내용:
{
  "history": """
    Human: Hi, I'm John
    AI: Hello John! How can I help you today?
    Human: What's my name?
    AI: Your name is John.
  """
}

내부 동작:

사용자 입력 → 메모리에 저장
전체 대화 이력 + 새 입력 → LLM에 전달
LLM 응답 → 메모리에 저장
응답 반환 토큰 사용량:

1턴: ~50 토큰
5턴: ~250 토큰
10턴: ~500 토큰
주의: 대화가 길어질수록 비용 증가

ConversationBufferWindowMemory

개념: ConversationBufferWindowMemory는 최근 N개 메시지만 유지하는 슬라이딩 윈도우 방식의 메모리입니다. 오래된 대화는 자동으로 삭제되어 토큰 비용을 절감합니다. 장점:

토큰 비용 예측 가능 (일정하게 유지)
긴 대화에서도 컨텍스트 길이 제한 회피
최근 맥락에 집중 단점:
오래된 정보 손실
대화 초반 내용 참조 불가 사용 시나리오:
긴 대화 (50턴 이상)
최근 맥락만 중요한 경우
비용 최적화가 필요한 경우

from langchain.memory import ConversationBufferWindowMemory
# 최근 5개 메시지만 유지 (5턴 = Human 5개 + AI 5개)
memory = ConversationBufferWindowMemory(k=5)
conversation = ConversationChain(
    llm=llm,
    memory=memory
)
# 예시: 10번 대화 후
# 메모리에는 최근 5턴만 남음 (6~10번째 대화)
# 1~5번째 대화는 자동 삭제됨

실행 예시:

# 1번째 대화
conversation.predict(input="My name is John")
# 2번째 대화
conversation.predict(input="I live in Seoul")
# 3번째 대화
conversation.predict(input="I like Python")
# 4번째 대화
conversation.predict(input="I'm 30 years old")
# 5번째 대화
conversation.predict(input="I work as a developer")
# 6번째 대화
conversation.predict(input="What's my name?")
# 출력: "I don't have that information in our recent conversation."
# 이유: 1번째 대화가 삭제되어 이름 정보 손실

k 값 선택 가이드:

k=3: 매우 짧은 맥락 (비용 최소화)
k=5: 일반적인 대화 (권장)
k=10: 긴 맥락 유지
k=20: 매우 긴 맥락 (비용 증가) 토큰 사용량:
k=5: 약 250 토큰 (일정)
k=10: 약 500 토큰 (일정)
장점: 대화가 100턴이어도 토큰은 일정

ConversationSummaryMemory

개념: ConversationSummaryMemory는 대화 이력을 LLM으로 요약하여 저장합니다. 전체 대화를 압축하여 토큰을 절약하면서도 핵심 정보를 유지합니다. 장점:

긴 대화에서도 토큰 절약
핵심 정보 유지
오래된 대화도 요약 형태로 보존 단점:
요약 과정에서 추가 LLM 호출 (비용 발생)
세부 정보 손실 가능
요약 품질이 LLM 성능에 의존 사용 시나리오:
매우 긴 대화 (100턴 이상)
전체 맥락이 필요하지만 토큰 절약이 중요한 경우
고객 상담 이력 요약 동작 원리:

새 메시지 추가
일정 길이 초과 시 자동 요약
요약본 + 최근 메시지 유지

from langchain.memory import ConversationSummaryMemory
# 대화를 요약하여 저장 (토큰 절약)
memory = ConversationSummaryMemory(llm=llm)
conversation = ConversationChain(
    llm=llm,
    memory=memory
)
# 예시 대화
conversation.predict(input="My name is John and I live in Seoul")
conversation.predict(input="I'm a Python developer with 5 years of experience")
conversation.predict(input="I'm working on a machine learning project")
# ... 많은 대화 ...
# 메모리 확인
print(memory.load_memory_variables({}))
# 출력: {
#   "history": "John is a Python developer with 5 years of experience, 
#               living in Seoul and working on a machine learning project."
# }

실행 결과 예시:

원본 대화 (200 토큰):
Human: My name is John and I live in Seoul
AI: Nice to meet you, John!
Human: I'm a Python developer with 5 years of experience
AI: That's impressive!
Human: I'm working on a machine learning project
AI: Sounds exciting!
요약 (50 토큰):
"John is a Python developer with 5 years of experience, 
living in Seoul and working on a machine learning project."

비용 분석:

요약 없이: 10턴 = 500 토큰
요약 사용: 10턴 = 100 토큰 (요약본) + 50 토큰 (요약 비용)
절감: 이 예시 숫자로는 약 70% 토큰 절약

단, 요약을 만드는 호출도 지금까지의 대화를 입력으로 받으므로 그 입력 토큰까지 비용에 들어갑니다. 대화가 짧으면 요약 비용이 절감분을 넘을 수 있고, 대화가 길어질수록 매 턴 전체 기록을 다시 보내는 방식과의 차이가 커집니다. 언제 사용할까?:

적합: 대화가 길어지는 경우(예: 20턴 이상), 전체 맥락이 중요한 경우, 토큰 비용을 줄여야 하는 경우
부적합: 요약 과정에서 빠지면 안 되는 세부 정보가 중요한 경우 (예: 법률 상담)

Agent와 Custom Tool

기본 Agent

개념: Agent는 LLM이 자율적으로 도구를 선택하고 사용하는 시스템입니다. 사용자 질문을 분석하여 어떤 도구를 어떤 순서로 사용할지 스스로 결정합니다. ReAct 패턴 (Reasoning + Acting):

Thought: 무엇을 해야 할지 생각
Action: 도구 선택 및 실행
Observation: 도구 실행 결과 확인
반복: 최종 답변까지 반복 Agent 타입:

ZERO_SHOT_REACT_DESCRIPTION: 도구 설명만 보고 사용 (가장 일반적)
CONVERSATIONAL_REACT_DESCRIPTION: 대화 이력 포함
STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION: 복잡한 입력 처리 사용 시나리오:
복잡한 질문 (여러 단계 필요)
외부 API 호출 (검색, 계산, 데이터베이스)
동적 워크플로우 (미리 정의할 수 없는 경우)

from langchain.agents import initialize_agent, AgentType, Tool
from langchain_openai import ChatOpenAI
# 도구 1: Wikipedia 검색
def search_wikipedia(query):
    # 실제로는 Wikipedia API 호출
    return f"Wikipedia result for: {query}"
# 도구 2: 계산기
def calculate(expression):
    try:
        return str(eval(expression))  # 실무에서는 ast.literal_eval 사용 권장
    except:
        return "Invalid expression"
# 도구 목록 정의
tools = [
    Tool(
        name="Wikipedia",  # 도구 이름 (Agent가 참조)
        func=search_wikipedia,  # 실행할 함수
        description="Search Wikipedia for information"  # LLM이 이 설명을 보고 도구 선택
    ),
    Tool(
        name="Calculator",
        func=calculate,
        description="Calculate mathematical expressions. Input should be a valid Python expression."
    )
]
# Agent 초기화
agent = initialize_agent(
    tools=tools,  # 사용 가능한 도구 목록
    llm=ChatOpenAI(model="gpt-4o-mini", temperature=0),  # temperature=0: 일관된 도구 선택
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,  # ReAct 패턴 사용
    verbose=True  # 내부 사고 과정 출력
)
# 실행
result = agent.run("What is the population of Seoul multiplied by 2?")

실행 결과 (verbose=True):

> Entering new AgentExecutor chain...
Thought: I need to find the population of Seoul first, then multiply it by 2.
Action: Wikipedia
Action Input: "Seoul population"
Observation: Seoul has a population of approximately 9.7 million people.
Thought: Now I need to multiply 9.7 million by 2.
Action: Calculator
Action Input: "9.7 * 2"
Observation: 19.4
Thought: I now know the final answer.
Final Answer: The population of Seoul multiplied by 2 is approximately 19.4 million.
> Finished chain.

내부 동작 흐름:

질문 분석: “서울 인구 × 2”
계획 수립: Wikipedia 검색 → 계산
Wikipedia 도구 실행: “Seoul population” 검색
결과 확인: 9.7 million
Calculator 도구 실행: “9.7 * 2” 계산
결과 확인: 19.4
최종 답변: “19.4 million” 주의사항:

temperature=0 권장 (일관된 도구 선택)
도구 description이 매우 중요 (LLM이 이를 보고 선택)
무한 루프 방지: max_iterations 설정 가능
비용: 여러 번 LLM 호출 (Thought마다 1회) 실무 팁:
도구 개수는 5~10개 이내 권장 (너무 많으면 선택 어려움)
도구 이름은 명확하고 직관적으로
description에 입력 형식 명시

Custom Tool

from langchain.tools import BaseTool
from typing import Optional
class WeatherTool(BaseTool):
    name = "weather"
    description = "Get current weather for a location"
    
    def _run(self, location: str) -> str:
        # 실제 날씨 API 호출
        return f"Weather in {location}: Sunny, 20°C"
    
    async def _arun(self, location: str) -> str:
        # 비동기 버전
        return self._run(location)
tools = [WeatherTool()]
agent = initialize_agent(tools, llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION)

RAG: 문서 로딩, 분할, 벡터 스토어, 검색

문서 로딩

from langchain.document_loaders import TextLoader, PyPDFLoader, WebBaseLoader
# 텍스트 파일
loader = TextLoader("document.txt")
documents = loader.load()
# PDF
loader = PyPDFLoader("document.pdf")
documents = loader.load()
# 웹 페이지
loader = WebBaseLoader("https://example.com")
documents = loader.load()

문서 분할

개념: 문서 분할(Chunking)은 긴 문서를 작은 조각으로 나누는 과정입니다. LLM의 컨텍스트 길이 제한과 검색 정확도를 위해 필수적입니다. 왜 필요한가?:

LLM 컨텍스트 길이 제한 (예: 8K, 128K 토큰)
검색 정확도 향상 (작은 청크가 더 정확)
비용 절감 (관련 부분만 LLM에 전달) 주요 파라미터:
chunk_size: 각 청크의 최대 크기 (문자 수)
chunk_overlap: 청크 간 겹치는 부분 (문맥 유지)
length_function: 길이 측정 함수 (보통 len 또는 토큰 카운터) RecursiveCharacterTextSplitter의 동작:

먼저 \n\n (문단)으로 분할 시도
안 되면 \n (줄)으로 분할
안 되면 . (문장)으로 분할
안 되면 공백으로 분할
최후에는 문자 단위로 분할

from langchain.text_splitter import RecursiveCharacterTextSplitter
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,        # 각 청크 최대 1000자
    chunk_overlap=200,      # 청크 간 200자 겹침 (문맥 유지)
    length_function=len     # 길이 측정: 문자 수 기준
)
# 문서 분할
chunks = text_splitter.split_documents(documents)
print(f"Split into {len(chunks)} chunks")
# 예시 출력
# Split into 45 chunks

실행 예시:

# 원본 문서 (3000자)
document = """
LangChain is a framework for developing applications powered by language models.
[... 2900자 더 ...]
"""
# 분할 결과
chunks = text_splitter.split_text(document)
# chunks[0]: "LangChain is a framework... (1000자)"
# chunks[1]: "... framework for developing... (1000자)" (200자 겹침)
# chunks[2]: "... applications powered... (1000자)" (200자 겹침)

chunk_overlap의 중요성:

청크 1: "LangChain은 LLM 애플리케이션 개발 프레임워크입니다."
청크 2 (overlap 없음): "주요 기능은 Chains, Agents, Memory입니다."
→ 문맥 단절!
청크 1: "LangChain은 LLM 애플리케이션 개발 프레임워크입니다."
청크 2 (overlap 200): "...개발 프레임워크입니다. 주요 기능은 Chains, Agents, Memory입니다."
→ 문맥 유지!

chunk_size 선택 가이드:

500자: 짧은 문서, 정확한 검색
1000자: 일반적인 경우 (권장)
2000자: 긴 문맥이 필요한 경우
4000자: 매우 긴 문맥 (비용 증가) chunk_overlap 선택 가이드:
0: 문맥 불필요 (예: 독립적인 Q&A)
100~200: 일반적인 경우 (권장)
chunk_size의 10~20%: 경험적 최적값 실무 팁:
토큰 기반 분할: tiktoken으로 정확한 토큰 수 측정
문서 타입별 최적화: PDF, HTML, 코드마다 다른 전략
메타데이터 유지: 청크에 원본 파일명, 페이지 번호 등 포함

벡터 스토어 생성

개념: 벡터 스토어는 텍스트를 벡터(숫자 배열)로 변환하여 저장하며, 유사도 검색을 가능하게 하는 데이터베이스입니다. 동작 원리:

임베딩: 텍스트 → 벡터 (예: 1536차원 배열)
저장: 벡터를 데이터베이스에 저장
검색: 질문을 벡터로 변환 → 유사한 벡터 찾기 임베딩이란?:

텍스트의 의미를 숫자로 표현
비슷한 의미 = 비슷한 벡터
예: “dog” ≈ “puppy” (벡터 거리가 가까움) Chroma의 특징:
로컬 파일 기반 (설치 간단)
빠른 검색 속도
메타데이터 필터링 지원
영구 저장 가능

from langchain_openai import OpenAIEmbeddings
from langchain.vectorstores import Chroma
# 임베딩 모델 초기화
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small"  # OpenAI 임베딩 모델
    # 1536차원 벡터 생성
    # 비용: $0.00002 / 1K 토큰
)
# 벡터 스토어 생성
vectorstore = Chroma.from_documents(
    documents=chunks,           # 분할된 문서 청크
    embedding=embeddings,       # 임베딩 모델
    persist_directory="./chroma_db"  # 저장 경로 (영구 저장)
)
# 저장 후 로드
# vectorstore = Chroma(
#     persist_directory="./chroma_db",
#     embedding_function=embeddings
# )

실행 과정:

1. 청크 45개 입력
   ↓
2. 각 청크를 임베딩 (45번 API 호출)
   청크 1: "LangChain is..." → [0.12, -0.34, 0.56, ...] (1536차원)
   청크 2: "Chains allow..." → [0.23, -0.11, 0.78, ...]
   ...
   ↓
3. 벡터 + 원본 텍스트를 Chroma DB에 저장
   ↓
4. 디스크에 영구 저장 (./chroma_db/)

비용 계산:

청크 45개 × 평균 500자 = 22,500자 ≈ 5,600 토큰
임베딩 비용: 5,600 토큰 × $0.00002 / 1K = $0.00011 (약 0.15원)

임베딩 모델 비교:

모델	차원	비용	성능
text-embedding-3-small	1536	$0.00002/1K	빠름, 저렴
text-embedding-3-large	3072	$0.00013/1K	느림, 고성능
text-embedding-ada-002	1536	$0.00010/1K	구버전

persist_directory의 역할:

# 첫 실행: 벡터 스토어 생성 및 저장
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)
# 이후 실행: 저장된 벡터 스토어 로드 (임베딩 재생성 불필요)
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings
)

실무 팁:

대량 문서: 배치 임베딩으로 속도 향상
비용 절감: 캐싱으로 중복 임베딩 방지
메타데이터 활용: 필터링으로 검색 정확도 향상

RAG Chain

개념: RAG (Retrieval-Augmented Generation)는 검색 + 생성을 결합한 기법입니다. 질문과 관련된 문서를 먼저 검색하며, 그 문서를 참고하여 LLM이 답변을 생성합니다. RAG의 장점:

최신 정보 제공 (LLM 학습 데이터 이후)
환각(Hallucination) 감소
특정 도메인 지식 활용
출처 추적 가능 동작 흐름:

질문 임베딩: 질문을 벡터로 변환
유사도 검색: 벡터 스토어에서 관련 문서 검색
컨텍스트 구성: 검색된 문서를 프롬프트에 포함
답변 생성: LLM이 문서를 참고하여 답변 chain_type 옵션:

stuff: 모든 문서를 한 번에 전달 (가장 간단, 권장)
map_reduce: 각 문서 개별 처리 후 결합 (긴 문서)
refine: 순차적으로 답변 개선 (정교한 답변)
map_rerank: 각 문서별 점수 매겨 최고 선택

from langchain.chains import RetrievalQA
# RAG Chain 생성
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,                    # 답변 생성 LLM
    chain_type="stuff",         # 모든 문서를 한 번에 전달
    retriever=vectorstore.as_retriever(
        search_kwargs={"k": 3}  # 상위 3개 문서 검색
    )
)
# 질문
question = "What is the main topic of the document?"
answer = qa_chain.run(question)
print(answer)

실행 과정:

1. 질문: "What is the main topic of the document?"
   ↓
2. 질문 임베딩: [0.45, -0.23, 0.67, ...] (1536차원)
   ↓
3. 유사도 검색 (코사인 유사도):
   청크 12: 유사도 0.89 ✓
   청크 7:  유사도 0.85 ✓
   청크 23: 유사도 0.82 ✓
   ↓
4. 프롬프트 구성:
   "Based on the following context, answer the question.
   
   Context:
   [청크 12 내용]
   [청크 7 내용]
   [청크 23 내용]
   
   Question: What is the main topic of the document?
   Answer:"
   ↓
5. LLM 답변 생성:
   "The main topic of the document is LangChain, 
    a framework for building LLM applications..."

k 값 선택 가이드:

k=1: 가장 관련 있는 문서만 (빠름, 저렴)
k=3: 일반적인 경우 (권장)
k=5: 복잡한 질문 (느림, 비용 증가)
k=10: 매우 복잡한 질문 (컨텍스트 길이 주의) chain_type 비교:

# stuff: 모든 문서를 한 번에 전달 (가장 빠름)
# 프롬프트: Context: [doc1] [doc2] [doc3]\nQuestion: ...
# map_reduce: 각 문서 개별 처리 후 결합
# 1단계: 각 문서별 답변 생성 (3번 LLM 호출)
# 2단계: 답변들을 종합 (1번 LLM 호출)
# 총 4번 LLM 호출 (비용 증가)
# refine: 순차적으로 답변 개선
# 1단계: doc1로 답변 생성
# 2단계: doc2를 보고 답변 개선
# 3단계: doc3를 보고 답변 개선
# 총 3번 LLM 호출

실무 팁:

대부분의 경우 chain_type="stuff" 사용
k=3이 비용과 성능의 균형점
return_source_documents=True로 출처 확인 가능

고급 RAG

from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferMemory
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)
qa = ConversationalRetrievalChain.from_llm(
    llm=llm,
    retriever=vectorstore.as_retriever(),
    memory=memory
)
# 대화형 질문
print(qa({"question": "What is RAG?"}))
print(qa({"question": "Can you explain more about it?"}))

벡터 스토어: Chroma, FAISS, Pinecone

벡터 임베딩과 유사도 검색 메커니즘:

임베딩(Embedding)이란?
텍스트를 고차원 숫자 벡터로 변환하는 과정

예시:
"LangChain is awesome"
↓ OpenAI Embeddings API
[0.023, -0.145, 0.892, ..., 0.234]
1536개 차원 (text-embedding-3-small)

의미:
- 유사한 의미의 문장 → 가까운 벡터
- 다른 의미의 문장 → 먼 벡터

벡터 공간:
         "AI framework"
              ●
             / \
            /   \
"LangChain" ●   ● "Machine learning"
            \   /
             \ /
              ●
        "Cooking recipe" (멀리 떨어짐)

임베딩 모델 내부 (Transformer):

입력: "LangChain is awesome"
↓
Tokenization:
["Lang", "Chain", " is", " awesome"] → [1234, 5678, 910, 1112]
↓
Transformer Encoder:
- Self-Attention (각 단어 간 관계 학습)
- Feed-Forward 네트워크
- 12~24개 레이어 (모델 크기 의존)
↓
[CLS] 토큰의 Hidden State 추출
↓
Pooling (평균 또는 마지막 토큰)
↓
1536차원 벡터 출력

벡터 유사도 계산 (Cosine Similarity):

두 벡터 A, B:
A = [a1, a2, ..., an]
B = [b1, b2, ..., bn]

코사인 유사도:
similarity = (A · B) / (||A|| × ||B||)

A · B = a1*b1 + a2*b2 + ... + an*bn (내적)
||A|| = sqrt(a1² + a2² + ... + an²) (벡터 크기)

결과: -1 ~ 1
- 1: 완전히 같은 방향 (매우 유사)
- 0: 직각 (관련 없음)
- -1: 정반대 방향 (반대 의미)

예시:
"LangChain tutorial" vs "LangChain guide"
→ 유사도: 0.92 (매우 유사)

"LangChain tutorial" vs "Cooking recipe"
→ 유사도: 0.15 (거의 무관)

벡터 검색 과정:

1. 쿼리 임베딩:
   "How to use LangChain?"
   → [0.12, -0.34, 0.78, ...]

2. 모든 문서 벡터와 비교:
   Doc 1: similarity(query, doc1) = 0.89
   Doc 2: similarity(query, doc2) = 0.34
   Doc 3: similarity(query, doc3) = 0.91
   ...
   Doc 1000: similarity(query, doc1000) = 0.12

3. Top-k 선택:
   k=3이면:
   - Doc 3 (0.91)
   - Doc 1 (0.89)
   - Doc 8 (0.87)

4. 결과 반환

성능 최적화:

Brute Force:
- 모든 벡터와 비교: O(n)
- 1536차원 내적: 1536번 곱셈
- 100만 문서: 15억 번 연산

근사 최근접 이웃 (ANN):
HNSW (Hierarchical Navigable Small World):
- 그래프 기반 검색
- O(log n)
- 정확도(recall) 약간 손실: 가장 가까운 문서를 가끔 놓칠 수 있음
- 문서 수가 많을수록 전수 비교 대비 속도 이득이 커짐
- recall과 속도는 M, ef_search 같은 파라미터로 맞바꿈 (ef_search를 올리면 정확해지고 느려짐)

FAISS (Facebook AI Similarity Search):
- IVF (Inverted File Index) + PQ (Product Quantization)
- 벡터를 클러스터링
- 가까운 클러스터만 검색
- 메모리 압축

Chroma 내부:
- SQLite (메타데이터)
- HNSW (벡터 인덱스)
- 영구 저장 (persist_directory)

Chroma

from langchain.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings()

# 생성
vectorstore = Chroma.from_texts(
    texts=["LangChain is awesome", "Python is great"],
    embedding=embeddings,
    persist_directory="./db"
)

# 유사도 검색
results = vectorstore.similarity_search("LangChain", k=2)
for doc in results:
    print(doc.page_content)

FAISS

from langchain.vectorstores import FAISS
# 생성
vectorstore = FAISS.from_texts(
    texts=["text1", "text2"],
    embedding=embeddings
)
# 저장/로드
vectorstore.save_local("faiss_index")
vectorstore = FAISS.load_local("faiss_index", embeddings)

Pinecone (클라우드)

from langchain.vectorstores import Pinecone
import pinecone
pinecone.init(api_key="...", environment="...")
vectorstore = Pinecone.from_texts(
    texts=["text1", "text2"],
    embedding=embeddings,
    index_name="my-index"
)

문서 Q&A 챗봇, 코드 생성, 번역기 예제

예제 1: 문서 Q&A 챗봇

목표: 여러 문서를 읽고 질문에 답하는 대화형 챗봇을 만듭니다. 이전 대화를 기억하여 “그것에 대해 더 설명해줘” 같은 후속 질문도 처리합니다. 핵심 기능:

디렉토리 내 모든 문서 자동 로딩
RAG로 관련 문서 검색 후 답변
대화 이력 기억 (ConversationBufferMemory)
후속 질문 처리 가능 사용 시나리오:
사내 문서 Q&A 시스템
기술 문서 도우미
고객 지원 챗봇

from langchain.document_loaders import DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.vectorstores import Chroma
from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferMemory
class DocumentQABot:
    def __init__(self, docs_path):
        """
        문서 Q&A 챗봇 초기화
        
        Args:
            docs_path: 문서가 있는 디렉토리 경로
        """
        # 1. 문서 로딩
        loader = DirectoryLoader(
            docs_path, 
            glob="**/*.txt"  # 모든 하위 디렉토리의 .txt 파일
        )
        documents = loader.load()
        print(f"Loaded {len(documents)} documents")
        
        # 2. 문서 분할
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,      # 각 청크 1000자
            chunk_overlap=200     # 200자 겹침
        )
        chunks = text_splitter.split_documents(documents)
        print(f"Split into {len(chunks)} chunks")
        
        # 3. 벡터 스토어 생성
        embeddings = OpenAIEmbeddings()
        self.vectorstore = Chroma.from_documents(
            chunks, 
            embeddings, 
            persist_directory="./db"  # 영구 저장
        )
        print("Vector store created")
        
        # 4. LLM 및 메모리 초기화
        llm = ChatOpenAI(
            model="gpt-4o-mini", 
            temperature=0  # 일관된 답변
        )
        memory = ConversationBufferMemory(
            memory_key="chat_history",  # 메모리 키
            return_messages=True        # 메시지 형식으로 반환
        )
        
        # 5. 대화형 RAG Chain 생성
        self.qa_chain = ConversationalRetrievalChain.from_llm(
            llm=llm,
            retriever=self.vectorstore.as_retriever(
                search_kwargs={"k": 3}  # 상위 3개 문서 검색
            ),
            memory=memory  # 대화 이력 저장
        )
    
    def ask(self, question):
        """
        질문하기
        
        Args:
            question: 사용자 질문
            
        Returns:
            답변 문자열
        """
        response = self.qa_chain({"question": question})
        return response["answer"]
# 사용 예시
bot = DocumentQABot("./docs")
# 첫 번째 질문
print(bot.ask("What is the main topic?"))
# 출력: "The main topic is LangChain, a framework for..."
# 후속 질문 (이전 대화 기억)
print(bot.ask("Can you explain more?"))
# 출력: "LangChain provides several key features..."

실행 과정:

1. 초기화 (bot = DocumentQABot("./docs"))
   ├─ 문서 로딩: ./docs/**/*.txt
   ├─ 분할: 1000자 청크로 분할
   ├─ 임베딩: 각 청크를 벡터로 변환
   └─ 벡터 스토어 저장: ./db/
2. 첫 번째 질문 ("What is the main topic?")
   ├─ 질문 임베딩
   ├─ 유사도 검색: 상위 3개 청크
   ├─ 프롬프트 구성: 청크 + 질문
   ├─ LLM 답변 생성
   └─ 메모리 저장: 질문 + 답변
3. 두 번째 질문 ("Can you explain more?")
   ├─ 이전 대화 로드 (메모리)
   ├─ "it"이 "LangChain"을 가리킴을 이해
   ├─ 관련 문서 재검색
   └─ 답변 생성

메모리 내용 예시:

{
  "chat_history": [
    HumanMessage(content="What is the main topic?"),
    AIMessage(content="The main topic is LangChain..."),
    HumanMessage(content="Can you explain more?"),
    AIMessage(content="LangChain provides...")
  ]
}

확장 아이디어:

# 출처 문서 반환
# 함수 정의 및 구현
def ask_with_sources(self, question):
    response = self.qa_chain({"question": question})
    return {
        "answer": response["answer"],
        "sources": response.get("source_documents", [])
    }
# 스트리밍 답변
def ask_stream(self, question):
    for chunk in self.qa_chain.stream({"question": question}):
        yield chunk
# 메모리 초기화
def reset(self):
    self.qa_chain.memory.clear()

실무 팁:

문서 업데이트 시 벡터 스토어 재생성 필요
persist_directory로 재시작 시 빠른 로딩
대량 문서는 배치 처리로 속도 향상

예제 2: 코드 생성 Assistant

from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain.schema.output_parser import StrOutputParser
class CodeGenerator:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0.2)
        
        self.prompt = ChatPromptTemplate.from_messages([
            ("system", """You are an expert programmer.
            Generate clean, well-commented code following best practices.
            Include error handling and type hints."""),
            ("user", """Language: {language}
            Task: {task}
            
            Generate code:""")
        ])
        
        self.chain = self.prompt | self.llm | StrOutputParser()
    
    def generate(self, task, language="python"):
        return self.chain.invoke({
            "language": language,
            "task": task
        })
# 사용
generator = CodeGenerator()
code = generator.generate(
    "Read a CSV file and calculate column averages",
    "python"
)
print(code)

예제 3: 다국어 번역기

from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
class Translator:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o-mini")
        
        self.prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a professional translator."),
            ("user", "Translate to {target_lang}:\n\n{text}")
        ])
        
        self.chain = self.prompt | self.llm
    
    def translate(self, text, target_lang):
        response = self.chain.invoke({
            "text": text,
            "target_lang": target_lang
        })
        return response.content
# 사용
translator = Translator()
result = translator.translate("Hello, world!", "Korean")
print(result)  # "안녕하세요, 세계!"

커스텀 Retriever와 Reranking

커스텀 Retriever

from langchain.schema import BaseRetriever, Document
from typing import List
class CustomRetriever(BaseRetriever):
    def _get_relevant_documents(self, query: str) -> List[Document]:
        # 커스텀 검색 로직
        # 예: 데이터베이스 쿼리, API 호출 등
        results = search_my_database(query)
        return [Document(page_content=r) for r in results]
    
    async def _aget_relevant_documents(self, query: str) -> List[Document]:
        return self._get_relevant_documents(query)

Reranking

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
# 기본 retriever
base_retriever = vectorstore.as_retriever(search_kwargs={"k": 10})
# Compressor (관련성 높은 것만 선택)
compressor = LLMChainExtractor.from_llm(llm)
# Compression retriever
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=base_retriever
)
# 사용
docs = compression_retriever.get_relevant_documents("What is LangChain?")

토큰 계산과 캐싱으로 비용 줄이기

토큰 계산

개념: LLM API는 토큰 단위로 과금됩니다. 토큰은 단어보다 작은 단위로, 영어는 약 0.75 단어 = 1 토큰, 한글은 약 0.5 글자 = 1 토큰입니다. 왜 중요한가?:

비용 예측 및 관리
컨텍스트 길이 제한 확인
최적화 지점 발견 토큰 계산 방법:

tiktoken: OpenAI 공식 토큰 카운터
get_openai_callback: LangChain 내장 콜백

import tiktoken
def count_tokens(text, model="gpt-4o-mini"):
    """
    텍스트의 토큰 수 계산
    
    Args:
        text: 계산할 텍스트
        model: 모델 이름 (토큰화 방식이 모델마다 다름)
        
    Returns:
        토큰 수 (int)
    """
    encoding = tiktoken.encoding_for_model(model)
    return len(encoding.encode(text))
# 예시
text = "LangChain is a framework for building LLM applications."
tokens = count_tokens(text)
print(f"Tokens: {tokens}")  # 출력: Tokens: 11
# Chain에서 토큰 추적
from langchain.callbacks import get_openai_callback
with get_openai_callback() as cb:
    result = chain.invoke({"input": "Explain LangChain"})
    
    print(f"Total Tokens: {cb.total_tokens}")           # 총 토큰
    print(f"Prompt Tokens: {cb.prompt_tokens}")         # 입력 토큰
    print(f"Completion Tokens: {cb.completion_tokens}") # 출력 토큰
    print(f"Total Cost: ${cb.total_cost:.6f}")          # 총 비용

실행 결과 예시:

Tokens: 11
Total Tokens: 523
Prompt Tokens: 45
Completion Tokens: 478
Total Cost: $0.000785

토큰 계산 상세:

# 영어
text_en = "Hello, how are you?"
tokens_en = count_tokens(text_en)  # 약 6 토큰
# 비율: 5 단어 / 6 토큰 ≈ 0.83 단어/토큰
# 한글
text_ko = "안녕하세요, 어떻게 지내세요?"
tokens_ko = count_tokens(text_ko)  # 약 20 토큰
# 비율: 10 글자 / 20 토큰 ≈ 0.5 글자/토큰

비용 계산:

# GPT-4o-mini 가격 (2026년 기준)
# 입력: $0.15 / 1M 토큰
# 출력: $0.60 / 1M 토큰
prompt_tokens = 45
completion_tokens = 478
# 비용 계산
input_cost = (prompt_tokens / 1_000_000) * 0.15
output_cost = (completion_tokens / 1_000_000) * 0.60
total_cost = input_cost + output_cost
print(f"Input Cost: ${input_cost:.6f}")   # $0.000007
print(f"Output Cost: ${output_cost:.6f}") # $0.000287
print(f"Total Cost: ${total_cost:.6f}")   # $0.000294

get_openai_callback 활용:

# 여러 Chain 호출의 총 비용 추적
with get_openai_callback() as cb:
    # 여러 작업 수행
    result1 = chain1.invoke({"input": "..."})
    result2 = chain2.invoke({"input": "..."})
    result3 = chain3.invoke({"input": "..."})
    
    # 전체 통계
    print(f"Total Calls: {cb.successful_requests}")
    print(f"Total Tokens: {cb.total_tokens}")
    print(f"Total Cost: ${cb.total_cost:.6f}")

실무 비용 예시:

RAG 시스템 (1000 질문/일):
- 평균 프롬프트: 500 토큰 (검색된 문서 포함)
- 평균 답변: 200 토큰
- 일일 비용: 1000 × (500×0.15 + 200×0.60) / 1M = $0.195
- 월간 비용: $5.85

최적화 팁:

프롬프트 길이 최소화
max_tokens 제한 설정
캐싱으로 중복 호출 방지
저렴한 모델 우선 사용 (gpt-4o-mini)

캐싱

개념: 캐싱은 동일한 입력에 대한 LLM 응답을 저장하여, 재호출 시 API를 사용하지 않고 저장된 결과를 반환합니다. 장점:

비용 절감 (API 호출 0회)
속도 향상 (즉시 응답)
API 제한 회피 단점:
메모리/디스크 사용
캐시 무효화 관리 필요
동일 입력만 캐시 적중 캐시 타입:

InMemoryCache: 메모리에 저장 (빠름, 재시작 시 삭제)
SQLiteCache: 파일에 저장 (영구, 재시작 후에도 유지)
RedisCache: Redis에 저장 (분산 시스템)

from langchain.cache import InMemoryCache, SQLiteCache
from langchain.globals import set_llm_cache
# 옵션 1: 메모리 캐시 (빠르지만 재시작 시 삭제)
set_llm_cache(InMemoryCache())
# 옵션 2: SQLite 캐시 (영구 저장)
set_llm_cache(SQLiteCache(database_path=".langchain.db"))
# 사용 예시
llm.invoke("What is 2+2?")  # 첫 호출: API 호출 (네트워크 왕복 + 생성 시간)
llm.invoke("What is 2+2?")  # 두 번째: 캐시 반환 (로컬 조회만)

실행 결과 예시 (시간·비용은 설명용 수치):

첫 번째 호출:
- API 호출: ✓
- 응답 시간: 0.5초
- 비용: $0.0001
- 결과: "2+2 equals 4"
두 번째 호출 (동일 입력):
- API 호출: ✗ (캐시 적중)
- 응답 시간: 0.001초 (메모리/SQLite 조회만 하므로 API 호출보다 수백 배 빠른 수준)
- 비용: $0 (무료)
- 결과: "2+2 equals 4" (동일)

캐시 키 생성:

# 캐시 키는 프롬프트 + 모델 + 파라미터로 생성
# 예시:
# 입력 1: "What is 2+2?" + gpt-4o-mini + temp=0.7
# → 캐시 키: hash("What is 2+2?gpt-4o-mini0.7")
# 입력 2: "What is 2+2?" + gpt-4o-mini + temp=0.7
# → 동일 캐시 키 → 캐시 적중!
# 입력 3: "What is 2+2?" + gpt-4o-mini + temp=0.8
# → 다른 캐시 키 → 캐시 미스 (API 호출)

실무 활용 예시:

# FAQ 챗봇: 자주 묻는 질문 캐싱
set_llm_cache(SQLiteCache(database_path="faq_cache.db"))
faq_questions = [
    "What are your business hours?",
    "How do I reset my password?",
    "What is your return policy?"
]
# 첫 실행: 각 질문마다 API 호출
for q in faq_questions:
    answer = llm.invoke(q)  # API 호출
# 이후 실행: 모두 캐시에서 반환 (무료)
for q in faq_questions:
    answer = llm.invoke(q)  # 캐시 반환

비용 절감 효과:

시나리오: FAQ 챗봇 (10개 질문, 각 100회/일 반복)
캐싱 없음:
- 일일 호출: 10 × 100 = 1,000회
- 일일 비용: 1,000 × $0.0001 = $0.10
- 월간 비용: $3.00
캐싱 사용:
- 첫날 호출: 10회 (캐시 생성)
- 이후 호출: 0회 (캐시 사용)
- 일일 비용: $0.001 (첫날만)
- 월간 비용: $0.001
- 절감: 99.97%

캐시 무효화:

from langchain.cache import SQLiteCache
from langchain.globals import set_llm_cache, clear_llm_cache
# 캐시 설정
cache = SQLiteCache(database_path=".langchain.db")
set_llm_cache(cache)
# 캐시 초기화 (모든 캐시 삭제)
clear_llm_cache()
# 특정 캐시만 삭제 (수동)
import os
os.remove(".langchain.db")

주의사항:

시간 민감 정보: 날씨, 주가 등은 캐싱 부적합
개인화 답변: 사용자별 다른 답변은 캐싱 효과 낮음
캐시 크기: SQLite 파일이 커질 수 있음 (주기적 정리 필요) 최적 사용 사례:
적합: FAQ 시스템, 같은 문서를 반복 요약하는 경우, 같은 문장을 반복 번역하는 경우
부적합: 실시간 정보 조회, 매번 다른 결과가 필요한 창의적 글쓰기

Retry와 Fallback 설정

Retry 설정

from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
    model="gpt-4o-mini",
    max_retries=3,
    timeout=30
)

Fallback

from langchain.chat_models import ChatOpenAI
from langchain.llms import OpenAI
# 메인 모델
primary_llm = ChatOpenAI(model="gpt-4o")
# 폴백 모델 (저렴)
fallback_llm = ChatOpenAI(model="gpt-4o-mini")
# Fallback chain
llm = primary_llm.with_fallbacks([fallback_llm])

문서 분석 API 만들기

from fastapi import FastAPI, UploadFile, HTTPException
from langchain.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA
import tempfile
import os
app = FastAPI()
class DocumentAnalyzer:
    def __init__(self):
        self.embeddings = OpenAIEmbeddings()
        self.llm = ChatOpenAI(model="gpt-4o-mini")
        self.vectorstores = {}
    
    def process_document(self, file_path, doc_id):
        # 문서 로딩
        loader = PyPDFLoader(file_path)
        documents = loader.load()
        
        # 분할
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200
        )
        chunks = text_splitter.split_documents(documents)
        
        # 벡터 스토어 생성
        vectorstore = Chroma.from_documents(
            chunks,
            self.embeddings,
            persist_directory=f"./db/{doc_id}"
        )
        
        self.vectorstores[doc_id] = vectorstore
        return len(chunks)
    
    def query(self, doc_id, question):
        if doc_id not in self.vectorstores:
            raise ValueError("Document not found")
        
        qa_chain = RetrievalQA.from_chain_type(
            llm=self.llm,
            retriever=self.vectorstores[doc_id].as_retriever()
        )
        
        return qa_chain.run(question)
analyzer = DocumentAnalyzer()
@app.post("/upload")
async def upload_document(file: UploadFile):
    # 임시 파일로 저장
    with tempfile.NamedTemporaryFile(delete=False, suffix=".pdf") as tmp:
        content = await file.read()
        tmp.write(content)
        tmp_path = tmp.name
    
    try:
        doc_id = file.filename
        chunks = analyzer.process_document(tmp_path, doc_id)
        return {"doc_id": doc_id, "chunks": chunks}
    finally:
        os.unlink(tmp_path)
@app.post("/query")
async def query_document(doc_id: str, question: str):
    try:
        answer = analyzer.query(doc_id, question)
        return {"answer": answer}
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

LangChain vs 직접 API 호출

직접 API 호출이 나은 경우

# 간단한 작업
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Translate to Korean: Hello"}]
)

LangChain이 나은 경우

# 복잡한 워크플로우
chain = (
    load_documents
    | split_documents
    | create_embeddings
    | store_in_vectordb
    | retrieve_relevant
    | generate_answer
)

운영 전에 챙길 것: 키, 예외, 로깅, 테스트

.env로 API 키 관리

# ✅ .env 파일
OPENAI_API_KEY=sk-...
PINECONE_API_KEY=...
# ✅ 코드에서 로드
from dotenv import load_dotenv
load_dotenv()

Chain 호출 예외 처리

# ✅ 모든 Chain에 try-except
try:
    result = chain.invoke({"input": "..."})
except Exception as e:
    logger.error(f"Chain error: {e}")
    # 폴백 로직

verbose와 콜백으로 로깅

# ✅ Verbose 모드로 디버깅
chain = ConversationChain(llm=llm, verbose=True)
# ✅ 커스텀 콜백
from langchain.callbacks import StdOutCallbackHandler
chain.invoke({"input": "..."}, callbacks=[StdOutCallbackHandler()])

Chain 단위 테스트

# 단위 테스트
def test_chain():
    chain = create_my_chain()
    result = chain.invoke({"input": "test"})
    assert len(result) > 0

이 글의 핵심

들어가며

설치와 모델 초기화

설치

기본 설정

LCEL로 Chain 연결하기

기본 Chain

Sequential Chain

Prompt Template과 Few-Shot

기본 템플릿

Chat Prompt Template

Few-Shot Prompt Template

Memory로 대화 이력 관리하기

ConversationBufferMemory

ConversationBufferWindowMemory

ConversationSummaryMemory

Agent와 Custom Tool

기본 Agent

Custom Tool

RAG: 문서 로딩, 분할, 벡터 스토어, 검색

문서 로딩

문서 분할

벡터 스토어 생성

RAG Chain

고급 RAG

벡터 스토어: Chroma, FAISS, Pinecone

Chroma

FAISS

Pinecone (클라우드)

문서 Q&A 챗봇, 코드 생성, 번역기 예제

예제 1: 문서 Q&A 챗봇

예제 2: 코드 생성 Assistant

예제 3: 다국어 번역기

커스텀 Retriever와 Reranking

커스텀 Retriever

Reranking

토큰 계산과 캐싱으로 비용 줄이기

토큰 계산

캐싱

Retry와 Fallback 설정

Retry 설정

Fallback

문서 분석 API 만들기

LangChain vs 직접 API 호출

직접 API 호출이 나은 경우

LangChain이 나은 경우

운영 전에 챙길 것: 키, 예외, 로깅, 테스트

.env로 API 키 관리

Chain 호출 예외 처리

verbose와 콜백으로 로깅

Chain 단위 테스트

참고 자료

같이 보면 좋은 글