crewAI/docs/ar/concepts/checkpointing.mdx

---
title: Checkpointing
description: حفظ حالة التنفيذ تلقائيا حتى تتمكن الطواقم والتدفقات والوكلاء من الاستئناف بعد الفشل.
icon: floppy-disk
mode: "wide"
---

الـ Checkpointing يحفظ لقطة من حالة التنفيذ أثناء التشغيل بحيث يمكن لطاقم أو تدفق أو وكيل الاستئناف بعد الفشل أو التفرع إلى فرع بديل.

<CardGroup cols={2}>
  <Card title="الشرح" icon="lightbulb" href="#الشرح">
    كيف يعمل الـ Checkpointing: الأحداث والتخزين والوراثة.
  </Card>
  <Card title="درس تطبيقي" icon="graduation-cap" href="#درس-تطبيقي-استئناف-طاقم-فاشل">
    دليل 5 دقائق: تشغيل، إيقاف، استئناف.
  </Card>
  <Card title="ادلة عملية" icon="screwdriver-wrench" href="#ادلة-عملية">
    وصفات مركزة على المهام لسير العمل الشائع.
  </Card>
  <Card title="المرجع" icon="book" href="#المرجع">
    `CheckpointConfig` والأحداث والمزودات وسطر الأوامر.
  </Card>
</CardGroup>

## الشرح

### ما هي نقطة الحفظ

نقطة الحفظ هي لقطة متسلسلة من `RuntimeState` تكتب في نقطة معينة من التنفيذ. تسجل أي المهام اكتملت ومخرجاتها والمدخلات الحالية ومعرف نسب يحدد التشغيل.

عند الاستعادة من نقطة حفظ، يعيد CrewAI بناء تلك الحالة ويتخطى العمل المكتمل ويستمر. عند التفرع، يستعيد CrewAI الحالة تحت نسب جديد بحيث لا يتداخل الفرع الجديد مع التشغيل الأصلي.

### متى تكتب نقاط الحفظ

الـ Checkpointing مدفوع بالأحداث. يشترك وقت التشغيل في الأحداث التي تحددها عبر `on_events` ويكتب نقطة حفظ عند إطلاق أحدها. الافتراضي `task_completed` ينتج نقطة حفظ لكل مهمة منتهية — توازن معقول بين الدقة واستخدام القرص. الأحداث عالية التردد مثل `llm_call_completed` متاحة للاستعادة الدقيقة لكنها تكتب ملفات أكثر بكثير.

### التخزين

يتضمن CrewAI مزودين:

- `JsonProvider` يكتب ملفا لكل نقطة حفظ. قابل للقراءة وسهل التفقد.
- `SqliteProvider` يكتب إلى قاعدة بيانات SQLite واحدة. أفضل لنقاط الحفظ عالية التردد.

كلاهما يحذف أقدم نقاط الحفظ عند تحديد `max_checkpoints`.

<Note>
كتابة نقاط الحفظ بأفضل جهد. فشل نقطة حفظ يسجل لكنه لا يقاطع التشغيل.
</Note>

### نموذج الوراثة

`Crew` و`Flow` و`Agent` كلها تقبل وسيط `checkpoint`. يرث الأبناء من الأب ما لم يحددوا قيمتهم الخاصة أو يمرروا `False` للانسحاب. فعل الـ Checkpointing مرة واحدة على الطاقم وتشارك كل الوكلاء، أو استبعد وكيلا واحدا بشكل انتقائي.

## درس تطبيقي: استئناف طاقم فاشل

هذا الدليل يستغرق حوالي 5 دقائق. ستشغل طاقما بمهمتين، توقفه في المنتصف، ثم تستأنف من نقطة الحفظ المحفوظة.

