Membangun Multi-Agent AI Workflow dengan LangGraph + FastAPI (Panduan Lengkap 2026)

1) Introduction — What & Why

Kalau kamu aktif ngikutin tren developer 2026, kamu pasti notice satu pola besar: agentic architecture lagi naik daun banget.

Dari riset cepat pagi ini:

GitHub Trending dipenuhi repo seputar agent orchestration, multi-agent framework, dan workflow automation (contoh: LangGraph ecosystem, Deep Agents discussion, sampai security tooling untuk AI pipeline).
Di X Explore, topik AI masih dominan (berita OpenAI, tooling baru, dan diskusi performa/quality AI-generated code).
Di DEV Community, artikel yang naik biasanya seputar:
- multi-agent deployment,
- AI coding workflow,
- keamanan agent,
- dan “how to build production-ready AI systems”.

Masalahnya: banyak tutorial berhenti di demo “chatbot hello world”. Padahal di dunia nyata, kita butuh sistem yang:

bisa dipecah jadi beberapa peran (planner, researcher, writer, reviewer),
stateful (punya memori proses),
bisa diobservasi,
dan aman untuk dijalankan di backend/API.

Di tutorial ini, kita akan bangun workflow multi-agent yang benar-benar runnable pakai:

LangGraph untuk orkestrasi graph,
FastAPI untuk expose endpoint,
Redis untuk cache + memory sederhana,
plus praktik error handling dan struktur kode yang maintainable.

Target: setelah selesai, kamu punya fondasi arsitektur yang bisa dipakai buat content generation, ticket triage, support automation, sampai internal copilots.

2) Prerequisites

Sebelum mulai, siapkan:

Skill teknis

Python dasar (function, class, typing, async dasar)
Paham API HTTP (GET/POST)
Familiar sama command line

Tools

Python 3.11+
pip / uv
Redis (lokal via Docker atau service)
Editor (VS Code, Cursor, dll)

Opsional (untuk lanjut ke production)

API key model provider (OpenAI/Anthropic/Gemini, dll)
Logging/observability stack (mis. LangSmith / OpenTelemetry)

3) Core Concepts (dengan analogi)

Bayangin kamu punya tim kecil di kantor:

Planner = PM yang memecah masalah
Researcher = analis yang ngumpulin fakta
Writer = eksekutor yang bikin output
Reviewer = QA yang ngecek kualitas

Nah, LangGraph itu seperti SOP + flowchart tim kamu.

Konsep utama

State
- “Papan kerja bersama” tempat semua agent baca/tulis data.
- Contoh: topik, outline, draft, warning, final output.
Node
- Satu unit kerja (fungsi agent).
- Misal planner_node, research_node, writer_node, reviewer_node.
Edge
- Jalur perpindahan antar node.
- Bisa linear, branching, atau conditional.
Durable mindset
- Workflow kompleks sering gagal di tengah (timeout, API error, data invalid).
- Desain sejak awal harus ada retry, fallback, dan observability.
Context control
- Salah satu alasan multi-agent populer: mencegah context bloat.
- Subtask dipisah biar tidak semua detail numpuk di satu prompt/state besar.

4) Architecture / Diagram

Kita pakai arsitektur sederhana tapi realistis:

Client (Web/App)
      |
      v
+-------------------+
| FastAPI Endpoint  |  POST /generate
+-------------------+
      |
      v
+-------------------+        +------------------+
| LangGraph Engine  | <----> | Redis Cache/State|
+-------------------+        +------------------+
      |
      v
[planner] -> [researcher] -> [writer] -> [reviewer]
      |                                          |
      +------------------(if fail)--------------+
                         retry/fix

Data flow ringkas

Client kirim request (topic, audience, tone).
FastAPI validasi input pakai Pydantic.
Workflow graph dijalankan.
Tiap node update state.
Hasil final disimpan ke Redis + dikembalikan ke client.

5) Step-by-Step Implementation (complete runnable code)

5.1 Struktur project

multi-agent-workflow/
├── app/
│   ├── main.py
│   ├── graph_workflow.py
│   ├── schemas.py
│   └── storage.py
├── requirements.txt
└── README.md

5.2 Install dependency

python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate
pip install -r requirements.txt

