Codebook Agent 2026: Arsitektur Amortized Topology Design, VQ-Codebook Compression, dan Reranking Tanpa Pesan untuk Sistem Multi-Agent LLM

1. Tiga Asumsi Palsu pada Generative Multi-Agent Topology Designers

Dalam kurun 2024 hingga 2026, orkestrator multi-agent seperti GPTSwarm, G-Designer, ARG-Designer, dan GTD mengusung filosofi bahwa setiap prompt pengguna yang berbeda memerlukan topologi komunikasi kustom yang digenerasikan secara kontinu di atas ruang matriks ketetanggaan N × N. Alur standarnya selalu seragam: decoder generatif (berbasis VAE, Autoregressive, atau Denoising Diffusion) melakukan pencarian acak, lalu Graph Neural Network (GNN) melakukan message passing pada profil agen untuk menilai kandidat mana yang memiliki edge count terendah dan utilitas tertinggi.

Eksperimen ketat tim UCLA membuktikan bahwa resep ini salah kaprah karena bertumpu pada premis fiktif:

2. Paradoks Konsumsi Token: Mengapa Graf Jarang Justru Mencekik Anggaran

Salah satu kontribusi analitis terpenting dari makalah ini adalah pembedahan paradoks biaya token. Perhatikan perbandingan dinamika berikut:

Dengan menggunakan normalisasi biaya tugas c_bar_i = c_i / mean(c_task), kesulitan intrinsik dari soal dihilangkan dari evaluasi, sehingga perankingan murni mencerminkan efisiensi arsitektur komunikasi tanpa bias dari kompleksitas prompt.

3. Arsitektur Tiga Komponen: VQ-VAE, Reward Prior, dan Flattened Adjacency Proxy

Alih-alih merancang generator generatif lain yang lambat, tim UCLA mengamortisasi perancangan topologi ke dalam tiga komponen linier terpisah:

4. Tolok Ukur Komparatif 6 Benchmark: GSM8K, MATH, SVAMP, MBPP, HumanEval

Evaluasi komprehensif dijalankan pada 6 benchmark penalaran matematika dan sintesis kode menggunakan backend gpt-4o-mini dengan embedding beku all-MiniLM-L6-v2. Hasilnya menetapkan standar baru di ranah orkestrasi multi-agent:

Tabel di atas mengungkap kenyataan penting: Codebook Agent mengungguli pendekatan difusi graf terkuat (GTD) sebesar +1.60 poin akurasi agregat sambil memangkas latensi penentuan arsitektur dari 385.6 ms menjadi hanya 2.4 ms (percepatan 160x lipat). Lebih jauh lagi, konsumsi token LLM terpangkas antara 21.9% hingga 33.2% di seluruh skenario.

5. Implementasi Referensi PyTorch: Mesin Inferensi Satu Siklus (2.4 ms)

Berikut implementasi arsitektur produksi PyTorch murni untuk mengintegrasikan Vector Quantizer, Reward Predictor, dan Flattened Adjacency Scorer:

"""
Codebook Agent Production Architecture: Amortized Topology Design Engine
Berdasarkan Inovasi arXiv:2609.02264 (Yu et al., UCLA - September 2026).
Kompresi Discrete VQ-VAE, Reward-Weighted Prior, dan One-Pass Flattened Adjacency Proxy.
"""

import torch
import torch.nn as nn
import torch.nn.functional as F
from typing import Tuple, List, Dict, Optional

class VectorQuantizer(nn.Module):
    """
    Vector Quantizer diskrit untuk mengindeks graf komunikasi sukses.
    Menggunakan Exponential Moving Average (EMA) dictionary updates & straight-through estimator.
    """
    def __init__(self, num_embeddings: int = 16, embedding_dim: int = 32, commitment_cost: float = 0.25, decay: float = 0.99):
        super().__init__()
        self.num_embeddings = num_embeddings
        self.embedding_dim = embedding_dim
        self.commitment_cost = commitment_cost
        self.decay = decay

        self.embedding = nn.Embedding(self.num_embeddings, self.embedding_dim)
        self.embedding.weight.data.normal_()

        self.register_buffer("ema_cluster_size", torch.zeros(num_embeddings))
        self.register_buffer("ema_w", self.embedding.weight.data.clone())

    def forward(self, inputs: torch.Tensor) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
        # inputs: [batch_size, embedding_dim]
        distances = (
            torch.sum(inputs ** 2, dim=1, keepdim=True)
            + torch.sum(self.embedding.weight ** 2, dim=1)
            - 2 * torch.matmul(inputs, self.embedding.weight.t())
        )

        encoding_indices = torch.argmin(distances, dim=1)
        quantized = self.embedding(encoding_indices)

        # Loss commitment
        e_latent_loss = F.mse_loss(quantized.detach(), inputs)
        loss = self.commitment_cost * e_latent_loss

        # Straight-through gradient estimator
        quantized = inputs + (quantized - inputs).detach()

        return quantized, loss, encoding_indices

