第08回 学習メモ: LangGraph によるユーザーインタビューグラフ¶
全体像¶
第04回・第05回で手書きしていた while True: ループによるエージェントを、LangGraph の宣言的な状態遷移グラフで書き直す回。LangGraph のデビュー作。
題材は「ユーザーインタビュー」のロールプレイ:
- LLM が「質問者役(コンサル)」と「回答者役(40代SEのペルソナ)」を交互に演じて対話を進める
- メッセージが一定数たまったら、別の「レポーター役」が総括レポートを作る
グラフ構造(手書き Mermaid)¶
flowchart TD
Start([START]) --> Q[question<br/>コンサルが質問を生成]
Q --> I[interview<br/>ペルソナが回答]
I --> Check{should_continue<br/>messages < 9?}
Check -->|continue| Q
Check -->|end| R[report<br/>コンサルが総括]
R --> End([END])
ノードは3つ(question / interview / report)、エッジは「無条件遷移」と「条件付き遷移」の混在。while True: を書かずに「メッセージが 9 件未満ならループ続行」をグラフのエッジとして宣言 できているのがポイント。
グラフ構造(LangGraph 自動生成)¶
LangGraph 0.2+ には グラフを自動で Mermaid / PNG にエクスポートする機能 が標準で入っている。compile() 後のエージェントから取得できる:
graph = UserInterviewGraph()
print(graph.agent.get_graph().draw_mermaid()) # Mermaid 文字列
png = graph.agent.get_graph().draw_mermaid_png() # PNG バイト列
このスクリプトに対する自動生成結果は以下:

