Batch Normalization

https://arxiv.org/pdf/1502.03167

Background

batch normalizaion 은 2015년에 제시된 ICS(Internal Covariate Shift) 문제를 줄일 수 있는 아이디어입니다. covariate shift 는 학습 때 활용한 데이터가 실제 추론에 사용되는 데이터간의 분포가 다르면 추론 성능에 악영향을 미칠 수 있다라는 주장인데 이게 신경망 내부에서도 발생할 것이다 라는 주장을 하며 생긴용어가 Internal Covariate Shift 라고 합니다. 아래 사진을 보면 직관적으로 이해가 될 것 같습니다. 신경망을 통과하면서 데이터의 분포가 달라지는 현상이 발생하는데

통과하는 레이어 수가 많아질수록 그 정도가 심해지기 때문에 당연히 추론이나 학습 성능에 문제가 생길 확률이 큽니다. Batch Normalizaion 은 기존의 정규화 과정에서 학습데이터마다 분포가 다른것을 배치별로 평균과 분산을 활용해 정규화하는 것 입니다.

나동빈님의 영상을 참고하여 알게 된 batch normalizaion가 현실에서는 하이퍼파라미터 의존도를 줄였으며, 학습속도를 향상시키고, 모델이 일반적으로 즉, 학습데이터에만 태스크를 잘 처리하도록 하는것이 아닌 실제 현상을 잘 반영시키게 된 효과가 있었다고 합니다.

그런데 논문에서는 ics 를 감소시킨다고 주장하였으나 실제로 증명하지는 못했다고 합니다. 그래서 그것을 증명하기 위한 How Does Batch Normalization Help Optimization?  라는 논문이 나왔습니다.

https://arxiv.org/pdf/1805.11604

우선 일반적으로 Batch Norm 을 적용시킨 네트워크가 Accuracy 가 가파른 폭으로 올라갔다는 것을 보여줍니다.

우측의 히스토그램을 보면 각 레이어의 분포를 나타내고 있는데요 가장우측의 Standard + Noisy BatchNorm 에서 Layer3 부터 분포가 갑작스럽게 변하여 ICS가 발생하고 있음을 볼 수 있습니다. ICS가 발생하고 있음에도 불구하고 왼쪽 그래프를 보면 학습성능이 우수함을 볼 수 있습니다.

즉 임의로 Batch Norm Layer 이후 바로 Noise 를 넣어 covariate shift 를 발생시켰을 때에도 BatchNorm 이 포함된 네트워크는 일반적인 네트워크보다 성능이 우수함을 보였습니다. 그래서 실험적으로 Batch Norm 이 ICS 문제를 해소할 수 있다는 이전 논문의 반박을 하였고, 심지어 ICS가 크게 발생함에도 불구하고 Batch Norm 이 있으면 성능이 좋아진다는 것을 보여준 사례가 되었습니다.

해당논문에서 ICS를 파라미터의 기울기 계산하여 ICS를 계산하는 방법을 제안했는데, 포스팅의 목적보다 너무 벗어나는것 같아 다루지 않겠습니다. 궁금하신분께서는 논문을 참고하시면 될 것 같습니다.

그렇다면 ICS 를 해소하지 못했음에도 불구하고 성능이 좋은 이유는 뭘까요? 논문에서는 Batch Norm 의 Smoothing 효과 때문이라고 설명합니다.

Loss Landscape 가 훨씬 더 예상 가능한 범위로 형성되면서 학습효과가 증대된다고 말하고 있습니다.

Batch Normalization Layer

미니배치의 평균값과 분산을 구해서 normalizaion 을 수행할 수 있습니다. 그리고 감마와 베타를 활용해 실제 output 을 내는데요, 여기서 감마와 베타가 실제 학습에 활용되는 파라미터입니다. 학습중에는 loss 를 최소화 하는 방향으로 감마와 베타를 찾아갈 것 입니다.

정규화에서 학습 파라미터를 사용하는 이유는 활성화 함수의 특징에 있습니다. sigmoid를 예시로 들면 어떤 구간에서는 매우 선형적으로 작동하기 때문에 표준정규분포로 정규화한 0과 1사이의 값에서 선형적으로 작동하게 됩니다. 그래서 감마와 베타를 활용해 non-linearity 를 지켜주고, 해당 정규화 레이어의 output 도 적절하게 내보낼 수 있게됩니다. 결론은 레이어의 입력을 정규화할 때는 linearity 를 주의해서 정규화 해야한다는 점 입니다.

Batch Normalization Layer 연산구분

batch normalization Layer 는 학습할때와 추론할 때 네트워크에서의 역할이 달라집니다. 학습할때 감마와 베타 파라미터를 학습시켜야 하지만 추론때에는 필요없습니다. 따라 해당 파라미터들을 고정하여 학습된 파라미터에 의한 값이 나와야합니다.

step 7 에서부터는 BN 이 training 모드로 네트워크에 있었던 것을 inference 모드로 바꿉니다. ( 파라미터 고정을 통해서 )

Batch Normalization Data Flow

입력 데이터 (X)

$$

X = \begin{bmatrix} [1,\ 2] \ [2,\ 4] \ [3,\ 6] \end{bmatrix}

$$

배치로 들어온 데이터

shape: (3, 2)

→ 샘플 3개, 각 샘플은 2차원 벡터

Linear Layer 통과

가중치와 bias를 이렇게 두겠습니:

$$ [ W = \begin{bmatrix} [1,0], \ [0,1] \end{bmatrix}, \quad b = [0,\ 0] ] $$

즉, 아무 변화 없는 선형층

$$ [ Z = XW + b = X ] $$

결과:

Z =
[
 [1, 2],
 [2, 4],
 [3, 6]
]

shape 그대로 (3, 2)

Batch Normalization

1️⃣ Batch Mean (μ)

feature별 평균:

$$ μ=[(1+2+3)/3, (2+4+6)/3]=[2, 4] $$

2️⃣ Batch Variance (σ²)

$$ σ2=[((1−2)2+(2−2)2+(3−2)2)/3,((2−4)2+(4−4)2+(6−4)2)/3]=[2/3, 8/3] $$

3️⃣ Normalize (x̂)

$$ \hat{x} = \frac{x - \mu}{\sqrt{\sigma^2 + \epsilon}} (ε 무시한다고 가정) $$

샘플별 계산

첫 번째 샘플

$$ [1,2] → [-1/\sqrt{2/3},\ -2/\sqrt{8/3}] ≈ [-1.22,\ -1.22] $$

두 번째

$$ [2,4] → [0,\ 0] $$

세 번째

$$ [3,6] → [1.22,\ 1.22] $$

결과:

X_hat =
[
 [-1.22, -1.22],
 [ 0.00,  0.00],
 [ 1.22,  1.22]
]

