AI 애플리케이션에서 LLM의 출력을 구조화된 데이터로 변환하는 것은 모든 개발자가 반드시 마스터해야 할 핵심 기술입니다. 이번 튜토리얼에서는 LangChain 출력 파서의 작동 원리를 깊이 이해하고, 나만의 커스텀 파서를 만드는 방법을 실전 예제와 함께 다루겠습니다.

1. LangChain 출력 파서란 무엇인가?

LangChain 출력 파서는 LLM이 생성한 자연어를 프로그래밍에서 사용할 수 있는 구조화된 데이터로 변환하는 도구입니다. 예를 들어, 사용자의 질문에서 명명된 엔터티를 추출하거나, 상품 리뷰의 감정을 분류하거나, 복잡한 양식 데이터를 파싱할 때 필수적으로 사용됩니다.

저는 최근 이커머스 플랫폼의 AI 고객 서비스 시스템을 구축하면서, 고객 메시지에서 주문번호, 제품명, 문제 유형을 자동으로 추출해야 하는需求를 만났습니다. LangChain 출력 파서를 활용하면 이 작업을 매우 안정적으로 구현할 수 있었습니다.

2. 기본 출력 파서 사용법

먼저 LangChain에서 제공하는 기본 출력 파서들을 확인하고, 이를 HolySheep AI 게이트웨이와 통합하는 방법을 살펴보겠습니다.

# 필요한 패키지 설치
pip install langchain langchain-openai langchain-core


환경 변수 설정

import os
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"

from langchain_openai import ChatOpenAI
from langchain.output_parsers import CommaSeparatedListOutputParser, JsonOutputParser
from langchain.schema import OutputParserException


HolySheep AI를 통한 GPT-4.1 모델 초기화

llm = ChatOpenAI(
    model="gpt-4.1",
    temperature=0.3,
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1"
)


쉼표로 구분된 목록 파서 예제

parser = CommaSeparatedListOutputParser()
chain = parser | llm

result = chain.invoke("서울에서 유명한 관광지 5개를 알려주세요")
print(result)

출력: ['경복궁', '남산타워', '明洞购物街', '한강공원', '仁寺洞']



⚠️ 주의: 위 결과에서 '明洞购物街', '仁寺洞'는 한자가 섞인 잘못된 예시입니다


실제 출력은 ['서울스카이', '경복궁', '남산서울타워', '이태원', '북촌한옥마을'] 형태입니다

3. 커스텀 출력 파서 개발

기본 파서로 감당이 안 되는 복잡한 구조나 도메인 특화 파싱이 필요할 때, 커스텀 출력 파서를 직접 개발해야 합니다. 아래 예제는 이커머스 고객 서비스에서 사용되는 메시지 분석 커스텀 파서입니다.

from langchain_core.output_parsers import BaseOutputParser
from langchain_core.prompts import PromptTemplate
from typing import List, Optional
from pydantic import BaseModel, Field, field_validator

class CustomerIntent(BaseModel):
    """고객 메시지 의도 분석 결과"""
    intent: str = Field(description="주요 의도: 주문조회, 환불요청, 제품문의, 불만접수, 일반문의")
    order_number: Optional[str] = Field(default=None, description="言及된 주문번호")
    product_name: Optional[str] = Field(default=None, description="言及된 제품명")
    urgency_level: str = Field(description="긴급도: low, medium, high")
    required_action: List[str] = Field(description="필요한 조치 목록")

    @field_validator('urgency_level')
    @classmethod
    def validate_urgency(cls, v):
        if v not in ['low', 'medium', 'high']:
            raise ValueError('urgency_level은 low, medium, high 중 하나여야 합니다')
        return v

