Tutorial AutoGen v0.4: Build Multi-Agent AI Asinkron Lokal Pakai Ollama

Latar Belakang Arsitektur AutoGen v0.4 & Tantangan Scalability Multi-Agent

Microsoft merombak total arsitektur AutoGen pada rilis v0.4. Versi terdahulu (v0.2/v0.3) mengandalkan pola komputasi sekuensial sinkron berbasis kelas ConversableAgent. Pola tersebut menimbulkan bottleneck pada skala produksi: thread terblokir saat menunggu respons LLM, state terpusat sulit dikelola, dan koordinasi agen bersifat kaku.

AutoGen v0.4 beralih ke paradigma Actor Model dan Event-Driven Architecture (EDA) yang dibangun di atas paket autogen-core. Setiap agen berfungsi sebagai actor independen dengan alur eksekusi asinkron (asyncio). Agen berkomunikasi melalui pengiriman pesan (message passing) tanpa saling mengekspos internal state.

Perbandingan Arsitektur: AutoGen Legacy vs v0.4 Event-Driven

• Arsitektur Legacy (v0.2): Topologi pengolahan linier/sinkron. Eksekusi agen A memblokir agen B. Sulit menangani alur kerja paralel skala besar. State tersimpan dalam memori terpusat, meningkatkan risiko race condition saat skalabilitas ditingkatkan.
• Arsitektur v0.4 Core: Topologi pub/sub (Publish/Subscribe) dan point-to-point asinkron. Runtime terdistribusi, non-blocking I/O, dan pengolahan pesan berbasis event topic. Setiap agen memproses event queue secara mandiri.

Tantangan utama sistem multi-agent konvensional adalah scalability deadlock dan token inflation. Saat jumlah agen bertambah, lalu lintas pesan secara sinkron membuat eksekusi lambat dan biaya API melonjak. Integrasi AutoGen v0.4 dengan model lokal via Ollama menyelesaikan masalah biaya API sekaligus memaksimalkan throughput melalui komputasi asinkron lokal.

Penggunaan Actor Model memisahkan lapisan logika agen dari infrastruktur pengiriman pesan. Runtime bertindak sebagai event bus yang mengatur siklus hidup agen, serialisasi pesan, serta pendaftaran alur topik. Hasilnya, arsitektur v0.4 mampu menjalankan ratusan agen bersamaan dalam satu event loop tanpa mengorbankan konsistensi data.

Praktek Setup Agent Asinkron, Routing Messaging & Ollama Integration

Pengembangan sistem multi-agent asinkron membutuhkan komponen dasar berikut: Ollama sebagai inference server lokal, LiteLLM sebagai OpenAI-compatible proxy bridge, serta framework autogen-core dan autogen-ext.

1. Persiapan Environment dan Inference Server

Untuk menyiapkan lingkungan kerja dan server inferensi lokal, ikuti langkah-langkah berikut secara berurutan:

1. Jalankan service Ollama di latar belakang dan pastikan port 11434 dapat diakses.
2. Unduh model LLM lokal yang dioptimalkan untuk fungsi coding atau instruksi teknis dengan menjalankan perintah ollama run qwen2.5-coder:7b.
3. Buat virtual environment Python baru untuk mencegah konflik dependensi paket.
4. Install library AutoGen v0.4 dan dependensi pendukung menggunakan manajer paket pip.

# Pull model lokal melalui Ollama
ollama run qwen2.5-coder:7b

# Install library AutoGen v0.4 dan dependensi pendukung
pip install autogen-core autogen-ext litellm pydantic

2. Implementasi Multi-Agent Asinkron

Kode Python di bawah mengimplementasikan runtime asinkron dengan dua agen: PlannerAgent (pembuat tugas) dan CoderAgent (eksekutor kode). Kedua agen berkomunikasi via kanal pesan terstruktur.

import asyncio
from pydantic import BaseModel
from autogen_core import (
    AgentId,
    MessageContext,
    RoutedAgent,
    SingleThreadedAgentRuntime,
    TopicId,
    default_subscription,
    message_handler,
)
from autogen_ext.models.openai import OpenAIChatCompletionClient

# Schema pesan terstruktur
class TaskRequest(BaseModel):
    task_id: str
    instruction: str

class TaskResponse(BaseModel):
    task_id: str
    result: str