그리고 해당값에 gamma 와 betta 연산을 통해 Layer 를 통과시킵니다. 이처럼 batch norm 은 미니 배치의 피처별로 평균, 분산을 구해서 원본 데이터에 대입시키는 방법으로 Normalizaion 을 수행하게 됩니다.

Layer Normalization

arxiv.org

Layer Normalization 은 Batch Norm 이 RNN 에 적용하기 어려운 문제점을 해소하기 위해 제시된 방법입니다. RNN은 시간단위로 계산을 합니다. 따라서 미니배치의 각 피쳐마다 통계를 이용해 정규화하는 BN 의 경우에는 해당 스트림의 맥락을 반영하지 못합니다.

가장 큰 문제는 RNN 이나 NLP, 혹은 음성데이터의 경우는 배치마다 길이가 다릅니다.

샘플 1: "나는 밥을 먹었다"        (길이 4)
샘플 2: "오늘"                    (길이 1)
샘플 3: "어제 비가 와서 우산을 썼다" (길이 6)

이것을 BN 을 활용한 Layer output 을 사용한다면 샘플2 의 2,3 샘플1의 3,4 가 0이 됩니다. 그렇기 때문에 데이터의 의미를 충분히 반영하지 못하는 문제가 발생합니다. 이 문제는 시계열 데이터에도 그대로 적용됩니다. 이미지나 성적통계(국어는 국어끼리, 수학은 수학끼리) 와 같은 데이터가 아니라 피쳐하나가 다른 피쳐나 데이터에도 영향을 주는경우는 Batch 사이즈에 영향을 받지 않고 데이터의 의미를 잘 반영할 수 있는 LN 이 성능이 좋다고 주장합니다.

BN 과의 차이점

Batch Normalization은 미니배치 단위로 평균과 분산을 계산하여 정규화를 수행합니다. 반면 **Layer Normalization(LN)**은 이름 그대로 레이어 단위, 정확히는 하나의 샘플 내부 feature들에 대해서만 정규화를 수행합니다. 즉, 정규화의 기준이 완전히 다릅니다.

• Batch Normalization
  • 평균, 분산 계산 축: batch 방향
  • 같은 feature를 가진 여러 샘플을 함께 사용
• Layer Normalization
  • 평균, 분산 계산 축: feature 방향
  • 하나의 샘플 안에서만 계산

하나의 샘플 x = [x₁, x₂, ..., xₐ]에 대해:

$$ \mu = \frac{1}{d} \sum_{i=1}^{d} x_i $$

$$ \sigma^2 = \frac{1}{d} \sum_{i=1}^{d} (x_i - \mu)^2 $$

$$ \hat{x}_i = \frac{x_i - \mu}{\sqrt{\sigma^2 + \epsilon}} $$

그리고 Batch Normalization과 동일하게 scale, shift 파라미터를 적용합니다:

$$ y_i = \gamma_i \hat{x}_i + \beta_i $$

여기서 중요한 점은 γ, β는 feature 차원에 대해서만 존재하며 batch 크기와 무관하다는 것입니다.

위의 수식대로 같은 샘플을 가지고 레이어를 통과하는 연산을 수행해보겠습니다.

Layer Normalization Data Flow

입력 데이터 (X)

$$ X = \begin{bmatrix} [1,\ 2] \\ [2,\ 4] \\ [3,\ 6] \end{bmatrix} $$

shape: (3, 2)

→ 샘플 3개, 각 샘플은 2차원 벡터

Linear Layer 통과

가중치와 bias는 이전과 동일하게 설정합니다.

$$ Z = X $$

Layer Normalization 적용

Layer Normalization은 각 샘플마다 독립적으로 평균과 분산을 계산합니다.

첫 번째 샘플 [1, 2]

$$ \mu = (1 + 2) / 2 = 1.5 $$

$$ \sigma^2 = ((1 - 1.5)^2 + (2 - 1.5)^2) / 2 = 0.25 $$

정규화 결과:

$$ [1, 2] \rightarrow [-1, 1] $$

두 번째 샘플 [2, 4]

$$ \mu = 3,\quad \sigma^2 = 1 $$

정규화 결과:

$$ [2, 4] \rightarrow [-1, 1] $$

세 번째 샘플 [3, 6]

$$ \mu = 4.5,\quad \sigma^2 = 2.25 $$

정규화 결과:

$$ [3, 6] \rightarrow [-1, 1] $$

Layer Normalization 결과

X_hat =
[
 [-1,  1],
 [-1,  1],
 [-1,  1]
]

Transformer 구조에서 Layer Normalization 이 Batch Normalization 보다 적합한 이유

1. 시퀀스 길이 가변성과 Masking 문제

Transformer의 Self-Attention은 가변 길이 시퀀스를 처리해야 합니다. 입력형태는 각 문장마다 길이가 다르다는 점입니다. 이를 해결하기 위해 짧은 문장에는 padding을 추가하 attention mask를 사용해야 합니다.

Batch Normalization을 이러한 구조에 적용하면 심각한 문제가 발생합니다. BN은 배치와 시퀀스 차원 전체에 걸쳐 평균과 분산을 계산하는데 위에서 봤던 것 처럼 의미 없는 padding 토큰의 0 벡터가 통계에 포함됩니다. 결과적으로 문장 길이에 따라 정규화 통계가 왜곡되고, 같은 내용의 문장이라도 padding의 양에 따라 다르게 정규화될 수 있습니다.

반면 Layer Normalization은 각 토큰의 feature 차원에 대해서만 정규화를 수행합니다. 즉, 하나의 토큰 내부에서만 평균과 분산을 계산하기 때문에 padding 토큰이나 시퀀스 길이가 정규화 통계에 전혀 영향을 미치지 않습니다. 각 토큰은 독립적으로 정규화되므로 데이터의 의미가 충실히 반영되고 배치나 시퀀스 구조와 무관하게 일관된 정규화가 가능합니다.

2. Autoregressive Decoding과 배치 크기 불일치

Transformer Decoder는 추론 시 미래의 정보를 참조하지 못하도록 autoregressive 방식으로 동작합니다. 즉, 이전에 생성한 토큰을 바탕으로 다음 토큰을 하나씩 순차적으로 생성합니다. 이 과정에서 대부분의 경우 배치 크기가 1이 됩니다. 이는 Layer Normalization 논문에서 보여준것처럼 Batch Normalization에 치명적인 문제를 야기합니다.

Layer Normalization은 배치 크기와 무관하게 안정적으로 동작합니다. 배치 크기가 1이든 32든 정규화 결과는 일관되며, 학습 시 관찰한 성능이 추론 시에도 그대로 유지됩니다. 이는 Transformer Decoder의 생성 품질에 결정적으로 중요한 특성입니다.

3. Residual Connection과의 구조적 불일치

