Tutorial Lengkap Membangun Multi-Agent AI Workflow dengan LangGraph + FastAPI (2026)
Panduan praktis membangun layanan AI agent modern dengan LangGraph dan FastAPI: dari konsep stateful graph, arsitektur API production-ready, implementasi kode runnable, hingga best practices, anti-pattern, dan langkah scaling.
Tutorial Lengkap Membangun Multi-Agent AI Workflow dengan LangGraph + FastAPI (2026)
1) Introduction — What and Why
Dalam 12 bulan terakhir, topik AI agent orchestration naik tajam di komunitas developer. Dari GitHub Trending, banyak repo tentang agent framework, managed agents, dan tool-calling workflow yang melesat. Di komunitas developer (termasuk artikel terbaru di DEV), pembahasannya juga bergeser: bukan lagi sekadar “call LLM sekali”, tapi bagaimana membangun alur kerja agent yang tahan gagal, bisa diobservasi, dan siap dipakai di production.
Masalahnya, banyak implementasi agent masih seperti skrip demo:
- state tidak persist
- error handling minim
- sulit di-debug
- sulit diintegrasikan ke backend aplikasi nyata
Di sinilah kombinasi LangGraph + FastAPI sangat kuat:
- LangGraph: mengatur alur agent sebagai graph (node + edge), cocok untuk workflow kompleks, stateful, dan long-running.
- FastAPI: API layer modern, cepat, type-safe, enak untuk production.
Analogi sederhananya:
- LangGraph itu otak alur kerja (siapa mikir dulu, kapan pakai tools, kapan selesai).
- FastAPI itu pintu layanan (bagaimana aplikasi/mobile/web kamu berinteraksi dengan otak tadi).
Di tutorial ini, kamu akan membangun satu service yang bisa:
- menerima request user,
- menjalankan alur agent bertahap,
- mengembalikan jawaban terstruktur,
- menangani error dengan rapi,
- siap dikembangkan ke multi-agent yang lebih advanced.
2) Prerequisites
Sebelum mulai, siapkan:
- Python 3.11+
- Basic pemahaman REST API
- Basic async/await Python
- API key provider LLM (opsional, karena kita siapkan fallback mock)
Install dependencies:
python -m venv .venv source .venv/bin/activate # Windows: .venv\Scripts\activate pip install --upgrade pip pip install fastapi uvicorn langgraph langchain pydantic python-dotenv
Buat struktur project:
agent-service/ ├── app/ │ ├── main.py │ ├── graph_builder.py │ ├── schemas.py │ └── settings.py ├── .env └── requirements.txt
Contoh .env:
APP_NAME=LangGraph FastAPI Agent Service APP_ENV=development OPENAI_API_KEY= DEFAULT_MODEL=gpt-4o-mini
Kalau API key kosong, service tetap jalan pakai mock-response agar tetap runnable.
3) Core Concepts
Sebelum ngoding, pahami fondasinya.
a) State Graph
Di LangGraph, setiap langkah adalah node yang menerima dan mengubah state.
Bayangkan seperti papan kanban otomatis:
- Kolom 1: baca intent user
- Kolom 2: tentukan perlu tools atau tidak
- Kolom 3: eksekusi tools
- Kolom 4: final response
Setiap perpindahan kolom = edge.
b) Orchestration vs Single Prompt
Single prompt itu seperti nanya 1 pertanyaan lalu selesai. Orchestration itu seperti punya tim mini:
- planner (rencana)
- worker (eksekusi)
- reviewer (cek hasil)
Ini penting untuk task nyata yang butuh beberapa langkah.
c) Typed State (Pydantic / TypedDict mindset)
Kalau state kamu liar (dict bebas), bug gampang masuk. Dengan schema yang jelas, data antar node jadi konsisten. Ini seperti kontrak antar tim.
d) Durable dan Observability
Di level production, pertanyaan utamanya bukan “jalan atau nggak”, tapi:
- kalau gagal di tengah, bisa resume?
- bisa tahu node mana paling sering error?
- latency per step berapa?
Makanya dari awal kita desain dengan error handling dan metadata.
4) Architecture / Diagram
Arsitektur minimal yang akan kita buat:
Client (Web/Mobile) | v +-------------------+ | FastAPI Endpoint | | POST /v1/agent/run| +-------------------+ | v +-------------------+ +----------------------+ | LangGraph Engine | ---> | Optional LLM Provider| | (planner->worker) | | (OpenAI/others) | +-------------------+ +----------------------+ | v +-------------------+ | Structured Result | | answer + metadata | +-------------------+
Flow request:
- Validasi payload (FastAPI + Pydantic model)
- Build initial state
- Jalankan graph invoke
- Tangkap exception dan map ke HTTP error yang jelas
- Return response terstruktur
5) Step-by-Step Implementation (Complete Runnable Code)
File 1 — app/settings.py
from pydantic import BaseModel from dotenv import load_dotenv import os load_dotenv() class Settings(BaseModel): app_name: str = os.getenv("APP_NAME", "LangGraph FastAPI Agent Service") app_env: str = os.getenv("APP_ENV", "development") openai_api_key: str = os.getenv("OPENAI_API_KEY", "") default_model: str = os.getenv("DEFAULT_MODEL", "gpt-4o-mini") settings = Settings()
File 2 — app/schemas.py
from pydantic import BaseModel, Field from typing import List, Dict, Any class AgentRunRequest(BaseModel): user_id: str = Field(..., min_length=2, max_length=100) query: str = Field(..., min_length=3, max_length=5000) class AgentRunResponse(BaseModel): success: bool answer: str steps: List[str] meta: Dict[str, Any]
File 3 — app/graph_builder.py
from typing import TypedDict, List, Dict, Any from langgraph.graph import StateGraph, START, END import time class AgentState(TypedDict, total=False): query: str plan: str answer: str steps: List[str] errors: List[str] meta: Dict[str, Any] def planner_node(state: AgentState) -> AgentState: # Menentukan strategi jawaban berdasarkan query user. steps = state.get("steps", []) steps.append("planner_node") query = state.get("query", "") if not query: errors = state.get("errors", []) errors.append("Query kosong pada planner_node") return {**state, "errors": errors, "steps": steps} plan = ( "1) Pahami intent user. " "2) Berikan jawaban teknis ringkas. " "3) Tambahkan rekomendasi next-step praktis." ) return {**state, "plan": plan, "steps": steps} def worker_node(state: AgentState) -> AgentState: # Mengeksekusi plan dengan fallback deterministic. steps = state.get("steps", []) steps.append("worker_node") query = state.get("query", "") plan = state.get("plan", "") if not query or not plan: errors = state.get("errors", []) errors.append("Data tidak lengkap pada worker_node") return {**state, "errors": errors, "steps": steps} started = time.time() answer = ( f"Pertanyaan kamu: {query} " f"Rencana eksekusi: {plan} " "Jawaban: Untuk implementasi production, pastikan kamu " "punya validasi input, observability, dan fallback strategy saat provider LLM gagal." ) duration_ms = int((time.time() - started) * 1000) meta = state.get("meta", {}) meta.update({"worker_duration_ms": duration_ms}) return {**state, "answer": answer, "steps": steps, "meta": meta} def reviewer_node(state: AgentState) -> AgentState: # Final quality gate sebelum output dikirim. steps = state.get("steps", []) steps.append("reviewer_node") answer = state.get("answer", "") errors = state.get("errors", []) if errors: answer = "Proses selesai dengan catatan error: " + "; ".join(errors) elif len(answer) < 30: errors.append("Jawaban terlalu pendek, kemungkinan proses tidak optimal") return {**state, "answer": answer, "steps": steps, "errors": errors} def build_agent_graph(): graph = StateGraph(AgentState) graph.add_node("planner", planner_node) graph.add_node("worker", worker_node) graph.add_node("reviewer", reviewer_node) graph.add_edge(START, "planner") graph.add_edge("planner", "worker") graph.add_edge("worker", "reviewer") graph.add_edge("reviewer", END) return graph.compile()
File 4 — app/main.py
from fastapi import FastAPI, HTTPException from fastapi.responses import JSONResponse from app.settings import settings from app.schemas import AgentRunRequest, AgentRunResponse from app.graph_builder import build_agent_graph import traceback app = FastAPI(title=settings.app_name) agent_graph = build_agent_graph() @app.get("/health") async def health_check(): return {"status": "ok", "env": settings.app_env} @app.post("/v1/agent/run", response_model=AgentRunResponse) async def run_agent(payload: AgentRunRequest): try: initial_state = { "query": payload.query, "steps": [], "errors": [], "meta": {"user_id": payload.user_id, "app_env": settings.app_env}, } result = agent_graph.invoke(initial_state) errors = result.get("errors", []) if errors: return AgentRunResponse( success=False, answer=result.get("answer", "Terjadi error saat memproses permintaan."), steps=result.get("steps", []), meta={**result.get("meta", {}), "errors": errors}, ) return AgentRunResponse( success=True, answer=result.get("answer", ""), steps=result.get("steps", []), meta=result.get("meta", {}), ) except Exception as exc: print("[ERROR] run_agent failed:", str(exc)) traceback.print_exc() raise HTTPException(status_code=500, detail="Internal agent processing error") from exc @app.exception_handler(Exception) async def global_exception_handler(_, exc: Exception): return JSONResponse( status_code=500, content={ "success": False, "error": "Unhandled server error", "detail": str(exc), }, )
Menjalankan aplikasi
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
Test endpoint:
curl -X POST http://localhost:8000/v1/agent/run -H 'Content-Type: application/json' -d '{ "user_id": "user-123", "query": "Bagaimana cara membuat API agent yang robust?" }'
Kalau semua benar, kamu dapat response berisi success, answer, steps, dan meta.
6) Best Practices (Tips dari Praktisi)
-
Pisahkan node berdasarkan tanggung jawab
- planner hanya merencanakan
- worker hanya eksekusi
- reviewer hanya quality gate
-
Gunakan schema ketat untuk input/output Ini menurunkan bug integrasi antar service.
-
Buat fallback strategy Misalnya provider LLM timeout, maka:
- retry terbatas (exponential backoff)
- fallback ke model lebih murah/cepat
- atau fallback ke jawaban template
-
Tambahkan correlation ID Simpan
request_iddimetauntuk tracing lintas service. -
Observability dari hari pertama Capture latency per node, error rate, token usage, dan success ratio.
-
Guardrail input Batasi panjang query, lakukan sanitasi, dan filter instruction berbahaya.
7) Common Mistakes (dan Cara Menghindarinya)
Mistake 1: Semua logic ditaruh di satu node
Akibat: susah debug, susah test, dan rawan side-effect.
Solusi: pecah node per concern.
Mistake 2: Tidak mendesain error path
Akibat: kalau satu step gagal, seluruh pipeline collapse.
Solusi: simpan daftar error di state + graceful response.
Mistake 3: State tidak konsisten
Contoh: node A menulis result_text, node B membaca answer.
Solusi: standar naming state dari awal + type hints.
Mistake 4: Over-engineering terlalu cepat
Langsung bikin 10 agent padahal use case belum valid.
Solusi: mulai dari 3 node minimal (planner-worker-reviewer), ukur hasil, baru scale.
Mistake 5: Tidak memperhatikan biaya
Workflow agent bisa mahal kalau loop terlalu panjang.
Solusi: batasi jumlah iterasi, caching prompt, dan model routing.
8) Advanced Tips
Kalau kamu mau naik level, coba ini:
-
Conditional routing Jika intent = troubleshooting, route ke node diagnosis. Jika intent = coding, route ke node code-generator.
-
Human-in-the-loop checkpoint Untuk aksi sensitif (misalnya update data penting), minta approval manusia sebelum lanjut.
-
Memory jangka panjang Simpan preferensi user di vector store / database agar jawaban makin personal.
-
Parallel branch execution Jalankan dua strategi agent sekaligus (misal retrieval branch dan reasoning branch), lalu merge.
-
A/B testing prompt strategy Uji dua planner prompt berbeda, bandingkan accuracy + latency + cost.
-
Production deployment pattern
- FastAPI sebagai stateless API layer
- worker queue (Celery/RQ/Arq) untuk task berat
- Redis/Postgres untuk state persistence
- dashboard monitoring untuk SLA
9) Summary & Next Steps
Kita sudah membangun fondasi agent service modern:
- LangGraph untuk orkestrasi alur kerja stateful
- FastAPI untuk API layer yang rapi dan scalable
- kode runnable dengan error handling
- praktik production: observability, fallback, dan struktur node yang sehat
Next steps yang saya sarankan:
- Tambahkan integrasi LLM provider real + timeout/retry policy.
- Simpan state ke persistence store (mis. Redis/Postgres).
- Tambahkan authentication + rate limiting di endpoint.
- Buat test otomatis (unit test per node + integration test endpoint).
- Integrasikan tracing (mis. LangSmith/OpenTelemetry) untuk melihat bottleneck nyata.
Kalau kamu menguasai pola ini, kamu bukan cuma bisa bikin demo agent — kamu bisa bikin platform agent yang siap dipakai tim produk secara berkelanjutan. 🚀
10) References
- LangGraph Overview: https://docs.langchain.com/oss/python/langgraph/overview
- LangGraph Quickstart: https://docs.langchain.com/oss/python/langgraph/quickstart
- LangGraph GitHub Repository: https://github.com/langchain-ai/langgraph
- FastAPI Official Docs: https://fastapi.tiangolo.com/
- FastAPI GitHub: https://github.com/fastapi/fastapi
- Pydantic Docs: https://docs.pydantic.dev/latest/
- DEV Community (tren artikel developer): https://dev.to
- GitHub Trending (sumber tren topik): https://github.com/trending