第6章 実践的な開発パターンとユースケース

6.1 カスタマーサポート用のチャットボット開発

実用的なAIアプリケーションの代表例が、問い合わせに対応するカスタマーサポートボットです。単に質問に答えるだけでなく、裏側で動作する顧客システムと安全に連携しながら、ユーザー個別の課題を解決するシステムをLangGraphで設計します。

このチャットボットは、まず顧客の「契約状態」や「過去の履歴」を状態（State）に読み込むことから始めます。ユーザーが「プランを解約したい」と言った場合、AIは社内の解約手続きに関するルール（RAGで検索）を参照し、必要な情報をユーザーから聞き取ります。途中でクレジットカード情報の登録変更などが発生する場合、前述の「Human-in-the-loop」を利用して、ユーザー自身が変更に同意したことを確認してから処理を実行するエッジを作ります。

以下は、サポートボットがユーザー情報の確認と解約などの手続きを安全に進めるための状態遷移コード例です。

from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

# カスタマーサポート用の状態定義
class SupportState(TypedDict):
    customer_id: str
    contract_status: str  # "active", "pending_cancel", "canceled" などの状況
    messages: Annotated[Sequence[BaseMessage], add_messages]
    billing_confirmed: bool  # ユーザー同意が得られたか

# 初期情報の取得ノード
def initialize_session(state: SupportState):
    # 顧客IDに基づいてデータベースから現在の契約状況を取得
    customer_info = get_customer_db(state["customer_id"])
    return {"contract_status": customer_info["status"]}

# 解約確認ノード
def request_cancel_confirmation(state: SupportState):
    # ユーザーに対し「本当に解約してもよろしいですか？」というメッセージを生成して返す
    message = ("user", "解約手続きを進めます。契約内容を確認し、同意する場合は承認してください。")
    return {"messages": [message], "billing_confirmed": False}

# 解約処理実行ノード（このノードの手前でHuman-in-the-loopによる一時停止を行います）
def execute_cancellation(state: SupportState):
    # データベースの契約状況を「キャンセル済み」に変更
    update_customer_db(state["customer_id"], "canceled")
    return {"contract_status": "canceled", "messages": [("assistant", "解約手続きが完了しました。")]}

# グラフの構築
workflow = StateGraph(SupportState)
workflow.add_node("initialize", initialize_session)
workflow.add_node("request_confirm", request_cancel_confirmation)
workflow.add_node("execute_cancel", execute_cancellation)

workflow.add_edge(START, "initialize")
workflow.add_edge("initialize", "request_confirm")

# 実行前に人間の確認（同意）を必要とするため、execute_cancelの直前で一時停止する
workflow.add_edge("request_confirm", "execute_cancel")
workflow.add_edge("execute_cancel", END)

# actionノードの前に一時停止（ユーザーまたは管理者の承認待ち）を挟んでコンパイル
app = workflow.compile(interrupt_before=["execute_cancel"])

このように、情報の検索、ユーザーへのヒアリング、安全な手続きの実行、といった一連のワークフローを状態遷移として定義することで、ロボット的ではない、顧客に寄り添った丁寧かつ確実なサポート自動化を実現できます。

6.2 自己修正型RAG（検索結果や生成結果を自己評価して修正するループ）

従来のRAGシステムは、検索したドキュメントの中に質問の答えが含まれていなかったり、LLMが検索内容を無視して誤った回答（ハルシネーション）を生成したりする問題がありました。これらを解決するために、検索と生成のプロセスを自己評価し、自動でリトライや修正を行う「自己修正型RAG（Corrective RAG）」をLangGraphのループ構造で構築します。

この仕組みでは、検索したドキュメントがユーザーの質問に対して「本当に役に立つか」を評価する評価ノードを設けます。もし役に立たないと判断された場合は、検索クエリ（キーワード）をAI自身に書き換えさせ、Web検索などを通じて不足している情報を再取得します。また、生成された回答が「元データに基づいて正しく書かれているか」を検証するノードも作成します。矛盾や事実誤認があれば、生成ノードに戻って書き直させます。

以下は、自己修正型RAGを定義するグラフコード例です。

from typing import List, TypedDict
from langgraph.graph import StateGraph, START, END

class RAGState(TypedDict):
    question: str
    documents: List[str]  # 検索された文書
    generation: str      # 生成された回答
    search_retry_count: int  # 検索クエリ書き換えの再試行回数

# 1. 外部データベースから情報を検索するノード
def retrieve(state: RAGState):
    docs = vector_store.similarity_search(state["question"])
    return {"documents": docs}

# 2. 検索したドキュメントの適合性を評価するノード
def grade_documents(state: RAGState):
    # LLMを用いて「質問の答えがドキュメントに含まれているか」をチェック
    grade = grader_llm.invoke({"question": state["question"], "docs": state["documents"]})
    if grade.is_relevant:
        return {"search_retry_count": state["search_retry_count"]}
    else:
        # 関連性が低いと判断された場合は、再検索フラグとしてカウントを増やす
        return {"search_retry_count": state["search_retry_count"] + 1}

