Session 04: LangGraph + RAG総合演習

HANDS-ON 01 / 15 MIN

LangGraphの基礎 ── State / Node / Edge

05_langgraph_basics.py

EXERCISE

python 05_langgraph_basics.py を実行

CONCEPT

State

グラフ全体で共有するデータ。辞書に型を付けたもの

CONCEPT

Node

処理の単位。普通のPython関数。Stateを受け取り、更新したい値を辞書で返す

CONCEPT

Edge

処理の順番を定義。add_edgeで繋ぐ

なぜLangGraphを使うのか ── 単純なスクリプトとの違い

Session 02-03で書いたコードは「上から順に実行するだけ」のスクリプトだった。しかし実際のAIシステムでは「検索結果が不十分なら再検索する」「エラーが出たら別の処理に分岐する」といった条件分岐が必要になる。LangGraphはこうした複雑な処理フローを、State（データ）+ Node（処理）+ Edge（順序）の3要素だけで見通しよく定義できる。

Step 1 ── Stateを定義する

from typing import TypedDict

class SimpleState(TypedDict):
    question: str   # ユーザーの質問
    answer: str     # AIの回答

TypedDict はPythonの型ヒント機能の一つ。辞書（dict）のキー名と値の型をあらかじめ宣言できる
class SimpleState(TypedDict): は「SimpleStateという名前の辞書で、question（文字列）とanswer（文字列）のキーを持つ」と定義している
LangGraphでは、このStateがグラフ全体の「共有メモ帳」として機能する。あるNodeがquestionを読み、別のNodeがanswerを書き込む、という形でデータが流れる
Session 02で学んだPythonの辞書（dict）に型情報を追加したもの、と理解すればよい

Python公式: TypedDict ドキュメント

Step 2 ── Nodeを2つ作る

def log_node(state: SimpleState) -> dict:
    print(f"受け取った質問: {state['question']}")
    return {}

def answer_node(state: SimpleState) -> dict:
    response = model.invoke(state["question"])
    return {"answer": response.content}

LangGraphのNodeは特別なものではなく、普通のPython関数。Session 02で学んだ def greet(name): と同じ
唯一のルール: 引数としてStateを受け取り、更新したいキーと値の辞書を返す
log_node は return {} で空辞書を返している。これは「Stateを何も変更しない」という意味。printで質問を表示するだけのノード
answer_node は return {"answer": response.content} で、Stateのanswerキーを更新している
ノードの中ではAPI呼び出し、データベース操作、ファイル読み書きなど、好きな処理を書ける

Step 3 ── グラフを組み立てる

from langgraph.graph import StateGraph, END

graph = StateGraph(SimpleState)
graph.add_node("log", log_node)
graph.add_node("answer", answer_node)
graph.set_entry_point("log")
graph.add_edge("log", "answer")
graph.add_edge("answer", END)

app = graph.compile()

log_node

→

answer_node

→

END

StateGraph(SimpleState): 「SimpleStateを共有データとして使うグラフ」を新規作成
add_node("log", log_node): "log"という名前でlog_node関数をグラフに登録
add_node("answer", answer_node): "answer"という名前でanswer_node関数を登録
set_entry_point("log"): 「最初に実行するノードはlogです」と指定
add_edge("log", "answer"): 「logの次はanswerを実行」と接続
add_edge("answer", END): 「answerの次は終了」と接続。ENDはLangGraphが提供する特殊な終了マーカー
compile(): 上の設定を確定し、実行可能なアプリケーションにする。コンパイル後に app.invoke() で実行できる

LangGraph: Low-level concepts

Step 4 ── 実行

result = app.invoke({
    "question": "LangGraphとは何ですか？",
    "answer": ""
})
print(f"回答: {result['answer']}")

HANDS-ON 02 / 15 MIN

総合演習 ── LangGraphでRAGシステムを作る

06_rag_with_langgraph.py

EXERCISE

python 06_rag_with_langgraph.py を実行

PDF読込

→

分割

→

ベクトル化

→

格納

→

質問で検索

→

LLMが回答

午前のすべてがここに集約される

このコードには Session 01 で確認したAPI接続、Session 02 で学んだChatOpenAI/Prompt Template/LCEL、Session 03 で体験した5つのRAGコンポーネント、そしてこのSession 04のLangGraphが全て含まれている。新しい要素は一つもない。既に学んだパーツを組み合わせているだけ。

