OpenAI Batch API로 LangChain 비용 80% 줄이기

서버에서 LangChain으로 문서 요약, 분류, 임베딩, 평가를 돌리다 보면 비용이 생각보다 빠르게 불어납니다. 특히 동일한 모델을 수천~수십만 번 호출하는 배치성 작업(예: 백필, 재색인, 로그 분석, 고객 티켓 라벨링)은 실시간 응답이 필요 없는데도 일반 API 호출로 처리하는 경우가 많습니다.

이 글에서는 OpenAI Batch API를 LangChain 워크플로에 접목해, 배치성 LLM 호출 비용을 크게(사례 기준 최대 80% 수준까지) 줄이는 방법을 다룹니다. 핵심은 다음 한 줄입니다.

• 실시간 chat.completions 호출을 비동기 대량 작업 큐로 바꾸고
• 요청을 JSONL로 모아 Batch로 제출한 뒤
• 완료 후 결과를 원본 입력과 조인해 파이프라인을 이어 붙입니다

주의: Batch는 “즉시 응답”이 아니라 “나중에 결과를 받는” 모델입니다. 사용자-facing 요청에는 부적합하지만, 백오피스/데이터 파이프라인에는 최적입니다.

왜 LangChain 비용이 커지는가

LangChain 자체가 비용을 발생시키는 건 아니고, 다음 패턴이 비용을 키웁니다.

1. 짧은 프롬프트를 매우 많이 호출

• 예: 1만 건 분류, 5만 건 요약, 20만 건 평가
• 호출 오버헤드와 단가가 누적됩니다

2. 동시성 제어가 약해 리트라이가 폭증

• 레이트 리밋, 네트워크 타임아웃으로 재시도
• 재시도는 곧 토큰 비용 중복

3. 실시간 API로 배치 작업을 억지로 처리

• 처리량을 올리려다 실패율이 오르고, 실패율을 낮추려다 워커 수를 줄이면 시간이 늘어납니다

Batch API는 이 상황에서 “대량 요청을 한 번에 맡기고”, 완료되면 결과를 받아 처리하는 방식이라 운영이 단순해지고 단가 측면 이점이 생깁니다.

Batch API가 맞는 작업/아닌 작업

Batch에 잘 맞는 작업

• 야간 배치 요약/분류/태깅
• 벡터DB 재색인 전처리(문서 정규화, 메타데이터 추출)
• 대량 평가(LLM-as-a-judge)
• 데이터 백필(backfill)

Batch에 안 맞는 작업

• 채팅/에이전트처럼 사용자 대기 시간이 중요한 경우
• 툴 호출을 반복하며 상태를 바꾸는 온라인 플로우
• 강한 순서 의존성이 있는 멀티턴 대화

에이전트가 무한 루프에 빠지는 문제를 겪고 있다면 비용 절감 이전에 Stop 조건 설계가 먼저입니다. 관련해서는 AutoGPT 툴 호출 무한루프 끊는 Stop조건 설계를 함께 참고하면 좋습니다.

아키텍처: LangChain 파이프라인에 Batch 끼워 넣기

전체 흐름

1. LangChain으로 입력을 만들고(문서 chunk, 질문 목록 등)
2. 각 입력을 OpenAI 요청 JSON으로 변환해 JSONL 파일 생성
3. Batch 제출
4. 완료 폴링
5. 결과 JSONL 다운로드
6. custom_id 기준으로 원본 입력과 조인
7. 후속 처리(저장, 인덱싱, 리포팅)

핵심은 custom_id를 설계하는 것입니다.

• 예: doc:{doc_id}:chunk:{chunk_id}
• 혹은 DB PK를 그대로 넣어도 됩니다

코드: JSONL 만들기 (Chat Completions 배치)

아래 예시는 “문장 분류”를 대량으로 돌리는 케이스입니다. Batch 입력은 보통 JSONL 한 줄당 하나의 요청입니다.

MDX 빌드 에러 방지를 위해, 본문에서 부등호가 들어갈 수 있는 표기는 모두 코드로 감쌉니다.

import json
from pathlib import Path