requirements.txt

fastapi==0.116.1
uvicorn==0.35.0
langgraph==0.3.5
pydantic==2.12.5
redis==6.2.0

5.3 Kode: `app/schemas.py`

from pydantic import BaseModel, Field, field_validator


class GenerateRequest(BaseModel):
    topic: str = Field(..., min_length=5, max_length=200)
    audience: str = Field(default="developer pemula")
    tone: str = Field(default="praktis")

    @field_validator("topic")
    @classmethod
    def topic_must_not_be_empty(cls, value: str) -> str:
        if not value.strip():
            raise ValueError("Topic tidak boleh kosong")
        return value.strip()


class GenerateResponse(BaseModel):
    request_id: str
    topic: str
    final_output: str
    quality_score: int
    warnings: list[str]

5.4 Kode: `app/storage.py`

import json
from redis import Redis
from redis.exceptions import RedisError


class Storage:
    def __init__(self, url: str = "redis://localhost:6379/0"):
        self.client = Redis.from_url(url, decode_responses=True)

    def save_state(self, key: str, data: dict) -> None:
        """Simpan state ke Redis dengan handling error yang jelas."""
        try:
            self.client.setex(key, 3600, json.dumps(data, ensure_ascii=False))
        except RedisError as err:
            # Jangan crash total workflow hanya karena cache gagal
            print(f"[WARN] Gagal simpan ke Redis: {err}")

    def load_state(self, key: str) -> dict | None:
        try:
            raw = self.client.get(key)
            return json.loads(raw) if raw else None
        except (RedisError, json.JSONDecodeError) as err:
            print(f"[WARN] Gagal baca dari Redis: {err}")
            return None

5.5 Kode: `app/graph_workflow.py`

from __future__ import annotations

from typing import TypedDict
from langgraph.graph import StateGraph, START, END


class WorkflowState(TypedDict, total=False):
    request_id: str
    topic: str
    audience: str
    tone: str
    outline: str
    research_notes: str
    draft: str
    final_output: str
    quality_score: int
    warnings: list[str]


def planner_node(state: WorkflowState) -> WorkflowState:
    topic = state["topic"]
    outline = (
        f"Outline untuk topik '{topic}':
"
        "1) Definisi
"
        "2) Arsitektur
"
        "3) Implementasi
"
        "4) Best practices
"
        "5) Risiko
"
    )
    return {"outline": outline}


