LangChain 완벽 가이드: AI 애플리케이션 개발의 표준


LangChain이란?

LangChain은 대규모 언어 모델(LLM)을 활용한 애플리케이션을 쉽게 개발할 수 있게 해주는 프레임워크입니다. 복잡한 AI 기능을 모듈화된 컴포넌트로 제공하여, 마치 레고 블록을 조립하듯 AI 앱을 구축할 수 있습니다.

왜 LangChain이 필요한가?

LangChain 없이 개발:

# 직접 구현해야 하는 것들
- LLM API 호출 및 에러 처리
- 프롬프트 템플릿 관리
- 대화 히스토리 저장
- 벡터 DB 연동
- 문서 로딩 및 처리
- 응답 파싱
- 재시도 로직
# ... 수백 줄의 보일러플레이트 코드

LangChain으로 개발:

from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.chains import LLMChain

# 간결한 코드로 동일한 기능
llm = ChatOpenAI()
prompt = ChatPromptTemplate.from_template("Tell me about {topic}")
chain = LLMChain(llm=llm, prompt=prompt)
result = chain.run(topic="AI")

LangChain 핵심 구성요소

1. Models - LLM 통합

다양한 LLM을 통일된 인터페이스로 사용합니다.

# OpenAI
from langchain.chat_models import ChatOpenAI
llm = ChatOpenAI(model="gpt-4", temperature=0.7)

# Anthropic Claude
from langchain.chat_models import ChatAnthropic
llm = ChatAnthropic(model="claude-3-opus-20240229")

# 로컬 모델 (Ollama)
from langchain.llms import Ollama
llm = Ollama(model="llama2")

# Google Gemini
from langchain.chat_models import ChatGoogleGenerativeAI
llm = ChatGoogleGenerativeAI(model="gemini-pro")

# 모두 동일한 방식으로 사용
response = llm.invoke("안녕하세요")

2. Prompts - 프롬프트 관리

재사용 가능한 프롬프트 템플릿을 만듭니다.

from langchain.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate

# 기본 템플릿
simple_template = ChatPromptTemplate.from_template(
    "다음 주제에 대해 설명하세요: {topic}"
)

# 복잡한 템플릿
system_template = SystemMessagePromptTemplate.from_template(
    "당신은 {role}입니다. {context}"
)
human_template = HumanMessagePromptTemplate.from_template(
    "질문: {question}"
)

chat_prompt = ChatPromptTemplate.from_messages([
    system_template,
    human_template
])

# 사용
messages = chat_prompt.format_messages(
    role="Python 전문가",
    context="초보자도 이해할 수 있게 설명하세요.",
    question="리스트 컴프리헨션이 뭔가요?"
)

Few-Shot 프롬프트:

from langchain.prompts import FewShotPromptTemplate, PromptTemplate

# 예제 정의
examples = [
    {"input": "happy", "output": "sad"},
    {"input": "tall", "output": "short"},
    {"input": "hot", "output": "cold"}
]

# 예제 포맷
example_prompt = PromptTemplate(
    input_variables=["input", "output"],
    template="Input: {input}\nOutput: {output}"
)

# Few-shot 템플릿
few_shot_prompt = FewShotPromptTemplate(
    examples=examples,
    example_prompt=example_prompt,
    prefix="Give the antonym of the word:",
    suffix="Input: {input}\nOutput:",
    input_variables=["input"]
)

print(few_shot_prompt.format(input="big"))

3. Chains - 워크플로우 구성

여러 단계를 연결하여 복잡한 작업을 수행합니다.

기본 Chain:

from langchain.chains import LLMChain

llm = ChatOpenAI()
prompt = ChatPromptTemplate.from_template(
    "Write a {length} word summary of {topic}"
)

chain = LLMChain(llm=llm, prompt=prompt)
result = chain.run(length=100, topic="climate change")

Sequential Chain:

from langchain.chains import SimpleSequentialChain

# Chain 1: 주제로 개요 작성
chain_1 = LLMChain(
    llm=llm,
    prompt=ChatPromptTemplate.from_template(
        "Write an outline for {topic}"
    )
)

# Chain 2: 개요를 바탕으로 본문 작성
chain_2 = LLMChain(
    llm=llm,
    prompt=ChatPromptTemplate.from_template(
        "Write a detailed article based on this outline:\n{outline}"
    )
)

# 체인 연결
overall_chain = SimpleSequentialChain(
    chains=[chain_1, chain_2],
    verbose=True
)

