LangChain 기초 학습 가이드

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