كيفية إنشاء نسخة احتياطية محلية لحفظ جلسات الوكلاء حتى في حالة توقف OpenAI Assistants API
TuBrief 편집팀
2026년 8월 7일
0
Computing/Software원본 영상을 바탕으로 AI의 도움을 받아 작성했습니다. 원본 영상이 기준입니다.
커뮤니티의 다른 글
댓글 (0)
Log in to leave a comment
아직 작성된 글이 없습니다
원본 영상을 바탕으로 AI의 도움을 받아 작성했습니다. 원본 영상이 기준입니다.
Log in to leave a comment
아직 작성된 글이 없습니다
عندما تقوم بتشغيل خدمة باستخدام وكلاء الذكاء الاصطناعي بمفردك، يكون الأمر مزعجًا للغاية إذا تم إيقاف واجهة برمجة التطبيقات (API) الخاصة بالمورد فجأة. إذا قامت OpenAI بإيقاف نقطة نهاية Assistants API v1 Beta تمامًا في 26 أغسطس 2026، فسيتم تبخر سلاسل المحادثات وسجل التنفيذ المخزن على الخادم تمامًا. إن الهيكل الذي يعتمد كليًا على خادم المورد لتخزين حالة الجلسة لا يختلف عن العمل أثناء حمل قنبلة موقوتة.
لكل شخص لا يرغب في أن يكون مقيدًا بمنصة معينة، يجب جلب حالة المحادثة إلى نظام الملفات في جهازك المحلي. من الأمان بكثير استخدام نماذج اللغات الكبيرة (LLM) كأدوات حسابية يمكن تبديلها في أي وقت، والاحتفاظ بسياق المحادثة وأصول التوجيه (Prompts) محليًا.
عند ترك الجلسات على خوادم موفرين مثل OpenAI أو Anthropic، لا توجد طريقة لمعرفة كيف يتم ضغط وتشفير حالة المحادثة داخليًا. إن إجراء تدقيق داخلي عند حدوث مشكلة أمر مستحيل تمامًا. ومع تراكم جولات المحادثة، تتضاعف تكاليف الرموز (Tokens) أضعافًا مضاعفة بسبب إعادة معالجة السياق بأكمله في كل مرة. كما أن تكلفة Code Interpreter التي تبلغ 0.03 دولار لكل جلسة أو تكلفة تخزين File Search التي تبلغ 0.10 دولار شهريًا لكل جيجابايت تستنزف ميزانيتك في خفاء.
| بند التقييم | الجلسات المدارة سحابيًا (OpenAI Assistants) | نظام ربط مستقل يعتمد على نظام الملفات المحلي |
|---|---|---|
| ملكية البيانات | معزولة على خادم المورد (لا يمكن التصدير) | مخزنة كملفات JSON في الدليل المحلي |
| استمرارية الخدمة | ضياع السلاسل عند توقف API في 26 أغسطس 2026 | التبديل الفوري إلى نموذج آخر عند تعطل خادم المورد |
| تكلفة السياق | هدر الرموز بسبب إعادة معالجة السلسلة بالكامل | تقليل الرموز عن طريق الحقن الانتقائي لتغييرات DiffMem |
| شفافية تصحيح الأخطاء | التحقق مستحيل بسبب الضغط الغامض من جانب الخادم | التحقق المباشر من عملية الاستدلال عبر سجلات الوسيط المحلي |
وكما ذكر الرئيس التنفيذي لشركة Microsoft، Satya Nadella، فإن الهيكل الذي يعتمد على إدارة الجلسات في البنية التحتية لموفر النموذج ينطوي على مخاطر عالية. والسبب في ذلك هو أن سياق المحادثة المملوك بشكل خاص، وليس النموذج نفسه، هو الأصل الحقيقي. حتى أن "التبعية السلوكية" التي أشار إليها 76% من المطورين في استطلاع Docker تنشأ نتيجة فقدان السيطرة على هذه الجلسة.
من خلال الجمع بين إطار عمل LangGraph وأداة Custom Checkpointer، يمكنك تثبيت حالة المحادثة بأمان في ملفات JSON محلية. عند كتابة البيانات في الملفات، يتم استخدام طريقة الكتابة الذرية (Atomic Write) باستخدام os.replace. حتى إذا تم قطع التيار الكهربائي أو إيقاف العملية قسريًا أثناء العمل، فلن تتلف البيانات.
`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 من كود Python أعلاه وربطه بحيث يتم استبدال لقطة (Snapshot) ملف JSON في كل مرة يتم فيها تحديث جولة المحادثة.GitSessionVersioner أدناه بخط أنابيب خطاف Git (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 فقط، يتم منع إنفاق الرموز غير الضرورية، ويتم تقليل وقت استعادة الجلسة بأكثر من ساعتين.
غالبًا ما تخفي واجهات برمجة التطبيقات السحابية تفاصيل الأفكار التي يفكر بها الوكلاء والأدوات التي يستدعونها. للتخلص من إحباط تصحيح الأخطاء، يجب وضع LiteLLM Proxy أو وسيط اعتراض (Middleware Interceptor) في المنتصف وتسجيل البيانات الواردة والصادرة مباشرة في ملف. إذا قمت بتشغيل أداة تتبع مفتوحة المصدر مثل Langfuse أو Weights & Biases Weave في Docker محلي، يمكنك جمع الاختلافات مثل type: function في OpenAI أو input_schema في 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)}")
`
يتم جمع سجلات التنفيذ في ثلاث خطوات:
AgentLoggingMiddleware في نقاط PreToolUse وPostToolUse لحلقة تنفيذ الوكلاء.مجرد تجميع السجلات بشكل صحيح يمكن أن يتيح لك العثور على سبب الخطأ الذي يحدث في وسائط الأداة دفعة واحدة. سيتم تقليل تكاليف رموز API المهدرة في الحلقات الخاطئة بشكل ملحوظ.
تعمل واجهة برمجة تطبيقات OpenAI على إرسال واستقبال وسائط الأدوات كسلاسل JSON مُسلسلة وتخصص دور tool بشكل منفصل. من ناحية أخرى، تستخدم واجهة برمجة تطبيقات Anthropic Messages كائنات قاموس (Dictionary) محللة وكتلة tool_result داخل رسالة user. نظرًا لاختلاف المواصفات، فإن إرسال بيانات 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]
هذه المحادثة هي عمل مستمر تمت ترقيلته من الجلسة السابقة (المعرف: sess_99812).
تتضمن تاريخ الرسائل نتائج تنفيذ الأدوات وسياق المحادثة مع النموذج السابق.
استمر في العمل من النقطة التي توقفت عندها بناءً على نتائج استدعاء الأدوات السابقة المقدمة (tool_result).
`
لا يهم إذا تعطل خادم موفر معين أو قام بتغيير سياسة الخدمة الخاصة به فجأة. يكفي الانتقال إلى نموذج لغات كبير آخر مع الحفاظ على تدفق المحادثة بنسبة 100% باستخدام بضعة أسطر من كود التحويل. عندما تحتفظ ببيانات الجلسة ماديًا في دليلك الخاص، يتم أخيرًا ضمان استمرارية الأعمال للمطور الفرد.