LangGraph4j は、ステートフルでグラフベースの AI アプリケーションを構築するための Java ライブラリです。グラフは、処理を実行するノード、次のノードを選択するエッジ、そしてグラフの実行に伴って変化する共有状態で構成されます。LangGraph に着想を得ており、LangChain4j および Spring AI と統合できますが、グラフノードには通常の Java コードを使用することもできます。
+------------+
| call model |
+------+-----+
|
tool requested?
/ \
yes no
/ \
+-------v-------+ +--v--+
| execute tool | | END |
+-------+-------+ +-----+
|
+----------> call modelサイクルは意図的なものです。サイクルによって、アプリケーションは推論し、行動し、結果を検査し、再び推論できます。LangGraph4j はこの制御フローと状態を明示化し、さらにリデューサー、条件付きルーティング、チェックポイント、ブレークポイント、ストリーミング、グラフ可視化、サブグラフを提供します。
このチュートリアルでは LangGraph4j 1.8.24 と Java 17 を使用します。特定のモデルプロバイダーや Web フレームワークに依存しないよう、コアモジュールに焦点を当てます。
LangGraph4j が解決する課題 #
小規模な AI 機能は、単一のリクエストで実装できる場合があります。
request -> prompt -> model -> responseエージェント型アプリケーションでは、次の処理が必要になることがあります。
- リクエストを分類する。
- 非公開ドキュメントを取得する。
- モデルにツールを選択させる。
- そのツールを検証して実行する。
- 回復可能な失敗を再試行する。
- リスクの高い操作の前に一時停止する。
- リクエストをまたいで進捗を保持する。
- 同じ実行を再開する。
- 処理中の進捗を報告する。
ループ、条件分岐、キュー、データベーステーブルを使えば、グラフライブラリなしでも実装できます。難しいのは、モデルの判断が非決定的であっても、実行を理解可能、割り込み可能、回復可能、かつ安全にすることです。
LangGraph4j は実行モデルを提供しますが、ワークフロー、プロンプト、ツール、認可ルール、ビジネスポリシーを選択するものではありません。これらは引き続きアプリケーションの責務です。
グラフは DAG に限定されない #
有向非巡回グラフ(DAG)は、一方向のパイプラインに適しています。
load -> transform -> validate -> publishエージェントループは通常、循環します。
model -> tool -> model -> tool -> model -> answerLangGraph4j は両方の形式をサポートします。グラフは分岐、ループ、独立した分岐の並列実行、結果の結合、サブグラフの呼び出し、設定済みノードでの一時停止、終了を行えます。ルーティングは決定論的な Java コードでも、モデル判断の検証済み結果でも実装できます。
ランタイムは Pregel に似たスーパーステップモデルに従います。1 つのステップで有効化されたノードは並行して実行され、その後、更新がマージされてから次の処理が実行されます。そのため、共有状態のマージ規則が重要になります。
コア概念 #
状態 #
AgentState は、グラフ実行全体で引き渡される共有データです。Map<String, Object> をラップし、通常は型付きアクセサーを追加して拡張します。
import java.util.List;
import java.util.Map;
import org.bsc.langgraph4j.state.AgentState;
final class SupportState extends AgentState {
static final String REQUEST = "request";
static final String DOCUMENTS = "documents";
static final String PROPOSED_ACTION = "proposed_action";
static final String APPROVED = "approved";
static final String ATTEMPTS = "attempts";
static final String ANSWER = "answer";
SupportState(Map<String, Object> data) {
super(data);
}
String request() {
return value(REQUEST).map(Object::toString).orElse("");
}
List<String> documents() {
return this.<List<String>>value(DOCUMENTS).orElse(List.of());
}
boolean approved() {
return this.<Boolean>value(APPROVED).orElse(false);
}
}ノードは完全な状態オブジェクトを置き換えるのではなく、更新を返します。
Map<String, Object> incrementAttempt(SupportState state) {
int attempts = state.<Integer>value(SupportState.ATTEMPTS).orElse(0);
return Map.of(SupportState.ATTEMPTS, attempts + 1);
}状態は単なるチャット履歴ではありません。判断、識別子、成果物、ステータスを名前付きフィールドに保持すると、文章メッセージから復元するよりも検証しやすくなります。
オープンな接続、エグゼキューター、リクエストオブジェクト、シークレットをチェックポイント対象の状態に格納しないでください。チェックポイントはシリアライズ可能である必要があり、オペレーターによって検査されたり、プロセス外に永続化されたりする可能性があります。
チャネルとリデューサー #
状態スキーマはキーを Channel<?> に対応付けます。チャネルはキーのデフォルト値と、更新をマージする方法を定義します。ベースチャネルでは、新しい値が古い値を置き換えます。Channels.appender は値をリストに蓄積します。
import java.util.ArrayList;
import java.util.Map;
import org.bsc.langgraph4j.state.Channel;
import org.bsc.langgraph4j.state.Channels;
static final Map<String, Channel<?>> SCHEMA = Map.of(
"status", Channels.base(() -> "new"),
"audit", Channels.appender(ArrayList::new));あるノードが Map.of("audit", "classified") を返し、後続ノードが Map.of("audit", "retrieved") を返す場合、appender チャネルは両方のエントリを保持します。標準の appender は重複を許可しません。重複する値に意味がある場合は、Channels.appenderWithDuplicate を使用してください。
並列分岐が同じキーに書き込む場合、リデューサーは不可欠です。チェックポイントからの回復やリプレイで同じ状態遷移が再適用される可能性があるため、リデューサーは決定論的で副作用を持たないものにしてください。
チャットメッセージには、メッセージ編集を単純なリスト連結として扱うのではなく、適切な LangChain4j または Spring AI 統合が提供するメッセージチャネル機能を使用してください。
ノード #
ノードは現在の状態を受け取り、更新のマップを返します。モデルの呼び出し、ベクトルストアへのクエリ、ツールの呼び出し、データの検証、監査エントリの出力、サブグラフの呼び出しを行えます。
NodeAction<S> は同期処理です。AsyncNodeAction<S> は CompletableFuture を返すため、ノンブロッキングなモデル、データベース、HTTP クライアントに適しています。node_async アダプターは、同期アクションを非同期ノードアクションに変換します。
ノードは小さく保ってください。分類、取得、複数ツールの呼び出し、行の書き込み、回答生成を 1 つのノードに詰め込むと、ワークフローが 1 つのメソッドに隠れ、チェックポイントや失敗の解釈が難しくなります。
エッジ #
通常のエッジは常に同じ宛先を選択します。
graph.addEdge("retrieve", "generate");条件付きエッジは状態を評価し、ルーティングラベルを返します。グラフはそのラベルを宣言済みの宛先に対応付けます。
import static org.bsc.langgraph4j.action.AsyncEdgeAction.edge_async;
graph.addConditionalEdges(
"review",
edge_async(state -> state.approved() ? "publish" : "reject"),
Map.of("publish", "publish", "reject", "reject"));この対応付けは許可リストです。モデルが生成した文字列を、制限のない宛先や関数名として使用しないでください。モデル出力を検証し、受け入れる判断の小さな集合をグラフノードに対応付けます。
StateGraph.START と StateGraph.END は、グラフの開始と終了を示します。ノードとエッジを定義した後に compile() を呼び出してください。コンパイルによって構造が検証され、実行可能な CompiledGraph が生成されます。
LangGraph4j のインストール #
LangGraph4j 1.8.x には Java 17 以降が必要です。コアモジュールとオプションモジュールのバージョンを揃えるため、BOM をインポートします。
<properties>
<langgraph4j.version>1.8.24</langgraph4j.version>
</properties>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.bsc.langgraph4j</groupId>
<artifactId>langgraph4j-bom</artifactId>
<version>${langgraph4j.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<dependencies>
<dependency>
<groupId>org.bsc.langgraph4j</groupId>
<artifactId>langgraph4j-core</artifactId>
</dependency>
</dependencies>統合モジュールは、アプリケーションで必要になった場合にのみ追加してください。LangGraph4j は、プロバイダー SDK、LangChain4j、Spring AI、データベースクライアント、通常のアプリケーションコードを直接使用するノードをオーケストレーションできます。
完全な Java グラフ #
次の決定論的なグラフは、リリース告知を下書きし、レビューし、公開または却下します。単純な下書きロジックによってグラフの仕組みを見やすくしています。実際のコードでは、下書きノードをモデル呼び出しに置き換えられます。
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import org.bsc.langgraph4j.GraphStateException;
import org.bsc.langgraph4j.StateGraph;
import org.bsc.langgraph4j.action.NodeAction;
import org.bsc.langgraph4j.state.AgentState;
import org.bsc.langgraph4j.state.Channel;
import org.bsc.langgraph4j.state.Channels;
import static org.bsc.langgraph4j.StateGraph.END;
import static org.bsc.langgraph4j.StateGraph.START;
import static org.bsc.langgraph4j.action.AsyncEdgeAction.edge_async;
import static org.bsc.langgraph4j.action.AsyncNodeAction.node_async;
final class ReleaseState extends AgentState {
static final String RELEASE_ID = "release_id";
static final String NOTES = "notes";
static final String DRAFT = "draft";
static final String APPROVED = "approved";
static final String STATUS = "status";
static final String AUDIT = "audit";
static final Map<String, Channel<?>> SCHEMA = Map.of(
STATUS, Channels.base(() -> "new"),
AUDIT, Channels.appender(ArrayList::new));
ReleaseState(Map<String, Object> data) {
super(data);
}
String releaseId() {
return value(RELEASE_ID).map(Object::toString).orElseThrow();
}
String notes() {
return value(NOTES).map(Object::toString).orElse("");
}
String draft() {
return value(DRAFT).map(Object::toString).orElseThrow();
}
boolean approved() {
return this.<Boolean>value(APPROVED).orElse(false);
}
List<String> audit() {
return this.<List<String>>value(AUDIT).orElse(List.of());
}
}
public final class ReleaseGraph {
private static final NodeAction<ReleaseState> DRAFT = state -> Map.of(
ReleaseState.DRAFT,
"Release %s%n%n%s".formatted(state.releaseId(), state.notes()),
ReleaseState.STATUS, "awaiting_review",
ReleaseState.AUDIT, "drafted");
private static final NodeAction<ReleaseState> REVIEW = state -> Map.of(
ReleaseState.AUDIT, "reviewed");
private static final NodeAction<ReleaseState> PUBLISH = state -> {
// Call a publication API here with releaseId() as an idempotency key.
return Map.of(
ReleaseState.STATUS, "published",
ReleaseState.AUDIT, "published");
};
private static final NodeAction<ReleaseState> REJECT = state -> Map.of(
ReleaseState.STATUS, "rejected",
ReleaseState.AUDIT, "rejected");
static StateGraph<ReleaseState> definition() throws GraphStateException {
return new StateGraph<>(
ReleaseState.SCHEMA,
ReleaseState::new)
.addNode("draft", node_async(DRAFT))
.addNode("review", node_async(REVIEW))
.addNode("publish", node_async(PUBLISH))
.addNode("reject", node_async(REJECT))
.addEdge(START, "draft")
.addEdge("draft", "review")
.addConditionalEdges(
"review",
edge_async(state ->
state.approved() ? "publish" : "reject"),
Map.of(
"publish", "publish",
"reject", "reject"))
.addEdge("publish", END)
.addEdge("reject", END);
}
public static void main(String[] args) throws GraphStateException {
var graph = definition().compile();
var input = Map.<String, Object>of(
ReleaseState.RELEASE_ID, "2026.08",
ReleaseState.NOTES, "Adds resumable exports and audit reports.",
ReleaseState.APPROVED, true);
for (var output : graph.stream(input)) {
System.out.println(output);
}
}
}グラフは draft から開始し、review に進み、条件付きエッジを評価します。実際のレビュー処理では、信頼できないブラウザーリクエストから approved を直接受け入れてはいけません。認証済みの承認者を特定の提案アクションに紐付け、その判断をサーバー上で検証する必要があります。
PUBLISH ノードは、モデルに面した処理から意図的に分離されています。release_id を冪等性キーとして扱い、最終ペイロードを検証し、外部副作用を生成する前に認可を実行する必要があります。
条件付きルーティングと有界ループ #
条件付きエッジによって、再試行やツールループを明示できます。
call model -> tool needed? -- yes --> execute tool --+
| |
no |
v |
answer <----------------------------------------+ルートアクションは、少数のラベルのいずれかを返すようにしてください。再試行ループでは、試行回数を状態に格納し、上限に達したら失敗ノードにルーティングします。CompileConfig.builder().recursionLimit(...) もグラフ反復回数の正の上限を設定します。デフォルトは 25 です。
高コストまたは危険なループの停止をモデルに依存しないでください。モデル呼び出し、ツール呼び出し、経過時間、トークン使用量、再試行回数をアプリケーション状態または可観測性データで追跡します。すべてのサイクルには、成功ルート、有界な再試行ルート、失敗ルートが必要です。
チェックポイント、ブレークポイント、再開 #
チェックポイントセーバーは、RunnableConfig のスレッド ID に基づいてスナップショットを保存します。これにより、アプリケーションは状態を検査し、設定済みノードで停止し、後で同じスレッドを継続できます。
import java.util.Map;
import org.bsc.langgraph4j.CompileConfig;
import org.bsc.langgraph4j.RunnableConfig;
import org.bsc.langgraph4j.checkpoint.MemorySaver;
var compiled = ReleaseGraph.definition()
.compile(CompileConfig.builder()
.checkpointSaver(new MemorySaver())
.interruptBefore("publish")
.build());
var thread = RunnableConfig.builder()
.threadId("release-2026.08")
.build();
for (var output : compiled.stream(
Map.of(
ReleaseState.RELEASE_ID, "2026.08",
ReleaseState.NOTES, "Adds resumable exports and audit reports.",
ReleaseState.APPROVED, true),
thread)) {
System.out.println(output);
}interruptBefore("publish") を指定すると、名前付きノードの前で実行が停止します。MemorySaver はプロセス終了時にすべての状態を失うため、ローカル開発専用です。LangGraph4j は PostgreSQL、SQLite、MySQL、Redis、DynamoDB、CockroachDB、Hazelcast、Oracle Database などのデータベースやストア向けセーバーモジュールも提供します。アプリケーションの可用性、保持期間、トランザクション、運用要件に基づいてバックエンドを選択してください。
一時停止したスレッドを編集するには、認証済みの承認判断を永続化し、チェックポイント対象の状態を更新します。updateState は更新されたチェックポイントを識別する構成を返します。判断を行ったノードを指定すると、ランタイムが次のルートを計算できます。
RunnableConfig resumed = compiled.updateState(
thread,
Map.of(ReleaseState.APPROVED, true),
"review");
for (var output : compiled.stream(null, resumed)) {
System.out.println(output);
}この stream オーバーロードに null を渡すと、保存されたグラフ入力から再開します。サービスでは、スレッド ID をテナントおよび所有者とともにサーバー側で保存してください。クライアントが任意に指定したスレッド ID を、アクセス権を確認せずに読み込んではいけません。
割り込みはノードレベルのブレークポイントであり、認可システムではありません。承認画面には、具体的で検証済みのアクションとその引数を表示する必要があります。サービスのデプロイ、データの削除、金銭の変更を行うアクションでは、「続行しますか?」だけでは有効な同意になりません。
Durable は Exactly Once を意味しない #
チェックポイントによって計算を再開できますが、チェックポイントと外部副作用は自動的に 1 つのトランザクションにはなりません。
1. publish node calls an external API
2. process fails before the following checkpoint
3. the node runs again during recovery冪等性キーまたは信頼できるトランザクション記録がなければ、副作用が 2 回発生する可能性があります。メール送信、インフラストラクチャの変更、業務システムの変更を行うノードは冪等にしてください。モデル呼び出しは、再実行可能ですが非決定的なものとして扱います。リプレイによって費用が発生し、異なる出力が返される可能性があります。
ストリーミングと状態検査 #
CompiledGraph.stream は、グラフの実行中に一連の NodeOutput 値を返します。完全な例では各値を表示していますが、Web アプリケーションでは安全な進捗イベントを Server-Sent Events や WebSocket メッセージに変換できます。
デフォルトでは、グラフストリーミングは CompiledGraph.StreamMode.VALUES を使用します。セーバーを構成すると、streamSnapshots または StreamMode.SNAPSHOTS を指定した RunnableConfig によって、チェックポイント指向の状態スナップショットが提供されます。
import org.bsc.langgraph4j.CompiledGraph;
import org.bsc.langgraph4j.RunnableConfig;
var config = RunnableConfig.builder()
.threadId("release-2026.08")
.streamMode(CompiledGraph.StreamMode.SNAPSHOTS)
.build();
for (var output : compiled.streamSnapshots(
Map.of(ReleaseState.APPROVED, true),
config)) {
System.out.println(output);
}すべての状態フィールドをブラウザーにシリアライズしないでください。状態には、取得したドキュメント、ツール引数、内部識別子、またはノードが誤って追加したデータが含まれる可能性があります。各受信者が参照できるフィールドだけを含む明示的なイベント DTO を定義してください。
CompiledGraph.getState、stateOf、lastStateOf、getStateHistory は、スレッドの保存済み状態を検査します。これらのインターフェースは特権を持つ運用ツールとして扱い、他のアプリケーションデータに適用するものと同じテナント分離、保持、削除、暗号化のルールを適用してください。
並列ノードとマージ規則 #
複数の出力エッジによって、同じスーパーステップ内で独立した処理を有効化できます。
+-> search internal docs -+
question -> route +-> search issue tracker --+-> combine -> answer
+-> search source code ----+並列化が有効なのは、分岐が独立している場合だけです。分岐の書き込みはマージ可能でなければなりません。チャネルは候補を追加でき、カスタムのベースチャネルリデューサーは名前空間付きマップをマージでき、単一の所有者が最終回答を書き込めます。
並列モデルリクエストはコストを増加させ、プロバイダーの制限を使い果たす可能性があります。アプリケーションのエグゼキューターで並行性を制限し、キャンセルを伝播し、ある分岐が失敗して別の分岐が成功した場合の動作を決定してください。並列ノード自体に互換性のない複数の下流ルートを持たせることはできないため、次の決定を決定論的に行う結合ノードを設計します。
サブグラフとマルチエージェントワークフロー #
StateGraph または CompiledGraph は、別のグラフのノードとして追加できます。サブグラフは、再利用可能な調査フロー、専門化されたツールループ、チームが所有するモジュール、または個別にテスト可能なワークフローに適しています。
StateGraph<ReleaseState> child = new StateGraph<>(
ReleaseState.SCHEMA,
ReleaseState::new);
StateGraph<ReleaseState> parent = new StateGraph<>(
ReleaseState.SCHEMA,
ReleaseState::new)
.addNode("prepare_release", child);親グラフとこの形式の子グラフは、同じ状態型を共有します。グラフを構成する前に、状態の所有権と狭いインターフェースを明確に定義してください。通常の決定論的関数に別個のモデル媒介エージェントを作成しないでください。制御フローが改善されないまま、遅延、コスト、非決定性が増加します。
プロセス内サブグラフは、独立したリモートエージェントではありません。A2A はエージェントサービス境界を越えた通信を扱います。MCP はエージェントにツールとリソースを公開します。LangGraph4j は、いずれかのプロトコルを使用する Java ワークフローをオーケストレーションします。
LangGraph4j、LangChain4j、Spring AI #
これらのコンポーネントは異なるレイヤーを担います。
| コンポーネント | 主な役割 |
|---|---|
| LangGraph4j | Java の状態グラフ、ルーティング、チェックポイント、ブレークポイント、実行 |
| LangChain4j | Java のモデル、メッセージ、ツール、取得、エージェント抽象化 |
| Spring AI | Spring のモデルクライアント、ツール、RAG、ベクトルストア、MCP、可観測性 |
| LangGraph4j Studio | グラフの可視化、実行、デバッグ UI |
LangGraph4j は LangGraph に着想を得た独立したプロジェクトです。別のランタイムのドロップイン実装ではなく、チェックポイントやデプロイに関する互換性保証も共有しません。
プロバイダーアダプター、メッセージ抽象化、ツール、取得機能がアプリケーションに適している場合は、グラフノード内で LangChain4j または Spring AI を使用します。より単純であれば、通常の Java クライアントを使用してください。明示的なサイクル、状態を持つルーティング、永続チェックポイント、制御された一時停止が必要な場合はグラフランタイムが有用ですが、基本的なチャットや取得処理には必要ありません。
デプロイと運用 #
LangGraph4j は、チームが構築・運用する Java サービスに組み込まれるライブラリです。チームは次の事項を管理します。
- チェックポイントストレージ、マイグレーション、保持、削除
- キューイング、並行性制限、キャンセル、タイムアウト
- 認証とスレッド認可
- 再試行ポリシーと冪等性
- ストリーミングエンドポイントと再接続動作
- 可観測性、アラート、コスト管理
- モデルとツールの認証情報
- 安全なシャットダウンと回復
LangGraph4j は Mermaid と PlantUML のグラフ表現サポート、および視覚的な検査のための組み込み可能な Studio プロジェクトを含みます。これらは開発時に有用ですが、本番メトリクス、トレース、監査ログ、認可チェックの代替にはなりません。
コンパイル済みグラフを Spring シングルトンとして登録する場合、ノードと依存関係をステートレスかつスレッドセーフに保ってください。実行ごとのデータは、シングルトンの可変フィールドではなく、グラフ入力とチェックポイントストレージに保持します。
設計ガイドライン #
決定論的な判断は Java に保持する #
モデルは意味的な分類と生成に使用します。認可、算術、クォータ、スキーマ検証、不変条件の適用には通常の Java コードを使用します。
状態を明示する #
古いメッセージを解析するのではなく、重要な判断を名前付きフィールドに格納します。各フィールドを所有するノードと、並行更新をマージする方法を文書化してください。
すべてのサイクルに上限を設ける #
試行回数、モデル呼び出し、経過時間、トークン、ツール呼び出しを追跡します。各ループに終了条件とエラールートを設定してください。
副作用を冪等にする #
再試行と回復は通常の動作です。外部から見える副作用には、安定した操作 ID と信頼できる記録を使用します。
提案と実行を分離する #
モデルにはアクションを提案させます。別のノードがアクションを実行する前に、検証、認可、必要な承認の取得を行います。
統制可能なデータだけを永続化する #
チェックポイントに保持、暗号化、削除、テナント分離を適用します。グラフ状態には、プロンプト、個人データ、取得したドキュメント、ツール結果が含まれる可能性があります。
軌跡を評価する #
グラフが誤ったツールを使い、禁止されたデータを取得し、過剰に再試行し、必要なレビューを無視していても、最終回答は正しく見えることがあります。最終出力に加えて、ルート、状態遷移、ツール引数、根拠、コスト、遅延を評価してください。
グラフと状態をバージョン管理する #
スレッドはデプロイ後も存続できます。後続リリースでノード名や状態スキーマが変更されても、保存済みチェックポイントは以前のグラフを参照している可能性があります。本番グラフを変更する前に、マイグレーションと互換性のポリシーを定義してください。
LangGraph4j を使用しない場合 #
次の場合、グラフランタイムを追加しないでください。
- モデル呼び出し 1 回で十分な場合
- 短い決定論的パイプラインがすでに明確な場合
- 通常の Spring AI または LangChain4j コードで要件を満たせる場合
- 従来型のワークフローエンジンが永続的な業務プロセスをすでに管理している場合
- リクエスト間で状態を保持する必要がない場合
- チームが追加のステートフルシステムを運用・評価できない場合
金融やフルフィルメントのプロセスでは、従来型のワークフローエンジンを信頼できるシステムとして残し、グラフで有界な AI 分析を行う構成にできます。
durable business workflow
-> starts an AI analysis graph
-> receives a proposed decision
-> applies deterministic policy and approval
-> performs the authoritative transaction必要な制御フローを表現できる、最小の抽象化を選択してください。ワークフローが実際に分岐、ループ、一時停止、再開を行う場合、グラフは有用です。それ以外では、すでに明確なコードの周囲に冗長なフローチャートを作ることになりかねません。
コメント
GitHub で閲覧・コメントする(投稿にはログインが必要です)