目次
- 目次
- はじめに：LLMは「なぜ？」をどこまで理解しているのか
- DAGと「調整」の基本
- ステップ1：DAGベースの「独立性＆バックドアチェッカー」をPythonで実装する
- ステップ2：LangGraphで「因果チェックAIエージェント」を組む
- ステップ3：広告の例で実際に動かしてみる
- 最後に
こんにちは、Insight Edge でリードデータサイエンティストをしている五十嵐です。 本記事は、Insight Edge Advent Calendar 2025の 3日目を担当してお届けします。上手く次の人へバトンを渡せるように頑張りますので、よろしくお願いします！！
今回は、LLM・LangGraph・因果グラフ（DAG） を組み合わせて、「広告データに対して LLM に調整すべき変数を選ばせ、その妥当性をコードで検証する」 というテーマを扱います。
-
ビジネスサイドの方へ： 「LLM に因果的な問いを投げるとき、どこまで“理由付け”を信頼して任せられるのか？」という検証として。
エンジニア・データサイエンティストの方へ： 「DAG や d-separation を実装し、LangGraph で実際に AI エージェント化する具体的な手法」として。
それぞれの視点で楽しんでいただける内容になっていますので、ぜひ最後までお付き合いください！
はじめに：LLMは「なぜ？」をどこまで理解しているのか
ChatGPT や Gemini のような大規模言語モデル（LLM）は、
- 質問に答える
- 文章を要約する
- コードを書く
といったことがとても得意です。
一方で、データサイエンス寄りの人からすると、
「このモデル、本当に“因果関係”を理解しているの？」
という疑問もあると思います。
たとえば、よくある問いとして、以下を例に挙げます。
「広告費を増やすと売上は上がりますか？」
LLM はおそらく、
「広告費と売上には正の相関が見られることが多い一方で、 季節要因やキャンペーンなど他の要因も影響しているため、 広告費だけの効果を切り出すには注意が必要です」
のように、かなりそれらしい答えを返してくれます。
しかし、どれだけ説明が精緻になっても、現実のデータには
- 季節（Season）
- キャンペーン
- 景気
といった、広告費と売上の両方に効いている要因が潜んでいます。 問題は、こうした要因をどう扱うかを グラフとして明示し、そのうえで「どこまで信じてよい説明なのか」をチェックできるか という点にあります。
そこで本記事では、
- LLMに「この因果グラフ（DAG）なら、どの変数を調整すべきか？」と考えさせて
- その答えが、因果推論のルールに照らして正しそうかどうかを、こちらが用意したPythonコードでチェックする
という “因果推論テスト用の AI エージェント” を作ります。
ここで LangGraph は、
「LLM に考えさせるステップ」と「Pythonで因果ルールチェックをするステップ」をつないでくれるワークフローエンジンとして使っています。
つまり、
- LLM = 因果関係について説明したり、「この変数を調整すべき」と 提案する役
- Pythonコード = 因果グラフ（DAG）にもとづいて、「その提案は理論上ちゃんと筋が通っているか」を 判定する役
という役割分担を、LangGraph でひとつのエージェントとしてまとめている、というイメージです。
なお、本記事のコードは Google Cloud の Vertex AI 上のノートブック環境（Python） で実行しています。同様の構成であれば、ローカル環境や他のクラウドでも基本的には同じように動かすことができます。
DAGと「調整」の基本
ここで、簡単に前提となるイメージをそろえておきます。
広告の簡単な例を DAG（因果グラフ）で描くと、次のようになります。
- Season … 季節（年末セール期かどうか等）
- AdSpend … 広告費
- Sales … 売上
ここでの直感的なイメージは次のようになります。
- 季節（Season）が良いと、自然と売上は上がりやすい
- 同時に、良い季節には広告費も増やしがち
- さらに、広告費を増やすと売上も増えるはず
ここで、
「広告費（AdSpend）の効果だけを、できるだけ素直に見たい」
と思ったら、
- 季節（Season）による差をできるだけ公平にそろえる必要があります。
→ これを統計の世界では 「調整する」 と呼びます。
本記事で登場する用語の説明
以降の説明を読みやすくするために、 先に本記事で登場する専門用語の意味を簡単に確認しておきます。 （厳密な定義よりも全体像の把握を優先しています）
DAG（Directed Acyclic Graph）
- 変数を丸、因果関係を矢印で表現した「因果マップ」です。
A → Bは「AがBに影響する（可能性がある）」という関係を表します。
調整する（adjustment）
「ある要因の違いをそろえて、公平に比べる」ことを指します。
例：
年齢が高い人と若い人で薬の効果を比べたい → 年齢をそろえて比べる
季節による売上の差をならし、広告の効果だけを見る
バックドアパス（backdoor path）
- 因果の矢印とは別に、“裏道”のように紛れ込んでくる経路です。
- 「Season → AdSpend」と「Season → Sales」でできる
AdSpend ← Season → Salesのようなパスは、
Season を通じて「広告費と売上が一緒に動いているだけ」のパスと解釈できます。 - これが残ったままだと、
「広告費が効いているのか、季節が効いているのか分からない」 という問題が生じます。
調整集合 Z（adjustment set）
- バックドアの“裏道”をふさぐために、
「条件として入れておくべき変数の集合」 です。 - 例：
Z = {Season}なら、
「季節が同じ状況で広告費の違いだけを見る」というイメージになります。
d-separation
- 「グラフ上で X と Y の間に、まだ情報が流れる道が残っているかどうか」 を
機械的にチェックするためのルールです。 - 全てのパスが“閉じている” → d-separated → その条件下では独立
- 1本でも“開いた”パスがある → d-connected → まだ依存が残っている
コライダー / 非コライダー
- パス上の真ん中の点の「矢印の入り方」による区別です。
A → C ← Bのように、両側から矢印が集まってくる C を
「コライダー（ぶつかり地点）」と呼びます。A ← C ← BやA ← C → Bのように、矢印が“通り抜ける”形は非コライダー です。
本記事では、d-separation に基づく判定ロジックを Python で実装し、 LLM が提案した調整集合が 「裏道を適切に遮断しているか」 を自動でチェックできるようにします。
実際に本番分析で使うときは、ここで紹介したロジックを DoWhy/EconML などのフレームワークと組み合わせるのがおすすめです。
ステップ1：DAGベースの「独立性＆バックドアチェッカー」をPythonで実装する
まずは、因果グラフを扱うための土台として、次の2つのクラスを実装します。
DAG の構造（親・子の関係）を保持する CausalDAG
d-separation とバックドア条件をチェックする DSeparationChecker
この2つは、あくまで「DAG 上でパスをたどって、因果推論のルールに沿ってチェックする」ためのユーティリティです。実データを学習したり推定する部分は含んでいません。
※「細かいロジックまでは追わないけど、全体の構成だけ知りたい」という方は、
以降のコードをざっと眺めてこういう裏道検査用のクラスがあるんだな、くらいに捉えて頂ければ十分です。
1-1. 因果グラフを扱うクラス：CausalDAG
最初に、DAG の構造を表現するクラスを定義します。 親ノードと子ノードの対応関係、ノードの一覧、祖先ノードの集合などを扱えるようにします。
from collections import defaultdict, deque from typing import Dict, List, Set, Iterable class CausalDAG: def __init__(self, edges: Iterable[tuple[str, str]]): """ edges: (parent, child) のペアのリストで DAG を定義する。 例: edges = [ ("Season", "AdSpend"), ("Season", "Sales"), ("AdSpend", "Sales"), ] """ self.parents: Dict[str, List[str]] = defaultdict(list) self.children: Dict[str, List[str]] = defaultdict(list) self.nodes: Set[str] = set() for u, v in edges: self.parents[v].append(u) self.children[u].append(v) self.nodes.add(u) self.nodes.add(v) def all_nodes(self) -> Set[str]: return set(self.nodes) def ancestors_of(self, zs: Iterable[str]) -> Set[str]: """ Z のすべての祖先ノード Anc(Z) を返す。 d-separation では、 「コライダーが Z または Z の祖先を持つとき、パスが開く」 というルールで必要になる。 """ zs = set(zs) visited: Set[str] = set() queue: deque[str] = deque(zs) while queue: z = queue.popleft() for p in self.parents[z]: if p not in visited: visited.add(p) queue.append(p) return visited
この CausalDAG クラスでは、
コンストラクタで (親, 子) のエッジ一覧から
各ノードの親リスト parents
各ノードの子リスト childrenを構築しています。
all_nodes() でノードの集合を取得し、
ancestors_of(zs) で、あるノード集合 Z の「祖先ノード集合」を求めます。
後で説明する d-separation の判定では、 「コライダーの祖先に条件づけされたノードが含まれているか」 を判断する必要があるため、この祖先集合を使います。
1-2. d-separation とバックドアパスを判定する：DSeparationChecker
次に、DAG の上で d-separation とバックドアパスの有無をチェックするクラスです。 ここでは、DAG を「無向グラフ」として見たときの全ての単純パスを列挙し、 各パスが d-separation のルールに照らして「開いているか／閉じているか」を判定します。
class DSeparationChecker: """ DAG に対して d-separation / バックドア条件を判定するクラス。 """ def __init__(self, dag: CausalDAG): self.dag = dag # ---------- d-separation 関連 ---------- def _is_collider_on_path(self, prev_node: str, mid_node: str, next_node: str) -> bool: """ パス上の3点 prev -> mid -> next において、mid がコライダーかどうかを判定。 定義: mid に2本の矢印が“向かっている”とき、mid はコライダー。 つまり (prev -> mid) かつ (next -> mid) のとき。 """ return (prev_node in self.dag.parents[mid_node]) and \ (next_node in self.dag.parents[mid_node]) def _compute_ancestors_of_Z(self, Z: Set[str]) -> Set[str]: """ コライダーが Z または Z の祖先に含まれるとき、 そのコライダーを通るパスは「開きうる」。 そのため Anc(Z) を前もって計算しておく。 """ return self.dag.ancestors_of(Z) def _find_all_simple_paths(self, start: str, goal: str, max_len: int = 10) -> List[List[str]]: """ 無向グラフとして見たときの単純パスをすべて列挙する。 DAG は小さい前提なので、深さ制限 max_len を軽くかけている。 """ neighbors: Dict[str, List[str]] = {} for n in self.dag.all_nodes(): neighbors[n] = list(set(self.dag.parents[n]) | set(self.dag.children[n])) paths: List[List[str]] = [] stack: List[tuple[str, List[str]]] = [(start, [start])] while stack: node, path = stack.pop() if node == goal: paths.append(path) continue if len(path) >= max_len: continue for nxt in neighbors[node]: if nxt in path: continue # simple path only stack.append((nxt, path + [nxt])) return paths def _path_is_active(self, path: List[str], Z: Set[str], ancestors_Z: Set[str]) -> bool: """ 与えられたパスが、条件集合 Z のもとでアクティブかどうかを判定。 ルール（縮約版）: - 非コライダー中間ノード j: j ∈ Z ならパスはブロック - コライダー中間ノード j: j ∈ Z または j ∈ Anc(Z) ならパスが開きうる それ以外ならブロック """ if len(path) <= 2: # 直接つながっている場合は、中間ノードがないので常に候補 return True for i in range(1, len(path) - 1): prev_node = path[i - 1] mid_node = path[i] next_node = path[i + 1] is_collider = self._is_collider_on_path(prev_node, mid_node, next_node) if not is_collider: # 非コライダーの場合、そのノードに条件づけるとパスはブロック if mid_node in Z: return False else: # コライダーの場合、 # そのノード自身 or その祖先が Z に含まれる場合にパスが開きうる。 if (mid_node not in Z) and (mid_node not in ancestors_Z): return False return True def d_separated(self, X: Iterable[str], Y: Iterable[str], Z: Iterable[str]) -> bool: """ X と Y が条件集合 Z のもとで d-separated かどうかを判定する。 戻り値: True -> X ⫫ Y | Z （独立） False -> X ̸⫫ Y | Z（依存） """ X = set(X) Y = set(Y) Z = set(Z) ancestors_Z = self._compute_ancestors_of_Z(Z) for x in X: for y in Y: paths = self._find_all_simple_paths( x, y, max_len=len(self.dag.all_nodes()) + 1 ) for p in paths: if self._path_is_active(p, Z, ancestors_Z): # 1本でもアクティブパスがあれば d-connected（依存） return False # アクティブパスが見つからなければ d-separated（独立） return True # ---------- バックドアパス関連 ---------- def has_active_backdoor_path( self, treatment: str, outcome: str, Z: Iterable[str], ) -> bool: """ treatment -> outcome の因果効果を推定したいときに、 「バックドアパス」が Z の下でアクティブかどうかを判定する。 バックドアパスとは: - treatment から outcome へのパスのうち、 - 最初のエッジが「親 -> treatment」になっているもの。 (例: Season -> AdSpend のように、最初が '入ってくる' パス) """ Z = set(Z) ancestors_Z = self._compute_ancestors_of_Z(Z) # treatment から outcome へのすべての単純パス paths = self._find_all_simple_paths( treatment, outcome, max_len=len(self.dag.all_nodes()) + 1, ) for p in paths: if len(p) < 2: continue first_neighbor = p[1] # 最初のエッジが「neighbor -> treatment」かをチェック # parent -> child の定義から、 # "neighbor -> treatment" なら neighbor は treatment の親であるはず if treatment not in self.dag.children[first_neighbor]: # neighbor -> treatment ではないのでバックドア候補ではない continue # このパスが Z のもとでアクティブかどうかを判定 if self._path_is_active(p, Z, ancestors_Z): return True # アクティブなバックドアパスが存在する return False # どのバックドアパスもアクティブではない def is_valid_backdoor_adjustment_set( self, treatment: str, outcome: str, Z: Iterable[str], ) -> bool: """ Z が treatment -> outcome の因果効果を推定するための 「妥当なバックドア調整集合」かどうかを判定する。 定義: - treatment と outcome の間に、Z のもとでアクティブなバックドアパスが存在しないとき True。 """ return not self.has_active_backdoor_path(treatment, outcome, Z)
このクラスでは、
DAG 上のすべてのパスを洗い出し、
各パスが d-separation のルールに従って「開いているか／閉じているか」を判定し、
その結果として
「X と Y が条件付きで独立になっているか（d_separated）」
「バックドアパスがすべて閉じていて、調整集合として妥当か（is_valid_backdoor_adjustment_set）」
を返す仕組みをまとめています。
ここまでで、DAG 上のパスに対して因果推論の基本ルールを機械的に適用し、 LLM の提案をチェックするための土台が整いました。
ステップ2：LangGraphで「因果チェックAIエージェント」を組む
次に、この d-separation チェッカーを LLM と組み合わせた AI エージェントとして動かすために、LangGraph を使ってワークフローを組み立てます。
このエージェントは、次の2ステップで動きます。
LLM に「調整すべき変数集合 Z」を提案させる
その提案 Z が、DAG に基づいてバックドアを閉じる集合になっているかどうかをチェックする
2-1. Stateの設計
LangGraph は「状態（State）を持つワークフローエンジン」というイメージです。 各ノードは State を受け取り、更新した State を次のノードへ渡します。
今回のエージェントでは、次のような State を定義します。
from typing import TypedDict, List, Optional class CausalAgentState(TypedDict, total=False): # 入力 question: str # ユーザーの因果的な問い（説明用） treatment: str # 介入変数 X target: str # 効果を知りたい変数 Y # LLM の出力 candidate_adjustment: List[str] # LLM が提案した調整集合 Z llm_raw_answer: str # LLM の生の回答 # 検査結果 d_separated: Optional[bool] # X と Y が Z で d-separated かどうか（参考値） backdoor_ok: Optional[bool] # Z が妥当なバックドア調整集合かどうか # ログ debug_log: List[str]
ここでは、
treatment/targetに「広告費」や「売上」などの変数名を入れ、candidate_adjustmentに LLM が提案する調整集合 Z を格納し、backdoor_okで「その Z がバックドア調整として妥当か」を記録します。
debug_log には、各ステップの内部状態や LLM の生出力の一部を文字列として残しておきます。
2-2. LLMに「調整すべき変数セット」を提案させる
次に、LLM に対して「どの変数で調整すべきか」を尋ねる部分です。 LangChain の ChatPromptTemplate を使い、「JSON 配列だけを返す」 ように強く指示します。
from langchain_core.prompts import ChatPromptTemplate import json ADJUST_PROMPT = ChatPromptTemplate.from_template( """ You are a careful causal inference assistant. We have a causal DAG over variables and we want to estimate the causal effect of {treatment} on {target}. Your task: 1. Propose a set of variables Z to adjust for (back-door adjustment set). 2. Return ONLY a JSON list of variable names, like: ["VarA", "VarB"] IMPORTANT: - Output MUST be a single JSON array. - Do NOT add any explanation. - Do NOT use Markdown code fences. - Do NOT wrap the JSON in ```json or ```. Variables available: {all_vars} Causal DAG description: {dag_text} """ ) def _extract_json_array_from_text(text: str) -> str: """ LLM が返したテキストから JSON 配列部分だけを抜き出すユーティリティ。 - ```json ... ``` のようなコードブロックを剥がす - テキスト中の最初の '[' から最後の ']' までを切り出す """ t = text.strip() # 1. コードブロック ```...``` を剥がす if t.startswith("```"): lines = t.splitlines() # 先頭の ```xxx を削る if lines and lines[0].startswith("```"): lines = lines[1:] # 末尾の ``` を削る if lines and lines[-1].startswith("```"): lines = lines[:-1] t = "\n".join(lines).strip() # 2. 最初の '[' と最後の ']' を探す start = t.find("[") end = t.rfind("]") if start != -1 and end != -1 and start < end: return t[start : end + 1] # 見つからなければそのまま返す（この後の json.loads で落ちてフォールバック） return t def propose_adjustment_node( state: CausalAgentState, dag: CausalDAG, dag_text: str, llm, ) -> CausalAgentState: treatment = state["treatment"] target = state["target"] all_vars = sorted(list(dag.all_nodes())) prompt = ADJUST_PROMPT.format( treatment=treatment, target=target, all_vars=", ".join(all_vars), dag_text=dag_text, ) resp = llm.invoke(prompt) raw_content = resp.content if hasattr(resp, "content") else str(resp) # JSON 配列部分だけにクリーニング cleaned = _extract_json_array_from_text(raw_content) candidate_Z: List[str] = [] try: parsed = json.loads(cleaned) if isinstance(parsed, list): # 文字列だけに揃えておく candidate_Z = [str(x) for x in parsed] except Exception: candidate_Z = [] debug_log = list(state.get("debug_log", [])) debug_log.append(f"[propose_adjustment_node] raw LLM: {raw_content[:120]}...") debug_log.append(f"[propose_adjustment_node] cleaned: {cleaned}") debug_log.append(f"[propose_adjustment_node] parsed Z: {candidate_Z}") new_state: CausalAgentState = { **state, "candidate_adjustment": candidate_Z, "llm_raw_answer": raw_content, "debug_log": debug_log, } return new_state
このノードは、DAG の情報（変数名や構造の説明）をプロンプトに埋め込んで LLM に渡し、「調整すべき変数の候補 Z を JSON 配列で返してもらう」 役割を持ちます。返ってきたテキストから JSON 配列の部分だけを抜き出してパースし、その結果を candidate_adjustment（LLM が提案した Z）として State に保存します。あわせて、元の出力や抽出結果は debug_log に記録しておきます。これにより、LLM の出力形式が多少ぶれても、「変数名の配列」だけを取り出して使えるようにしています。
2-3. DAG側でその提案をチェックする
このノードは、LLM が提案した調整集合 Z について、
バックドアパスがすべて閉じているかどうか（
backdoor_ok）参考として、X と Y が Z のもとで d-separated になっているかどうか（
d_separated）
を DSeparationChecker で判定し、その結果を State に書き込むだけのシンプルなチェック役です。
def check_adjustment_node( state: CausalAgentState, checker: DSeparationChecker, ) -> CausalAgentState: treatment = state["treatment"] target = state["target"] Z = state.get("candidate_adjustment", []) # 1. バックドア調整として妥当か？ backdoor_ok = checker.is_valid_backdoor_adjustment_set( treatment=treatment, outcome=target, Z=Z, ) # 2. オプション: d-separation もログとして残しておく（X と Y が完全独立かどうか） d_sep = checker.d_separated([treatment], [target], Z) debug_log = list(state.get("debug_log", [])) debug_log.append( f"[check_adjustment_node] X={treatment}, Y={target}, Z={Z}, " f"backdoor_ok={backdoor_ok}, d_separated={d_sep}" ) new_state: CausalAgentState = { **state, "d_separated": d_sep, # これは参考値 "backdoor_ok": backdoor_ok, # 実際に見たいのはこちら "debug_log": debug_log, } return new_state
ここで行っていることはシンプルです。
checker.is_valid_backdoor_adjustment_set(...)で、 LLM が提案した Z が 「バックドアパスをすべて閉じているか」 を判定します。 → これがbackdoor_okです。checker.d_separated(...)は、X と Y が Z のもとで完全に独立になるかどうかを判定します。 実務上は常に独立である必要はなく、ここではあくまで参考値としてログに残しています。
2-4. LangGraphでノードをつなぐ
最後に、LangGraph の StateGraph を使って、
propose_adjustment_node（LLMに調整集合を提案させる）check_adjustment_node（DAGでその提案を検査する）
という2つのノードを一つのワークフローとしてつなぎます。
from langgraph.graph import StateGraph, END def build_causal_langgraph( dag: CausalDAG, dag_text: str, llm, ): graph = StateGraph(CausalAgentState) # d-separation / バックドアチェッカー checker = DSeparationChecker(dag) # 部分適用で dag / dag_text / llm を閉じ込めたノード関数を定義 def _propose_node(s: CausalAgentState) -> CausalAgentState: return propose_adjustment_node( s, dag=dag, dag_text=dag_text, llm=llm, ) def _check_node(s: CausalAgentState) -> CausalAgentState: return check_adjustment_node(s, checker=checker) # ノードを登録 graph.add_node("propose_adjustment", _propose_node) graph.add_node("check_adjustment", _check_node) # フローを定義 graph.set_entry_point("propose_adjustment") graph.add_edge("propose_adjustment", "check_adjustment") graph.add_edge("check_adjustment", END) # 実行可能なアプリケーションを返す app = graph.compile() return app
この build_causal_langgraph 関数は、「LLM に調整変数を考えさせて、DAG 側でチェックする」ための因果チェック用エージェントを組み立てる関数です。
エージェントに question（説明用の問い）、treatment（介入したい変数）、target（効果を知りたい変数）を渡すと、
まず LLM が「調整すべき変数の候補 Z」を提案し、
そのあと
DSeparationCheckerが「バックドアが閉じているかどうか」を判定し、
その結果として、LLM の回答内容や提案された Z、判定結果 backdoor_ok などが final_state にまとまって返ってきます。
ステップ3：広告の例で実際に動かしてみる
ここからは、実際に広告の DAG を使ってエージェントを動かしてみます。 LLM には Vertex AI の Gemini を利用します。
3-0. LLM（Gemini）のセットアップ
まず、Vertex AI 上で Gemini を呼び出すための設定を行います。 本記事のコードは Vertex AI のノートブック環境（Python）で実行していますが、 適切な認証とプロジェクト設定を行えば、ローカル環境などからでも同様のコードで呼び出すことができます。
from langchain_google_vertexai import ChatVertexAI llm = ChatVertexAI( model="gemini-2.5-flash", project="your-gcp-project-id", # あなたの GCP プロジェクト ID location="us-central1", temperature=0, )
ここでは、モデル名やリージョン、プロジェクトIDなどを指定しています。
temperature=0 としているのは、因果推論のように「論理的な一貫性」を重視したいケースでは、ランダム性を抑えた方が望ましいためです。
3-1. 広告費と売上（AdSpend→Sales）の例
先ほど説明した広告の DAG を、そのままコードに落とし込みます。
# 1. DAG とその説明テキスト edges = [ ("Season", "AdSpend"), ("Season", "Sales"), ("AdSpend", "Sales"), ] dag = CausalDAG(edges) dag_text = """ Variables: - Season: categorical (e.g., 'Holiday', 'Normal', ...) - AdSpend: continuous, amount of advertising spend - Sales: continuous, sales amount Causal structure (DAG): - Season -> AdSpend - Season -> Sales - AdSpend -> Sales Goal: We want to estimate the causal effect of AdSpend on Sales. """ # 2. LangGraph アプリケーションを構築 causal_app = build_causal_langgraph(dag, dag_text, llm=llm) # 3. 初期状態を定義して実行 initial_state: CausalAgentState = { "question": "広告費(AdSpend)の売上(Sales)への因果効果を推定したい。", "treatment": "AdSpend", "target": "Sales", "debug_log": [], } final_state = causal_app.invoke(initial_state) print("=== [AdSpend→Sales] LLM の生回答 ===") print(final_state.get("llm_raw_answer", "")) print("\n=== LLM が提案した調整集合 Z ===") print(final_state.get("candidate_adjustment")) print("\n=== バックドア調整として妥当か？ ===") print(f"backdoor_ok -> {final_state.get('backdoor_ok')}") print("\n=== d-separation 判定結果（参考値） ===") print(f"(AdSpend ⫫ Sales | Z) ? -> {final_state.get('d_separated')}") print("\n=== Debug log ===") for log in final_state.get("debug_log", []): print(log)
このコードでは、まず edges で広告の因果構造（DAG）を定義し、それを CausalDAG に渡しています。dag_text には DAG の意味を英語でまとめておき、LLM に渡すプロンプトの一部として使います。build_causal_langgraph(...) で因果チェック用のエージェントを作成し、initial_state に質問文・介入変数 AdSpend・目的変数 Sales をセットして causal_app.invoke(initial_state) を呼び出すと、一連のフローが実行されます。
実行結果として、LLM の生回答や提案された調整集合 Z、その Z がバックドア調整として妥当かどうか（backdoor_ok）、d-separation の判定結果などが得られます。この例では、LLM が Season を含むような調整集合を提案し、backdoor_ok -> True となることを期待しています。
以下が、上記コードの実行結果です。
この出力は、次のことを示しています。
LLM は、広告効果を評価するために Season を調整すべき変数として正しく提案している
Python 側の d-separation チェッカーも、「Season を調整すればバックドアパス（AdSpend ← Season → Sales）は閉じる」と判断し、backdoor_ok -> True になっている
一方で、AdSpend → Sales という因果パスは残っているため、Season で調整しても AdSpend と Sales は独立にはならない（d_separated=False）
つまりこの仕組みは、 「LLM が DAG を踏まえて妥当な調整集合を提案できているか？」 を、コード側で機械的にチェックできている ことを、シンプルな例で確認できた、という結果になっています。
今回の実装はあくまで、
因果構造（DAG）は人間または別プロセスが与える
LLM は「どの変数で調整するか」を提案する
Python（d-separation チェッカー）が、その提案が因果論的に妥当かどうかを検証する
という、ごく小さなパイプラインです。それでも、
LLM に自由にしゃべらせるのではなく、 「DAG に沿った因果的な一貫性」 をチェックする枠組みを足す
LangGraph で「LLM に考えさせるステップ」と「ルールベースで検証するステップ」をきれいに分離する
という設計の手応えは十分に感じられると思います。
最後に
本記事では、LangGraphを用いた実装コードを交えつつ、AIエージェントと因果グラフを組み合わせて「調整すべき変数」を選ばせるアプローチを紹介しました。
あらためて補足しますと、今回扱った範囲はあくまで 調整集合のチェック までです。 実データから因果効果を推定したり、反実仮想を評価したりする段階では、厳密な統計的推定や感度分析が不可欠です。実務での分析においては、今回紹介したロジックを DoWhy や EconML などの既存フレームワークと組み合わせて活用することをおすすめします。
今後の発展としては、
もう少し複雑な DAG（多段の交絡、コライダー、介在変数など）で LLM をテストする
調整集合の候補を複数出させ、どれがミニマルかをチェックする
実データと接続し、DoWhy/EconML 側で推定した結果を LLM に要約させる
といった方向性が考えられます。
LLM に「なぜ？」を語らせつつ、その裏側で 因果グラフと Python のロジックで足場を固める── そうした組み合わせ方の一例として、本記事が何かのヒントになれば幸いです。
最後まで読んでいただき、ありがとうございました! それでは、引き続きよい Advent Calendar ライフ（？）をお過ごしください!!
