챗봇개발과 API 과금 줄이기

채팅 앱에서 대화의 연속성을 위해 챗 메모리를 사용합니다.

 

먼저 용어부터...
    - 턴 : 사용자의 요청과 봇의 답변을 한 쌍으로 묶음

 

아무 조건없이 메모리를 사용하면 모든 대화 내용을 기억합니다.
챗봇에서 '기억'이라는 의미는 좀 다릅니다.

예를 들어 LLM에 전달되는 내용으로 설명하겠습니다.

1. LLM에 전달(API) : 사용자 질문1과 챗봇의 답변1 -> 턴1
2. LLM에 전달 : (턴1 의 내용) + 질문2 + 답변2 -> 턴2
3. LLM에 전달 : (턴1 + 턴2) + 질문3 + 답변3 -> 턴3

이런 형태로 LLM에 전달됩니다. 

 

LLM에 전달된다는 것이 바로 입력토큰(Input Token)이 됩니다.
답변은 물론 출력토큰(Output Token)이구요.
11번째 대화가 이루어지면 1턴에서 10턴까지 내용이 입력토큰이 되어 토큰 사용량이 증가하여 API 과금이 많아집니다.
대화가 길어지면 길어질수록 입력토큰량이 많아져 API 사용 과금이 증가하는 것입니다.

이것을 해소하는 방법입니다.

 

첫째. 챗메모리에 기억하는 턴 수에 제한을 둡니다. 아래 코드는 턴 수를 20개 제한을 둔 것입니다.

def trim_session_history(self, session_id: str, keep_last: int = 5):
    if session_id in self.session_histories:
        history = self.session_histories[session_id]
        if len(history) > keep_last:
            self.session_histories[session_id] = history[-keep_last:]

trim_session_history(session_id, keep_last=20)

 

둘째. 채팅에서 챗봇의 답변에 그래프와 표가 있습니다. 이것은 값 이외의 것들이 많이 포함되어 있습니다. 값 이외의 것들이 메모리에 저장되지 않도록 코드를 구현합니다. html 코드 등을 제외합니다.

history_answer = answer_text + "".join(
    f"\n<q_data>{block}</q_data>" for block in q_data_blocks
)

add_to_history(session_id, question, history_answer, current_query) # 데이터 값들만 메모리에 저장합니다.

 

셋째. 프롬프트 캐싱 적용하기

1. 챗봇의 답변을 구하기 위한 프롬프트 모두 프롬프트 캐싱 적용하기

SystemMessage(content=[{"type": "text", "text": content, "cache_control": {"type": "ephemeral"}}])

    - 시스템 프롬프트 안에 있는 DB/엑셀/CSV 파일의 메타 정보도 1회 캐싱됩니다.

 

2. 챗 메모리 내용을 프롬프트 캐싱 적용하기 - 뒤에 자세히 설명하겠습니다.

if history_messages:
    last_hist_msg = history_messages[-1]
    last_hist_msg.content = [
        {"type": "text", "text": last_hist_msg.content, "cache_control": {"type": "ephemeral"}}
    ]
input_messages = history_messages + [HumanMessage(content=question)]

 

    첫째, 둘째는 적용한 상태였구요. 셋째 적용하여 과금을 측정해봤습니다.
        적용 전 : 대화는 7개 창. 평균 턴수 5개. 50개 질문.  --> 6.2$
        적용 후 : 대화창 1개. 15개 질문  --> 0.7$
    적용 후 과금이 적용 전 금액의 35% 정도입니다. 

 

넷째.batch API 사용하기

    - API 호출을 모아서 한꺼번에 LLM 실행. 50% 줄일 수 있다고 합니다.
    - 실제 적용해보지 않았습니다. 채팅에서는 사용자가 요청 후 바로 답을 기다리는 상황이라 적용하기 애매했습니다.
    - 정기보고서에 적용하기 : "매월 6일 09시에 보고서를 작성해주세요"라고 요청한 경우에도 문제가 생길 수 있습니다.
        - 배치 처리이기 때문에 6일 9시 이후에 보고서가 작성될 수 있습니다. 9시에 생성이 된다는 보장을 할 수 없습니다.
        - 보고서 작성 요청을 6일 01시로 조정하여 이 문제를 해결하려 시도할 수 있습니다. 이럴 경우에도 보고서에 필요한 데이터들 집계 시간이 작성 요청시각보다 늦어질 경우는 문제가 발생합니다.
    - batch API는 실제 적용해보지 않았어요.

 

<대화 턴에 프롬프트 캐싱 적용>
0. 캐싱 : Write 일 때는 1.25배 / Read 일 때는 0.1배
1. 20개 턴으로 제한을 걸었다고 가정하겠습니다.
    - 턴1 : 프롬프트 캐싱 Write
    - 턴2 : 턴1 Read + 턴2 Write
    - 턴3 : 턴1/턴2 Read + 턴3 Write
    - 턴20 : 턴1~턴19 Read + 턴20 Write
    - 턴21 : 턴1이 지워짐 --> 턴2~턴21 Write --> 과금 과다
    - 턴22 : 턴2가 지워짐 --> 턴3~턴22 Write --> 과금 과다
2. 턴20이 넘어가면 과금이 더 많아집니다. 
    - 제한된 턴까지는 캐싱에 추가되는 형태입니다.
    - 턴이 지워지면 프롬프트 캐싱을 다시 등록해야 합니다.
3. 버퍼를 줍니다. 버퍼를 10개
    - 턴30 : 턴1~턴29 Read + 턴30 Write
    - 턴31 : 턴1~턴10 삭제 + 턴11~턴31 Write
    - 턴32 : 턴11~턴31 Read + 턴32 Write
    - 턴41에서도 앞의 턴 10개 삭제, 턴 20개를 Write
이렇게 진행하도록 구현했습니다.

TARGET_TURNS = 20
THRESHOLD_TURNS = 28    
# 20블럭 한도가 있어요. 한 턴이 2개(질문/답변) 블럭을 차지하기에 꽉 채우는 것보다 여유를 줄려고 8개를 버퍼로 뒀어요.

    def trim_session_history(self, session_id: str, target: int = TARGET_TURNS, threshold: int = THRESHOLD_TURNS):
        """버퍼 트림: threshold를 넘기 전까지는 자르지 않고 계속 누적하다가,
        넘는 순간 한 번에 target 턴만 남긴다.
        (매 턴 target으로 슬라이딩하면 프롬프트 캐싱의 히스토리 시작 지점이 매 턴 바뀌어
        캐시가 계속 무효화되므로, 트림을 threshold 시점 1번으로 몰아서 캐시 유지 구간을 늘린다.)
        """
        if session_id in self.session_histories:
            history = self.session_histories[session_id]
            if len(history) > threshold:
                print(f"[TRIM] session={session_id}: {len(history)}턴 → {target}턴으로 정리 "
                      f"(threshold={threshold} 초과) - 이 턴은 캐시 프리픽스가 바뀌어 write가 발생함")
                self.session_histories[session_id] = history[-target:]

 

그리고 알아두면 좋은 것 두 가지.

1. 캐싱은 "압축"이 아니라 "미리 계산해둔 결과를 재사용"하는 것입니다.
2. 캐싱은 토큰수를 줄이는 것이 아닙니다. 토큰수는 그대로이고 과금이 적게 적용되는 것입니다.