Learn Harness Engineering プロジェクト08：ワークフローをグラフとして描く——ループからグラフエンジニアリングへの実践ガイド

本記事は Learn Harness Engineering の演習プロジェクト 08（docs/ja/projects/project-08-graph-engineering-first-graph/index.md）を主体的な骨格として、対応講義 第14回 単一ループからグラフエンジニアリングへ の概念と、実際の参照実装 maker_checker_graph.py のソースコードを併せて深掘りする技術記事です。

P07 で構築した maker-checker ループ（実装→検証→フィードバック→再実装）は、すべての決定が一つのエージェントのコンテキストウィンドウの中に閉じていました。本プロジェクトは、そのループの内部に隠れていた構造——ノード、エッジ、共有状態、ルーティングルール——を一言一言明示的に書き出し、3つの発展的実験（明示グラフ化・並列 fan-out/fan-in・条件付きフォールバック＋人間承認）を通じて「グラフは新しい発明ではなく、ループが一定の複雑さに達したときに自然に変わる姿である」ことを体感することを目指します。読み終えると、あなたは自分のワークフローをgraph.md一枚に描き起こし、LangGraph等のグラフ実行エンジン上で実行可能なプログラムへ変換する一連の実践スキルを身につけられます。

このプロジェクトの位置づけ：Loop から Graph への躍進

第13回の講義 手動プロンプトから自律ループへ で構築した maker-checker ループは、/goalを入力すれば検証可能な停止条件に達するまで自律的にループし続ける「一つのエージェント」でした。しかし、タスクが複雑になると、単一ループには以下の4つの問いが浮上します（第14回講義 より）：

1. 分業：要件を研究するエージェント、コードを書くエージェント、テストするエージェント——誰が先に始めるのか？
2. 並列：どの作業を同時に進められるのか？
3. フォールバック：テストが失敗したらどこへ戻るのか——実装ノードか、それとも研究ノードか？
4. 引き継ぎ：複数のエージェントが同じ要件・ノート・テスト結果をどう共有するのか？レビュアーと実装者が対立したら、どちらが勝つのか？

ループは「先延ばしの決定」であり、アーキテクチャを後回しにできる代わりに失敗モードが見えません。グラフは「先回りの決定」であり、構造全体を事前に宣言することで読みやすく、監査しやすく、部分修復が可能になります。ループは問題をループの中に隠し、グラフは問題を紙の上に並べる——本プロジェクトはこの転換を、机上の理論ではなく手を動かす実験として行います。

グラフの4つの部品を理解する

第14回講義はグラフを最も素朴な4つの部品に還元します。本プロジェクトの全実験はこの4つの部品を操作する練習です。

典型的な開発グラフは以下のようになります（第14回講義より）：

前回のループ図（発見→ディスパッチ→検証→永続化→再び発見の一つの環）と比べてください。環はまだ存在しますが、明示的なノードとエッジに分解されています。検証ノードは失敗を直接実装ノードに打ち返せますし、実装ノードは情報不足で研究ノードに戻れます——こうした「フォールバックエッジ」は単一ループでは暗黙的で、エージェント自身がコンテキストの中で「戻るべきだ」と記憶しているだけでした。

準備：3つのブランチと共有状態ファイルstate.md

P07 完了後のリポジトリ（または現在実行中のエージェントワークフロー）から始め、以下の準備を行います。

git checkout -b p08-explicit-graph git checkout -b p08-parallel git checkout -b p08-human-in-the-loop

state.mdを共有状態ファイルとして用意します。要件・進捗・検証結果をすべてここに書きます。これがグラフの「公共の作業台」です。ノード同士は直接会話せず、すべて同じ状態を読み書きします。第14回講義の言葉を借りれば、グラフレイヤーで共有されるのは状態だけであり、ノードのコンテキストはプライベート——ループはノードの私物、グラフはそれらが引き継ぐ公共の台です。

# state.md — 共有状態 - 要件（requirements）: ... - 進捗（progress）: ... - 検証結果（verification）: ... - レビュー結論（review）: pass / fail / unclear

実験1：Loop を明示的なグラフに描く（p08-explicit-graphブランチ）

ステップ1-1：すべてのノードを列挙する

P07 の maker-checker ループの各ステップを1つのノードとして書き出します。各ノードに必ず以下を明記します：その責務、その入力、その出力、それがエージェントか決定論的コードか。

第14回講義のノード表がそのまま雛形になります：

特にverify ノードに注意してください。これはグラフの中で最も間違えやすいノードです。モノリシックエージェントでは「レビュー」も同じコンテキストを使い、自分で自分を審査してしまいます。グラフでは verify は必ず新しいコンテキストを持ち、実装者の思考過程は見えず、共有状態のcodeだけが見えます。コンテキストの隔離は副作用ではなく設計なのです——これは第9回講義 エージェントが早すぎる完了宣言をする理由 が「検証ノードは実装ノードから独立していなければならない」と述べた構造的な理由そのものです。