Transformer의 각 블록은 residual connection을 사용합니다: y = x + Sublayer(LN(x)). 이 구조가 중요한 이유는 gradient의 흐름 때문입니다. 역전파 시 ∂y/∂x = 1 + ∂Sublayer/∂x 가 되어, gradient가 항상 직접 흐를 수 있는 경로(identity mapping)가 보장됩니다. 이는 깊은 네트워크에서 gradient vanishing 문제를 해결하는 핵심 메커니즘입니다.

만약 Batch Normalization을 residual path에 사용하면, BN의 출력이 배치 통계에 의존하기 때문에 residual path에 batch-dependent noise가 주입됩니다. 이는 gradient flow를 불안정하게 만들고, 특히 깊은 Transformer에서는 gradient 폭발이나 소실을 일으킬 수 있습니다. 실제로 Post-LN Transformer(residual 후에 LN을 적용)는 레이어가 깊어질수록 학습이 불안정해지는 것으로 알려져 있으며, Pre-LN Transformer(residual 전에 LN을 적용)가 더 안정적인 학습을 보입니다. BN은 이러한 residual connection의 특성과 근본적으로 충돌합니다.

Layer Normalization은 각 샘플을 독립적으로 정규화하기 때문에 배치에 의존하지 않습니다. 따라서 residual path의 gradient flow를 방해하지 않으며, 수십 개의 레이어로 이루어진 깊은 Transformer에서도 안정적인 학습이 가능합니다. 이러한 구조적 조화가 Transformer가 Layer Normalization을 사용하는 또 다른 중요한 이유입니다.

배경

사내 LLM 서비스 개발 중 vLLM 이 병렬처리 되지 않는 현상이 발생했습니다. vLLM 로그를 보면 vLLM 서버에 요청이 하나씩 전송되어 처리되고 있는것을 알 수 있었는데, 처음엔 vLLM 내부에서 multi GPU 인식을 하지 못해 vram 을 과다하게 점유하여 병렬처리가 되지 않는 문제라고 생각했습니다.

하지만, vLLM 실행시 multi gpu 옵션을 줬고, 로그를 찍어보아도 2개의 gpu 가 잘 인식되어 있는것을 확인하고 문제를 찾다 FastAPI 에서 vLLM 에 요청을 보낼 때 openai 의 라이브러리를 사용했던것이 문제임을 알게 되었습니다. openai 라이브러리 중 OpenAI 를 사용하면 동기 Request 로 작동하고 AysncOpenAI 를 사용해야 비동기 작동을 하는 것을 알게 되었습니다.

해당내용을 정리할 겸 Request 를 사용한 방식가 httpx 를 사용한 요청방식의 차이점 그리고 FastAPI 의 동기/비동기, 병렬과 비동기의 작동방식을 정리하려고 합니다.

1. FastAPI 동기 / 비동기 처리 방식

FastAPI는 엔드포인트 함수가 def 인지 async def 인지에 따라 완전히 다른 방식으로 동작합니다.

1.1 동기 엔드포인트 (def)

from fastapi import FastAPI
import time

app = FastAPI()

@app.get("/sync")
def sync_endpoint():
    time.sleep(5)
    return {"msg": "done"}

동기 엔드포인트의 경우 FastAPI는 내부적으로 ThreadPoolExecutor를 사용해 요청을 처리합니다.

즉, 요청 하나당 스레드 하나를 점유하게 됩니다. 이 방식의 문제는 외부 API 호출과 같이 I/O 대기 시간이 긴 작업이 있을 경우입니다. 응답이 올 때까지 스레드가 점유되기 때문에, 동시에 처리할 수 있는 요청 수가 급격히 줄어들게 됩니다. 이 경우 vLLM 입장에서는 요청이 하나씩 순차적으로 들어오는 것처럼 보이게 됩니다.

1.2 비동기 엔드포인트 (async def)

from fastapi import FastAPI
import asyncio

app = FastAPI()

@app.get("/async")
async def async_endpoint():
    await asyncio.sleep(5)
    return {"msg": "done"}

비동기 엔드포인트는 이벤트 루프 기반으로 동작합니다. I/O 작업을 기다리는 동안 제어권을 이벤트 루프에 반환하고, 다른 요청을 처리할 수 있습니다. 다만 여기서 중요한 점은, async def로 선언했다고 해서 자동으로 비동기가 되는 것은 아닙니다. 엔드포인트 내부에서 사용하는 모든 I/O 작업이 비동기여야만 의미 있는 비동기 처리가 됩니다.

아래에서 추가적으로 설명하겠지만, 비동기 작업은 병렬과 다릅니다. 비동기작업은 동시성 작업으로 동시에 처리되는 것 처럼 보이게 됩니다.

2. FastAPI에서의 동시성과 병렬성

이번 이슈를 이해하기 위해서는 동시성과 병렬성의 차이를 명확히 구분할 필요가 있습니다.

2.1 동시성 (Concurrency)

동시성은 여러 작업을 번갈아가며 처리하는 개념입니다.

실제로 동시에 실행되는 것은 아니지만, 동시에 처리되는 것처럼 보이게 됩니다.FastAPI의 비동기 처리는 여기에 해당합니다.

2.2 병렬성 (Parallelism)

병렬성은 여러 작업을 실제로 동시에 실행하는 개념입니다.

FastAPI 공식문서에 귀여운 burger 예시가 있는데요

1.동시성

2. 병렬성 

자세한 내용은 위 링크에서 한번 확인해보시기 바랍니다.

3. OpenAI 라이브러리가 병목이 된 이유

3.1 OpenAI (동기 SDK) 사용 시

from openai import OpenAI

client = OpenAI(
    base_url="<http://vllm:8000/v1>",
    api_key="EMPTY"
)

@app.post("/chat")
def chat():
    response = client.chat.completions.create(
        model="qwen",
        messages=[{"role": "user", "content": "hello"}]
    )
    return response.choices[0].message.content

OpenAI 클래스는 동기 방식으로 동작합니다.

즉, 응답이 돌아올 때까지 FastAPI 스레드를 완전히 점유합니다.

이로 인해 발생한 현상은 다음과 같습니다.

• FastAPI 요청이 직렬화됨
• vLLM 서버 로그에 요청이 하나씩 찍힘
• GPU가 충분히 있음에도 batching이 발생하지 않음

처음에는 vLLM 설정 문제로 오해하기 쉬운 부분이었습니다.

3.2 AsyncOpenAI 사용 시 (해결)

from openai import AsyncOpenAI

client = AsyncOpenAI(
    base_url="<http://vllm:8000/v1>",
    api_key="EMPTY"
)

@app.post("/chat")
async def chat():
    response = await client.chat.completions.create(
        model="qwen",
        messages=[{"role": "user", "content": "hello"}]
    )
    return response.choices[0].message.content

AsyncOpenAI를 사용하면서 문제가 해결되었습니다.

