⚡ KHÓA HỌC HARNESS ENGINEERING & AI CODING AGENT (CLAUDE CODE) BÀI 4 / 4 (HOÀN THÀNH)
Harness Engineering Series • Bài 4 (Cuối)

Khóa Học Harness Engineering (Phần 4): Đội Ngũ Agent Teams, Đồ Thị Nhiệm Vụ & Vòng Lặp Goal Loop

Ở các phần trước, chúng ta đã xây dựng một tác nhân đơn lẻ có khả năng suy luận, quản lý ngữ cảnh sạch bằng Subagents và cô lập nhánh lập trình với Git Worktrees. Tuy nhiên, khi đối mặt với một chiến dịch tái cấu trúc mã nguồn toàn diện (Codebase Refactoring) hoặc phát triển một hệ thống giao dịch tự động từ con số 0, một tác nhân đơn lẻ — dù thông minh đến đâu — cũng sẽ gặp giới hạn về năng lực xử lý tuần tự.

Đỉnh cao của kiến trúc tác nhân hiện đại nằm ở Sự Phối Hợp Đa Tác Nhân (Multi-Agent Systems). Bài viết cuối cùng của chuỗi Harness Engineering sẽ vén màn 3 cơ chế tối tân nhất của Claude Code và shareAI-lab/learn-claude-code: Đội ngũ tác nhân chuyên trách (Agent Teams), Đồ thị phụ thuộc nhiệm vụ (Task Graph với blockedBy) và Vòng lặp mục tiêu tự phục hồi (Goal Loop).

1. Đội Ngũ Tác Nhân Bất Đồng Bộ (Agent Teams & Async Mailbox)

Mô hình cộng tác đa tác nhân truyền thống thường mắc lỗi đồng bộ (synchronous blocking): Agent A phải đợi Agent B trả lời rồi mới được chạy tiếp, gây lãng phí tài nguyên và dễ rơi vào bế tắc (deadlock). Claude Code giải quyết triệt để vấn đề này bằng mô hình Async Mailbox (Hộp Thư Bất Đồng Bộ):

2. Hệ Thống Nhiệm Vụ Với Đồ Thị Phụ Thuộc (Task Graph & blockedBy)

Làm thế nào để 3–5 tác nhân cùng làm việc trên 20 đầu việc mà không xảy ra tình trạng "giẫm chân lên nhau" (tranh giành làm cùng 1 việc) hoặc làm sai thứ tự logic?

Claude Code sử dụng cấu trúc dữ liệu TaskRecord lưu trữ trực tiếp trên đĩa cứng (Disk Persistence tại ~/.claude/tasks/):

