11. סוכנים אוטונומיים

"חברי צוות סורקים את הלוח ותופסים משימות בעצמם"

30 דקות קריאה
💡חדש בנושא?

מה זה תפיסת משימה?

כשחבר צוות משנה אטומית את סטטוס המשימה מ-'pending' ל-'in_progress' וכותב את שמו עצמו כמוקצה. מכיוון שכל משימה היא קובץ JSON נפרד, שני סוכנים לא יכולים לתפוס את אותה משימה בו-זמנית — אחד ינצח, השני ינסה שוב.

מה זה הזרקת זהות מחדש?

לאחר שדחיסת context (סשן דחיסת הקשר) מוחקת את היסטוריית השיחה, הסוכן עשוי לשכוח מי הוא. ה-harness מזריק בלוק <identity> בתחילת ה-context הדחוס: 'אתה Alice, מומחית backend, עובדת על משימה 3.' זה משחזר את תחושת הזהות של הסוכן.

מה זה מחזור סרק?

כשחבר צוות לא מוצא משימות מוכנות, הוא ממתין בקצרה (ישן 2 שניות) ואז בודק שוב. לולאת הסקירה הזו היא המנגנון שמאפשר לחברי צוות לעבוד ללא הגבלת זמן מבלי שהמוביל יקצה כל משימה בנפרד.

הבעיה

בסשנים צוותי הסוכנים ופרוטוקולי הצוות, חברי הצוות עובדים רק כשנאמר להם במפורש. המוביל חייב לשגר כל אחד עם פרומפט ספציפי. 10 משימות לא תפוסות בלוח? המוביל מקצה כל אחת ידנית. לא מתרחב.

אוטונומיה אמיתית: חברי הצוות סורקים את לוח המשימות בעצמם, תופסים משימות לא תפוסות, עובדים עליהן, ואז מחפשים עוד.

פרט עדין אחד: לאחר דחיסת context (סשן דחיסת הקשר), הסוכן עשוי לשכוח מי הוא. הזרקת זהות מחדש פותרת זאת.

הפתרון

Teammate lifecycle with idle cycle:

+-------+
| spawn |
+---+---+
    |
    v
+-------+   tool_use     +-------+
| WORK  | <------------- |  LLM  |
+---+---+                +-------+
    |                       ^
    | done or no tasks      |
    v                       |
+-------+   poll tasks      |
| IDLE  | --check_ready()---+
+-------+   (wait 2s)
    |
    | shutdown_req received
    v
+----------+
| SHUTDOWN |
+----------+

איך זה עובד

  1. תפיסת משימה היא אטומית — כותבים בעלים + סטטוס יחד.
def claim_task(name: str) -> dict | None:
    """Find and claim the first ready task. Returns the task or None."""
    tasks_dir = Path(".tasks")
    ready = get_ready_tasks()  # reads all task files
    if not ready:
        return None

    task = ready[0]
    task_path = tasks_dir / f"task_{task['id']}.json"

    # Atomic claim: read current state, update only if still pending
    current = json.loads(task_path.read_text())
    if current["status"] != "pending":
        return None  # someone else claimed it

    current["status"] = "in_progress"
    current["assignee"] = name
    task_path.write_text(json.dumps(current, indent=2))
    return current
  1. לולאת חבר הצוות סוקרת משימות ועובדת דרכן.
def autonomous_teammate(name: str, role: str) -> None:
    system = build_system_with_identity(name, role)

    while True:
        # Check inbox for shutdown requests
        process_inbox(name)
        if get_status(name) == "SHUTDOWN":
            break

        task = claim_task(name)
        if task is None:
            # No ready tasks — idle cycle
            update_status(name, "IDLE")
            time.sleep(2)
            continue

        update_status(name, "WORKING")
        history = [{
            "role": "user",
            "content": f"Work on task {task['id']}: {task['title']}\n{task.get('description','')}"
        }]
        run_agent_with_identity(history, system, name, task["id"])
        complete_task(task["id"])
  1. הזרקת זהות מחדש עוטפת את פונקציית ה-hard_compact מסשן דחיסת הקשר.
def build_identity_block(name: str, role: str, task_id: int) -> str:
    return (
        f"<identity>\n"
        f"You are {name}, a {role}.\n"
        f"You are currently working on task {task_id}.\n"
        f"After completing it, claim another ready task from .tasks/.\n"
        f"</identity>"
    )

def hard_compact_with_identity(messages: list, name: str, role: str, task_id: int) -> list:
    # Regular hard compact...
    compacted = hard_compact(messages)
    # Prepend identity to the summary
    identity = build_identity_block(name, role, task_id)
    first_msg = compacted[0]
    first_msg["content"] = identity + "\n\n" + first_msg["content"]
    return compacted

מה השתנה מ-פרוטוקולי צוות

רכיבלפני (פרוטוקולי צוות)אחרי (סוכנים אוטונומיים)
הקצאת משימותהמוביל שולח במפורשחברי הצוות תופסים אוטונומית
מצב סרקהמתנה להודעהסקירת משימות מוכנות כל 2 שניות
זהותטיפול מיוחד איןהוזרקת מחדש לאחר דחיסה
תפקיד המובילמתאם + מקצהמתאם בלבד (תצוגת רמת לוח)

מסקנה מרכזית

תפיסת משימות אוטונומית היא מה שמבדיל צוות ממרוץ שליחים. כל חבר צוות הוא סוכן מכוון עצמי שסורק את הלוח, תופס מה שהוא יכול לעשות, עובד, ומבצע לולאה. תפקיד המוביל מצטמצם ליצירת משימות וטיפול בפרוטוקולים. הזרקת הזהות מחדש היא הדבק שמשמר את הסוכן קוהרנטי לאורך גבולות הדחיסה.

מדריך קוד אינטראקטיבי

מחזור סרק ותפיסת משימות אוטונומית
1def claim_task(name: str) -> dict | None:
2 ready = get_ready_tasks()
3 if not ready:
4 return None
5 task = ready[0]
6 task_path = Path(".tasks") / f"task_{task['id']}.json"
7 current = json.loads(task_path.read_text())
8 if current["status"] != "pending":
9 return None # someone else claimed it
10 current["status"] = "in_progress"
11 current["assignee"] = name
12 task_path.write_text(json.dumps(current, indent=2))
13 return current
14 
15def autonomous_teammate(name: str, role: str) -> None:
16 system = build_system_with_identity(name, role)
17 while True:
18 process_inbox(name)
19 if get_status(name) == "SHUTDOWN":
20 break
21 task = claim_task(name)
22 if task is None:
23 update_status(name, "IDLE")
24 time.sleep(2)
25 continue
26 update_status(name, "WORKING")
27 history = [{"role": "user", "content":
28 f"Work on task {task['id']}: {task['title']}"}]
29 run_agent_with_identity(history, system, name, task["id"])
30 complete_task(task["id"])
31 
claim_task() מוצאת תחילה את כל המשימות המוכנות. אם אין, היא מחזירה None מיד — הקורא יכניס את הסוכן למצב סרק.
שלב 1 מתוך 5
🧪 נסו בעצמכם

הגדירו 3 סוכנים אוטונומיים עם לוח משימות ריק, ואז הוסיפו 5 משימות. צפו בהם תופסים ומשלימים.

רמז

ה-idle_cycle של כל סוכן יסרוק ויתפוס אוטומטית

מצאתם טעות? דווחו ←