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

1. איפה מוצאים סוכנים מנוהלים

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

סוכן מנוהל נותן לכם את השירות הבא:

1. קוד שלכם יכול ליצור מכונה וירטואלית חדשה בענן או להפעיל אחת שמורה.

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

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

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

2. בואו ניצור סוכן מנוהל כדי לראות איך זה עובד

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

/claude-api managed-agents-onboard

נפעיל אותה מתוך קלוד קוד ונקבל מדריך התחלה מהירה עם המנגנון. קלוד עונה לי את התשובה הראשונה:

❯ /claude-api managed-agents-onboard

  Listed 1 directory

⏺ Managed Agents is hosted. Anthropic runs the agent loop, the sandbox and the infrastructure, and you just define the
  agent. We'll set it up in three steps: the agent, the environment it runs in, and then a live test session.

  To start, describe the agent you want:

  - What should it do? For example, "triage new Linear issues and draft fixes as PRs" or "research competitors and
    produce a weekly xlsx summary".
  - What starts it? A person chatting with it, an event from your app or a webhook, or a schedule such as every night.

  Mention any services it has to work with, like GitHub, Slack or your own API, and anything it should produce, like a
  report, a PR or files. I'll use that to write a draft config for you to react to.

✻ Sautéed for 9s · done 18:21

בשביל הדוגמה אני מבקש סוכן שמנהל רשימת קניות. נוכל לבקש ממנו להוסיף פריטים לרשימה או להציג את הרשימה (נכון אין אופציה למחוק מהרשימה. אנחנו רק בדוגמה בינתיים). אני כותב לקלוד:

create a shopping list agent.
  the agent handles "add item" and "show items"
  users should get to it via a python web app (fastapi)
  and create a simple HTML/JS/CSS UI for it

קלוד כתב הרבה על המסך אבל עדיין לא יצר שום קובץ בתיקייה והוא ממשיך עם שאלות:

1. מי המודל?
2. צריך איזה פרומפט מערכת מיוחד?
3. איזה כלים?
4. איפה הרשימה תישמר?
5. מאיפה לוקחים מפתח API?
6. האם כל המשתמשים מקבלים את אותה רשימה?

אני עונה שימשיך עם ברירות המחדל, שאת המפתח אכתוב ישירות לקובץ .env בהמשך ושלכולם יש את אותה רשימה:

continue with the defaults; everyone has the same list; I'll put the API key in a .env file after you finish

סוף סוף קלוד התחיל לכתוב קוד. התוצאה היא עץ הקבצים הבא:

.
├── agents
│   └── shopping-list.md
├── app
│   ├── __pycache__
│   │   └── main.cpython-314.pyc
│   └── main.py
├── environments
│   └── shopping-list.yaml
├── pyproject.toml
├── README.md
├── static
│   ├── app.js
│   ├── index.html
│   └── style.css
└── uv.lock

אני מתחיל עם main.py וכבר שמח לראות כמה דברים שקלוד עשה נכון, בלי שהייתי צריך להגיד מילה:

1. השתמש ב uv לנהל את הפרויקט

2. השתמש ב Type Hints בכל הפונקציות, לדוגמה:

def add_item(name: str, quantity: int = 1) -> dict:

def get_items() -> list[dict]:

יש גם בדיקת פרמטרים שמוודאת ששם הפריט נמצא ושהכמות גדולה מאחד ואפילו טיפול במקביליות עם threading.Lock.

בצד השלילי שמתי לב שבדרך כלל קוד fastapi יהיה אסינכרוני. כאן יש לנו קוד סינכרוני שמשתמש ב thread pool ובגלל זה הנעילה. החלפה מקוד סינכרוני לקוד אסינכרוני דורשת שינוי משמעותי בקובץ. זו אחת הבעיות בעבודה עם Coding Agent שמצד אחד אנחנו מקבלים קוד משמעותי עובד אבל מצד שני לפעמים מדלגים על החלטות חשובות. ברור שאפשר לחזור אחורה ולתקן את הפרומפט או לבקש מהסוכן עכשיו לעשות את השינוי אבל אף אחד מהאופציות לא מלהיבה.

הרשימה עצמה נשמרת בקובץ שזו גם החלטה סבירה בהתחשב בנסיבות.

הקוד הכי חשוב ב main.py הוא הפונקציה run_turn שמפעילה את הסוכן המנוהל:

def run_turn(session_id: str, text: str) -> list[str]:
    """Send one user message and drive the session until the agent's turn ends."""
    replies: list[str] = []
    # Stream-first: open the stream before sending so no early events are missed.
    with client.beta.sessions.events.stream(session_id=session_id) as stream:
        client.beta.sessions.events.send(
            session_id=session_id,
            events=[{"type": "user.message", "content": [{"type": "text", "text": text}]}],
        )
        for event in stream:
            if event.type == "agent.message":
                replies.extend(b.text for b in event.content if b.type == "text")
            elif event.type == "agent.custom_tool_use":
                result, is_error = run_custom_tool(event.name, event.input or {})
                client.beta.sessions.events.send(
                    session_id=session_id,
                    events=[
                        {
                            "type": "user.custom_tool_result",
                            "custom_tool_use_id": event.id,
                            "content": [{"type": "text", "text": result}],
                            "is_error": is_error,
                        }
                    ],
                )
            elif event.type == "session.status_terminated":
                break
            elif event.type == "session.status_idle":
                # requires_action = waiting on a tool result we are sending; keep going.
                if event.stop_reason.type == "requires_action":
                    continue
                if event.stop_reason.type == "retries_exhausted":
                    raise HTTPException(502, "The agent failed to finish this turn.")
                break
    return replies

3. הפעלת הסוכן

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

    if not session_id:
        session = client.beta.sessions.create(
            agent=AGENT_ID, environment_id=ENVIRONMENT_ID, title="Shopping list chat"
        )
        session_id = session.id
        print(f"New session: https://platform.claude.com/workspaces/default/sessions/{session_id}")

לפני שאני יכול להפעיל את התוכנית עליי ליצור את הסביבה והסוכן במערכות של אנתרופיק ולשם כך יש תוכנית שורת פקודה של אנתרופיק בשם ant. קלוד קוד הדפיס לי את ההוראות ואני מפעיל משורת הפקודה:

ant apply agents/shopping-list.md environments/shopping-list.yaml

זה שומר את מזהה הסוכן ומזהה הסביבה לקובץ בשם claude-lock.json ובשביל להפעיל אני צריך לקחת משם את המזהים ולכתוב אותם יחד עם מפתח אנתרופיק לקובץ .env.

4. מה קיבלנו

בסופו של דבר הצלחנו להפעיל את השרת וקיבלנו מערכת סוכנים שיודעת לענות למספר משתמשים במקביל:

1. המערכת מנהלת רשימת קניות בקובץ JSON מקומי.
2. כל הודעה נשלחת לסוכן שרץ על סוכן במכונה וירטואלית בענן של אנתרופיק.
3. הסוכן מבין מתוך ההודעה איזה פריטים להוסיף לרשימה או אם צריך להציג את הפריטים ברשימה ומפעיל את הפונקציות המתאימות בפייתון שנכתבו ב main.py.
4. אנתרופיק מבטיחה שכל סוכן רץ בסביבה מבודדת ומקבל את המשאבים להגיב מהר.

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

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