🧩 Graph Engineering入門：AIエージェントのループを明示グラフへ変える

AIエージェントが長いタスクを担うと、1つのloopへ判断・状態・副作用が集中し、再開や監査が難しくなります。Graph Engineeringは、agent、決定的関数、evaluator、人間承認を明示的な実行グラフへ分離し、型付きstateと検査可能なedgeで接続する方法論です。

検証日: 2026-07-27
対象: AIエージェント文脈のGraph Engineering

設計方式を選びたい方は「概要」「反証・限界・適用条件」から読むと、graphを採用すべき条件を短時間で判断できます。実装へ進みたい方は「構造」「データ」「構築方法」を読み、運用担当者は「運用」「トラブルシューティング」を参照してください。

本稿の Graph Engineering は Knowledge Graph Engineering や GraphRAG ではありません。agent、決定的関数、evaluator、人間承認を実行nodeとして接続し、状態・遷移・権限・予算・回復を明示的に設計する方法論を扱います。

この記事の全体像。以下、順に解説します。

概要

調査対象と用語の境界

本稿が扱う Graph Engineering は、AI エージェントの実行を明示的なグラフとして設計する方法論です。Knowledge Graph Engineering、GraphRAG、グラフデータベースのスキーマ設計、グラフアルゴリズムの実装は別の領域です。

出典の種類を分けた定義

この統合定義では「複数 agent」を必須条件にしません。単一 agent、決定的関数、evaluator、人間を組み合わせた graph も対象です。また、edge の型は単なる矢印の名前ではなく、入力・出力 schema、遷移条件、許可される副作用、失敗時の扱いを含む境界契約として扱います。

起点と普及の時系列

「誰が用語を提唱したか」は、確認できた一次資料と普及の契機を分けて扱う必要があります。

以上から、Josh C. Simmons は少なくとも 2026 年 7 月 4 日に AI agent 文脈の Graph Engineering を明示的に定義した人物と確認できます。ただし、公開 Web 全体で最初に語を使った人物だと証明する資料は確認できません。Peter Steinberger は 7 月中旬の議論を広げた人物ですが、当該投稿は定義を提示していません。「Steinberger が発明し、Simmons が後から定義した」という時系列は一次資料と整合しません。

主要主張の証拠マトリクス

技術的な系譜

Graph Engineering の構成要素は新規発明の集合ではありません。AI agent 固有の不確実性を、既存の software / distributed systems engineering の技法へ接続した再編成です。

関連概念との違い

実行構成の明示度を比較する

以下は方法論を排他的に比べる表ではありません。loopをnode内部で使い、workflowをgraphの実行基盤にする包含関係を前提に、制御構造をどこまで明示するかを比較します。「起動」はtaskの設計から最初の有効な実行までを指します。実測値はmodel、tool、runtime、並列上限に依存します。

技術的には、Graph Engineering の overhead は node 数そのものより、node 境界の永続化、serialization、schema validation、routing、join、trace、retry に生じます。一方、独立 branch の wall-clock time は並列化で短縮できます。依存が密な task、rate limit が厳しい tool、共有 context が大きい task では、この利点が小さくなります。

Anthropic は、Claude Opus 4 を lead agent、Claude Sonnet 4 を subagent とする multi-agent 構成が、単体の Claude Opus 4 を社内 research eval で 90.2% 上回ったと報告しています。同記事は別途、breadth-first query が multi-agent に適する傾向と、通常の agent が chat の約 4 倍、multi-agent が chat の約 15 倍の token を使うという観測も述べています。いずれも Anthropic Research の構成と内部評価に限られるため、Graph Engineering 一般の性能値には転用できません。

適用ユースケース

Graph Engineering を選ぶ前に単純な方式を優先するケース

採用判断チェックリスト

次の問いのうち「はい」が少なければ、単一 call、loop、または通常の workflow を先に選びます。graph を採用する場合も、最初に必要な node と edge だけを実装します。

特徴

• loop を node 内へ保存します。 Graph Engineering は loop の置換ではなく、複数 loop と決定的処理の外側に協調層を置きます。
• 異質な node を同じ実行モデルで扱います。 LLM agent、通常関数、retrieval、evaluator、人間を同じ graph に配置できます。
• state を transcript から分離します。 schema を持つ task state、artifact、evidence、budget、approval を checkpoint 可能な値として扱います。
• edge を判断と契約として扱います。 固定遷移、条件分岐、model-decided routing、human approval を区別し、決定的に表現できる箇所へ決定的規則を使います。
• 並列性を明示します。 ready な node を fan-out し、all-of、any-of、quorum などの join policy で fan-in します。
• 評価を出力から trajectory へ広げます。 最終回答だけでなく、どの経路を、どの cost と authority で通ったかを評価対象にします。
• failure domain を node 境界へ限定します。 idempotency、retry、timeout、compensation、replan、escalation を node / edge ごとに定義できます。
• 人間を例外処理ではなく実行主体として扱います。 approval、編集、拒否、追加証拠の要求を永続状態を伴う遷移として設計できます。
• budget と authority を状態へ含めます。 token、金額、wall-clock、tool permission、remaining retries を edge crossing で検査できます。
• static と dynamic の trade-off を選択できます。 静的 DAG は監査性を高め、dynamic graph は探索の柔軟性を高めます。
• framework から独立した方法論です。 LangGraph、Google ADK、Microsoft Agent Framework、Temporal、通常の queue / database の組み合わせでも実装できます。

主張に対する批判・反証

arXiv:2604.11378 の扱いに関する注意

Hu Wei の論文は、Graph Engineering の理論的背景として有用です。一方、著者自身が position paper / design proposal と明記し、production implementation と empirical result を提供していません。静的 DAG が loop より優れるという性能主張は、将来検証する仮説です。