result = overall_chain.run("AI in healthcare")

Router Chain (조건부 라우팅):

from langchain.chains.router import MultiPromptChain
from langchain.chains import ConversationChain

# 도메인별 프롬프트
physics_template = """당신은 물리학 전문가입니다...
질문: {input}"""

math_template = """당신은 수학 전문가입니다...
질문: {input}"""

prompt_infos = [
    {
        "name": "physics",
        "description": "물리학 관련 질문에 답합니다",
        "prompt_template": physics_template
    },
    {
        "name": "math",
        "description": "수학 관련 질문에 답합니다",
        "prompt_template": math_template
    }
]

# 라우터 체인 생성
router_chain = MultiPromptChain.from_prompts(
    llm=llm,
    prompt_infos=prompt_infos
)

# 자동으로 적절한 전문가에게 라우팅
print(router_chain.run("What is Newton's law?"))  # → physics
print(router_chain.run("Solve x^2 + 5x + 6 = 0"))  # → math

4. Memory - 대화 기억

대화 컨텍스트를 유지합니다.

ConversationBufferMemory:

from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain

memory = ConversationBufferMemory()
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

conversation.predict(input="안녕! 내 이름은 철수야")
# AI: 안녕하세요, 철수님!

conversation.predict(input="내 이름이 뭐였지?")
# AI: 철수라고 하셨습니다.

ConversationBufferWindowMemory (최근 K개만):

from langchain.memory import ConversationBufferWindowMemory

# 최근 5개 대화만 기억
memory = ConversationBufferWindowMemory(k=5)

ConversationSummaryMemory (요약):

from langchain.memory import ConversationSummaryMemory

# 긴 대화를 요약하여 저장 (토큰 절약)
memory = ConversationSummaryMemory(llm=llm)

Vector Store Memory (의미 기반):

from langchain.memory import VectorStoreRetrieverMemory
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings

# 벡터 DB에 대화 저장, 관련성 높은 것만 검색
vectorstore = Chroma(embedding_function=OpenAIEmbeddings())
memory = VectorStoreRetrieverMemory(
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5})
)

5. Document Loaders - 데이터 로딩

다양한 소스에서 데이터를 불러옵니다.

# PDF
from langchain.document_loaders import PyPDFLoader
loader = PyPDFLoader("document.pdf")
pages = loader.load()

# 웹페이지
from langchain.document_loaders import WebBaseLoader
loader = WebBaseLoader("https://example.com")
docs = loader.load()

# 디렉토리 (여러 파일)
from langchain.document_loaders import DirectoryLoader
loader = DirectoryLoader('./documents', glob="**/*.txt")
docs = loader.load()

# Notion
from langchain.document_loaders import NotionDirectoryLoader
loader = NotionDirectoryLoader("notion_export")
docs = loader.load()

# CSV
from langchain.document_loaders import CSVLoader
loader = CSVLoader("data.csv")
docs = loader.load()

# YouTube 자막
from langchain.document_loaders import YoutubeLoader
loader = YoutubeLoader.from_youtube_url("https://youtube.com/watch?v=...")
docs = loader.load()

6. Text Splitters - 문서 분할

큰 문서를 적절한 크기로 나눕니다.

from langchain.text_splitter import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,  # 청크 크기
    chunk_overlap=200,  # 중복 크기 (컨텍스트 보존)
    separators=["\n\n", "\n", " ", ""]  # 우선순위
)

chunks = text_splitter.split_documents(documents)

코드 전용 Splitter:

from langchain.text_splitter import (
    PythonCodeTextSplitter,
    JavaScriptCodeTextSplitter
)

# Python 코드를 함수/클래스 단위로 분할
python_splitter = PythonCodeTextSplitter(chunk_size=500)
chunks = python_splitter.split_text(python_code)

7. Vector Stores - 벡터 DB 연동

from langchain.vectorstores import Chroma, Pinecone, Weaviate
from langchain.embeddings import OpenAIEmbeddings

embeddings = OpenAIEmbeddings()

# Chroma (로컬)
vectorstore = Chroma.from_documents(
    documents=docs,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# Pinecone (클라우드)
import pinecone
pinecone.init(api_key="your-api-key", environment="us-west1-gcp")
vectorstore = Pinecone.from_documents(
    docs, embeddings, index_name="my-index"
)

# 검색
results = vectorstore.similarity_search("query", k=5)

8. Retrievers - 검색 전략

# 기본 유사도 검색
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# MMR (최대 marginal relevance) - 다양성 확보
retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 5, "fetch_k": 20}
)