Code コードの全体構成

# --- 準備 ---
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

loader = PyPDFLoader("data/sample.pdf")
pages = loader.load()
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
chunks = splitter.split_documents(pages)
vectorstore = FAISS.from_documents(chunks, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# --- State ---
class RAGState(TypedDict):
    question: str
    documents: List[str]
    answer: str

# --- Node ---
def retrieve_node(state):
    docs = retriever.invoke(state["question"])
    return {"documents": [doc.page_content for doc in docs]}

def generate_node(state):
    context = "\n\n".join(state["documents"])
    prompt = ChatPromptTemplate.from_messages([
        ("system", "以下の文書を参考に質問に回答してください。\n\n{context}"),
        ("human", "{question}")
    ])
    chain = prompt | model | StrOutputParser()
    return {"answer": chain.invoke({"context": context, "question": state["question"]})}

# --- Graph ---
graph = StateGraph(RAGState)
graph.add_node("retrieve", retrieve_node)
graph.add_node("generate", generate_node)
graph.set_entry_point("retrieve")
graph.add_edge("retrieve", "generate")
graph.add_edge("generate", END)
app = graph.compile()

retrieve_node

→

generate_node

→

END

retrieve_node: Stateからquestion（質問文）を取り出し → Retriever（Session 03のStep 5）に渡して検索 → 検索結果をdocumentsとしてStateに書き込む
generate_node: Stateからdocuments（検索結果）とquestion（質問）を取り出す → documentsを改行で連結してcontextを作る → ChatPromptTemplate（Session 02のStep 2）でsystemメッセージに文書を埋め込む → LCEL（Session 02のStep 3）でLLMに渡す → 回答をanswerとしてStateに書き込む
つまり retrieve_node = RAGの「検索」担当、generate_node = RAGの「生成」担当。Session 01の冒頭で示したRAGの全体像がそのまま実現されている

Map 各パーツの対応関係

コード	学んだ場所
PyPDFLoader, TextSplitter, FAISS, Retriever	Session 03
ChatPromptTemplate, LCEL	Session 02
StateGraph, add_node, add_edge	この演習の前半

# 期待される出力
質問: PCが起動しない場合の対処手順を教えてください

回答:
PCが起動しない場合の対処手順は以下の通りです。
1. 電源ケーブルが正しく接続されているか確認する。
2. 電源ボタンを10秒間長押しして強制シャットダウン...

確認ポイント

PDFの内容に基づいた回答が返ってくれば成功。

EXERCISE

自分で試してみよう

質問を変えてみる
Retrieverの検索件数を変えてみる（k=5）
SystemMessageを変えて口調の違いを確認する

難易度: 低

A. AIの人格を変える

SystemMessageを書き換えて、丁寧語やフランクな口調など異なるトーンで回答させてみる

難易度: 中

B. 検索結果を回答に添付する

generate_nodeの出力にdocumentsを含め、回答の根拠となった文書チャンクも表示する

難易度: 中

C. 複数の質問に連続で答える

質問リストをループで回し、同じグラフに複数回invokeする処理を書いてみる

午前の部終了

午前の部はここで終了。午後はAI駆動開発（Claude Code）に入ります。

午前の振り返りチェックリスト

LLMをPythonから呼び出せた（model.invoke）
temperatureの意味がわかった
Prompt TemplateとLCELでChainを組めた
PDFをDocument Loaderで読み込めた
テキストをEmbeddingでベクトルに変換する意味がわかった
FAISSでベクトル検索ができた
LangGraphでState/Node/Edgeを使ったワークフローを組めた
RAGシステム全体が動作するのを確認できた

Session 04 参考リンク

LangGraph 公式ドキュメント LangChain RAG チュートリアル LCEL 概要 LangGraph サンプル集（GitHub） LangGraph Quickstart Python TypedDict RAG概念の解説（LangChain）

LangGraph +RAG総合演習

python 05_langgraph_basics.py を実行

State

Node

Edge

python 06_rag_with_langgraph.py を実行

自分で試してみよう

A. AIの人格を変える

B. 検索結果を回答に添付する

C. 複数の質問に連続で答える

LangGraph +
RAG総合演習