同論文の 70-project survey は、著者自身が peer-reviewed survey ではなく qualitative evidence だと明記しています。また、本文は Agent Loop を 42 / 70 と記載する一方、Appendix A.7 の見出しは 41 projects と記載します。Appendix の category 件数は 41 + 11 + 4 + 5 + 7 = 68 で、同じ project 名が複数 category に現れる箇所もあります。したがって、「60%」は産業全体の普及率として一般化せず、著者の sample における概算として扱います。

構造

ここでいう Graph Engineering の対象は、AI エージェントの制御フローです。制御をnode、型付きedge、状態、ゲートとして外在化します。Knowledge Graph Engineering の対象は、知識をnodeと関係として格納する情報構造です。Josh C. Simmons は、nodeを交換可能な能力単位、edgeを意思決定、状態をチェックポイント可能なスキーマとして捉えています。一方、arXiv:2604.11378 の Structured Graph Harness は、静的 DAG、計画・実行・復旧の分離、段階的復旧を採用する厳格な設計案です。以下の図は Graph Engineering 全般を表し、論文固有の制約は境界として明記します。

システムコンテキスト図

この境界では、LLM は制約された推論コンポーネントとして、計画生成、分類、評価、node内部の推論に参加します。検査可能なグラフとポリシーが、実行可否、依存関係、予算、権限、承認を制御します。

コンテナ図

コンテナ境界は方法論上の責務分離です。単一プロセスにも分散サービスにも実装できます。特に、チェックポイントの粒度は製品ごとに異なります。LangGraph はグラフの各 step に状態スナップショットを保存します。Microsoft Agent Framework は superstep の完了時に全 executor 状態、次のメッセージ、保留要求、共有状態を保存します。Temporal は Workflow と Activity の履歴、再試行、待機を耐久化する下位実行基盤を担い、agent graph のモデルは上位アプリケーションやフレームワークが定義します。

コンポーネント図

コンポーネント図は実装可能な共通パターンを示します。各フレームワークの機能と Structured Graph Harness の保証は別々に追跡します。LangGraph と Google ADK は条件付き back-edge や動的経路を表現できます。Structured Graph Harness は計画版内の静的 DAG を選び、動的な構造変更を新しい計画版へ限定します。したがって、Graph Engineering の設計時には、探索性を優先する動的グラフと、監査性を優先する静的グラフのどちらを採用するかを明示します。

また、「各 edge で checkpoint する」は Josh C. Simmons の実践上の推奨です。各製品は独自の保存境界を定義します。実装では、チェックポイント境界、再開時に再実行される範囲、副作用の冪等性をフレームワークの契約に合わせます。LangGraph の interrupt は再開時に中断した node を先頭から実行します。Microsoft Agent Framework は superstep 境界で保存します。Temporal は耐久実行の基盤として Workflow state と Activity 結果を復元します。

データ

本節の Graph Engineering は、AI エージェントの制御フローをグラフとして設計し、実行履歴を追跡するためのデータ設計を指します。ドメイン知識を triple や ontology として表現する Knowledge Graph Engineering とは対象が異なります。ここで扱うグラフの頂点は実行単位であり、辺は依存関係またはルーティング条件です。

起点論文の Graph Harness は、実行計画を Π=(id, version, V, E, σ, κ) と定義します。論文はnode状態、回復プロトコル、予算、承認待ちを定義しています。一方、永続化スキーマや監査台帳は将来の engineering details としています。そのため、以下では論文の概念と、公式フレームワークから補完した「実装モデル案」を区別します。

モデルの出典境界

概念モデル

定義所有

実行所有

保証所有

情報モデル

以下の3図は、ZennのMermaid上限に合わせて定義・実行・保証へ分割した情報モデルです。図には主要属性だけを示し、追加属性と制約は各図の直後の表で補足します。

定義エンティティ

実行エンティティ

保証エンティティ

実行時遷移モデル

Edge は graph version に属する不変の定義、EdgeTraversal は Run に属する実行記録です。この分離により、「経路が定義されていたこと」と「その経路が実際に、どの判定で選ばれたか」を混同しません。

状態更新と reducer

起点論文は global state の merge 方式を定義していません。以下は framework 公式実装から補完した実装モデル案です。

Google ADK の state は serializable な key-value です。session: 相当の無接頭辞、user:、app:、temp: の scope を持ちます。本モデルの StateSchema は scope を明示し、Run 内状態と長期状態の誤混入を防ぎます。Artifact は state にバイナリを埋め込まず、版と URI を state から参照します。

edge condition と終端状態

LangGraph は通常 edge、conditional edge、START、END を持ちます。通常 edge と動的 routing を同じ node から併用すると両経路が動くため、公式資料は一つの routing mechanism を選ぶよう求めています。本モデルでも Node ごとに static または conditional の一方を選びます。

versioning と checkpoint

LangGraph は checkpoint を thread と checkpoint ID で識別し、StateSnapshot に values、next、config、metadata、tasks を保持します。過去 checkpoint の replay では、その checkpoint より前を再利用し、後続 node を再実行します。Microsoft Agent Framework は各 superstep の終了時に全 executor state、次 superstep の pending messages、pending requests and responses、shared state を保存します。したがって Checkpoint は state map だけでなく、次の dispatch と未解決 approval を含む必要があります。

Graph migration の実務では、paused thread が次に参照する node の rename や delete が危険です。LangGraph 公式資料も interrupted thread では node の rename と remove を制約し、state key の rename は保存値を失うと説明しています。本モデルでは進行中 Run を旧 GraphDefinition version に固定し、移行は新規 Run または明示的な checkpoint migration として扱います。

provenance と idempotency

Evidence は「誰が、どの定義で、何を入力し、何を生成し、何で成功と判定したか」を追跡します。最低限、runId、attemptId、graphVersion、policyVersion、sourceRef、contentHash、capturedAt を保持します。Artifact は内容本体または URI を持ち、Evidence はその特定 version と digest を引用します。これにより、可変 URL や latest artifact への参照だけで監査根拠が変わる事態を防ぎます。