• FastAPI 이벤트 루프가 block되지 않음
• 여러 요청이 동시에 vLLM으로 전달됨
• vLLM batching 정상 동작
• multi GPU 사용 확인

결과적으로 병렬처리가 되지 않는 것처럼 보였던 문제의 원인은

FastAPI와 vLLM 사이의 요청 방식이었습니다.

4. requests와 httpx 차이

4.1 requests

import requests

def call_vllm():
    r = requests.post(url, json=payload)
    return r.json()

• 동기 전용 라이브러리
• async def 내부에서 사용 시 이벤트 루프를 block
• FastAPI 비동기 구조와 맞지 않음

4.2 httpx (비동기 권장)

import httpx

async def call_vllm():
    async with httpx.AsyncClient(timeout=60) as client:
        r = await client.post(url, json=payload)
        return r.json()

• 비동기 I/O 지원
• connection pooling 제공
• FastAPI와 궁합이 매우 좋음

4.3 잘못된 예와 올바른 예

❌ 잘못된 예

@app.post("/bad")
async def bad():
    r = requests.post(url, json=payload)
    return r.json()

⭕ 올바른 예

@app.post("/good")
async def good():
    async with httpx.AsyncClient() as client:
        r = await client.post(url, json=payload)
        return r.json()

배경

대고객 챗봇 개발 당시 hallucination 에 관한 기준이 엄격해 모르는 답변은 모른다고 답변하고 상담원 연결로 돌리는 로직으로 설계되어 있었습니다.

때문에 고객이 chain 구조에서 조금만 예상에 어긋나는 행동을 하면 답변을 회피(모르겠다 답변 후 상담원 연결) 해 상담 만족도가 떨어지는 문제가 발생했는데요, 그래서 질문에 유연하게 대응하기 위해 체인구조에서 ReAct agent 로 마이그레이션 하기로 했습니다.

체인에 도달할 때에는 정해진 DTO 를 지켜야 했는데 체인이 있는 Tool 까지 도달할 때에는 이미 LLM 에 의해 DTO 가 뭉개져 Tool 에 인자를 전달하지 못하는 문제가 발생했습니다. 이때 프롬프트로만 출력을 제어했었는데, 답변을 잘하는 것 처럼 보였지만 Langsmith 로 agent tool calling을 추적한 결과 내부적으로는 일부 데이터들을 누락되고 calling 을 반복 하는 문제가 발생했습니다. 아마 강력한 프롬프트를 쓰면 좀 나아졌겠지만 결과적으로 이 문제는 응답시간 지연과, 토큰 비용 증가로 이어졌습니다. 하지만 고객반응과 실제 비지니스 문제해결에는 문제가 없었기 때문에 우선순위에 밀려 기술부채로 남게되었습니다.

지금 회사에 오게 되면서 structured output 에 관한 개념을 접하게 되고 신뢰가능한지, 실제 대고객 업무에서 사용할 만큼 신뢰도 있는지 확인해보려고 합니다.

Structured Output 의 작동원리

먼저 structured output 은 LLM 의 output 을 Json 이나 Pydantic 혹은 dataclass 같은 형태로 받을 수 있는 기능입니다. 또한 에러처리가 가능한데, 모델이 범위를 어긋나게 응답하거나 자료형을 틀리게 매칭한다면 validation error 를 만들 수 있어 에러메시지 유도가 가능합니다.

이것을 잘 활용하면 특정 경우에만 (format 이 맞지 않는 경우, 필드에 값이 잘못 들어가는 경우) Error를 발생시킬 수 있습니다. 일반적으로는 재시도를 하게되고 재시도 하는 경우 대부분 잘 매칭이 됩니다. 가장 치명적인 것은 structure 에 맞게 데이터를 넣기는 하지만, 그 값이 실제로 맞는지는 보장하지 않는 다는 것을 고려해야합니다.

작동 순서

1. 모델과 스키마를 입력받는다.
2. langchain 내부에서 전략을 선택함
   1. toolcalling strategy : 모델이 structured output 지원하지 않는 경우
   2. langchain 이 도구호출 JSON 형태로 반환하고 langchain 에서 파싱해서 스키마에 맞는 개체로 변환하는데 도구 호출 자체가 토큰을 더 쓰기때문에 비용증가/응답시간 증가가 발생합니다  https://platform.openai.com/docs/guides/structured-outputs
3. provider strategy : 모델이 structured output 지원하는 경우
4. langchain or agent 응답 생성
5. 결과물 유효성 검증 : 스키마에 맞게 파싱이 되었는지 Pydantic 이나 json 기반 파서 사
6. 파싱 성공하면 structured_response 에 넣어서 최종결과 반환

스키마입력 / 전략선택

스키마를 입력받는 부분부터 살펴보겠습니다. 아래의 예제가 있습니다.

Pydantic 스키마로 예시를 작성했는데 with_structured_output 메소드의 인자로 Pydantic 이 스키마로 넘어가게 됩니다.

class ReviewSummary(BaseModel):
    title: str = Field(..., description="리뷰 제목")
    sentiment: str = Field(..., description="긍정/부정/중립 중 하나")
    score: float = Field(..., description="0~1 사이의 감정 점수")
    
from langchain_openai import ChatOpenAI

# OpenAI API 또는 vLLM OpenAI 서버 URL로 자동 연결됨
model = ChatOpenAI(
    model="gpt-4o-mini",  # 아무 모델 가능
    temperature=0
)

structured_model = model.with_structured_output(ReviewSummary)

result = structured_model.invoke(user_input)

print(result)
print(type(result))
------------
title='영화 리뷰 요약'
sentiment='부정'
score=0.15
<class '__main__.ReviewSummary'>
------------

structured output 지원하는 일부 모델들은 아래처럼 벤더사가 지원하는 스키마에 맞게 변환하는 도구만을 bind 한 채로 끝나게 됩니다.

class ChatAnthropic(BaseChatModel):
#----------중략----------
	def with_structured_output():
	#----------중략----------
        if method == "function_calling":
            formatted_tool = **convert_to_anthropic_tool(schema)**
            tool_name = formatted_tool["name"]
            if self.thinking is not None and self.thinking.get("type") == "enabled":
                llm = self._get_llm_for_structured_output_when_thinking_is_enabled(
                    schema,
                    formatted_tool,
                )
            else:
                llm = self.bind_tools(
                    [schema],
                    tool_choice=tool_name,
                    ls_structured_output_format={
                        "kwargs": {"method": "function_calling"},
                        "schema": formatted_tool,
                    },
                )

@dataclass(init=False)
class ProviderStrategy(Generic[SchemaT]):
    """Use the model provider's native structured output method."""

    schema: type[SchemaT]
    """Schema for native mode."""

    schema_spec: _SchemaSpec[SchemaT]
    """Schema spec for native mode."""

    def __init__(
        self,
        schema: type[SchemaT],
    ) -> None:
        """Initialize ProviderStrategy with schema."""
        self.schema = schema
        self.schema_spec = _SchemaSpec(schema)