<Steps>
  <Step title="أنشئ الطاقم مع تفعيل الـ Checkpointing">
    ```python
    from crewai import Agent, Crew, Task

    researcher = Agent(role="Researcher", goal="Research", backstory="Expert")
    writer = Agent(role="Writer", goal="Write", backstory="Expert")

    crew = Crew(
        agents=[researcher, writer],
        tasks=[
            Task(description="Research AI trends", agent=researcher, expected_output="bullets"),
            Task(description="Write a summary", agent=writer, expected_output="paragraph"),
        ],
        checkpoint=True,
    )
    ```
  </Step>
  <Step title="شغله وأوقفه بعد المهمة الأولى">
    ```python
    result = crew.kickoff()
    ```

    اضغط `Ctrl+C` بعد انتهاء المهمة الأولى. في `./.checkpoints/`، الملف بصيغة `<timestamp>_<uuid>.json` هو نقطة الحفظ.
  </Step>
  <Step title="استأنف من نقطة الحفظ">
    ```python
    from crewai import CheckpointConfig

    result = crew.kickoff(
        from_checkpoint=CheckpointConfig(
            restore_from="./.checkpoints/<timestamp>_<uuid>.json",
        ),
    )
    ```

    يتم تخطي مهمة البحث، ويعمل الكاتب على مخرجات البحث المحفوظة، وينتهي الطاقم.
  </Step>
</Steps>

## ادلة عملية