durable execution は node の一回だけの実行を自動的には保証しません。LangGraph の interrupt は resume 時に node の先頭から再実行します。checkpoint より後の処理や完了記録前に落ちた task も再実行されます。そのため、外部副作用には次の実装モデルを適用します。

• idempotencyKey = hash(runId, graphVersion, nodeId, logicalEffectId) とし、retry sequence をキーに含めません。
• 外部 API が idempotency key を受け付ける場合は同じ key を渡します。
• DB 更新は unique constraint 付き upsert または compare-and-set を使います。
• 副作用の結果を Artifact と Evidence に保存し、再実行前に確認します。
• approval 前の副作用を避けます。必要な準備操作は別 Node に分け、冪等にします。

budget accounting

Graph Harness は各 Node に timeout と retry budget を持たせ、execution context に budget を含めます。実装では一つの数値へ丸めず、次の ledger を分けます。

並列 dispatch では、実行前に各 NodeAttempt の worst-case cost を reserved に計上します。完了後に実使用量へ精算し、未使用予約を解放します。この予約方式と費用元帳の属性は論文未記載の実装モデル案です。

approval と evidence

Approval は boolean だけで表現しません。request payload digest、対象 action、side-effect level、適用 Policy version、要求者、判断者、判断時刻、decision、任意のコメントを保持します。Microsoft Agent Framework の checkpoint が pending requests and responses を保存し、LangGraph の interrupt が serializable payload と同じ thread ID による resume を要求するため、Approval ID を request と response の相関キーにします。

承認済みでも出力契約の検証は省略しません。Approval は「副作用を実行してよい」という判断です。Evidence は「何が実行され、どの結果が得られ、どの契約に合格したか」という観測です。Evaluation は Evidence を参照して verdict を生成します。この三つを分けることで、人の許可、実行事実、品質判定を混同せずに追跡できます。

構築方法

実装案の位置づけ: Graph Engineering は、AI エージェントを型付き状態と明示的な遷移で構成する方法論です。以下はその意図を LangGraph に写像した参考実装です。Graph Engineering の「標準実装」を示すコードではありません。

ここからのコードは、外部LLMやAPIキーを使わずに制御フローだけを検証します。LLMやtoolを組み込む場合も、state・edge・停止条件の契約を維持したままnode内部を差し替えます。

前提条件とバージョン固定

• Python 3.12 の独立した仮想環境を使います。
• LangGraph 公式 Graph API は StateGraph、START、END、Send、reducer、conditional edge を提供します。
• 本稿の最小例は外部 LLM を呼び出しません。API キーを用意せずに再現できます。
• 調査時の再現確認には langgraph==1.2.9 を使いました。実プロジェクトは lock file で検証済みバージョンを固定します。

uv と pip のどちらか一方を使います。

# uv
uv init --python 3.12 graph-engineering-sample
cd graph-engineering-sample
uv add 'langgraph==1.2.9'
uv run python -c 'import importlib.metadata as m; print(m.version("langgraph"))'

# pip
python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install 'langgraph==1.2.9'
python -c 'import importlib.metadata as m; print(m.version("langgraph"))'

loop から graph への段階移行

出典: Google Developers Blog image（取得日: 2026-07-27）

最初から大きな graph を設計せず、既存 loop の安定した責務を一つずつ明示化します。

出典: Google Developers Blog image（取得日: 2026-07-27）。この図は ADK の例であり、後続の LangGraph コードの API 図ではありません。

typed edge contract を先に固定する

edge を単なる source -> target とせず、遷移を許可する入力状態、判定の版、出力、失敗時の扱いを版管理します。次はフレームワーク非依存の最小例です。

edge_id: evaluate_to_publish
version: 3
source: evaluate
target: publish
preconditions:
  state.status: evaluated
  evaluation.verdict: accepted
  approval.status: approved
condition_ref: policy://publish-gate@7
emits:
  state.status: ready_to_publish
requires_evidence:
  - evaluation_id
  - approval_id
on_contract_violation: stop
on_transient_failure: retry
retry_limit: 2
side_effect_class: external_publish

この契約に対して、存在しない node、到達不能な target、schema 不整合、版なし condition、承認を迂回する path、上限なし retry を静的 lint で拒否します。実行時には EdgeTraversal と Evidence を原子的に保存してから target を dispatch します。

API キー不要の最小 graph

以下の例は、契約抽出、typed state、deterministic node、conditional edge、Send による fan-out、reducer による fan-in、evaluator、gate、budget/stop を一つの graph にまとめます。並列 branch が results を同時更新するため、Annotated[list[str], operator.add] を明示します。

# graph_example.py
import operator
from typing import Annotated, Literal
from typing_extensions import NotRequired, TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import Send


class GraphState(TypedDict):
    request: str
    work_items: list[str]
    work_item: NotRequired[str]
    results: Annotated[list[str], operator.add]
    score: int
    required_score: int
    attempt: int
    max_attempts: int
    status: str


def extract_contract(state: GraphState):
    items = [x.strip() for x in state["request"].split(",") if x.strip()]
    return {"work_items": items, "status": "planned" if items else "invalid"}


def dispatch(state: GraphState):
    if state["status"] == "invalid":
        return END
    return [Send("worker", {"work_item": item}) for item in state["work_items"]]


def worker(state: GraphState):
    # 実際の LLM ・ツール呼び出しと置換する境界です。
    suffix = "failed" if state["work_item"].startswith("fix:") else "done"
    return {"results": [f'{state["work_item"]}:{suffix}']}


def aggregate(state: GraphState):
    return {"status": f'aggregated:{len(state["results"])}'}


def evaluate(state: GraphState):
    completed = {
        item.removesuffix(":done")
        for item in state["results"]
        if item.endswith(":done")
    }
    score = len(completed & set(state["work_items"]))
    return {"score": score}


