Como criar um backup local para salvar sessões de agentes caso a OpenAI Assistants API pare
TuBrief 편집팀
2026년 8월 7일
0
Computing/Software원본 영상을 바탕으로 AI의 도움을 받아 작성했습니다. 원본 영상이 기준입니다.
커뮤니티의 다른 글
댓글 (0)
Log in to leave a comment
아직 작성된 글이 없습니다
원본 영상을 바탕으로 AI의 도움을 받아 작성했습니다. 원본 영상이 기준입니다.
Log in to leave a comment
아직 작성된 글이 없습니다
Rodar serviços com agentes de IA sozinho e de repente ver a API do fornecedor cair é desesperador. Se a OpenAI desligar completamente o endpoint beta da Assistants API v1 em 26 de agosto de 2026, os threads de conversação e o histórico de execução salvos no servidor evaporarão instantaneamente. Uma estrutura que confia todo o estado da sessão ao servidor do fornecedor é o equivalente a trabalhar sentado em cima de uma bomba.
Para não ficar refém de plataformas, você precisa trazer o estado da conversa para o sistema de arquivos do seu próprio computador. É muito mais seguro usar o LLM como uma ferramenta de computação substituível a qualquer momento e manter o contexto da conversa e os ativos de prompt sob seu controle direto localmente.
Quando você deixa sessões nos servidores de provedores como OpenAI ou Anthropic, não há como saber como o estado da conversa é compactado e criptografado internamente. Se algo der errado, uma auditoria interna é simplesmente impossível. Como todo o contexto é reprocessado a cada turno de conversa acumulado, os custos de token aumentam exponencialmente. Custos do Code Interpreter a $0,03 por sessão ou armazenamento do File Search a $0,10 por GB ao mês também drenam sua conta bancária sem que você perceba.
| Item de Avaliação | Sessão Gerenciada na Nuvem (OpenAI Assistants) | Harness Independente em Sistema de Arquivos Local |
|---|---|---|
| Propriedade de Dados | Isola-se no servidor do fornecedor (impossível exportar) | Salvo como arquivo JSON em diretório local |
| Continuidade do Serviço | Perda de threads se a API cair em 26 de agosto de 2026 | Transição imediata para outro modelo se o servidor do fornecedor cair |
| Custo de Contexto | Desperdício de tokens com reprocessamento total do thread por turno | Injeção seletiva apenas de alterações via DiffMem para economizar tokens |
| Transparência de Depuração | Impossível verificar devido à compactação opaca do servidor | Verificação direta do processo de inferência via logs de middleware local |
Como mencionou o CEO da Microsoft, Satya Nadella, a estrutura de delegar o gerenciamento de sessões para a infraestrutura de provedores de modelos é arriscada. Isso acontece porque o verdadeiro ativo não é o modelo em si, mas sim o seu contexto de conversas de propriedade exclusiva. A "dependência comportamental" apontada por 76% dos desenvolvedores na pesquisa do Docker também surge, no fim das contas, por perder o controle sobre essas sessões.
A combinação do framework LangGraph com um Custom Checkpointer permite fixar o estado da conversa com segurança em arquivos JSON locais. Ao gravar dados em arquivos, utiliza-se o método de gravação atômica (Atomic Write) com os.replace. Mesmo que a energia caia ou o processo seja encerrado à força durante o trabalho, os dados não serão corrompidos.
`python
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
`
Basta configurar o ambiente de backup de sessões da seguinte forma:
agent_workspace dentro da pasta do projeto e, dentro dela, coloque os diretórios prompts/, sessions/ e logs/. As regras operacionais são organizadas em AGENTS.md.LocalFileCheckpointer do código Python acima e conecte-o para sobrescrever o snapshot em um arquivo JSON sempre que a conversa for atualizada a cada turno.GitSessionVersioner abaixo ao pipeline do Git post-hook para que, sempre que houver alterações nos arquivos de sessão, as mudanças sejam automaticamente comitadas no Git.`python
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)
`
Ao estabelecer essa estrutura, mesmo que o thread de conversação fique longo, não será mais necessário empurrar todo o histórico à força para dentro da janela de contexto. Como apenas as alterações do DiffMem são selecionadas e consultadas, evita-se gastos desnecessários com tokens e o tempo de restauração da sessão é reduzido em mais de 2 horas.
APIs em nuvem costumam ocultar completamente qual é o raciocínio do agente ou quais ferramentas ele está chamando. Para eliminar a frustração na depuração, coloque um LiteLLM Proxy ou um interceptador de middleware no meio do caminho e grave os dados de entrada e saída diretamente em arquivos. Se você rodar ferramentas de rastreamento de código aberto como Langfuse ou Weights & Biases Weave em um Docker local, poderá unificar e visualizar facilmente desde as diferenças de type: function da OpenAI até o input_schema da Anthropic.
`python
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)}")
`
A coleta de logs de execução é feita em três etapas:
AgentLoggingMiddleware nos pontos PreToolUse e PostToolUse do loop de execução do agente.Apenas acumulando logs corretamente, você consegue encontrar a causa dos erros gerados nos argumentos das ferramentas de uma só vez. O custo em tokens de API desperdiçados em loops incorretos diminui visivelmente.
A API da OpenAI envia e recebe argumentos de ferramentas como strings JSON serializadas e mantém o papel (role) tool separadamente. Em contrapartida, a API de Mensagens da Anthropic usa objetos de dicionário analisados e um bloco tool_result dentro da mensagem user. Como os formatos são diferentes, enviar dados JSON de backup diretamente para a Anthropic resultará em um erro HTTP 400. É preciso haver um adaptador no meio do caminho para ajustar o formato.
`python
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
`
O processo de migração também é simples:
sessions/{session_id}.json.CrossModelSessionAdapter para converter o formato da OpenAI para o padrão da Anthropic.`text
[System Context Injection]
Esta conversa é um trabalho contínuo migrado de uma sessão anterior (ID: sess_99812).
O contexto de conversa e os resultados da execução de ferramentas com o modelo anterior estão incluídos no histórico de mensagens.
Com base nos resultados de chamadas de ferramentas anteriores fornecidos (tool_result), continue o trabalho a partir do ponto onde foi interrompido.
`
Não importa se o servidor de um fornecedor específico cair ou se as políticas de serviço mudarem repentinamente. Basta migrar para outro LLM mantendo 100% do fluxo de conversação com algumas linhas de código de conversão. A continuidade de negócios de um desenvolvedor solo só é garantida quando você mantém fisicamente os dados de sessão em seu próprio diretório.