LangChain Output Parser
아웃풋 형식 지정하기 : Output Parser
llm 이 내뱉는 답변 자체를 서비스에서 마지막 단으로 제공하는 워크플로우는 이제 흔하지 않다. ai 프로덕트가 다양해지면서 워크플로우가 복잡해지고 있으므로, 1번 노드에서 받은 응답을 가공해서 2번에 인풋으로 넣고.. 하는 과정이 많아지기 때문이다. 이런 과정에서 언제나 모델에 맞춰서 다시 파싱하고.. 타입 유효성 검증하고.. 하는 과정을 손수 하나하나 만드는건 정말 쉽지 않다. 그래서 아웃풋 파서는 랭체인 프레임워크에서 유용하게 잘 쓰고 있는 기능이다.
Pydantic Output Parser 사용법
PydanticOutputParser는 langchain의 output parser 중 하나로 LLM 출력을 파이썬 객체로 변환한다.
# 출력하게 할 형식을 pydantic 모델로 지정
from pydantic import BaseModel, Field
class City(BaseModel):
country: str = Field(description = "The country the city is in")
city: str = Field(description = "The city name")
# pydantic 모델을 사용하는 프롬프트 작성
from langchain_core.output_parsers import PydanticOutputParser
output_parser = PydanticOutputParser(pydantic_object = City)
format_instructions = output_parser.get_format_instructions()
print(format_instructions)
from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages([
("system",
"사용자가 입력한 국가의 수도를 출력해주세요.",
"{format_instructions}"),
("human", "{input}")
])
prompt_value = prompt.partial(
format_instructions = format_instructions
)
- output 형식을 pydantic object로 변환하는 과정에서 자연스럽게 구조화, 형식 검증을 수행하게 된다.
-> 결과 확인해보기
prompt.invoke({"input": "France"})
print(prompt_value.messages[0].content)
print(prompt_value.messages[1].content)
String Output Parser 사용법
LLM의 출력을 텍스트로 변환하는데 사용된다.
from langchain_core.output_parsers import StrOutputParser
from langchain_core.messages import AIMessage, HumanMessage
output_parser = StrOutputParser()
ai_message = AIMessage(content = "The capital of France is Paris.")
ai_message = output_parser.invoke(ai_message)
print(ai_message)
JsonOutputParser
from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field
class Recipe(BaseModel):
name: str = Field(description="요리 이름")
ingredients: list[str] = Field(description="재료 목록")
steps: list[str] = Field(description="조리 과정")
parser = JsonOutputParser(pydantic_object=Recipe)
# 프롬프트에 포맷 지시사항 자동 추가
prompt = ChatPromptTemplate.from_template(
"사용자 질문: {query}\n{format_instructions}"
)
pydantic output parser 랑 json output parser의 차이점
PydanticOutputParser:
- Pydantic 모델 기반 검증
- 타입 안전성 보장
- 복잡한 중첩 구조 지원
- 자동 필드 검증
JsonOutputParser:
- 단순 JSON 파싱
- Pydantic 모델 선택적 사용
- 더 빠른 처리 속도
- 유연한 구조
CommaSeparatedListOutputParser
from langchain_core.output_parsers import CommaSeparatedListOutputParser
output_parser = CommaSeparatedListOutputParser()
format_instructions = output_parser.get_format_instructions()
# 결과: "Your response should be a list of comma separated values, eg: `foo, bar, baz`"
DatetimeOutputParser
from langchain_core.output_parsers import DatetimeOutputParser
output_parser = DatetimeOutputParser()
# ISO 8601 형식의 datetime 객체로 변환
2. Custom Output Parser 만들기
from langchain_core.output_parsers import BaseOutputParser
from langchain_core.exceptions import OutputParserException
from typing import List
import re
class EmailListParser(BaseOutputParser[List[str]]):
"""이메일 주소 목록을 파싱하는 커스텀 파서"""
def parse(self, text: str) -> List[str]:
# 이메일 패턴 정규식
email_pattern = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'
emails = re.findall(email_pattern, text)
if not emails:
raise OutputParserException(f"No valid emails found in: {text}")
return emails
def get_format_instructions(self) -> str:
return "응답에는 유효한 이메일 주소들을 포함해주세요."
# 사용 예시
email_parser = EmailListParser()
- 응답을 파싱하는걸 랭체인의 runnable 객체로 사용할 수 있다. 랭체인 프레임워크 밖에서 이런 작업을 하려고 하면 응답의 유효성부터 예외처리까지 다 하나하나 만들어줘야겟지 . . .
3. Output Parser와 Pydantic 고급 활용
from pydantic import BaseModel, Field, validator
from typing import Optional, List
from enum import Enum
class Priority(str, Enum):
HIGH = "high"
MEDIUM = "medium"
LOW = "low"
class Task(BaseModel):
title: str = Field(description="작업 제목")
description: str = Field(description="작업 설명")
priority: Priority = Field(description="작업 우선순위")
due_date: Optional[str] = Field(description="마감일 (YYYY-MM-DD)")
tags: List[str] = Field(description="작업 태그 목록")
@validator('due_date')
def validate_date_format(cls, v):
if v is not None:
import datetime
try:
datetime.datetime.strptime(v, '%Y-%m-%d')
except ValueError:
raise ValueError('날짜는 YYYY-MM-DD 형식이어야 합니다')
return v
2. Format Instructions의 중요성
# 잘못된 예시
prompt = ChatPromptTemplate.from_template("날씨 정보를 JSON으로 주세요")
# 올바른 예시
prompt = ChatPromptTemplate.from_template(
"날씨 정보를 다음 형식으로 주세요:\n{format_instructions}\n질문: {query}"
)
prompt = prompt.partial(format_instructions=parser.get_format_instructions())
두 가지 방식 모두 형식에 대한 인스트럭션을 제공합니다. 그렇지만 강제성이 있는 방식이 언제나 더 낫습니다.. llm은 예측할 수 없는 결과를 보여주기도 하니까요…
3. 파싱 에러 처리 전략
from langchain_core.output_parsers import OutputFixingParser
from langchain_openai import ChatOpenAI
# 자동 수정 파서
fixing_parser = OutputFixingParser.from_llm(
parser=PydanticOutputParser(pydantic_object=Recipe),
llm=ChatOpenAI(temperature=0)
)
# 재시도 파서
from langchain_core.output_parsers import RetryOutputParser
retry_parser = RetryOutputParser.from_llm(
parser=PydanticOutputParser(pydantic_object=Recipe),
llm=ChatOpenAI(temperature=0)
)
try:
result = retry_parser.parse_with_prompt(malformed_output, original_prompt)
except Exception as e:
# 최종 실패 처리
print(f"파싱 실패: {e}")
4. 성능 최적화 팁
1. 간단한 구조 선호:
# 복잡한 중첩보다는 평면적 구조
class SimpleTask(BaseModel):
title: str
priority: str
completed: bool = False
2. 선택적 필드 활용:
class FlexibleResponse(BaseModel):
required_field: str
optional_field: Optional[str] = None # LLM이 생략 가능
3. Enum 활용으로 값 제한:
class Status(str, Enum):
PENDING = "pending"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
5. 실제 프로덕션 환경에서의 주의사항
토큰 비용 고려:
- 복잡한 format_instructions는 토큰을 많이 소모
- 자주 사용되는 파서는 캐싱 고려
에러 복구:
def robust_parse(parser, text, max_attempts=3):
for attempt in range(max_attempts):
try:
return parser.parse(text)
except OutputParserException as e:
if attempt == max_attempts - 1:
# 마지막 시도 실패시 기본값 또는 부분 파싱 결과 반환
return extract_partial_data(text)
# 다음 시도를 위한 텍스트 전처리
text = preprocess_for_retry(text)
로깅과 모니터링:
- 파싱 실패율 추적
- 자주 실패하는 패턴 분석
- LLM 응답 품질 개선을 위한 프롬프트 조정