RAG 19. 실시간 문서 수집: 늦게 온 이벤트와 오래된 근거 구분하기
새 문서가 들어왔다고 검색 결과가 최신이라는 뜻은 아닙니다. 순서가 뒤바뀐 갱신 이벤트, 중복 전달, 오래된 원문을 구분하는 작은 수집기를 만듭니다.
새 문서가 들어왔다고 검색 결과가 최신이라는 뜻은 아닙니다. 순서가 뒤바뀐 갱신 이벤트, 중복 전달, 오래된 원문을 구분하는 작은 수집기를 만듭니다.
그림은 처리 구조를 설명하는 개념도입니다. 실제 실행 화면이나 측정 결과가 아닙니다.
오늘 만들 결과
버전 2를 받은 뒤 버전 1이 도착해도 내용이 되돌아가지 않습니다. 원문 수정 후 5분이 지나면 검색에서 제외하며, 삭제 이후 늦게 들어온 옛 이벤트도 문서를 되살리지 못합니다. 네트워크 스트리밍 대신 메모리 안에서 이벤트 도착 순서를 재현합니다.
먼저 이해할 것
수집 시각은 우리 시스템이 이벤트를 받은 때이고, 원문 수정 시각은 내용이 바뀐 때입니다. 오래된 원문을 방금 가져와도 내용 자체가 최신이 되지는 않습니다. 이번 예제의 신선도 기준은 “원문이 수정된 지 300초 이내”입니다. 변하지 않는 정책 문서에는 너무 엄격할 수 있고, 짧게 유효한 배송 공지에는 유용할 수 있습니다.
버전은 같은 문서 안에서 증가하는 정수입니다. 원본 시스템이 이 순서를 신뢰할 수 있게 제공한다고 가정합니다. 벽시계 시각만 비교하면 시계 오차나 지연 때문에 순서를 잘못 판단할 수 있습니다. 삭제 이벤트도 버전을 남깁니다. 이 삭제 표식을 tombstone이라고 부릅니다. 내용은 검색하지 않되 예전 버전의 부활을 막는 경계로 사용합니다.
실행 준비와 전체 코드
Python 3.12.14에서 실행했습니다. Python 3.12와 표준 라이브러리만 있으면 됩니다. 빈 실습 폴더를 만들고 아래 파일들을 같은 폴더에 저장하세요. 다른 편의 코드나 유료 API 키는 필요하지 않습니다.
시각을 코드에 고정해 누구나 같은 결과를 재현하게 했습니다. 실제 현재 시각을 읽거나 외부 서버를 감시하는 코드가 아닙니다.
freshness.py
from datetime import datetime, timezone
def instant(value):
result = datetime.fromisoformat(value)
if result.tzinfo is None:
raise ValueError('시간대가 필요합니다.')
return result.astimezone(timezone.utc)
class Index:
def __init__(self):
self.documents = {}
def ingest(self, event, received_at):
received = instant(received_at)
updated = instant(event['updated_at'])
if updated > received or type(event['version']) is not int or event['version'] < 1:
raise ValueError('잘못된 시각 또는 버전입니다.')
if event['operation'] not in {'upsert', 'delete'}:
raise ValueError('지원하지 않는 이벤트입니다.')
old = self.documents.get(event['id'])
if old and event['version'] < old['version']:
return 'ignored_old'
if old and event['version'] == old['version']:
same = all(old[k] == event[k] for k in ('operation', 'text', 'updated_at'))
if not same:
raise ValueError('동일 버전에 서로 다른 내용이 들어왔습니다.')
return 'duplicate'
self.documents[event['id']] = dict(event, indexed_at=received.isoformat())
return 'applied'
def search(self, word, now, max_age_seconds=300):
current = instant(now)
hits, excluded = [], []
for doc in self.documents.values():
if doc['operation'] == 'delete' or word not in doc['text']:
continue
source_age = (current - instant(doc['updated_at'])).total_seconds()
if source_age < 0 or source_age > max_age_seconds:
excluded.append(doc['id'])
continue
hits.append(dict(doc, source_age_seconds=int(source_age)))
return hits, excluded
def event(version, updated_at, text, operation='upsert'):
return dict(id='notice', version=version, updated_at=updated_at,
text=text, operation=operation)
if __name__ == '__main__':
index = Index()
first = event(1, '2026-01-01T09:00:00+00:00', '배송 지연 2일')
second = event(2, '2026-01-01T09:04:00+00:00', '배송 정상화')
for item in (first, second, first, second):
status = index.ingest(item, '2026-01-01T09:04:30+00:00')
print(f"version={item['version']} {status}")
for now in ('2026-01-01T09:05:00+00:00', '2026-01-01T09:12:00+00:00'):
hits, stale = index.search('배송', now)
print(now, 'hits=', [(h['version'], h['text'], h['source_age_seconds']) for h in hits],
'excluded_stale=', stale)
index.ingest(event(3, '2026-01-01T09:13:00+00:00', '', 'delete'),
'2026-01-01T09:13:01+00:00')
print('late replay:', index.ingest(second, '2026-01-01T09:14:00+00:00'))
print('after delete:', index.search('배송', '2026-01-01T09:14:00+00:00')[0])
ingest는 버전이 낮으면 버리고, 같으면 내용까지 비교합니다. 같은 버전에 다른 본문이 들어오면 원본 시스템의 약속이 깨진 것이므로 조용히 덮어쓰지 않습니다. 새 버전만 사전에 저장합니다.
search는 삭제 표식을 먼저 제외한 뒤, 원문 나이를 초 단위로 계산합니다. 300초를 넘긴 일치 문서는 검색 결과에 넣지 않고 excluded에 구분합니다. “관련 자료 없음”과 “자료는 있지만 오래되어 제외됨”은 사용자에게 다른 설명이 필요한 상태이기 때문입니다. indexed_at도 보관하므로 수집 지연을 별도로 분석할 수 있습니다.
이 구현은 한 프로세스의 메모리 실습입니다. 재시작하면 사라지고, 동시 쓰기를 처리하지 않습니다. 실제 큐를 연결할 때는 이벤트와 처리 위치를 영속 저장하고, 장애 복구 시 재전달되어도 같은 결과가 되게 해야 합니다. 이 글에서 즉시 처리되는 출력은 외부 수집 지연을 측정한 실시간 성능 수치가 아닙니다.
직접 실행한 결과
python freshness.py
아래는 이 글의 가상 데이터로 로컬에서 실제 실행한 출력입니다. 언어 모델이나 임베딩 API를 호출한 결과가 아닙니다.
version=1 applied
version=2 applied
version=1 ignored_old
version=2 duplicate
2026-01-01T09:05:00+00:00 hits= [(2, '배송 정상화', 60)] excluded_stale= []
2026-01-01T09:12:00+00:00 hits= [] excluded_stale= ['notice']
late replay: ignored_old
after delete: []
09:05에는 원문 수정 후 60초라 검색됩니다. 09:12에는 480초가 지나 제외됩니다. 버전 3 삭제 뒤 버전 2를 다시 받아도 ignored_old로 처리됩니다.
테스트로 경계를 고정하기
다음을 test_example.py로 저장합니다.
import unittest
from freshness import Index, event
class FreshnessTests(unittest.TestCase):
def setUp(self):
self.index = Index()
self.item = event(2, '2026-01-01T09:04:00+00:00', '배송 정상화')
self.received = '2026-01-01T09:04:30+00:00'
self.index.ingest(self.item, self.received)
def test_duplicate_and_out_of_order(self):
self.assertEqual(self.index.ingest(self.item, self.received), 'duplicate')
old = event(1, '2026-01-01T09:00:00+00:00', '배송 지연')
self.assertEqual(self.index.ingest(old, self.received), 'ignored_old')
def test_stale_source_is_not_made_fresh_by_ingestion(self):
self.index.ingest(self.item, '2026-01-01T10:00:00+00:00')
hits, stale = self.index.search('배송', '2026-01-01T10:00:00+00:00')
self.assertEqual(hits, [])
self.assertEqual(stale, ['notice'])
def test_conflict(self):
with self.assertRaises(ValueError):
self.index.ingest(dict(self.item, text='다른 본문'), self.received)
def test_delete_keeps_version_barrier(self):
deleted = event(3, '2026-01-01T09:05:00+00:00', '', 'delete')
self.index.ingest(deleted, '2026-01-01T09:05:01+00:00')
self.assertEqual(self.index.ingest(self.item, '2026-01-01T09:06:00+00:00'), 'ignored_old')
self.assertEqual(self.index.search('배송', '2026-01-01T09:06:00+00:00')[0], [])
def test_future_time_rejected(self):
with self.assertRaises(ValueError):
self.index.ingest(self.item, '2026-01-01T08:00:00+00:00')
if __name__ == '__main__':
unittest.main()
python -m unittest -v
다섯 테스트가 역순·중복 전달, 오래된 근거, 같은 버전 충돌, 삭제 후 재전달, 미래 시각을 검사합니다.
test_conflict (test_example.FreshnessTests.test_conflict) ... ok
test_delete_keeps_version_barrier (test_example.FreshnessTests.test_delete_keeps_version_barrier) ... ok
test_duplicate_and_out_of_order (test_example.FreshnessTests.test_duplicate_and_out_of_order) ... ok
test_future_time_rejected (test_example.FreshnessTests.test_future_time_rejected) ... ok
test_stale_source_is_not_made_fresh_by_ingestion (test_example.FreshnessTests.test_stale_source_is_not_made_fresh_by_ingestion) ... ok
----------------------------------------------------------------------
Ran 5 tests in 0.000s
OK
직접 바꿔 볼 것
max_age_seconds를 30으로 줄여 09:05 검색이 제외되는지 확인하세요.- 배송 공지와 영구 정책에 서로 다른 신선도 기준을 두세요. 내용 수정 시각 대신 “원본을 마지막으로 확인한 시각”이 필요한 문서도 구분하세요.
- 프로세스를 껐다 켜도 버전 경계가 남도록 SQLite에 저장해 보세요.
- 삭제 표식을 언제 지울지 정하기 전에, 원본 큐가 얼마나 오래된 이벤트를 재전달할 수 있는지 확인하세요.
참고 자료
- Python datetime: 시간대가 있는 시각과 UTC 변환에 대한 공식 문서.
RAG 실전 포트폴리오 19/25
이전: 18. 재무 보고서 검색: 숫자와 기간·통화·단위를 함께 꺼내자