# 유사도 점수 기반 필터링
retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"score_threshold": 0.8}
)

# 앙상블 (여러 검색기 결합)
from langchain.retrievers import EnsembleRetriever

ensemble_retriever = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    weights=[0.7, 0.3]
)

실전 패턴

1. RAG (Retrieval-Augmented Generation)

from langchain.chains import RetrievalQA
from langchain.document_loaders import DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.chat_models import ChatOpenAI

# 1. 문서 로드
loader = DirectoryLoader('./docs', glob="**/*.pdf")
documents = loader.load()

# 2. 분할
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
splits = text_splitter.split_documents(documents)

# 3. 벡터 저장
vectorstore = Chroma.from_documents(
    documents=splits,
    embedding=OpenAIEmbeddings()
)

# 4. RAG 체인 생성
qa_chain = RetrievalQA.from_chain_type(
    llm=ChatOpenAI(model="gpt-4"),
    chain_type="stuff",  # stuff, map_reduce, refine, map_rerank
    retriever=vectorstore.as_retriever(),
    return_source_documents=True
)

# 5. 질의
result = qa_chain({"query": "What is the main topic?"})
print(result['result'])
print("Sources:", result['source_documents'])

2. Conversational RAG

from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

qa_chain = ConversationalRetrievalChain.from_llm(
    llm=ChatOpenAI(model="gpt-4"),
    retriever=vectorstore.as_retriever(),
    memory=memory
)

# 대화 가능
qa_chain({"question": "What is RAG?"})
qa_chain({"question": "Can you explain more about it?"})  # 이전 맥락 유지

3. Agent (도구 사용)

from langchain.agents import initialize_agent, Tool
from langchain.agents import AgentType
from langchain.tools import DuckDuckGoSearchRun
from langchain.utilities import WikipediaAPIWrapper

# 도구 정의
search = DuckDuckGoSearchRun()
wikipedia = WikipediaAPIWrapper()

tools = [
    Tool(
        name="Search",
        func=search.run,
        description="인터넷 검색. 최신 정보가 필요할 때 사용"
    ),
    Tool(
        name="Wikipedia",
        func=wikipedia.run,
        description="위키백과 검색. 일반적인 지식이 필요할 때 사용"
    )
]

# Agent 생성
agent = initialize_agent(
    tools=tools,
    llm=ChatOpenAI(model="gpt-4"),
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# 실행 - Agent가 자동으로 도구 선택
agent.run("2024년 노벨 물리학상 수상자는 누구인가?")

커스텀 도구 만들기:

from langchain.tools import BaseTool

class CalculatorTool(BaseTool):
    name = "Calculator"
    description = "수학 계산을 수행합니다. 입력: 수식(예: '2 + 2')"

    def _run(self, query: str) -> str:
        try:
            return str(eval(query))
        except:
            return "계산 오류"

    async def _arun(self, query: str) -> str:
        return self._run(query)

tools.append(CalculatorTool())

4. SQL Agent

from langchain.utilities import SQLDatabase
from langchain.agents import create_sql_agent
from langchain.agents.agent_toolkits import SQLDatabaseToolkit

# DB 연결
db = SQLDatabase.from_uri("sqlite:///database.db")

# SQL Agent 생성
toolkit = SQLDatabaseToolkit(db=db, llm=ChatOpenAI(model="gpt-4"))
agent = create_sql_agent(
    llm=ChatOpenAI(model="gpt-4"),
    toolkit=toolkit,
    verbose=True
)

# 자연어로 SQL 쿼리 실행
agent.run("2024년에 가입한 사용자 수는?")
# Agent가 자동으로 SQL 생성 및 실행
# SELECT COUNT(*) FROM users WHERE YEAR(created_at) = 2024

고급 기능

1. Streaming

실시간으로 응답을 받습니다.

from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

llm = ChatOpenAI(
    model="gpt-4",
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()]
)

# 응답이 토큰 단위로 실시간 출력됨
llm.invoke("긴 이야기를 들려주세요")

2. Caching

동일한 쿼리의 재호출을 방지합니다.

from langchain.cache import InMemoryCache, SQLiteCache
import langchain

# 메모리 캐시
langchain.llm_cache = InMemoryCache()

