LangChain 정리
Posted by Albert 67Day 8Hour 55Min 11Sec ago [2025-12-01]
1. 기초: LangChain 시작하기
설치 및 기본 설정
' 설치
pip install langchain langchain-openai langchain-anthropic
' 기본 LLM 호출
from langchain_openai import ChatOpenAI
import os
os.environ["OPENAI_API_KEY"] = "your-api-key"
llm = ChatOpenAI(model="gpt-4", temperature=0.7)
response = llm.invoke("안녕하세요! LangChain이 뭔가요?")
print(response.content)
프롬프트 템플릿 활용
from langchain.prompts import PromptTemplate, ChatPromptTemplate
' 간단한 프롬프트 템플릿
template = PromptTemplate(
input_variables=["topic"],
template="다음 주제에 대해 간단히 설명해주세요: {topic}"
)
prompt = template.format(topic="인공지능")
response = llm.invoke(prompt)
print(response.content)
' 채팅 프롬프트 템플릿
chat_template = ChatPromptTemplate.from_messages([
("system", "당신은 친절한 AI 어시스턴트입니다."),
("human", "{question}")
])
chain = chat_template | llm
response = chain.invoke({"question": "파이썬이란?"})
print(response.content)
2. 중급: 체인과 메모리
Simple Chain 구성
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
' 번역 체인
translate_prompt = PromptTemplate(
input_variables=["text", "language"],
template="다음 텍스트를 {language}로 번역해주세요: {text}"
)
translate_chain = LLMChain(llm=llm, prompt=translate_prompt)
result = translate_chain.run(text="Hello, how are you?", language="한국어")
print(result)
Sequential Chain으로 복잡한 작업
from langchain.chains import SequentialChain
' 1단계: 주제 생성
topic_prompt = PromptTemplate(
input_variables=["interest"],
template="다음 관심사에 관련된 블로그 주제를 하나 제안해주세요: {interest}"
)
topic_chain = LLMChain(llm=llm, prompt=topic_prompt, output_key="topic")
' 2단계: 개요 작성
outline_prompt = PromptTemplate(
input_variables=["topic"],
template="다음 주제로 블로그 글의 개요를 작성해주세요: {topic}"
)
outline_chain = LLMChain(llm=llm, prompt=outline_prompt, output_key="outline")
' Sequential Chain 연결
overall_chain = SequentialChain(
chains=[topic_chain, outline_chain],
input_variables=["interest"],
output_variables=["topic", "outline"]
)
result = overall_chain({"interest": "머신러닝"})
print("주제:", result["topic"])
print("\n개요:", result["outline"])
대화 메모리 활용
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
' 대화 기록 저장
memory = ConversationBufferMemory()
conversation = ConversationChain(
llm=llm,
memory=memory,
verbose=True
)
' 연속된 대화
response1 = conversation.predict(input="제 이름은 김철수입니다.")
print(response1)
response2 = conversation.predict(input="제 이름이 뭐라고 했죠?")
print(response2)
' 대화 기록 확인
print("\n대화 기록:")
print(memory.load_memory_variables({}))
3. 중급-고급: RAG (Retrieval-Augmented Generation)
문서 로드 및 벡터 저장소 구축
from langchain.document_loaders import TextLoader, PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain.vectorstores import FAISS
' 문서 로드 및 분할
loader = TextLoader("document.txt", encoding="utf-8")
documents = loader.load()
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
length_function=len
)
splits = text_splitter.split_documents(documents)
' 벡터 저장소 생성
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_documents(splits, embeddings)
' 유사도 검색
query = "AI의 미래는?"
docs = vectorstore.similarity_search(query, k=3)
for doc in docs:
print(doc.page_content[:200])
RAG 체인 구성
from langchain.chains import RetrievalQA
' RAG 체인 생성
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
return_source_documents=True
)
' 질의응답
result = qa_chain({"query": "문서에서 주요 내용을 요약해주세요."})
print("답변:", result["result"])
print("\n참조 문서:")
for doc in result["source_documents"]:
print(f"- {doc.page_content[:100]}...")
4. 고급: 에이전트와 도구
커스텀 도구 생성
from langchain.agents import Tool, AgentExecutor, create_react_agent
from langchain.tools import BaseTool
from langchain import hub
import requests
' 날씨 도구 예제
class WeatherTool(BaseTool):
name = "weather"
description = "도시의 현재 날씨를 알려줍니다. 입력은 도시 이름입니다."
def _run(self, city: str) -> str:
' 실제로는 API 호출
return f"{city}의 날씨는 맑고 기온은 20도입니다."
async def _arun(self, city: str) -> str:
return self._run(city)
' 계산기 도구
def calculator(expression: str) -> str:
try:
return str(eval(expression))
except:
return "계산 오류"
tools = [
WeatherTool(),
Tool(
name="Calculator",
func=calculator,
description="수학 계산을 수행합니다. 입력은 파이썬 표현식입니다."
)
]
ReAct 에이전트 구성
from langchain_openai import ChatOpenAI
' 에이전트 프롬프트 가져오기
prompt = hub.pull("hwchase17/react")
' 에이전트 생성
agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
handle_parsing_errors=True
)
' 에이전트 실행
result = agent_executor.invoke({
"input": "서울의 날씨는 어떤가요? 그리고 25 곱하기 4는 얼마인가요?"
})
print(result["output"])
5. 고급: 커스텀 체인과 LCEL
LangChain Expression Language (LCEL)
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
' LCEL로 복잡한 체인 구성
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
rag_chain = (
{
"context": vectorstore.as_retriever() | format_docs,
"question": RunnablePassthrough()
}
| ChatPromptTemplate.from_template(
"다음 문맥을 바탕으로 질문에 답하세요:\n\n{context}\n\n질문: {question}"
)
| llm
| StrOutputParser()
)
' 실행
response = rag_chain.invoke("AI의 장점은 무엇인가요?")
print(response)
병렬 처리와 라우팅
from langchain_core.runnables import RunnableParallel, RunnableLambda
' 병렬 처리
parallel_chain = RunnableParallel(
summary=ChatPromptTemplate.from_template("다음을 요약: {text}") | llm,
keywords=ChatPromptTemplate.from_template("다음에서 키워드 추출: {text}") | llm,
sentiment=ChatPromptTemplate.from_template("다음의 감정 분석: {text}") | llm
)
result = parallel_chain.invoke({"text": "오늘은 정말 즐거운 하루였어요!"})
print("요약:", result["summary"].content)
print("키워드:", result["keywords"].content)
print("감정:", result["sentiment"].content)
6. 고급: 스트리밍과 콜백
스트리밍 응답
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
' 스트리밍 LLM
streaming_llm = ChatOpenAI(
model="gpt-4",
streaming=True,
callbacks=[StreamingStdOutCallbackHandler()]
)
' 실시간 응답
chain = ChatPromptTemplate.from_template("{topic}에 대해 설명해주세요") | streaming_llm
response = chain.invoke({"topic": "양자컴퓨팅"})
커스텀 콜백
from langchain.callbacks.base import BaseCallbackHandler
class MyCustomCallback(BaseCallbackHandler):
def on_llm_start(self, serialized, prompts, **kwargs):
print(f"[시작] LLM 호출 시작")
def on_llm_end(self, response, **kwargs):
print(f"[완료] LLM 호출 완료")
print(f"토큰 사용: {response.llm_output}")
def on_chain_start(self, serialized, inputs, **kwargs):
print(f"[체인 시작] {serialized.get('name', 'Unknown')}")
llm_with_callback = ChatOpenAI(callbacks=[MyCustomCallback()])
result = llm_with_callback.invoke("간단한 질문입니다")
7. 실전 활용: 전체 통합 예제
고급 챗봇 시스템
from langchain.memory import ConversationSummaryBufferMemory
from langchain.chains import ConversationalRetrievalChain
' 문서 기반 대화형 챗봇
memory = ConversationSummaryBufferMemory(
llm=llm,
memory_key="chat_history",
return_messages=True,
max_token_limit=100
)
qa_chatbot = ConversationalRetrievalChain.from_llm(
llm=llm,
retriever=vectorstore.as_retriever(),
memory=memory,
verbose=True
)
' 대화 진행
print(qa_chatbot({"question": "문서의 주요 내용이 뭔가요?"}))
print(qa_chatbot({"question": "좀 더 자세히 설명해주세요"}))
print(qa_chatbot({"question": "처음 질문이 뭐였죠?"}))
이 가이드는 LangChain의 핵심 기능들을 단계별로 다룹니다. 각 예제는 실제로 작동하는 코드이며, 프로젝트에 맞게 수정하여 사용할 수 있습니다.