오픈소스를 위한 기초 상식
LangChain 기초 학습 가이드
지나가는 프로도
2025. 3. 30. 18:57
1. 기본 컴포넌트 이해
LangChain은 언어 모델(LLMs)을 이용한 애플리케이션 개발을 위한 프레임워크입니다. 다음은 LangChain의 핵심 컴포넌트들입니다.
모델 (Models)
LangChain은 다양한 언어 모델을 지원합니다:
LLMs (Large Language Models)
텍스트를 입력받아 텍스트를 생성하는 모델입니다.
from langchain_openai import OpenAI
llm = OpenAI(openai_api_key="your-api-key")
result = llm.invoke("인공지능이란 무엇인가요?")
print(result)
채팅 모델 (Chat Models)
대화형 메시지를 입력받아 메시지를 생성합니다.
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
chat = ChatOpenAI(openai_api_key="your-api-key")
messages = [
SystemMessage(content="당신은 도움이 되는 AI 어시스턴트입니다."),
HumanMessage(content="파이썬에서 리스트와 튜플의 차이점은 무엇인가요?")
]
response = chat.invoke(messages)
print(response.content)
프롬프트 (Prompts)
언어 모델에 전달되는 입력 텍스트의 구조를 정의합니다.
프롬프트 템플릿 (PromptTemplates)
변수가 포함된 템플릿을 정의하고 실행 시 변수를 입력합니다.
from langchain_core.prompts import PromptTemplate
template = PromptTemplate.from_template(
"다음 주제에 대한 5가지 핵심 포인트를 알려주세요: {topic}"
)
prompt = template.format(topic="블록체인 기술")
print(prompt)
출력 파서 (Output Parsers)
언어 모델의 출력을 구조화된 형식으로 변환합니다.
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_core.pydantic_v1 import BaseModel, Field
# 문자열 파서
parser = StrOutputParser()
# JSON 파서
class Article(BaseModel):
title: str = Field(description="기사 제목")
content: str = Field(description="기사 내용")
tags: list[str] = Field(description="관련 태그 목록")
json_parser = JsonOutputParser(pydantic_object=Article)
메모리 (Memory)
대화 이력을 저장하고 관리합니다.
from langchain.memory import ConversationBufferMemory
memory = ConversationBufferMemory()
memory.chat_memory.add_user_message("안녕하세요!")
memory.chat_memory.add_ai_message("안녕하세요! 무엇을 도와드릴까요?")
print(memory.load_memory_variables({}))
체인 (Chains)
여러 컴포넌트를 연결하여 복잡한 작업을 수행합니다.
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
prompt = ChatPromptTemplate.from_template("다음에 대해 설명해주세요: {topic}")
model = ChatOpenAI()
output_parser = StrOutputParser()
chain = prompt | model | output_parser
result = chain.invoke({"topic": "양자 컴퓨팅"})
print(result)
도구와 에이전트 (Tools & Agents)
언어 모델이 외부 도구를 사용할 수 있게 합니다.
from langchain import agents
from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
# 검색 도구 설정
search = TavilySearchResults(tavily_api_key="your-tavily-api-key")
# 에이전트 설정
llm = ChatOpenAI(model="gpt-3.5-turbo")
agent = agents.create_openai_functions_agent(llm, [search])
agent_executor = agents.AgentExecutor(agent=agent, tools=[search])
# 에이전트 실행
response = agent_executor.invoke({"input": "최근 AI 기술 트렌드는 무엇인가요?"})
print(response["output"])
검색기 (Retrievers)
외부 데이터 소스에서 관련 정보를 검색합니다.
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import CharacterTextSplitter
from langchain_community.document_loaders import TextLoader
# 문서 로드 및 분할
loader = TextLoader("data.txt")
documents = loader.load()
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
texts = text_splitter.split_documents(documents)
# 벡터 저장소 생성
embeddings = OpenAIEmbeddings()
db = Chroma.from_documents(texts, embeddings)
# 검색기 사용
retriever = db.as_retriever()
docs = retriever.get_relevant_documents("프로그래밍 언어의 역사")
2. 프롬프트 템플릿 작성
프롬프트 템플릿은 LLM에 전달할 메시지를 동적으로 구성하는 기능을 제공합니다.
기본 프롬프트 템플릿
from langchain_core.prompts import PromptTemplate
# 단일 변수를 포함한 기본 템플릿
template = PromptTemplate.from_template(
"다음 주제에 대한 짧은 설명을 작성해주세요: {topic}"
)
prompt = template.format(topic="인공지능의 윤리적 문제")
print(prompt)
채팅 프롬프트 템플릿
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import SystemMessage, HumanMessagePromptTemplate
# 시스템 메시지와 사용자 메시지를 포함한 템플릿
chat_template = ChatPromptTemplate.from_messages([
SystemMessage(content="당신은 {role} 전문가입니다."),
HumanMessagePromptTemplate.from_template("{query}")
])
messages = chat_template.format_messages(
role="프로그래밍",
query="파이썬에서 비동기 프로그래밍을 어떻게 구현하나요?"
)
print(messages)
복잡한 프롬프트 템플릿
from langchain_core.prompts import PromptTemplate, FewShotPromptTemplate
# 예시 목록
examples = [
{"input": "행복", "output": "행복은 만족과 즐거움을 느끼는 정서적 상태입니다."},
{"input": "성장", "output": "성장은 시간이 지남에 따라 발전하고 향상되는 과정입니다."},
{"input": "지식", "output": "지식은 경험, 교육, 이해를 통해 얻은 정보와 사실의 집합입니다."}
]
# 예시 형식 템플릿
example_formatter_template = """
입력: {input}
출력: {output}
"""
example_prompt = PromptTemplate.from_template(example_formatter_template)
# Few-shot 학습 템플릿
few_shot_prompt = FewShotPromptTemplate(
examples=examples,
example_prompt=example_prompt,
prefix="다음은 개념에 대한 정의 예시입니다:",
suffix="입력: {concept}\n출력:",
input_variables=["concept"]
)
print(few_shot_prompt.format(concept="창의성"))
조건부 프롬프트 템플릿
from langchain_core.prompts import PromptTemplate
# 조건부 로직이 포함된 템플릿
template = """
다음 {language} 코드를 분석해주세요:
```{language}
{code}
{%- if explanation_level == "beginner" %} 초보자도 이해할 수 있게 기본 개념부터 설명해주세요. {%- elif explanation_level == "intermediate" %} 중급 수준의 개발자를 위한 분석을 제공해주세요. {%- else %} 고급 개발자를 위한 심층 분석과 최적화 방안을 제시해주세요. {%- endif %} """
prompt_template = PromptTemplate.from_template(template) prompt = prompt_template.format( language="Python", code="def fibonacci(n):\n if n <= 1:\n return n\n return fibonacci(n-1) + fibonacci(n-2)", explanation_level="beginner" ) print(prompt)
## 3. 체인 구성 방법
체인은 여러 컴포넌트를 연결하여 복잡한 작업을 수행하는 구조입니다.
### 기본 체인 (LCEL 방식)
LCEL(LangChain Expression Language)을 사용한 체인 구성:
```python
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
# 컴포넌트 정의
prompt = ChatPromptTemplate.from_template("다음에 대해 설명해주세요: {topic}")
model = ChatOpenAI()
output_parser = StrOutputParser()
# 파이프 연산자(|)로 체인 구성
chain = prompt | model | output_parser
# 체인 실행
result = chain.invoke({"topic": "블록체인"})
print(result)
분기 체인 (Branching Chain)
조건에 따라 다른 경로로 실행되는 체인:
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableBranch
# 언어 감지 함수
def detect_language(text):
# 실제로는 언어 감지 로직이 들어갑니다
if "안녕" in text or "한국" in text:
return "korean"
else:
return "english"
# 언어별 프롬프트
korean_prompt = ChatPromptTemplate.from_template("한국어 응답: {query}")
english_prompt = ChatPromptTemplate.from_template("English response: {query}")
# 모델 및 파서
model = ChatOpenAI()
parser = StrOutputParser()
# 분기 체인 구성
branch = RunnableBranch(
(lambda x: detect_language(x["query"]) == "korean", korean_prompt | model | parser),
(lambda x: detect_language(x["query"]) == "english", english_prompt | model | parser),
# 기본 분기
english_prompt | model | parser
)
# 실행
chain = {"query": RunnablePassthrough()} | branch
print(chain.invoke({"query": "한국의 역사에 대해 알려주세요"}))
print(chain.invoke({"query": "Tell me about history of USA"}))
순차 체인 (Sequential Chain)
여러 체인을 순차적으로 실행하는 구조:
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
# 첫 번째 체인: 주제 요약
summary_prompt = PromptTemplate.from_template(
"다음 주제에 대한 간략한 요약을 작성하세요: {topic}"
)
summary_chain = summary_prompt | ChatOpenAI() | StrOutputParser()
# 두 번째 체인: 질문 생성
question_prompt = PromptTemplate.from_template(
"다음 내용을 바탕으로 5가지 중요한 질문을 생성하세요:\n\n{summary}"
)
question_chain = question_prompt | ChatOpenAI() | StrOutputParser()
# 순차 체인 실행
def run_sequential_chain(topic):
# 첫 번째 체인 실행
summary = summary_chain.invoke({"topic": topic})
print(f"요약:\n{summary}\n")
# 두 번째 체인 실행
questions = question_chain.invoke({"summary": summary})
print(f"질문들:\n{questions}")
run_sequential_chain("인공지능의 미래")
에이전트 체인 (Agent Chain)
도구를 사용하여 자율적으로 작업을 수행하는 에이전트:
from langchain_openai import ChatOpenAI
from langchain.agents import load_tools, initialize_agent, AgentType
# 언어 모델 초기화
llm = ChatOpenAI(temperature=0)
# 도구 로드 (여기서는 기본 수학 도구 사용)
tools = load_tools(["llm-math"], llm=llm)
# 에이전트 초기화
agent = initialize_agent(
tools,
llm,
agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
verbose=True
)
# 에이전트 실행
agent.run("23462 곱하기 56271은 얼마인가요?")
메모리를 활용한 체인
대화 이력을 유지하는 체인:
from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferMemory
from langchain_openai import ChatOpenAI
# 메모리 초기화
memory = ConversationBufferMemory()
# 대화 체인 생성
conversation = ConversationChain(
llm=ChatOpenAI(),
memory=memory,
verbose=True
)
# 대화 진행
print(conversation.predict(input="안녕하세요!"))
print(conversation.predict(input="제 이름은 김철수입니다."))
print(conversation.predict(input="제 이름이 뭐라고 했죠?"))
검색 증강 생성 (RAG) 체인
외부 문서에서 정보를 검색하여 응답을 생성하는 체인:
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import WebBaseLoader
# 웹 페이지에서 문서 로드
loader = WebBaseLoader("https://en.wikipedia.org/wiki/Artificial_intelligence")
documents = loader.load()
# 문서 분할
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
splits = text_splitter.split_documents(documents)
# 벡터 저장소 생성
vectorstore = FAISS.from_documents(documents=splits, embedding=OpenAIEmbeddings())
retriever = vectorstore.as_retriever()
# RAG 체인 구성
template = """다음 컨텍스트를 사용하여 질문에 답변해주세요:
{context}
질문: {question}
"""
prompt = ChatPromptTemplate.from_template(template)
model = ChatOpenAI()
parser = StrOutputParser()
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
rag_chain = (
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| prompt
| model
| parser
)
# 체인 실행
response = rag_chain.invoke("인공지능의 정의는 무엇인가요?")
print(response)
복잡한 워크플로우 체인
여러 단계와 로직이 포함된 복잡한 체인:
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.pydantic_v1 import BaseModel, Field
from typing import List
from langchain_core.runnables import RunnablePassthrough
# 출력 구조 정의
class AnalysisResult(BaseModel):
summary: str = Field(description="주제에 대한 간략한 요약")
key_points: List[str] = Field(description="핵심 포인트 목록")
sentiment: str = Field(description="긍정적, 부정적 또는 중립적")
# 언어 모델 및 파서
llm = ChatOpenAI(temperature=0)
json_parser = JsonOutputParser(pydantic_object=AnalysisResult)
# 1단계: 초기 분석 수행
analysis_prompt = ChatPromptTemplate.from_template("""
다음 텍스트를 분석하세요:
"{text}"
요약, 핵심 포인트 및 감정 분석을 제공하세요.
""")
analysis_chain = analysis_prompt | llm | json_parser
# 2단계: 추가 정보를 기반으로 확장
expansion_prompt = ChatPromptTemplate.from_template("""
다음 초기 분석을 확장하세요:
{analysis}
다음 카테고리에 대한 정보를 추가하세요: {category}
""")
expansion_chain = (
{
"analysis": lambda x: x["initial_analysis"],
"category": lambda x: x["category"]
}
| expansion_prompt
| llm
| StrOutputParser()
)
# 전체 워크플로우
def workflow(text, category):
# 초기 분석 수행
initial_analysis = analysis_chain.invoke({"text": text})
print(f"초기 분석: {initial_analysis}")
# 확장 수행
expanded_result = expansion_chain.invoke({
"initial_analysis": initial_analysis,
"category": category
})
print(f"확장된 분석: {expanded_result}")
return {
"initial_analysis": initial_analysis,
"expanded_analysis": expanded_result
}
# 실행 예
result = workflow(
"인공지능 기술은 빠르게 발전하고 있으며, 많은 산업에 혁명적인 변화를 가져오고 있습니다.",
"경제적 영향"
)
4. 고급 LangChain 활용
벡터 저장소와 임베딩 활용
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader
import os
# 문서 로드
loader = TextLoader("data.txt")
documents = loader.load()
# 문서 분할
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
texts = text_splitter.split_documents(documents)
# 임베딩 및 벡터 저장소 생성
embeddings = OpenAIEmbeddings()
db = Chroma.from_documents(texts, embeddings, persist_directory="./chroma_db")
# 유사 문서 검색
query = "인공지능의 미래 전망"
docs = db.similarity_search(query)
print(docs[0].page_content)
데이터베이스 기반 에이전트
from langchain_community.utilities import SQLDatabase
from langchain_core.tools import Tool
from langchain_community.agent_toolkits import SQLDatabaseToolkit
from langchain.agents import create_sql_agent
from langchain_openai import ChatOpenAI
# 데이터베이스 연결
db = SQLDatabase.from_uri("sqlite:///chinook.db")
# SQL 에이전트 생성
llm = ChatOpenAI(temperature=0)
toolkit = SQLDatabaseToolkit(db=db, llm=llm)
agent = create_sql_agent(
llm=llm,
toolkit=toolkit,
verbose=True
)
# 에이전트 실행
agent.run("customers 테이블에서 미국 고객의 수를 알려주세요.")
여러 소스의 데이터 통합 체인
from langchain_community.utilities import WikipediaAPIWrapper
from langchain_community.utilities import SerpAPIWrapper
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from operator import itemgetter
from langchain_core.runnables import RunnablePassthrough
# 도구 설정
wikipedia = WikipediaAPIWrapper()
search = SerpAPIWrapper()
# 통합 체인 구성
def fetch_wikipedia(query):
return wikipedia.run(query)
def fetch_search_results(query):
return search.run(query)
# 프롬프트 템플릿
template = """다음 정보를 바탕으로 {query}에 대한 종합적인 답변을 제공하세요:
웹 검색 결과:
{search_results}
위키피디아 정보:
{wikipedia_results}
"""
prompt = ChatPromptTemplate.from_template(template)
# 체인 구성
integrated_chain = (
{
"query": RunnablePassthrough(),
"search_results": lambda x: fetch_search_results(x),
"wikipedia_results": lambda x: fetch_wikipedia(x)
}
| prompt
| ChatOpenAI()
| StrOutputParser()
)
# 실행
response = integrated_chain.invoke("양자 컴퓨팅의 최근 발전")
print(response)
실시간 데이터 처리 체인
import yfinance as yf
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from datetime import datetime, timedelta
# 주식 데이터 가져오는 함수
def get_stock_data(symbol, days=7):
end_date = datetime.now()
start_date = end_date - timedelta(days=days)
data = yf.download(symbol, start=start_date, end=end_date)
return data.reset_index().to_dict(orient='records')
# 프롬프트 템플릿
template = """
다음은 {symbol} 주식의 최근 {days}일 데이터입니다:
{stock_data}
이 데이터를 분석하고 다음 사항에 답변해주세요:
1. 주가 추세는 어떠한가요?
2. 거래량에 특이점이 있나요?
3. 이 데이터를 바탕으로 간단한 투자 조언을 제공해주세요.
"""
prompt = ChatPromptTemplate.from_template(template)
# 체인 구성
llm = ChatOpenAI()
chain = prompt | llm | StrOutputParser()
# 실행
def analyze_stock(symbol, days=7):
stock_data = get_stock_data(symbol, days)
response = chain.invoke({
"symbol": symbol,
"days": days,
"stock_data": stock_data
})
return response
# 예시 실행
print(analyze_stock("AAPL"))
5. 실전 활용 예제
다중 API 통합 비서 에이전트
from langchain_openai import ChatOpenAI
from langchain_community.tools import tool
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
import requests
import json
import datetime
# 도구 정의
@tool
def get_weather(location: str) -> str:
"""지정된 위치의 현재 날씨 정보를 가져옵니다."""
# 실제 구현에서는 날씨 API 사용
return f"{location}의 현재 날씨: 맑음, 기온 22°C"
@tool
def search_news(query: str) -> str:
"""특정 주제에 관한 최신 뉴스를 검색합니다."""
# 실제 구현에서는 뉴스 API 사용
return f"{query}에 관한 최신 뉴스: 최근 관련 발표가 있었습니다."
@tool
def set_reminder(title: str, time: str) -> str:
"""새 알림을 설정합니다."""
# 실제 구현에서는 캘린더/알림 API 사용
return f"알림 '{title}'이(가) {time}에 설정되었습니다."
# 프롬프트 템플릿
prompt = ChatPromptTemplate.from_messages([
("system", "당신은 날씨, 뉴스, 알림 기능을 갖춘 지능형 비서입니다."),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
# 에이전트 설정
llm = ChatOpenAI(temperature=0)
tools = [get_weather, search_news, set_reminder]
agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
# 실행
result = agent_executor.invoke({
"input": "서울의 오늘 날씨를 알려주고, AI에 관한 최신 뉴스를 검색해줘.",
"chat_history": []
})
print(result["output"])
문서 QA 시스템
from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
# PDF 파일 로드
loader = PyPDFLoader("document.pdf")
pages = loader.load()
# 문서 분할
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
splits = text_splitter.split_documents(pages)
# 벡터 저장소 생성
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_documents(documents=splits, embedding=embeddings)
retriever = vectorstore.as_retriever()
# 프롬프트 템플릿
template = """다음 문서 내용을 바탕으로 질문에 답변하세요:
{context}
질문: {question}
"""
prompt = ChatPromptTemplate.from_template(template)
# 체인 구성
llm = ChatOpenAI()
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
qa_chain = (
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
# 실행 예시
def ask_document(question):
return qa_chain.invoke(question)
print(ask_document("이 문서의 주요 주제는 무엇인가요?"))
대화형 데이터 분석 에이전트
import pandas as pd
from langchain_experimental.agents import create_pandas_dataframe_agent
from langchain_openai import ChatOpenAI
# 데이터 로드
df = pd.read_csv("sales_data