<AccordionGroup>
  <Accordion title="تفعيل الـ Checkpointing بالإعدادات الافتراضية" icon="play">
    ```python
    crew = Crew(agents=[...], tasks=[...], checkpoint=True)
    ```

    يكتب إلى `./.checkpoints/` عند كل `task_completed`.
  </Accordion>

  <Accordion title="تخصيص التخزين والتردد" icon="sliders">
    ```python
    from crewai import Crew, CheckpointConfig

    crew = Crew(
        agents=[...],
        tasks=[...],
        checkpoint=CheckpointConfig(
            location="./my_checkpoints",
            on_events=["task_completed", "crew_kickoff_completed"],
            max_checkpoints=5,
        ),
    )
    ```
  </Accordion>

  <Accordion title="اختيار مزود التخزين" icon="database">
    <CodeGroup>
    ```python JsonProvider
    from crewai import Crew, CheckpointConfig
    from crewai.state import JsonProvider

    crew = Crew(
        agents=[...],
        tasks=[...],
        checkpoint=CheckpointConfig(
            location="./my_checkpoints",
            provider=JsonProvider(),
            max_checkpoints=5,
        ),
    )
    ```
    ```python SqliteProvider
    from crewai import Crew, CheckpointConfig
    from crewai.state import SqliteProvider

    crew = Crew(
        agents=[...],
        tasks=[...],
        checkpoint=CheckpointConfig(
            location="./.checkpoints.db",
            provider=SqliteProvider(),
            max_checkpoints=50,
        ),
    )
    ```
    </CodeGroup>

    <Tip>
    SQLite يفعل وضع journal WAL للقراءات المتزامنة. يفضل لنقاط الحفظ عالية التردد.
    </Tip>
  </Accordion>

  <Accordion title="استبعاد وكيل واحد" icon="user-slash">
    ```python
    crew = Crew(
        agents=[
            Agent(role="Researcher", ...),
            Agent(role="Writer", ..., checkpoint=False),
        ],
        tasks=[...],
        checkpoint=True,
    )
    ```
  </Accordion>

  <Accordion title="الاستئناف عبر classmethod" icon="rotate-left">
    ```python
    config = CheckpointConfig(restore_from="./my_checkpoints/<file>.json")
    crew = Crew.from_checkpoint(config)
    result = crew.kickoff()
    ```
  </Accordion>

  <Accordion title="التفرع إلى فرع جديد" icon="code-branch">
    `fork()` يستعيد نقطة حفظ تحت نسب جديد بحيث لا يتصادم التشغيل الجديد مع الأصلي.

    ```python
    config = CheckpointConfig(restore_from="./my_checkpoints/<file>.json")
    crew = Crew.fork(config, branch="experiment-a")
    result = crew.kickoff(inputs={"strategy": "aggressive"})
    ```

    تسمية `branch` اختيارية؛ يتم إنشاء واحدة إذا أغفلت.
  </Accordion>

  <Accordion title="Checkpointing لـ Crew أو Flow أو Agent" icon="cubes">
    <Tabs>
      <Tab title="Crew">
        ```python
        crew = Crew(
            agents=[researcher, writer],
            tasks=[research_task, write_task, review_task],
            checkpoint=CheckpointConfig(location="./crew_cp"),
        )
        ```

        المشغل الافتراضي: `task_completed`.
      </Tab>
      <Tab title="Flow">
        ```python
        from crewai.flow.flow import Flow, start, listen
        from crewai import CheckpointConfig

        class MyFlow(Flow):
            @start()
            def step_one(self):
                return "data"

            @listen(step_one)
            def step_two(self, data):
                return process(data)

        flow = MyFlow(
            checkpoint=CheckpointConfig(
                location="./flow_cp",
                on_events=["method_execution_finished"],
            ),
        )
        result = flow.kickoff()

        config = CheckpointConfig(restore_from="./flow_cp/<file>.json")
        flow = MyFlow.from_checkpoint(config)
        result = flow.kickoff()
        ```
      </Tab>
      <Tab title="Agent">
        ```python
        agent = Agent(
            role="Researcher",
            goal="Research topics",
            backstory="Expert researcher",
            checkpoint=CheckpointConfig(
                location="./agent_cp",
                on_events=["lite_agent_execution_completed"],
            ),
        )
        result = agent.kickoff(messages=[{"role": "user", "content": "Research AI trends"}])
        ```
      </Tab>
    </Tabs>
  </Accordion>

  <Accordion title="كتابة نقطة حفظ يدويا" icon="code">
    سجل معالجا على أي حدث واستدع `state.checkpoint()`.

    <CodeGroup>
    ```python Sync
    from crewai.events.event_bus import crewai_event_bus
    from crewai.events.types.llm_events import LLMCallCompletedEvent

    @crewai_event_bus.on(LLMCallCompletedEvent)
    def on_llm_done(source, event, state):
        path = state.checkpoint("./my_checkpoints")
        print(f"تم حفظ نقطة الحفظ: {path}")
    ```
    ```python Async
    from crewai.events.event_bus import crewai_event_bus
    from crewai.events.types.llm_events import LLMCallCompletedEvent

    @crewai_event_bus.on(LLMCallCompletedEvent)
    async def on_llm_done_async(source, event, state):
        path = await state.acheckpoint("./my_checkpoints")
        print(f"تم حفظ نقطة الحفظ: {path}")
    ```
    </CodeGroup>

    يتم تمرير وسيط `state` تلقائيا عندما يقبل المعالج ثلاثة معاملات. راجع [Event Listeners](/ar/concepts/event-listener) لقائمة الأحداث الكاملة.
  </Accordion>

  <Accordion title="التصفح والاستئناف والتفرع من سطر الأوامر" icon="terminal">
    ```bash
    crewai checkpoint                              # كشف تلقائي لـ .checkpoints/ أو .checkpoints.db
    crewai checkpoint --location ./my_checkpoints
    crewai checkpoint --location ./.checkpoints.db
    ```

    <Frame>
      <img src="/images/checkpointing.png" alt="Checkpoint TUI" />
    </Frame>

    اللوحة اليسرى تجمع نقاط الحفظ حسب الفرع؛ التفرعات تتداخل تحت أبيها. اختيار نقطة حفظ يعرض بياناتها الوصفية وحالة الكيان وتقدم المهام. **Resume** يكمل التشغيل؛ **Fork** يبدا فرعا جديدا.

    لوحة التفاصيل تعرض منطقتين قابلتين للتحرير:

    - **Inputs** — مدخلات الـ kickoff الأصلية، معبأة مسبقا وقابلة للتحرير.
    - **مخرجات المهام** — مخرجات المهام المكتملة. تحرير مخرج والضغط على **Fork** يبطل المهام التابعة لتعاد بالسياق المعدل.

    <Tip>
    مفيد لاستكشاف "ماذا لو": تفرع، عدل، راقب.
    </Tip>
  </Accordion>

  <Accordion title="تفقد نقاط الحفظ بدون TUI" icon="magnifying-glass">
    ```bash
    crewai checkpoint list ./my_checkpoints
    crewai checkpoint info ./my_checkpoints/<file>.json
    crewai checkpoint info ./.checkpoints.db
    ```
  </Accordion>