그리고 Provider 에 없는 모델은 ToolStrategy 를 사용하게 되는데 vllm 같은 로컬 서빙 프레임워크에서 작동시키는 모델들이 대체로 그러합니다.

class ChatOllama(BaseChatModel):
   #---중략----
   def with_structurd_output():
	   #---중략----
     if is_pydantic_schema:
            schema = cast("TypeBaseModel", schema)
            if issubclass(schema, BaseModelV1):
                response_format = schema.schema()
            else:
                response_format = schema.model_json_schema()
            llm = self.bind(
                format=response_format,
                ls_structured_output_format={
                    "kwargs": {"method": method},
                    "schema": schema,
                },
            )

@dataclass(init=False)
class ToolStrategy(Generic[SchemaT]):
    """Use a tool calling strategy for model responses."""

    schema: type[SchemaT]
    """Schema for the tool calls."""

    schema_specs: list[_SchemaSpec[SchemaT]]
    """Schema specs for the tool calls."""

    tool_message_content: str | None
    """The content of the tool message to be returned when the model calls
    an artificial structured output tool."""

    handle_errors: (
        bool | str | type[Exception] | tuple[type[Exception], ...] | Callable[[Exception], str]
    )
    

ToolStrategy 는 bind 메서드를 사용하여 러너블 객체에 접근하고 그 지점에 툴콜링을 하게 됩니다.

사람이 개입하여 벤더사의 툴을 호출할 수 있지만 전략선택의 결정적으로 큰 차이는 결국 with_structured_output 함수를 호출할 때 기본으로 선택되는 method 가 다르다는 것입니다.

모델이 structured output 지원하지 않는 경우

def with_structured_output(
        self,
        schema: dict | type,
        *,
        method: Literal["function_calling", "json_mode", "json_schema"] = "json_schema",
        include_raw: bool = False,
        **kwargs: Any,
    ) -> Runnable[LanguageModelInput, dict | BaseModel]:
        r"""Model wrapper that returns outputs formatted to match the

structured output 지원하는 경우

def with_structured_output(
        self,
        schema: dict | type,
        *,
        include_raw: bool = False,
        method: Literal["function_calling", "json_schema"] = "function_calling",
        **kwargs: Any,
    ) -> Runnable[LanguageModelInput, dict | BaseModel]:
        """Model wrapper that returns outputs formatted to match the given schema.

structured output 을 지원하는 경우에는 method 가 function_calling 으로 api 제공 벤더사의 function calling 형태로 처리하고

if method == "function_calling":
    formatted_tool = convert_to_anthropic_tool(schema)
    tool_name = formatted_tool["name"]
    if self.thinking is not None and self.thinking.get("type") == "enabled":
        llm = self._get_llm_for_structured_output_when_thinking_is_enabled(
            schema,
            formatted_tool,
        )
    else:
        llm = self.bind_tools(
            [schema],
            tool_choice=tool_name,
            ls_structured_output_format={
                "kwargs": {"method": "function_calling"},
                "schema": formatted_tool,
            },
        )

bind_tools 메서드를 사용하고 있습니다.

반대의 경우에는 json_schema 가 기본 선택되어 bind 메서드를 사용해서 tool calling 형태가 아니라 runnable sequence 에 새로운 객체를 만들어 호출 옵션을 재정의는 것입니다.

elif method == "json_schema":
            if schema is None:
                msg = (
                    "schema must be specified when method is not 'json_mode'. "
                    "Received None."
                )
                raise ValueError(msg)
            if is_pydantic_schema:
                schema = cast("TypeBaseModel", schema)
                if issubclass(schema, BaseModelV1):
                    response_format = schema.schema()
                else:
                    response_format = schema.model_json_schema()
                llm = self.bind(
                    format=response_format,
                    ls_structured_output_format={
                        "kwargs": {"method": method},
                        "schema": schema,
                    },
                )
                output_parser = PydanticOutputParser(pydantic_object=schema)  # type: ignore[arg-type]

##bind example
"""
        Example:
            ```python
            from langchain_ollama import ChatOllama
            from langchain_core.output_parsers import StrOutputParser

            model = ChatOllama(model="llama3.1")

            # Without bind
            chain = model | StrOutputParser()

            chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
            # Output is 'One two three four five.'

            # With bind
            chain = model.bind(stop=["three"]) | StrOutputParser()

            chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
            # Output is 'One two'
            
"""

자체적으로 response_format 을 세팅하고 있는 것을 볼 수 있습니다. 이렇게 스키마를 입력받고 전략을 선택하는 로직을 거치게 됩니다.

이제 전략별로 어떻게 structured output 을 만들어 내는지 살펴보겠습니다.

전략별 응답생성 과정

1. ToolcallingStrategy

   class ToolStrategy(Generic[SchemaT]):
       schema: type[SchemaT]
       schema_specs: list[_SchemaSpec[SchemaT]]
       tool_message_content: str | None
       handle_errors: bool | ...
   

   langchain 은 schema_spec 을 이용해서 fake tool schema 를 생성하고 이 fake tool 이름이 structured output 같은 형태로 모델에게 전달됩니다. 그럼 모델은 아래와 같은 형태로 응답합니다.이제 json 을 파싱해서 pydantic 이나 dataclass 검증을 하고 실패하면 Validation Error 을 뱉어내고 다시 모델에게 요청을 하게 됩니다.

   {
     "tool": "structured_output",
     "arguments": {
         "title": "some text",
         "score": 0.82
     }
   }
   ​

   이 error 이후 다시 모델에게 요청하는 과정에서 만약 모든 컨텍스트를 포함한 체인이나 노드라면 정말 많은 토큰이 낭비되고, 응답시간이 지연되게 됩니다.모델이 native 하게 structured output 을 지원하지 않는 경우 Toolcalling strategy 를 선택하게 됩니다.
2. ProviderStrategy

   @dataclass(init=False)
   class ProviderStrategy(Generic[SchemaT]):
       """Use the model provider's native structured output method."""
   
       schema: type[SchemaT]
       """Schema for native mode."""
   
       schema_spec: _SchemaSpec[SchemaT]
       """Schema spec for native mode."""
   

   langchain 은 스키마만 그대로 모델에게 전달하고 응답받아서 파싱만 수행합니다. openAI 와 anthropic gemini 의 응답은 안정적으로 다시 모델에게 요청하는 경우가 적습니다. 모델이 자체적으로 structured output 을 지원하는 경우입니다. 이때 langchain 은 각 벤더사에 맞는 형태로 변환/파싱을 수행합니다.

Structured Output 테스트