class TopologyVQVAE(nn.Module):
    """
    Autoencoder yang memetakan matriks ketetanggaan (adjacency matrix)
    ke ruang diskrit 16-codebook tanpa kondisioning query.
    """
    def __init__(self, num_agents: int = 4, latent_dim: int = 32, num_codes: int = 16):
        super().__init__()
        self.num_agents = num_agents
        self.edge_dim = num_agents * (num_agents - 1)  # Off-diagonal directed edges
        self.latent_dim = latent_dim

        self.encoder = nn.Sequential(
            nn.Linear(self.edge_dim, 64),
            nn.LayerNorm(64),
            nn.ReLU(),
            nn.Linear(64, latent_dim),
        )
        self.vq = VectorQuantizer(num_embeddings=num_codes, embedding_dim=latent_dim)
        self.decoder = nn.Sequential(
            nn.Linear(latent_dim, 64),
            nn.LayerNorm(64),
            nn.ReLU(),
            nn.Linear(64, self.edge_dim),
        )

    def decode_code(self, code_idx: int) -> torch.Tensor:
        """Decode satu indeks kode menjadi matriks ketetanggaan biner [N, N]."""
        emb = self.vq.embedding(torch.tensor([code_idx], device=self.vq.embedding.weight.device))
        edge_logits = self.decoder(emb)
        binary_edges = (torch.sigmoid(edge_logits) > 0.5).float()

        # Rekonstruksi ke matriks N x N dengan diagonal nol
        adj = torch.zeros((self.num_agents, self.num_agents), device=edge_logits.device)
        idx = 0
        for i in range(self.num_agents):
            for j in range(self.num_agents):
                if i != j:
                    adj[i, j] = binary_edges[0, idx]
                    idx += 1
        return adj

class RewardWeightedCodePredictor(nn.Module):
    """
    Pemeta embedding query ke distribusi probabilitas atas 16 kode topologi.
    Memasukkan preferensi komputasi hemat sejak tahap prior.
    """
    def __init__(self, query_dim: int = 384, num_codes: int = 16):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(query_dim, 128),
            nn.ReLU(),
            nn.Dropout(0.1),
            nn.Linear(128, num_codes),
        )

    def forward(self, query_emb: torch.Tensor) -> torch.Tensor:
        return F.softmax(self.net(query_emb), dim=-1)

class ExecutionGroundedProxy(nn.Module):
    """
    Scorer cepat berbasis MLP atas representasi flattened adjacency.
    Memprediksi utilitas nyata dan normalized token cost tanpa message passing.
    """
    def __init__(self, num_agents: int = 4):
        super().__init__()
        edge_dim = num_agents * (num_agents - 1)
        self.shared_backbone = nn.Sequential(
            nn.Linear(edge_dim, 64),
            nn.ReLU(),
            nn.Linear(64, 32),
            nn.ReLU(),
        )
        self.utility_head = nn.Linear(32, 1)      # Prediksi akurasi / task success
        self.cost_head = nn.Linear(32, 1)         # Prediksi normalized token cost c_bar

    def forward(self, flattened_adjacencies: torch.Tensor) -> Tuple[torch.Tensor, torch.Tensor]:
        feat = self.shared_backbone(flattened_adjacencies)
        u_pred = torch.sigmoid(self.utility_head(feat))
        c_pred = F.softplus(self.cost_head(feat))
        return u_pred.squeeze(-1), c_pred.squeeze(-1)

class CodebookAgentOrchestrator:
    """
    Mesin inferensi satu-siklus (one-pass) berlatensi 2.4 ms.
    Menghilangkan loop difusi, autoregressive branching, dan message-passing overhead.
    """
    def __init__(
        self,
        vqvae: TopologyVQVAE,
        predictor: RewardWeightedCodePredictor,
        proxy: ExecutionGroundedProxy,
        beta_cost_weight: float = 0.2,
    ):
        self.vqvae = vqvae
        self.predictor = predictor
        self.proxy = proxy
        self.beta = beta_cost_weight

    @torch.no_grad()
    def select_topology(self, query_embedding: torch.Tensor, top_k_codes: int = 4) -> Dict[str, any]:
        # 1. Prediksi distribusi kode dari embedding query
        code_probs = self.predictor(query_embedding).squeeze(0)
        top_indices = torch.topk(code_probs, k=top_k_codes).indices.tolist()

        # 2. Decode kandidat topologi unik
        unique_adjacencies = []
        flattened_list = []
        for code in top_indices:
            adj = self.vqvae.decode_code(code)
            # Ambil off-diagonal flattened
            flat_edges = []
            for i in range(adj.shape[0]):
                for j in range(adj.shape[1]):
                    if i != j:
                        flat_edges.append(adj[i, j])
            flat_tensor = torch.tensor(flat_edges, device=query_embedding.device)
            unique_adjacencies.append(adj)
            flattened_list.append(flat_tensor)

        batch_flat = torch.stack(flattened_list, dim=0)

        # 3. Satu evaluasi batched proxy execution-grounded
        u_preds, c_preds = self.proxy(batch_flat)
        composite_rewards = u_preds - self.beta * c_preds

        best_idx = torch.argmax(composite_rewards).item()
        winner_adj = unique_adjacencies[best_idx]

        return {
            "selected_code": top_indices[best_idx],
            "adjacency_matrix": winner_adj.cpu().numpy().tolist(),
            "predicted_utility": round(u_preds[best_idx].item(), 4),
            "predicted_cost": round(c_preds[best_idx].item(), 4),
            "composite_reward": round(composite_rewards[best_idx].item(), 4),
            "inference_mode": "one_pass_amortized",
        }

6. Panduan Praktis AI Engineering untuk Infrastruktur Multi-Agent 2026