Written with AI assistance from the source video. The video is the authority.
21:24Maximilian Schwarzmüller
Log in to leave a comment
No posts yet
独自运行 AI 代理服务时，如果供应商 API 突然宕机，就会陷入进退两难的境地。如果 OpenAI 在 2026 年 8 月 26 日直接关闭 Assistants API v1 Beta 端点，保存在服务器中的对话线程和执行历史将会瞬间蒸发。将所有的会话状态全部托付给供应商服务器的架构，无异于抱着定时炸弹工作。
要想不被平台牵着鼻子走，就必须将对话状态拉取到自己的计算机文件系统中。将 LLM 用作随时可以替换的计算工具，而在本地直接掌控对话上下文和提示词资产，要安全得多。
如果将会话托付给 OpenAI 或 Anthropic 等提供商的服务器，就无法得知其内部是如何压缩和加密对话状态的。当出现问题时，甚至无法进行内部审计。由于每积累一轮对话，都需要重新处理整个上下文，导致 Token 成本呈平方级增长。每个会话 0.03 美元的 Code Interpreter 费用，或每 GB 每月 0.10 美元的 File Search 存储费用，也会在不知不觉中蚕食资金。
| 评估项目 | 云端托管会话 (OpenAI Assistants) | 本地文件系统独立架构 |
|---|---|---|
| 数据所有权 | 隔离在供应商服务器中（无法导出） | 以 JSON 文件形式保存在本地目录中 |
| 服务连续性 | 2026 年 8 月 26 日 API 停机时线程丢失 | 供应商服务器崩溃时可立即切换至其他模型 |
| 上下文成本 | 进行多轮对话时重新处理整个线程导致 Token 浪费 | 仅选择性注入 DiffMem 变更点以节省 Token |
| 调试透明度 | 因服务器端不透明压缩而无法验证 | 通过本地中间件日志直接确认推理过程 |
正如微软 CEO Satya Nadella 所言，将会话管理托付给模型提供商基础设施的架构存在风险。因为真正宝贵的资产不是模型本身，而是独家拥有的对话上下文。正如 Docker 调查中 76% 的开发者指出的那样，“行为依赖性”归根结底也是因为失去了会话主导权而产生的。
将 LangGraph 框架与 Custom Checkpointer 相结合，可以安全地将对话状态固定为本地 JSON 文件。向文件写入数据时，采用利用 os.replace 的原子写入（Atomic Write）方式。即使在操作过程中断电或进程被强制终止，数据也不会损坏。
`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
`
只需按如下方式设置会话备份环境即可。
agent_workspace，并在其下方分别放置 prompts/、sessions/ 和 logs/ 目录。运营规则整理在 AGENTS.md 中。LocalFileCheckpointer，并进行连接，以便每当对话更新一轮时，就将快照覆盖写入 JSON 文件。GitSessionVersioner 附加到 Git 后置钩子（Post-Hook）管道中，使得会话文件每次发生变更时，都会自动将变更事项提交到 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)
`
建立此架构后，即使对话线程变长，也无需将整个历史记录强行塞入上下文窗口。由于只挑选查询 DiffMem 变更点，因此可以阻止不必要的 Token 支出，会话恢复工作时间也能缩短 2 小时以上。
云端 API 常常将代理的思考过程以及调用的工具完全隐蔽。为了消除调试时的苦恼，可以将 LiteLLM Proxy 或中间件拦截器置于中间，直接将传入和传出的数据写入文件。如果在本地 Docker 中运行 Langfuse 或 Weights & Biases Weave 等开源追踪工具，就可以在一处干净地收集 OpenAI 的 type: function 或 Anthropic 的 input_schema 之间的差异。
`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)}")
`
执行日志的收集分为三个步骤完成：
PreToolUse 和 PostToolUse 节点中插入 AgentLoggingMiddleware。只要妥善积累日志，就能一次性找出在工具参数中爆发的错误原因。因在错误的循环中盘旋而浪费的 API Token 费用将显著减少。
OpenAI API 以序列化 JSON 字符串的形式收发工具参数，并单独设置 tool 角色。相比之下，Anthropic Messages API 使用解析后的字典对象以及 user 消息内部的 tool_result 块。由于规范不同，如果将备份的 JSON 数据原封不动地提交给 Anthropic，就会产生 HTTP 400 错误。因此，中间必须有一个适配器来匹配规范。
`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
`
迁移工作也很简单：
sessions/{session_id}.json 文件中读取对话历史和工具执行数据。CrossModelSessionAdapter 将 OpenAI 格式转换为 Anthropic 规范。`text
[System Context Injection]
本对话是自原有会话（ID: sess_99812）迁移而来的连续工作。
消息历史中包含与先前模型的工具执行结果以及对话上下文。
请基于所提供的先前工具调用结果（tool_result），从中断的位置继续开展工作。
`
特定供应商服务器崩溃或突然更改服务政策都无所谓。只需几行转换代码，就能在 100% 保持对话流向的情况下更换为其他 LLM。当将会话数据物理掌控在自己的目录中时，独立开发者的业务连续性方能得到切实保障。