def researcher_node(state: WorkflowState) -> WorkflowState:
    topic = state["topic"].lower()

    # Simulasi hasil riset (runnable tanpa API eksternal)
    notes = [
        "Gunakan pemisahan peran agent agar konteks tidak menumpuk.",
        "Tambahkan validasi input-output antar node.",
        "Sediakan retry + fallback saat node gagal.",
        "Gunakan cache untuk hasil subtask yang mahal.",
    ]

    if "security" in topic:
        notes.append("Lakukan sanitasi prompt/tool input untuk mencegah injection.")

    return {"research_notes": "
".join(f"- {n}" for n in notes)}


def writer_node(state: WorkflowState) -> WorkflowState:
    draft = (
        f"Topik: {state['topic']}

"
        f"Target pembaca: {state['audience']}
"
        f"Nada: {state['tone']}

"
        f"{state['outline']}
"
        "Catatan riset:
"
        f"{state['research_notes']}

"
        "Implementasi dimulai dari desain state, lalu node, lalu observability."
    )
    return {"draft": draft}


def reviewer_node(state: WorkflowState) -> WorkflowState:
    draft = state["draft"]
    warnings: list[str] = []
    score = 90

    if len(draft) < 300:
        warnings.append("Draft terlalu pendek")
        score -= 20

    if "error handling" not in draft.lower():
        warnings.append("Belum menekankan error handling")
        score -= 10

    final_output = draft + "

[Reviewed] Tambahkan logging, metrics, dan test sebelum production."

    return {
        "final_output": final_output,
        "quality_score": max(score, 0),
        "warnings": warnings,
    }


def build_graph():
    graph = StateGraph(WorkflowState)

    graph.add_node("planner", planner_node)
    graph.add_node("researcher", researcher_node)
    graph.add_node("writer", writer_node)
    graph.add_node("reviewer", reviewer_node)

    graph.add_edge(START, "planner")
    graph.add_edge("planner", "researcher")
    graph.add_edge("researcher", "writer")
    graph.add_edge("writer", "reviewer")
    graph.add_edge("reviewer", END)

    return graph.compile()

5.6 Kode: `app/main.py`

import uuid
from fastapi import FastAPI, HTTPException
from pydantic import ValidationError

from app.schemas import GenerateRequest, GenerateResponse
from app.graph_workflow import build_graph
from app.storage import Storage


app = FastAPI(title="Multi-Agent Workflow API", version="1.0.0")
workflow = build_graph()
storage = Storage()


@app.get("/health")
def health_check():
    return {"status": "ok"}


@app.post("/generate", response_model=GenerateResponse)
def generate_content(payload: GenerateRequest):
    request_id = str(uuid.uuid4())

    initial_state = {
        "request_id": request_id,
        "topic": payload.topic,
        "audience": payload.audience,
        "tone": payload.tone,
    }

    try:
        result = workflow.invoke(initial_state)

        # Simpan hasil untuk audit/debug sederhana
        storage.save_state(f"workflow:{request_id}", result)

        return GenerateResponse(
            request_id=request_id,
            topic=payload.topic,
            final_output=result.get("final_output", ""),
            quality_score=result.get("quality_score", 0),
            warnings=result.get("warnings", []),
        )

    except ValidationError as err:
        raise HTTPException(status_code=422, detail=str(err))
    except Exception as err:
        # Error handling global supaya API tetap informatif
        raise HTTPException(
            status_code=500,
            detail=f"Workflow gagal diproses: {err}",
        )

5.7 Run project

uvicorn app.main:app --reload --port 8000

Test endpoint:

curl -X POST http://127.0.0.1:8000/generate \
  -H 'Content-Type: application/json' \
  -d '{
    "topic": "Membangun multi-agent AI untuk support ticket",
    "audience": "backend engineer",
    "tone": "praktis"
  }'

Kalau sukses, kamu dapat JSON response berisi final_output, quality_score, dan warnings.

6) Best Practices (tips industri)

Pisahkan domain agent dengan jelas
- Jangan bikin semua node “serba bisa”.
- Deskripsi tugas harus spesifik (planner merencanakan, reviewer menilai).
Treat state as contract
- Anggap struktur state seperti API contract.
- Versioning state kalau workflow makin besar.
Tambah guardrails di tiap node
- Validasi input node.
- Batasi panjang output.
- Normalisasi format (JSON schema kalau perlu).
Observability itu wajib, bukan bonus
- Simpan trace node mana yang gagal.
- Catat latency per node.
- Log prompt/tool calls (tanpa bocorin secret).
Idempotency untuk endpoint produksi
- Kalau user retry request, jangan selalu generate ulang dari nol.
- Gunakan request fingerprint + cache result.
Fail soft, not fail silent
- Saat Redis down, workflow bisa lanjut dengan in-memory fallback.
- Saat satu tool gagal, coba fallback strategy.

7) Common Mistakes (dan cara hindarinya)

Mistake 1: “Multi-agent” tapi sebenarnya monolitik

Gejala: semua logic di satu node besar.
Dampak: susah debug, context bloat, sulit scaling.
Solusi: pecah node per tanggung jawab.

Mistake 2: Tidak ada schema validation

Gejala: hasil antar node tidak konsisten.
Dampak: error random di tengah workflow.
Solusi: validasi ketat dengan Pydantic/TypedDict.

Mistake 3: Mengabaikan timeout/retry

Gejala: request menggantung lama.
Dampak: UX buruk, biaya API naik.
Solusi: set timeout eksplisit, retry terbatas (exponential backoff).

Mistake 4: Langsung production tanpa test path gagal

Gejala: happy path jalan, edge case hancur.
Dampak: incident sulit didiagnosis.
Solusi: test skenario API gagal, cache mati, output invalid.

Mistake 5: Prompt/tool injection diabaikan

Gejala: agent mengikuti instruksi berbahaya dari input/tool result.
Dampak: kebocoran data / tindakan tidak aman.
Solusi: sanitasi input, policy enforcement, allowlist tool action.