openai 의 structured otutput은 아래의 장점을 갖고 있는데, 특히 세번째 부분이 인상적이었습니다. 이전 챗봇 개발당시 레거시는 이 기능을 몰랐던것인지 프롬프트로 출력을 강제하고 있었는데, structured output 을 사용하면format 을 지키기 위해서 강력한 프롬프트를 하지 않아도 되기 때문입니다.

structured output 이 언제 지원되도록 포함되었는지 확인해보니 Toolcalling strategy 는 2023년 중후반쯤 그리고 ProviderStrategy는 2024년 8월 6일 gpt-4o 모델을 시작으로 openai 가 가장먼저 지원했습니다. 그 다음 anthropic 과 gemini 가 차례로 지원하기 시작했습니다.

langchain 스테이블버전이 2024년 1월에 배포되고, 그때부터 챗봇 레거시가 개발되기 시작했으니 최초 시스템 개발 이후 신기술 추적을 1년 6개월 가까이 하지 않았다는 것을 알 수 있었습니다.

그럼 실제로 프롬프트로 출력을 강제하는 것과 structured output 으로 output 형태를 파싱하는 것이 얼마나 다른지 확인해보겠습니다.

프롬프트 엔지니어링으로 output format 강제 테스트

Structured Output Pydantic 파라미터 전달 테스트

structured output 은 공식문서에서도 “structured output 은 실수할 수 있다” , “최대한 스키마에 대한 정보를 잘 작성해라” 라고 말하고 있습니다. 그래서 따라서 LLM 이 분류하거나, 어떤 포맷에 입력을 강제해야한다면 Pydantic 사용하기를 권장합니다.

간단한 프롬프트의 경우 둘다 잘 뱉어내는 것을 볼 수 있습니다.

그럼 실무에서 발생하는 시나리오를 생각해보고 테스트 해보겠습니다. LLM 이 섭취하게 될 데이터는 생각보다 복잡할 수 있습니다. 특히 여러개 DTO 가 섞여있는 경우 DTO 가 기하급수적으로 커지게 되는데요 3개의 DTO를 예시로 하여 json 타입이 아닌 자연어로 데이터를 주었을 때 잘 파싱하는지 확인해보겠습니다.

class Gender(str, Enum):
    male = "male"
    female = "female"
    other = "other"

class Address(BaseModel):
    street: str = Field(description="Street name and number")
    city: str = Field(description="City name")
    state: str = Field(description="State/Province")
    postal_code: str = Field(description="Postal/ZIP code")
    country: str = Field(description="Country name")

class UserProfile(BaseModel):
    name: str = Field(description="The user's full name")
    age: int = Field(description="The user's age")
    gender: Gender = Field(description="The user's gender")
    email: str = Field(description="The user's email address")
    phone_number: str = Field(description="The user's primary phone number")
    addresses: List[Address] = Field(description="List of user's addresses")
    date_of_birth: date = Field(description="The user's birth date")
    interests: List[str] = Field(default_factory=list, description="List of user's interests")
    is_active: bool = Field(default=True, description="Whether the user is active")
    bio: Optional[str] = Field(default=None, description="Short biography of the user")
    friends_ids: Optional[List[int]] = Field(default_factory=list, description="List of friend's user IDs")
    account_created: date = Field(description="Date when the user account was created")

Input은 아래와 같이 했다. 

박준호라는 사용자의 정보를 JSON으로 만들어주세요. 
나이는 24세, 남성이며, 이메일은 park.junho@example.com, 
전화번호는 010-1111-2222입니다.  
주소는 부산 해운대구 마린시티 5번지와 대구 수성구 범어로 88번지입니다.
생일은 2000년 12월 1일, 관심사는 게임, 코딩, 축구입니다. 
사용자는 비활성 상태(False)이며, 자기소개는 게임 개발자를 꿈꾸고 있는 대학생입니다.
친구 ID는 301, 302, 계정 생성일은 2021년 6월 20일입니다.

복잡한 구조적 데이터를 프롬프트로 형태를 강제한 것도 대체로 잘 파싱하는 것을 볼 수 있습니다. 하지만 결과물을 보면 postal code에 포함되어 있지 않은 데이터가 들어있습니다. 

그렇다면 structured output 을 사용한 쿼리는 어떨까요?

찬가지로 잘 파싱합니다. DTO가 복잡해지더라도 좋은 모델인 경우에는 거의 다 파싱을 해내는 것을 볼 수 있었습니다. 하지만 한가지 차이점이 발생했는데요 with structured output 메서드는 postal_code 가 빈칸인 것을 볼 수 있습니다. 하지만 prompt 로 강제한 경우에는 실제 데이터에 postal code 가 없음에도 불구하고 dummy 데이터가 들어가있는 것을 볼 수 있습니다.

Structured Outpu는 신뢰할 수 있을까

지금까지 내용으로 structured output 을 사용할 때 조금 더 잘 파싱이 되는 것을 볼 수 있었는데요, 훨신 간결하고 성능이 좋으니 따라서 프롬프트로 강제하는 것보다 structured output 기능을 사용하는것이 훨씬 더 유리할 것 같습니다.

2025년 12월 2일 기준으로

https://llm-stats.com/

실험에 사용한 gpt-4o 모델보다 괜찮은 로컬 모델들이 많은데요 30B 정도 되는 모델들을 사용한다면 로컬에서 돌리는 모델들도 잘 작동할 것이라 예상합니다. 그래서 성능측면에서 어떤 전략이 더 우월하다는 것은 큰 의미가 없어보입니다.

하지만 그럼에도 불구하고 Toolcalling Strategy 의 경우는 retry 가 자주 발생할 수 있기 때문에 structured output 지원되는 api 를 사용할 수 있는 환경이라면 ProviderStrategy 를 사용할 수 있는 방법으로 시도해야 한다고 생각합니다.

이제 출력 구조를 프롬프트로 강제하는 것보다 structured output 을 사용하는 것이 좋다는 것은 알게 되었습니다. 그럼 결정적으로 structured output 을 신뢰할 수 있을까? 에 대한 답을 구해야하는데, 최근 아래의 글을 읽게 되었습니다.

https://www.philschmid.de/why-engineers-struggle-building-agents

시니어 개발자들이 주니어 개발자들보다 AI Agent 를 개발하는게 느리다는 주제로 시작한 글인데 그 이유를 생각하면 사뭇 철학적으로 받아들여야 할 부분이 있습니다.

이유는 전통적인 소프트웨어 엔지니어링(엄격한 제어, 결정론적) 그러니까 맞으면 맞는거고 틀리면 틀린거지, 틀리면 고쳐야지 라는 전통적인 엔지니어링의 철학과 습관이 AI 에이전트 개발에 방해가 되고 있다는 겁니다. 글의 저자인 Phillipp Schmid 는 시니어일수록 LLM 의 불확실성을 코드로 제거하려고 하는 경향이 있어 주니어보다 느려진다는 것입니다.

