How to Build an AI Agent
How to Build an AI Agent
s11

Автономные агенты

Коллаборация

Scan Board, Claim Tasks

499 LOC14 инструментовTask board polling + timeout-based self-governance
Teammates scan the board and claim tasks themselves; no need for the lead to assign each one

s01 > s02 > s03 > s04 > s05 > s06 | s07 > s08 > s09 > s10 > [ s11 ] s12

"Со-работники сканируют доску и сами забирают задачи" -- не нужно, чтобы lead назначал каждого.

Проблема

В s09-s10 со-работники работают только когда явно сказано. Lead должен генерировать каждого со специфичным промптом. 10 невостребованных задач на доске? Lead назначает каждую вручную. Это не масштабируется.

Настоящая автономия: со-работники сами сканируют доску задач, забирают невостребованные задачи, работают над ними, затем ищут ещё.

После сжатия контекста (s06) агент может забыть, кто он. Повторное внедрение идентичности исправляет это.

Решение

Жизненный цикл со-работника с циклом idle:

+-------+
| spawn |
+---+---+
    |
    v
+-------+   tool_use     +-------+
| WORK  | <------------- |  LLM  |
+-------+                +-------+
    |
    | stop_reason != tool_use (or idle tool called)
    v
+--------+
|  IDLE  |  poll every 5s for up to 60s
+---+----+
    |
    +---> check inbox --> message? ----------> WORK
    |
    +---> scan .tasks/ --> unclaimed? -------> claim -> WORK
    |
    +---> 60s timeout ----------------------> SHUTDOWN

Повторное внедрение идентичности после сжатия:
  if len(messages) <= 3:
    messages.insert(0, identity_block)

Как Это Работает

  1. Цикл со-работника имеет две фазы: WORK и IDLE. Когда LLM перестаёт вызывать инструменты (или вызывает idle), со-работник входит в IDLE.
def _loop(self, name, role, prompt):
    while True:
        # -- WORK PHASE --
        messages = [{"role": "user", "content": prompt}]
        for _ in range(50):
            response = client.messages.create(...)
            if response.stop_reason != "tool_use":
                break
            # execute tools...
            if idle_requested:
                break

        # -- IDLE PHASE --
        self._set_status(name, "idle")
        resume = self._idle_poll(name, messages)
        if not resume:
            self._set_status(name, "shutdown")
            return
        self._set_status(name, "working")
  1. Фаза idle опрашивает почтовый ящик и доску задач в цикле.
def _idle_poll(self, name, messages):
    for _ in range(IDLE_TIMEOUT // POLL_INTERVAL):  # 60s / 5s = 12
        time.sleep(POLL_INTERVAL)
        inbox = BUS.read_inbox(name)
        if inbox:
            messages.append({"role": "user",
                "content": f"<inbox>{inbox}</inbox>"})
            return True
        unclaimed = scan_unclaimed_tasks()
        if unclaimed:
            claim_task(unclaimed[0]["id"], name)
            messages.append({"role": "user",
                "content": f"<auto-claimed>Task #{unclaimed[0]['id']}: "
                           f"{unclaimed[0]['subject']}</auto-claimed>"})
            return True
    return False  # timeout -> shutdown
  1. Сканирование доски задач: найти задачи pending, без owner, не заблокированные.
def scan_unclaimed_tasks() -> list:
    unclaimed = []
    for f in sorted(TASKS_DIR.glob("task_*.json")):
        task = json.loads(f.read_text())
        if (task.get("status") == "pending"
                and not task.get("owner")
                and not task.get("blockedBy")):
            unclaimed.append(task)
    return unclaimed
  1. Повторное внедрение идентичности: когда контекст слишком короткий (пришло сжатие), вставить блок идентичности.
if len(messages) <= 3:
    messages.insert(0, {"role": "user",
        "content": f"<identity>You are '{name}', role: {role}, "
                   f"team: {team_name}. Continue your work.</identity>"})
    messages.insert(1, {"role": "assistant",
        "content": f"I am {name}. Continuing."})

Что Изменилось с s10

КомпонентДо (s10)После (s11)
Инструменты1214 (+idle, +claim_task)
АвтономияУправляется leadСамоорганизация
Фаза idleNoneОпрос почтового ящика + доски задач
За claim задачиТолько вручнуюАвтоматически забирать невостребованные
ИдентичностьСистемный промпт+ повторное внедрение после сжатия
TimeoutNone60s idle -> автоматическая остановка

Попробуйте

python agents/s11_autonomous_agents.py
  1. Создать 3 задачи на доске, затем создать alice и bob. Наблюдать, как они сами забирают.
  2. Создать товарища-кодера и позволить ему найти работу на доске задач
  3. Создать задачи с зависимостями. Наблюдать, как со-работники уважают порядок блокировки.
  4. Введите /tasks, чтобы увидеть доску задач с владельцами
  5. Введите /team, чтобы следить за тем, кто работает vs idle