문서 수십 개를 AI에게 하나씩 요약시키는 스크립트를 돌리다가 중간에 오류로 멈추면, 다시 돌릴 때 이미 받은 답까지 처음부터 또 요청하게 됩니다. 시간도 들고, 유료 API라면 같은 답에 돈을 두 번 냅니다. AI 응답 캐시는 이 문제를 막는 가장 간단한 장치입니다. 질문을 키로 삼아 답을 SQLite 파일에 저장해 두고, 같은 질문이 다시 오면 AI 대신 파일에서 꺼내 줍니다. 이 글에서는 표준 라이브러리만으로 캐시를 만들고, 이 블로그의 실제 원고 30개로 시험했습니다. 첫 실행 30.32초가 재실행 때 0.23초가 됐는데, 두 프로세스를 동시에 돌리자 캐시가 있는데도 AI 호출이 60회로 두 배가 되는 문제가 나와서 고친 과정까지 정리합니다. 환경은 윈도우 11, Python 3.12.10입니다.
미리 밝혀 둘 점이 있습니다. 유료 API 키 없이 캐시 동작만 정확히 재려고, 실제 AI 대신 1초 기다렸다가 답을 돌려주는 대역 함수(fake_ai)를 썼습니다. 그래서 아래 “AI 호출” 시간은 실제 API 응답 시간이 아니라 이 대역 함수의 1초입니다. 반면 캐시 쪽(SQLite 조회·저장) 시간과 호출 횟수는 실제로 잰 값입니다. 쓰실 때는 fake_ai 자리에 각자 쓰는 API 호출 함수만 넣으면 됩니다.
캐시 키에 무엇을 넣어야 하나
캐시의 핵심은 “같은 질문”을 어떻게 판단하느냐입니다. 질문 글자만 보면 안 됩니다. 같은 질문이라도 모델, temperature, 시스템 프롬프트가 다르면 답이 달라지기 때문입니다. 그래서 답을 바꿀 수 있는 값을 전부 JSON으로 묶고 SHA-256 해시를 키로 썼습니다. 질문은 앞뒤 공백과 연속 공백만 한 칸으로 정리했습니다. 복사해 붙이다 줄바꿈이 하나 더 들어가는 정도로는 다른 질문 취급을 하지 않게 하려는 것입니다.
"""AI 답변 캐시 — 같은 질문을 두 번 보내지 않게 SQLite에 저장 (표준 라이브러리만)"""
import hashlib
import json
import sqlite3
import time
def fake_ai(prompt, model, temperature):
"""실제 AI API 대신 쓰는 대역 함수. 1초 기다렸다가 답을 돌려준다.
실제로 쓸 때는 이 함수만 각자 쓰는 API 호출로 바꾸면 된다."""
time.sleep(1.0)
return f'[{model}] {prompt[:40]} 에 대한 답변 (생성 시각 {time.strftime("%H:%M:%S")})'
def normalize(prompt):
# 앞뒤 공백 제거 + 연속 공백·줄바꿈을 한 칸으로. 내용은 건드리지 않는다.
return ' '.join(prompt.split())
def make_key(prompt, model, temperature, system):
# 답을 바꿀 수 있는 값은 전부 키에 넣는다. 하나라도 빠지면 엉뚱한 답이 재사용된다.
raw = json.dumps(
{'p': normalize(prompt), 'm': model, 't': temperature, 's': system},
ensure_ascii=False, sort_keys=True,
)
return hashlib.sha256(raw.encode('utf-8')).hexdigest()
class AICache:
def __init__(self, path, ttl_days=None, call=fake_ai):
self.db = sqlite3.connect(path)
self.db.execute(
'CREATE TABLE IF NOT EXISTS cache ('
' key TEXT PRIMARY KEY, model TEXT, prompt TEXT, answer TEXT,'
' created REAL, hits INTEGER DEFAULT 0)'
)
self.db.commit()
self.ttl = ttl_days * 86400 if ttl_days else None
self.call = call
self.hit = self.miss = 0
def ask(self, prompt, model='model-a', temperature=0, system=''):
key = make_key(prompt, model, temperature, system)
row = self.db.execute(
'SELECT answer, created FROM cache WHERE key = ?', (key,)
).fetchone()
if row and (self.ttl is None or time.time() - row[1] < self.ttl):
self.db.execute('UPDATE cache SET hits = hits + 1 WHERE key = ?', (key,))
self.db.commit()
self.hit += 1
return row[0]
answer = self.call(prompt, model, temperature)
self.db.execute(
'INSERT OR REPLACE INTO cache (key, model, prompt, answer, created, hits)'
' VALUES (?, ?, ?, ?, ?, 0)',
(key, model, normalize(prompt), answer, time.time()),
)
self.db.commit()
self.miss += 1
return answer
def close(self):
self.db.close()이 첫 버전으로 같은 질문을 조금씩 바꿔 보냈습니다.
$ python test_basic.py
처음 질문 AI 호출 1009.6ms
똑같이 다시 캐시 적중 6.9ms
앞뒤·중간 공백만 다름 캐시 적중 6.8ms
물음표 하나 추가 AI 호출 1010.0ms
모델만 다름 AI 호출 1010.0ms
temperature만 다름 AI 호출 1009.0ms
적중 2회, 호출 4회공백만 다른 질문은 캐시에서 나왔고, 물음표 하나를 붙이거나 모델·temperature를 바꾸면 새로 호출했습니다. 의도한 대로입니다. 물음표 하나 차이까지 같은 질문으로 보고 싶을 수도 있지만, 정규화를 세게 할수록 다른 질문에 엉뚱한 답이 나갈 위험이 커집니다. 저는 공백 정리까지만 하는 쪽을 택했습니다. 그런데 캐시 적중인데도 6.9ms가 걸린 게 눈에 걸렸습니다. 이 이유는 뒤에서 확인합니다.
원고 30개 배치 실측 — 30초가 0.23초로
실제 쓰임새에 가깝게, 이 블로그 저장소의 원고 파일 30개(합계 453,470바이트)를 docs 폴더에 복사하고 하나씩 “세 줄로 요약해 줘”를 보내는 배치를 만들었습니다.
"""폴더의 문서마다 AI에게 '요약해 줘'를 보내는 배치 작업
python batch.py docs
"""
import pathlib
import sys
import time
from ai_cache import AICache
sys.stdout.reconfigure(encoding='utf-8')
folder = pathlib.Path(sys.argv[1])
cache = AICache('batch.db')
s = time.perf_counter()
for f in sorted(folder.glob('*.js')):
text = f.read_text(encoding='utf-8')
cache.ask('다음 원고를 세 줄로 요약해 줘.\n\n' + text)
took = time.perf_counter() - s
print(f'문서 {cache.hit + cache.miss}개: AI 호출 {cache.miss}회, 캐시 적중 {cache.hit}회, {took:.2f}초')
cache.close()$ python batch.py docs
문서 30개: AI 호출 30회, 캐시 적중 0회, 30.32초
$ python batch.py docs
문서 30개: AI 호출 0회, 캐시 적중 30회, 0.23초
# 앞 3개 파일 끝에 한 줄씩 덧붙인 뒤
$ python batch.py docs
문서 30개: AI 호출 3회, 캐시 적중 27회, 3.26초첫 실행은 대역 함수 1초씩 30번이라 30.32초, 바로 다시 돌리면 30개 모두 캐시에서 나와 0.23초였습니다. 파일 3개 끝에 한 줄씩 덧붙이고 돌리니 바뀐 3개만 다시 호출했습니다. 문서 내용이 질문에 들어가니 내용이 바뀌면 키도 자동으로 바뀝니다. 중간에 멈춘 배치를 다시 돌릴 때 이미 끝난 부분은 공짜로 지나가는 셈입니다.
문제 1 — 답은 2,475자인데 DB가 528KB
배치를 돌린 뒤 DB를 열어 보니 생각보다 컸습니다.
$ python inspect_db.py batch.db
batch.db: 행 33개, 파일 528,384바이트, 질문 글자 합 294,447, 답변 글자 합 2,475
한 번도 재사용 안 된 행: 3답변은 다 합쳐 2,475자인데 파일이 528KB입니다. 확인용으로 질문 전문을 prompt 칸에 저장하게 해 둔 탓에 원고 내용이 통째로 DB에 들어간 것입니다. 키가 이미 질문의 해시라서 조회에는 전문이 필요 없습니다. 또 수정하기 전 원고 3개에 대한 옛 행이 한 번도 쓰이지 않은 채 남아 있었습니다. 문서를 자주 고치면 이런 행이 계속 쌓입니다.
문제 2 — 동시에 돌리면 AI 호출이 두 배
같은 배치를 두 창에서 실수로 동시에 실행하는 상황을 재현해 봤습니다. 빈 DB에서 batch.py 두 개를 동시에 띄웠습니다.
$ python parallel.py # 빈 DB에서 batch.py 두 개를 동시에 실행
문서 30개: AI 호출 30회, 캐시 적중 0회, 30.30초
문서 30개: AI 호출 30회, 캐시 적중 0회, 30.31초
종료 코드 [0, 0], 전체 30.39초SQLite의 “database is locked” 오류는 나지 않았습니다. 대신 두 프로세스가 같은 순간에 “아직 캐시에 없다”고 판단해서 같은 질문을 각자 보냈고, AI 호출이 30회면 될 일에 60회가 됐습니다. 오류 없이 조용히 비용만 두 배가 되는 경우라 실제로 겪으면 알아채기 어려울 것 같습니다. 조회하고 나서 저장할 때까지 1초 동안 다른 프로세스가 끼어들 틈이 있는 게 원인입니다.
문제 3 — 적중 6.9ms의 정체는 commit
처음에 본 6.9ms를 따로 재 봤습니다. 같은 키를 200번 조회하면서, 적중 횟수를 올리고 매번 commit하는 경우와 조회만 하는 경우를 비교했습니다.
$ python hit_cost.py # 같은 키를 200번 조회
조회+적중수 commit: 6.83ms/회
조회만: 0.071ms/회두 번 돌려도 각각 6.65ms, 0.070ms로 비슷했습니다. 적중 시간의 거의 전부가 “몇 번 재사용됐나”를 기록하는 commit이었습니다. commit은 디스크에 확실히 쓰기를 기다리기 때문에 느립니다. 통계용 숫자 하나 때문에 조회보다 약 95배 느려진 셈입니다.
개선판 — 자리 맡기, 적중 수 모아 쓰기
세 문제를 고친 개선판입니다. 위의 ai_cache.py와 같은 폴더에 두고, fake_ai·make_key·normalize는 거기서 가져다 씁니다.
"""ai_cache.py 개선판
- 질문 전문 대신 앞 100자만 저장 (DB 크기)
- 적중 횟수는 매번 commit 하지 않고 모아 두었다가 close()에서 한 번에
- 여러 프로세스가 같은 질문을 동시에 보내지 않도록 '자리 맡기' 행을 먼저 넣음
"""
import sqlite3
import time
from ai_cache import fake_ai, make_key, normalize
class AICache:
def __init__(self, path, ttl_days=None, call=fake_ai, wait_limit=120):
self.db = sqlite3.connect(path, timeout=30)
self.db.execute(
'CREATE TABLE IF NOT EXISTS cache ('
' key TEXT PRIMARY KEY, model TEXT, preview TEXT, answer TEXT,'
' created REAL, hits INTEGER DEFAULT 0)'
)
self.db.commit()
self.ttl = ttl_days * 86400 if ttl_days else None
self.call = call
self.wait_limit = wait_limit
self.pending_hits = {}
self.hit = self.miss = self.waited = 0
def _fresh(self, row):
return row and row[0] is not None and (self.ttl is None or time.time() - row[1] < self.ttl)
def ask(self, prompt, model='model-a', temperature=0, system=''):
key = make_key(prompt, model, temperature, system)
sql = 'SELECT answer, created FROM cache WHERE key = ?'
row = self.db.execute(sql, (key,)).fetchone()
if self._fresh(row):
self.pending_hits[key] = self.pending_hits.get(key, 0) + 1
self.hit += 1
return row[0]
if row and row[0] is not None: # 유효기간 지난 답 → 지우고 새로
self.db.execute('DELETE FROM cache WHERE key = ?', (key,))
self.db.commit()
# 자리 맡기: answer가 NULL인 행을 먼저 넣는다. 이미 누가 넣었으면 rowcount가 0.
cur = self.db.execute(
'INSERT OR IGNORE INTO cache (key, model, preview, answer, created) VALUES (?, ?, ?, NULL, ?)',
(key, model, normalize(prompt)[:100], time.time()),
)
self.db.commit()
if cur.rowcount == 0: # 다른 프로세스가 지금 묻는 중 → 답이 들어올 때까지 기다림
deadline = time.time() + self.wait_limit
while time.time() < deadline:
time.sleep(0.2)
row = self.db.execute(sql, (key,)).fetchone()
if self._fresh(row):
self.pending_hits[key] = self.pending_hits.get(key, 0) + 1
self.waited += 1
self.hit += 1
return row[0]
# 상대가 죽었다고 보고 직접 호출 (아래로 진행)
try:
answer = self.call(prompt, model, temperature)
except Exception:
self.db.execute('DELETE FROM cache WHERE key = ? AND answer IS NULL', (key,))
self.db.commit()
raise
self.db.execute('UPDATE cache SET answer = ?, created = ? WHERE key = ?', (answer, time.time(), key))
self.db.commit()
self.miss += 1
return answer
def close(self):
if self.pending_hits:
self.db.executemany(
'UPDATE cache SET hits = hits + ? WHERE key = ?',
[(n, k) for k, n in self.pending_hits.items()],
)
self.db.commit()
self.db.close()- 질문은 앞 100자만
preview에 저장합니다. 사람이 DB를 열어 볼 때 어떤 질문인지 알아볼 정도면 충분합니다. - 적중 횟수는 메모리에 모았다가
close()에서executemany한 번으로 씁니다. - 자리 맡기: AI를 부르기 전에 답이 빈(
NULL) 행을INSERT OR IGNORE로 먼저 넣습니다. 키가 PRIMARY KEY라서 먼저 넣은 프로세스만 성공하고, 늦은 쪽은rowcount가 0이 됩니다. 늦은 쪽은 0.2초 간격으로 답이 채워지길 기다립니다. - 호출이 실패하면 빈 행을 지웁니다. 안 지우면 다음 실행이 존재하지 않는 답을 120초 동안 기다리게 됩니다.
처음 만든 개선판으로 동시 실행을 다시 했더니 호출은 30회로 줄었는데, 결과에 이상한 숫자가 있었습니다.
$ python parallel2.py # 개선판으로 동시 실행 (첫 시도)
문서 30개: AI 호출 30회, 캐시 적중 0회(그중 남의 답 기다림 0회), 30.53초
문서 30개: AI 호출 0회, 캐시 적중 30회(그중 남의 답 기다림 30회), 30.52초
종료 코드 [0, 0], 전체 30.62초
행 30 파일 24576 바이트, hits합 0두 번째 프로세스가 30번 적중했는데 hits 합이 0입니다. 남의 답을 기다려서 받은 경우에는 적중 수를 모으는 줄을 빠뜨렸던 것입니다. 위 코드처럼 기다림 분기에도 pending_hits를 올리는 한 줄을 넣고 다시 돌렸습니다.
# 기다림 분기에 pending_hits 한 줄 추가 후
$ python parallel2.py
문서 30개: AI 호출 30회, 캐시 적중 0회(그중 남의 답 기다림 0회), 30.54초
문서 30개: AI 호출 0회, 캐시 적중 30회(그중 남의 답 기다림 30회), 30.52초
종료 코드 [0, 0], 전체 30.63초
$ python batch2.py docs
문서 30개: AI 호출 0회, 캐시 적중 30회(그중 남의 답 기다림 0회), 0.02초
행 30 파일 24576 바이트, hits합 60| 항목 | 첫 버전 | 개선판 |
|---|---|---|
| 동시 2프로세스 AI 호출 | 60회 | 30회 |
| DB 크기 (30행) | 475,136바이트 | 24,576바이트 |
| 30개 전부 적중하는 재실행 | 0.23초 | 0.02초 |
| 적중 수 기록 | 적중마다 commit | close()에서 한 번 |
한 가지 기대와 달랐던 점은 전체 시간입니다. 두 번째 프로세스는 첫 번째가 받는 답을 순서대로 기다리기만 해서 둘 다 30초가 걸렸습니다. 중복 호출은 막았지만 일을 나눠서 빨라진 건 아닙니다. 빨리 끝내는 게 목적이라면 캐시가 아니라 작업을 나눠 맡기는 쪽을 따로 만들어야 합니다.
유효기간과 호출 실패도 확인
환율이나 날씨처럼 시간이 지나면 답이 달라지는 질문은 캐시에 유효기간을 둬야 합니다. 시험용으로 유효기간을 2초로 두고, 실패하는 호출 함수도 넣어 봤습니다.
$ python test_ttl_fail.py
1차: [model-a] 오늘 환율 알려줘 에 대한 답변 (생성 시각 10:04:48)
즉시 2차: [model-a] 오늘 환율 알려줘 에 대한 답변 (생성 시각 10:04:48) (호출 1, 적중 1)
2.5초 뒤: [model-a] 오늘 환율 알려줘 에 대한 답변 (생성 시각 10:04:51) (호출 2, 적중 1)
호출 실패: 응답 없음
남은 빈 자리 행: 0
다시 시도: [model-a] 실패할 질문 에 대한 답변 (생성 시각 10:04:53) (호출 1)바로 다시 물으면 생성 시각이 같은 답(캐시)이 나오고, 2.5초 뒤에는 새로 호출해 생성 시각이 바뀌었습니다. 호출이 TimeoutError로 실패했을 때는 빈 자리 행이 0개로 정리됐고, 다음 실행에서 정상으로 다시 물었습니다. 실제로 쓸 때는 ttl_days=7처럼 일 단위로 넣으면 됩니다.
주의할 점
- 답을 바꾸는 값은 전부 키에 넣으세요. 모델 이름을 빼먹으면 모델을 바꿔도 예전 모델의 답이 계속 나옵니다.
- temperature가 높은 질문은 캐시가 “다양성”을 없앱니다. 매번 다른 아이디어를 받으려는 요청이라면 캐시를 끄거나 키에 회차 번호를 넣어야 합니다.
- 같은 DB를 여러 프로세스가 쓰면 자리 맡기가 필요합니다. 오류 없이 호출만 두 배가 됐습니다.
- 질문 전문을 저장하지 마세요. 문서를 통째로 넣는 질문이면 DB가 답보다 수백 배 커집니다. 개인정보가 든 문서라면 그 내용이 캐시 파일에 그대로 남는 문제도 생깁니다.
- 통계용 commit을 조심하세요. 적중 한 번이 0.07ms에서 6.7ms가 됐습니다.
- 캐시 파일은 git에 올리지 마세요. 답변과 질문 일부가 들어 있으니
.gitignore에*.db를 넣어 두는 게 안전합니다.
마무리
질문·모델·옵션을 묶은 해시를 키로, 답은 SQLite에. 재실행 30초가 0.02초가 됐고, 동시 실행 때 두 배로 나가던 호출도 자리 맡기 행으로 막았습니다.
가장 작은 첫걸음은 이렇습니다.
- 위
ai_cache.py와 개선판을 저장하고,fake_ai자리에 지금 쓰는 AI 호출 함수를 넣습니다. - 반복해서 돌리는 배치 스크립트에서 AI를 직접 부르던 줄을
cache.ask(...)로 바꿉니다. - 두 번 연속 실행해 두 번째에 “AI 호출 0회”가 나오는지 확인합니다.
다음 글에서는 파이썬 파일 경로를 다루는 os.path 코드를 pathlib로 바꿔 보면서, 실제로 얼마나 짧아지고 어디서 헷갈리는지 직접 확인해 보겠습니다.