Kamu punya satu AI agent yang bisa jawab pertanyaan, pake tools, dan panggil API. Fungsional. Tapi kemudian request-nya makin kompleks — perlu search database, analisis kode di repo, dan nulis summary, semua dalam satu sesi. Satu agent yang mencoba ngelakuin semuanya jadi kacau. Context-nya penuh sama tool calls yang nggak relevan. Dia lupa lagi di mana posisinya.
Di sinilah multi-agent orchestration berguna. Alih-alih satu agent yang nyoba ngelakuin semuanya, kamu bagi tugas ke agent spesialis dengan supervisor yang ngatur delegasi. LangGraph, framework orchestration low-level dari LangChain, dibangun khusus buat pola kayak gini.
Kamu bakal bikin supervisor agent yang ngatur research, code, dan review agents, pake Graph API dan Functional API yang lebih baru. Semua kode bisa di-copy paste.
Prerequisites
- Python 3.10+
- OpenAI API key (atau LLM provider lain yang di-support LangChain)
- Familiar dasar dengan Python async
pip install -U langgraph langchain-openai
Versi yang dipake di tutorial ini: LangGraph 1.2.10 (minimal Python 3.10+).
Langkah 1: Building Block Agent
Setiap LangGraph agent ngikutin loop yang sama: panggil LLM, cek apakah mau pake tool, jalanin tool kalau perlu, dan ulangi. Yuk bikin sebagai komponen yang bisa dipake ulang.
Buat file agent.py:
from typing import Literal
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
# Tool sederhana
def get_weather(location: str) -> str:
"""Get the current weather in a given location."""
return f"Sunny, 72°F in {location}"
tools = [get_weather]
llm = ChatOpenAI(model="gpt-4o-mini")
llm_with_tools = llm.bind_tools(tools)
# Node: panggil LLM
def call_model(state: MessagesState):
response = llm_with_tools.invoke(state["messages"])
return {"messages": [response]}
# Node: jalanin tool
def call_tool(state: MessagesState):
last_msg = state["messages"][-1]
results = []
for tc in last_msg.tool_calls:
if tc["name"] == "get_weather":
result = get_weather(**tc["args"])
results.append({
"role": "tool",
"content": result,
"tool_call_id": tc["id"],
})
return {"messages": results}
# Conditional edge: lanjut kalau ada tool call, berhenti kalau nggak
def should_continue(state: MessagesState) -> Literal["tools", END]:
last_msg = state["messages"][-1]
if last_msg.tool_calls:
return "tools"
return END
# Bangun graph
builder = StateGraph(MessagesState)
builder.add_node("agent", call_model)
builder.add_node("tools", call_tool)
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue)
builder.add_edge("tools", "agent")
graph = builder.compile()
# Jalanin
result = graph.invoke({
"messages": [HumanMessage(content="What is the weather in Jakarta?")]
})
print(result["messages"][-1].content)
Jalankan:
python agent.py
Kamu bakal liat response soal cuaca di Jakarta. Loop agent-nya sudah jalan.
Langkah 2: Supervisor Pattern
Satu agent cukup buat query sederhana. Tapi workflow beneran butuh banyak spesialis. Supervisor pattern solusinya: supervisor agent yang ngecek setiap user request dan routing ke spesialis yang tepat.
Buat supervisor.py:
from typing import Literal, Sequence
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI
# State bersama
class AgentState(TypedDict):
messages: Sequence[BaseMessage]
next: str # agent mana yang harus jalan berikutnya
members = ["researcher", "coder", "reviewer"]
options = members + ["FINISH"]
system_prompt = f"""
You are a supervisor managing the following workers: {members}.
Given the user request, decide which worker should act next.
Each worker will do their task and report back.
Respond with the worker name or "FINISH" when the task is complete.
"""
llm = ChatOpenAI(model="gpt-4o-mini")
# Node supervisor
def supervisor(state: AgentState):
response = llm.invoke([
SystemMessage(content=system_prompt),
] + state["messages"])
return {"next": response.content.strip()}
# Node worker (simplifikasi)
def researcher(state: AgentState):
return {"messages": [
HumanMessage(content="I am researching: " + state["messages"][-1].content)
]}
def coder(state: AgentState):
return {"messages": [
HumanMessage(content="I am coding: " + state["messages"][-1].content)
]}
def reviewer(state: AgentState):
return {"messages": [
HumanMessage(content="I am reviewing: " + state["messages"][-1].content)
]}
# Bangun workflow
builder = StateGraph(AgentState)
builder.add_node("supervisor", supervisor)
builder.add_node("researcher", researcher)
builder.add_node("coder", coder)
builder.add_node("reviewer", reviewer)
# Semua worker lapor balik ke supervisor
for member in members:
builder.add_edge(member, "supervisor")
# Supervisor yang mutusin langkah berikutnya
builder.add_conditional_edges(
"supervisor",
lambda state: state["next"],
{m: m for m in members} | {END: END}
)
builder.add_edge(START, "supervisor")
# Compile dengan memory biar supervisor inget percakapan
graph = builder.compile(checkpointer=MemorySaver())
# Jalanin
config = {"configurable": {"thread_id": "1"}}
result = graph.invoke({
"messages": [HumanMessage(content="Build a REST API that returns weather data from a SQLite database")],
"next": "",
}, config)
for msg in result["messages"]:
print(f"{msg.type}: {msg.content[:80]}...")
Langkah 3: Kenapa Supervisor Pattern Efektif
Supervisor pattern nyelesain tiga masalah yang susah ditangani single-agent:
Management context. Setiap spesialis cuma liat bagian percakapan yang relevan sama tugasnya. Researcher nggak perlu liat import statements punya coder. Supervisor yang megang big picture.
Isolasi tool. Setiap spesialis bawa cuma tool yang dia butuhin. Coder nggak perlu web search tool. Researcher nggak perlu code execution sandbox. Ini ngurangin hallucination karena LLM nggak bisa milih tool yang salah buat job yang salah.
Jalur eskalasi. Kalau tugas mentok di sesuatu yang nggak bisa ditangani spesialis, supervisor bisa routing ke worker lain atau minta klarifikasi. Single agent yang megang semua tool bakal coba ngelakuin semuanya dan hasilnya mediocre.
Langkah 4: Sub-Agent Spesialis dengan Tool Beneran
Sekarang kasih masing-masing spesialis tool yang real. Researcher pake web search (kita pake mock buat contoh ini), coder nulis dan baca file, reviewer ngecek kualitas kode.
Buat specialized_agents.py:
from typing import Literal, Sequence, Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END, add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage, AIMessage
from langchain_openai import ChatOpenAI
# --- Tools ---
def search_web(query: str) -> str:
"""Search the web for information."""
return f"Search results for '{query}': Found 3 relevant articles about best practices."
def write_file(filename: str, content: str) -> str:
"""Write content to a file."""
with open(f"/tmp/{filename}", "w") as f:
f.write(content)
return f"Written {len(content)} bytes to {filename}"
def read_file(filename: str) -> str:
"""Read content from a file."""
try:
with open(f"/tmp/{filename}") as f:
return f.read()
except FileNotFoundError:
return f"File {filename} not found."
def lint_code(code: str) -> str:
"""Check code for common issues."""
issues = []
if "import" not in code:
issues.append("No import statements found")
if len(code.split("\n")) < 5:
issues.append("Code is very short")
if not issues:
issues.append("No obvious issues found")
return "\n".join(issues)
# --- Agents ---
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], add_messages]
next: str
members = ["researcher", "coder", "reviewer"]
system_prompt = f"Route to one of: {members}. Say FINISH when done."
# Bikin masing-masing specialist agent
def make_agent(tools, system_message):
llm = ChatOpenAI(model="gpt-4o-mini")
llm_with_tools = llm.bind_tools(tools)
def agent_node(state: AgentState):
sys_msg = SystemMessage(content=system_message)
response = llm_with_tools.invoke([sys_msg] + list(state["messages"]))
tool_results = []
if hasattr(response, "tool_calls") and response.tool_calls:
for tc in response.tool_calls:
for tool in tools:
if tool.__name__ == tc["name"]:
result = tool(**tc["args"])
tool_results.append({
"role": "tool",
"content": str(result),
"tool_call_id": tc["id"],
})
return {"messages": [response] + tool_results}
return agent_node
researcher_agent = make_agent(
[search_web],
"You are a researcher. Search the web to find information. Report findings clearly."
)
coder_agent = make_agent(
[write_file, read_file],
"You are a Python developer. Write clean, working code. Read files when needed."
)
reviewer_agent = make_agent(
[lint_code],
"You are a code reviewer. Check code quality and suggest improvements."
)
# --- Supervisor dengan structured output ---
from langchain_core.pydantic_v1 import BaseModel, Field
class Router(BaseModel):
next: str = Field(description=f"One of: {options}")
llm_supervisor = ChatOpenAI(model="gpt-4o-mini")
supervisor_llm = llm_supervisor.with_structured_output(Router)
def supervisor(state: AgentState):
response = supervisor_llm.invoke([
SystemMessage(content=system_prompt),
] + list(state["messages"]))
return {"next": response.next}
# Bangun graph
builder = StateGraph(AgentState)
builder.add_node("supervisor", supervisor)
builder.add_node("researcher", researcher_agent)
builder.add_node("coder", coder_agent)
builder.add_node("reviewer", reviewer_agent)
for member in members:
builder.add_edge(member, "supervisor")
builder.add_conditional_edges(
"supervisor",
lambda s: s["next"],
{m: m for m in members} | {END: END}
)
builder.add_edge(START, "supervisor")
graph = builder.compile(checkpointer=MemorySaver())
Versi ini pake structured output (model Router) buat keputusan supervisor, bukan parsing teks biasa. Ini lebih reliable di production karena format output dijamin sama LLM provider.
Langkah 5: Functional API (Alternatif yang Lebih Simpel)
LangGraph juga punya Functional API pake decorator @task dan @entrypoint. Ini lebih simpel buat linear workflow atau star-shaped orchestration di mana supervisor jalan sebelum dan sesudah worker, bukan di sela-sela.
from langgraph.func import entrypoint, task
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini")
@task
def research(query: str) -> str:
"""Find relevant information."""
return f"Research findings for: {query}"
@task
def generate_code(spec: str) -> str:
"""Write code based on the spec and research."""
response = llm.invoke([
HumanMessage(content=f"Write Python code for: {spec}")
])
return response.content
@task
def review(code: str) -> str:
"""Review the generated code."""
response = llm.invoke([
HumanMessage(content=f"Review this code and suggest improvements: {code}")
])
return response.content
@entrypoint()
def build_feature(user_request: str):
findings = research(user_request).result()
code = generate_code(f"{user_request}\n\nResearch: {findings}").result()
feedback = review(code).result()
return {
"findings": findings,
"code": code,
"review": feedback,
}
# Jalanin
result = build_feature.invoke("Build a REST API endpoint for user authentication")
print(result["code"])
Functional API lebih bersih kalau workflow kamu adalah urutan yang sudah diketahui, bukan loop terbuka. Pake ini kalau kamu udah tahu urutan eksekusinya dari awal. Pake Graph API kalau agent perlu mutusin langkah berikutnya berdasarkan hasil.
Kapan Pake LangGraph vs Alternatif
| Tool | Cocok Buat | Kurang Cocok Buat |
|---|---|---|
| LangGraph | Multi-agent stateful; workflow panjang dengan human-in-the-loop; branching kompleks | Single-agent chat sederhana (overkill); prototype cepat yang cukup pake CrewAI |
| CrewAI | Prototype multi-agent cepat; agent role-based dengan delegasi simpel | Kontrol state detail; routing kondisional kompleks; durability production |
| AutoGen | Pattern agent-to-agent conversational; tim yang negosiasi dan debat | Orchestration terawasi; workflow yang butuh persistence dan checkpointing |
| OpenAI Assistants | Single-agent dengan built-in retrieval dan code interpreter | Orchestration multi-agent; custom state management; setup framework-agnostic |
Kekuatan LangGraph ada di kontrol — setiap edge, setiap state transition, setiap keputusan persistence harus eksplisit. Konsekuensinya lebih banyak kode dibanding CrewAI. Tapi kamu bisa debug dengan tepat kenapa suatu agent salah ambil keputusan.
Masalah yang Sering Muncul
Supervisor selalu milih worker yang sama. System prompt kamu perlu kriteria yang lebih jelas. Tambah aturan kayak "kalau user nanya soal kode, routing ke coder. Kalau nanya informasi, routing ke researcher. Kalau nanya kualitas, routing ke reviewer."
Agents kehilangan context antar turn. Pake MemorySaver() sebagai checkpointer dan kirim thread_id yang sama di config. Ini nyimpen percakapan antar invocations.
Tool execution blocking graph. Tools jalan synchronous secara default. Buat task yang lama (API call, file processing), bungkus pake asyncio atau pake decorator @task biar jalan concurrent.
Structured output gagal. Beberapa model handle structured output lebih baik dari yang lain. gpt-4o-mini work dengan baik. gpt-3.5-turbo kadang return JSON yang salah. Test router kamu dengan model yang bakal dipake di production.
Langkah Selanjutnya
- Tambah human-in-the-loop pake mekanisme
interruptLangGraph buat jeda eksekusi menunggu approval - Pake subgraphs buat enkapsulasi setiap specialist agent sebagai graph sendiri — ini bikin testing dan deploy independen
- Tambah persistence pake PostgreSQL atau SQLite checkpointer biar agent survive server restart
- Instrument pake LangSmith buat trace setiap keputusan dan tool call antar agent
Referensi
- Dokumentasi LangGraph — Docs resmi termasuk quickstart, tutorial, dan API reference
- GitHub Repository LangGraph — Source code, contoh, dan issue tracker
- Panduan Quickstart LangGraph — Panduan step-by-step dengan Graph API dan Functional API