# Konfigurasi LLM Client menunjuk ke Ollama Endpoint (via OpenAI format API)
def get_ollama_client():
    return OpenAIChatCompletionClient(
        model="qwen2.5-coder:7b",
        api_key="ollama",
        base_url="http://localhost:11434/v1",
        model_info={
            "vision": False,
            "function_calling": True,
            "json_output": True,
            "family": "r1",
        },
    )

@default_subscription
class PlannerAgent(RoutedAgent):
    def __init__(self, model_client: OpenAIChatCompletionClient) -> None:
        super().__init__("PlannerAgent")
        self._model_client = model_client

    @message_handler
    async def handle_task(self, message: TaskRequest, ctx: MessageContext) -> None:
        print(f"[Planner] Menerima instruksi: {message.instruction}")
        prompt = f"Break down this task into execution steps: {message.instruction}"
        
        response = await self._model_client.create(
            messages=[{"role": "user", "content": prompt}]
        )
        
        plan = response.content
        print(f"[Planner] Rencana dibuat. Mengirim ke CoderAgent...")
        
        # Publish pesan ke topic agen eksekutor
        await self.publish_message(
            TaskResponse(task_id=message.task_id, result=plan),
            topic_id=TopicId("coder_topic", source=self.id.key),
        )

@default_subscription
class CoderAgent(RoutedAgent):
    def __init__(self, model_client: OpenAIChatCompletionClient) -> None:
        super().__init__("CoderAgent")
        self._model_client = model_client

    @message_handler
    async def handle_plan(self, message: TaskResponse, ctx: MessageContext) -> None:
        print(f"[Coder] Menerima rencana untuk Task ID: {message.task_id}")
        prompt = f"Write Python code based on this plan:\n{message.result}"
        
        response = await self._model_client.create(
            messages=[{"role": "user", "content": prompt}]
        )
        print(f"[Coder] Hasil Eksekusi Kode:\n{response.content}")

async def main():
    runtime = SingleThreadedAgentRuntime()
    client = get_ollama_client()

    # Registrasi Agen ke Runtime
    await PlannerAgent.register(
        runtime, 
        "planner", 
        lambda: PlannerAgent(model_client=client)
    )
    await CoderAgent.register(
        runtime, 
        "coder", 
        lambda: CoderAgent(model_client=client)
    )

    runtime.start()

    # Routing setup: Hubungkan topic ke agen penerima
    await runtime.add_subscription(
        subscription=default_subscription.get_subscription(PlannerAgent),
        agent_type="planner"
    )
    await runtime.add_subscription(
        subscription=default_subscription.get_subscription(CoderAgent),
        agent_type="coder"
    )

    # Kirim event awal
    await runtime.publish_message(
        TaskRequest(task_id="REQ-001", instruction="Buat script async web scraper dengan httpx"),
        topic_id=TopicId("planner_topic", source="user")
    )

    await runtime.stop_when_idle()

if __name__ == "__main__":
    asyncio.run(main())

3. Detail Analisis Komponen Kode

Kelas RoutedAgent menyediakan kapabilitas routing otomatis berdasarkan annotation @message_handler. Ketika kelas turunan didaftarkan pada runtime, instance akan mendengarkan event yang dikirimkan ke TopicId yang sesuai.

Penggunaan SingleThreadedAgentRuntime mengelola eksekusi pesan dalam satu event loop asyncio. runtime ini menjamin pemrosesan non-blocking tanpa overhead multithreading kompleks. Integrasi dengan OpenAIChatCompletionClient memetakan panggilan metode AutoGen ke API endpoint Ollama lokal pada http://localhost:11434/v1.

4. Checklist Tahapan Eksekusi

• Pastikan service Ollama berjalan pada port 11434.
• Verifikasi kecocokan tipe payload pesan menggunakan Pydantic BaseModel.
• Daftarkan agen ke AgentRuntime sebelum memanggil method runtime.start().
• Tentukan TopicId yang jelas untuk mengatur alur messaging publikasi dan penanganan event.

Best Practice Handling State Drift & Troubleshooting Error Agent Loop

Penggunaan LLM lokal pada arsitektur multi-agent rentan terhadap dua masalah mendasar: State Drift (pergeseran konteks percakapan) dan Infinite Routing Loop (agen saling berkirim pesan tanpa henti).

