前回の記事では、LangGraphの概要とその特徴について紹介しました。
LangGraphとは?マルチステップLLMアプリを構築できるフレームワークを試す
今回はその続編として、実際にLangGraphを使った動くコードを紹介しながら、ノードの構成や状態遷移の流れを解説していきます。
簡単な質問応答アプリを例に、LangGraphの使い方と考え方を整理していきます。
1. 環境構築
今回の実装は、Google Coloboratory(以下Colab) 上で手軽に実行できます。
特別なローカル環境の準備は不要なので、Pythonが少しわかる方であればすぐに試せます。
使用した主なパッケージとバージョンは以下の通りです。
pip install langchain==0.3.0 langchain-openai==0.2.0 langgraph==0.2.22
依存関係の違いによるエラーを防ぐため、なるべくバージョンを揃えることをおすすめします。
以降では、Colab上で実行できるコードを紹介します。
2. 実行コード: LangGraphの処理を構築
本章では、LangGraphを使って質疑応答アプリの処理をどのように定義していくか、実際のコードをベースに解説します。
今回使用するノートブック全体はこちらで確認できます: [LangGraphTutrial](https://github.com/dgkmtu/LangGraphTutrial/blob/main/LangGraphTutrial/tutrial.ipynb)
2.1 処理の全体像
この質問応答アプリは、以下の3ステップで構成されています。
- ロール選定 (質問に応じて、回答担当の"専門家"を選ぶ)
- 回答生成 (選ばれたロールに応じた回答を生成)
- 品質チェック (回答が十分かを確認し、回答が不十分なら再回答)
この一連の流れをLangGraphで定義し、状態繊維として管理していきます。
2.2 コードの構成
2.2.1 ライブラリとAPIキーの準備
APIキーの設定と必要なライブラリをインポートします。
# OpenAIのAPIキーを環境変数に設定(ハードコードは避けてください)
import os
os.environ["OPENAI_API_KEY"] = "YOUR_API_KEY" # セキュリティのため実際は.envファイルなどが望ましい
import operator
from typing import Annotated, Any
from langchain_core.pydantic_v1 import BaseModel, Field # 状態モデル用
from langgraph.graph import StateGraph, END # グラフ構築用
from langchain_openai import ChatOpenAI # OpenAIのチャットモデル
from langchain_core.runnables import ConfigurableField # LLM設定変更用
from langchain_core.prompts import ChatPromptTemplate # プロンプトテンプレート
from langchain_core.output_parsers import StrOutputParser # LLMの文字列出力を扱う
2.2.2 回答ロールの定義
# 回答ロールの定義(番号付き)
ROLES = {
"1": {
"name": "一般知識エキスパート",
"description": "幅広い分野の一般的な質問に答える",
"details": "幅広い分野の一般的な質問に対して、正確でわかりやすい回答を提供してください。"
},
"2": {
"name": "生成AI製品エキスパート",
"description": "生成AIや関連製品、技術に関する専門的な質問に答える",
"details": "生成AIや関連製品、技術に関する専門的な質問に対して、最新の情報と深い洞察を提供してください。"
},
"3": {
"name": "カウンセラー",
"description": "個人的な悩みや心理的な問題に対してサポートを提供する",
"details": "個人的な悩みや心理的な問題に対して、共感的で支援的な回答を提供し、可能であれば適切なアドバイスも行なってください。"
}
}
どの専門家(ロール)に回答を任せるかを、LangGraph上で自動的に判断するためのマッピングです。
2.2.3 状態の定義 (Stateモデル)
# LangGraphに渡す状態の定義
class State(BaseModel):
query: str = Field(..., description="ユーザーからの質問")
current_role: str = Field(
default="", description="選定された回答ロール"
)
messages: Annotated[list[str], operator.add] = Field(
default=[], description="回答履歴"
)
current_judge: bool = Field(
default=False, description="品質チェックの結果"
)
judgment_reason: str = Field(
default="", description="品質チェックの判定理由"
)
LangGraphに渡す状態(質問文・選ばれたロール・回答など)をまとめて管理します。
2.2.4 各ノードの関数定義
- selection_node → 質問をもとにロール番号を判定
- answering_node → 選ばれたロールに応じた回答を生成
- check_node → 回答の品質をチェック
# OpenAIのチャットモデル(gpt-4o-mini)を初期化
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)
llm = llm.configurable_fields(max_tokens=ConfigurableField(id='max_tokens'))
# ノード①:ロールの選定
def selection_node(state: State) -> dict[str, Any]:
query = state.query
role_options = "\n".join([f"{k}. {v['name']}: {v['description']}" for k, v in ROLES.items()])
prompt = ChatPromptTemplate.from_template(
"""質問を分析し、最も適切な回答担当ロールを選択してください。
選択肢:
{role_options}
回答は選択肢の番号(1、2、または3)のみを返してください。
質問: {query}
""".strip()
)
# max_tokens=1 に設定して、1文字(番号)のみ返すよう制限
chain = prompt | llm.with_config(configurable=dict(max_tokens=1)) | StrOutputParser()
role_number = chain.invoke({"role_options": role_options, "query": query})
selected_role = ROLES[role_number.strip()]["name"]
return {"current_role": selected_role}
# ノード②:ロールに応じた回答の生成
def answering_node(state: State) -> dict[str, Any]:
query = state.query
role = state.current_role
role_details = "\n".join([f"- {v['name']}: {v['details']}" for v in ROLES.values()])
prompt = ChatPromptTemplate.from_template(
"""あなたは{role}として回答してください。以下の質問に対して、あなたの役割に基づいた適切な回答を提供してください。
役割の詳細:
{role_details}
質問: {query}
回答:""".strip()
)
chain = prompt | llm |StrOutputParser()
answer = chain.invoke({"role": role, "role_details": role_details, "query": query})
return {"messages": [answer]}
# 回答品質の評価スキーマ
class Judgement(BaseModel):
judge: bool = Field(default=False, description="判定結果")
reason: str = Field(default="", description="判定理由")
# ノード③:品質チェック(OKなら終了、NGなら再実行)
def check_node(state: State) -> dict[str, Any]:
query = state.query
answer = state.messages[-1]
prompt = ChatPromptTemplate.from_template(
"""以下の回答の品質をチェックし、問題がある場合は'False'、問題がない場合は'True'を回答してください。
また、その判断理由も説明してください。
ユーザーからの質問: {query}
回答: {answer}
""".strip()
)
chain = prompt | llm.with_structured_output(Judgement)
result: Judgement = chain.invoke({"query": query, "answer": answer})
return {
"current_judge": result.judge,
"judgement_reason": result.reason
}
2.2.5 グラフの構築と遷移設定
# LangGraphグラフの構築
workflow = StateGraph(State)
# 各ノードをグラフに追加
workflow.add_node("selection", selection_node)
workflow.add_node("answering", answering_node)
workflow.add_node("check", check_node)
# 処理開始点を指定(最初はロール選定から)
workflow.set_entry_point("selection")
# 各ノード間の遷移(エッジ)を定義
workflow.add_edge("selection", "answering")
workflow.add_edge("answering", "check")
# 条件分岐:checkノードの結果によって終了 or やり直し
workflow.add_conditional_edges(
"check",
lambda state: state.current_judge, # Trueなら終了、Falseなら再実行
{True: END, False: "selection"}
)
# グラフの最終的なコンパイル(実行可能状態に)
compiled = workflow.compile()
ここでは、LangGraphのグラフ構文を使って、3つの処理ノード(選定・回答・評価)をどのように接続・遷移させるかを定義していきます。
2.2.6 実行と結果の出力
# 初期状態としてユーザーの質問を与える
initial_state = State(query="生成AIについて教えてください")
# 実行(LangGraphが自動でループや遷移を管理)
result = compiled.invoke(initial_state)
# 最終出力(最終回答メッセージ)
print(result["messages"][-1])
ユーザーからの質問に対して、選ばれた"専門家"が答えた結果が出力されます。
2.3 実際の出力例
生成AIとは、人工知能の一分野であり、特にデータを基に新しいコンテンツを生成する能力を持つ技術を指します。これには、テキスト、画像、音声、動画など、さまざまな形式のコンテンツが含まれます。生成AIは、機械学習アルゴリズム、特に深層学習を利用して、既存のデータからパターンを学習し、それを基に新しいデータを生成します。 ### 主な技術と応用 1. **自然言語処理(NLP)**: テキスト生成や翻訳、要約などに使用されます。例えば、GPT(Generative Pre-trained Transformer)シリーズは、文章を生成する能力に優れています。 2. **画像生成**: GAN(Generative Adversarial Networks)やVQ-VAE(Vector Quantized Variational Autoencoders)などの技術を用いて、リアルな画像を生成することができます。DALL-EやMidjourneyなどが代表的な例です。 3. **音声生成**: 音声合成技術を用いて、人間の声に似た音声を生成することができます。これにより、ナレーションや対話システムなどが実現されています。 4. **動画生成**: まだ発展途上ですが、AIを用いて短い動画を生成する技術も進化しています。 ### 利点と課題 - **利点**: コンテンツの迅速な生成、コスト削減、クリエイティブなアイデアの支援などが挙げられます。 - **課題**: 偽情報の生成、著作権の問題、倫理的な懸念などがあり、これらに対する対策が求められています。 生成AIは、ビジネス、エンターテインメント、教育など多くの分野で革新をもたらす可能性を秘めていますが、その利用には慎重なアプローチが必要です。
2.4 グラフ描画で流れを可視化
以下のコードで、LangGraphがどのようにノードを遷移しているかを画像で確認できます:
▪️必要パッケージのインストール
# LangGraphのグラフ画像描画に必要なパッケージをインストール
!apt-get install graphviz libgraphviz-dev pkg-config
!pip install pygraphviz
▪️画像化
from IPython.display import Image
Image(compiled.get_graph().draw_png())
3. フロー図で処理の流れを視覚化
LangGraphの強みのひとつは、「状態遷移の構造をグラフとして視覚化できること」です。
以下は、実際に生成された状態遷移のフロー図です。
この図では以下のような流れで処理が進みます:
-
selection
:質問内容に応じて適切なロール(専門家)を選定 -
answering
:選ばれたロールに基づいて回答を生成 -
check
:回答の品質を判定 -
True
(品質OK)→ 終了 /False
(品質NG)→ ロール選定に戻って再実行
このように、品質が不十分だった場合は再びロール選定から処理を繰り返すようになっています。
これは「適切な回答者が選ばれていなかった可能性がある」という前提に基づき、役割選定を見直す設計です。
4. いろいろな質問で試してみる
以下のような質問を入れると、選ばれるロールや回答が変わってくるはずです。
- 「最近よく聞く生成AIってなんですか?」(→ 生成AIエキスパートが選ばれる)
- 「仕事のストレスが多くて困っています」(→ カウンセラー)
- 「チョコレートの歴史について教えてください」(→ 一般知識エキスパート)
ぜひ自分でもいろいろな質問で試してみてください。
次回予告
今回の流れでLangGraphの基礎はわかりました。
次回は、このLangGraphを応用してもう少し高度なグラフを構築していきたいと思います。