def choose_next(state: GraphState) -> Literal["gate", "revise", "stop"]:
    if state["score"] >= state["required_score"]:
        return "gate"
    if state["attempt"] >= state["max_attempts"]:
        return "stop"
    return "revise"


def revise(state: GraphState):
    failed = next(
        (item.removesuffix(":failed") for item in state["results"]
         if item.endswith(":failed")),
        None,
    )
    update = {"attempt": state["attempt"] + 1}
    if failed is not None:
        update["results"] = [f"{failed}:done"]
    return update


def gate(state: GraphState):
    return {"status": "accepted"}


def stop(state: GraphState):
    return {"status": "budget_exhausted"}


builder = StateGraph(GraphState)
for name, fn in {
    "extract_contract": extract_contract,
    "worker": worker,
    "aggregate": aggregate,
    "evaluate": evaluate,
    "revise": revise,
    "gate": gate,
    "stop": stop,
}.items():
    builder.add_node(name, fn)

builder.add_edge(START, "extract_contract")
builder.add_conditional_edges("extract_contract", dispatch)
builder.add_edge("worker", "aggregate")
builder.add_edge("aggregate", "evaluate")
builder.add_conditional_edges("evaluate", choose_next)
builder.add_edge("revise", "evaluate")
builder.add_edge("gate", END)
builder.add_edge("stop", END)

graph = builder.compile()

worker は同じ入力に同じ更新を返す決定的な境界です。fix: で始まるitemを失敗させ、revise が同じitemの成功結果を追加します。evaluate はwork itemごとの成功を一度だけ数えるため、再試行回数だけでscoreは増えません。LLMや外部APIを組み込む場合は、その呼び出しをnode内部の明示的な副作用として扱います。

graph definition の静的 lint

compile() はgraph構造を検査します。次の例は、宣言的なgraph仕様に対するlintの最小形で、直前のgraph_example.pyとは独立しています。実プロジェクトでは、この仕様からStateGraphとlint入力の両方を生成するか、graph.get_graph()から実node・edgeを取得し、定義と検査の二重管理を避けます。

# lint_graph.py
from collections import defaultdict, deque
from langgraph.graph import END, START

NODES = {
    "extract_contract", "worker", "aggregate", "evaluate",
    "revise", "gate", "stop",
}
FIXED_EDGES = {
    (START, "extract_contract"),
    ("worker", "aggregate"),
    ("aggregate", "evaluate"),
    ("revise", "evaluate"),
    ("gate", END),
    ("stop", END),
}
ROUTE_TARGETS = {
    "extract_contract": {"worker", END},
    "evaluate": {"gate", "revise", "stop"},
}


def lint_graph_definition() -> list[str]:
    errors: list[str] = []
    allowed = NODES | {START, END}
    edges = set(FIXED_EDGES)
    for source, targets in ROUTE_TARGETS.items():
        edges.update((source, target) for target in targets)

    unknown = {x for edge in edges for x in edge if x not in allowed}
    if unknown:
        errors.append(f"unknown nodes: {sorted(unknown)}")

    forward = defaultdict(set)
    reverse = defaultdict(set)
    for source, target in edges:
        forward[source].add(target)
        reverse[target].add(source)

    def walk(root, adjacency):
        seen, queue = {root}, deque([root])
        while queue:
            for node in adjacency[queue.popleft()]:
                if node not in seen:
                    seen.add(node)
                    queue.append(node)
        return seen

    reachable = walk(START, forward)
    can_finish = walk(END, reverse)
    if unreachable := NODES - reachable:
        errors.append(f"unreachable nodes: {sorted(unreachable)}")
    if trapped := (NODES & reachable) - can_finish:
        errors.append(f"nodes without END path: {sorted(trapped)}")
    if "stop" not in ROUTE_TARGETS["evaluate"]:
        errors.append("evaluate route requires a budget stop target")
    return errors


assert lint_graph_definition() == []

利用方法

必須パラメータと契約

基本実行と budget stop

次の入力は3件をfan-outします。fix:validateだけが初回に失敗し、reviseが同じitemを修復します。3つのwork itemが成功した段階でgateへ進みます。

from graph_example import graph

result = graph.invoke({
    "request": "parse,fix:validate,publish",
    "work_items": [],
    "results": [],
    "score": 0,
    "required_score": 3,
    "attempt": 0,
    "max_attempts": 1,
    "status": "new",
})

assert set(result["results"]) == {
    "parse:done", "fix:validate:failed",
    "fix:validate:done", "publish:done",
}
assert result["attempt"] == 1
assert result["status"] == "accepted"

required_scoreをwork item数より大きい4にすると、再作業後もscoreは3から増えません。1回の再作業でmax_attemptsへ達し、budget_exhaustedで終了します。

stopped = graph.invoke({
    "request": "parse,fix:validate,publish",
    "work_items": [],
    "results": [],
    "score": 0,
    "required_score": 4,
    "attempt": 0,
    "max_attempts": 1,
    "status": "new",
})
assert stopped["attempt"] == 1
assert stopped["status"] == "budget_exhausted"

persistence と human-in-the-loop

interrupt を使う graph は checkpointer 付きで compile します。同じ thread_id と Command(resume=...) を使うと、承認値が interrupt() の戻り値になります。InMemorySaver は局所的な再現用です。実運用では SQLite、PostgreSQL 等の持続 checkpointer を選びます。

from typing_extensions import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt


class ApprovalState(TypedDict):
    change_id: str
    approved: bool
    status: str


pending: dict[str, str] = {}


def upsert_pending(change_id: str) -> None:
    # 同じ key の更新なので再実行時も結果が安定します。
    pending[change_id] = "pending_approval"


def approval(state: ApprovalState):
    upsert_pending(state["change_id"])
    approved = interrupt({
        "question": "Apply change?",
        "change_id": state["change_id"],
    })
    return {
        "approved": bool(approved),
        "status": "approved" if approved else "rejected",
    }


