【技術書アウトプット】LangGraphの本読んだら、これからの時代はLangGraphだってなったのでアウトプットするッッ
背景
この記事は『LangChain と LangGraph による RAG・AI エージェント［実践］入門』という書籍のアウトプット記事になります。
結論として、本書はかなり学びになる本でした。
LangChainの基礎、AIエージェントとは何か、LangGraphのハンズオン形式の実装方法などかなり深く解説しており、LangChain・LangGraphを用いたアプリケーション開発の基礎を理解できる内容になっています。
LangChainについてもっと理解を深めたい、LangGraphを学んでみたいと思っている人には非常にお勧めできる一冊となっています。
本書を読んで、自分ももっとLangChain, LangGraphに対する理解を深めたいなと思ったのでアウトプットとして記事を書くことにしました。
もし興味がある方は最後までお読みいただけますと幸いです。🐶
この記事を読むことで以下を理解することができます！
- LangGraphとは何か
- Graph
- State
- Node
- Edge
- どのようにしてLangGraphを実装すればよいか
- 「質問に答えるだけでライフプランを提案してくれるエージェント」のコードを使って解説
本の概要
本書ではLangChain、LangGraphを用いてAIエージェントを実装するための基礎が学べます。
LangChain・LangGraph・RAGを用いたAIアプリケーションの開発をしたい方には非常におすすめな一冊となっています。
個人的には「LangChainの全体構成がどうなっているか」、「なぜChainをパイプラインで繋げることができるのか」、「ChainはRunnable型について」など学びになる内容が盛りだくさんでとても学びになりました。
その他にもRAGを実行する上でのさまざまなパターンの解説、LangGraphの基礎およびハンズオン形式の解説もあり、非常におすすめです。
LangGraphとは
LangGraphの5つの特徴
LangGraphには5つの特徴があります。
それは、「①明示的なステート管理」「②条件分岐とループ」「③拡張が容易」「④デバッグとテストの容易さ」「⑤チェックポイントとリカバリ」です。
順に解説していきます。
1. 明示的なステート管理
LangChainではStateという構造データを管理する仕組みがあります。
Stateには会話履歴や中間結果といった情報を保持することができます。
このStateは処理が終わるたびに更新され、後続の処理に渡すことができます。
そのため、LangGraph長期的なタスクにおいて一貫性のある情報の受け渡しが可能になります。
例えばですが、Stateでは以下のような構造化データを定義できます。
class ProfileState(BaseModel):
user_input: str = Field(..., description="ユーザーの質問の回答結果")
profile: str = Field(default="", description="ユーザープロファイル情報")
recommendations: Annotated[list[str], operator.add] = Field(
default_factory=list, description="推奨されたアクティビティのリスト"
)
schedule: str = Field(default="", description="生成されたスケジュール")
is_sufficient: bool = Field(default=False, description="情報が十分かどうか")
iteration: int = Field(default=0, description="プロファイル生成と推奨の反復回数")
final_output: str = Field(
default="", description="最終的な成果物。ユーザーに推奨する行動が書かれてある"
)
userからの入力、生成した情報が十分かどうか（条件分岐の際に使用）、ループが何回行われたかといった情報をStateでは保持することができます。
これらのデータをそれぞれの処理の時に渡すことで、フローの処理が長くなったとしても一貫したコンテキストを渡せるようになります。
2. 条件分岐とループ
LangGraphはノードやエッジなどのコンポーネントを用いて、以下のようなワークフローを構築します。
※ノードとエッジは後ほど解説します。
このようなグラフ構造のため、条件分岐を簡単に導入することができます。
「Xという条件を満たすまで繰り返す」のようなループ処理が簡単に構築することができるのがLangGraphの強みです。
これは従来のLangChainのChainでは難しかったことです。
LangChainでもLangGraphのように条件分岐をつかしたり、ループ処理を実装することは可能でしたが、それには複雑な処理が必要でした。
LangGraphを用いることでその複雑な処理を簡単に実装できるようになったのです！
3. 拡張が容易
LangGraphはノードとエッジを組み合わせてワークフローを構築することから、新しい処理を追加しようと思ったらノードとエッジを追加するだけでいいので、拡張が容易です。
4. デバッグとテストの容易さ
LangGraphはノードを独立してテストすることができるので、テストやデバッグが容易にできる。
※今回は使用しません
5. チェックポイントとリカバリ
LangGraphはステートのチェックポイントを作成することができます。
そのため長期的なタスクを一時中断し、しばらくたってから再開するみたいなことができる。
※今回は使用しません
LangGraphを使う上で覚えておくべき概念
ここからはLangGraphを使う上で覚えておくべき概念を解説していきます。
1. Graph
GraphはLangGraphのグラフ全体を管理するコンポーネントになります。
以下のようにStateGraphを呼び出し、引数としてStateを渡すことで初期化を行います。
# グラフの初期化
workflow = StateGraph(ProfileState)
ここで定義したGraphに対して後述するNodeやEdgeを追加していくことでワークフローを構築していきます。
2. State
StateはGraphのNodeにて更新された値を保持する仕組みのこと。
Stateでは、会話履歴やループ回数、情報が十分かどうかを示すフラグ（条件分岐の際に使用）などの中間データなどを渡すことができます。
Stateを用いることで各NodeはStateのデータを読み書きしながら処理を進めることが可能になるのです。
Stateは以下のように定義します。
class ProfileState(BaseModel):
user_input: str = Field(..., description="ユーザーの質問の回答結果")
profile: str = Field(default="", description="ユーザープロファイル情報")
recommendations: Annotated[list[str], operator.add] = Field(
default_factory=list, description="推奨されたアクティビティのリスト"
)
schedule: str = Field(default="", description="生成されたスケジュール")
is_sufficient: bool = Field(default=False, description="情報が十分かどうか")
iteration: int = Field(default=0, description="プロファイル生成と推奨の反復回数")
final_output: str = Field(
default="", description="最終的な成果物。ユーザーに推奨する行動が書かれてある"
)
定義したStateはGraphを初期化する際に、引数として渡します。
# グラフの初期化
workflow = StateGraph(ProfileState)
こうすることで、LangGraphはStateをNode内で参照することができるようになるのです。
3. Node
LangGraphのGraphはNodeの集合体です。
LangGraphではGraphに対して、NODE_A, NODE_B, NODE_Cを追加していくことで、ワークフローを作成していきます。
GraphにNodeを追加する方法は非常に簡単です。
以下のように{graph_name}.add_node(NODE_NAME, 関数)とすれば完了です。
# 各ノードの追加
workflow.add_node("analysis_user", self._analysis_user) # Graphにanalysis_userというグラフを追加する
workflow.add_node("generate_recommendations", self._generate_recommendations)
workflow.add_node("evaluate_recommendations", self._evaluate_recommendations)
workflow.add_node("finalize_output", self._finalize_output)
第2引数に渡す関数は以下のように引数にStateを受け取り、更新したいStateのオブジェクトを返します。
def _analysis_user(self, state: ProfileState) -> dict[str, Any]:
# stateの値を使って処理を実行するコードを記述
profile = self.analysis_user.run(state.user_input)
# レスポンスは更新したいStateの辞書型のオブジェクト
return {
"profile": profile,
"iteration": state.iteration + 1,
}
Stateのオブジェクトを返すことで、LangGraphではノードの処理後に毎回Stateが更新されるようになるのです。
4. Edge
EdgeはNode同士の推移を定義するコンポーネントです。
GraphにNODE_A, NODE_Bを追加しただけではNode同士の繋がり、つまりEdgeが定義されていないのでNODE_Aの後にNODE_Bの処理を実行するという流れを定義できていません。
NODE_Aの処理の後にはNODE_Bに遷移するとしたい場合はEdgeを使ってNODE_AとNODE_Bを繋げる必要があります。
そのためには以下のようにadd_edge(from, to)を用いてNode同士を繋げる必要があります。
workflow.add_edge("NODE_A", "NODE_B")
workflow.add_edge("NODE_B", "NODE_C")
こうすることで、NODE_Aの処理が終わったら、NODE_Bへ、NODE_Bの処理が終わったら、NODE_Cへというルーティングを定義することができます。
Edgeには3種類のEdgeが存在します。
それは、「①エントリーポイント」「②通常エッジ」「③条件付きエッジ」の3つです。
-
エントリーポイント
# エントリーポイント設定 workflow.set_entry_point("NODE_A")グラフの開始を定義するエッジです。
このように書くことで、「NODE_Aが初めのNodeだよ」ということを定義できます。 -
通常エッジ
「NODE_Aの処理が完了したらNODE_Bに遷移する」という基礎的なルーティングを行いたい場合に使うエッジです。
# 通常エッジ workflow.add_edge("NODE_A", "NODE_B")このように書くことで、NODE_Aの処理が完了した後に無条件でNODE_Bの処理が実行されるようになります。
-
条件付きエッジ
条件に基づいて遷移先を選択するエッジです。
add_conditional_edges()で定義することができます。# 条件付きエッジの追加 workflow.add_conditional_edges( "NODE_B", lambda state: state.judge_flag, {True: "NODE_C", False: "NODE_A"}, )第1引数にはノードの名前、第2引数には条件のロジック、第3引数には第2引数の結果どのNodeに遷移するかを定義します。
サンプルコードでは、第2引数で条件を定義、その結果Trueの場合はNODE_Cへ、Falseの場合はNODE_Aに処理を戻すということを行っています。
以上でLangGraphの基本的な概念の説明は終わりです。
まとめると、LangGraphはNodeとEdgeを組み合わせることでGraphを作成する。
Nodeの処理時にはStateが渡され、一貫した値を参照した上で処理を進めることができる。
Edgeを使ってNode同士を連結させ、条件分岐などを組み合わせることでループ処理が可能になる。
という感じです。
次からはLangGraphを用いて簡単なアプリケーションを実装しましたので、そのコードを解説しつつ、LangGraphの理解を一緒に深めていきましょうっ！
LangGraphを使って『人生戦略提案AIエージェント』を作ってみる
どんなものを作ったか
書籍では、入力したアプリケーションの簡単な要件からペルソナを作成し、ペルソナにインタビューを行い、最終的に要件定義署を作成するというアプリケーションを開発していました。
同じのを解説してもつまらないので、今回「人生戦略提案AIエージェント」を作ってみました。
作ったコードは以下のリポジトリで管理しております。
アプリの動作についてざっと説明します。
以下のようにpythonファイルを実行すると質問がされるので、よしなに答えたら…
Q1. あなたの性別は？ -> 男性
Q2. あなたの年齢は？ -> ２４歳
Q3. あなたの性格は？ -> おとなしめ、仲良くなった人にはとことんオープンになる、内省しがち、かなりマメ、日記とか毎日書くタイプ
Q4. あなたのMBTIは？ -> INTJ
Q5. 好きなもの・ことは？ -> 学習すること、読書すること、テニスをすること、ラーメン、自分が朝に決めたタスクや目標を着実に達成すること、プログラミングでものを作ること
Q6. 嫌いなもの・ことは？ -> 人生をプラスにしない行動、活動、やりたくないのにやってしまう悪習慣、辛いもの、
Q7. 得意なことは？ -> スケジュールを管理すること、細かくタスクを管理すること
Q8. 苦手なことは？ -> 同じ作業の繰り返しなど
Q9. 現在の仕事は？ -> エンジニア
Q10. 将来どうなっていたい？ -> エンジニアとしてものを作り続けたい、技術力を高め人に毎日使ってもらうサービスを作りたい
Q11. どんな人に憧れる？ -> サービスを0から作り、それを改善し続けている人、人に使ってもらえるサービスを開発した人
Q12. これは長く続けられるなってことは？ -> 学習、読書、テニス、プログライング
最終的にこんな感じで人生戦略を提案してくれます。
### 人生戦略: 目的別・優先度別
#### 目的1: 技術力の向上とスキルセットの拡充
1. **新しいプログラミング言語の習得**
- **実行時期**: 24歳から25歳
- **方法**: 彼が興味を持っている分野に関連するプログラミング言語を選び、オンラインコースや書籍を活用して学習を進める。
- **達成基準**: 新しい言語で小規模なプロジェクトを完成させることができる。
- **理由**: 学習を楽しむ彼にとって、新しい言語の習得は論理的思考を深め、将来的なプロジェクトに役立つ。
2. **個人プロジェクトの開始**
- **実行時期**: 25歳から26歳
- **方法**: 興味のある分野で小規模なプロジェクトを立ち上げ、独立して作業する。
- **達成基準**: プロジェクトを公開し、ユーザーからのフィードバックを得る。
- **理由**: 影響力のあるサービスを作るための実践的なスキルを磨く。
#### 目的2: 効率的な作業環境の構築
3. **効率的なタスク管理ツールの導入**
- **実行時期**: 24歳
- **方法**: 市場で評価の高いタスク管理ツールを試し、最も使いやすいものを選定する。
- **達成基準**: 日々のタスク管理がよりスムーズになり、時間の有効活用が実感できる。
- **理由**: 既に得意なスケジュール管理をさらに効率化し、時間を最大限に活用する。
#### 目的3: 創造的な問題解決能力の向上
4. **創造的な問題解決が求められるプロジェクトに参加**
- **実行時期**: 26歳から27歳
- **方法**: オンラインプラットフォームやハッカソンで創造的なプロジェクトに参加する。
- **達成基準**: プロジェクトでの貢献が認められ、チームメンバーからの評価を得る。
- **理由**: 独創的な思考を活かし、自己成長を促進する。
#### 目的4: ネットワークの拡大と新しい視点の獲得
5. **技術系のオンラインコミュニティに参加**
- **実行時期**: 24歳から継続
- **方法**: 興味のある技術分野のフォーラムやコミュニティに参加し、定期的に意見交換を行う。
- **達成基準**: 定期的な参加と貢献により、コミュニティ内での存在感を高める。
- **理由**: 新しい視点を得て、論理的思考をさらに深める。
#### 目的5: 長期的なビジョンの設定と自己成長
6. **定期的な自己反省と目標設定**
- **実行時期**: 24歳から継続
- **方法**: 毎月の終わりに自己反省を行い、次の月の目標を設定する。
- **達成基準**: 設定した目標を達成し、自己成長を実感する。
- **理由**: 戦略的思考を活かし、計画的に目標に向かって進む。
### 総括
この戦略は、彼の論理的で戦略的な思考を活かし、技術力の向上、効率的な作業環境の構築、創造的な問題解決能力の向上、ネットワークの拡大、そして長期的なビジョンの設定を目的としています。各行動は彼の性格や得意分野に基づいており、達成基準を明確にすることで、具体的な目標に向かって着実に進むことができます。
グラフの構成
グラフの構成は以下のようになっています。
- ユーザーの分析
- ユーザー分析結果より、推奨されるアクションプラン
- アクションプランが適切かどうかをチェック（もし十分でないなら1からやり直し）
- 最終的なユーザーの人生戦略の提案
となっています。
ユーザーの特徴分析では、ユーザーのMBTIや日々の行動を分析し、どのような行動が性格的な特徴にマッチしているかを理解した上で特徴を導き出してくれるようになっています。
では次からはLangGraphのワークフローを実行するまでの流れを説明していきます！
main.py
# 環境変数をロードするためのdotenvを使用
from dotenv import load_dotenv
# OpenAIのチャットモデルを使用するためのライブラリ
from langchain_openai import ChatOpenAI
# 自作のライフプランナーエージェントをインポート
from lang_graph.life_planner_agent import LifePlannerAgent
# .envファイルから環境変数（APIキーなど）を読み込み
load_dotenv()
def main():
print("Hello LangGraph Practicing Project")
# ユーザープロフィール情報を収集するための質問リスト
# 性格、好み、スキル、キャリア目標などの情報を集める
questions = [
"あなたの性別は？",
"あなたの年齢は？",
"あなたの性格は？",
"あなたのMBTIは？",
"好きなもの・ことは？",
"嫌いなもの・ことは？",
"得意なことは？",
"苦手なことは？",
"現在の仕事は？",
"将来どうなっていたい？",
"どんな人に憧れる？",
"これは長く続けられるなってことは？",
]
# ユーザーの回答を保存するリスト
answers = []
# 質問を順番に表示し、ユーザーからの入力を受け取る
for idx, question in enumerate(questions, start=1):
ans = input(f"Q{idx}. {question} -> ")
# 質問と回答をペアにして保存
answers.append(f"{question} {ans}")
# 回答を改行区切りで一つの文字列にまとめる
user_profile = "\n".join(answers)
# 集めたユーザープロフィールを表示
print("\n=== User Profile ===")
print(user_profile)
# GPT-4oモデルを使用するLLMを初期化（temperature=0で決定論的な出力に）
llm = ChatOpenAI(model="gpt-4o", temperature=0.0)
# 作成したLLMを使ってライフプランナーエージェントを初期化
agent = LifePlannerAgent(llm=llm)
# ユーザープロフィール情報を基にエージェントを実行し、プランを生成
final_output = agent.run(user_input=user_profile)
# 生成されたプランを表示
print("\n=== Generated Plan ===")
print(final_output)
# スクリプトが直接実行された場合にのみmain関数を実行
if __name__ == "__main__":
main()
このアプリケーションで重要なのは、agent = LifePlannerAgent()とagent.run()の処理です。
こいつらがこの人生戦略提案アプリの核を担います。
ぶっちゃけmain.pyは詳しく理解しなくていいのでWorkflowを構築する部分を詳しくみていきましょう！
LangGraphを構築する部分
LangGraphを実装する上で以下のコードはめっちゃ重要なので、一緒に深く理解していきましょう。
まずは、全体のコードをお見せします。
# 型ヒントのために Any 型をインポート
from typing import Any
# OpenAI の ChatGPT モデルを利用するためのクライアント
from langchain_openai import ChatOpenAI
# LangGraph のグラフ構造を定義するためのクラスと終了ノードのシンボル
from langgraph.graph import END, StateGraph
# 自作の各処理コンポーネントをインポート
from lang_chain.information_evaluator import InformationEvaluator
from lang_chain.generate_recommendations import GenerateRecommendations
from lang_chain.analysis_user import AnalysisUser
from lang_chain.finalize_output import FinalizeOutput
# グラフの状態を管理するクラス
from lang_graph.state import ProfileState
class LifePlannerAgent:
def __init__(self, llm: ChatOpenAI):
# 各処理ステップを担当するコンポーネントを初期化
self.analysis_user = AnalysisUser(llm=llm) # ユーザー分析担当
self.generate_recommendations = GenerateRecommendations(llm=llm) # 提案生成担当
self.information_evaluator = InformationEvaluator(llm=llm) # 提案評価担当
self.finalize_output = FinalizeOutput(llm=llm) # 最終出力生成担当
# ワークフローのグラフ構造を作成
self.graph = self._create_graph()
def _create_graph(self) -> StateGraph:
# ProfileState を状態として持つグラフを初期化
workflow = StateGraph(ProfileState)
# 各処理ステップをグラフのノードとして追加
workflow.add_node("analysis_user", self._analysis_user) # ユーザー分析ノード
workflow.add_node("generate_recommendations", self._generate_recommendations) # 提案生成ノード
workflow.add_node("evaluate_recommendations", self._evaluate_recommendations) # 提案評価ノード
workflow.add_node("finalize_output", self._finalize_output) # 最終出力ノード
# グラフの開始ノードを設定
workflow.set_entry_point("analysis_user")
# ノード間の基本的な遷移パスを定義
workflow.add_edge("analysis_user", "generate_recommendations") # 分析→提案生成
workflow.add_edge("generate_recommendations", "evaluate_recommendations") # 提案生成→評価
# 条件付き遷移パスを追加 (フィードバックループを形成)
# 提案が不十分かつ5回未満の繰り返しなら分析に戻り、そうでなければ最終出力へ
workflow.add_conditional_edges(
"evaluate_recommendations",
lambda state: not state.is_sufficient and state.iteration < 5,
{True: "analysis_user", False: "finalize_output"},
)
workflow.add_edge("finalize_output", END) # 最終出力後に処理終了
# グラフを実行可能形式にコンパイル
return workflow.compile()
def _analysis_user(self, state: ProfileState) -> dict[str, Any]:
# ユーザー情報を分析してプロファイルを生成
profile = self.analysis_user.run(state.user_input)
# 状態を更新（プロファイル情報と反復回数の増加）
return {
"profile": profile,
"iteration": state.iteration + 1,
}
def _generate_recommendations(self, state: ProfileState) -> dict[str, Any]:
# プロファイルに基づいて具体的な行動提案リストを生成
recommendations: list[str] = self.generate_recommendations.run(state.profile)
# 生成した提案リストを状態に追加
return {"recommendations": recommendations}
def _evaluate_recommendations(self, state: ProfileState) -> dict[str, Any]:
# 生成された提案内容の十分性を評価
result = self.information_evaluator.run(state.recommendations)
# 評価結果（十分か否かのフラグと理由）を状態に追加
return {
"is_sufficient": result.is_sufficient,
"reason": result.reason,
}
def _finalize_output(self, state: ProfileState) -> dict[str, Any]:
# 提案と元のプロファイルを基に最終的なライフプランを作成
plan: str = self.finalize_output.run(state.recommendations, state.profile)
# 最終プランを状態に追加
return {"final_output": plan}
def run(self, user_input: str) -> str:
# ユーザー入力を元に初期状態を作成
initial_state = ProfileState(user_input=user_input)
# グラフを実行して最終状態を取得
final_state = self.graph.invoke(initial_state)
# 最終プランを返却
return final_state["final_output"]
まず、このコードはLifePlanを提案するLifePlannerAgentを定義しています。
このクラスではLangGraphを用いて人生戦略の提案を行うAgent Workflowの構築とその実行を行う関数を定義しています。
import operator
from typing import Annotated
from pydantic import BaseModel, Field
class ProfileState(BaseModel):
user_input: str = Field(..., description="ユーザーの質問の回答結果")
profile: str = Field(default="", description="ユーザープロファイル情報")
recommendations: Annotated[list[str], operator.add] = Field(
default_factory=list, description="推奨されたアクティビティのリスト"
)
schedule: str = Field(default="", description="生成されたスケジュール")
is_sufficient: bool = Field(default=False, description="情報が十分かどうか")
iteration: int = Field(default=0, description="プロファイル生成と推奨の反復回数")
final_output: str = Field(
default="", description="最終的な成果物。ユーザーに推奨する行動が書かれてある"
)
こちらのコードではLangGraphを実行する上で重要なStateの構成を定義しています。
ワークフローを実行しているときに、常に保持しておくべき情報をStateでは定義します。
今回の場合は、ユーザーの入力やユーザーのプロフィール情報、推奨されるアクション、情報が十分かどうかのステータス、実行回数などを保持する必要があることがわかります。
LangGraphの構築
class LifePlannerAgent:
def __init__(self, llm: ChatOpenAI):
# 各処理ステップを担当するコンポーネントを初期化
self.analysis_user = AnalysisUser(llm=llm) # ユーザー分析担当
self.generate_recommendations = GenerateRecommendations(llm=llm) # 提案生成担当
self.information_evaluator = InformationEvaluator(llm=llm) # 提案評価担当
self.finalize_output = FinalizeOutput(llm=llm) # 最終出力生成担当
# ワークフローのグラフ構造を作成
self.graph = self._create_graph()
def _create_graph(self) -> StateGraph:
# ProfileState を状態として持つグラフを初期化
workflow = StateGraph(ProfileState)
# 各処理ステップをグラフのノードとして追加
workflow.add_node("analysis_user", self._analysis_user) # ユーザー分析ノード
workflow.add_node("generate_recommendations", self._generate_recommendations) # 提案生成ノード
workflow.add_node("evaluate_recommendations", self._evaluate_recommendations) # 提案評価ノード
workflow.add_node("finalize_output", self._finalize_output) # 最終出力ノード
# グラフの開始ノードを設定
workflow.set_entry_point("analysis_user")
# ノード間の基本的な遷移パスを定義
workflow.add_edge("analysis_user", "generate_recommendations") # 分析→提案生成
workflow.add_edge("generate_recommendations", "evaluate_recommendations") # 提案生成→評価
# 条件付き遷移パスを追加 (フィードバックループを形成)
# 提案が不十分かつ5回未満の繰り返しなら分析に戻り、そうでなければ最終出力へ
workflow.add_conditional_edges(
"evaluate_recommendations",
lambda state: not state.is_sufficient and state.iteration < 5,
{True: "analysis_user", False: "finalize_output"},
)
workflow.add_edge("finalize_output", END) # 最終出力後に処理終了
# グラフを実行可能形式にコンパイル
return workflow.compile()
この部分でLangGraphを用いたAgentワークフローの構築を行なっています。
このワークフローは
「①ユーザーの分析を行うノード」
「②推奨される行動を提案するノード」
「③推奨行動の評価を行うノード」
「④最終的な出力を作成するノード」
で構築されます。
LangGraphでは「ノードの作成」→「エッジの追加」→「実行可能な形式にコンパイル」というフローを踏むことで、動作するワークフローの構築が可能になります。
単純なエッジの追加はworkflow.add_edge(NODE_A, NODE_B)を実行することで可能です。
こうすることで、NODE_AからNODE_Bへのエッジを追加することができます。
また、LangGraphにはworkflow.add_conditional_edges()を用いることで、ノード同士に条件分岐を追加することができるようになります。
# 条件付き遷移パスを追加 (フィードバックループを形成)
# 提案が不十分かつ5回未満の繰り返しなら分析に戻り、そうでなければ最終出力へ
workflow.add_conditional_edges(
"evaluate_recommendations",
lambda state: not state.is_sufficient and state.iteration < 5,
{True: "analysis_user", False: "finalize_output"},
)
構文は以下のとおりです。
workflow.add_conditional_edges(
NODE_A,
条件判定を行う関数,
{True: NODE_B, False: NODE_C},
)
add_conditional_edgesを用いることで、条件判定を行う関数の結果がTrueの場合はNODE_Bへ、Falseの場合はNODE_Cへというロジックをワークフローに追加することができるようになります。
これらのノードとエッジを組み合わせることで、最終的に以下のようなワークフローを構築することができます。
確かにevaluate_recommendationsノードにて、is_sufficientがtrueの場合、再度ユーザーの分析を行うようになっていることがわかるかと思います。
それぞれのノードの詳細
def _analysis_user(self, state: ProfileState) -> dict[str, Any]:
# ユーザー情報を分析してプロファイルを生成
profile = self.analysis_user.run(state.user_input)
# 状態を更新（プロファイル情報と反復回数の増加）
return {
"profile": profile,
"iteration": state.iteration + 1,
}
def _generate_recommendations(self, state: ProfileState) -> dict[str, Any]:
# プロファイルに基づいて具体的な行動提案リストを生成
recommendations: list[str] = self.generate_recommendations.run(state.profile)
# 生成した提案リストを状態に追加
return {"recommendations": recommendations}
def _evaluate_recommendations(self, state: ProfileState) -> dict[str, Any]:
# 生成された提案内容の十分性を評価
result = self.information_evaluator.run(state.recommendations)
# 評価結果（十分か否かのフラグと理由）を状態に追加
return {
"is_sufficient": result.is_sufficient,
"reason": result.reason,
}
def _finalize_output(self, state: ProfileState) -> dict[str, Any]:
# 提案と元のプロファイルを基に最終的なライフプランを作成
plan: str = self.finalize_output.run(state.recommendations, state.profile)
# 最終プランを状態に追加
return {"final_output": plan}
続いて、それぞれのノードの詳細がどうなっているかを見ていきます。
ノードの詳細では、「①具体的な処理」「②Stateの値を更新するためのデータを返す」が基本となります。
def _analysis_user(self, state: ProfileState) -> dict[str, Any]:
# ユーザー情報を分析してプロファイルを生成
profile = self.analysis_user.run(state.user_input)
# 状態を更新（プロファイル情報と反復回数の増加）
return {
"profile": profile,
"iteration": state.iteration + 1,
}
例えば、こちらのコードはユーザーの分析を行う（analysis_user）ノードのコードです。
中身は、ユーザーの分析を行うメソッドの実行とstateを更新するのに必要なデータをレスポンスするという内容となっています。
ノードが返す辞書のキーは、更新したいStateオブジェクトのフィールド名に対応している必要があり、そのキーに対応する値でStateの該当フィールドが更新されます。
Stateの全てのフィールドを返す必要はありません。
Stateの値を見てみると、確かにprofileとiterationが存在していますよね。
ノードの処理が実行したタイミングでこのレスポンスがStateに渡されるようになっています。
Stateはこの値を受け取り、情報を更新し次のステップへと移行します。
import operator
from typing import Annotated
from pydantic import BaseModel, Field
class ProfileState(BaseModel):
user_input: str = Field(..., description="ユーザーの質問の回答結果")
profile: str = Field(default="", description="ユーザープロファイル情報")
recommendations: Annotated[list[str], operator.add] = Field(
default_factory=list, description="推奨されたアクティビティのリスト"
)
schedule: str = Field(default="", description="生成されたスケジュール")
is_sufficient: bool = Field(default=False, description="情報が十分かどうか")
iteration: int = Field(default=0, description="プロファイル生成と推奨の反復回数")
final_output: str = Field(
default="", description="最終的な成果物。ユーザーに推奨する行動が書かれてある"
)
LangGraphの実行
def run(self, user_input: str) -> str:
# 初期状態設定＆グラフ走査
initial_state = ProfileState(user_input=user_input)
final_state = self.graph.invoke(initial_state)
return final_state["final_output"]
LangGraphの実行では、Stateのインスタンスを引数に渡すことで実行されています。
最終的なアウトプットはStateのfinal_outputフィールドに保存されているので、そこから値を取得し、レスポンスしています。
以上がLangGraphのワークフローを構築するコードの解説になります。
基本的には以下のステップさえ覚えておけばOKかと思います。
- Stateの定義
- ノードをグラフに追加していく
- ノード同士を繋げるエッジを定義していく
- 最終的なワークフローはコンパイルする
この大まかな流れを理解しておけば、AI駆使すればそれなりのLangGraphを用いたワークフローは作成できるかと思います。
また、それぞれのノードのレスポンスはStateのフィールドと一致している必要があることを覚えておきましょう！
ノードの具体的な処理
続いて、ノードの具体的な処理の内容について解説していきます。
今回作成したライフプランナーエージェントには、以下のノードが存在します。
- ユーザーの分析を行う
analysis_userノード - ユーザーの分析結果から推奨アクションを作成する
generate_recommendationsノード - 推奨アクションの内容が適切かを評価する
evaluate_recommendationsノード - 最終的なアウトプットを作成する
finalize_outputノード
です。
今回はその中の一つであるanalysis_userノードの具体的な処理について説明します。
class AnalysisUser:
def __init__(self, llm: ChatOpenAI):
# ChatOpenAIモデルのインスタンスを受け取り、プロパティとして保存
self.llm = llm
def run(self, user_input: str) -> str:
# ユーザーの回答を受け取り、分析結果を返すメソッド
# プロンプトテンプレートを定義
# ChatPromptTemplate.from_messagesは、会話形式のプロンプトを作成する
prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"あなたはユーザーの質問から、ユーザーがどんな人かを分析する専門家です。",
),
(
"human",
"以下のユーザーが答えた質問から、ユーザーがどのような人物かを分析してください。\n\n"
"ユーザーの回答: {user_input}\n\n"
"ユーザーの回答より、このユーザーがどういった傾向の人物かを分析してください。また、以下のMBTIのリストより、その人物の得意なことなどを提案してください。"
"最低でも、ユーザーの得意なこと不得意なこと、思考の傾向やタイプ、パフォーマンスが高まるアプローチなどを分析してください。"
"\n\nMBTI別特徴：\n\n"
"---\n\n"
"## 🧠 Analysts（分析家タイプ）\n\n"
"---\n\n"
"### **INTJ（建築家）**\n"
"- **どんな人？**：戦略家。未来の可能性を常に見据え、冷静に計画を立てて行動するタイプ。\n"
"- **傾向**：感情よりも論理を重視。独創的で一人で黙々と作業できる環境が好き。\n"
"- **特徴**：長期的なビジョンに忠実。非合理なものが大嫌い。\n"
"- **著名人**：イーロン・マスク、ニーチェ、坂本龍馬\n\n"
"---\n\n"
"### **INTP（論理学者）**\n"
"- **どんな人？**：独自の理論とアイデアを愛する思想家タイプ。\n"
"- **傾向**：好奇心旺盛でマイペース。正解より「なぜ？」を重視する。\n"
"- **特徴**：型にハマらない天才肌。外界より内面に熱中。\n"
"- **著名人**：アルベルト・アインシュタイン、村上春樹、渋沢栄一\n\n"
"---\n\n"
"### **ENTJ（指揮官）**\n"
"- **どんな人？**：行動的なリーダー。最短でゴールに向かう戦略家。\n"
"- **傾向**：計画・実行・成果を重視。誰よりも行動が早い。\n"
"- **特徴**：自他ともに厳しい完璧主義者。弱音は吐かない。\n"
"- **著名人**：スティーブ・ジョブズ、明石家さんま、ナポレオン\n\n"
"---\n\n"
"### **ENTP（討論者）**\n"
"- **どんな人？**：自由奔放な発明家。議論で世界を広げる挑戦者。\n"
"- **傾向**：アイデアが止まらない。刺激と新しさを求めて動き続ける。\n"
"- **特徴**：ルール破りの天才。マンネリが一番の敵。\n"
"- **著名人**：トーマス・エジソン、アドラー、爆笑問題・太田\n\n"
"---\n\n"
"## 🌱 Diplomats（理想主義者タイプ）\n\n"
"---\n\n"
"### **INFJ（提唱者）**\n"
"- **どんな人？**：世の中を良くしたいと本気で思っている理想主義者。\n"
"- **傾向**：少人数の深い人間関係を重視。内に熱い志を持つ。\n"
"- **特徴**：共感力の塊。感情と論理をバランスよく使いこなす。\n"
"- **著名人**：マザー・テレサ、ガンジー、スラムダンクの赤木剛憲\n\n"
"---\n\n"
"### **INFP（仲介者）**\n"
"- **どんな人？**：心優しく、理想と現実のギャップに悩むロマンチスト。\n"
"- **傾向**：マイワールド持ち。共感力が高く、人の痛みに敏感。\n"
"- **特徴**：芸術肌。自分の「好き」を大切に生きる。\n"
"- **著名人**：ジョン・レノン、シェイクスピア、星野源\n\n"
"---\n\n"
"### **ENFJ（主人公）**\n"
"- **どんな人？**：人の力を引き出す天才。カリスマ性があり人気者。\n"
"- **傾向**：人間関係に長けていて、人をまとめるのが得意。\n"
"- **特徴**：「どうやったらこの人が輝けるか」を常に考えている。\n"
"- **著名人**：オプラ・ウィンフリー、バラク・オバマ、松岡修造\n\n"
"---\n\n"
"### **ENFP（運動家）**\n"
"- **どんな人？**：感情豊かで、好奇心が爆発している自由人。\n"
"- **傾向**：ノリと勢いで新しいことに飛び込む。人の話をよく聞く。\n"
"- **特徴**：アイデア豊富、だけど飽き性。多才なエネルギー体。\n"
"- **著名人**：ロビン・ウィリアムズ、ディズニー、ひろゆき\n\n"
"---\n\n"
"## 🛡️ Sentinels（現実主義者タイプ）\n\n"
"---\n\n"
"### **ISTJ（管理者）**\n"
"- **どんな人？**：真面目で堅実、責任感の塊。\n"
"- **傾向**：ルールと信頼が大事。計画的に物事を進める。\n"
"- **特徴**：職人肌。淡々と、でも着実に成果を出すタイプ。\n"
"- **著名人**：ジョージ・ワシントン、ナウシカのクロトワ、乃木希典\n\n"
"---\n\n"
"### **ISFJ（擁護者）**\n"
"- **どんな人？**：思いやりと献身に満ちた癒し系。\n"
"- **傾向**：縁の下の力持ち。裏で支えることに喜びを感じる。\n"
"- **特徴**：控えめだけど芯は強い。記憶力も良い。\n"
"- **著名人**：ビヨンセ、エリザベス女王、田中みな実\n\n"
"---\n\n"
"### **ESTJ（幹部）**\n"
"- **どんな人？**：組織やルールを大切にする実務家。\n"
"- **傾向**：リーダー気質で、責任感が強く頼られる存在。\n"
"- **特徴**：効率最優先。感情よりも実績を重視する。\n"
"- **著名人**：ミシェル・オバマ、エイブラハム・リンカーン\n\n"
"---\n\n"
"### **ESFJ（領事）**\n"
"- **どんな人？**：社交性と面倒見の良さで人を引きつけるムードメーカー。\n"
"- **傾向**：周囲の期待に応えようとする努力家。\n"
"- **特徴**：親しみやすく、礼儀正しく、伝統を大切にする。\n"
"- **著名人**：テイラー・スウィフト、ホイットニー・ヒューストン\n\n"
"---\n\n"
"## 🎨 Explorers（冒険家タイプ）\n\n"
"---\n\n"
"### **ISTP（巨匠）**\n"
"- **どんな人？**：クールで現場主義、実際にやってみないと気が済まない。\n"
"- **傾向**：無口だけど器用。突発的な問題に強い。\n"
"- **特徴**：マニュアルより触って覚えるタイプ。\n"
"- **著名人**：クリント・イーストウッド、リヴァイ兵長\n\n"
"---\n\n"
"### **ISFP（冒険家）**\n"
"- **どんな人？**：感性と自由を愛するアーティスト。\n"
"- **傾向**：美しいものが好き。マイペースで心優しい。\n"
"- **特徴**：目立たないけど、芯がある。行動で語るタイプ。\n"
"- **著名人**：マイケル・ジャクソン、オードリー・ヘプバーン\n\n"
"---\n\n"
"### **ESTP（起業家）**\n"
"- **どんな人？**：即行動！今この瞬間を楽しむエンターテイナー。\n"
"- **傾向**：リスクを恐れず突き進む。説得力がある。\n"
"- **特徴**：「まずやってみる」精神。カリスマ性高め。\n"
"- **著名人**：アーネスト・ヘミングウェイ、ジョン・F・ケネディ\n\n"
"---\n\n"
"### **ESFP（エンターテイナー）**\n"
"- **どんな人？**：その場の雰囲気を一気に明るくする人間太陽☀️\n"
"- **傾向**：自分も周りも楽しませたい。おしゃべり大好き。\n"
"- **特徴**：人気者だけど繊細。感情に正直。\n"
"- **著名人**：マリリン・モンロー、フレディ・マーキュリー",
),
]
)
# LangChainのチェーン処理を作成
# prompt | self.llm：プロンプトをLLMに送信するパイプライン
chain = prompt | self.llm
# チェーンを実行し、結果を返す
# invoke関数でプロンプト内の{user_input}を実際の値で置換して実行
return chain.invoke({"user_input": user_input})
解説とはいってもやっていることはLCEL（LangChain Expression Language）を用いたChainの実行です。
- システムプロンプトの定義（MBTIの定義や特徴など）
- LCELを用いたchainの作成およびその実行
非常にシンプルであることがわかります。
このChainの内容を実行することで、ユーザーの質問の回答から、MBTIごとの特徴を理解し、このユーザーがどのような特徴を持つ人なのかのプロフィールを作成します。
def _analysis_user(self, state: ProfileState) -> dict[str, Any]:
# ユーザープロファイルを解析
profile = self.analysis_user.run(state.user_input)
return {
"profile": profile,
"iteration": state.iteration + 1,
}
analysis_userノードでは、このChainの実行結果を受け取り、それをStateのprofileというフィールドに保存していましたよね。
LangGraphではノードごとの処理はLangChainや計算などのさまざまな処理を記述します。
これらのノードを組み合わせて最終的なワークフローを作成、実行することでAIエージェントを用いたワークフローの実行が可能になるのです。
そのほかのノードの具体的な処理が気になる方は、以下のリンクより調べてみてください！
まとめ
最後までお読みいただきありがとうございました！
この記事を読んでLangGraphに対する理解を深めてもらい、ちょっとでもLangGraphを勉強しようと思ってくださったら幸いです！
『LangChain と LangGraph による RAG・AI エージェント［実践］入門』という書籍はめちゃくちゃためになる内容なので、「ちょっとやってみようかな」という方にはおすすめです！
これは余談なのですが、会社のAIエンジニアの方も「LangChainを学ぶよりもLangGraphを学んだほうが良い」と言っていたくらいには今後はLangGraphは伸びてくるかと思いますので、ぜひ学ぶことをお勧めします！
それではみなさま、ぜひ楽しいAIエージェントライフをお過ごしくださいませ〜
Discussion