목차
- 오늘 강의 개요
- LangChain이란? - OpenAI API vs LangChain 비교
- LangChain 메시지 클래스 - SystemMessage, HumanMessage, AIMessage
- LLM 호출 3가지 방식 - invoke / stream / batch
- llm.bind() - 파라미터 동적 고정
- ChatPromptTemplate - 프롬프트 템플릿화
- partial() - 변수 사전 고정
- StrOutputParser - 문자열 파싱
- JsonOutputParser - JSON 구조 파싱
- PydanticOutputParser - 타입 검증 파싱
- 파이프라인(|) 연산자 - 체인 구성
- 핵심 개념 총정리
- 자주 나오는 실수 / 주의사항
1. 오늘 강의 개요
오늘부터 LangChain을 본격적으로 사용한다. OpenAI API를 직접 호출하던 방식에서, LangChain의 구조화된 방식으로 전환.
LangChain을 쓰는 이유
OpenAI API 직접: messages = [{“role”: “user”, “content”: “…”}] response = client.chat.completions.create(model=…, messages=messages) text = response.choices[0].message.content ← 매번 이렇게
LangChain: chain = prompt | llm | StrOutputParser() text = chain.invoke({“변수”: “값”}) ← 한 줄로 처리
2. LangChain이란? - OpenAI API vs LangChain 비교
LangChain의 역할
- OpenAI API의 “래퍼(Wrapper)” + 확장 기능 모음
- 다양한 LLM(Claude, Gemini 등)을 같은 인터페이스로 사용 가능
- 프롬프트 관리, 출력 파싱, 체인 구성 등을 구조화
기본 설정
1
2
3
4
5
6
7
| from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
from dotenv import load_dotenv
load_dotenv()
llm = ChatOpenAI(model='gpt-4o-mini', temperature=0.7)
|
가장 단순한 호출
1
2
3
| response = llm.invoke('LangChain을 한 문장으로 설명해주세요')
print(type(response)) # <class 'langchain_core.messages.ai.AIMessage'>
print(response.content) # 'LangChain은 LLM 기반 애플리케이션...'
|
llm.invoke() 반환값은 AIMessage 객체 .content 속성으로 텍스트 추출 (OpenAI 직접 호출과 다름!)
3. LangChain 메시지 클래스 - SystemMessage, HumanMessage, AIMessage
메시지 클래스 종류
| 클래스 | 역할 | OpenAI API 대응 |
|---|
| SystemMessage | 시스템 프롬프트 (AI 역할 지정) | {“role”: “system”, …} |
| HumanMessage | 사용자 질문 | {“role”: “user”, …} |
| AIMessage | LLM 응답 객체 | {“role”: “assistant”, …} |
사용 예시
1
2
3
4
5
6
7
8
9
10
| from langchain_core.messages import SystemMessage, HumanMessage
persona = "MLOps 리더"
messages = [
SystemMessage(content=f"당신은 {persona}입니다. 실무 팁 위주로 두 문장 이하로 답변해주세요"),
HumanMessage(content="Langchain 체인을 배포할 때 고려해야 하는 리스크 두 가지는?")
]
response = llm.invoke(messages)
print(response.content) # 실제 텍스트
|
응답 메타데이터 확인
1
2
3
4
5
6
7
8
9
| response.response_metadata
# {
# 'model_name': 'gpt-4o-mini',
# 'token_usage': {'prompt_tokens': 45, 'completion_tokens': 68, 'total_tokens': 113},
# 'finish_reason': 'stop'
# }
response.response_metadata['token_usage'] # 토큰 수 모니터링 → 비용 관리
response.response_metadata.get('model_name') # 'gpt-4o-mini'
|
유용한 함수 패턴
1
2
3
4
5
6
7
8
9
10
| def ask_expert(persona, question):
messages = [
SystemMessage(content=f"당신은 {persona}입니다. 간결하게 답변해주세요"),
HumanMessage(content=question)
]
return llm.invoke(messages).content
# 다양한 전문가에게 같은 질문
print(ask_expert("파이썬 전문가", "리스트 컴프리헨션이 뭔가요?"))
print(ask_expert("수학 선생님", "미적분이 뭔가요?"))
|
4. LLM 호출 3가지 방식 - invoke / stream / batch
방식 1: invoke - 단일 동기식 호출
1
2
3
| response = llm.invoke("파이썬이 뭔지 한 줄로 설명해주세요")
print(response.content)
# '파이썬은 간결한 문법과 강력한 라이브러리를 가진 프로그래밍 언어입니다.'
|
방식 2: stream - 실시간 스트리밍
1
2
3
4
5
6
7
| stream_chunks = []
for chunk in llm.stream("머신러닝의 3가지 유형을 나열해줘"):
stream_chunks.append(chunk.content)
print(chunk.content, end="", flush=True)
# 토큰 단위로 실시간 출력됨
result = ''.join(stream_chunks) # 전체 응답 문자열 조합
|
chunk.content : 이번 청크에 새로 생성된 텍스트
방식 3: batch - 다중 질문 병렬 처리
1
2
3
4
5
6
7
8
9
10
| questions = [
"파이썬의 장점 하나만 말해줘",
"자바스크립트의 장점 하나만 말해줘",
"러스트의 장점 하나만 말해줘"
]
results = llm.batch(questions) # 병렬로 3개 동시 처리
for answer in results:
print(answer.content)
print()
|
batch()는 내부적으로 병렬 처리 → 순차 실행보다 빠름 대량 데이터 처리에 효과적
3가지 방식 비교
| 방식 | 입력 | 출력 | 사용 상황 |
|---|
| invoke | 단일 질문 | AIMessage | 일반적인 단건 처리 |
| stream | 단일 질문 | iterator | 실시간 타이핑 효과 (UX) |
| batch | 질문 리스트 | [AIMessage] | 대량 처리, 성능 최적화 |
5. llm.bind() - 파라미터 동적 고정
개념
- 특정 파라미터를 고정한 새로운 LLM 인스턴스 생성
- 원본 llm 객체는 변경되지 않음
사용 예시
1
2
3
4
5
6
7
8
| # temperature=0, max_tokens=80 고정
stream_llm = llm.bind(temperature=0, max_tokens=80)
# JSON 응답 강제
json_llm = llm.bind(response_format={"type": "json_object"})
# bind된 llm은 일반 llm과 동일하게 사용
response = stream_llm.invoke("파이썬의 특징을 설명해주세요")
|
JSON 응답 예시
1
2
3
4
5
6
7
8
9
10
| from langchain_core.messages import SystemMessage, HumanMessage
import json
response = json_llm.invoke([
SystemMessage(content="JSON으로 응답하세요"),
HumanMessage(content="서울의 정보를 알려주세요")
])
data = json.loads(response.content)
# {'name': '서울', 'country': '대한민국', 'population': '약 950만'}
|
6. ChatPromptTemplate - 프롬프트 템플릿화
왜 필요한가?
1
2
3
4
5
6
7
8
9
10
11
| # 이렇게 매번 쓰는 건 비효율적
response = llm.invoke("당신은 파이썬 전문가입니다. 함수 어떻게 쓰나요?")
response = llm.invoke("당신은 자바 전문가입니다. 함수 어떻게 쓰나요?")
response = llm.invoke("당신은 파이썬 전문가입니다. 클래스 어떻게 쓰나요?")
# ChatPromptTemplate으로 변수화하면
prompt = ChatPromptTemplate.from_messages([
("system", "당신은 {role} 전문가입니다. 핵심만 3줄로 답변해주세요"),
("human", "{question}")
])
# role과 question만 바꿔서 재사용 가능!
|
기본 사용법
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
| from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages([
("system", "당신은 {role} 전문가입니다. 핵심만 3줄로 답변해주세요"),
("human", "{question}")
])
# 어떤 변수가 필요한지 확인
print(prompt.input_variables) # ['role', 'question']
# 변수 채우기
result = prompt.invoke({
"role" : "파이썬",
"question": "함수 어떻게 쓰나요?"
})
# ChatPromptValue(messages=[SystemMessage(...), HumanMessage(...)])
# LLM에 넘기기
chain = prompt | llm
response = chain.invoke({"role": "파이썬", "question": "함수 어떻게 쓰나요?"})
|
{role}, {question} : 중괄호로 감싼 부분이 변수 prompt.input_variables : 어떤 변수를 채워야 하는지 확인 prompt | llm : 파이프 연산자로 체인 구성 (LCEL)
7. partial() - 변수 사전 고정
개념
- 자주 쓰는 변수를 미리 고정해서 재사용
- partial이 적용된 새로운 프롬프트 반환 (원본 불변)
사용 예시
1
2
3
4
5
6
7
8
9
10
11
12
13
14
| prompt = ChatPromptTemplate.from_messages([
("system", "당신은 {role} 전문가입니다. 핵심만 3줄로 답변해주세요"),
("human", "{question}")
])
# role을 "파이썬"으로 미리 고정
partial_prompt = prompt.partial(role="파이썬")
# 이제 question만 입력하면 됨
response = (partial_prompt | llm).invoke({"question": "함수 어떻게 쓰나요?"})
# input_variables 비교
print(prompt.input_variables) # ['role', 'question']
print(partial_prompt.input_variables) # ['question'] ← role은 고정됨
|
실무에서: 같은 역할의 챗봇에서 role은 고정, question만 변경할 때 유용
PM 챗봇 실전 예시
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
| from langchain_core.output_parsers import StrOutputParser
prompt = ChatPromptTemplate.from_messages([
("system", "당신은 {role}입니다. 액션 아이템을 불릿 3개 이하로 제시해주세요.\n"
"[Context]\n{context}"),
("human", "{question}")
])
parser = StrOutputParser()
pm_chain = prompt.partial(role="프로덕트 매니저") | llm | parser
result = pm_chain.invoke({
"question": "LangChain 워크샵을 90분 세션으로 구성하려 할 때 어떤 활동이 필요한가요?",
"context" : "- 대상: 사내 주니어 엔지니어 25명\n- 장비: 노트북 12대"
})
print(result)
|
8. StrOutputParser - 문자열 파싱
왜 필요한가?
llm.invoke() 반환값 → AIMessage 객체 StrOutputParser 거치면 → 순수 문자열(str)
사용 예시
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| from langchain_core.output_parsers import StrOutputParser
prompt = ChatPromptTemplate.from_messages([
("human", "{topic}에 대해 한 줄로 설명해주세요")
])
parser = StrOutputParser()
# 파이프라인 구성
chain = prompt | llm | parser
# 호출
result = chain.invoke({"topic": "LangChain"})
print(type(result)) # <class 'str'> ← AIMessage가 아닌 순수 문자열!
print(result) # 'LangChain은 LLM 기반 애플리케이션 개발 프레임워크입니다.'
|
StrOutputParser 없이: chain.invoke() → AIMessage 객체 StrOutputParser 붙이면: chain.invoke() → str 문자열 대부분의 경우 StrOutputParser를 체인 마지막에 붙이는 것이 편리
9. JsonOutputParser - JSON 구조 파싱
개념
- LLM에게 JSON 형식으로 응답하도록 지시 + 결과를 dict로 변환
- Pydantic 모델로 스키마 정의 가능
사용 예시
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
| from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field
# JSON 스키마 정의
class CityInfo(BaseModel):
name : str = Field(description="도시이름")
country : str = Field(description="국가")
population : str = Field(description="인구 (대략적)")
famous_for : list[str] = Field(description="유명한 것 3가지")
# 파서 생성
parser = JsonOutputParser(pydantic_object=CityInfo)
# 형식 지시 프롬프트 (LLM에게 JSON 형식 알려줌)
print(parser.get_format_instructions())
# "The output should be formatted as a JSON instance..."
prompt = ChatPromptTemplate.from_messages([
("system", "정확한 정보를 json 형식으로 응답해주세요.\n{format_instructions}"),
("human", "{city}에 대해서 알려주세요")
])
chain = prompt | llm | parser
result = chain.invoke({
"city" : "서울",
"format_instructions": parser.get_format_instructions()
})
print(result)
# {'name': '서울', 'country': '대한민국', 'population': '약 960만',
# 'famous_for': ['경복궁', '남산타워', '한강']}
print(type(result)) # <class 'dict'>
print(result['name']) # '서울'
|
get_format_instructions() : 파서가 LLM에게 “이런 형식으로 출력해”라고 알려주는 문자열 반환값이 dict이므로 result[‘name’] 처럼 딕셔너리로 접근 가능
10. PydanticOutputParser - 타입 검증 파싱
JsonOutputParser와의 차이
- JsonOutputParser → dict 반환
- PydanticOutputParser → Pydantic 모델 객체 반환 (속성 접근 + 타입 검증)
사용 예시
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
| from langchain_core.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
class BookReview(BaseModel):
title : str = Field(description="책 제목")
author : str = Field(description="저자")
rating : int = Field(description="평점 (1-5)", ge=1, le=5) # 1~5 제약
summary : str = Field(description="한줄요약")
recommended: bool = Field(description="추천 여부")
parser = PydanticOutputParser(pydantic_object=BookReview)
prompt = ChatPromptTemplate.from_messages([
("system", "도서 리뷰를 작성해주세요.\n{format_instructions}"),
("human", "{book} 책에 대한 리뷰를 작성해주세요")
])
chain = prompt | llm | parser
result = chain.invoke({
"book" : "장발장",
"format_instructions": parser.get_format_instructions()
})
print(result)
# BookReview(title='장발장', author='빅토르 위고', rating=5, ...)
print(type(result)) # <class '__main__.BookReview'> Pydantic 객체!
print(result.title) # '장발장' (속성 접근)
print(result.author) # '빅토르 위고'
print(result.rating) # 5
print(result.recommended) # True
|
ge=1, le=5 : Pydantic 제약 조건 (1 이상 5 이하, 벗어나면 ValidationError) result.title 처럼 점(.) 으로 속성 접근 (dict[‘key’] 보다 편리)
11. 파이프라인(|) 연산자 - 체인 구성
기본 구조
1
2
3
4
| Prompt | LLM | Parser
= Prompt의 출력 → LLM의 입력
LLM의 출력 → Parser의 입력
|
단계별 역할
| 단계 | 입력 | 출력 |
|---|
| Prompt | 딕셔너리 {변수: 값} | ChatPromptValue (messages 리스트) |
| LLM | ChatPromptValue | AIMessage (응답 객체) |
| Parser | AIMessage | str / dict / Pydantic 객체 |
예시 모음
1
2
3
4
5
6
7
8
9
10
11
| # 단순 텍스트 체인
chain1 = prompt | llm | StrOutputParser()
result = chain1.invoke({"topic": "LangChain"}) # → str
# JSON 체인
chain2 = prompt | llm | JsonOutputParser(pydantic_object=CityInfo)
result = chain2.invoke({"city": "서울"}) # → dict
# Pydantic 체인
chain3 = prompt | llm | PydanticOutputParser(pydantic_object=BookReview)
result = chain3.invoke({"book": "장발장"}) # → BookReview 객체
|
| 연산자가 없었다면:
1
2
3
4
5
| # LCEL 없이 직접 구현한다면 이렇게 해야 함
prompt_value = prompt.invoke({"topic": "LangChain"})
llm_response = llm.invoke(prompt_value)
final_result = parser.invoke(llm_response)
# LCEL: chain = prompt | llm | parser → chain.invoke({"topic": "LangChain"})
|
12. 핵심 개념 총정리
| 개념 | 핵심 내용 |
|---|
| ChatOpenAI | OpenAI API 래퍼, llm.invoke() |
| invoke/stream/batch | 단건/스트리밍/다중 병렬 처리 |
| llm.bind() | 파라미터 고정한 새 LLM 인스턴스 생성 |
| ChatPromptTemplate | 변수 {role} {question} 있는 프롬프트 템플릿 |
| partial() | 자주 쓰는 변수 미리 고정 |
| StrOutputParser | AIMessage → str 변환 |
| JsonOutputParser | AIMessage → dict 변환 (스키마 검증) |
| PydanticOutputParser | AIMessage → Pydantic 객체 변환 (타입 검증) |
| | 연산자 (LCEL) | prompt | llm | parser 파이프라인 구성 |
| get_format_instructions | LLM에게 출력 형식을 알려주는 지시문 |
Parser 비교
| Parser | 반환 타입 | 언제 쓰나? |
|---|
| StrOutputParser | str | 대부분의 경우 (텍스트 응답) |
| JsonOutputParser | dict | JSON 데이터 추출 |
| PydanticOutputParser | Pydantic 객체 | 타입 검증 필요, 속성 접근 편의 |
13. 자주 나오는 실수 / 주의사항
실수 1: LangChain 응답에서 .choices[0].message.content 접근
1
2
3
4
5
6
7
8
9
10
| # LangChain llm.invoke() 반환값은 AIMessage!
# OpenAI 직접 호출과 다름!
response = llm.invoke("질문")
# 틀린 코드 (OpenAI 직접 호출 방식)
text = response.choices[0].message.content # AttributeError!
# 올바른 코드 (LangChain)
text = response.content
|
1
2
3
4
5
6
7
8
9
| # JsonOutputParser/PydanticOutputParser는 LLM에게 형식을 알려줘야 함!
# format_instructions 없이 쓰면 LLM이 형식을 모름 → 파싱 실패
# 올바른 코드
chain = prompt | llm | parser
result = chain.invoke({
"city" : "서울",
"format_instructions": parser.get_format_instructions() # ← 필수!
})
|
실수 3: partial()과 invoke()의 변수 혼동
1
2
3
4
5
6
7
8
9
10
11
12
| prompt = ChatPromptTemplate.from_messages([
("system", "당신은 {role} 전문가입니다"),
("human", "{question}")
])
partial_prompt = prompt.partial(role="파이썬")
# role은 이미 고정됐으므로 invoke에서 role 또 주면 에러 or 무시됨
result = (partial_prompt | llm).invoke({
"question": "함수 어떻게 쓰나요?"
# role 은 이미 고정됨 → 여기서 주지 않아도 됨
})
|
실수 4: batch() 결과 순서
1
2
3
4
5
6
| questions = ["질문1", "질문2", "질문3"]
results = llm.batch(questions)
# 병렬 처리지만 순서는 입력과 동일하게 유지됨
for q, r in zip(questions, results):
print(f"Q: {q} → A: {r.content[:30]}")
|
끝