텍스트 데이터의 맥락을 구조화 된 것으로 강제하면 LLM이 잘하는것을 오히려 더 못하게 하면서 성능이 떨어지고 성능이 떨어지는 이유를 코드로 제거하려 하니 수렁에 빠지게 된다의 의미인 것 같습니다.

그래서 저자는 agent를 개발할 때 아래의 정신을 갖추어야 한다고 제안합니다.

1. 텍스트가 새로운 상태(State)
   • 함정: 자연어 입력을 구조화된 데이터(예: true/false)로 강제하면 맥락 상실.
   • 해결: 피드백(예: “승인, 미국 시장 집중”)을 텍스트로 보존해 동적 조정 가능.
2. 제어권을 넘겨라
   • 함정: 흐름을 하드코딩(예: 구독 취소 루트)하면 비직선적 상호작용 대응 실패.
   • 해결: 에이전트(LLM)가 맥락 기반으로 의도 판단하도록 신뢰.
3. 에러는 그냥 입력이다
   • 함정: 에러 발생 시 프로그램 중단(전통 방식)으로 고비용 실행 낭비.
   • 해결: 에러를 피드백으로 제공해 에이전트가 자가 복구 시도.
4. 유닛 테스트에서 Eval로
   • 함정: 이진 테스트(TDD) 적용 시 확률적 시스템에서 무의미(무한 유효 답변).
   • 해결: 신뢰성(Pass@k), 품질(LLM Judge), 추적(Eval)로 변동성 관리.
5. 에이전트는 진화하고, API는 그렇지 않다
   • 함정: 인간 중심 API(암묵적 맥락) 사용 시 에이전트 환각 발생.
   • 해결: 상세 시맨틱 타이핑(예: “user_email_address”)과 독스트링으로 명확화. 에이전트는 도구 변화에 적응 가능.

결론은 엔지니어링 산물의 확률성을 받아들이고 edge case 들을 강제하는 것이 아닌 그 마저도 LLM 이 자기피드백을 할 수 있는 탄력적 시스템 구축으로 만들고 그 과정을 관리하라는 말 입니다.

그래서 다시 요점으로 돌아와 structured output 을 신뢰할 수 있는가? 에 대한 답은 신뢰할 수 있다에 가깝다. 인 것 같습니다. 그리고 신뢰할 수 있다 없다 로 판단하는것이 아니라 얼마나 신뢰할 수 있는가? 에 집중해야 하겠습니다.

기능은 대체로 잘 작동하니(gpt-4o 이상의 모델), 각자의 프로덕션 환경에서 테스트해보고 관리 가능한 edge case 인지 파악하고 사용하는 것이 좋을 것 같습니다.

이 부분에 대한 생각은 사람마다 많이 다를 것 같습니다. 다른 의견들을 댓글로 남겨주세요!

1. 소개

Python 3.12는 여러 새로운 기능과 개선 사항을 도입하여 개발자들의 생산성과 코드 품질을 향상시킵니다. 특히, 향상된 오류 메시지, 더 강력한 f-strings, 더 빠른 Python 실행 속도, 전용 타입 변수 구문, Linux perf 프로파일러 지원 등 다양한 기능이 추가되었습니다.
Real Python

2. 설치 단계
3. /tmp 디렉토리로 이동

먼저, /tmp 디렉토리로 이동합니다.

cd /tmp/

2. Python 3.12.0 버전 다운로드 및 압축 해제

Python 3.12.0 버전을 다운로드하고 압축을 해제합니다.

wget https://www.python.org/ftp/python/3.12.0/Python-3.12.0.tgz
tar -xzvf Python-3.12.0.tgz
cd Python-3.12.0/

3. 필요한 빌드 의존성 패키지 설치

Python을 빌드하기 위해 필요한 의존성 패키지를 설치합니다.

sudo apt update
sudo apt install build-essential zlib1g-dev libncurses5-dev libgdbm-dev libnss3-dev libssl-dev libreadline-dev libffi-dev pkg-config

4. Python 구성 및 빌드

Python을 구성하고 빌드합니다.

./configure --enable-optimizations
make -j $(nproc)

5. Python 설치

Python을 시스템에 설치합니다.

sudo make altinstall

6. 심볼릭 링크 추가

설치된 Python 실행 파일에 심볼릭 링크를 추가합니다.

ln -s /usr/local/bin/python3.12 /usr/local/bin/python
ls -al /usr/local/bin/python

7. 설치 확인

설치된 Python 버전을 확인합니다.

python -V

3. 마무리

이제, 원하는 Python 버전이 설치되었고 실행 파일에 적절한 심볼릭 링크가 추가되었습니다. 이로써 Python 3.12 환경이 준비되었습니다.

추가 정보

Python 3.12는 다양한 새로운 기능과 개선 사항을 도입하여 개발자들의 생산성과 코드 품질을 향상시킵니다. 특히, 향상된 오류 메시지, 더 강력한 f-strings, 더 빠른 Python 실행 속도, 전용 타입 변수 구문, Linux perf 프로파일러 지원 등 다양한 기능이 추가되었습니다.
Real Python

LangChain을 활용한 LLM(Large Language Model) 애플리케이션 개발에서 기본 LLM 체인은 가장 핵심적인 개념입니다. 이 체인은 사용자의 입력(프롬프트)을 받아 LLM을 통해 원하는 응답을 생성하는 간단하면서도 강력한 구조를 말해요. 대화형 AI부터 자동 문서 요약까지 다양한 LLM 기반 애플리케이션의 기반이 됩니다.

1. 기본 LLM 체인의 핵심 구성 요소

기본 LLM 체인은 크게 두 가지 요소로 이루어져 있습니다.

• 프롬프트(Prompt): LLM에게 어떤 작업을 수행할지 알려주는 지시문이에요. 질문, 명령, 특정 맥락을 제공하는 문장 등 다양한 형태를 가질 수 있으며, LLM의 응답 품질을 결정하는 가장 중요한 요소입니다. 효과적인 프롬프트는 LLM이 의도한 방향으로 정확히 응답하도록 유도합니다.
• LLM(Large Language Model): GPT-3.5, GPT-4, Gemini 등 대규모 언어 모델을 의미합니다. 방대한 양의 텍스트 데이터로 학습되어 언어를 이해하고 새로운 텍스트를 생성하는 능력을 갖추고 있습니다. 프롬프트를 분석하고, 학습된 지식을 바탕으로 요청된 작업을 수행하거나 적절한 정보를 제공하는 역할을 합니다.

2. 작동 방식

기본 LLM 체인의 작동 방식은 다음과 같습니다.

