LangGraphを使った３つのAgentによるマルチエージェントのサンプルを記述します。
以前投稿した　LangGraph簡単なサンプル3つ　の続き（４つ目）のサンプルでもあります。今回はツールを使うAgentを入れ、簡単なTipsを入れています

設計方針

Agent_01 : queryとして都市名を入れると、観光スポットを２つ、理由と伴に提案する
- １つ目は、天候が晴れていた場合の観光スポット
- ２つ目は、天候が雨天の場合の観光スポット
- 回答はspot名と理由を辞書型として２つをリスト形式で回答させる
  - [{"spot":"観光スポット名1","reason":"晴れた場合..."},{"spot":"観光スポット名2","reaseon":"雨天の場合..."}
Agent_02 : ツールを使ったAgentを中間にいれる。RAGやTAVILYを使ったネット検索のサンプルはよく見かけるので、趣向を変えてOpenWeatherMapを用いて、queryで入れた場所の天候を取得する
Agent_03 :Agent01での提案と、Agent_02での天候情報を踏まえて、最終的なおすすめの提案をしてもらう

Tips

yaspinを入れてAgentの思考中を可視化させる
streamによる出力を、enumerate()でくくることで、途中経過の進捗率を表示させる

環境

langchain                                0.2.16
langchain-community                      0.2.17
langchain-core                           0.2.41
langchain-openai                         0.1.25
langgraph                                0.2.26
langgraph-checkpoint                     1.0.12
langsmith                                0.1.147
openai                                   1.59.3
pydantic                                 2.10.4
pydantic_core                            2.27.2

.envの準備

.env

# OPENAI
OPENAI_API_KEY = xxxxxxxxxxxx

# OpenWeatherMap (取得してください）
OPENWEATHERMAP_API_KEY = xxxxxxxxxx

# LANGSMITH(なくても問題ないですが、あるとLangGraphの構築には便利です)
LANGCHAIN_TRACING_V2=true
LANGCHAIN_ENDPOINT="https://api.smith.langchain.com"
LANGCHAIN_API_KEY="lsv2_pt_xxxxxxxx"
LANGCHAIN_PROJECT="sample_progect_name"

# TAVILY(今回はつかいません）
TAVILY_API_KEY = xxxxxxxxxxx

python

import os
from os.path import join, dirname, abspath
from dotenv import load_dotenv

dir_path = dirname(abspath("__file__"))
dotenv_path = join(dir_path, '../your_dir_name/.env')
load_dotenv(dotenv_path, verbose=True)

#print(os.environ['OPENAI_API_KEY'])
#print(os.environ["OPENWEATHERMAP_API_KEY"]) #Agent02で使います
#print(os.environ["LANGCHAIN_API_KEY"]) #任意です
#print(os.environ['TAVILY_API_KEY']) #今回は使いません

Library

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from typing_extensions import TypedDict
from typing import List
from langgraph.graph import StateGraph, START, END

llm = ChatOpenAI(temperature=0, model_name="gpt-4o-mini")

State

State作成の正攻法がわかってないですが、私の場合、最初に全体が見えてStateを作成するというより、Agentを１つずつ作りながら、取得する変数を追記していく感じでつくっています。
例えば、最初のAgentは、queryを入れて、その答えをanswerで返す。ということだけ決めている場合は、初期のclsass State(TypedDict)には、queryとanswerしか入れずにコードを書きはじめています。
Agent_01の関数ができたら、動作確認します。初期値として initial_state(query="東京")をつくり、Agent_01(initial_state)に入れて動作確認します。うまく動作したら、Agent_02を作るときに、また入力と出力の変数を考えて class State(TypedDict):に追記していきます。最終的にできたのが以下です。

class State(TypedDict):
    """
    グラフの状態を表す(Dict型)
    
    Attributes:
      query:str 質問(ロケーション)
      answer:List[str] 観光スポットの提案２つ
      owm_result:天候情報（OpenWeatherMapで取得)
      today_str:今日の日付 例:2025年01月05日
      final_suggestion:提案する観光スポット
    """
    query:str
    answer:List[str]
    owm_result:str
    today_str:str
    final_suggestion:str

yaspin

いれるとAgentの途中経過を可視化できるので入れています。なくても動作します

%pip install yaspin
#Successfully installed termcolor-2.3.0 yaspin-3.1.0

こちらで教えてもらいました

python

from yaspin import yaspin

Agent_01 (agent_suggest)

観光スポットを提案するAgentです。晴れた場合と、雨天の場合の２箇所を提案してもらいます
回答を文字列StrOutputParser(）で返してもらってもよいですが、今回はPydanticOutputParserを使って、辞書型を入れたリスト形式で回答してもらうようにしています。この辺は好みの問題ですが、複数案を出させる場合は、このやり方のほうがよいかと思っています

from langchain_core.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field, model_validator

# 辞書型(リストのinnner)
class Suggest(BaseModel):
    sport: str = Field(description="観光スポット")
    reason: str = Field(description="観光スポットを提案した理由")

# リスト(辞書型のOuter)
class dict_list(BaseModel):  #複数の観光名所を持つクラス
    sightseeing: List[Suggest]      #観光名所のリスト

# Set up a OutputParser
# parserだけだと何のParserかわからなくなるので、
# StrOutputParser()の場合、str_parserなど、わかりやすい名前をつけてます
suggest_parser = PydanticOutputParser(pydantic_object=dict_list)

Agent_01 agent_suggest

from langchain_core.prompts import PromptTemplate
from langchain_core.prompts import SystemMessagePromptTemplate
from langchain_core.prompts import HumanMessagePromptTemplate

def agent_suggest(state:State):#観光地を提案する関数（後でノードとして定義し利用）
    query=state['query']
    # template
    system_prompt = PromptTemplate(
        template="""あなたは優秀な観光アドバイザーです。
                {query}における、観光スポットを２つ提案してください。1つ目は天候が良い場合、2つ目は雨天の場合です。
                spot and reason ：{format_instructions}""",
        input_variables=["query"],
        partial_variables={"format_instructions": suggest_parser.get_format_instructions()},
    )
    
    system_message_prompt = SystemMessagePromptTemplate(prompt=system_prompt)
    human_message_prompt = HumanMessagePromptTemplate.from_template('観光の場所：\n{query}')
    prompt1 = ChatPromptTemplate.from_messages([system_message_prompt, human_message_prompt])
    
    #agentの作成
    agent_suggest = (
        prompt1
        | llm
        | suggest_parser
    )
    with yaspin(text="Agent_suggest Processing") as spinner:
        answer = agent_suggest.invoke(query)
        spinner.ok("✅ ")
    # answerを文字列で取得する場合はここでreturn
    #return {"query":query,"answer":answer}
    
    # answerを、辞書形式のリストのまま取り出したい場合
    # pydantic v1
      # pydantic v1はBaseModelを使ったクラスを作成した際、
      # dict()で辞書変換を行うことが出来てましたのでこうします
    #spot_list=answer.dict()['sightseeing']
    #
    # pydantic v2
    # v2の場合、辞書変換時に使用するのはdict()ではなくmodel_dump()になるのでこうします
    spot_list=answer.model_dump()['sightseeing']
    return {"query":query,"answer":spot_list}

pydantic_v1とv2で記述の仕方が異なるので注意が必要です
ここまでで動作確認したい場合はこんな感じで試験します

動作確認

# 初期値
initial_state = State(query='東京')
# 動作試験
agent_suggest(initial_state)

最初のAgentが考えている最中は、 Agent_suggest Processingの前のアイコンがくるくる回転し、処理が完了すると✅️チェックマークがついたあとに結果が表示されます

実行結果

✅  Agent_suggest Processing
{'query': '東京',
 'answer': [{'sport': '上野恩賜公園',
   'reason': '天候が良い場合、広大な公園内を散策し、桜や美術館、動物園を楽しむことができます。特に春には桜が美しく、ピクニックにも最適です。'},
  {'sport': '東京タワー',
   'reason': '雨天の場合でも、東京タワーの展望台からの景色を楽しむことができます。屋内での観光が可能で、雨を気にせずに東京の景色を一望できます。'}]}

ちなみに回答の型は、dictが入ったリスト形式なのでこんなふうに表示して確認も可能です

ans_list = agent_suggest(initial_state)
query = ans_list['query']
print("query:",query)
ans_dict = ans_list['answer']
ans_dict

query: 東京
[{'sport': '上野恩賜公園', 'reason': '天候が良い場合、広大な公園内を散策し、桜や自然を楽しむことができるため。'},
 {'sport': '東京タワー', 'reason': '雨天の場合でも、展望台からの景色を楽しむことができ、屋内の施設も充実しているため。'}]

# テーブルにして確認が可能
import pandas as pd
pd.DataFrame(ans_dict)

	sport	reason
0	上野恩賜公園	天候が良い場合、広大な公園内を散策し、桜や自然を楽しむことができるため。
1	東京タワー	雨天の場合でも、展望台からの景色を楽しむことができ、屋内の施設も充実しているため。

Agent_02 (agent_owm)

OpenWeatherMapを使った天気を取得するAgentです。
作り方は、前回の投稿で記載したとおり、以下になります。それを今回はLangGraphのAgent用の関数にしてます。
ポイントは、このAgent内で、create_react_agentの仕組みを使っているところです。

Agent_02 agent_owm

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langgraph.prebuilt import create_react_agent
from langchain_openai import OpenAI
from langchain.agents import load_tools
import datetime

def agent_owm(state:State): #OpenWeatherMapで天候を取得する関数（後でノードとして定義し利用）
    query = state['query']
    answer = state['answer']
    
    owm_tools = load_tools(["openweathermap-api"])
    llm = ChatOpenAI(temperature=0,model="gpt-4o-mini")
    
    # プロンプトの定義
    owm_prompt = ChatPromptTemplate.from_messages(
        [('system','与えられたinputに従って天気の関数を呼び出してください',),
         ('placeholder', '{messages}'),])
    
    agent_owm = create_react_agent(
        model=llm,
        tools=owm_tools,
        state_modifier=owm_prompt
    )

    # 今日の日付も取得して聞くことにします。(なくても動作しますが、最終回答に日付が欲しかったので）
    today_str = datetime.datetime.today().date().strftime("%Y年%m月%d日")
    weather_query=f"現在は{today_str}です。現在の{query}の天気はどうですか？"

    with yaspin(text="agent_owm Processing") as spinner:
        owm_result = agent_owm.invoke({"messages":weather_query})
        spinner.ok("✅ ")
    owm_result_str=owm_result['messages'][-1].content
    return {"query":query,"today_str":today_str,"owm_result":owm_result_str}

動作確認

# 初期値
initial_state = State(query='東京',
                      answer=[
                          {'sport': '上野恩賜公園','reason': '天候が良い場合、広大な公園内を散策し、桜や美術館、動物園を楽しむことができます。特に春には桜が美しく、ピクニックにも最適です。'},
                          {'sport': '東京タワー','reason': '雨天の場合でも、東京タワーの展望台からの景色を楽しむことができます。屋内での観光が可能で、雨を気にせず東京の景色を一望できます。'}
                      ])
# 動作確認
agent_owm(initial_state)

実行結果

✅  agent_owm Processing
{'query': '東京',
 'today_str': '2025年01月05日',
 'owm_result': '現在の東京の天気は以下の通りです：\n\n- 天気: 晴れ\n- 気温: 4.95°C\n- 最高気温: 5.79°C\n- 最低気温: 2.58°C\n- 体感温度: 3.25°C\n- 湿度: 59%\n- 風速: 2.06 m/s（風向き: 290°）\n- 雲のカバー: 0%\n\n特に雨は降っていません。'}

Agent_03 (agent_final) 最終提案するAgent

Agent_03 agent_final

def agent_final(state:State): #最終提案する関数（後でノードとして定義し利用）
    query = state['query']
    answer = state['answer']
    owm_result = state['owm_result']
    today_str = state['today_str']

    answer_table=pd.DataFrame(answer).to_html()
    prompt = ChatPromptTemplate([
      ('system', 'あなたは優秀なアシスタントです。'),
      ('user', """以下の{today_str}現在の天候と観光スポット情報から、{today_str}の{query}でのおすすめの観光スポットを提案してください。\n 観光スポット候補:{answer_table} \n 天候情報:{owm_result}""")
    ])
    agent_final = (
        prompt
        | llm
        | StrOutputParser()
    )
    with yaspin(text="agent_final Processing") as spinner:
        final_suggestion = agent_final.invoke({"query":query,"today_str":today_str,"answer_table":answer_table,"owm_result":owm_result})
        spinner.ok("✅ ")
    #return {"query":query,"answer":answer,"final_sugesstion":final_suggestion}
    return {"final_suggestion":final_suggestion}

動作確認

# 動作確認のための入力値
test_state = State(
    query='東京',
    answer=[{'sport': '上野恩賜公園',
  'reason': '天候が良い場合、広大な公園内を散策し、桜や美術館、動物園を楽しむことができます。特に春には桜が美しく、ピクニックにも最適です。'},
 {'sport': '東京タワー',
  'reason': '雨天の場合でも、東京タワーの展望台からの景色を楽しむことができます。屋内での観光が可能で、雨を気にせず東京の景色を一望できます。'}],
    owm_result="東京の現在の天気は以下の通りです：\n\n- **天候**: 晴れ\n- **風速**: 3.09 m/s（北の方向）\n- **湿度**: 63%\n- **気温**: \n  - 現在: 6.35°C\n  - 最高: 7.79°C\n  - 最低: 3.67°C\n  - 体感温度: 4.07°C\n- **降水**: なし\n- **雲のカバー**: 0%\n\n快適な晴れの日ですね！",
    today_str = "2025年01月05日"
)
# 動作確認
agent_final(test_state)

実行結果

✅  agent_final Processing
{'final_suggestion': '2025年01月05日の東京でのおすすめの観光スポットは、以下の2つです。\n\n1. **上野恩賜公園**\n   - **理由**: 天候が晴れで、気温も比較的低めですが、風が穏やかなので公園内を散策するには最適です。広大な公園内では、桜や美術館、動物園を楽しむことができ、特に自然を感じながらの散策やピクニックにぴったりです。\n\n2. **東京タワー**\n   - **理由**: 晴れた日には、東京タワーの展望台からの景色が非常に美しいです。屋内での観光が可能なので、寒さを気にせずに東京の素晴らしい景色を楽しむことができます。\n\nこの2つのスポットは、晴れた日の観光に最適ですので、ぜひ訪れてみてください！'}

ポイントとしては、結果がdictを格納したListなので、一旦テーブル表記に変換してからプロンプトに埋め込んだところでしょうか

テーブル変換方法

# Agent_02から、こんな感じで回答が返って来ている場合
answer_tmp=[{'sport': '上野恩賜公園',
  'reason': '天候が良い場合、広大な公園内を散策し、桜や美術館、動物園を楽しむことができます。特に春には桜が美しく、ピクニックにも最適です。'},
 {'sport': '東京タワー',
  'reason': '雨天の場合でも、東京タワーの展望台からの景色を楽しむことができます。屋内での観光が可能で、雨を気にせず東京の景色を一望できます。'}]
# こんなふうに変換すれば、テーブル情報としてプロンプトに渡せます
pd.DataFrame(answer_tmp).to_html()

出力結果

'<table border="1" class="dataframe">\n  <thead>\n    <tr style="text-align: right;">\n      <th></th>\n      <th>sport</th>\n      <th>reason</th>\n    </tr>\n  </thead>\n  <tbody>\n    <tr>\n      <th>0</th>\n      <td>上野恩賜公園</td>\n      <td>天候が良い場合、広大な公園内を散策し、桜や美術館、動物園を楽しむことができます。特に春には桜が美しく、ピクニックにも最適です。</td>\n    </tr>\n    <tr>\n      <th>1</th>\n      <td>東京タワー</td>\n      <td>雨天の場合でも、東京タワーの展望台からの景色を楽しむことができます。屋内での観光が可能で、雨を気にせず東京の景色を一望できます。</td>\n    </tr>\n  </tbody>\n</table>'

LangGraph

ここまでで各種Agentの準備ができたので、ようやくLangGraphとしてまとめていきます
まずは、すでに記述しましたが、各Agentの出力を振り返って、最初のclass State(TypedDict)の中をただしく記述しておきます

あとはいつもどおりLangGraphの記述手法に則って記載していきます。

LangGraphの作成

# グラフの初期化
workflow = StateGraph(State)

# Add node and edge
workflow.add_node("agent_01_suggest", agent_suggest)
workflow.add_node("agent_02_owm", agent_owm)
workflow.add_node("agent_03_final", agent_final)
workflow.add_edge(START, "agent_01_suggest")
workflow.add_edge("agent_01_suggest","agent_02_owm")
workflow.add_edge("agent_02_owm","agent_03_final")
workflow.add_edge("agent_03_final", END) 

# not memory（今回はメモリーは使わずに作ります）
graph_app = workflow.compile()

可視化

from IPython.display import Image, display
try:
    display(Image(graph_app.get_graph().draw_mermaid_png()))
except Exception:
    # This requires some extra dependencies and is optional
    pass

今回のAgent_02は、create_react_agentのライブラリを使っているので、内部的には左のような処理が入っていますが、上の単純な可視化だとそこは表現されないようです。

agent_02:create_react_aget	今回のLangGraph

create_react_agentの詳細は以下を参照

実行

さて、これで完成したので最終的に完成したものを使って試験してみます。

入力値

# 初期値
initial_state = State(query='東京')

stream_modeはupdateにしています。

実行

# streamモードをeventsと名付けたGenerator objectにする
events = graph_app.stream(initial_state, stream_mode="updates")
#print(events)
#<generator object Pregel.stream at 0x121a18580>

# 確認
total_step = 3
for i,event in enumerate(events):
    # graph全体の処理がながく途中経過を表示したい場合
    progress = (i+1)/total_step * 100
    print(f"進捗率: {progress:.1f}%")
    print(event)
    #　最終結果を表示したい場合、stream_mode="updates"には Agent名がつくレスポンスで返り値がくるためこんな感じのif文で取得する
    if 'agent_03_final' in event:
        print("==="*50)
        #print(event)
        print(event['agent_03_final']['final_suggestion'])

上記のように記述すれば、以下のように途中経過の進捗率を表示させながら、最終回答を得ることができます

✅  Agent_suggest Processing
進捗率: 33.3%
{'agent_01_suggest': {'query': '東京', 'answer': [{'sport': '上野恩賜公園', 'reason': '天候が良い場合、広大な公園内を散策し、桜や美術館、動物園を楽しむことができます。特に春には桜が美しく、ピクニックにも最適です。'}, {'sport': '東京タワー', 'reason': '雨天の場合でも、東京タワーの展望台からの景色を楽しむことができます。屋内での観光が可能で、雨に濡れずに東京の景色を一望できます。'}]}}
✅  agent_owm Processing
進捗率: 66.7%
{'agent_02_owm': {'query': '東京', 'owm_result': '現在の東京の天気は以下の通りです：\n\n- 天気: 晴れ\n- 風速: 2.06 m/s（西北西から）\n- 湿度: 59%\n- 現在の気温: 4.95°C\n- 最高気温: 5.79°C\n- 最低気温: 2.58°C\n- 体感温度: 3.25°C\n- 雨: なし\n- 雲のカバー: 0% \n\n快適な晴れの日ですね！', 'today_str': '2025年01月05日'}}
✅  agent_final Processing
進捗率: 100.0%
{'agent_03_final': {'final_suggestion': '2025年01月05日の東京でのおすすめの観光スポットは「上野恩賜公園」です。\n\n### おすすめ理由:\n- **天候が良い**: 現在の東京は晴れで、風も穏やかです。広大な公園内を散策するには最適な日です。\n- **多様な楽しみ**: 上野恩賜公園内には美術館や動物園があり、文化的な体験や動物とのふれあいを楽しむことができます。\n- **自然を満喫**: 冬の公園は静かで、散策しながら自然を楽しむことができます。特に、晴れた日には美しい景色を堪能できるでしょう。\n\nもし、天候が急変した場合や屋内での観光を希望される場合は「東京タワー」も良い選択肢です。展望台からの景色を楽しむことができ、雨に濡れる心配もありません。\n\nぜひ、素敵な一日をお過ごしください！'}}
======================================================================================================================================================
2025年01月05日の東京でのおすすめの観光スポットは「上野恩賜公園」です。

### おすすめ理由:
- **天候が良い**: 現在の東京は晴れで、風も穏やかです。広大な公園内を散策するには最適な日です。
- **多様な楽しみ**: 上野恩賜公園内には美術館や動物園があり、文化的な体験や動物とのふれあいを楽しむことができます。
- **自然を満喫**: 冬の公園は静かで、散策しながら自然を楽しむことができます。特に、晴れた日には美しい景色を堪能できるでしょう。

もし、天候が急変した場合や屋内での観光を希望される場合は「東京タワー」も良い選択肢です。展望台からの景色を楽しむことができ、雨に濡れる心配もありません。

ぜひ、素敵な一日をお過ごしください！

stream_mode="updates"にしています。これによって、Agent名がついたdict形式で回答が届き、かつupdateなので更新されたところだけを取得できます
langgraphをstream出力させていくのは、pythonのgeneratorにあたりますので、enumerate()でくくってあげれば、何番目の処理か取得できますので、それを全体のプロセスで割ってあげれば進捗率が求まるわけですので、これを入れています
Agent名付きの応答が返ることを利用すれば、if文で、そのAgent名がついた場合の処理を別にして結果を出すことができるわけです。例えば今回はAgent03を通常のテキスト出力にしましたが、Agent01のようにdictのListを出力にすれば、それをHTML変換させてレポート的な出力にすることも可能です