8) Advanced Tips

Kalau fondasi di atas sudah stabil, ini upgrade path yang worth it:

Conditional routing
- Tambahkan edge bercabang: topik teknis masuk researcher A, topik bisnis ke researcher B.
Parallel sub-agents
- Riset sumber A/B/C paralel, lalu agregasi.
- Ini bisa menurunkan total latency end-to-end.
Human-in-the-loop approval
- Untuk langkah sensitif (mis. publish otomatis), sisipkan node “approval required”.
Quality gates berbasis skor
- Kalau quality_score < 80, otomatis loop balik ke writer untuk revisi.
Hybrid memory strategy
- Redis buat short-term cache.
- DB (Postgres) buat audit trail dan long-term memory.
Model specialization
- Planner pakai model cepat/murah.
- Reviewer pakai model lebih kuat untuk evaluasi kualitas.
Security hardening
- Secret di vault/env manager.
- Rate limit endpoint.
- Input/output redaction untuk data sensitif.

9) Summary & Next Steps

Kita baru saja bangun sistem multi-agent yang:

punya arsitektur jelas,
runnable lokal,
ada error handling,
dan siap jadi fondasi production.

Inti pentingnya:

LangGraph cocok untuk orkestrasi workflow kompleks berbasis state.
FastAPI memudahkan expose service yang cepat dan terstruktur.
Redis membantu cache/memory untuk performa dan reliability.

Next steps yang saya rekomendasikan

Tambah test otomatis (unit + integration).
Pasang tracing/metrics untuk tiap node.
Implement retry policy + circuit breaker.
Sambungkan ke provider LLM nyata (dengan fallback model).
Tambah CI/CD + security scanning sebelum production.

Kalau kamu serius bangun AI system tahun ini, belajar orchestration + reliability + security akan jauh lebih berharga daripada sekadar “prompting”.

10) References

LangGraph Overview (official docs): https://docs.langchain.com/oss/python/langgraph/overview
LangGraph GitHub: https://github.com/langchain-ai/langgraph
LangChain Blog — Building Multi-Agent Applications with Deep Agents: https://blog.langchain.com/building-multi-agent-applications-with-deep-agents/
FastAPI Documentation: https://fastapi.tiangolo.com/
Pydantic Documentation: https://docs.pydantic.dev/latest/
Redis Documentation: https://redis.io/docs/latest/develop/get-started/
LangGraph Examples (archived path info): https://github.com/langchain-ai/langgraph/tree/main/examples

Penutup: Tren 2026 jelas: tim engineering bergerak dari single-agent demo ke multi-agent systems yang bisa dioperasikan. Semoga tutorial ini bukan cuma bikin kamu “paham konsep”, tapi juga punya baseline implementasi yang siap kamu evolve ke production 🚀

Membangun Multi-Agent AI Workflow dengan LangGraph + FastAPI (Panduan Lengkap 2026)

Membangun Multi-Agent AI Workflow dengan LangGraph + FastAPI (Panduan Lengkap 2026)

1) Introduction — What & Why

2) Prerequisites

Skill teknis

Tools

Opsional (untuk lanjut ke production)

3) Core Concepts (dengan analogi)

Konsep utama

4) Architecture / Diagram

Data flow ringkas

5) Step-by-Step Implementation (complete runnable code)

5.1 Struktur project

5.2 Install dependency

5.3 Kode: app/schemas.py

5.4 Kode: app/storage.py

5.5 Kode: app/graph_workflow.py

5.6 Kode: app/main.py

5.7 Run project

6) Best Practices (tips industri)

7) Common Mistakes (dan cara hindarinya)

Mistake 1: “Multi-agent” tapi sebenarnya monolitik

Mistake 2: Tidak ada schema validation

Mistake 3: Mengabaikan timeout/retry

Mistake 4: Langsung production tanpa test path gagal

Mistake 5: Prompt/tool injection diabaikan

8) Advanced Tips

9) Summary & Next Steps

Next steps yang saya rekomendasikan

10) References

5.3 Kode: `app/schemas.py`

5.4 Kode: `app/storage.py`

5.5 Kode: `app/graph_workflow.py`

5.6 Kode: `app/main.py`