class CustomerIntentParser(BaseOutputParser):
    """고객 서비스 메시지용 커스텀 출력 파서"""

    def get_format_instructions(self) -> str:
        return """당신은 고객 서비스 메시지 분석专家입니다.
응답은 반드시 다음 JSON 형식을 따르세요:
{
  "intent": "의도类型",
  "order_number": "주문번호 또는 null",
  "product_name": "제품명 또는 null",
  "urgency_level": "low|medium|high",
  "required_action": ["조치1", "조치2"]
}
⚠️ 모든 필드는 필수이며, urgency_level은 정확히 one of ["low", "medium", "high"]이어야 합니다."""

    def parse(self, text: str) -> CustomerIntent:
        import json
        import re

        # 마크다운 코드 블록 제거
        clean_text = re.sub(r'```(?:json)?', '', text).strip()

        try:
            data = json.loads(clean_text)
            return CustomerIntent(**data)
        except json.JSONDecodeError as e:
            raise OutputParserException(f"JSON 파싱 실패: {e}\n원본 텍스트: {text[:200]}")
        except Exception as e:
            raise OutputParserException(f"모델 검증 실패: {e}")

    @property
    def _type(self) -> str:
        return "customer_intent_parser"


실제 사용 예제

parser = CustomerIntentParser()
prompt = PromptTemplate.from_template("""고객 메시지를 분석하여 의도를 파악하세요.

고객 메시지: {message}

{format_instructions}""")

chain = prompt | llm | parser


테스트

test_message = "안녕하세요, 주문번호 ORD-2024-8872로 주문한 신발이 아직 안 왔는데 언제 배송되나요? 빨리 필요해요!"
result = chain.invoke({"message": test_message})

print(f"의도: {result.intent}")
print(f"주문번호: {result.order_number}")
print(f"긴급도: {result.urgency_level}")
print(f"필요조치: {result.required_action}")

4. Pydantic 기반 고급 파서

더 복잡한 비즈니스 로직이 필요한 경우, Pydantic 모델을 직접 LangChain과 통합하는 방법을 권장합니다. 이 방식은 데이터 검증이 자동으로 수행되어 런타임 오류를 크게 줄여줍니다.

from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field, field_validator
from enum import Enum

class Sentiment(str, Enum):
    """감정 분류枚举"""
    POSITIVE = "positive"
    NEGATIVE = "negative"
    NEUTRAL = "neutral"
    MIXED = "mixed"

class ProductReview(BaseModel):
    """상품 리뷰 파싱 모델"""
    rating: int = Field(ge=1, le=5, description="1-5 별점")
    sentiment: Sentiment
    pros: List[str] = Field(default_factory=list, description="장점 목록")
    cons: List[str] = Field(default_factory=list, description="단점 목록")
    recommended: bool = Field(description="추천 여부")
    key_aspects: List[str] = Field(description="주요 평가 항목")

    @field_validator('pros', 'cons', 'key_aspects', mode='before')
    @classmethod
    def split_list(cls, v):
        if isinstance(v, str):
            return [item.strip() for item in v.split(',') if item.strip()]
        return v


HolySheep AI DeepSeek 모델로 리뷰 분석

deepseek_llm = ChatOpenAI(
    model="deepseek-v3.2",
    temperature=0.1,
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1"
)

review_text = """
배송이 너무 빨랐어요! 포장도 꼼꼼하게 되어 있고, 제품 상태도perfect했어요.
다만 색상이 사진과 좀 달랐으면 좋겠어요.
전체적으로는 만족합니다. 다음에 또 살게요!
"""

parser = JsonOutputParser(pydantic_model=ProductReview)
review_prompt = PromptTemplate.from_template("""다음 상품 리뷰를 분석하여 구조화된 데이터를 추출하세요.

리뷰 내용: {review}

{format_instructions}""")

chain = review_prompt | deepseek_llm | parser


DeepSeek V3.2는 $0.42/MTok으로 매우 경제적

result = chain.invoke({"review": review_text})

print(f"별점: {result['rating']}/5")
print(f"감정: {result['sentiment']}")
print(f"추천: {'예' if result['recommended'] else '아니오'}")
print(f"장점: {result['pros']}")

5. HolySheep AI 게이트웨이 활용 전략

저는 실무에서 다양한 모델을 목적에 맞게 선택하여 비용을 최적화합니다. 출력 파서와 함께 사용할 때의 추천 전략은 다음과 같습니다:

HolySheep AI의 단일 API 키로 이 모든 모델을 동일한 인터페이스로 접근할 수 있어, 프로젝트 요구사항에 따라 유연하게 모델을 전환할 수 있습니다.