# 3. 再検索のためにクエリを書き換えるノード
def rewrite_query(state: RAGState):
    # LLMを使って新しい検索キーワードを生成
    better_question = query_llm.invoke(state["question"])
    new_docs = vector_store.similarity_search(better_question)
    return {"documents": new_docs, "question": better_question}

# 4. 回答を生成するノード
def generate(state: RAGState):
    answer = generator_llm.invoke({"question": state["question"], "docs": state["documents"]})
    return {"generation": answer}

# グラフの定義
workflow = StateGraph(RAGState)
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade_docs", grade_documents)
workflow.add_node("rewrite_query", rewrite_query)
workflow.add_node("generate", generate)

workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "grade_docs")

# 適合性評価に基づいて分岐させる条件付きエッジ
def decide_to_generate(state: RAGState):
    # ドキュメントが不適合で、再試行回数が上限に達していなければクエリ書き換えに進む
    if state["search_retry_count"] > 0 and state["search_retry_count"] < 3:
        return "rewrite_query"
    return "generate"

workflow.add_conditional_edges(
    "grade_docs",
    decide_to_generate,
    {
        "rewrite_query": "rewrite_query",
        "generate": "generate"
    }
)
workflow.add_edge("rewrite_query", "grade_docs")  # 再検索後は再度評価を行う
workflow.add_edge("generate", END)

app = workflow.compile()

この一連の「検索 -> 評価 -> （失敗なら再クエリ・再検索） -> 生成 -> 検証 -> （失敗なら再生成） -> 完了」というループ処理を実装することで、誤情報の極めて少ない、極めて信頼性の高い知識検索システムを構築することが可能になります。

6.3 コードの自動生成・実行・修正ループ

AIにプログラムのコードを書かせる際、一発で完璧に動作するコードを出力することは稀です。プログラミング言語の細かな仕様変更やライブラリの依存関係により、軽微なエラーが発生することが多いためです。そこで、AIが書いたコードを実際にテスト環境で実行し、エラーが出た場合はそのエラーログをAIにフィードバックして自動修正させる「コーディングループ」を実装します。

まず、AIがユーザーの要望に従ってコードを出力する生成ノードを実行します。次に、そのコードをサンドボックス（安全な実行環境）で実行し、テストを行う実行ノードを作成します。テストが成功すれば処理は終了しますが、エラーが発生した場合は、エラーメッセージとスタックトレース（エラーの詳細な発生場所）を状態（State）に書き込み、生成ノードへと差し戻す条件付きエッジを張ります。差し戻された生成ノードでは、エラーの原因をLLMが分析し、コードの修正版を出力します。

以下は、コード自動生成・実行・修正ループの実装コード例です。

import subprocess
from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class CodingState(TypedDict):
    task: str       # 実装したいプログラムの要件
    code: str       # LLMが生成したPythonコード
    error: str      # 実行した際のエラー内容
    iterations: int # 修正ループの回数

# 1. コード生成（または修正）ノード
def generate_code(state: CodingState):
    # エラーがある場合はエラーを修正するためのプロンプトでLLMを呼び出す
    if state["error"]:
        prompt = f"タスク: {state['task']}\n生成したコード:\n{state['code']}\nエラー内容:\n{state['error']}\n上記エラーを修正してください。"
    else:
        prompt = f"タスク: {state['task']}\nこの要件を満たすPythonコードを書いてください。"
    
    code_output = coding_llm.invoke(prompt)
    return {"code": code_output, "iterations": state["iterations"] + 1, "error": ""}

# 2. コードを実行してテストするノード
def execute_test(state: CodingState):
    try:
        # 生成されたコードを一時ファイルに保存して実行する（安全な環境を想定）
        with open("temp_script.py", "w", encoding="utf-8") as f:
            f.write(state["code"])
        
        # サブプロセスとして実行し、結果を待つ
        result = subprocess.run(["python", "temp_script.py"], capture_output=True, text=True, timeout=5)
        
        if result.returncode == 0:
            return {"error": ""}  # テスト成功
        else:
            # 実行エラー（例外など）が発生した場合
            return {"error": result.stderr}
    except Exception as e:
        return {"error": str(e)}

# グラフの定義
workflow = StateGraph(CodingState)
workflow.add_node("generator", generate_code)
workflow.add_node("tester", execute_test)

workflow.add_edge(START, "generator")
workflow.add_edge("generator", "tester")

# テスト結果に基づいてループか終了かを分岐する条件付きエッジ
def check_test_result(state: CodingState):
    if state["error"] and state["iterations"] < 5:
        # エラーがあり、実行回数が上限（5回）未満であれば、再生成（修正）に戻る
        return "generator"
    return "end"

workflow.add_conditional_edges(
    "tester",
    check_test_result,
    {
        "generator": "generator",
        "end": END
    }
)

app = workflow.compile()

この修正ループを最大3〜5回などと決めて回すことで、人間の手を煩わせることなく、最終的に「構文エラーがなくテストを通過したプログラム」だけをユーザーに提供する自律的なコード開発ツールを実現できます。