Membangun Multi-Agent Systems dengan LangGraph dan Deep Agents: Panduan Lengkap 2026
Pelajari cara membangun sistem multi-agent yang powerful menggunakan LangGraph dan Deep Agents. Tutorial lengkap ini mencakup arsitektur, implementasi step-by-step, best practices, dan contoh kode yang bisa langsung dijalankan untuk mengotomatisasi tugas kompleks.
Membangun Multi-Agent Systems dengan LangGraph dan Deep Agents: Panduan Lengkap 2026
Pendahuluan
Bayangkan Anda memiliki tim asisten virtual yang bekerja sama secara otomatis—satu agen bertugas riset, yang lain menulis laporan, dan yang lain lagi melakukan review. Ini bukan lagi sekadar konsep futuristik, melainkan realitas yang bisa Anda bangun hari ini dengan LangGraph dan Deep Agents.
Multi-agent systems adalah arsitektur di mana beberapa AI agent bekerja bersama untuk menyelesaikan tugas kompleks. Berbeda dengan single agent yang harus menangani segalanya sendiri, multi-agent systems memungkinkan spesialisasi, paralelisme, dan hasil yang lebih baik untuk tugas-tugas yang membutuhkan berbagai keahlian.
LangGraph adalah framework orchestration dari LangChain yang memungkinkan Anda membangun workflow agent yang stateful, durable, dan bisa di-deploy ke production. Sementara Deep Agents adalah "agent harness" yang sudah siap pakai dengan berbagai built-in tools seperti planning, filesystem access, dan subagent spawning.
Mengapa ini penting di 2026?
- GitHub Trending: Repository
langchain-ai/deepagentsmeroket dengan 14,516+ stars dan 1,415 stars baru dalam sehari - Kebutuhan Industri: Perusahaan seperti Klarna, Replit, dan Elastic sudah mengadopsi LangGraph untuk production systems
- Efisiensi: Multi-agent systems bisa mengurangi waktu development dari minggu menjadi jam untuk tugas-tugas kompleks
Prerequisites
Sebelum memulai, pastikan Anda memiliki:
1. Pengetahuan Dasar
- Python 3.10+ (pengalaman dengan async/await akan membantu)
- Pemahaman dasar tentang LLM (Large Language Models) seperti GPT-4, Claude, atau Gemini
- Konsep dasar graph dan state machine
2. Environment Setup
# Buat virtual environment python -m venv venv source venv/bin/activate # Linux/Mac # atau .\\venv\\Scripts\\activate # Windows # Install dependencies utama pip install -U langgraph langchain langchain-openai deepagents # Optional: Untuk development dan debugging pip install langsmith jupyter
3. API Keys
Anda membutuhkan API key dari salah satu LLM provider:
# OpenAI export OPENAI_API_KEY=\"sk-your-key-here\" # Atau untuk Google Gemini export GOOGLE_API_KEY=\"your-key-here\" # Untuk tracing dan debugging (recommended) export LANGSMITH_TRACING=true export LANGSMITH_API_KEY=\"your-langsmith-key\"
Core Concepts: Memahami Fondasi LangGraph
Sebelum terjun ke kode, mari pahami konsep-konsep fundamental yang membuat LangGraph powerful.
1. State - Memory Bersama
State adalah seperti "notepad bersama" yang dibawa oleh seluruh agent dalam sistem. Setiap node (langkah) bisa membaca dan menulis ke state ini.
from typing import TypedDict, Annotated from langgraph.graph import add_messages class AgentState(TypedDict): \\\"\\\"\\\"State yang dibagikan antar semua agent dalam sistem.\\\"\\\"\\\" messages: Annotated[list, add_messages] # History percakapan current_task: str # Tugas yang sedang dikerjakan research_results: str # Hasil riset draft_content: str # Draft yang dihasilkan review_feedback: str # Feedback dari reviewer iterations: int # Counter untuk tracking
Analogi: Bayangkan state seperti whiteboard di ruang meeting. Setiap orang (agent) bisa melihat apa yang sudah ditulis dan menambahkan informasi baru.
2. Nodes - Langkah-langkah Individual
Nodes adalah fungsi Python yang melakukan pekerjaan spesifik. Setiap node menerima state, melakukan sesuatu, dan mengembalikan update ke state.
from langgraph.graph import StateGraph, MessagesState def research_node(state: MessagesState) -> dict: \\\"\\\"\\\"Node untuk melakukan riset topik.\\\"\\\"\\\" # Extract task dari state task = state[\\\"messages\\\"][-1].content # Simulasi riset (dalam praktiknya, gunakan tools sebenarnya) research_result = f\\\"Research completed for: {task}\\\" # Return update ke state return {\\\"messages\\\": [{\\\"role\\\": \\\"assistant\\\", \\\"content\\\": research_result}]} def write_node(state: MessagesState) -> dict: \\\"\\\"\\\"Node untuk menulis konten berdasarkan riset.\\\"\\\"\\\" research = state[\\\"messages\\\"][-1].content draft = f\\\"Based on research: {research}\\\ \\\ Here's the written content...\\\" return {\\\"messages\\\": [{\\\"role\\\": \\\"assistant\\\", \\\"content\\\": draft}]}
3. Edges - Koneksi dan Alur
Edges menentukan kemana flow berlanjut setelah sebuah node selesai. Ada dua jenis:
- Normal Edges: Selalu ke target yang sama
- Conditional Edges: Bercabang berdasarkan kondisi state
from langgraph.graph import START, END # Normal edge - selalu ke target yang sama graph.add_edge(START, \\\"research\\\") graph.add_edge(\\\"research\\\", \\\"write\\\") graph.add_edge(\\\"write\\\", END) # Conditional edge - bercabang berdasarkan kondisi def should_continue(state: MessagesState) -> str: \\\"\\\"\\\"Decide whether to continue or end.\\\"\\\"\\\" if state.get(\\\"iterations\\\", 0) < 3: return \\\"continue\\\" return \\\"end\\\" graph.add_conditional_edges( \\\"write\\\", should_continue, {\\\"continue\\\": \\\"research\\\", \\\"end\\\": END} )
4. StateGraph - Orkestrator Utama
StateGraph adalah container yang menggabungkan semua komponen menjadi satu sistem yang executable.
from langgraph.graph import StateGraph, MessagesState, START, END # Inisialisasi graph dengan state type graph = StateGraph(MessagesState) # Tambahkan nodes graph.add_node(\\\"research\\\", research_node) graph.add_node(\\\"write\\\", write_node) # Definisikan edges (flow) graph.add_edge(START, \\\"research\\\") graph.add_edge(\\\"research\\\", \\\"write\\\") graph.add_edge(\\\"write\\\", END) # Compile menjadi executable app = graph.compile() # Jalankan result = app.invoke({\\\"messages\\\": [{\\\"role\\\": \\\"user\\\", \\\"content\\\": \\\"Write about AI\\\"}]})
Arsitektur Multi-Agent Systems
Mari kita lihat arsitektur lengkap sistem multi-agent yang akan kita bangun:
┌─────────────────────────────────────────────────────────────────────┐ │ USER INPUT │ │ \\\"Research dan tulis artikel\\\" │ └─────────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────────┐ │ ORCHESTRATOR AGENT │ │ - Menganalisis task │ │ - Membuat plan dengan write_todos │ │ - Mendistribusikan ke sub-agents │ └─────────────────────────────────────────────────────────────────────┘ │ ┌───────────────┼───────────────┐ ▼ ▼ ▼ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ RESEARCH │ │ WRITER │ │ REVIEWER │ │ AGENT │ │ AGENT │ │ AGENT │ ├──────────────┤ ├──────────────┤ ├──────────────┤ │ - Web search │ │ - Draft │ │ - Quality │ │ - Doc search │ │ - Edit │ │ check │ │ - Summarize │ │ - Format │ │ - Feedback │ └──────────────┘ └──────────────┘ └──────────────┘ │ │ │ └───────────────┼───────────────┘ ▼ ┌─────────────────────────────────────────────────────────────────────┐ │ SHARED STATE │ │ - messages: conversation history │ │ - research_results: collected data │ │ - draft_content: written content │ │ - review_feedback: improvement suggestions │ └─────────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────────┐ │ OUTPUT │ │ Final polished content with sources │ └─────────────────────────────────────────────────────────────────────┘
Step-by-Step Implementation
Step 1: Setup Project Structure
Buat struktur folder yang terorganisir:
multi_agent_project/ ├── agents/ │ ├── __init__.py │ ├── orchestrator.py │ ├── researcher.py │ ├── writer.py │ └── reviewer.py ├── tools/ │ ├── __init__.py │ ├── search_tools.py │ └── file_tools.py ├── graph/ │ ├── __init__.py │ └── workflow.py ├── config.py ├── main.py └── requirements.txt
Step 2: Define State dan Configuration
Buat file config.py untuk konfigurasi:
# config.py import os from typing import TypedDict, Annotated, Sequence from langchain_core.messages import BaseMessage from langgraph.graph import add_messages class MultiAgentState(TypedDict): \\\"\\\"\\\" State yang dibagikan antar semua agents dalam sistem. Attributes: messages: History percakapan dengan add_messages reducer task: Tugas utama dari user research_data: Data hasil riset draft: Draft konten feedback: Feedback dari reviewer iterations: Jumlah iterasi yang sudah dilakukan final_output: Output final yang disetujui \\\"\\\"\\\" messages: Annotated[Sequence[BaseMessage], add_messages] task: str research_data: str draft: str feedback: str iterations: int final_output: str # Configuration class Config: \\\"\\\"\\\"Configuration untuk agents dan LLM.\\\"\\\"\\\" MODEL_NAME = os.getenv(\\\"MODEL_NAME\\\", \\\"gpt-4o\\\") MAX_ITERATIONS = 3 TEMPERATURE = 0.7 # Prompts ORCHESTRATOR_PROMPT = \\\"\\\"\\\"Anda adalah orchestrator yang mengkoordinasi tim agents. Tugas Anda adalah memecah request user menjadi sub-tasks dan mendistribusikannya ke agent yang tepat. Gunakan format yang jelas dan structured.\\\"\\\"\\\" RESEARCHER_PROMPT = \\\"\\\"\\\"Anda adalah research specialist. Tugas Anda adalah mencari informasi yang relevan dan akurat. Selalu sertakan sumber.\\\"\\\"\\\" WRITER_PROMPT = \\\"\\\"\\\"Anda adalah content writer profesional. Tugas Anda adalah menulis konten yang engaging dan informatif berdasarkan research yang diberikan.\\\"\\\"\\\" REVIEWER_PROMPT = \\\"\\\"\\\"Anda adalah quality assurance specialist. Tugas Anda adalah mereview konten dan memberikan feedback konstruktif untuk perbaikan.\\\"\\\"\\\"
Step 3: Build Tools untuk Agents
Buat file tools/search_tools.py:
# tools/search_tools.py from langchain_core.tools import tool from typing import Optional import json @tool def web_search(query: str, num_results: int = 5) -> str: \\\"\\\"\\\" Melakukan pencarian web untuk mendapatkan informasi terkini. Args: query: Query pencarian num_results: Jumlah hasil yang diinginkan (default: 5) Returns: String berisi hasil pencarian dengan URL dan snippet \\\"\\\"\\\" # Dalam praktiknya, gunakan API seperti Serper, Tavily, atau Brave Search # Ini adalah simulasi untuk demonstrasi # Simulasi hasil pencarian mock_results = [ { \\\"title\\\": f\\\"Understanding {query} - Complete Guide\\\", \\\"url\\\": \\\"https://example.com/guide\\\", \\\"snippet\\\": f\\\"Comprehensive guide about {query} with examples...\\\" }, { \\\"title\\\": f\\\"Best Practices for {query}\\\", \\\"url\\\": \\\"https://example.com/best-practices\\\", \\\"snippet\\\": f\\\"Industry best practices and recommendations...\\\" } ] results = [] for i, result in enumerate(mock_results[:num_results]): results.append( f\\\"{i+1}. {result['title']}\\\ \\\" f\\\" URL: {result['url']}\\\ \\\" f\\\" Summary: {result['snippet']}\\\ \\\" ) return \\\"\\\ \\\".join(results) @tool def scrape_webpage(url: str) -> str: \\\"\\\"\\\" Mengambil konten dari sebuah webpage. Args: url: URL webpage yang akan di-scrape Returns: Konten text dari webpage \\\"\\\"\\\" # Dalam praktiknya, gunakan library seperti BeautifulSoup atau Playwright return f\\\"Content from {url}: [Extracted article content would appear here...]\\\" # tools/file_tools.py from langchain_core.tools import tool import os @tool def read_file(filepath: str) -> str: \\\"\\\"\\\" Membaca konten dari file lokal. Args: filepath: Path ke file yang akan dibaca Returns: Konten file sebagai string \\\"\\\"\\\" try: with open(filepath, 'r', encoding='utf-8') as f: return f.read() except FileNotFoundError: return f\\\"Error: File not found at {filepath}\\\" except Exception as e: return f\\\"Error reading file: {str(e)}\\\" @tool def write_file(filepath: str, content: str) -> str: \\\"\\\"\\\" Menulis konten ke file lokal. Args: filepath: Path ke file tujuan content: Konten yang akan ditulis Returns: Status operasi \\\"\\\"\\\" try: # Create directory if not exists os.makedirs(os.path.dirname(filepath), exist_ok=True) with open(filepath, 'w', encoding='utf-8') as f: f.write(content) return f\\\"Successfully wrote to {filepath}\\\" except Exception as e: return f\\\"Error writing file: {str(e)}\\\" @tool def list_files(directory: str) -> str: \\\"\\\"\\\" Menampilkan daftar file dalam directory. Args: directory: Path ke directory Returns: Daftar file dalam directory \\\"\\\"\\\" try: files = os.listdir(directory) return \\\"\\\ \\\".join(files) except Exception as e: return f\\\"Error listing files: {str(e)}\\\"
Step 4: Build Individual Agents
Buat file agents/researcher.py:
# agents/researcher.py from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage from langchain_core.prompts import ChatPromptTemplate from tools.search_tools import web_search, scrape_webpage def create_research_agent(): \\\"\\\"\\\" Membuat research agent yang bertugas mengumpulkan informasi. Returns: Tuple of (llm, tools) untuk research agent \\\"\\\"\\\" llm = ChatOpenAI( model=\\\"gpt-4o\\\", temperature=0.3 # Lower temperature untuk factual accuracy ) tools = [web_search, scrape_webpage] system_prompt = \\\"\\\"\\\"Anda adalah Research Specialist dengan keahlian: 1. Mencari informasi dari berbagai sumber 2. Memvalidasi kredibilitas sumber 3. Menyintesis informasi dari multiple sources 4. Mendokumentasikan sumber dengan benar Saat melakukan riset: - Gunakan web_search untuk menemukan sumber relevan - Scrape halaman untuk detail lebih lanjut - Kategorikan temuan Anda - Selalu sertakan URL sumber - Tandai jika informasi tidak pasti Format output: ## Temuan Utama [Key findings] ## Detail [Detailed information] ## Sumber - [URL 1] - [URL 2] \\\"\\\"\\\" return llm.bind_tools(tools), tools, system_prompt def research_node(state): \\\"\\\"\\\" Node function untuk research agent. Args: state: Current state dengan task dan messages Returns: Updated state dengan research_data \\\"\\\"\\\" llm, tools, system_prompt = create_research_agent() # Get task from state task = state.get(\\\"task\\\", \\\"\\\") messages = state.get(\\\"messages\\\", []) # Create prompt research_prompt = f\\\"\\\"\\\"Research task: {task} Lakukan riset mendalam dan kumpulkan informasi yang relevan. Fokus pada: 1. Definisi dan konsep dasar 2. Best practices dan implementasi 3. Contoh nyata dan case studies 4. Tren dan perkembangan terkini \\\"\\\"\\\" # Invoke LLM with tools response = llm.invoke([ SystemMessage(content=system_prompt), HumanMessage(content=research_prompt) ]) # Extract research data research_data = response.content return { \\\"research_data\\\": research_data, \\\"messages\\\": [response] }
Buat file agents/writer.py:
# agents/writer.py from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage def create_writer_agent(): \\\"\\\"\\\" Membuat writer agent yang bertugas menulis konten. Returns: LLM configured for writing \\\"\\\"\\\" llm = ChatOpenAI( model=\\\"gpt-4o\\\", temperature=0.7 # Higher temperature untuk creativity ) system_prompt = \\\"\\\"\\\"Anda adalah Content Writer profesional dengan keahlian: 1. Menulis konten yang engaging dan mudah dipahami 2. Menyusun struktur artikel yang logis 3. Mengadaptasi tone sesuai audience 4. SEO-friendly writing Panduan penulisan: - Gunakan heading dan subheading yang jelas - Sertakan contoh konkret - Gunakan bullet points untuk readability - Hindari jargon berlebihan - Selalu sertakan call-to-action Format artikel: # [Title] ## Introduction [Hook dan overview] ## [Main Section 1] [Content] ## [Main Section 2] [Content] ## Conclusion [Summary dan next steps] \\\"\\\"\\\" return llm, system_prompt def writer_node(state): \\\"\\\"\\\" Node function untuk writer agent. Args: state: Current state dengan research_data Returns: Updated state dengan draft \\\"\\\"\\\" llm, system_prompt = create_writer_agent() # Get research data from state research_data = state.get(\\\"research_data\\\", \\\"\\\") task = state.get(\\\"task\\\", \\\"\\\") feedback = state.get(\\\"feedback\\\", \\\"\\\") iterations = state.get(\\\"iterations\\\", 0) # Build prompt if iterations > 0 and feedback: # Revision mode prompt = f\\\"\\\"\\\"Berdasarkan feedback berikut, revisi draft artikel: FEEDBACK: {feedback} DRAFT SEBELUMNYA: {state.get(\\\"draft\\\", \\\"\\\")} RESEARCH DATA: {research_data} Lakukan perbaikan sesuai feedback sambil mempertahankan kualitas konten. \\\"\\\"\\\" else: # Initial writing mode prompt = f\\\"\\\"\\\"Berdasarkan research data berikut, tulis artikel yang komprehensif: TASK: {task} RESEARCH DATA: {research_data} Tulis artikel dalam Bahasa Indonesia yang informatif dan engaging. \\\"\\\"\\\" response = llm.invoke([ SystemMessage(content=system_prompt), HumanMessage(content=prompt) ]) return { \\\"draft\\\": response.content, \\\"messages\\\": [response], \\\"iterations\\\": iterations + 1 }
Buat file agents/reviewer.py:
# agents/reviewer.py from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage def create_reviewer_agent(): \\\"\\\"\\\" Membuat reviewer agent untuk quality assurance. Returns: LLM configured for reviewing \\\"\\\"\\\" llm = ChatOpenAI( model=\\\"gpt-4o\\\", temperature=0.2 # Low temperature untuk consistency ) system_prompt = \\\"\\\"\\\"Anda adalah Quality Assurance Specialist dengan keahlian: 1. Mengevaluasi akurasi dan kelengkapan konten 2. Memastikan kualitas penulisan 3. Mengidentifikasi area yang perlu improvement 4. Memberikan feedback konstruktif Kriteria review: - Akurasi informasi - Kejelasan struktur - Grammar dan spelling - Engagement dan readability - Kelengkapan konten Format feedback: ## Overall Assessment [Score: X/10 dan summary] ## Strengths - [Point 1] - [Point 2] ## Areas for Improvement - [Issue]: [Suggestion] - [Issue]: [Suggestion] ## Final Recommendation [APPROVED / NEEDS_REVISION] \\\"\\\"\\\" return llm, system_prompt def reviewer_node(state): \\\"\\\"\\\" Node function untuk reviewer agent. Args: state: Current state dengan draft Returns: Updated state dengan feedback dan decision \\\"\\\"\\\" llm, system_prompt = create_reviewer_agent() draft = state.get(\\\"draft\\\", \\\"\\\") research_data = state.get(\\\"research_data\\\", \\\"\\\") iterations = state.get(\\\"iterations\\\", 1) prompt = f\\\"\\\"\\\"Review artikel berikut: DRAFT: {draft} RESEARCH DATA (untuk cross-check): {research_data} Lakukan review menyeluruh dan berikan feedback. \\\"\\\"\\\" response = llm.invoke([ SystemMessage(content=system_prompt), HumanMessage(content=prompt) ]) feedback = response.content # Determine if approved is_approved = \\\"APPROVED\\\" in feedback.upper() and iterations >= 2 return { \\\"feedback\\\": feedback, \\\"messages\\\": [response], \\\"final_output\\\": draft if is_approved else None }
Step 5: Build the Orchestration Graph
Buat file graph/workflow.py:
# graph/workflow.py from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver from agents.researcher import research_node from agents.writer import writer_node from agents.reviewer import reviewer_node from config import MultiAgentState, Config def should_continue(state: MultiAgentState) -> str: \\\"\\\"\\\" Conditional edge untuk menentukan apakah perlu revisi. Args: state: Current state Returns: \\\"approved\\\" jika sudah OK, \\\"revise\\\" jika perlu revisi \\\"\\\"\\\" # Check if approved in feedback feedback = state.get(\\\"feedback\\\", \\\"\\\") iterations = state.get(\\\"iterations\\\", 0) # Approved jika: # 1. Reviewer sudah approve # 2. Sudah minimal 2 iterasi (ensure quality) if \\\"APPROVED\\\" in feedback.upper(): if iterations >= Config.MAX_ITERATIONS: return \\\"approved\\\" return \\\"approved\\\" # Max iterations reached - force approve if iterations >= Config.MAX_ITERATIONS: return \\\"approved\\\" return \\\"revise\\\" def create_multi_agent_workflow(): \\\"\\\"\\\" Membuat dan meng-compile workflow graph untuk multi-agent system. Returns: Compiled LangGraph application \\\"\\\"\\\" # Initialize graph workflow = StateGraph(MultiAgentState) # Add nodes workflow.add_node(\\\"research\\\", research_node) workflow.add_node(\\\"writer\\\", writer_node) workflow.add_node(\\\"reviewer\\\", reviewer_node) # Define edges - linear flow first workflow.set_entry_point(\\\"research\\\") workflow.add_edge(\\\"research\\\", \\\"writer\\\") workflow.add_edge(\\\"writer\\\", \\\"reviewer\\\") # Conditional edge after review workflow.add_conditional_edges( \\\"reviewer\\\", should_continue, { \\\"approved\\\": END, \\\"revise\\\": \\\"writer\\\" # Loop back for revision } ) # Compile with memory saver for persistence memory = MemorySaver() app = workflow.compile(checkpointer=memory) return app # Convenience function untuk menjalankan workflow def run_research_workflow(task: str, thread_id: str = \\\"default\\\") -> dict: \\\"\\\"\\\" Menjalankan multi-agent workflow untuk research dan writing. Args: task: Tugas yang akan dikerjakan thread_id: Thread ID untuk session (untuk persistence) Returns: Dictionary dengan hasil workflow \\\"\\\"\\\" app = create_multi_agent_workflow() # Initial state initial_state = { \\\"task\\\": task, \\\"messages\\\": [], \\\"research_data\\\": \\\"\\\", \\\"draft\\\": \\\"\\\", \\\"feedback\\\": \\\"\\\", \\\"iterations\\\": 0, \\\"final_output\\\": \\\"\\\" } # Configuration for this run config = {\\\"configurable\\\": {\\\"thread_id\\\": thread_id}} # Run workflow result = app.invoke(initial_state, config) return result
Step 6: Build Main Application
Buat file main.py:
#!/usr/bin/env python3 \\\"\\\"\\\" Multi-Agent Research & Writing System ===================================== Aplikasi yang menggunakan LangGraph untuk menjalankan multi-agent workflow untuk research dan content writing. Usage: python main.py \\\"Research tentang AI Agents dan tulis artikelnya\\\" \\\"\\\"\\\" import sys import asyncio from graph.workflow import run_research_workflow, create_multi_agent_workflow from langgraph.pregel import GraphRecursionError def main(): \\\"\\\"\\\"Main entry point.\\\"\\\"\\\" # Get task from command line or use default if len(sys.argv) > 1: task = \\\" \\\".join(sys.argv[1:]) else: task = \\\"Research tentang perkembangan AI Agents di 2026 dan tulis artikel lengkap dalam Bahasa Indonesia\\\" print(\\\"=\\\"*60) print(\\\"Multi-Agent Research & Writing System\\\") print(\\\"=\\\"*60) print(f\\\"\\\ Task: {task}\\\ \\\") print(\\\"Starting workflow...\\\ \\\") try: # Run the workflow result = run_research_workflow(task, thread_id=\\\"session-001\\\") # Print results print(\\\"\\\ \\\" + \\\"=\\\"*60) print(\\\"HASIL AKHIR\\\") print(\\\"=\\\"*60) if result.get(\\\"final_output\\\"): print(result[\\\"final_output\\\"]) else: print(result.get(\\\"draft\\\", \\\"No output generated\\\")) print(\\\"\\\ \\\" + \\\"=\\\"*60) print(\\\"STATISTICS\\\") print(\\\"=\\\"*60) print(f\\\"Iterations: {result.get('iterations', 0)}\\\") print(f\\\"Final Status: {'Approved' if result.get('final_output') else 'Needs Review'}\\\") except GraphRecursionError as e: print(f\\\"Error: Maximum recursion limit reached. {e}\\\") except Exception as e: print(f\\\"Error running workflow: {e}\\\") raise async def stream_workflow_example(): \\\"\\\"\\\" Example of streaming workflow execution. Berguna untuk real-time updates di UI. \\\"\\\"\\\" app = create_multi_agent_workflow() initial_state = { \\\"task\\\": \\\"Write about LangGraph multi-agent systems\\\", \\\"messages\\\": [], \\\"research_data\\\": \\\"\\\", \\\"draft\\\": \\\"\\\", \\\"feedback\\\": \\\"\\\", \\\"iterations\\\": 0, \\\"final_output\\\": \\\"\\\" } config = {\\\"configurable\\\": {\\\"thread_id\\\": \\\"stream-demo\\\"}} print(\\\"Streaming workflow execution:\\\ \\\") async for event in app.astream_events(initial_state, config): kind = event[\\\"event\\\"] if kind == \\\"on_chain_start\\\": print(f\\\"Starting: {event['name']}\\\") elif kind == \\\"on_chain_end\\\": print(f\\\"Completed: {event['name']}\\\") elif kind == \\\"on_llm_stream\\\": # Stream token output content = event[\\\"data\\\"][\\\"chunk\\\"].content if content: print(content, end=\\\"\\\", flush=True) if __name__ == \\\"__main__\\\": main()
Step 7: Using Deep Agents (Alternative yang Lebih Cepat)
Untuk development yang lebih cepat, gunakan Deep Agents SDK:
#!/usr/bin/env python3 \\\"\\\"\\\" Deep Agents Example - Simpler approach dengan built-in capabilities \\\"\\\"\\\" from deepagents import create_deep_agent from langchain_openai import ChatOpenAI # Custom tools def research_tool(topic: str) -> str: \\\"\\\"\\\"Research a topic and return findings.\\\"\\\"\\\" # Implementasi research logic return f\\\"Research findings for: {topic}\\\" def write_tool(content: str, style: str = \\\"professional\\\") -> str: \\\"\\\"\\\"Write content in specified style.\\\"\\\"\\\" return f\\\"Written content ({style}): {content}\\\" # Create deep agent dengan custom tools agent = create_deep_agent( model=ChatOpenAI(model=\\\"gpt-4o\\\"), tools=[research_tool, write_tool], system_prompt=\\\"\\\"\\\"Anda adalah AI assistant yang bisa melakukan research dan writing. Gunakan tools yang tersedia untuk menyelesaikan tugas.\\\"\\\"\\\" ) # Jalankan result = agent.invoke({ \\\"messages\\\": [{ \\\"role\\\": \\\"user\\\", \\\"content\\\": \\\"Research tentang LangGraph dan tulis summary dalam Bahasa Indonesia\\\" }] }) print(result[\\\"messages\\\"][-1].content)
Best Practices dari Industry Experts
1. State Design Principles
DO:
# TypedDict dengan field yang jelas class GoodState(TypedDict): messages: Annotated[list, add_messages] task: str results: dict[str, str] # Namespace untuk clarity
DON'T:
# Menggunakan dict biasa tanpa type hints state = {} # Error-prone dan tidak self-documenting
2. Tool Design
# GOOD: Tool dengan docstring yang jelas @tool def search_database(query: str, limit: int = 10) -> str: \\\"\\\"\\\" Search the company database for relevant records. Args: query: SQL-like query string limit: Maximum number of results (default: 10) Returns: JSON string with matching records Example: search_database(\\\"customers WHERE region='APAC'\\\", limit=5) \\\"\\\"\\\" # Implementation pass
3. Error Handling
def safe_node(state: State) -> dict: \\\"\\\"\\\"Node dengan error handling yang proper.\\\"\\\"\\\" try: # Main logic result = perform_operation(state) return {\\\"result\\\": result, \\\"error\\\": None} except SpecificError as e: # Handle specific errors return {\\\"result\\\": None, \\\"error\\\": f\\\"Specific error: {e}\\\"} except Exception as e: # Catch-all untuk unexpected errors return {\\\"result\\\": None, \\\"error\\\": f\\\"Unexpected error: {e}\\\"}
4. Streaming untuk UX yang Lebih Baik
async def stream_response(app, task: str): \\\"\\\"\\\"Stream output untuk real-time feedback.\\\"\\\"\\\" config = {\\\"configurable\\\": {\\\"thread_id\\\": \\\"stream\\\"}} async for chunk in app.astream({\\\"task\\\": task}, config): for node_name, node_output in chunk.items(): print(f\\\"[{node_name}] {node_output}\\\")
5. Persistence dengan Checkpointer
from langgraph.checkpoint.sqlite import SqliteSaver # SQLite untuk production persistence with SqliteSaver.from_conn_string(\\\"checkpoints.db\\\") as checkpointer: app = workflow.compile(checkpointer=checkpointer) # Resume dari checkpoint result = app.invoke( {\\\"task\\\": \\\"Continue previous task\\\"}, config={\\\"configurable\\\": {\\\"thread_id\\\": \\\"session-123\\\"}} )
Common Mistakes dan Cara Menghindarinya
Mistake 1: State Terlalu Besar
Problem: Menyimpan semua data di state menyebabkan memory issues dan context window overflow.
Solution:
# WRONG class BadState(TypedDict): all_documents: list[str] # Bisa ribuan dokumen! # RIGHT class GoodState(TypedDict): document_summaries: list[str] # Summary saja document_paths: list[str] # Path untuk lazy loading
Mistake 2: Infinite Loops
Problem: Conditional edges tanpa proper termination condition.
Solution:
def should_continue(state: State) -> str: # SELALU punya exit condition if state[\\\"iterations\\\"] >= MAX_ITERATIONS: return \\\"end\\\" if state[\\\"is_complete\\\"]: return \\\"end\\\" return \\\"continue\\\"
Mistake 3: Tidak Ada Timeout
Problem: Agent bisa hang indefinitely.
Solution:
# Set timeout pada invocation result = app.invoke( initial_state, config={ \\\"configurable\\\": {\\\"thread_id\\\": \\\"123\\\"}, \\\"recursion_limit\\\": 50 # Max steps } )
Mistake 4: Hardcoded Prompts
Problem: Prompts di-hardcode sulit di-maintain.
Solution:
# Gunakan file terpisah atau configuration PROMPTS = { \\\"researcher\\\": \\\"\\\"\\\"...\\\"\\\"\\\", \\\"writer\\\": \\\"\\\"\\\"...\\\"\\\"\\\", } # Atau load dari file import yaml with open(\\\"prompts.yaml\\\") as f: PROMPTS = yaml.safe_load(f)
Advanced Tips
1. Subgraph untuk Modularity
from langgraph.graph import StateGraph def create_research_subgraph(): \\\"\\\"\\\"Subgraph untuk research workflow.\\\"\\\"\\\" subgraph = StateGraph(ResearchState) subgraph.add_node(\\\"search\\\", search_node) subgraph.add_node(\\\"summarize\\\", summarize_node) subgraph.add_edge(\\\"search\\\", \\\"summarize\\\") return subgraph.compile() # Gunakan sebagai node dalam graph utama main_graph.add_node(\\\"research_subgraph\\\", create_research_subgraph())
2. Human-in-the-Loop
from langgraph.types import interrupt def human_review_node(state: State) -> dict: \\\"\\\"\\\"Node untuk mendapatkan approval dari human.\\\"\\\"\\\" # Interrupt execution dan tunggu human input approval = interrupt({ \\\"question\\\": \\\"Approve this draft?\\\", \\\"draft\\\": state[\\\"draft\\\"] }) return {\\\"approved\\\": approval == \\\"yes\\\"}
3. Parallel Execution
from langgraph.graph import StateGraph # Multiple edges dari satu node = parallel execution workflow.add_edge(\\\"research\\\", \\\"writer_1\\\") workflow.add_edge(\\\"research\\\", \\\"writer_2\\\") workflow.add_edge(\\\"research\\\", \\\"writer_3\\\")
4. Memory Management
from langgraph.store.memory import InMemoryStore store = InMemoryStore() # Save ke long-term memory store.put( namespace=(\\\"user\\\", \\\"123\\\", \\\"preferences\\\"), # Namespaced key key=\\\"writing_style\\\", value={\\\"tone\\\": \\\"professional\\\", \\\"language\\\": \\\"id\\\"} ) # Retrieve dari memory prefs = store.get((\\\"user\\\", \\\"123\\\", \\\"preferences\\\"), \\\"writing_style\\\")
Summary dan Next Steps
Apa yang Sudah Dipelajari
- Core Concepts: State, Nodes, Edges, dan StateGraph sebagai building blocks
- Architecture: Multi-agent system dengan orchestrator, researcher, writer, dan reviewer
- Implementation: Step-by-step pembuatan workflow lengkap
- Best Practices: State design, tool design, error handling, dan streaming
- Common Mistakes: Infinite loops, state bloat, missing timeouts
- Advanced Features: Subgraphs, human-in-the-loop, parallel execution, memory
Next Steps untuk Mendalami
-
Deploy ke Production
- Gunakan LangSmith untuk observability
- Deploy dengan LangGraph Cloud atau self-hosted
-
Integrasi dengan Tools Lain
- MCP (Model Context Protocol) untuk tool integration
- RAG dengan vector databases
- External APIs dan webhooks
-
Optimisasi Performance
- Caching dengan Redis
- Parallel processing dengan async
- Cost optimization dengan model selection
-
Advanced Patterns
- Hierarchical agents
- Agent swarms
- Reflection dan self-improvement loops