Arsitektur AI Modern: End-to-End

Dari query user pertama hingga response tersimpan ke database. Panduan lengkap layer, tools, request flow, learning roadmap, dan penerapan nyata di production — termasuk Google ADK yang baru rilis.

Arsitektur AI Modern: End-to-End

Dari query user pertama hingga response tersimpan ke database. Panduan lengkap layer, tools, request flow, learning roadmap, dan penerapan nyata di production — termasuk Google ADK yang baru rilis.

Mengapa Perlu Memahami Ini?

LLM hanya bisa melakukan satu hal: text in → text out. Ia tidak bisa ingat percakapan sebelumnya, akses database, eksekusi kode, atau ambil keputusan multi-step. Semua "kecanggihan" AI yang kamu lihat di production bukan karena model-nya pintar sendirian — tapi karena ada arsitektur di sekitar model yang mengkoordinasikan semua kemampuan itu.

Memahami arsitektur ini memberi kamu kemampuan untuk: memilih tool yang tepat untuk problem yang tepat, mendesain sistem AI yang bisa di-maintain, dan debug ketika sesuatu tidak bekerja seperti yang diharapkan.

Layer Stack — 8 Lapisan Arsitektur

Sistem AI modern terdiri dari 8 layer yang bekerja bersama. Klik setiap layer di bawah ini untuk melihat detail penjelasan dan tools-nya.

Request Flow — Perjalanan Satu Query

Berikut adalah perjalanan lengkap satu query user dari awal hingga response diterima. Tidak ada step yang terjadi secara ajaib — semuanya eksplisit dan bisa di-trace.

Konsep Fundamental

Sebelum masuk ke framework spesifik, ini 6 konsep yang harus benar-benar dipahami. Semua framework — LangGraph, Google ADK, CrewAI — dibangun di atas konsep ini.

4.1 ReAct Loop — Pola Dasar Semua Agent

ReAct (Reason + Act) adalah pola fundamental yang dipakai semua agentic framework. Agent loop ini terus berulang sampai agent yakin jawabannya cukup.

4.2 Context Window — Apa yang Dilihat LLM

LLM adalah fungsi matematika: $f(\text{tokens\_in}) \rightarrow \text{probability\_distribution}(\text{tokens\_out})$. Tidak ada memori, tidak ada state. Setiap call, ia membaca seluruh context dari awal. Semua "pengetahuan" LLM dalam satu request ada di dalam context window ini.

4.3 RAG Pipeline — Cara Kerja Detail

RAG terdiri dari dua fase yang terpisah: Indexing (offline, sekali jalan) dan Retrieval (online, setiap request).

4.4 Tool Calling — Mekanisme Lengkap

Ini yang membuat LLM bisa berinteraksi dengan dunia nyata. Bukan magic — ini sequence yang eksplisit dan deterministic dari sisi orchestration.

# Step 1: Define tool sebagai Python function biasa
@tool
def check_inventory(product_id: str) -> dict:
    """Cek stok produk dari database. Args: product_id (str)"""
    return db.query("SELECT * FROM products WHERE id = %s", product_id)

# Step 2: LLM menerima tool schema di context
# {"name": "check_inventory", "description": "Cek stok...", "parameters": {...}}

# Step 3: LLM output (bukan eksekusi!) — JSON tool call
# {"type": "tool_call", "name": "check_inventory", "args": {"product_id": "A"}}

# Step 4: Orchestration layer eksekusi fungsi Python
result = check_inventory(product_id="A")  # {"stock": 142, "safety_stock": 200}

# Step 5: Inject hasil sebagai tool_result ke context
# {"type": "tool_result", "content": {"stock": 142, "safety_stock": 200}}

# Step 6: LLM dipanggil LAGI dengan context lengkap + tool result
# Kali ini LLM bisa generate jawaban final berdasarkan data nyata

4.5 Structured Output — Dari Teks ke Data

Tanpa structured output, LLM adalah sumber teks. Dengan structured output, LLM menjadi fungsi yang bisa dipanggil seperti API — output-nya langsung bisa diproses kode tanpa parsing manual.

import instructor
from pydantic import BaseModel, Field
from typing import Literal

# 1. Define schema yang diinginkan
class InventoryAlert(BaseModel):
    product_id: str
    current_stock: int = Field(ge=0)
    status: Literal["ok", "low", "critical", "stockout"]
    reorder_suggested: bool
    suggested_quantity: int | None = None
    reason: str = Field(max_length=200)

# 2. Instructor wrap LLM client
client = instructor.from_anthropic(Anthropic())

# 3. Response langsung berupa Python object — bukan string
alert = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=500,
    response_model=InventoryAlert,  # inject schema ke tool calling
    messages=[{"role": "user", "content": context}]
)

# alert.status == "low" — type-safe, langsung pakai
if alert.reorder_suggested:
    trigger_purchase_order(alert.product_id, alert.suggested_quantity)

4.6 Observability dengan LangFuse

LangFuse adalah observability layer khusus AI — analoginya Grafana untuk infra, tapi untuk LLM calls. Hierarki datanya: Trace (satu end-to-end request) $\rightarrow$ Span (unit kerja) $\rightarrow$ Generation (satu LLM call) $\rightarrow$ Score (evaluasi).