# 영구 캐시
langchain.llm_cache = SQLiteCache(database_path=".langchain.db")

# 첫 호출: API 요청
llm.invoke("What is AI?")  # 느림

# 두 번째 호출: 캐시에서 반환
llm.invoke("What is AI?")  # 빠름, 비용 절약

3. Callbacks

실행 과정을 모니터링합니다.

from langchain.callbacks import get_openai_callback

with get_openai_callback() as cb:
    result = chain.run("Tell me about AI")

    print(f"Tokens used: {cb.total_tokens}")
    print(f"Cost: ${cb.total_cost}")

커스텀 Callback:

from langchain.callbacks.base import BaseCallbackHandler

class MyCallback(BaseCallbackHandler):
    def on_llm_start(self, serialized, prompts, **kwargs):
        print(f"LLM 시작: {prompts}")

    def on_llm_end(self, response, **kwargs):
        print(f"LLM 완료: {response}")

    def on_chain_start(self, serialized, inputs, **kwargs):
        print(f"Chain 시작: {inputs}")

llm = ChatOpenAI(callbacks=[MyCallback()])

4. Output Parsers

LLM 출력을 구조화합니다.

from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field

class Person(BaseModel):
    name: str = Field(description="사람 이름")
    age: int = Field(description="나이")
    hobbies: list[str] = Field(description="취미 목록")

parser = PydanticOutputParser(pydantic_object=Person)

prompt = ChatPromptTemplate.from_template(
    "다음 정보를 추출하세요:\n{format_instructions}\n\n{text}"
)

chain = prompt | llm | parser

result = chain.invoke({
    "text": "철수는 25살이고 축구와 독서를 좋아합니다.",
    "format_instructions": parser.get_format_instructions()
})

print(result.name)  # "철수"
print(result.age)   # 25
print(result.hobbies)  # ["축구", "독서"]

LangChain Expression Language (LCEL)

체인을 더 간결하게 작성하는 새로운 문법입니다.

# 기존 방식
chain = LLMChain(llm=llm, prompt=prompt)

# LCEL 방식
chain = prompt | llm

# 복잡한 체인도 간결하게
chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

result = chain.invoke("What is AI?")

병렬 실행:

from langchain.schema.runnable import RunnableParallel

chain = RunnableParallel(
    summary=summary_chain,
    translation=translation_chain,
    sentiment=sentiment_chain
)

# 3개 체인이 동시에 실행됨
result = chain.invoke({"text": "..."})
# {"summary": "...", "translation": "...", "sentiment": "..."}

실전 팁

1. 에러 처리

from langchain.chains import LLMChain
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
def run_chain(query):
    return chain.run(query)

try:
    result = run_chain("query")
except Exception as e:
    print(f"Error: {e}")
    # 폴백 로직

2. 비용 최적화

# 1. 캐싱 사용
langchain.llm_cache = SQLiteCache(database_path=".cache.db")

# 2. 저렴한 모델 먼저 시도
cheap_llm = ChatOpenAI(model="gpt-3.5-turbo")
expensive_llm = ChatOpenAI(model="gpt-4")

try:
    result = cheap_llm.invoke(query)
    if confidence_score(result) < 0.8:  # 신뢰도 낮으면
        result = expensive_llm.invoke(query)
except:
    result = expensive_llm.invoke(query)

# 3. 프롬프트 압축
from langchain.prompts import PromptTemplate

# 불필요한 단어 제거, 간결하게 작성

3. 디버깅

# Verbose 모드
chain = LLMChain(llm=llm, prompt=prompt, verbose=True)

# LangSmith (공식 디버깅 도구)
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"

# 모든 실행이 LangSmith에 로깅됨

결론

LangChain은 AI 애플리케이션 개발의 사실상 표준입니다. 이를 통해:

  • 🚀 빠른 개발: 보일러플레이트 코드 제거
  • 🔧 모듈화: 재사용 가능한 컴포넌트
  • 🔄 유연성: 다양한 LLM/도구 통합
  • 📊 확장성: 간단한 챗봇부터 복잡한 Agent까지

시작하기:

pip install langchain langchain-openai

# 추가 패키지 (필요시)
pip install chromadb  # 벡터 DB
pip install pypdf  # PDF 로더
pip install wikipedia  # Wikipedia 도구

LangChain을 마스터하면 며칠이 걸릴 작업을 몇 시간만에 완성할 수 있습니다!