builder = StateGraph(ApprovalState)
builder.add_node("approval", approval)
builder.add_edge(START, "approval")
builder.add_edge("approval", END)
approval_graph = builder.compile(checkpointer=InMemorySaver())

config = {"configurable": {"thread_id": "change-42"}}
paused = approval_graph.invoke(
    {"change_id": "chg-42", "approved": False, "status": "pending"},
    config,
)
print(paused["__interrupt__"])

resumed = approval_graph.invoke(Command(resume=True), config)
assert resumed["status"] == "approved"

resume 時は interrupt() を呼んだ node の先頭から再実行します。そのため interrupt 前の副作用には upsert と idempotency key を使います。新規作成や外部通知は interrupt 後の別 node に配置すると、重複実行の境界が明確になります。

node 契約テスト

node 単体の決定性、route の列挙範囲、reducer 必須キーを、外部 API から切り離して検査します。

# test_graph_contracts.py
from typing import Annotated, get_origin, get_type_hints

from graph_example import GraphState, choose_next, extract_contract, worker
from lint_graph import lint_graph_definition


def test_nodes_are_deterministic_and_schema_bounded():
    state = {
        "request": "parse,validate",
        "work_items": [],
        "results": [],
        "score": 0,
        "required_score": 2,
        "attempt": 0,
        "max_attempts": 1,
        "status": "new",
    }
    assert extract_contract(state) == extract_contract(state)
    assert worker({**state, "work_item": "parse"}) == {"results": ["parse:done"]}
    assert set(extract_contract(state)) <= set(GraphState.__annotations__)


def test_routes_and_reducer_contract():
    state = {
        "request": "x", "work_items": ["x"], "results": ["x:done"],
        "score": 1, "required_score": 1, "attempt": 0,
        "max_attempts": 1, "status": "aggregated:1",
    }
    assert choose_next(state) in {"gate", "revise", "stop"}
    hints = get_type_hints(GraphState, include_extras=True)
    assert get_origin(hints["results"]) is Annotated
    assert lint_graph_definition() == []

故障注入と checkpoint からの recovery

故障注入では、成功済み node の状態が checkpoint に残り、同じ thread_id と None 入力で未完了 node から再開することを確認します。

from typing_extensions import TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph


class RecoveryState(TypedDict):
    trace: list[str]


failure = {"remaining": 1}


def stable(state: RecoveryState):
    return {"trace": state["trace"] + ["stable"]}


def flaky(state: RecoveryState):
    if failure["remaining"]:
        failure["remaining"] -= 1
        raise TimeoutError("injected")
    return {"trace": state["trace"] + ["flaky"]}


def finish(state: RecoveryState):
    return {"trace": state["trace"] + ["finish"]}


b = StateGraph(RecoveryState)
b.add_node("stable", stable)
b.add_node("flaky", flaky)
b.add_node("finish", finish)
b.add_edge(START, "stable")
b.add_edge("stable", "flaky")
b.add_edge("flaky", "finish")
b.add_edge("finish", END)
recovery_graph = b.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "fault-1"}}

try:
    recovery_graph.invoke({"trace": []}, config)
except TimeoutError:
    pass

snapshot = recovery_graph.get_state(config)
assert snapshot.values["trace"] == ["stable"]
assert snapshot.next == ("flaky",)

recovered = recovery_graph.invoke(None, config)
assert recovered["trace"] == ["stable", "flaky", "finish"]

履歴からの replay

get_state_history() は新しい checkpoint から順に返します。調査したい checkpoint の config を invoke(None, checkpoint.config) に渡すと、その分岐から実行を再現できます。本番データに影響する replay は、副作用をモックまたは隔離環境へ差し替えます。

history = list(recovery_graph.get_state_history(config))
before_flaky = next(item for item in history if item.next == ("flaky",))
replayed = recovery_graph.invoke(None, before_flaky.config)

assert replayed["trace"] == ["stable", "flaky", "finish"]

調査環境での再現確認

掲載コードは構文例として眺めるだけでなく、2026-07-27 に隔離した uv 環境で実行しました。外部 API と API key は使用していません。

依存関係を固定して一時 harness を実行したコマンドの基準形は次です。graph_example.py 単体は graph を定義・compile するコードであり、経路 assertion は「基本実行と budget stop」のコードも同じファイルへ続けて保存して実行します。approval と recovery は state 型が異なるため別スクリプトとして検証します。

uv run --with 'langgraph==1.2.9' python graph_example.py
npx md-mermaid-lint 'articles/graph-engineering_20260727.md'

langgraph==1.2.9 は検証時点の PyPI 公開版です。将来の最新版での動作を保証する記述ではないため、導入時には lock file と CI で再検証します。

実装選択の読み替え

Graph Engineering の設計要素は複数の実行基盤に写像できます。各製品の API を共通仕様のように混ぜず、以下の対応関係だけを利用します。

Temporal では Workflow コードの決定性を保ち、LLM や外部 API の呼び出しを Activity に配置します。この境界は、Graph Engineering の deterministic control plane と probabilistic work node の分離に応用できます。

運用

運用上の前提

Graph Engineering は、AI エージェントの実行単位をnode、許可された遷移をedgeとして扱う設計上の呼称です。2026 年 7 月時点では標準規格や確立済みの独立分野ではありません。arXiv:2604.11378 の Structured Graph Harness は position paper と設計提案であり、本番実装と実験結果を伴いません。論文が主張する性能向上、安定性向上、コスト効率は将来の実験で検証する仮説です。

本番運用では、Graph Engineering 固有の用語より、durable execution、イベント履歴、冪等実行、分散トレーシング、バックプレッシャー、dead-letter queue、SLO、canary、incident response という既存の分散システム原語を正本にします。論文自身も、永続化、ハートビート、ワーカー再配置、リーダー選出を既存の workflow engine と distributed systems が扱ってきた課題として位置づけています。

