OpenAI Assistants API가 멈춰도 에이전트 세션을 살리는 로컬 백업 구축법
TuBrief 편집팀
2026년 8월 7일
0
컴퓨터/소프트웨어원본 영상을 바탕으로 AI의 도움을 받아 작성했습니다. 원본 영상이 기준입니다.
커뮤니티의 다른 글
댓글 (0)
Log in to leave a comment
아직 작성된 글이 없습니다
원본 영상을 바탕으로 AI의 도움을 받아 작성했습니다. 원본 영상이 기준입니다.
Log in to leave a comment
아직 작성된 글이 없습니다
혼자서 AI 에이전트로 서비스를 돌리다가 벤더 API가 느닷없이 셧다운되면 난감해집니다. 당장 2026년 8월 26일 OpenAI가 Assistants API v1 베타 엔드포인트를 완전히 꺼버리면 서버에 저장해둔 대화 스레드와 실행 이력은 그대로 증발합니다. 벤더 서버에 세션 상태를 전부 맡겨놓는 구조는 폭탄을 안고 일하는 것과 다름없습니다.
플랫폼에 끌려다니지 않으려면 대화 상태를 내 컴퓨터 파일 시스템으로 가져와야 합니다. LLM은 언제든 갈아끼우는 연산 도구로 쓰고, 대화 맥락과 프롬프트 자산은 로컬에서 직접 쥐고 흔드는 편이 훨씬 안전합니다.
OpenAI나 Anthropic 같은 제공업체의 서버에 세션을 맡기면 내부에서 대화 상태를 어떻게 압축하고 암호화하는지 알 길이 없습니다. 문제가 생겼을 때 내부를 들여다보는 감사 자체가 불가능합니다. 대화 턴이 쌓일 때마다 전체 맥락을 매번 재처리하는 바람에 토큰 비용은 제곱으로 늘어납니다. 세션당 $0.03씩 붙는 Code Interpreter 비용이나 1GB당 월 $0.10인 File Search 저장 비용도 쥐도 새도 모르게 통장을 갉아먹습니다.
| 평가 항목 | 클라우드 관리형 세션 (OpenAI Assistants) | 로컬 파일 시스템 독립형 하네스 |
|---|---|---|
| 데이터 소유권 | 벤더 서버에 고립 (내보내기 불가) | 로컬 디렉토리에 JSON 파일로 저장 |
| 서비스 연속성 | 2026년 8월 26일 API 셧다운 시 스레드 소실 | 벤더 서버가 터져도 타 모델로 즉시 전환 |
| 컨텍스트 비용 | 턴 진행 시 스레드 전체 재처리로 토큰 낭비 | DiffMem 변경점만 선택 주입해 토큰 절감 |
| 디버깅 투명성 | 서버 측 불투명 압축으로 검증 불가 | 로컬 미들웨어 로그로 추론 과정 직접 확인 |
Microsoft CEO Satya Nadella가 언급했듯, 모델 제공업체 인프라에 세션 관리를 맡기는 구조는 위험합니다. 모델 자체가 아니라 독자 소유의 대화 컨텍스트가 진짜 자산이기 때문입니다. Docker 설문조사에서 개발자의 76%가 지적한 '행동적 종속성'도 결국 이 세션 주도권을 뺏긴 탓에 생깁니다.
LangGraph 프레임워크와 Custom Checkpointer를 조합하면 대화 상태를 로컬 JSON 파일로 안전하게 고정할 수 있습니다. 파일에 데이터를 쓸 때는 os.replace를 활용한 원자적 쓰기(Atomic Write) 방식을 씁니다. 작업 도중 전원이 꺼지거나 프로세스가 강제로 죽어도 데이터가 깨지지 않습니다.
import json
import os
from datetime import datetime
from typing import Any, Dict, List
from dataclasses import dataclass, asdict
@dataclass
class LocalSessionState:
session_id: str
model_provider: str
model_name: str
system_prompt: str
messages: List[Dict[str, Any]]
metadata: Dict[str, Any]
updated_at: str
class LocalFileCheckpointer:
def __init__(self, base_dir: str = "./agent_workspace/sessions"):
self.base_dir = base_dir
os.makedirs(self.base_dir, exist_ok=True)
def _get_file_path(self, session_id: str) -> str:
return os.path.join(self.base_dir, f"{session_id}.json")
def save_state(self, state: LocalSessionState) -> str:
file_path = self._get_file_path(state.session_id)
temp_path = f"{file_path}.tmp"
state.updated_at = datetime.utcnow().isoformat()
data = asdict(state)
with open(temp_path, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
os.replace(temp_path, file_path)
return file_path
이렇게 세션 백업 환경을 세팅하면 됩니다.
agent_workspace를 만들고, 그 밑에 prompts/, sessions/, logs/ 디렉토리를 각각 둡니다. 운영 규칙은 AGENTS.md에 정리합니다.LocalFileCheckpointer를 가져와 대화가 한 턴씩 갱신될 때마다 JSON 파일로 스냅샷을 덮어쓰도록 연결합니다.GitSessionVersioner를 붙여 세션 파일에 변경이 생길 때마다 변경사항을 자동으로 Git에 커밋하도록 만듭니다.import subprocess
class GitSessionVersioner:
def __init__(self, repo_dir: str = "./agent_workspace"):
self.repo_dir = repo_dir
self._ensure_git_repo()
def _ensure_git_repo(self):
if not os.path.exists(os.path.join(self.repo_dir, ".git")):
subprocess.run(["git", "init"], cwd=self.repo_dir, check=True)
def commit_session(self, session_id: str, commit_message: str):
file_name = f"sessions/{session_id}.json"
subprocess.run(["git", "add", file_name], cwd=self.repo_dir, check=True)
status = subprocess.run(["git", "status", "--porcelain", file_name], cwd=self.repo_dir, capture_output=True, text=True)
if status.stdout.strip():
subprocess.run(["git", "commit", "-m", f"session({session_id}): {commit_message}"], cwd=self.repo_dir, check=True)
이 구조를 잡아두면 대화 스레드가 길어져도 전체 이력을 컨텍스트 윈도우에 억지로 밀어 넣을 필요가 없어집니다. DiffMem 변경점만 골라서 조회하니 불필요한 토큰 지출을 막고, 세션 복원 작업 시간도 2시간 이상 줄어듭니다.
클라우드 API는 에이전트가 무슨 생각으로 어떤 도구를 호출하는지 통째로 은폐하곤 합니다. 디버깅 답답함을 없애려면 LiteLLM Proxy나 미들웨어 인터셉터를 중간에 두고 들어오고 나가는 데이터를 직접 파일로 찍어봐야 합니다. Langfuse나 Weights & Biases Weave 같은 오픈소스 트레이싱 도구를 로컬 Docker로 띄우면 OpenAI의 type: function이나 Anthropic의 input_schema 차이까지 한곳에서 깔끔하게 모아볼 수 있습니다.
import json
import logging
from typing import Dict, Any, Optional
class AgentLoggingMiddleware:
def __init__(self, log_file_path: str):
self.logger = logging.getLogger("AgentLogger")
self.logger.setLevel(logging.DEBUG)
handler = logging.FileHandler(log_file_path, encoding="utf-8")
handler.setFormatter(logging.Formatter('[%(asctime)s] [%(levelname)s] %(message)s'))
self.logger.addHandler(handler)
def on_pre_tool_execution(self, tool_name: str, tool_args: Dict[str, Any], tool_call_id: str):
log_entry = {"event": "PreToolUse", "tool_call_id": tool_call_id, "tool_name": tool_name, "arguments": tool_args}
self.logger.info(f"TOOL_CALL_INIT: {json.dumps(log_entry, ensure_ascii=False)}")
def on_post_tool_execution(self, tool_call_id: str, result: Any, error: Optional[str] = None):
log_entry = {"event": "PostToolUse", "tool_call_id": tool_call_id, "result": result, "error": error}
if error:
self.logger.error(f"TOOL_CALL_FAILED: {json.dumps(log_entry, ensure_ascii=False)}")
else:
self.logger.info(f"TOOL_CALL_SUCCESS: {json.dumps(log_entry, ensure_ascii=False)}")
실행 로그 수집은 세 단계로 마칩니다.
PreToolUse와 PostToolUse 지점에 AgentLoggingMiddleware를 끼워 넣습니다.로그만 제대로 쌓아도 도구 인자에서 터지는 에러 원인을 한 번에 찾아낼 수 있습니다. 엉뚱한 루프를 돌며 버리는 API 토큰 비용이 눈에 띄게 줄어듭니다.
OpenAI API는 도구 인자를 직렬화된 JSON 문자열로 주고받으며 tool 역할을 따로 둡니다. 반면 Anthropic Messages API는 파싱된 딕셔너리 객체와 user 메시지 내부의 tool_result 블록을 사용합니다. 규격이 다르다 보니 백업된 JSON 데이터를 그대로 Anthropic에 찌르면 HTTP 400 에러가 납니다. 중간에서 규격을 맞춰주는 어댑터가 있어야 합니다.
import json
from typing import List, Dict, Any, Tuple
class CrossModelSessionAdapter:
@staticmethod
def openai_to_anthropic_format(system_prompt: str, openai_messages: List[Dict[str, Any]]) -> Tuple[str, List[Dict[str, Any]]]:
anthropic_messages = []
i = 0
while i < len(openai_messages):
msg = openai_messages[i]
role = msg.get("role")
if role == "system":
system_prompt = msg.get("content", system_prompt)
i += 1
elif role == "user":
anthropic_messages.append({"role": "user", "content": msg.get("content")})
i += 1
elif role == "assistant":
content_blocks = []
if msg.get("content"):
content_blocks.append({"type": "text", "text": msg.get("content")})
if "tool_calls" in msg and msg["tool_calls"]:
for tc in msg["tool_calls"]:
args = tc["function"]["arguments"]
parsed_args = json.loads(args) if isinstance(args, str) else args
content_blocks.append({"type": "tool_use", "id": tc["id"], "name": tc["function"]["name"], "input": parsed_args})
anthropic_messages.append({"role": "assistant", "content": content_blocks})
i += 1
elif role == "tool":
tool_results = []
while i < len(openai_messages) and openai_messages[i].get("role") == "tool":
t_msg = openai_messages[i]
tool_results.append({"type": "tool_result", "tool_use_id": t_msg.get("tool_call_id"), "content": t_msg.get("content")})
i += 1
anthropic_messages.append({"role": "user", "content": tool_results})
return system_prompt, anthropic_messages
마이그레이션 작업도 간단합니다.
sessions/{session_id}.json 파일에서 대화 이력과 도구 실행 데이터를 읽어옵니다.CrossModelSessionAdapter를 실행해 OpenAI 포맷을 Anthropic 규격으로 바꿉니다.[System Context Injection]
본 대화는 기존 세션(ID: sess_99812)에서 마이그레이션된 연속된 작업입니다.
이전 모델과의 도구 실행 결과 및 대화 맥락이 메시지 이력에 포함되어 있습니다.
제시된 이전 도구 호출 결과(tool_result)를 바탕으로 중단된 지점부터 작업을 이어가십시오.
특정 벤더 서버가 먹통이 되거나 서비스 정책을 갑자기 바꿔도 상관없습니다. 몇 줄의 변환 코드로 대화 흐름을 100% 유지한 채 다른 LLM으로 갈아타면 그만입니다. 세션 데이터를 내 디렉토리에 물리적으로 쥐고 있을 때 1인 개발자의 비즈니스 연속성이 비로소 확보됩니다.