</AccordionGroup>

## المرجع

### `CheckpointConfig`

<ParamField path="location" type="str" default='"./.checkpoints"'>
  وجهة التخزين. مجلد لـ `JsonProvider`، مسار ملف قاعدة بيانات لـ `SqliteProvider`.
</ParamField>

<ParamField path="on_events" type="list[str]" default='["task_completed"]'>
  أنواع الأحداث التي تطلق نقطة حفظ. راجع [أنواع الأحداث](#أنواع-الأحداث).
</ParamField>

<ParamField path="provider" type="BaseProvider" default="JsonProvider()">
  واجهة التخزين. `JsonProvider` أو `SqliteProvider`.
</ParamField>

<ParamField path="max_checkpoints" type="int | None" default="None">
  الحد الاقصى لنقاط الحفظ المحتفظ بها. الأقدم تحذف بعد كل كتابة.
</ParamField>

<ParamField path="restore_from" type="Path | str | None" default="None">
  نقطة الحفظ المراد استعادتها عند تمريرها عبر `from_checkpoint`.
</ParamField>

### قيم حقل `checkpoint`

مقبولة في `Crew` و`Flow` و`Agent`.

<ParamField path="None" type="افتراضي">
  يرث من الأب.
</ParamField>

<ParamField path="True" type="bool">
  تفعيل بالإعدادات الافتراضية.
</ParamField>

<ParamField path="False" type="bool">
  انسحاب صريح. يوقف الوراثة.
</ParamField>

<ParamField path="CheckpointConfig(...)" type="CheckpointConfig">
  إعدادات مخصصة.
</ParamField>

### أنواع الأحداث

قيم شائعة لـ `on_events`:

| حالة الاستخدام | الأحداث |
|:---------------|:--------|
| بعد كل مهمة | `["task_completed"]` |
| بعد كل طريقة في التدفق | `["method_execution_finished"]` |
| بعد تنفيذ الوكيل | `["agent_execution_completed"]`, `["lite_agent_execution_completed"]` |
| عند اكتمال الطاقم فقط | `["crew_kickoff_completed"]` |
| بعد كل استدعاء LLM | `["llm_call_completed"]` |
| كل شيء | `["*"]` |

<Warning>
`["*"]` والأحداث عالية التردد مثل `llm_call_completed` تكتب نقاط حفظ كثيرة وقد تضر بالاداء. استخدمها مع `max_checkpoints`.
</Warning>

### مزودات التخزين

<ParamField path="JsonProvider" type="provider">
  ملف واحد لكل نقطة حفظ بصيغة `<timestamp>_<uuid>.json` داخل `location`.
</ParamField>

<ParamField path="SqliteProvider" type="provider">
  ملف قاعدة بيانات واحد في `location` مع journaling WAL.
</ParamField>

### سطر الأوامر

| الامر | الغرض |
|:------|:------|
| `crewai checkpoint` | تشغيل TUI؛ كشف التخزين تلقائيا. |
| `crewai checkpoint --location <path>` | تشغيل TUI على موقع محدد. |
| `crewai checkpoint list <path>` | سرد نقاط الحفظ. |
| `crewai checkpoint info <path>` | تفقد ملف نقطة حفظ أو آخر مدخل في قاعدة بيانات SQLite. |