LangChain은 대규모 언어 모델(LLM)을 기반으로 한 애플리케이션을 구축하기 위한 오픈소스 프레임워크다. 체인, 에이전트, 메모리, RAG 파이프라인 구축을 추상화한다.
LangChain 핵심 구성 요소
| 구성 요소 | 역할 |
|---|
| LLM/Chat Models | OpenAI, Anthropic, Ollama 등 래핑 |
| Prompts | 프롬프트 템플릿 관리 |
| Chains | 여러 단계를 파이프라인으로 연결 |
| Memory | 대화 기록 관리 |
| Agents | 도구를 사용하는 자율 실행 |
| Tools | 웹 검색, 계산기, DB 쿼리 등 |
| Vector Stores | 임베딩 기반 유사 검색 |
기본 LCEL 체인
python
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
prompt = ChatPromptTemplate.from_messages([
("system", "당신은 {language} 전문가입니다."),
("human", "{topic}에 대해 간단히 설명해 주세요."),
])
chain = prompt | llm | StrOutputParser()
result = chain.invoke({
"language": "Python",
"topic": "데코레이터 패턴",
})
print(result)
# 스트리밍
for chunk in chain.stream({"language": "Python", "topic": "제너레이터"}):
print(chunk, end="", flush=True)
RAG 파이프라인
python
from langchain_community.document_loaders import WebBaseLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
# 문서 로드 및 청킹
loader = WebBaseLoader("https://docs.example.com")
docs = loader.load()
splits = RecursiveCharacterTextSplitter(chunk_size=500).split_documents(docs)
# 벡터 저장소
vectorstore = Chroma.from_documents(splits, OpenAIEmbeddings())
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})
# RAG 체인
qa_prompt = ChatPromptTemplate.from_template("""
컨텍스트를 참고하여 질문에 답하세요.
컨텍스트: {context}
질문: {input}
""")
document_chain = create_stuff_documents_chain(llm, qa_prompt)
rag_chain = create_retrieval_chain(retriever, document_chain)
result = rag_chain.invoke({"input": "주요 기능이 무엇인가요?"})
print(result["answer"])
LangGraph (에이전트 오케스트레이션)
python
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
class AgentState(TypedDict):
messages: list
next_step: str
graph = StateGraph(AgentState)
graph.add_node("retrieve", retrieve_node)
graph.add_node("generate", generate_node)
graph.add_conditional_edges("retrieve", router, {"generate": "generate", "end": END})
app = graph.compile()