def build_requests_jsonl(items, out_path: str):
    """items: [{"id": "...", "text": "..."}, ...]"""
    p = Path(out_path)
    p.parent.mkdir(parents=True, exist_ok=True)

    with p.open("w", encoding="utf-8") as f:
        for it in items:
            req = {
                "custom_id": it["id"],
                "method": "POST",
                "url": "/v1/chat/completions",
                "body": {
                    "model": "gpt-4o-mini",
                    "temperature": 0,
                    "messages": [
                        {
                            "role": "system",
                            "content": "You are a strict classifier. Output JSON only.",
                        },
                        {
                            "role": "user",
                            "content": (
                                "Classify the following text into one of: "
                                "billing, bug, feature, other. "
                                "Return {\"label\": \"...\"}.\n\n"
                                f"TEXT: {it['text']}"
                            ),
                        },
                    ],
                    "response_format": {"type": "json_object"},
                },
            }
            f.write(json.dumps(req, ensure_ascii=False) + "\n")


items = [
    {"id": "ticket:1001", "text": "I was charged twice this month."},
    {"id": "ticket:1002", "text": "App crashes when I click export."},
]

build_requests_jsonl(items, "./batch/input.jsonl")

이 단계는 LangChain을 써도 되고, 안 써도 됩니다. LangChain을 쓰는 경우 보통 다음에서 생성됩니다.

• TextSplitter로 chunk 생성
• 메타데이터에 doc_id, chunk_id를 넣고
• 그 조합으로 custom_id를 만듭니다

코드: Batch 제출/폴링/다운로드

OpenAI Python SDK 기준 예시입니다(버전에 따라 메서드명이 다를 수 있어, 개념적으로 이해하고 적용하세요).

import time
from openai import OpenAI

client = OpenAI()

# 1) JSONL 파일 업로드
input_file = client.files.create(
    file=open("./batch/input.jsonl", "rb"),
    purpose="batch",
)

# 2) Batch 생성
batch = client.batches.create(
    input_file_id=input_file.id,
    endpoint="/v1/chat/completions",
    completion_window="24h",
)

print("batch_id=", batch.id)

# 3) 완료까지 폴링
while True:
    b = client.batches.retrieve(batch.id)
    status = b.status
    print("status=", status)

    if status in ("completed", "failed", "cancelled", "expired"):
        break

    time.sleep(10)

if b.status != "completed":
    raise RuntimeError(f"batch not completed: {b.status}")

# 4) 결과 파일 다운로드
output_file_id = b.output_file_id
content = client.files.content(output_file_id).read()

with open("./batch/output.jsonl", "wb") as f:
    f.write(content)

운영에서는 폴링을 워커에서 돌리기보다, Batch ID를 DB에 저장해두고 스케줄러가 주기적으로 상태를 확인하는 방식이 깔끔합니다.

코드: 결과 조인(custom_id로 원본과 매칭)

Batch 결과는 각 줄에 custom_id와 응답/에러가 포함됩니다. 이를 원본 입력과 조인해 파이프라인을 이어갑니다.

import json

def load_jsonl(path: str):
    with open(path, "r", encoding="utf-8") as f:
        for line in f:
            yield json.loads(line)

# 원본 입력을 dict로 만들어둔다
inputs = {
    "ticket:1001": {"text": "I was charged twice this month."},
    "ticket:1002": {"text": "App crashes when I click export."},
}

results = []
for row in load_jsonl("./batch/output.jsonl"):
    cid = row.get("custom_id")

    # 성공 응답
    if "response" in row and row["response"].get("status_code") == 200:
        body = row["response"]["body"]
        msg = body["choices"][0]["message"]["content"]
        results.append({
            "id": cid,
            "input": inputs.get(cid),
            "raw": msg,
        })
    else:
        results.append({
            "id": cid,
            "input": inputs.get(cid),
            "error": row.get("error") or row.get("response"),
        })

print(results)

이제 results를 DB에 적재하거나, 벡터DB 메타데이터로 저장하거나, LangChain의 후속 체인으로 넘기면 됩니다.

LangChain과 연결하는 실전 패턴 3가지

1) MapReduce 요약을 “배치 요약”으로 바꾸기

대량 문서 요약을 LangChain map_reduce로 실시간 호출하면 비용이 급증합니다. 대신 다음 구조가 효율적입니다.

• map 단계(각 chunk 요약)를 Batch로 돌림
• 결과 요약들을 모아 reduce 단계만 실시간(또는 또 한 번 Batch)으로 처리

이때 custom_id를 doc:{doc_id}:chunk:{chunk_id}로 만들면 reduce 조인이 쉽습니다.

