LLM 프롬프트 엔지니어링 고급 기법: 실전에서 바로 쓰는 프롬프트 패턴
Chain of Thought, Few-Shot Learning, Self-Consistency 등 실무에서 즉시 활용 가능한 고급 프롬프트 엔지니어링 기법을 코드 예제와 함께 소개합니다.
LangChain은 대규모 언어 모델(LLM)을 활용한 애플리케이션을 쉽게 개발할 수 있게 해주는 프레임워크입니다. 복잡한 AI 기능을 모듈화된 컴포넌트로 제공하여, 마치 레고 블록을 조립하듯 AI 앱을 구축할 수 있습니다.
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")
다양한 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("안녕하세요")
재사용 가능한 프롬프트 템플릿을 만듭니다.
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"))
여러 단계를 연결하여 복잡한 작업을 수행합니다.
기본 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
대화 컨텍스트를 유지합니다.
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})
)
다양한 소스에서 데이터를 불러옵니다.
# 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()
큰 문서를 적절한 크기로 나눕니다.
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)
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)
# 기본 유사도 검색
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]
)
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'])
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?"}) # 이전 맥락 유지
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())
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
실시간으로 응답을 받습니다.
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
llm = ChatOpenAI(
model="gpt-4",
streaming=True,
callbacks=[StreamingStdOutCallbackHandler()]
)
# 응답이 토큰 단위로 실시간 출력됨
llm.invoke("긴 이야기를 들려주세요")
동일한 쿼리의 재호출을 방지합니다.
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?") # 빠름, 비용 절약
실행 과정을 모니터링합니다.
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()])
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) # ["축구", "독서"]
체인을 더 간결하게 작성하는 새로운 문법입니다.
# 기존 방식
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": "..."}
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}")
# 폴백 로직
# 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
# 불필요한 단어 제거, 간결하게 작성
# 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 애플리케이션 개발의 사실상 표준입니다. 이를 통해:
시작하기:
pip install langchain langchain-openai
# 추가 패키지 (필요시)
pip install chromadb # 벡터 DB
pip install pypdf # PDF 로더
pip install wikipedia # Wikipedia 도구
LangChain을 마스터하면 며칠이 걸릴 작업을 몇 시간만에 완성할 수 있습니다!