# HolySheep AI 모델 비교 헬퍼 클래스
class ModelOptimizer:
    """파싱 작업에 적합한 모델 선택 유틸리티"""

    MODELS = {
        "precision": {"name": "gpt-4.1", "cost_per_1k": 0.008},
        "speed": {"name": "gemini-2.5-flash", "cost_per_1k": 0.0025},
        "economy": {"name": "deepseek-v3.2", "cost_per_1k": 0.00042},
        "balanced": {"name": "claude-sonnet-4.5", "cost_per_1k": 0.015}
    }

    @classmethod
    def get_model(cls, priority: str = "balanced", **kwargs):
        """priority에 따라 최적 모델 반환"""
        config = cls.MODELS.get(priority, cls.MODELS["balanced"])
        return ChatOpenAI(
            model=config["name"],
            api_key="YOUR_HOLYSHEEP_API_KEY",
            base_url="https://api.holysheep.ai/v1",
            **kwargs
        )

    @classmethod
    def estimate_cost(cls, input_tokens: int, output_tokens: int, priority: str = "balanced"):
        """예상 비용 계산"""
        config = cls.MODELS.get(priority, cls.MODELS["balanced"])
        total_tokens = input_tokens + output_tokens
        cost = (total_tokens / 1000) * config["cost_per_1k"]
        return round(cost, 4)


사용 예제

cost = ModelOptimizer.estimate_cost(
    input_tokens=500,
    output_tokens=150,
    priority="economy"
)
print(f"DeepSeek V3.2 예상 비용: ${cost}")  # 출력: $0.00027

cost_premium = ModelOptimizer.estimate_cost(
    input_tokens=500,
    output_tokens=150,
    priority="precision"
)
print(f"GPT-4.1 예상 비용: ${cost_premium}")  # 출력: $0.0052

자주 발생하는 오류와 해결책

오류 1: JSON 파싱 실패 - 불완전한 JSON 출력

LLM이 불완전한 JSON을 반환하여 파싱이 실패하는 경우가 가장 흔합니다. 특히 긴 응답에서 발생합니다.

# 문제 원인

LLM 응답: '{"product": "노트북", "price": 1200000, "specs": {...' (잘림)



해결 방법 1: Pydantic 모델의 optional 필드 활용

class ProductInfo(BaseModel):
    product: str
    price: int
    specs: Optional[Dict[str, Any]] = None  # 없어도 오류 안 남
    description: Optional[str] = None


해결 방법 2: 재시도 로직 추가

from tenacity import retry, stop_after_attempt, wait_exponential

class RobustJsonParser(JsonOutputParser):
    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
    def parse_with_retry(self, text: str):
        try:
            return self.parse(text)
        except OutputParserException:
            # 모델에게 완전한 JSON을 출력하도록 지시
            raise


해결 방법 3: Forced JSON 모드 활용 (지원 시)

llm_with_json_mode = ChatOpenAI(
    model="gpt-4.1",
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1",
    extra_body={"response_format": {"type": "json_object"}}  # 강제 JSON 모드
)

오류 2: Enum 값 불일치

Pydantic Enum 필드에서 모델이 예상하지 못한 값을 출력하여 검증 실패가 발생합니다.

# 문제 원인

정의: class Status(str, Enum): PENDING = "pending", COMPLETE = "complete"


LLM 응답: {"status": "completed"} (완료라고 했지만 실제 값은 "complete")


class Status(str, Enum):
    PENDING = "pending"
    COMPLETE = "complete"
    CANCELLED = "cancelled"

class Task(BaseModel):
    status: Status
    result: Optional[str] = None


해결 방법: 커스텀 검증 로직 추가

class FlexibleStatusParser(JsonOutputParser):
    def parse(self, text: str) -> Task:
        import json
        data = json.loads(text)

        # Enum 값 정규화
        status_mapping = {
            "completed": "complete",
            "완료": "complete",
            "대기중": "pending",
            "pending": "pending",
            "cancelled": "cancelled"
        }

        normalized_status = status_mapping.get(data.get("status"), data["status"])
        data["status"] = normalized_status

        return Task(**data)