運用ポリシーを graph definition から分離する

topology と、環境ごとに変わる予算・retry・承認閾値を別 artifact として版管理します。Run は両方の版を固定し、途中変更を反映しません。次は設定例であり、標準仕様ではありません。

policy_id: production-default
version: 12
admission:
  max_active_runs_per_tenant: 20
budgets:
  run_token_limit: 250000
  run_cost_usd_limit: 25
  wall_clock_minutes: 120
recovery:
  node_retry_limit: 3
  graph_replan_limit: 1
  require_idempotency_for_side_effects: true
approval:
  external_publish: human
  destructive_action: two_person
backpressure:
  queue_age_seconds_warn: 300
  queue_age_seconds_reject: 1800
retention:
  event_days: 90
  redact_model_content: true

policy 更新は canary run で旧版と比較し、成功率だけでなく token、遅延、escalation、human override を判定します。上限超過は例外ログだけでなく、budget_exhausted や escalated という明示的な終端状態へ遷移させます。

run・node・edge のトレース

• 1 回の graph run を root span にします。各 node 実行を child span にします。
• edge は、単なる静的定義ではなく「どの条件、どの証拠、どの plan version により遷移したか」をイベントとして記録します。
• 並列分岐は共通の parent span から分けます。別 trace に分離する場合は span link で因果関係を保持します。
• trace_id、run_id、graph_version、node_id、attempt、input_artifact_ids、output_artifact_ids を相関キーにします。
• OpenTelemetry の invoke_agent、chat、execute_tool と token usage 属性を利用できます。Graph 固有 ID はアプリケーション側の custom attribute として名前空間を固定します。
• prompt、completion、tool result は機微情報を含みます。本文記録を opt-in にし、既定では hash、サイズ、分類ラベル、artifact ID を記録します。

以下は JSONL の実行履歴から、node の開始と終端が対応するかを確認する診断例です。これは Graph Engineering の規格ではなく、運用実装例です。

#!/usr/bin/env python3
import json
import sys
from collections import defaultdict

events = defaultdict(list)
for line in open(sys.argv[1], encoding="utf-8"):
    event = json.loads(line)
    key = (event["run_id"], event["node_id"], event["attempt"])
    events[key].append(event["type"])

terminal = {"node.succeeded", "node.failed", "node.skipped", "node.cancelled"}
broken = []
for key, types in events.items():
    starts = types.count("node.started")
    ends = sum(types.count(t) for t in terminal)
    if starts != 1 or ends != 1:
        broken.append({"key": key, "starts": starts, "ends": ends, "types": types})

print(json.dumps(broken, ensure_ascii=False, indent=2))
raise SystemExit(1 if broken else 0)

state・checkpoint・replay

• graph definition、runtime state、artifact を分離します。graph definition は versioned immutable artifact にします。
• run state は node transition の append-only event と定期 checkpoint で保存します。checkpoint 単体より event history を監査の正本にします。
• LangGraph の checkpointer は thread 単位の graph state snapshot を保存し、fault tolerance、time travel、human-in-the-loop に利用できます。長期・thread 横断データは store に分けます。
• Temporal は Workflow の進行を永続化し、障害後に同じ履歴を replay して再開します。LLM 呼び出しや外部 API のような非決定的処理は Activity に隔離し、結果を履歴へ固定します。
• replay はモデルを再呼び出す操作と区別します。監査 replay は保存済み output を読む処理にします。再評価は別 run ID と parent run ID を持つ新規実行にします。
• checkpoint には prompt や secret の平文を保存せず、暗号化、短期 retention、参照 ID、redaction を組み合わせます。

以下は checkpoint と node event の整合性を確認する SQL 例です。テーブル名は実装例です。

WITH latest_checkpoint AS (
  SELECT run_id, MAX(sequence_no) AS checkpoint_sequence
  FROM graph_checkpoints
  GROUP BY run_id
), latest_event AS (
  SELECT run_id, MAX(sequence_no) AS event_sequence
  FROM graph_events
  GROUP BY run_id
)
SELECT e.run_id, c.checkpoint_sequence, e.event_sequence,
       e.event_sequence - c.checkpoint_sequence AS unapplied_events
FROM latest_event e
LEFT JOIN latest_checkpoint c USING (run_id)
WHERE c.checkpoint_sequence IS NULL
   OR e.event_sequence - c.checkpoint_sequence > 1000
ORDER BY unapplied_events DESC;

SLO と予算

可用性だけでなく、終了品質と資源消費を同じ run に結び付けます。初期値は実測に基づき、例示値をそのまま本番目標に転用しません。

Google SRE の error budget と同様に、SLO 超過時の行動を先に文書化します。Graph Engineering では「品質を満たさない run」「deadline 超過」「予算超過」も error budget 消費へ含めます。予算は token だけでなく、wall-clock、tool call、並列 node 数、外部副作用回数を同時に制限します。

以下は run budget を診断し、次の node を許可するかを決める実装例です。

from dataclasses import dataclass

@dataclass(frozen=True)
class Budget:
    max_input_tokens: int
    max_output_tokens: int
    max_wall_seconds: float
    max_tool_calls: int

def admission(used: dict, estimate: dict, budget: Budget) -> tuple[bool, list[str]]:
    checks = {
        "input_tokens": used["input_tokens"] + estimate["input_tokens"] <= budget.max_input_tokens,
        "output_tokens": used["output_tokens"] + estimate["output_tokens"] <= budget.max_output_tokens,
        "wall_seconds": used["wall_seconds"] + estimate["wall_seconds"] <= budget.max_wall_seconds,
        "tool_calls": used["tool_calls"] + estimate["tool_calls"] <= budget.max_tool_calls,
    }
    violations = [name for name, ok in checks.items() if not ok]
    return not violations, violations

queue・backpressure・dead-letter・escalation