ステップ1-2：すべてのエッジを描く

ノード間の各エッジを列挙し、2つの特別なエッジに重点を置いて印を付けます：

• 条件エッジ：検証の合格/不合格——それぞれどちらへ進むか
• フォールバックエッジ：失敗がどのノードに戻るか

ステップ1-3：共有状態を書く

状態にどのフィールドがあるか（要件、コード、テスト結果、レビュー結論）、誰が読み誰が書くかを明確に列挙します。第14回講義のstate定義が参考になります：

state = { "requirements": テキスト, # 研究ノードが書き込む "code": テキスト, # 実装ノードが書き込む "review": "pass" | "fail", # レビューノードが書き込む "attempts": 数値, # 失敗ごとに +1（並列書き込み時は「合計」でマージ） }

ステップ1-4：ルーティングルールを書く

「次にどこへ行くか」のルールを最も簡単な if-then の言葉で書きます：

if 検証合格 → マージノード if 検証失敗 → 実装ノード if 実装ノード情報不足 → 研究ノード

ステップ1-5：graph.mdにまとめる

上記の内容を1つのドキュメントにまとめます。mermaid で図を描き、ノード表とルーティングルールを添えます。このgraph.mdが以後の設計図（ブループリント）になります——後述の実験3後の最終提出物でも、このドキュメントが「グラフの記述と実際の実行が一致するか」を検証する基準になります。

ステップ1-6：この問いに答える——暗黙のエッジを探す

描き終わったら、もともと暗黙的だったエッジを少なくとも1つ見つけます。それは以前はエージェントのコンテキストの中に隠れていて、あなた自身も存在を知らなかった決定経路です。第14回講義はこう述べます：単一ループでは「検証ノードが失敗を実装ノードに打ち返す」「実装ノードが情報不足で研究ノードに戻る」といったフォールバックエッジは、エージェントが「戻るべきだ」とコンテキスト内で記憶しているだけ——グラフ化とは、この「記憶」を紙の上に固定することです。図を描いて初めて、自分のワークフローが実際にはどれだけ未モデル化だったかを認めることになります（Luis Catacora の言う「ループには大量の容錯余地がある。グラフは、ワークフローの中にまだ本当にモデル化されていない部分がどれだけあるかを、あなたに認めさせてしまう」）。

実験2：並列の Fan-out / Fan-in ノードを追加する（p08-parallelブランチ）

ステップ2-1：並列にできる箇所を選ぶ

タスクの中で2つの独立した部分に分割できる場所を見つけます。例：

• 実装の並列化：実装を2つの独立したモジュールに分割し、2つのエージェントが並列で書く
• 検証の並列化：検証を2つの独立したレビューに分割——1つはテストと lint を実行、もう1つはコードレビュー（異なる指示、異なる注目点）
• 研究の並列化：研究を2つの方向に分割し、2つのエージェントがそれぞれ1ルートずつ調べる

ステップ2-2：fan-out ルールを書く

共有状態に「このタスクが N 個の並列サブタスクに分割された」ことを記録します。各サブタスクは独立したコンテキスト、独立したノードを持ちます。エッジの「並列」表現は「A が完了したら、B と C が同時に始まる」という形です。

ステップ2-3：fan-in ルールを書く

すべてのサブタスクが完了した後、誰が結果をマージするのか？マージの基準は何か？（例：両方のレビューが通ってからマージするのか、それとも1つ通ればよいのか）。第14回講義はこの点を「各フィールドに『どうマージされるか』を宣言する——複数の並列ノードが同時に同じフィールドに書き込むとき、上書きか、追加か、それとも合計か」と説明します。attemptsフィールドのように並列書き込み時に「合計」でマージするフィールドと、「上書き」でよいフィールドを、あらかじめgraph.mdに書き込んでおくのです。

ステップ2-4：worktree で隔離する

各並列サブタスクを独立した git worktree で実行し、物理的にファイルの衝突を避けます（第13回講義の Worktree プリミティブを復習）。複数のエージェントを実行すると、ファイルの衝突が必然的な故障モードになります。git worktreeは各エージェントに独自のディレクトリ・独自のブランチを与え、物理的にお互いのチェックアウトに触れさせません。

ステップ2-5：一度実行して記録する

並列化の前後でwall-clock 時間、token 消費、結果の品質を記録します。並列は本当に速くなったのか？それとも調整のオーバーヘッドが節約した時間を食い潰したのか？第14回講義の「オーケストレーション税」（Addy Osmani）を思い出してください：エージェントを起動するのは安いが、ループを閉じるのは高い——あなたのレビュー帯域幅は依然として天井であり、ノードを足しても並列化されない直列リソースです。この記録は、あなた自身がオーケストレーション税を実測する機会です。

実験3：フォールバックエッジと人間による承認ノードを追加する（p08-human-in-the-loopブランチ）

これは3つの実験の中で最も重要なものです。グラフに2種類のノードを追加します。

ステップ3-1：条件付きフォールバックエッジ

検証ノードに「部分合格」の経路を追加します——全体を実装ノードに打ち返すのではなく、具体的なフィードバックを添えて問題を生んだそのノードに戻します。例：テストは全部通ったが、コードレビューで要件の理解に誤りがあると判明した場合、実装ノードではなく研究ノードに戻します。これには、共有状態に「問題がどのレイヤーで起きたか」を記録する必要があります。

第14回講義の言葉で言えば、これは「ルーティングルールが返すのはノードの名前」という原則の応用です。検証ノードは「マージ」に直接つなぐのではなく、一つの決定につなぎ、そこで次の行き先を決めます。表にすると：

ステップ3-2：人間による承認ノード（Human-in-the-loop）

マージノードの前に人間ノードを追加します。ここまで来たら、グラフは停止して、あなたがstate.mdに「承認」または「打ち返し」を書くのを待ちます。第14回講義はこれを checkpoint の応用として説明します：「merge の前に『一時停止して人間の承認を待つ』ノードを挿入することもできます。これが前回の講義の『人間による承認』がグラフ上でどう見えるかです」。

checkpoint = on(graph, every_step) # 各ステップの状態を保存する graph.pause_before("merge") # マージ前に停止し、人間の承認を待つ

承認ノードにはタイムアウトルールを持たせられます：N 時間後に応答がなければ、自動的に打ち返すか自動的にエスカレーションします。

ステップ3-3：interrupt のフォーマットを書く

承認リクエストをどう明確に書くか——何が起きたか、何を変更したか、なぜ人間が必要か、承認/打ち返しの結果はそれぞれ何か。これは第14回講義が言う「グラフが複雑になるほど、各ノードが何をしているかを見る必要がある」（第11回講義 観測可能性）と直結します。承認リクエストが不明確なら、人間ノードは観測不可能なブラックボックスになります。

ステップ3-4：完全なフローを少なくとも2ラウンド実行する

各ラウンドで人間による承認ノードまで進み、あなた自身が1回承認または打ち返しを行います。記録すべきことは：あなたの承認判断は検証ノードの判断と一致したか？承認ノードが、検証ノードが止められなかった何かを止めたか？これは第9回講義の「検証ノードの独立」と第14回講義の「verify は fresh context を持つ」が、実際にどの程度機能しているかを検証する実験です。

結果の測り方

提出物

• graph.md（実験1の完全なグラフ記述：mermaid 図 + ノード表 + エッジ表 + 共有状態フィールド + ルーティングルール）
• 実験1で発見した暗黙のエッジのリスト（少なくとも1本）
• 実験2の fan-out/fan-in ルールと一回の並列実行記録（時間/コスト/品質の比較）
• 実験3のフォールバックエッジのルール、承認ノードのフォーマット、2ラウンドの人間と機械の協働記録
• 最終的な振り返り：loop から graph へ、あなたの働き方はどう変わったか？どのタスクが図を描く価値があり、どのタスクが価値がないのか？

補足：graph.mdを実行可能なプログラムにする（講義14の6ステップと参照実装）

本プロジェクトのgraph.mdは設計図にすぎません。第14回講義の演習5に従えば、この設計図を実際に動くグラフとして実装できます。どのエンジンにも依存しない6ステップが示されています：状態の定義 → ノードの列挙 → エッジの接続 → ルーティングの記述 → checkpoint の設置 → 実行。

その参照実装が docs/en/lectures/lecture-14-graph-engineering/code/maker_checker_graph.py（LangGraph 製）です。ソースコードは上記6ステップと一対一で対応しています：

from typing import Annotated, TypedDict import operator from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver # ---------- Step 1: 共有状態を定義する ---------- class GraphState(TypedDict): requirements: str # research ノードが書く code: str # implement ノードが書く review: str # pass / fail / unclear attempts: Annotated[int, operator.add] # 失敗ごとに +1（並列書き込み時は「合計」でマージ） # ---------- Step 2: ノードを列挙する ---------- def research(state: GraphState) -> dict: # agent ノード: 問題を特定し要件文を作る requirements = call_model("You are a research agent", f"Analyze this problem: {state.get('requirements', '')}") return {"requirements": requirements} def implement(state: GraphState) -> dict: # agent ノード: コード + テストを書く code = call_model("You are an implementation agent", f"Implement against: {state['requirements']}") return {"code": code} def verify(state: GraphState) -> dict: # agent ノード: 独立レビュー + テスト実行（実装者のコンテキストを継承しない） review = call_model("You are an independent reviewer", f"Review this code: {state['code']}") passed = tests_pass(state["code"]) verdict = "pass" if passed and "approved" in review else "fail" return {"review": verdict} def merge(state: GraphState) -> dict: # 決定論的ノード: commit print(f"Merging code (passed after {state['attempts']} attempts)") return {} # ---------- Step 4: ルーティングルールを書く（最も重要なステップ） ---------- def route_after_verify(state: GraphState) -> str: if state["review"] == "fail": return "implement" # verify 失敗 → implement に戻る return "merge" # verify 合格 → merge # ---------- Step 3: エッジを結ぶ ---------- graph = StateGraph(GraphState) graph.add_node("research", research) graph.add_node("implement", implement) graph.add_node("verify", verify) graph.add_node("merge", merge) graph.add_edge(START, "research") graph.add_edge("research", "implement") graph.add_edge("implement", "verify") graph.add_conditional_edges("verify", route_after_verify, {"implement": "implement", "merge": "merge"}) graph.add_edge("merge", END) # ---------- Step 5: checkpoint を付ける ---------- app = graph.compile(checkpointer=MemorySaver()) # ---------- Step 6: グラフを実行する ---------- if __name__ == "__main__": result = app.invoke( {"requirements": "fix the login page bug", "attempts": 0}, config={"configurable": {"thread_id": "session-1"}}, ) print(result)

この実装から読み取れるポイント：

• GraphStateのattempts: Annotated[int, operator.add]が、講義で述べた「並列書き込み時のマージ方法の宣言」そのものです（graph.mdに書き込むルールが、エンジン上では型アノテーションになる）。
• route_after_verifyが「ルーティングルールが返すのはノードの名前」という原則の実装です。ここにresearchを追加すれば、実験3の「部分合格フォールバック」は即座に実現できます。
• verifyは実装者のコンテキストを一切受け取らず、共有状態のcodeだけを見ます——グラフにおける「独立レビュー」の実装的な意味です。
• compile(checkpointer=MemorySaver())とthread_idにより、プロセスが落ちても checkpoint から再開でき、実行インスタンスを区別できます。

実行後、あなたのgraph.mdとこのコードを照合し、最初に対応しない場所を見つけてください——図の描き方が間違っているのか、コードの書き方が間違っているのか。それこそが「グラフは問題を紙の上に並べる」という言葉の意味です：以前は対応しなくても誰も気づきませんでしたが、今は一目でわかります。

どのタスクに図を描く価値があるか（判断基準）

第14回講義は「すべてのタスクが図を描く価値があるわけではない」と明言し、5つの判断基準を提示します。少なくとも3つ満たしてから手を付けるべきです：

1. タスクを独立した複数の作業単位に分割できる——分割した部分が互いに依存せず、並列できる
2. 分岐またはフォールバック経路が存在する——テスト失敗でどこに戻るか、情報不足でどこに戻るか、そうした経路を明示的に宣言する価値がある
3. 中間状態を保存する価値がある——checkpoint の後で停止でき、復元でき、最初からやり直す必要がない
4. 結果を明確に受け入れできる——各ノードが自動チェック可能な完了基準を持つ
5. 協力の利益 > 調整のコスト——並列で節約した時間が、グラフ自体と共有状態がもたらすオーバーヘッドより多い

「複雑」は「ステップが多い」と同じではありません。20ステップの線形パイプラインにはグラフは不要で、それはワークフローか単なるスクリプトです。5つのノードしかないが、互いにフォールバック・並列・承認がある構造こそグラフが必要です。判断基準は規模ではなく、分岐とフォールバックの存在です。

関連講義

• Lecture 14 — 単一ループからグラフエンジニアリングへ：本プロジェクトの概念的な土台。グラフの4つの部品、ループの3つの構造的失敗（Goodhart・上方向の失明・衝突）、Graph ≠ Workflow、ゼロから最初のグラフを構築する6ステップ
• Lecture 13 — 手動プロンプトから自律ループへ：あなたの loop はグラフの中の一つのノード。このプロジェクトはノードの内部構造を広げるもの
• Lecture 09 — エージェントが早すぎる完了宣言をする理由：検証ノードがなぜ実装ノードから独立していなければならないのか——グラフでは構造の問題
• Lecture 11 — 観測可能性がハーネスの一部である理由：グラフが複雑になるほど、各ノードが何をしているかを見る必要がある