取り組みの背景：なぜ「ステートフルなAIエージェント」が必要なのか
AIエージェントの開発が本格化するにつれ、単純なチェーン処理では対応しきれない複雑なシナリオが増えてきましました。たとえば、コードレビューエージェントが途中で人間の承認を待つ必要があったり、データ分析エージェントが前のステップの結果に応じて動的に処理を分岐させたりといったケースです。
こうした課題を解決するのがLangGraphです。LangGraphはLangChain社が開発したライブラリで、AIエージェントをグラフ構造（ノードとエッジ）として定義し、状態（State）を明示的に管理しながら複雑なワークフローを構築できます。
この記事で学べること：
- LangGraphのコア概念（StateGraph、ノード、エッジ、チェックポイント）
- Antigravityを使ったLangGraphコードの効率的な生成・リファクタリング
- プロダクション品質のエージェント設計パターン
- ヒューマンインザループ実装
- エラーハンドリングとロールバック戦略
LangGraphの基本概念を理解する
StateGraph：エージェントの骨格
LangGraphの中心となるのはStateGraphクラスです。エージェントの「状態」をTypedDictまたはPydanticモデルで定義し、その状態を変換するノード（関数）をグラフに登録します。
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
import operator
# エージェントの状態定義
class AgentState(TypedDict):
messages: Annotated[List[BaseMessage], operator.add] # メッセージ履歴（追記モード）
current_step: str # 現在のステップ名
tool_results: dict # ツール実行結果
iteration_count: int # ループ回数（無限ループ防止）
error: str | None # エラー情報
# グラフの初期化
graph = StateGraph(AgentState)Annotated[List[BaseMessage], operator.add]という記述がポイントです。これはLangGraphの「Reducer」機能で、各ノードが返す部分的な状態をどのようにマージするかを指定します。operator.addはリストの追記を意味し、メッセージ履歴が自動的に蓄積されます。
ノードとエッジ：処理フローの設計
ノードは状態を受け取り、更新された部分的な状態を返す関数です。エッジはノード間の接続を定義します。
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
# LLMの初期化
llm = ChatOpenAI(model="gemini-2.0-flash", temperature=0)
# ツール定義
@tool
def search_codebase(query: str) -> str:
"""コードベースを検索して関連コードを返す"""
# 実際の実装では Antigravity の MCP ツールと連携
return f"Search results for: {query}"
@tool
def run_tests(test_path: str) -> dict:
"""指定パスのテストを実行して結果を返す"""
return {"passed": 10, "failed": 2, "errors": ["test_auth.py:45"]}
tools = [search_codebase, run_tests]
llm_with_tools = llm.bind_tools(tools)
# エージェントノード（LLMの呼び出し）
def agent_node(state: AgentState) -> dict:
"""メインのLLMエージェント処理"""
response = llm_with_tools.invoke(state["messages"])
return {
"messages": [response],
"current_step": "agent",
"iteration_count": state.get("iteration_count", 0) + 1
}
# ツール実行ノード
from langgraph.prebuilt import ToolNode
tool_node = ToolNode(tools)
# 条件分岐関数（エッジのルーティング）
def should_continue(state: AgentState) -> str:
"""次のノードを決定する条件分岐ロジック"""
messages = state["messages"]
last_message = messages[-1]
# 最大イテレーション数チェック（無限ループ防止）
if state.get("iteration_count", 0) >= 10:
return "end"
# ツール呼び出しがあれば tools ノードへ
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
# なければ終了
return "end"
# グラフの構築
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.set_entry_point("agent")
graph.add_conditional_edges(
"agent",
should_continue,
{"tools": "tools", "end": END}
)
graph.add_edge("tools", "agent") # ツール実行後は必ずエージェントに戻る
app = graph.compile()Antigravityを活用したLangGraph開発フロー
プロジェクトのセットアップをAntigravityに任せる
LangGraphプロジェクトの初期セットアップは依存関係が複雑です。Antigravityを使えば、適切なプロジェクト構造とconfigファイルを一気に生成できます。
Antigravityのチャットに以下のように指示してみましょう：
以下の要件でLangGraphエージェントプロジェクトをセットアップしてください：
- Python 3.11+
- LangGraph 0.2+, LangChain 0.3+
- Gemini 2.0 Flash をLLMとして使用
- PostgreSQL + pgvector をチェックポイントストアとして使用
- 開発環境: Docker Compose
- テスト: pytest + pytest-asyncio
pyproject.toml, docker-compose.yml, .env.example,
src/agents/__init__.py の初期ファイルを作成してください
Antigravityは適切なバージョン指定と依存関係を持つpyproject.tomlを生成してくれます。
# pyproject.toml（Antigravity生成例）
[project]
name = "langgraph-agent"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = [
"langgraph>=0.2.0",
"langchain>=0.3.0",
"langchain-google-genai>=2.0.0",
"langchain-community>=0.3.0",
"psycopg[binary]>=3.1.0",
"langgraph-checkpoint-postgres>=2.0.0",
"python-dotenv>=1.0.0",
"pydantic>=2.0.0",
]
[project.optional-dependencies]
dev = [
"pytest>=8.0.0",
"pytest-asyncio>=0.23.0",
"pytest-mock>=3.12.0",
"ruff>=0.3.0",
]コード生成とリファクタリングのパターン
Antigravityはエディタ上でLangGraphコードのリファクタリングを強力にサポートします。たとえば、既存の単純なLLMチェーンをステートフルなLangGraphエージェントに変換する場合：
- 既存コードをAntigravityのエディタで開く
- コメント
# TODO: LangGraphのStateGraphに変換を追加 Cmd+I（またはCtrl+I）でAntigravityのインラインチャットを起動- 「このLLMチェーンをLangGraphのStateGraphに変換してください。状態にはmessages, current_step, error フィールドを含め、ToolNodeも使えるようにしてください」と指示
Antigravityはコンテキストを理解した上で、適切な変換コードを提案してくれます。
チェックポイント：長時間タスクの耐障害性
PostgreSQLチェックポイントの設定
実際のプロダクション環境では、エージェントの途中状態を永続化する「チェックポイント」が不可欠です。LangGraphはPostgreSQL、SQLite、Redisなど複数のバックエンドをサポートしています。
import asyncio
from psycopg_pool import AsyncConnectionPool
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
async def create_agent_with_checkpoints():
"""PostgreSQLチェックポイント付きエージェントの作成"""
# PostgreSQL接続プールの設定
connection_string = "postgresql://user:password@localhost:5432/langgraph_db"
async with AsyncConnectionPool(
conninfo=connection_string,
max_size=20,
kwargs={"autocommit": True, "prepare_threshold": 0}
) as pool:
# チェックポインターの初期化
checkpointer = AsyncPostgresSaver(pool)
await checkpointer.setup() # テーブルの自動作成
# チェックポインター付きグラフのコンパイル
app = graph.compile(checkpointer=checkpointer)
# スレッドIDを使って会話の継続が可能
config = {
"configurable": {
"thread_id": "user-123-task-456",
"checkpoint_ns": "code_review_agent"
}
}
# 初回実行
initial_state = {
"messages": [HumanMessage(content="このPRのコードレビューをしてください")],
"current_step": "start",
"tool_results": {},
"iteration_count": 0,
"error": None
}
result = await app.ainvoke(initial_state, config=config)
print(f"First run result: {result['messages'][-1].content}")
# 後から同じスレッドIDで再開（前の状態から継続）
followup = {
"messages": [HumanMessage(content="security の部分を特に詳しくレビューして")]
}
result2 = await app.ainvoke(followup, config=config)
return result2チェックポイントの状態を確認・デバッグする
Antigravityのデバッグ機能と組み合わせることで、チェックポイントの状態を視覚的に確認できます。
async def inspect_checkpoints(app, thread_id: str):
"""チェックポイントの履歴を確認する"""
config = {"configurable": {"thread_id": thread_id}}
# 現在の状態を取得
current_state = await app.aget_state(config)
print("Current state values:", current_state.values)
print("Next nodes:", current_state.next)
print("Created at:", current_state.created_at)
# チェックポイント履歴を取得（最新3件）
history = []
async for state in app.aget_state_history(config, limit=3):
history.append({
"checkpoint_id": state.config["configurable"]["checkpoint_id"],
"step": state.values.get("current_step"),
"message_count": len(state.values.get("messages", [])),
"created_at": state.created_at
})
return history
# 特定のチェックポイントに戻す（ロールバック）
async def rollback_to_checkpoint(app, thread_id: str, checkpoint_id: str):
"""特定のチェックポイントにロールバックして再実行"""
config = {
"configurable": {
"thread_id": thread_id,
"checkpoint_id": checkpoint_id # 戻りたいチェックポイントID
}
}
# このconfigで再度invokeすることで、指定時点から再実行できる
result = await app.ainvoke(
{"messages": [HumanMessage(content="別のアプローチで試してください")]},
config=config
)
return resultヒューマンインザループ：人間が介入するエージェント
interrupt_beforeによる承認フローの実装
複雑なタスクを自動化する際、特定のアクション（本番データベースの更新、外部APIへのPOSTなど）の前に人間の承認を挟みたいケースがあります。LangGraphのinterrupt_before機能がこれを実現します。
from langgraph.graph import StateGraph, END
# 危険な操作を行うノード
def deploy_to_production(state: AgentState) -> dict:
"""本番環境へのデプロイ（要承認）"""
# このノードは interrupt_before で実行前に一時停止される
deploy_result = execute_deployment(state["tool_results"]["build_artifact"])
return {
"messages": [AIMessage(content=f"デプロイ完了: {deploy_result}")],
"current_step": "deployed"
}
def prepare_deployment(state: AgentState) -> dict:
"""デプロイの準備（ビルド、テスト等）"""
# ビルドとテストを実行
return {
"tool_results": {"build_artifact": "dist/app.zip", "test_passed": True},
"current_step": "deploy_ready",
"messages": [AIMessage(content="ビルドとテストが完了しました。本番環境へのデプロイを実行しますか？")]
}
# グラフの構築
deploy_graph = StateGraph(AgentState)
deploy_graph.add_node("prepare", prepare_deployment)
deploy_graph.add_node("deploy", deploy_to_production)
deploy_graph.set_entry_point("prepare")
deploy_graph.add_edge("prepare", "deploy")
deploy_graph.add_edge("deploy", END)
# interrupt_before でdeployノードの前に一時停止
deploy_app = deploy_graph.compile(
checkpointer=checkpointer,
interrupt_before=["deploy"] # デプロイノードの前で一時停止
)
# 実行
config = {"configurable": {"thread_id": "deploy-task-001"}}
state = await deploy_app.ainvoke(initial_state, config=config)
# この時点でエージェントは "deploy" ノードの前で一時停止している
# 現在の状態を確認
current = await deploy_app.aget_state(config)
print("Waiting for approval:", current.next) # ['deploy']
print("Build result:", current.values["tool_results"])
# 人間が確認して承認する場合
# Noneを渡すことで「前の状態を変更せずに続行」を指示
await deploy_app.ainvoke(None, config=config)
# 人間が修正を加えて承認する場合
await deploy_app.aupdate_state(
config,
{"messages": [HumanMessage(content="承認します。stagingも同時にデプロイしてください")]}
)
await deploy_app.ainvoke(None, config=config)Antigravityを使ったヒューマンインザループUIの生成
Antigravityに依頼することで、このヒューマンインザループをFastAPIのWebAPIとして包むコードも生成できます。
以下のLangGraphエージェントにFastAPIのWebAPIエンドポイントを追加してください：
1. POST /tasks - 新しいタスクを開始し、task_idを返す
2. GET /tasks/{task_id}/status - タスクの状態を返す（waiting_approval フラグ含む）
3. POST /tasks/{task_id}/approve - 承認してエージェントを再開
4. POST /tasks/{task_id}/reject - 拒否して代替アクションを指示
非同期処理にはasyncioとBackgroundTasksを使ってください
複数エージェントのオーケストレーション
スーパーバイザーパターン
大規模なタスクでは、複数の専門エージェントを調整する「スーパーバイザーエージェント」を使うパターンが効果的です。
from langgraph.graph import StateGraph, END
from typing import Literal
# 各専門エージェントの定義
class SupervisorState(TypedDict):
messages: Annotated[List[BaseMessage], operator.add]
next_agent: str
task_context: dict
completed_tasks: List[str]
# スーパーバイザーノード
def supervisor_node(state: SupervisorState) -> dict:
"""次に実行する専門エージェントを決定"""
system_prompt = """あなたはエージェントオーケストレーターです。
以下の専門エージェントを適切に組み合わせてタスクを実行してください：
- researcher: 情報収集とリサーチ
- coder: コードの実装と修正
- reviewer: コードレビューとテスト
- documenter: ドキュメント生成
- FINISH: 全タスク完了
次に実行すべきエージェントを1単語で答えてください。
"""
response = llm.invoke([
{"role": "system", "content": system_prompt},
*state["messages"]
])
return {"next_agent": response.content.strip()}
# 専門エージェントの実装（researcher の例）
def researcher_agent(state: SupervisorState) -> dict:
"""リサーチ専門エージェント"""
research_llm = llm_with_tools # web_search, read_doc ツールを持つLLM
result = research_llm.invoke([
{"role": "system", "content": "あなたは技術リサーチの専門家です。"},
*state["messages"]
])
return {
"messages": [result],
"completed_tasks": state.get("completed_tasks", []) + ["research"]
}
# ルーティング関数
def route_to_agent(state: SupervisorState) -> Literal["researcher", "coder", "reviewer", "documenter", "end"]:
agent = state["next_agent"].lower()
if agent == "finish":
return "end"
return agent if agent in ["researcher", "coder", "reviewer", "documenter"] else "end"
# スーパーバイザーグラフの構築
supervisor_graph = StateGraph(SupervisorState)
supervisor_graph.add_node("supervisor", supervisor_node)
supervisor_graph.add_node("researcher", researcher_agent)
supervisor_graph.add_node("coder", coder_agent) # 同様に実装
supervisor_graph.add_node("reviewer", reviewer_agent)
supervisor_graph.add_node("documenter", documenter_agent)
supervisor_graph.set_entry_point("supervisor")
supervisor_graph.add_conditional_edges(
"supervisor",
route_to_agent,
{
"researcher": "researcher",
"coder": "coder",
"reviewer": "reviewer",
"documenter": "documenter",
"end": END
}
)
# 各専門エージェントは処理後スーパーバイザーに戻る
for agent in ["researcher", "coder", "reviewer", "documenter"]:
supervisor_graph.add_edge(agent, "supervisor")
multi_agent_app = supervisor_graph.compile(checkpointer=checkpointer)エラーハンドリングと再試行戦略
堅牢なエラーハンドリングパターン
プロダクション環境では、LLMのレート制限やツール実行エラーに対する堅牢なハンドリングが必要です。
import time
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from langchain_core.exceptions import OutputParserException
class AgentError(Exception):
"""エージェント固有のエラー"""
pass
def agent_node_with_retry(state: AgentState) -> dict:
"""リトライロジック付きエージェントノード"""
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10),
retry=retry_if_exception_type((Exception,)),
reraise=True
)
def _invoke_with_retry():
return llm_with_tools.invoke(state["messages"])
try:
response = _invoke_with_retry()
return {
"messages": [response],
"current_step": "agent",
"error": None,
"iteration_count": state.get("iteration_count", 0) + 1
}
except Exception as e:
error_msg = f"エージェント実行エラー: {str(e)}"
return {
"messages": [AIMessage(content=f"エラーが発生しました: {error_msg}")],
"current_step": "error",
"error": error_msg
}
# エラーハンドリングノード
def error_handler_node(state: AgentState) -> dict:
"""エラー状態を処理し、回復を試みる"""
error = state.get("error", "Unknown error")
recovery_response = llm.invoke([
*state["messages"],
HumanMessage(content=f"エラーが発生しました: {error}\n代替手段で目標を達成してください。")
])
return {
"messages": [recovery_response],
"current_step": "recovery",
"error": None # エラーをクリア
}
# エラー時のルーティング
def route_on_error(state: AgentState) -> str:
if state.get("error"):
return "error_handler"
return should_continue(state)
graph.add_node("error_handler", error_handler_node)
graph.add_conditional_edges("agent", route_on_error, {
"tools": "tools",
"end": END,
"error_handler": "error_handler"
})
graph.add_edge("error_handler", "agent") # 回復後は再試行LangSmithによる観測可能性
トレーシングのセットアップ
LangGraphエージェントのデバッグには、LangSmithを使ったトレーシングが非常に有効です。Antigravityを使ってトレーシング設定も効率よく構成できます。
import os
from langsmith import Client
# 環境変数でLangSmithを有効化
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "YOUR_LANGSMITH_API_KEY"
os.environ["LANGCHAIN_PROJECT"] = "antigravitylab-agent"
# カスタムメタデータの追加（コスト計算に有用）
from langchain_core.callbacks import CallbackManager
from langsmith.run_helpers import traceable
@traceable(name="code_review_workflow", metadata={"version": "1.0", "env": "production"})
async def run_code_review_agent(pr_url: str, thread_id: str) -> dict:
"""コードレビューエージェントの実行（LangSmithトレース付き）"""
config = {"configurable": {"thread_id": thread_id}}
result = await app.ainvoke(
{"messages": [HumanMessage(content=f"PRをレビューしてください: {pr_url}")]},
config=config
)
return {
"review": result["messages"][-1].content,
"steps_taken": result.get("iteration_count", 0)
}LangSmithのダッシュボードでは、各エージェント実行のトークン使用量、レイテンシ、エラー率をリアルタイムで確認できます。月次のAPIコスト分析にも活用できます。
テスト戦略
LangGraphエージェントのユニットテスト
Antigravityを使って、適切なテストコードを生成することも重要な開発工程です。
import pytest
from unittest.mock import AsyncMock, MagicMock, patch
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
@pytest.mark.asyncio
async def test_agent_node_normal_flow():
"""エージェントノードの正常系テスト"""
# モックレスポンスの設定
mock_response = AIMessage(
content="コードを検索します",
tool_calls=[{
"name": "search_codebase",
"args": {"query": "authentication"},
"id": "call_001"
}]
)
with patch.object(llm_with_tools, "invoke", return_value=mock_response):
state = AgentState(
messages=[HumanMessage(content="認証コードを調べて")],
current_step="start",
tool_results={},
iteration_count=0,
error=None
)
result = agent_node(state)
assert result["current_step"] == "agent"
assert result["iteration_count"] == 1
assert len(result["messages"]) == 1
assert result["messages"][0].tool_calls[0]["name"] == "search_codebase"
@pytest.mark.asyncio
async def test_infinite_loop_prevention():
"""無限ループ防止のテスト"""
state = AgentState(
messages=[HumanMessage(content="test")],
current_step="agent",
tool_results={},
iteration_count=10, # 最大値に達している
error=None
)
mock_response = AIMessage(
content="",
tool_calls=[{"name": "search_codebase", "args": {}, "id": "call_999"}]
)
with patch.object(llm_with_tools, "invoke", return_value=mock_response):
next_node = should_continue(state)
# iteration_count が 10 に達しているので "end" を返すべき
assert next_node == "end"
@pytest.mark.asyncio
async def test_checkpoint_persistence():
"""チェックポイントの永続化テスト（インメモリSQLiteを使用）"""
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
async with AsyncSqliteSaver.from_conn_string(":memory:") as checkpointer:
test_app = graph.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "test-thread-001"}}
# 1回目の実行
await test_app.ainvoke(
{"messages": [HumanMessage(content="Hello")]},
config=config
)
# チェックポイントが保存されているか確認
state = await test_app.aget_state(config)
assert state is not None
assert len(state.values["messages"]) > 0コスト最適化とトークン管理
メッセージ履歴の効率的な削減
LangGraphエージェントのランニングコストを抑える上で最も重要なのは、メッセージ履歴の管理です。LangChainが提供するtrimMessages関数を使うことで、重要な情報を保持しながら古いメッセージを適切に削除できます。
from langchain_core.messages import trim_messages, SystemMessage
def trim_agent_messages(state: AgentState) -> dict:
"""メッセージ履歴をトリミングしてトークン数を削減"""
trimmed = trim_messages(
state["messages"],
max_tokens=4096, # モデルの入力上限の50%程度が目安
strategy="last", # 最新のメッセージを優先して保持
token_counter=llm, # モデルのトークナイザーでカウント
include_system=True, # システムメッセージは必ず保持
allow_partial=False, # メッセージを途中で切らない
start_on="human" # HumanMessageから始まるよう調整
)
return {"messages": trimmed}
# トリミングノードをグラフに追加
graph.add_node("trim_history", trim_agent_messages)
graph.add_edge("tools", "trim_history") # ツール実行後にトリミング
graph.add_edge("trim_history", "agent") # トリミング後にエージェントへツール呼び出しのキャッシュ実装
同じ引数でツールが繰り返し呼ばれるケースでは、キャッシュが大幅なコスト削減につながります。
import hashlib
import json
from functools import lru_cache
from datetime import datetime, timedelta
class CachedToolWrapper:
"""TTL付きキャッシュを持つツールラッパー"""
def __init__(self, tool_func, ttl_minutes: int = 10):
self.tool_func = tool_func
self.ttl = timedelta(minutes=ttl_minutes)
self._cache: dict = {}
def _make_cache_key(self, **kwargs) -> str:
"""引数からキャッシュキーを生成"""
return hashlib.md5(json.dumps(kwargs, sort_keys=True).encode()).hexdigest()
def __call__(self, **kwargs) -> str:
cache_key = self._make_cache_key(**kwargs)
now = datetime.now()
# キャッシュヒットチェック
if cache_key in self._cache:
result, timestamp = self._cache[cache_key]
if now - timestamp < self.ttl:
print(f"Cache hit for {self.tool_func.__name__}")
return result
# キャッシュミス：実際にツールを実行
result = self.tool_func(**kwargs)
self._cache[cache_key] = (result, now)
return result
# キャッシュ付きツールの作成
cached_search = CachedToolWrapper(search_codebase, ttl_minutes=5)モデル選択戦略でコストを最小化
同じタスクでも、必要な精度に応じてモデルを使い分けることでコストを大幅に削減できます。
from langchain_google_genai import ChatGoogleGenerativeAI
class AdaptiveModelSelector:
"""タスクの複雑度に応じてモデルを動的に選択"""
def __init__(self):
# コスト効率重視（単純タスク用）
self.flash_model = ChatGoogleGenerativeAI(
model="gemini-2.0-flash",
temperature=0
)
# 高精度（複雑タスク用）
self.pro_model = ChatGoogleGenerativeAI(
model="gemini-2.0-pro",
temperature=0
)
def select_model(self, task_complexity: str, requires_reasoning: bool = False):
"""タスクの種類に応じて最適なモデルを返す"""
if requires_reasoning or task_complexity == "high":
return self.pro_model
return self.flash_model
# エージェントノード内でのモデル選択
def adaptive_agent_node(state: AgentState) -> dict:
"""複雑度に応じてモデルを切り替えるエージェントノード"""
selector = AdaptiveModelSelector()
# メッセージ数やタスクタイプでモデルを選択
iteration = state.get("iteration_count", 0)
is_complex = iteration > 3 or "architecture" in str(state["messages"]).lower()
model = selector.select_model(
task_complexity="high" if is_complex else "low",
requires_reasoning=is_complex
)
model_with_tools = model.bind_tools(tools)
response = model_with_tools.invoke(state["messages"])
return {
"messages": [response],
"current_step": "agent",
"iteration_count": iteration + 1
}プロダクション展開のベストプラクティス
FastAPIによるエージェントAPIサーバーの構築
LangGraphエージェントをWebAPIとして公開する際のベストプラクティスを示します。
from fastapi import FastAPI, BackgroundTasks, HTTPException
from pydantic import BaseModel
import uuid
import asyncio
from typing import Optional
app_server = FastAPI(title="LangGraph Agent API")
# タスク状態の管理
task_registry: dict = {}
class TaskCreateRequest(BaseModel):
task_description: str
user_id: str
priority: str = "normal" # normal / high
class TaskStatusResponse(BaseModel):
task_id: str
status: str # running / waiting_approval / completed / failed
current_step: str
result: Optional[str] = None
waiting_for: Optional[str] = None # 承認待ちの場合の説明
@app_server.post("/tasks", response_model=dict)
async def create_task(
request: TaskCreateRequest,
background_tasks: BackgroundTasks
):
"""新しいエージェントタスクを開始"""
task_id = str(uuid.uuid4())
task_registry[task_id] = {
"status": "running",
"current_step": "start",
"result": None,
"error": None
}
# バックグラウンドでエージェントを実行
background_tasks.add_task(
run_agent_task,
task_id=task_id,
description=request.task_description,
thread_id=f"{request.user_id}-{task_id}"
)
return {"task_id": task_id, "status": "started"}
async def run_agent_task(task_id: str, description: str, thread_id: str):
"""バックグラウンドでエージェントを実行"""
config = {"configurable": {"thread_id": thread_id}}
try:
result = await app.ainvoke(
{
"messages": [HumanMessage(content=description)],
"current_step": "start",
"tool_results": {},
"iteration_count": 0,
"error": None
},
config=config
)
# interrupt_before で停止しているか確認
state = await app.aget_state(config)
if state.next: # まだ次のノードが残っている = 承認待ち
task_registry[task_id].update({
"status": "waiting_approval",
"current_step": state.next[0],
"waiting_for": f"Approval required for: {state.next[0]}"
})
else:
task_registry[task_id].update({
"status": "completed",
"result": result["messages"][-1].content
})
except Exception as e:
task_registry[task_id].update({
"status": "failed",
"error": str(e)
})
@app_server.get("/tasks/{task_id}/status", response_model=TaskStatusResponse)
async def get_task_status(task_id: str):
"""タスクの現在状態を取得"""
if task_id not in task_registry:
raise HTTPException(status_code=404, detail="Task not found")
task = task_registry[task_id]
return TaskStatusResponse(
task_id=task_id,
status=task["status"],
current_step=task.get("current_step", "unknown"),
result=task.get("result"),
waiting_for=task.get("waiting_for")
)
@app_server.post("/tasks/{task_id}/approve")
async def approve_task(task_id: str, background_tasks: BackgroundTasks):
"""承認待ちタスクを再開"""
if task_id not in task_registry:
raise HTTPException(status_code=404, detail="Task not found")
if task_registry[task_id]["status"] != "waiting_approval":
raise HTTPException(status_code=400, detail="Task is not waiting for approval")
task_registry[task_id]["status"] = "running"
thread_id = f"{task_id}" # 実際の実装ではuser_idも含める
background_tasks.add_task(resume_agent_task, task_id=task_id, thread_id=thread_id)
return {"message": "Task resumed", "task_id": task_id}ヘルスチェックとグレースフルシャットダウン
from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app: FastAPI):
"""アプリケーションのライフサイクル管理"""
# 起動時: データベース接続プールの初期化
print("🚀 Starting LangGraph Agent API Server...")
await checkpointer.setup()
print("✅ Checkpoint store initialized")
yield # ← アプリケーションが実行中
# シャットダウン時: 実行中タスクの安全な停止
print("🔄 Shutting down gracefully...")
running_tasks = [
tid for tid, t in task_registry.items()
if t["status"] == "running"
]
if running_tasks:
print(f"⚠️ {len(running_tasks)} tasks still running, saving state...")
# チェックポインターに状態が保存されているため、再起動後に再開可能
print("✅ Shutdown complete")
app_server = FastAPI(title="LangGraph Agent API", lifespan=lifespan)
@app_server.get("/health")
async def health_check():
"""ヘルスチェックエンドポイント"""
try:
# チェックポインターへの接続確認
# await checkpointer.pool.getconn()
return {
"status": "healthy",
"active_tasks": len([t for t in task_registry.values() if t["status"] == "running"]),
"pending_approvals": len([t for t in task_registry.values() if t["status"] == "waiting_approval"])
}
except Exception as e:
return {"status": "unhealthy", "error": str(e)}まとめ
ここではLangGraphとAntigravityを組み合わせてプロダクション品質のステートフルAIエージェントを構築する方法を解説しました。
要点をまとめると次のとおりです。LangGraphのStateGraphとReducerにより、複雑なエージェントのデータフローを宣言的に定義できます。PostgreSQLチェックポインターを使うことで、長時間タスクの耐障害性と会話継続性が実現します。interrupt_beforeを活用したヒューマンインザループは、高リスクな自動化タスクを安全に運用するための必須パターンです。スーパーバイザーパターンにより、専門エージェントを組み合わせた高度なオーケストレーションが可能になります。そして、Antigravityはコードの文脈を理解した上でLangGraphコードの生成・リファクタリング・デバッグを強力にサポートしてくれます。
LangGraphとAntigravityの組み合わせは、AIエージェント開発の生産性を大幅に向上させる強力な組み合わせです。ぜひ本記事のコードを参考に、自身のプロジェクトに応用してみてください。
LangGraphを使ったシステム設計の基礎を固めたい方には、先にAIエージェントシステム設計完全ガイドをご覧いただくと、本記事の内容がより深く理解できるでしょう。また、Antigravity AgentKit 2.0 でマルチエージェント開発を始めるの記事も、本記事と組み合わせることでマルチエージェント開発の全体像が把握できます。
LangGraphの理論的背景