1. 프롬프트 생성: 사용자의 요구사항이나 수행할 작업을 정의하는 프롬프트를 만듭니다. 이 프롬프트는 LLM이 더 정확하게 응답할 수 있도록 명확한 지침과 맥락을 포함하도록 최적화할 수 있어요.
2. LLM 처리: 생성된 프롬프트는 LLM에게 전달됩니다. LLM은 프롬프트를 분석하고, 내부적으로 학습된 지식과 패턴을 활용하여 응답을 생성합니다.
3. 응답 반환: LLM이 생성한 응답은 사용자에게 전달됩니다. 이 응답은 단순한 답변, 요약된 정보, 혹은 생성된 텍스트 등 다양한 형태를 띨 수 있습니다.

3. 실습 예제: LangChain으로 LLM 체인 만들기

이제 실제 코드를 통해 LangChain에서 기본 LLM 체인을 구성하는 방법을 살펴보겠습니다.

예제 1: 단순 LLM 호출

가장 기본적인 방법으로, ChatOpenAI 함수를 사용하여 OpenAI의 LLM 모델에 직접 프롬프트를 전달하는 예제입니다.

from langchain_openai import ChatOpenAI

# LLM 모델 인스턴스 생성
llm = ChatOpenAI(model="gpt-4o-mini")

# 모델에 직접 프롬프트 전달 및 실행
llm.invoke("지구의 자전 주기는?")

위 코드를 실행하면, llm 객체가 "지구의 자전 주기는?" 라는 질문을 받아 답변을 생성하고, AIMessage 객체 형태로 반환합니다.

AIMessage(content='지구의 자전 주기는 약 23시간 56분 4초입니다. 이것을 항성일(sidereal day)이라고 합니다. 우리가 흔히 말하는 하루 24시간은 태양일(solar day)로, 지구가 자전하면서 공전하기 때문에 태양을 기준으로 하루가 24시간이 됩니다.')

예제 2: 프롬프트 템플릿 사용

이번에는 더 체계적인 접근을 위해 프롬프트 템플릿을 사용해 보겠습니다. 프롬프트 템플릿은 프롬프트의 형식을 미리 정의해두고, 필요한 부분만 변수로 채워 넣어 사용하는 방식입니다.

ChatPromptTemplate.from_template() 메서드를 이용해 프롬프트 템플릿을 생성할 수 있습니다. 아래 예제는 LLM에게 "천문학 전문가" 역할을 부여하여 질문에 답변하도록 지시하는 템플릿입니다.

from langchain_core.prompts import ChatPromptTemplate

# 프롬프트 템플릿 정의
prompt = ChatPromptTemplate.from_template(
    "You are an expert in astronomy. Answer the question. <Question>: {input}"
)

# 템플릿 객체 확인
prompt

결과를 보면 input_variables=['input']을 통해 input이라는 변수를 받는 프롬프트 객체가 생성된 것을 확인할 수 있습니다.

ChatPromptTemplate(input_variables=['input'], messages=[HumanMessagePromptTemplate(prompt=PromptTemplate(input_variables=['input'], template='You are an expert in astronomy. Answer the question. <Question>: {input}'))])

예제 3: LCEL을 활용한 체인 구성

LangChain Expression Language (LCEL)은 파이프(|) 연산자를 이용해 프롬프트, 모델, 출력 파서를 간편하게 연결하여 하나의 체인으로 만드는 강력한 기능입니다.

다음 코드는 앞서 정의한 prompt와 llm을 연결하고, 최종적으로 StrOutputParser를 통해 LLM의 응답을 깔끔한 문자열 형태로 변환하는 과정을 보여줍니다.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 프롬프트, 모델, 출력 파서 정의
prompt = ChatPromptTemplate.from_template("You are an expert in astronomy. Answer the question. <Question>: {input}")
llm = ChatOpenAI(model="gpt-4o-mini")
output_parser = StrOutputParser()

# LCEL로 체인 연결
chain = prompt | llm | output_parser

# 체인 호출
chain.invoke({"input": "지구의 자전 주기는?"})

이 코드를 실행하면, 프롬프트 템플릿이 먼저 질문을 완성하고, LLM이 답변을 생성하며, 마지막으로 StrOutputParser가 그 답변을 순수한 텍스트로 변환하여 반환합니다.

지구의 자전 주기는 약 24시간입니다. 이것은 하루의 길이를 결정하는데 중요한 역할을 합니다.

분류 : 딕셔너리

표면적으로 딕셔너리를 사용해서 푸는 문제이지만, 두가지 해결해야 하는 이슈들이 더 있다.

1. Value 로 Key 값을 찾기.

2. input() 에서 발생하는 시간초과 문제 해결하기

keypoint : python input / value로 key 찾기 

code

import sys 
n,m = map(int,input().split(' '))

pocketmon_list = dict()
rev_poecketmon_list = dict()
cnt=1
for i in range(0,n):
    name = sys.stdin.readline().strip()
    pocketmon_list[str(cnt)] = name
    rev_poecketmon_list[name] = str(cnt)
    cnt+=1



for i in range(0,m):
    tmp_input = sys.stdin.readline().strip()
    if tmp_input.isdigit():
            print(pocketmon_list[tmp_input])
    else:
         print(rev_poecketmon_list[tmp_input])

중요한 내용 

1. Value 로 Key 찾기 

위 글을 참고하기 바란다. 결론은 value로 key 를 직접 찾는 것은 for 문을 사용한 완전탐색밖에 없다. 

2. 왜 input() 이 sys.stdin.readline().stirp() 보다 느릴까?

input() 함수는 Python 에서 기본적으로 제공하는 사용자 입력 함수인데 다음과 같은 특징을 갖고 있다.

1. 입력 된 값을 '문자열로 반환' 하고 '자동으로 개행 문자 제거' 를 한다.

2. 프롬프트 메시지를 인자로 받을 수 있다.

여기서 이 문자열로 변환하여 반환하고 자동으로 개행 문자를 제거하는게 물리적으로 시간이 대단히 오래걸린다.

그에 반해 readline() 함수는 개행 문자를 포함하여 문자열을 반환한다. 그렇기 때문에 그 시간 차이가 발생하는데 readline 함수에서는 strip() 을 사용하여 개행문자를 지울 수 있다. 

두 함수의 시간차이를 보여주는 함수이다.

import sys
import time

# sys.stdin.readline() 사용
start = time.time()
for _ in range(100000):
    line = sys.stdin.readline().strip()
end = time.time()
print(f'sys.stdin.readline() 사용 시간: {end - start}초')

# input() 사용
start = time.time()
for _ in range(100000):
    line = input()
end = time.time()
print(f'input() 사용 시간: {end - start}초')

100000줄의 입력을 처리하는 데 걸리는 시간:
input() 함수: 12.3456초
sys.stdin.readline() 함수: 0.4567초

결과값은 어마어마하게 차이가 난다. 따라서 python 에서 시간초과문제를 겪을 때 input 을 sys.stdin.readline().strip() 으로 변경해보자.

import sys
input = sys.stdin.readline().strip

이렇게 하면 코드변경 없이도 기존 input 함수에 적용하여 사용 할 수 있다.