오늘은 22일차입니다.
LangChain을 활용하다 보면, 모델이나 체인을 저장해두었다가 다시 불러와야 할 때가 있습니다. 이를 위해 LangChain은 객체를 직렬화(Serialization)하여 JSON, 딕셔너리 형태로 저장하고, 필요할 때 역직렬화하여 재사용할 수 있는 기능을 제공합니다.
또한, LLM을 호출할 때는 항상 토큰 사용량을 확인하는 것이 중요합니다. 토큰 사용량은 곧 API 비용과 직결되며, 프롬프트 최적화에도 핵심 지표가 됩니다.
이번 글에서는 두 가지 주제를 다룹니다.
- LangChain 객체를 직렬화/역직렬화하는 방법
- LLM 호출 시 토큰 사용량을 추적하는 방법
1. 직렬화(Serialization)
직렬화란?
직렬화는 Python 객체를 JSON, YAML 같은 형식으로 변환하여 저장하거나 전송할 수 있도록 만드는 과정입니다.
LangChain의 프롬프트, 체인, 모델 객체는 모두 직렬화가 가능하며, 이를 통해 구성한 워크플로우를 파일 형태로 관리할 수 있습니다.
1) 직렬화 가능 여부 확인
LangChain의 객체가 모두 직렬화 가능한 것은 아닙니다. 따라서 우선 is_lc_serializable() 메서드로 직렬화 지원 여부를 확인해야 합니다.
import os
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
# 프롬프트 템플릿 생성
prompt = PromptTemplate.from_template("{fruit}의 색상이 무엇입니까?")
# ChatOpenAI 클래스 자체가 직렬화 가능한지 확인
print(f"ChatOpenAI: {ChatOpenAI.is_lc_serializable()}")
# 출력: True
True가 나오면 해당 클래스는 직렬화를 지원합니다.
2) LLM 객체 직렬화 가능 여부
클래스 자체뿐 아니라, 인스턴스를 생성한 뒤에도 직렬화가 가능한지 확인할 수 있습니다.
# gpt-3.5-turbo 모델을 사용하는 LLM 객체 생성
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
# 인스턴스 단위에서도 직렬화 가능 여부 확인
print(f"ChatOpenAI: {llm.is_lc_serializable()}")
# 출력: True
3) 체인 직렬화 가능 여부
프롬프트와 LLM을 연결해 체인을 만든 뒤에도 직렬화가 지원되는지 확인할 수 있습니다.
# 체인을 생성 (PromptTemplate | LLM)
chain = prompt | llm
# 직렬화 가능 여부 확인
print(chain.is_lc_serializable())
# 출력: True
👉 여기까지는 체인을 직렬화할 수 있는 준비가 되었다는 것을 확인하는 과정입니다.
4) 직렬화 실행
LangChain에서는 dumpd와 dumps 두 가지 직렬화 함수를 제공합니다.
- dumpd: 객체 → Python dict
- dumps: 객체 → JSON 문자열
from langchain_core.load import dumpd, dumps
# 체인을 딕셔너리로 직렬화
dumpd_chain = dumpd(chain)
print(type(dumpd_chain)) # dict
print(dumpd_chain)
출력 예시(일부 축약):
{
'lc': 1,
'type': 'constructor',
'id': ['langchain', 'schema', 'runnable', 'RunnableSequence'],
'kwargs': {
'first': {...}, # PromptTemplate 정보
'last': {...} # ChatOpenAI 정보
}
}
이번에는 JSON 문자열로 직렬화해봅니다.
# JSON 문자열로 직렬화
dumps_chain = dumps(chain)
print(type(dumps_chain)) # str
print(dumps_chain[:100]) # 앞부분만 출력
5) Pickle / JSON 파일 저장
직렬화한 결과는 파일로 저장해둘 수도 있습니다.
import pickle, json
# Pickle 파일로 저장
with open("fruit_chain.pkl", "wb") as f:
pickle.dump(dumpd_chain, f)
# JSON 파일로 저장
with open("fruit_chain.json", "w") as fp:
json.dump(dumpd_chain, fp)
⚠️ 주의: Pickle은 임의 코드 실행 위험이 있으므로, 신뢰할 수 없는 환경에서는 사용하지 않는 것이 권장됩니다. 안전하게 공유하려면 JSON 직렬화를 사용하는 것이 좋습니다.
6) 역직렬화(Deserialization)
저장된 파일은 load / loads를 사용해 다시 LangChain 객체로 복원할 수 있습니다.
import pickle
from langchain_core.load import load, loads
# Pickle 파일 로드
with open("fruit_chain.pkl", "rb") as f:
loaded_chain = pickle.load(f)
# 체인 복원 (역직렬화)
chain_from_file = load(
loaded_chain, secrets_map={"OPENAI_API_KEY": os.environ["OPENAI_API_KEY"]}
)
print(chain_from_file.invoke({"fruit": "사과"}))
또는 JSON 파일에서 복원할 수도 있습니다.
import json
with open("fruit_chain.json", "r") as fp:
loaded_from_json_chain = json.load(fp)
loads_chain = load(loaded_from_json_chain)
print(loads_chain.invoke({"fruit": "사과"}).content)
👉 이렇게 하면 JSON → LangChain 객체로 다시 불러와, 동일한 체인을 재사용할 수 있습니다.
✨ 중간 요약
- is_lc_serializable(): 직렬화 가능 여부 확인
- dumpd → dict, dumps → JSON 문자열 변환
- Pickle/JSON 파일로 저장 가능
- load / loads로 다시 LangChain 객체 복원
2. 토큰 사용량 확인
1) 왜 토큰 사용량을 확인해야 할까?
LLM을 활용할 때는 항상 토큰(token) 개념을 염두에 두어야 합니다.
- 프롬프트 토큰: 우리가 모델에 입력한 글자 수(토큰화 기준).
- 응답 토큰: 모델이 생성해낸 출력.
- 총 토큰 수 = 입력 + 출력
토큰은 곧 API 비용과 직결되며, 과도하게 사용하면 예상치 못한 요금 폭탄을 맞을 수 있습니다. 또한, 불필요하게 긴 프롬프트를 줄이는 데에도 토큰 분석이 필수적입니다.
2) LanChain에서 토큰 사용량 확인하기
LangChain은 get_openai_callback()이라는 유틸을 제공합니다.
with 블록 내부에서 실행되는 모든 LLM 호출이 자동으로 집계됩니다.
현재 이 기능은 OpenAI API 전용으로 제공됩니다.
먼저 단일 Chat 모델 호출에 대한 토큰 사용량을 추적하는 매우 간단한 예를 살펴보겠습니다.
from langchain.callbacks import get_openai_callback
from langchain_openai import ChatOpenAI
# 모델을 불러옵니다.
llm = ChatOpenAI(model_name="gpt-4o")
with get_openai_callback() 구문안에서 실행되는 모든 토큰 사용량/요금이 추적됩니다.
# callback을 사용하여 추적합니다.
with get_openai_callback() as cb:
result = llm.invoke("대한민국의 수도는 어디야?")
print(cb)
출력 결과:
Tokens Used: 23
Prompt Tokens: 15
Prompt Tokens Cached: 0
Completion Tokens: 8
Reasoning Tokens: 0
Successful Requests: 1
Total Cost (USD): $0.00011750000000000001
# callback을 사용하여 추적합니다.
with get_openai_callback() as cb:
result = llm.invoke("대한민국의 수도는 어디야?")
result = llm.invoke("대한민국의 수도는 어디야?")
print(f"총 사용된 토큰수: \\t\\t{cb.total_tokens}")
print(f"프롬프트에 사용된 토큰수: \\t{cb.prompt_tokens}")
print(f"답변에 사용된 토큰수: \\t{cb.completion_tokens}")
print(f"호출에 청구된 금액(USD): \\t${cb.total_cost}")
출력 결과:
총 사용된 토큰수: 46
프롬프트에 사용된 토큰수: 30
답변에 사용된 토큰수: 16
호출에 청구된 금액(USD): $0.00023500000000000002
➡️ get_openai_callback()은 단일 호출뿐만 아니라, 세션 단위에서의 총 사용량과 비용도 확인할 수 있습니다.
3. 마무리
오늘은 LangChain의 두 가지 중요한 기능을 살펴보았습니다.
- 직렬화/역직렬화: 프롬프트와 체인을 JSON이나 Pickle 형태로 저장하고 다시 불러와 재사용 가능
- 토큰 사용량 추적: OpenAI API 호출 시 사용된 토큰 수와 비용을 간단히 확인 가능
👉 직렬화는 워크플로우를 재사용할 때, 토큰 추적은 비용 관리와 최적화에 필수적인 기능입니다.
앞으로 RAG 파이프라인을 만들 때도 이 두 기능은 기본으로 챙겨야 할 요소가 될 것입니다 🚀
읽어주셔서 감사합니다.