Strategi Pencegahan State Drift

1. Strict Schema Validation: Gunakan Pydantic model untuk mengunci format input/output antar agen. Hindari pengiriman string mentah (raw text) tanpa struktur. Validasi tipe data pada runtime mencegah penyebaran data korup antar eksekutor.
2. Window Context Trimming: Potong riwayat percakapan secara berkala. Simpan state sistem pada database eksternal (misal: Redis atau PostgreSQL) daripada memuat seluruh histori ke dalam memory context window LLM lokal. Pendekatan ini menstabilkan penggunaan token dan menjaga latensi inferensi tetap konstan.
3. Immutable Agent State: Desain state dalam agen bersifat read-only per transaksi event untuk mencegah race condition saat eksekusi asinkron. Objek pesan tidak boleh diubah di tengah alur rantai event.

Troubleshooting Agent Loop dan Error Handling

Loop tanpa henti terjadi ketika agen gagal mengidentifikasi kondisi berhenti (termination condition). Kegagalan ini kerap dipicu oleh ketidakmampuan LLM lokal mengenali token akhir atau instruksi penutupan tugas.

Untuk mengimplementasikan mekanisme guard control pada handler pesan agen, jalankan langkah-langkah berikut:

1. Definisikan konstanta ambang batas maksimum interaksi (misalnya MAX_ITERATIONS = 5).
2. Tambahkan atribut counter pada variabel instansi agen untuk mencatat jumlah eksekusi pesan.
3. Periksa apakah nilai counter telah mencapai batas maksimum di setiap awal metode handler.
4. Hentikan alur eksekusi pesan dan cetak log peringatan jika ambang batas terlampaui.
5. Bungkus pemanggilan LLM dalam blok try-except untuk menangkap exception runtime dan mengarahkannya ke kanal penanganan error khusus.

# Contoh penambahan guard control dalam handler pesan agen
MAX_ITERATIONS = 5

@message_handler
async def handle_message_with_guard(self, message: TaskResponse, ctx: MessageContext) -> None:
    # Check turn count untuk mencegah infinite loop
    if self.iteration_counter >= MAX_ITERATIONS:
        print("[System Guard] Batas maksimum interaksi tercapai. Menghentikan loop.")
        return

    self.iteration_counter += 1
    
    try:
        # Eksekusi logika LLM
        result = await self._execute_llm_call(message)
        if "COMPLETE" in result:
            print("[System] Tugas selesai.")
            return
        
        # Teruskan ke agen lain jika belum selesai
        await self.publish_message(
            TaskResponse(task_id=message.task_id, result=result),
            topic_id=TopicId("coder_topic", source=self.id.key),
        )
    except Exception as e:
        print(f"[Error Handler] Gagal memproses pesan: {str(e)}")
        # Implementasikan fallback logic atau pengiriman pesan ke ErrorTopic

Kesimpulan & Roadmap Implementasi di Environment Production

AutoGen v0.4 mengubah lanskap pembuatan aplikasi Agentic AI. Pendekatan berbasis event dan Actor Model memberikan kontrol penuh atas alur kerja agen asinkron. Kombinasi AutoGen v0.4 dan Ollama lokal menghadirkan arsitektur multi-agent performa tinggi tanpa ketergantungan pada API berbayar.

Roadmap Implikasi Production

Untuk memindahkan sistem multi-agent lokal ke lingkungan produksi terdistribusi, ikuti tahapan berikut:

1. Tahap 1 (Containerization): Bungkus service Ollama dan runtime Python AutoGen ke dalam Docker container terpisah. Gunakan Docker Compose untuk mengelola dependensi jaringan antar container.
2. Tahap 2 (Distributed Runtime): Ganti SingleThreadedAgentRuntime dengan distributed runtime berbasis gRPC atau RabbitMQ via autogen-ext untuk penskalaan multi-node horizontal.
3. Tahap 3 (Observability): Integrasikan OpenTelemetry untuk melakukan tracing lalu lintas pesan dan mengukur latensi respons tiap agen secara real-time.
4. Tahap 4 (Model Optimization): Gunakan teknik quantization (Q4_K_M atau Q8_0) pada Ollama untuk mengoptimalkan pemanfaatan VRAM GPU di server produksi.