RAG 25. 로컬 API 운영: 상태 확인, 입력 제한, 본문 없는 로그와 스모크 테스트
터미널 검색기를 로컬 HTTP API로 감싸고 정상 동작을 다시 확인하는 방법을 만듭니다. 상태 확인, 입력 제한, 근거 출처, 본문 없는 로그까지 붙여 25편의 실습을 마무리합니다.
터미널 검색기를 로컬 HTTP API로 감싸고 정상 동작을 다시 확인하는 방법을 만듭니다. 상태 확인, 입력 제한, 근거 출처, 본문 없는 로그까지 붙여 25편의 실습을 마무리합니다.
그림은 처리 구조를 설명하는 개념도입니다. 실제 실행 화면이나 측정 결과가 아닙니다.
오늘 만들 결과
GET /health는 인덱스 상태를, POST /search는 질문과 관련된 근거를 반환합니다. smoke.py를 실행하면 임시 로컬 서버를 켜고 네 요청을 보낸 뒤 종료합니다. 인터넷에 배포하거나 유료 서비스를 만들 필요가 없습니다.
먼저 이해할 것
HTTP API는 정해진 주소와 요청 형식으로 프로그램 기능을 호출하는 인터페이스입니다. 상태 확인(health check)은 서버가 응답하고 필요한 자료가 준비되었는지 짧게 확인하는 요청입니다. 여기서는 메모리에 가상 문서 두 개가 있는지를 readiness, 즉 요청을 처리할 준비 상태로 사용합니다.
스모크 테스트(smoke test)는 핵심 기능이 크게 망가지지 않았는지 빠르게 확인하는 작은 통합 검사입니다. 이번에는 함수만 호출하지 않고 실제 로컬 TCP 연결로 HTTP 요청을 보냅니다. 검색 품질 전체나 대규모 동시 접속을 검증하는 부하 테스트는 아닙니다.
Python http.server는 학습용 로컬 실행에 사용합니다. 공식 문서는 운영용 사용을 권장하지 않습니다. 아래 서버는 인증·TLS·서비스 수준 요청 제한이 없고 공개 인터넷에 노출하면 안 됩니다. 127.0.0.1에만 바인딩해 같은 컴퓨터에서 실습합니다.
실행 준비와 전체 코드
Python 3.12.14에서 실행했습니다. Python 3.12와 표준 라이브러리만 있으면 됩니다. 빈 실습 폴더를 만들고 아래 파일들을 같은 폴더에 저장하세요. 다른 편의 코드나 유료 API 키는 필요하지 않습니다.
다음 두 파일을 모두 저장하세요. smoke.py는 자동으로 빈 포트를 골라 실행하고 종료합니다. service.py를 직접 실행할 때는 기본 포트 8765를 사용하며 RAG_PORT 환경변수로 바꿀 수 있습니다. API 키는 쓰지 않습니다.
smoke.py
import json
from threading import Thread
from urllib.error import HTTPError
from urllib.request import Request, urlopen
from service import make_server
def request(base, path, payload=None, raw=None):
data = raw if raw is not None else (json.dumps(payload).encode() if payload is not None else None)
req = Request(base + path, data=data, headers={'Content-Type': 'application/json'})
try:
with urlopen(req, timeout=3) as response:
return response.status, json.load(response)
except HTTPError as error:
with error:
return error.code, json.load(error)
def main():
records = []
server = make_server(audit=records.append)
worker = Thread(target=server.serve_forever, daemon=True)
worker.start()
base = f'http://127.0.0.1:{server.server_port}'
try:
checks = [('/health', None), ('/search', {'question': 'refunds days'}),
('/search', {'question': 'warranty'}), ('/search', {'question': ''})]
for path, payload in checks:
status, data = request(base, path, payload)
print(status, json.dumps(data, ensure_ascii=False))
assert len(records) == 4
assert all(set(row) == {'request_id', 'route', 'status', 'duration_ms'} for row in records)
print('smoke: 4 requests; audit has no question or document text')
finally:
server.shutdown()
server.server_close()
worker.join()
if __name__ == '__main__':
main()
service.py
import json
import os
import re
import time
import uuid
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
DOCUMENTS = (
{'id': 'policy-p1', 'page': 1, 'text': 'Refunds are available within 14 days.'},
{'id': 'support-p2', 'page': 2, 'text': 'Support hours are 09:00 to 18:00 UTC.'},
)
INDEX_VERSION = 'fictional-v1'
def tokens(text):
return set(re.findall(r'[a-z0-9]+', text.lower()))
def search(question):
terms = tokens(question)
scores = [(len(terms & tokens(doc['text'])), doc) for doc in DOCUMENTS]
scores.sort(key=lambda pair: (-pair[0], pair[1]['id']))
return [dict(doc) for score, doc in scores[:2] if score > 0]
class Handler(BaseHTTPRequestHandler):
server_version = 'LocalRAG/1'
sys_version = ''
def setup(self):
self.request.settimeout(2)
super().setup()
self.started = time.perf_counter()
def log_message(self, format, *args):
pass # No raw request path or body in access logs.
def reply(self, status, data, route):
request_id = uuid.uuid4().hex
body = json.dumps(data, ensure_ascii=False).encode('utf-8')
self.send_response(status)
self.send_header('Content-Type', 'application/json; charset=utf-8')
self.send_header('Content-Length', str(len(body)))
self.send_header('Cache-Control', 'no-store')
self.send_header('X-Request-ID', request_id)
self.end_headers()
self.server.audit({'request_id': request_id, 'route': route, 'status': status,
'duration_ms': round((time.perf_counter()-self.started)*1000, 3)})
self.wfile.write(body)
def do_GET(self):
if self.path != '/health':
return self.reply(404, {'error': 'not_found'}, 'unknown')
ready = bool(DOCUMENTS)
self.reply(200 if ready else 503,
{'ready': ready, 'index_version': INDEX_VERSION,
'documents': len(DOCUMENTS)}, 'health')
def do_POST(self):
if self.path != '/search':
return self.reply(404, {'error': 'not_found'}, 'unknown')
if self.headers.get('Transfer-Encoding'):
return self.reply(400, {'error': 'unsupported_transfer_encoding'}, 'search')
lengths = self.headers.get_all('Content-Length', [])
if len(lengths) != 1:
return self.reply(411, {'error': 'length_required'}, 'search')
try:
size = int(lengths[0])
except ValueError:
return self.reply(400, {'error': 'invalid_length'}, 'search')
if not 1 <= size <= 4096:
return self.reply(413, {'error': 'body_size'}, 'search')
if self.headers.get_content_type() != 'application/json':
return self.reply(415, {'error': 'json_required'}, 'search')
try:
raw = self.rfile.read(size)
if len(raw) != size:
raise ValueError('short body')
data = json.loads(raw.decode('utf-8'))
except TimeoutError:
return self.reply(408, {'error': 'read_timeout'}, 'search')
except (ValueError, UnicodeDecodeError):
return self.reply(400, {'error': 'invalid_json'}, 'search')
if not isinstance(data, dict) or set(data) != {'question'}:
return self.reply(400, {'error': 'invalid_fields'}, 'search')
question = data['question']
if not isinstance(question, str) or not 1 <= len(question.strip()) <= 200:
return self.reply(400, {'error': 'invalid_question'}, 'search')
hits = search(question)
self.reply(200, {'mode': 'retrieval_only', 'index_version': INDEX_VERSION,
'hits': hits}, 'search')
def make_server(port=0, audit=None):
server = ThreadingHTTPServer(('127.0.0.1', port), Handler)
server.audit = audit if audit is not None else lambda record: print(json.dumps(record), flush=True)
return server
def configured_port():
port = int(os.getenv('RAG_PORT', '8765'))
if not 1 <= port <= 65535:
raise ValueError('RAG_PORT must be between 1 and 65535')
return port
if __name__ == '__main__':
with make_server(configured_port()) as server:
print(f'local-only: http://127.0.0.1:{server.server_port}', flush=True)
try:
server.serve_forever()
except KeyboardInterrupt:
print('stopped')
service.py의 입력 경계는 본문 4,096바이트, 질문 1~200자, question 필드 하나입니다. JSON 파싱 오류나 필드 오류는 제한된 오류 코드만 반환합니다. 원본 예외·요청 본문·환경변수 전체를 응답에 넣지 않습니다. 2초 연결 타임아웃은 입력을 무한히 기다리는 일을 줄이지만 운영용 공격 방어를 완성하지는 않습니다.
reply는 요청 id, 경로 종류, 상태 코드, 처리 시간만 로그에 남깁니다. 사용자 질문이나 검색된 문서 본문은 기록하지 않습니다. 요청 id는 응답 헤더에도 넣어 사용자가 문의한 요청과 로그를 연결할 수 있게 했습니다. 여기서 duration_ms는 로깅 직전까지의 처리 시간으로, 클라이언트가 전체 응답을 받는 종단 간 시간과 다릅니다.
search 응답에는 mode=retrieval_only를 넣었습니다. 검색된 문장을 언어 모델 답변으로 오해하지 않게 하는 장치입니다. id와 page도 그대로 남깁니다. 문서가 비어 있으면 /health가 503을 반환합니다. 실제 서비스에서는 저장소 연결·인덱스 버전·최근 갱신 지연 등을 readiness 조건에 추가할 수 있습니다.
smoke.py는 서버를 별도 스레드로 띄우고 finally에서 shutdown, server_close, join을 호출합니다. 테스트에 실패해도 서버 포트를 계속 붙잡지 않게 하는 정리 코드입니다. 이 API는 가상 공개 자료만 다루며 22편의 사용자 권한 검사나 실제 인증을 통합한 서비스는 아닙니다.
직접 실행한 결과
python smoke.py
아래는 이 글의 가상 데이터로 로컬에서 실제 실행한 출력입니다. 언어 모델이나 임베딩 API를 호출한 결과가 아닙니다.
200 {"ready": true, "index_version": "fictional-v1", "documents": 2}
200 {"mode": "retrieval_only", "index_version": "fictional-v1", "hits": [{"id": "policy-p1", "page": 1, "text": "Refunds are available within 14 days."}]}
200 {"mode": "retrieval_only", "index_version": "fictional-v1", "hits": []}
400 {"error": "invalid_question"}
smoke: 4 requests; audit has no question or document text
200은 정상 처리, 400은 잘못된 입력을 뜻합니다. warranty에는 빈 hits가 돌아오고, 빈 질문은 invalid_question으로 거절됩니다. smoke 실행이 끝나면 임시 서버도 종료됩니다.
테스트로 경계를 고정하기
다음을 test_example.py로 저장합니다.
import json
import unittest
from threading import Thread
from smoke import request
from service import make_server
class ServiceTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.records = []
cls.server = make_server(audit=cls.records.append)
cls.worker = Thread(target=cls.server.serve_forever, daemon=True)
cls.worker.start()
cls.base = f'http://127.0.0.1:{cls.server.server_port}'
@classmethod
def tearDownClass(cls):
cls.server.shutdown()
cls.server.server_close()
cls.worker.join()
def test_loopback_binding_and_health(self):
self.assertEqual(self.server.server_address[0], '127.0.0.1')
status, data = request(self.base, '/health')
self.assertEqual(status, 200)
self.assertTrue(data['ready'])
def test_search_has_page_source(self):
status, data = request(self.base, '/search', {'question': 'refunds days'})
self.assertEqual(status, 200)
self.assertEqual(data['hits'][0]['id'], 'policy-p1')
self.assertEqual(data['hits'][0]['page'], 1)
def test_unknown_has_no_hits(self):
self.assertEqual(request(self.base, '/search', {'question': 'warranty'})[1]['hits'], [])
def test_invalid_json_and_fields(self):
self.assertEqual(request(self.base, '/search', raw=b'{broken')[0], 400)
for payload in ({'question': ''}, {'question': 123}, {'question': 'ok', 'key': 'secret'}):
self.assertEqual(request(self.base, '/search', payload)[0], 400)
def test_body_limit(self):
self.assertEqual(request(self.base, '/search', raw=b'x'*4097)[0], 413)
def test_unknown_path(self):
self.assertEqual(request(self.base, '/missing')[0], 404)
def test_safe_audit_fields(self):
request(self.base, '/search', {'question': 'private_marker_123'})
serialized = json.dumps(self.records)
self.assertNotIn('private_marker_123', serialized)
self.assertNotIn('14 days', serialized)
self.assertTrue(self.records)
self.assertTrue(all(set(row) == {'request_id', 'route', 'status', 'duration_ms'} for row in self.records))
if __name__ == '__main__':
unittest.main()
python -m unittest -v
일곱 테스트가 루프백 바인딩, 상태 확인, 페이지 근거, 빈 검색, 잘못된 JSON·필드, 큰 본문, 알 수 없는 경로, 로그의 본문 제외를 확인합니다. 일부 항목은 한 테스트에서 함께 검사합니다. 외부 배포·TLS·실제 모델 API·부하 테스트는 실행하지 않았습니다.
test_body_limit (test_example.ServiceTests.test_body_limit) ... ok
test_invalid_json_and_fields (test_example.ServiceTests.test_invalid_json_and_fields) ... ok
test_loopback_binding_and_health (test_example.ServiceTests.test_loopback_binding_and_health) ... ok
test_safe_audit_fields (test_example.ServiceTests.test_safe_audit_fields) ... ok
test_search_has_page_source (test_example.ServiceTests.test_search_has_page_source) ... ok
test_unknown_has_no_hits (test_example.ServiceTests.test_unknown_has_no_hits) ... ok
test_unknown_path (test_example.ServiceTests.test_unknown_path) ... ok
----------------------------------------------------------------------
Ran 7 tests in 0.513s
OK
직접 바꿔 볼 것
- 터미널 하나에서
python service.py를 실행하고, 브라우저에서 http://127.0.0.1:8765/health를 여세요. 서버는Ctrl+C로 종료합니다. 브라우저 주소창은GET요청이라/search실습에는 별도POST요청이 필요합니다. - 테스트를 깨뜨려 보세요. 응답의
page를 제거하거나 본문 제한을 없애면 어떤 검사가 잡는지 확인하세요. - 실제 서비스로 옮기기 전에 인증·권한 검사, 운영용 웹 서버, TLS, 동시성 제한, 비밀 관리, 로그 보관 정책, 인덱스 백업·복구 절차를 정하세요.
- 새 버전 배포 전에는 고정 질문 평가와 스모크 테스트를 실행하고, 실패하면 이전 코드와 인덱스 버전으로 되돌릴 절차를 준비하세요.
25편의 마지막 완성 기준은 “질문에 답이 나온다”를 넘어섭니다. 근거와 버전이 남고, 문서가 바뀌면 오래된 청크가 사라지며, 볼 수 없는 자료는 검색되지 않고, 오류와 비용을 관측할 수 있어야 합니다. 이번 로컬 API는 그 확인을 반복할 수 있는 작은 출발점입니다.
참고 자료
- Python http.server: HTTP 처리기·서버와 운영용 사용 제한.
- Python socketserver: serve_forever와 shutdown의 스레드 사용 조건.
RAG 실전 포트폴리오 25/25 · 완결