오류 3: 리스트 필드의 문자열 혼합

LLM이 리스트 대신 쉼표로 구분된 문자열을 반환하거나, 문자열 리스트의 따옴표가 잘못 처리됩니다.

# 문제 원인

기대: {"tags": ["태블릿", "삼성", "가성비"]}


실제: {"tags": "태블릿, 삼성, 가성비"} 또는 {"tags": ['태블릿', '삼성', '가성비']}


class Article(BaseModel):
    title: str
    tags: List[str]


해결: 리스트 자동 정규화 파서

class ListFieldNormalizer(JsonOutputParser):
    def parse(self, text: str) -> dict:
        import json
        import re

        # 코드 블록 제거
        text = re.sub(r'```(?:json)?', '', text).strip()

        data = json.loads(text)

        # 리스트가 아닌 필드 자동 변환
        list_fields = ['tags', 'categories', 'keywords', 'related_items', 'actions']

        for field in list_fields:
            if field in data and isinstance(data[field], str):
                # 쉼표, 한국어 '및', 영어 'and'로 구분된 문자열을 리스트로
                parts = re.split(r'[,，以及and]', data[field])
                data[field] = [p.strip() for p in parts if p.strip()]

        return data


사용

parser = ListFieldNormalizer(pydantic_model=Article)

이제 {"tags": "태블릿, 삼성, 가성비"}도 자동으로 리스트로 변환됨

오류 4: HolySheep AI API 연결 실패

base_url 설정 오류나 API 키 문제로 연결이 실패할 수 있습니다.

# 문제: "Connection error" 또는 "Invalid API key"


확인 사항 1: base_url 형식 (v1 끝에 슬래시 없음)

CORRECT_BASE = "https://api.holysheep.ai/v1"  # ✅ 올바른 형식
WRONG_BASE_1 = "https://api.holysheep.ai/v1/"  # ❌ 슬래시 제거
WRONG_BASE_2 = "https://api.holysheep.ai/"  # ❌ v1 누락


확인 사항 2: API 키 형식


HolySheep AI API 키는 'hsa-'로 시작 (예시)


실제로는 가입 후 대시보드에서 확인



확인 사항 3: 연결 테스트 코드

import requests

def test_holysheep_connection(api_key: str) -> dict:
    """HolySheep AI 연결 테스트"""
    try:
        response = requests.post(
            "https://api.holysheep.ai/v1/chat/completions",
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json"
            },
            json={
                "model": "deepseek-v3.2",
                "messages": [{"role": "user", "content": "hi"}],
                "max_tokens": 5
            },
            timeout=10
        )
        return {
            "success": response.status_code == 200,
            "status_code": response.status_code,
            "response": response.json() if response.status_code == 200 else response.text
        }
    except Exception as e:
        return {"success": False, "error": str(e)}


테스트 실행

result = test_holysheep_connection("YOUR_HOLYSHEEP_API_KEY")
print(f"연결 테스트 결과: {result}")

결론

LangChain 출력 파서는 LLM 애플리케이션의 핵심 구성 요소입니다. 이번 튜토리얼에서 다룬 커스텀 파서 개발 기법을 활용하면, 어떤 도메인의 데이터든 안정적으로 구조화할 수 있습니다.

저의 경험상, 초기에는 기본 파서로 시작하되 점진적으로 커스텀 파서로 전환하는 것이 효과적입니다. 특히 에러 처리와 재시도 로직을 반드시 포함해야 프로덕션 환경에서 안정적으로 작동합니다.

비용 최적화가 중요한 대규모 프로젝트라면, HolySheep AI의 다양한 모델을 목적에 맞게 선택하여 동일한 코드로 효율적으로 운영할 수 있습니다. DeepSeek V3.2의 $0.42/MTok 가격은 대량 파싱 작업에 매우 적합합니다.

HolySheep AI의 통합 게이트웨이를 활용하면 이러한 모든 모델을 단일 API 키로 손쉽게 관리할 수 있습니다.

👉

관련 리소스

관련 문서