• ready node を直接無制限に起動せず、priority queue と admission control を通します。
• provider、model、tool、tenant、failure domain ごとに concurrency limit と rate limit を設定します。
• queue depth より queue age を主な利用者影響指標にします。古い task が新しい task に埋もれる starvation を監視します。
• fan-out 上限を run 単位と全体単位に持たせます。下流の許容量に合わせて親 node の生成速度を落とします。
• transient error、contract error、policy denial、budget exhaustion、human decision required を別の error class にします。
• retry は指数 backoff、jitter、maximum attempts、maximum elapsed time を持ちます。Temporal の Activity は既定で retry されるため、agent workload では明示上限を設定します。
• 同一 payload が上限回数を超えた場合は DLQ に移します。AWS SQS の maxReceiveCount と redrive は、この既存パターンの具体例です。
• DLQ は放置場所ではなく、owner、SLA、redrive 条件、廃棄条件を持つ triage queue とします。
• 論文の三段階 recovery は、local retry、bounded patch、full replan の順です。本番では、その外側に policy denial、human approval、incident escalation を追加します。

以下は queue、trace、budget の観測値から stalled run を列挙する診断例です。

SELECT run_id,
       graph_version,
       COUNT(*) FILTER (WHERE state = 'ready') AS ready_nodes,
       COUNT(*) FILTER (WHERE state = 'blocked') AS blocked_nodes,
       COUNT(*) FILTER (WHERE state = 'running') AS running_nodes,
       MAX(EXTRACT(EPOCH FROM (CURRENT_TIMESTAMP - ready_at)))
         FILTER (WHERE state = 'ready') AS oldest_ready_seconds,
       MAX(EXTRACT(EPOCH FROM (CURRENT_TIMESTAMP - state_entered_at)))
         FILTER (WHERE state = 'blocked') AS oldest_blocked_seconds,
       SUM(input_tokens + output_tokens) AS tokens_used,
       MAX(attempt) AS max_attempt
FROM graph_node_attempts
WHERE run_state IN ('queued', 'running', 'waiting')
GROUP BY run_id, graph_version
HAVING MAX(EXTRACT(EPOCH FROM (CURRENT_TIMESTAMP - ready_at)))
         FILTER (WHERE state = 'ready') > 300
    OR MAX(EXTRACT(EPOCH FROM (CURRENT_TIMESTAMP - state_entered_at)))
         FILTER (WHERE state = 'blocked') > 300
    OR MAX(attempt) > 3
ORDER BY oldest_ready_seconds DESC NULLS LAST;

graph version migration・canary・replay

• graph definition、node contract、policy、prompt、model routing、evaluator を個別に version 管理します。run には解決済みの version set を固定します。
• in-flight run は開始時の graph version で完了させる方式を基本にします。強制 migration は変換関数、互換性検証、rollback point を用意した場合に限定します。
• LangGraph は完了済み thread の topology 変更を扱えます。interrupt 中の thread では、次に進む可能性がある node の rename と remove に制約があります。state key の rename は既存 state を失うため、copy-and-deprecate migration を使います。
• canary は run の一部を新しい graph version に固定して流します。Anthropic が multi-agent research system で説明する rainbow deployment と同様に、旧版と新版を並行稼働させ、長時間 run を途中で壊さず traffic を移します。
• canary 判定は success、quality、latency、token、tool error、human override、DLQ の差を同じ task class で比較します。
• replay 検証は保存済み nondeterministic output を使います。モデルを再呼び出した結果との差は regression test として別管理します。

以下は version 間で同じ保存済み node output を流し、routing 差分を検査する擬似コードです。

def replay_route(graph, event_history):
    state = graph.initial_state(event_history[0].input)
    route_log = []
    for event in event_history:
        if event.type == "node.output.recorded":
            state = graph.apply_recorded_output(state, event.node_id, event.output)
            route_log.append(graph.next_nodes(state, event.node_id))
    return route_log

old_routes = replay_route(load_graph("2026-07-20"), load_history("run-123"))
new_routes = replay_route(load_graph("2026-07-27"), load_history("run-123"))
for index, (old, new) in enumerate(zip(old_routes, new_routes)):
    if old != new:
        print({"step": index, "old": old, "new": new})

incident response

1. run intake を止め、影響する graph version、tenant、tool、model provider を切り分けます。
2. in-flight run を freeze、cancel、safe-complete のいずれかに分類します。副作用済み node は compensation の要否を確認します。
3. trace から最初の failure node、直前 edge、共有 state、同一 idempotency key を確認します。
4. credential leak または privilege misuse を含む場合は、secret rotation と downstream revoke を先に実行します。
5. replay は保存済み出力で再現します。外部世界の状態は reality anchor の時点情報と突き合わせます。
6. canary を旧安定版へ戻し、DLQ と保留中 human approval を棚卸しします。
7. blameless postmortem に、detection、containment、recovery、graph design、contract、budget、permission の改善項目を残します。

ベストプラクティス

実行の安全性

契約と評価

• node 入出力を typed contract にします。JSON Schema、Pydantic、Protocol Buffers などを使い、自然言語の期待だけに依存しません。
• schema validation と semantic validation を分離します。形式合格を事実合格として扱いません。
• evaluator は可能な限り deterministic にします。unit test、SQL constraint、policy engine、checksum、compiler、external API status を優先します。
• LLM evaluator を使う場合は、実行 agent と model、context、prompt を分離し、calibration set で再現性を測ります。
• external reality anchor を保存します。URL、取得時刻、commit SHA、artifact digest、database version、ticket ID を結果へ結び付けます。
• approval と audit を実行 agent から独立させます。高リスク action では同一 agent の自己承認を許可しません。

権限とデータ保護