# Cara paling simple: LiteLLM integration (1 baris)
import os
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-..."
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-..."
litellm.success_callback = ["langfuse"]  # semua LLM call auto-traced

# Atau via decorator untuk custom spans
from langfuse.decorators import observe

@observe(name="inventory-analysis")
def analyze_inventory(query: str) -> str:
    chunks = retrieve_docs(query)     # jadi child span
    answer = generate_answer(query, chunks)  # jadi child span
    return answer

# Yang di-capture per generation:
# latency, input/output tokens, cost, faithfulness score, user feedback

Google ADK — Framework Multi-Agent Baru

from google.adk.agents import LlmAgent
from google.adk.tools import tool
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService

# 1. Define tool
@tool
def check_stock(product_id: str) -> dict:
    """Cek stok produk dari inventory database"""
    return db.get_stock(product_id)

# 2. Create agent
inventory_agent = LlmAgent(
    name="inventory-analyst",
    model="gemini-2.0-flash",
    tools=[check_stock],
    instruction="Kamu adalah analis inventory. Bantu user memahami kondisi stok."
)

# 3. Run
session_service = InMemorySessionService()
runner = Runner(agent=inventory_agent, session_service=session_service)

response = runner.run(
    user_id="user_123",
    session_id="session_abc",
    message="Berapa stok produk A sekarang?"
)

ADK vs Framework Lain

Format Response AI di Production

Format response yang baik bukan tentang estetika — tapi tentang predictability, parsability, dan contract yang tidak berubah saat model di-update.

class InventoryInsight(BaseModel):
    product_id: str
    action: Literal["reorder", "hold", "liquidate"]
    urgency: Literal["immediate", "this_week", "this_month"]
    confidence: float = Field(ge=0.0, le=1.0)
    suggested_quantity: int | None = None

insight = client.messages.create(
    response_model=InventoryInsight,
    messages=[{"role": "user", "content": context}]
)
# insight.action == "reorder" — langsung gunakan, no parsing

@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    async def generate():
        with client.messages.stream(
            model="claude-sonnet-4-6",
            messages=request.messages
        ) as stream:
            for text in stream.text_stream:
                yield f"data: {json.dumps({'token': text})}\n\n"
        yield "data: [DONE]\n\n"
    return StreamingResponse(generate(), media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})

class AIResponse[T](BaseModel):
    request_id: str        # untuk tracing
    trace_id: str | None   # LangFuse trace ID
    model: str
    latency_ms: int
    input_tokens: int
    output_tokens: int
    cost_usd: float
    data: T                  # payload utama (InventoryInsight, dll)
    confidence: float
    fallback_used: bool = False
    retry_count: int = 0
    context_sources: list[str] = []  # chunk ID dari RAG (audit trail)
    generated_at: datetime

class AIErrorCode(str, Enum):
    VALIDATION_FAILED   = "validation_failed"   # Instructor retry habis
    CONTEXT_TOO_LONG    = "context_too_long"    # melebihi context window
    MODEL_UNAVAILABLE   = "model_unavailable"   # semua provider down
    RATE_LIMITED        = "rate_limited"        # quota habis
    TIMEOUT             = "timeout"             # inference terlalu lama

class AIErrorResponse(BaseModel):
    error_code: AIErrorCode
    retryable: bool         # True = rate_limited/timeout; False = validation
    retry_after_seconds: int | None
    request_id: str

Learning Roadmap — 8 Minggu

Urutan belajar yang optimal berdasarkan dependency antar konsep. Tidak ada shortcut — tiap minggu membangun fondasi untuk minggu berikutnya.

Production Readiness Checklist

Gunakan ini sebelum deploy AI feature ke production. Setiap item yang terlewat adalah potensi incident.

SCHEMA

• Semua field typed, tidak ada dict atau Any
• Field nullable di-mark Optional / None eksplisit
• Field numerik punya constraint (ge=0, le=1.0, dll)
• Literal atau Enum untuk field yang nilainya terbatas

DELIVERY

• Batch endpoint untuk data pipeline (JSON + Pydantic)
• Streaming endpoint untuk chat interface (SSE)
• SSE header benar: Cache-Control: no-cache, X-Accel-Buffering: no
• Timeout di-set (bukan infinite wait)
• Max tokens di-set eksplisit di setiap LLM call

ERROR HANDLING

• Validation error ter-catch sebelum masuk DB
• Retry logic ada (Instructor 3x default, atau manual)
• Error response punya retryable flag
• request_id ada di setiap response untuk tracing
• Fallback provider dikonfigurasi di LiteLLM

OBSERVABILITY

• trace_id link ke LangFuse di setiap response
• latency_ms di-log per request
• Input/output tokens di-log untuk cost monitoring
• context_sources (chunk ID dari RAG) di-log untuk audit trail
• PII masking dikonfigurasi di LangFuse sebelum go-live

SECURITY

• Prompt injection protection — validasi input user sebelum masuk ke system prompt
• Rate limiting per user / per endpoint
• API key tidak hardcoded — gunakan env var atau secret manager
• Output validation — AI output yang masuk ke DB harus melalui Pydantic