{ "id": "task-03", "title": "Viết Unit Test cho Module Xác Thực JWT", "status": "pending", # pending | running | completed | failed "owner": null, # Tác nhân nào đang nhận việc (Locking) "blockedBy": ["task-01"], # Chỉ được nhận khi task-01 đã hoàn thành "deliverables": ["tests/test_jwt.py"] }

Các quy tắc vận hành của Task Graph:

  1. Atomic Task Claims (Khóa việc nguyên tử): Khi một Agent quyết định nhận task-03, hệ thống kiểm tra điều kiện nguyên tử: Nếu task-03.blockedBy chưa xong hoặc đã có Agent khác claim, lệnh nhận việc lập tức bị từ chối.
  2. Tự động mở khóa (Unblocking Trigger): Ngay khi task-01 chuyển trạng thái sang completed, hệ thống lập tức thông báo tới các Agent đang rảnh rỗi để bắt đầu tranh việc task-03.

3. Vòng Lặp Mục Tiêu (Goal Loop) & Tự Động Hóa Qua Đêm

Một trong những nỗi thất vọng lớn nhất của lập trình viên khi dùng AI là bot dừng lại nửa chừng: "Tôi đã sửa xong file này, mời bạn kiểm tra và gõ lệnh tiếp theo."

Với Goal Loop (`/goal <condition>`), tác nhân được đặt trong một trạng thái tự trị tuyệt đối:

Cơ Chế 1: Cổng Đánh Giá
Evaluation Gate
Mỗi khi Agent muốn kết thúc câu trả lời, Harness sẽ tự động kích hoạt Cổng Đánh Giá: Chạy lại toàn bộ test suite và linter để kiểm chứng kết quả thực tế.
Cơ Chế 2: Tự Động Tiếp Diễn
Automatic Continuation
Nếu các bài test vẫn báo lỗi hoặc tiêu chuẩn nghiệm thu chưa đạt 100%, Harness sẽ tự động nạp stack trace lỗi vào prompt và yêu cầu Agent tự sửa tiếp mà không cần con người can thiệp.
Cơ Chế 3: Lập Lịch Định Kỳ
Scheduled Cron Loops
Cho phép Agent thức dậy định kỳ mỗi 15 phút hoặc vào 2 giờ sáng (`/schedule`) để kiểm tra log hệ thống, cập nhật dữ liệu và gửi báo cáo tóm tắt.

4. Sơ Đồ Kiến Trúc Luồng Phối Hợp Agent Teams & Task Graph

Dưới đây là sơ đồ kiến trúc thể hiện toàn bộ vòng lặp phối hợp giữa Điều Phối Viên (Coordinator), các Tác Nhân Chuyên Trách (Worker Agents) thông qua Async Mailbox và Task Graph:

KIẾN TRÚC PHỐI HỢP AGENT TEAMS & TASK DEPENDENCY GRAPH Hộp thư bất đồng bộ (Async Mailbox) · Khóa nhiệm vụ nguyên tử · Vòng lặp nghiệm thu TASK DEPENDENCY GRAPH Task 1: Tạo Data Schema Status: COMPLETED ✓ Task 2: API Endpoints blockedBy: [Task 1] (Đã mở) Owner: Coder Agent 1 Task 3: Unit Tests E2E blockedBy: [Task 2] (Đang khóa) Chờ Task 2 hoàn thành Đồ thị nhiệm vụ hướng mục tiêu Claims ĐỘI NGŨ AGENT TEAMS Team Lead (Orchestrator) Phân bổ nhiệm vụ · Đánh giá goal Điều phối luồng công việc Worker Agent 1 (Coder) Đang viết code Task 2 trong Worktree Tự do gọi Bash / Edit tool Worker Agent 2 (Tester) Đang đợi thư hoàn thành Task 2 Sẵn sàng chạy suite kiểm thử Vận hành đồng thời đa luồng Verify GOAL LOOP & GATE Cổng Kiểm Định Test Chạy pytest & npm test Nếu Fail: Đẩy lại cho Coder Async Mailbox Queue Trao đổi thông điệp JSON Không bao giờ bị block luồng Nghiệm Thu Hoàn Thành Mọi bài test pass 100% Dừng vòng lặp / Bàn giao Tự trị xử lý trọn vẹn mục tiêu

5. Triển Khai Code: Bộ Điều Phối TaskGraph Với Cơ Chế blockedBy

Dưới đây là mã nguồn Python triển khai hệ thống quản trị đồ thị nhiệm vụ cho phép phân phối việc tự động và khóa nguyên tử cho nhiều Agent:

from typing import Dict, List, Optional class TaskRecord: def __init__(self, task_id: str, title: str, blocked_by: List[str] = None): self.task_id = task_id self.title = title self.blocked_by = blocked_by or [] self.status = "pending" # pending | running | completed | failed self.owner = None class TaskGraphEngine: """ Hệ thống Task Graph hỗ trợ khóa nhiệm vụ nguyên tử (Atomic Task Claims) """ def __init__(self): self.tasks: Dict[str, TaskRecord] = {} def add_task(self, task_id: str, title: str, blocked_by: List[str] = None): self.tasks[task_id] = TaskRecord(task_id, title, blocked_by) def claim_next_available_task(self, agent_name: str) -> Optional[TaskRecord]: """Tìm và khóa nhiệm vụ khả dụng đầu tiên cho Agent rảnh rỗi""" for t in self.tasks.values(): if t.status == "pending" and t.owner is None: # Kiểm tra xem toàn bộ các task chặn (blockedBy) đã hoàn thành chưa all_dependencies_met = all( self.tasks[dep_id].status == "completed" for dep_id in t.blocked_by if dep_id in self.tasks ) if all_dependencies_met: t.status = "running" t.owner = agent_name print(f"[TASK GRAPH] {agent_name} đã nhận thành công: [{t.task_id}] {t.title}") return t return None def complete_task(self, task_id: str): """Đánh dấu hoàn thành và tự động mở khóa các nhiệm vụ tiếp theo""" if task_id in self.tasks: self.tasks[task_id].status = "completed" print(f"[TASK GRAPH] Đã hoàn tất [{task_id}]. Đang mở khóa các đầu việc phụ thuộc...") def all_tasks_completed(self) -> bool: return all(t.status == "completed" for t in self.tasks.values())
🏆
CHÚC MỪNG BẠN ĐÃ HOÀN THÀNH KHÓA HỌC HARNESS ENGINEERING (CLAUDE CODE): Bạn đã làm chủ toàn bộ tư duy kỹ thuật nền tảng: từ bản chất của vòng lặp Agent Loop tối giản, nghệ thuật quản trị ngữ cảnh với Subagents và Skills động, quy trình SDLC chuẩn hóa với Git Worktrees cho đến cơ chế tự trị đa tác nhân Agent Teams và Goal Loop. Hãy áp dụng bộ khung này để xây dựng các trợ lý AI lập trình và định lượng vượt trội!
⚡

Biên Soạn: Học Viện Giao Dịch — Hàng Hóa Plus

Chuỗi bài giảng Harness Engineering được phát triển dựa trên các nghiên cứu chuyên sâu về kiến trúc tác nhân lập trình AI của Anthropic Claude Code và shareAI-lab.