• least privilege を node 単位で適用します。graph 全体で 1 個の強権限 credential を共有しません。
• read、propose、approve、execute を別 capability にします。子 agent の権限は親の権限以下に制限します。
• OWASP の Excessive Agency 対策に合わせ、利用可能 tool、tool の機能、downstream identity、scope を必要最小限にします。
• PII、secret、prompt、tool result は収集前に redaction します。trace backend に原文を送る場合は opt-in、暗号化、retention、access audit を必須にします。
• checkpoint と long-term memory を信頼済みデータとして扱いません。provenance、TTL、write authority、quarantine を持たせます。

並列処理と merge

• 並列化は独立性を証明できる node に限定します。共有ファイル、同一 database row、同一 checkpoint namespace への write は直列化または reducer を使います。
• merge policy を edge より先に定義します。append、set union、last-write-wins、rank-and-select、human resolution を型ごとに固定します。
• Anthropic の実運用報告では、multi-agent system は chat の約 15 倍の token を消費し、非同期化には state consistency と error propagation の課題があります。この値は同社の research system における観測であり、一般的な固定倍率ではありません。
• LangGraph も parallel branch が同じ state key を reducer なしで更新すると concurrent update error を返します。parallelism は latency の短縮と同時に、誤りの増幅、merge 競合、token・tool cost の増加を持ち込みます。

反証・限界・適用条件

主張を過大評価しないための反証

iii.dev の批判はベンダ自身の立場を含む論説です。ただし「新しいラベルの下で queue、state、retry、trace、dead-letter を再発明しない」という検査観点は、Temporal、OpenTelemetry、SQS、SRE の一次資料と整合します。

方式の適用判断

選択順序は、単一 prompt、loop、決定的 workflow、graph の順に複雑性を追加します。複数 agent の存在だけを graph 採用理由にしません。既知の決定的処理へ LLM node を追加する場合は、workflow を基礎にして該当 step だけを非決定的 Activity として隔離します。

トラブルシューティング

症状からの切り分け

並列 write の診断

以下は同じ run・step・state key に複数 node が書いた箇所を検出します。

SELECT run_id, superstep, state_key,
       COUNT(DISTINCT node_id) AS writers,
       ARRAY_AGG(DISTINCT node_id ORDER BY node_id) AS node_ids
FROM graph_state_writes
GROUP BY run_id, superstep, state_key
HAVING COUNT(DISTINCT node_id) > 1
ORDER BY run_id, superstep, state_key;

effect ledger の重複診断

以下は同じ idempotency key に異なる payload digest が対応する危険な状態を検出します。

SELECT idempotency_key,
       COUNT(*) AS attempts,
       COUNT(DISTINCT payload_digest) AS payload_variants,
       COUNT(*) FILTER (WHERE effect_status = 'committed') AS commits
FROM tool_effect_ledger
GROUP BY idempotency_key
HAVING COUNT(DISTINCT payload_digest) > 1
    OR COUNT(*) FILTER (WHERE effect_status = 'committed') > 1;

予算超過 run の停止確認

以下は予算を超えた後にも node が開始された run を検出します。

#!/usr/bin/env python3
import json
import sys

events = [json.loads(line) for line in open(sys.argv[1], encoding="utf-8")]
exhausted_at = {}
violations = []
for event in sorted(events, key=lambda x: (x["run_id"], x["sequence_no"])):
    run_id = event["run_id"]
    if event["type"] == "budget.exhausted":
        exhausted_at.setdefault(run_id, event["sequence_no"])
    if event["type"] == "node.started" and run_id in exhausted_at:
        violations.append({
            "run_id": run_id,
            "node_id": event["node_id"],
            "started_after": exhausted_at[run_id],
        })

print(json.dumps(violations, ensure_ascii=False, indent=2))
raise SystemExit(1 if violations else 0)

まとめ

Graph Engineeringの本質は、loopを捨てることではありません。loopをnode内部へ閉じ込め、node間のstate、権限、予算、停止、回復を検査可能な契約として設計することです。まず単一call、loop、決定的workflowを検討し、並列分岐・長期再開・監査可能なhuman gateが必要になった段階でgraphを採用してください。

この記事が参考になった、あるいは改善点が見つかった場合は、リアクションやコメント、SNSで共有していただけると励みになります。

関連記事

• Loop Engineering入門 — graph が node 内部へ閉じ込める loop そのものを設計する
• AI Harness Engineering入門 — node を実行する層の責務と証跡を契約として定義する
• Context Engineering入門 — 型付き state に載せる情報を、何を渡すかの側から設計する
• プロンプトエンジニアリングの進化 — node 内部の 1 回の model 呼び出しに与える指示を設計する
• LangGraph技術調査

参考リンク

論文・記事

• Peter Steinberger の起点投稿

• We Are Entering the Graph Engineering Phase — Josh C. Simmons

• From Agent Loops to Structured Graphs — arXiv abstract

• From Agent Loops to Structured Graphs — arXiv HTML full text

• Loops, Graphs, and the Layer That Matters — iii

• Building effective agents — Anthropic

• How we built our multi-agent research system — Anthropic

• Why Do Multi-Agent LLM Systems Fail? v1 — arXiv

• Unifying Language Agent Algorithms with Graph-based Orchestration Engine — arXiv

公式ドキュメント

• LangGraph overview

• LangGraph Graph API

• LangGraph use the Graph API

• LangGraph Persistence

• LangGraph Interrupts

• LangGraph Test

• Microsoft Agent Framework Workflows

• Microsoft Workflow Builder and Execution

• Microsoft Workflow Checkpoints

• Google ADK Graph workflows

• Google ADK Template workflows

• Why we built ADK 2.0 — Google Developers Blog

• Temporal Workflow Execution

• Temporal Retry Policies

• OpenTelemetry GenAI observability

• OpenTelemetry trace semantic conventions

• Google SRE Workbook — Error Budget Policy

• Google SRE Workbook — Canarying Releases

• Amazon SQS visibility timeout

• Amazon SQS dead-letter queues

• OWASP LLM06:2025 Excessive Agency