LangChainのLangGraphは、サイクリックな処理が可能なためループ状態になる可能性があります。また、処理内容によりトークン数が増大（コストが増大）する可能性があります。実行の際はご注意下さい。この記事の内容により事故や損失等が発生しても責任は負いかねますことを予めご了承下さい。コードを実行される場合は、langchain blogやgithub上のexampleにも是非当たって頂ければと思います。
はじめに
先日、LangGraphを利用したAIエージェント操作について、次の記事を書かせて頂きました。ご一読頂けると幸いです。本日は、その第2弾となります。
次の図は、langchain blogに載っているnodeとedgeの表現が含まれた図です。前回は左の図のイメージでAIエージェントを動かしました。今回は右の図をイメージしAIエージェントを動かしてみたいと思います。
引用元：https://blog.langchain.dev/langgraph-multi-agent-workflows/
LangGraphとは
LangChainドキュメントには、次のように書かれています。
「LangGraphは、LLMでステートフルなマルチアクターアプリケーションを構築するためのライブラリで、LangChainでの使用を意図しています。 LangChain式言語（LCEL）を拡張し、複数のチェーン(またはアクター)を複数の計算ステップで循環的に調整する機能を備えています。 」
つまり、LangGraphは、LangChainフレームワーク内でLCELを使い構築できるAgent Orchestratorです。（"つまり"以下は、筆者の勝手な表現です。）
LangGraphの特徴とは
LangGraphは、グラフ理論（グラフ理論についてはここでは割愛させて頂きます）に導かれており、Nodeと呼ばれるエージェントと、Edgeと呼ばれる線で繋がれた構造をとることが特徴となっています。
LangGraphでAIエージェントを動かしてみる
先ほどの図の内、右の図についてgithub上のexample（※1）を頼りに、langgraphのAIエージェントを動かしてみたいと思います。exampleでは、AIエージェント向け検索エンジン（Tavily）やLangSmith（LLM アプリケーションの開発、監視、およびテストツール）が使われていますが、ここではそれらは使わず、アウトバウンドしないシンプルな形で、AIエージェントに役割（プロンプトによる役割）を与え動かしてみます。
▼全体のイメージ
ある会社における今年の販売戦略をAIエージェントとツールを使い、立ててみたいと思います。
・過去5年間のA、B、C商品の売上を分析し、セールススタッフ、セールスマネージャーが、意見の述べあい、今年の販売戦略を立てる。
・商品分析用として、データ分析ツールを用意する。
・過去5年間の商品データは、CSVファイルを用意する。
＜イメージ図＞
▼用意するエージェント
・ルーター：router
・エージェント1 : sales_staff
・エージェント2 : sales_manager
▼用意するツール
・データ分析ツール
（ここでは、簡単に基本統計と相関関係だけが分析できるもの）
▼用意するデータ
・次のデータを入力した"sales_data.csv"ファイルを、実行ファイルと同じフォルダに配置。
| Year | Product_A | Product_B | Product_C |
|---|---|---|---|
| 2019 | 1091 | 540 | 802 |
| 2020 | 979 | 527 | 782 |
| 2021 | 970 | 521 | 816 |
| 2022 | 1070 | 559 | 752 |
| 2023 | 1059 | 533 | 730 |
▼会話のテーマ
「当社の商品A、B、Cの過去5年分のデータを調べ、今年の販売戦略を立てる。」
環境
-
Windows10
-
Python v3.11.4
-
主なlangchainライブラリバージョン
・langchain-core==0.1.27
・langgraph==0.0.26
・langchain==0.1.9 -
APIキー等の環境変数は、試したコードと同じフォルダに".env"ファイルを作り、その中に記述しています。
-
requirements.txt は、最後の方に載せています。
AZURE_OPENAI_TYPE = "azure"
AZURE_OPENAI_KEY = "YOUR AZURE OPENAI KEY"
azure_endpoint = "YOUR AZURE ENDPOINT URL"
AZURE_OPENAI_DEPLOYMENT_NAME = "YOUR AZURE DEPLOYMENT NAME"
AZURE_OPENAI_VERSION = "2024-02-15-preview"
※ AZURE_OPENAI_DEPLOYMENT_NAMEは、チャット用モデルのデプロイ名（筆者はgpt-4を利用）
LangGraphを使ったコード
＜主な処理の流れ＞
- 環境設定
- エージェント設定
- ツール設定
- グラフ設定
- 状態の定義
- エージェントノードの定義
- ツールノードの定義
- エッジロジックの定義
- グラフの定義
- 実行
# main_multi.py
# 1.環境設定
import os
from dotenv import load_dotenv
load_dotenv("./.env")
# 2.エージェント設定
import json
from langchain_core.messages import (
AIMessage,
BaseMessage,
ChatMessage,
FunctionMessage,
HumanMessage,
)
from langchain_core.utils.function_calling import convert_to_openai_function
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langgraph.graph import END, StateGraph
from langgraph.prebuilt.tool_executor import ToolExecutor, ToolInvocation
from langchain_openai import AzureChatOpenAI
llm = AzureChatOpenAI(
api_version=os.getenv("AZURE_OPENAI_VERSION"),
azure_endpoint=os.getenv("azure_endpoint"),
api_key=os.getenv("AZURE_OPENAI_KEY"),
azure_deployment=os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME"),
temperature=0.7,
streaming=True,
)
def create_agent(llm, tools, system_message: str):
functions = [convert_to_openai_function(t) for t in tools]
prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"You are a helpful AI assistant, collaborating with other assistants."
" Use the provided tools to progress towards answering the question."
" If you are unable to fully answer, that's OK, another assistant with different tools "
" will help where you left off. Execute what you can to make progress."
" If you or any of the other assistants have the final answer or deliverable,"
" prefix your response with FINAL ANSWER so the team knows to stop."
" You have access to the following tools: {tool_names}.\n{system_message}",
),
MessagesPlaceholder(variable_name="messages"),
]
)
prompt = prompt.partial(system_message=system_message)
prompt = prompt.partial(tool_names=", ".join([tool.name for tool in tools]))
return prompt | llm.bind_functions(functions)
# 3.ツール設定
from langchain_core.tools import tool
from typing import Annotated
from langchain_experimental.utilities import PythonREPL
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
repl = PythonREPL()
@tool
def python_repl(
code: Annotated[str, "The python code to execute to generate your chart."]
):
"""Use this to execute python code. If you want to see the output of a value,
you should print it out with `print(...)`. This is visible to the user."""
try:
result = repl.run(code)
except BaseException as e:
return f"Failed to execute. Error: {repr(e)}"
return f"Succesfully executed:\n```python\n{code}\n```\nStdout: {result}"
@tool
def data_analysis_tool():
"""This tool enhances analysis by providing detailed statistics and visualizations."""
try:
df = pd.read_csv("./sales_data.csv")
# 基本統計の表示
analysis_result = df.describe()
basic_stats_response = f"Basic statistics:\n{analysis_result.to_string()}"
# 相関関係の分析
correlation = df.corr()
correlation_response = f"\nCorrelation matrix:\n{correlation.to_string()}"
response = f"{basic_stats_response}\n{correlation_response}"
except Exception as e:
response = f"Failed to analyze data. Error: {str(e)}"
return response
# 4.グラフ設定
# ### 状態の定義
import operator
from typing import Annotated, List, Sequence, Tuple, TypedDict, Union
from langchain.agents import create_openai_functions_agent
from langchain_core.utils.function_calling import convert_to_openai_function
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI
from typing_extensions import TypedDict
# グラフの各ノード間で渡されるオブジェクトを定義する。エージェントとツールごとに異なるノードを作成する。
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add]
sender: str
# ### エージェントノードの定義
import functools
# 与えられたエージェントのノードを作成するヘルパー関数
def agent_node(state, agent, name):
print(f"Executing {name} node!")
result = agent.invoke(state)
# エージェントの出力を、グローバル・ステートに追加するのに適したフォーマットに変換
if isinstance(result, FunctionMessage):
pass
else:
result = HumanMessage(**result.dict(exclude={"type", "name"}), name=name)
return {
"messages": [result],
# 厳密なワークフローがあるので、senderを追跡することが可能。
# senderを追跡することで、次に誰に渡すべきかを知ることが可能。
"sender": name,
}
# sales_staff
sales_staff = create_agent(
llm,
[python_repl],
system_message="顧客対応と製品、サービス提案を担当。顧客からの質問に答え、適切な製品、サービスを推薦し、商談データ、売上予定データをシステムに記録します。",
)
sales_staff_node = functools.partial(agent_node, agent=sales_staff, name="sales_staff")
# sales_manager
sales_manager = create_agent(
llm,
[python_repl],
system_message="チームの管理と指導を担当。販売目標の設定、販売戦略の策定、パフォーマンスの監視、そしてチームメンバーへのフィードバック提供を行います。",
)
sales_manager_node = functools.partial(agent_node, agent=sales_manager, name="sales_manager")
# ### ツールノードの定義
tools = [python_repl, data_analysis_tool]
tool_executor = ToolExecutor(tools)
def tool_node(state):
# print(f"Executing tool_node! state is {state}!")
"""This runs tools in the graph
It takes in an agent action and calls that tool and returns the result."""
messages = state["messages"]
last_message = messages[-1]
# function_callからToolInvocationを作成
tool_input = json.loads(
last_message.additional_kwargs["function_call"]["arguments"]
)
# 単一引数渡す
if len(tool_input) == 1 and "__arg1" in tool_input:
tool_input = next(iter(tool_input.values()))
tool_name = last_message.additional_kwargs["function_call"]["name"]
action = ToolInvocation(
tool=tool_name,
tool_input=tool_input,
)
# tool_executorを呼び出し、レスポンスを返す。
response = tool_executor.invoke(action)
# レスポンスを使い、FunctionMessageを作成する。
function_message = FunctionMessage(
content=f"{tool_name} response: {str(response)}", name=action.tool
)
# 既存のリストに追加
return {"messages": [function_message]}
# ### エッジロジックの定義
def router(state):
messages = state["messages"]
last_message = messages[-1]
if "function_call" in last_message.additional_kwargs:
return "call_tool"
if "FINAL ANSWER" in last_message.content:
return "end"
return "continue"
# ### グラフの定義
workflow = StateGraph(AgentState)
workflow.add_node("sales_staff", sales_staff_node)
workflow.add_node("sales_manager", sales_manager_node)
workflow.add_node("call_tool", tool_node)
workflow.add_conditional_edges(
"sales_staff",
router,
{"continue": "sales_manager", "call_tool": "call_tool", "end": END},
)
workflow.add_conditional_edges(
"sales_manager",
router,
{"continue": "sales_staff", "call_tool": "call_tool", "end": END},
)
workflow.add_conditional_edges(
"call_tool",
# 各エージェントノードは'sender'フィールドを更新する。
# ツール呼び出しノードは更新しない。
# つまり このエッジは、ツールを呼び出した元のエージェントにルーティングされる。
lambda x: x["sender"],
{
"sales_staff": "sales_staff",
"sales_manager": "sales_manager",
},
)
workflow.set_entry_point("sales_staff")
graph = workflow.compile()
# 5.実行（Invoke）
for s in graph.stream(
{
"messages": [
HumanMessage(
content="このコードに事前に用意されたagentとtoolを利用します。"
"会話のテーマは、「当社の商品A、B、Cの過去5年分のデータを調べ、今期の販売戦略を立てる。」です。"
"会話は、sales_staffから始めます。"
"データ分析ツールは、必ず'./sales_data.csv' ファイルを利用して下さい。"
"データ分析ツールは、'./sales_data.csv' ファイルからデータを読み込み、基本統計と相関関係を分析します。"
"データ分析ツールは、分析結果をテキストとして出力し、分析結果に基づく洞察も提供します。"
"次に、データ分析ツールが出した分析結果と洞察は、sales_staffに渡されます。"
"次に、sales_staffはとsales_managerは、データ分析ツールが出した分析結果と洞察に基づき、sales_staffと意見を交わし、今期の販売戦略を立てます。"
"sales_staffとsales_managerの会話合計数は、最大20回までです。"
"最後に、sales_managerが、全て会話のを踏まえた上で、重要なポイントを箇条書きでまとめて終了します。"
)
],
},
# グラフ内の最大ステップ数
{"recursion_limit": 25},
):
for key in ["sales_staff", "sales_manager"]:
if key in s:
messages = s[key]["messages"]
for msg in messages:
# エージェントからのメッセージ内容を出力
print(msg.content)
print("----\n") # セクションの終わり
実行結果
Executing sales_staff node!
まずは、'./sales_data.csv' ファイルからデータを読み込み、基本統計と相関関係を分析することから始めましょう。Pythonを使用して分析を行います。データファイ
ルの内容については、具体的な構造や列名をまだ知らないため、初めにデータを読み込みその概要を確認します。
ここで行うステップは次の通りです：
1. データファイルを読み込む。
2. データの概要（最初の数行、列名、データ型）を確認。
3. 基本統計量（平均、中央値、標準偏差など）を計算。
4. 商品A、B、C間の相関関係を分析。
このステップを実行していきます。
----
Python REPL can execute arbitrary code. Use with caution.
Executing sales_staff node!
データの読み込みに成功しました。データ概要は以下の通りです：
- 列名: 'Year', 'Product_A', 'Product_B', 'Product_C'
- データ型: すべての列が整数型（int64）です。
- 最初の5行の概要:
| Year | Product_A | Product_B | Product_C |
|------|-----------|-----------|-----------|
| 2019 | 1091 | 540 | 802 |
| 2020 | 979 | 527 | 782 |
| 2021 | 970 | 521 | 816 |
| 2022 | 1070 | 559 | 752 |
| 2023 | 1059 | 533 | 730 |
次に、基本統計量（平均、中央値、標準偏差など）を計算し、商品A、B、C間の相関関係を分析します。
----
Executing sales_staff node!
基本統計量と商品間の相関関係の分析結果は以下の通りです：
### 基本統計量
- **平均**：
- Product_A: 1033.8
- Product_B: 536.0
- Product_C: 776.4
- **標準偏差**：
- Product_A: 55.43
- Product_B: 14.66
- Product_C: 35.37
- **最小値**：
- Product_A: 970
- Product_B: 521
- Product_C: 730
- **最大値**：
- Product_A: 1091
- Product_B: 559
- Product_C: 816
### 商品間の相関関係
- Product_AとProduct_B間の相関係数は0.749、つまり正の強い相関があり、一方の販売数が増えるともう一方も増える傾向にあります。
- Product_AとProduct_C間の相関係数は-0.436、つまり負の相関があり、一方の販売数が増えるともう一方は減少する傾向にあります。
- Product_BとProduct_C間の相関係数は-0.464、同様に負の相関があり、一方の販売数が増えるともう一方は減少する傾向にあります。
これらの結果から、商品AとBは互いに売り上げが連動している可能性が高く、商品Cの売り上げはAやBの売り上げとは逆の動きをする傾向があります。この分析を基に、
販売戦略を考える際には、商品AとBを連動させて推進する戦略が良さそうで、商品Cについては独立した戦略を考える必要がありそうです。
----
Executing sales_manager node!
これらの分析結果を踏まえて、商品AとBの販売促進策について話し合いましょう。これらの商品に関して、互いに連動して売り上げを伸ばす戦略を考えることが重要で
す。また、商品Cについては、独立した販売戦略を立てるべきです。商品AとBの促進策について何か具体的なアイデアがありますか？また、商品Cの販売促進については
どのように考えていますか？
----
Executing sales_staff node!
商品AとBに関しては、互いに正の強い相関があるため、以下のような販売促進策を提案します。
### 商品AとBの販売促進策
1. **セット販売の実施**：商品AとBをセットで割引価格にすることで、顧客に両商品を同時に購入するインセンティブを与えます。これにより、一方の商品の購入がも
う一方の商品の購入を促進することが期待できます。
2. **共同プロモーション**：商品AとBの共同プロモーションを実施し、両商品の連動性を強調するキャンペーンを展開します。例えば、特定のイベントや季節に合わせ
たプロモーションを行い、両商品の同時購入を促します。
3. **クロスマーケティング**：商品Aを購入した顧客に対して、商品Bの割引クーポンを提供するなど、クロスマーケティングを実施します。これにより、既存の顧客を
活用して他の商品の販売を促進することができます。
### 商品Cの販売促進策
商品Cについては、AやBと異なる販売戦略を立てる必要があります。商品Cの独自性やターゲット市場を考慮した上で、以下のような戦略を提案します。
1. **ターゲット市場の特定と集中**：商品Cの主な購買層を特定し、そのニーズに合わせたマーケティング活動を展開します。例えば、年齢層、興味・関心、ライフス
タイルなどに基づいてターゲット市場を絞り込みます。
2. **独自の価値提案**：商品Cが提供する独自の価値や利点を明確に打ち出し、競合他社の商品との差別化を図ります。これにより、商品C独自の魅力を前面に出し、顧
客の購買意欲を喚起します。
3. **オンラインマーケティングの強化**：SNSやオンライン広告を活用し、商品Cの認知度向上と購買促進を図ります。特に、ターゲットとする顧客層がオンライン上で
活動している場合、効果的なオンラインマーケティング戦略が重要となります。
以上のような販売促進策を通じて、商品AとBの相乗効果を最大化し、商品Cの市場でのポジショニングを強化することが期待できます。
これらの販売促進策に加えて、商品AとBについては、顧客の声やフィードバックを積極的に収集し、製品の改善や新しいプロモーションのアイデアに反映させることも
重要です。また、商品Cの場合は、特定のニッチ市場に注力し、その市場におけるリーダーとなることを目指す戦略も有効かもしれません。
商品AとBの促進策に関しては、セット販売や共同プロモーションの内容をさらに具体化する必要があります。例えば、どのようなイベントや季節に合わせたプロモーシ
ョンが考えられるか、また、クロスマーケティングの具体的な方法についても詳細を検討する必要があります。
商品Cに関しては、オンラインマーケティングの強化に加えて、インフルエンサーや意見リーダーとのコラボレーションを検討することも一つの手段です。これにより、
商品Cの認知度と魅力をより効果的に伝えることができるかもしれません。
これらの戦略を実行するためには、市場調査や顧客アンケートなどを通じて、ターゲット顧客のニーズや期待をより深く理解することが必要です。また、競合分析も重
要で、どのようにして他の商品と差別化を図るか、独自の価値を提供するかについても検討が必要です。
これらの戦略に基づき、具体的なアクションプランを立てることが、次のステップとなります。
----
Executing sales_manager node!
これらの提案は非常に有用で、各商品の販売促進における方向性を明確に示しています。以下に、これまでの議論を要約し、結論へと移ります。
### 重要なポイント
1. **商品AとBの販売促進戦略**：
- セット販売の実施：商品AとBをセットで割引価格に設定し、同時購入を促進。
- 共同プロモーション：特定のイベントや季節に合わせた共同プロモーションを展開し、両商品の相乗効果を狙う。
- クロスマーケティング：商品A購入者への商品B割引クーポン提供など、互いの顧客基盤を活用した販売促進。
2. **商品Cの販売促進戦略**：
- ターゲット市場の特定と集中：商品Cの主な購買層を明確にし、そのニーズに合わせたマーケティングを実施。
- 独自の価値提案：商品Cの独自性を強調し、競合との差別化を図る。
- オンラインマーケティングの強化：SNSやオンライン広告を活用した商品Cの認知度向上と購買促進。
### 結論
商品AとBに関しては、互いに連動する販売促進戦略を採用することで、相乗効果を最大限に引き出すことが重要です。一方、商品Cについては、その独自性を生かした販
売戦略を立てることが求められます。これらの戦略を実行するためには、市場調査や顧客アンケートを通じて、ターゲット顧客のニーズを深く理解し、競合分析を行う
ことが不可欠です。具体的なアクションプランの策定と実行に向けて、これらのポイントを踏まえて計画を進めていくことが今後の課題となります。
FINAL ANSWER
----
requirements.txt
aiohttp==3.9.3
aiosignal==1.3.1
annotated-types==0.6.0
anyio==4.3.0
asttokens==2.4.1
attrs==23.2.0
black==24.2.0
certifi==2024.2.2
charset-normalizer==3.3.2
click==8.1.7
colorama==0.4.6
comm==0.2.1
contourpy==1.2.0
cycler==0.12.1
dataclasses-json==0.6.4
debugpy==1.8.1
decorator==5.1.1
distro==1.9.0
executing==2.0.1
fonttools==4.49.0
frozenlist==1.4.1
greenlet==3.0.3
h11==0.14.0
httpcore==1.0.4
httpx==0.27.0
idna==3.6
ipykernel==6.29.3
ipython==8.22.1
jedi==0.19.1
jsonpatch==1.33
jsonpointer==2.4
jupyter_client==8.6.0
jupyter_core==5.7.1
kiwisolver==1.4.5
langchain==0.1.9
langchain-community==0.0.24
langchain-core==0.1.27
langchain-experimental==0.0.52
langchain-openai==0.0.8
langgraph==0.0.26
langsmith==0.1.10
marshmallow==3.21.0
matplotlib==3.8.3
matplotlib-inline==0.1.6
multidict==6.0.5
mypy-extensions==1.0.0
nest-asyncio==1.6.0
numpy==1.26.4
openai==1.13.3
orjson==3.9.15
packaging==23.2
pandas==2.2.1
parso==0.8.3
pathspec==0.12.1
pillow==10.2.0
platformdirs==4.2.0
prompt-toolkit==3.0.43
psutil==5.9.8
pure-eval==0.2.2
pydantic==2.6.3
pydantic_core==2.16.3
Pygments==2.17.2
pyparsing==3.1.1
python-dateutil==2.8.2
python-dotenv==1.0.1
pytz==2024.1
pywin32==306
PyYAML==6.0.1
pyzmq==25.1.2
regex==2023.12.25
requests==2.31.0
seaborn==0.13.2
six==1.16.0
sniffio==1.3.1
SQLAlchemy==2.0.27
stack-data==0.6.3
tenacity==8.2.3
tiktoken==0.6.0
tornado==6.4
tqdm==4.66.2
traitlets==5.14.1
typing-inspect==0.9.0
typing_extensions==4.10.0
tzdata==2024.1
urllib3==2.2.1
wcwidth==0.2.13
yarl==1.9.4
さいごに
LangGraphによるAIエージェントとツールを使った会話でした。routerを中心に配置し、左右に置いたAIエージェント（sales_staff、sales_manager）と、下に配置したツールを使い、自律的に意思決定を行うプロセスを確認することができました。なお、私の理解不足もあると思いますので、実際動作させる場合は、前述のオリジナルGithubコードも必ず参照下さい。）
前回と今回を通じてLangGraphを使ったAIエージェントパワーを感じることができました。アイデア次第で、多くの業務やタスクの効率化および品質向上が期待できそうです。