Mermaid 版(条件付きエッジが点線 -.-> で表示される):
---
config:
flowchart:
curve: linear
---
graph TD;
__start__([__start__]):::first
question(question)
interview(interview)
report(report)
__end__([__end__]):::last
__start__ --> question;
interview -. continue .-> question;
interview -. end .-> report;
question --> interview;
report --> __end__;
classDef default fill:#f2f0ff,line-height:1.2
classDef first fill-opacity:0
classDef last fill:#bfb6fc
手書き Mermaid との違い:
| 観点 | 手書き Mermaid | LangGraph 自動生成 |
|---|---|---|
| ノード説明 | 「コンサルが質問を生成」など意味的な注釈付き | ノード名(question / interview / report)のみ |
should_continue |
ひし形ノードで明示 | エッジのラベルとして continue / end を点線エッジに埋め込み |
__start__ / __end__ |
START / END 表記 |
内部表現そのまま |
| 用途 | 概念の理解 | 実装の検証(コードと図のズレを検出) |
LangGraph 自動生成図は「コードが意図通りグラフに変換されたかを確認するツール」、手書き図は「読み手に意味を伝えるツール」と使い分けるのが良い。
可視化を使うタイミング¶
- デバッグ: 「ノードを追加したのにエッジを繋ぎ忘れた」「条件付きエッジの分岐先を typo した」などをグラフ図で即発見
- PR レビュー: コードと一緒に自動生成図を貼れば、レビュワーが構造を一目で確認できる
- 設計レビュー: 実装前にノード名と接続を Markdown に書き、合意してからコードに落とす逆順アプローチも可能
04/05 との対比(書き方の進化)¶
flowchart LR
subgraph T04["04/05: 手書きループ"]
L1["while True:"] --> L2["process_step()"]
L2 --> L3{is_final?}
L3 -->|No| L1
L3 -->|Yes| L4[break]
end
subgraph T08["08: LangGraph"]
G1[StateGraph 定義] --> G2[add_node × 3]
G2 --> G3[add_edge / conditional_edges]
G3 --> G4["compile() → 実行は stream()"]
end
T04 -->|宣言化| T08
「ループ条件をどこに書くか」が大きく変わる:
- 04/05: Python の while/if で制御。読むときコードを上から下に追う必要がある
- 08: グラフのエッジに条件を埋め込む。構造が一目で分かる(Mermaid で図示できる構造化)
使用ライブラリ・原理¶
StateGraph と AgentState¶
class AgentState(TypedDict):
mission: str
persona: str
messages: Annotated[Sequence[BaseMessage], operator.add]
LangGraph の 「状態」 は TypedDict で定義する。グラフを流れる「便箋」のようなもので、各ノードが「便箋の一部を書き換える」というメンタルモデル。
Annotated[Sequence[BaseMessage], operator.add] が肝心:
| キー | 動作 |
|---|---|
mission: str |
ノードが返した値で 完全に上書き(普通の Python dict と同じ) |
persona: str |
同上 |
messages: Annotated[..., operator.add] |
ノードが返した値で old + new(リスト結合)して保存 |
つまり messages は 追記専用 のフィールドで、ノードは「自分が増やしたい分」だけ返せば LangGraph が state["messages"] += new 相当を自動でやってくれる。
Reducer(畳み込み関数)として operator.add を使う発想で、関数型寄りのデザイン。
ノード関数の規約¶
ノードは「state を受け取り、state の部分更新を dict で返す」関数。
def generate_question(self, state):
mission = state["mission"]
# ... LLM 呼び出し
return {"messages": [response]} # ← 部分更新だけ返す
- 全フィールドを返す必要はない。変更したいフィールドだけ 返す
- 戻り値は 次の状態への差分(reducer に渡される)
これは Redux の reducer や Elm のメッセージ駆動アーキテクチャに近い設計思想。
グラフ組み立ての5ステップ¶
workflow = StateGraph(AgentState) # ① 状態の型でグラフを初期化
workflow.add_node("question", self.generate_question) # ② ノード追加
workflow.add_node("interview", self.generate_interview)
workflow.add_node("report", self.generate_report)
workflow.set_entry_point("question") # ③ 入口
workflow.add_edge("question", "interview") # ④ 無条件遷移
workflow.add_edge("report", END) # ⑤ 終端
workflow.add_conditional_edges( # ⑥ 条件付き遷移
"interview", self.should_continue,
{"continue": "question", "end": "report"},
)
self._agent = workflow.compile() # ⑦ コンパイル(最適化+実行可能化)
compile() は「LCEL のチェイン化」に似ていて、ここで初めて実行可能な Runnable になる。
should_continue の役割¶
def should_continue(self, state):
if len(state["messages"]) < 9:
return "continue"
else:
return "end"
- 戻り値は文字列キー(任意のラベルで OK)
add_conditional_edges(..., {"continue": "question", "end": "report"})で キー → 次ノード のマッピングを定義- これで
while True:を データドリブンな分岐 に置き換えられる
stream() でノード実行をストリーミング¶
for s in graph.agent.stream({"mission": "...", "persona": "...", "messages": []}):
if "question" in s:
print(s["question"]["messages"][0].content)
if "interview" in s:
...
stream()は ノードが1つ実行されるたびに その結果を yield する- 戻り値は
{"<実行ノード名>": <ノードが返した dict>}の形 - これで Chainlit のような UI に「考えている過程」をリアルタイム表示できる
ファイル別の役割¶
| ファイル | 役割 |
|---|---|
user_interview_graph.py |
UserInterviewGraph クラス本体。StateGraph 定義 + 3 ノード + 条件付きエッジ + if __name__ == "__main__" のサンプル実行 |
requirements.txt |
連載原典の依存固定(langchain==0.1.12, langgraph==0.0.28)。現代版(langgraph>=0.2)とは API が異なるので動かす際は注意 |
README.md |
環境構築と実行方法のメモのみ(極めて簡潔) |
学んだこと(要点)¶
while True:から宣言的グラフへ: 04/05 はループ条件を Python の制御フローで書いていたが、LangGraph ではadd_conditional_edgesでエッジに条件を埋め込む。コードの可読性と図示可能性が劇的に上がるAnnotated[..., operator.add]は reducer: TypedDict のフィールド型に「合流時の畳み込み方」を埋め込める Python の型システム小技。関数型寄りの発想- ノードは状態の部分更新を返すだけ: 副作用なし、全体状態を意識しなくてよい設計。Redux の reducer と同じ
- LLM は同じモデルだが、プロンプトで役割を切り替えている:
generate_questionではコンサル、generate_interviewではペルソナ。1つの LLM が複数役を演じる ロールプレイの実装手法 stream()で UI 表示が自然になる: 04/05 のasync for step in agent.run(...)と同じ哲学だが、LangGraph では標準機能として提供されるshould_continueの戻り値は任意のキー:"continue"/"end"でなくても"loop"/"finish"でもよい。マッピング辞書で次ノードに紐づくだけ
拡張アイデア¶
- メッセージ件数ではなく LLM 自己判定でループ終了:
should_continueを別の LLM 呼び出しにして「もう十分情報集まった?」を判定させる - 複数ペルソナのインタビュー: ノードを増やしてペルソナA・ペルソナBの両方を順番にインタビュー → レポート統合
- Human-in-the-Loop ノード:
interpretノードの代わりに人間が回答するノードを差し込む(LangGraph 0.2+ のinterruptで可能) - 質問の重複検出ノード:
questionノードの後に「過去と同じ質問じゃないか」をチェックするノードを挟む - 可視化のさらなる活用: 既に本メモに自動生成図を貼ったが、ノード追加時に PNG を再生成する仕組み(例:
make graphで再生成)をプロジェクトに入れると、設計とコードのズレを継続的に検出できる
連載原典からの移植で行った変更(langgraph 0.0.28 → 0.2.x)¶
連載原典の langgraph==0.0.28 / langchain==0.1.12 から、現行の langgraph>=0.2,<1.0 / langchain>=0.3,<1.0 に書き換えた。最大の変更点は Annotated[..., operator.add] → Annotated[..., add_messages]。
| 箇所 | 原典 | 本フォルダ |
|---|---|---|
messages の畳み込み |
Annotated[Sequence[BaseMessage], operator.add] |
Annotated[list[BaseMessage], add_messages](from langgraph.graph.message import add_messages) |
| エントリポイント | workflow.set_entry_point("question") |
workflow.add_edge(START, "question")(START を langgraph.graph から import) |
| LLM | gpt-4-0125-preview |
gpt-4o-mini(温度 0.7 でロールプレイの幅を確保) |
| プロンプト import | from langchain.prompts import ChatPromptTemplate |
from langchain_core.prompts import ChatPromptTemplate |
| BaseMessage import | from langchain_core.messages import BaseMessage |
同じ(変更なし) |
| API キー取得 | 環境変数 / ハードコード前提 | 1Password CLI 経由(op read op://Personal/openAI_API/credential) |
| 型ヒント | 戻り値型なし | def generate_question(self, state: AgentState) -> dict のように明示 |
| 依存可視化ライブラリ | grandalf を別途インストール |
不要(graph.get_graph().draw_mermaid() が標準装備) |
operator.add → add_messages の違い(重要)¶
両方とも「リストの結合」をする reducer だが、add_messages のほうが メッセージ専用に最適化されている:
| 観点 | operator.add |
add_messages |
|---|---|---|
| 動作 | 単純な old + new(list concat) |
old + new + ID 重複の de-duplication + dict → Message 自動変換 |
tool_calls のマージ |
不可(同じ tool_call が重複しうる) | 同じ id のメッセージは新しい方で上書き |
| 辞書形式の入力 | そのままlist に積まれる(型不整合の元) | 自動で HumanMessage 等にキャストされる |
add_messages を使うと、たとえばノードが {"messages": [{"role": "user", "content": "..."}]}(dict 形式)を返しても、reducer が自動で HumanMessage に変換してくれる。これは Tool 呼び出しや Human-in-the-Loop でメッセージ ID を再利用する場面で特に効く。
langgraph 0.0.28 → 0.2+ のその他変更¶
pregel(LangGraph の実行エンジン)概念が公開ドキュメントに登場。stream_mode="values"/"updates"/"messages"の使い分けが学べるasyncノードが標準サポート。Chainlit 連携時はasync def化するのが自然(このスクリプトは CLI 実行なので同期のまま)- グラフ可視化が公式で
get_graph().draw_mermaid()/draw_png()に統合(grandalf不要)
起動方法¶
詳細は README.md を参照。サマリ:
cd 08
uv run --no-project \
--with "openai>=1.0" \
--with "langchain>=0.3,<1.0" \
--with "langchain-openai>=0.2,<1.0" \
--with "langgraph>=0.2,<1.0" \
python user_interview_graph.py
実行すると、コンサル → ペルソナ の対話が4ターン進み、最後にレポートが出力される。
記事参照¶
- Software Design 2024年〜の連載 第08回「LangGraphの応用」
- 連載タイトル上は 07 が「LangGraph によるユーザーインタビューグラフ」、08 が「LangGraph の応用」だが、コード実体は 07 が LCEL の予習回、08 が LangGraph デビュー作 という非対称がある(README ベースの誌面構造と GitHub 構造のズレ)
作成: 2026-05-19 / 最終更新: 2026-06-10