2) RAG 전처리를 Batch로 밀어 넣기

RAG에서 비용이 숨어 있는 구간은 “검색”이 아니라 “전처리”인 경우가 많습니다.

• 문서 정규화(제목 추출, 섹션 라벨링)
• PII 마스킹
• QnA 생성

이런 작업은 대부분 오프라인으로 돌려도 되므로 Batch에 적합합니다.

3) 평가(Eval) 파이프라인을 Batch로

LLM 평가를 LangChain으로 붙여두면 실험 1회에 수천 호출이 나갑니다. 평가 프롬프트는 짧고 반복적이라 Batch로 단가 최적화가 잘 됩니다.

비용을 80%까지 줄이려면: 숫자보다 중요한 운영 포인트

“80%”는 누구나 동일하게 나오는 숫자는 아닙니다. 하지만 아래를 지키면 체감 비용이 크게 내려갑니다.

1. 배치로 옮길 수 있는 호출을 최대한 옮기기

• 실시간이 아닌 작업을 구분해 큐로 분리

2. 중복 호출 제거(캐시/디듀프)

• 동일 입력 해시로 결과 재사용
• 실패 재시도 시에도 동일 custom_id로 재처리 여부 추적

3. 프롬프트 압축과 구조화 출력 강제

• 불필요한 지시문 제거
• response_format을 json_object로 강제해 파싱 비용과 재시도 감소

4. 재시도는 “요청 단위”로, 무한 재시도 금지

• Batch 결과에서 실패만 골라 새 JSONL로 재제출
• 재시도 횟수 상한 설정

5. 동시성 폭주로 생기는 이벤트 루프 이슈 방지

• 배치 제출/폴링/다운로드를 asyncio로 섞어 구현하다 보면 로컬/CI에서 Event loop is closed류의 문제가 자주 납니다. 필요하면 Python asyncio RuntimeError - Event loop is closed 해결을 참고해 런타임을 안정화하세요.

운영 체크리스트

관측(Observability)

• Batch ID, input 파일 ID, output 파일 ID를 모두 저장
• custom_id 기준 성공/실패 카운트
• 실패 사유(레이트 리밋, 정책, 입력 포맷 에러)별 분류

데이터 품질

• JSON 출력 강제 후 스키마 검증
• 잘못된 JSON은 “재시도”가 아니라 “프롬프트/제약 수정”이 필요할 때가 많음

배포/파이프라인

• GitHub Actions로 야간 배치를 돌리면 권한 이슈로 막히는 경우가 있습니다. 예를 들어 레포 권한/토큰 스코프 문제는 GitHub Actions GITHUB_TOKEN 403 권한 오류 해결을 참고해 빠르게 점검할 수 있습니다.

마이그레이션 가이드: 기존 LangChain 코드에서 무엇을 바꾸나

1. 기존

• chain.invoke() 또는 chain.batch()로 즉시 호출

2. 변경

• invoke() 대신 “요청 생성기”로 바꿈
• 생성된 요청을 JSONL로 저장
• Batch 제출 후 결과 파일을 받아 파싱
• 파싱된 결과를 다음 단계로 전달

즉, LangChain은 여전히 유용합니다.

• 입력 문서 처리(로더/스플리터)
• 프롬프트 템플릿 관리
• 후처리 체인

다만 LLM 호출 자체는 Batch로 우회하는 것이 포인트입니다.

결론

OpenAI Batch API는 “대량 호출을 싸고 안정적으로” 처리하는 데 초점이 맞춰져 있습니다. LangChain 기반 시스템에서 비용이 커지는 구간은 대개 배치성 작업이므로, 해당 호출을 Batch로 옮기면 비용과 운영 복잡도를 동시에 낮출 수 있습니다.

정리하면 다음 순서로 적용하세요.

• 배치로 전환 가능한 호출을 분리한다
• custom_id 설계를 먼저 고정한다
• JSONL 생성, 제출, 폴링, 결과 조인까지 파이프라인화한다
• 실패만 골라 재제출하는 재시도 전략을 둔다

이 구조를 한 번 만들어두면, 요약/분류/평가/전처리 등 대부분의 LLM 작업을 같은 프레임으로 흡수할 수 있고, LangChain 프로젝트의 월 비용을 체감 가능한 수준으로 떨어뜨릴 수 있습니다.