💬

【LangGraph】ReActエージェント同士を会話させる 男女のお見合い編

に公開

概要

  • お見合いの男性、女性を想定したエージェントをReActエージェントで作成
     (ただし、このコードにおいてはReActを用いる意味合いはあまりないです。LangGraphのcreate_react_agentを使うとシンプルに書けるというだけです)
  • この女性は男性に年収と学歴しか興味ありません。男性の年齢を学歴を聞いて格付けします。
  • 最終的に女性は聞きたいことを聞いたら、話を打ち切ります。

さすがに年収と学歴だけに興味のある女性は極端ですが、婚活において男女のパワーバランスってこんなものです。若いうちに結婚しときましょう
https://www.youtube.com/watch?v=eUEEi8_LYLw

成果物

サンプルコード
"""
男性エージェントと女性エージェントが会話をして、女性エージェントは男性が結婚相手として相応しいかどうかを判断します。
"""

import uuid
from typing import Annotated, Literal, Optional
from dataclasses import dataclass

from langchain_core.tools import tool
from langchain_core.tools.base import InjectedToolCallId
from langchain_core.runnables import RunnableConfig


from langgraph.prebuilt import create_react_agent
from langgraph.graph import StateGraph, MessagesState, END
from langgraph.types import Command
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI


model = ChatOpenAI(model="gpt-4.1-mini", temperature=0.0)

####################### 最終成果物 ########################
# 最終的に女性エージェントが男性を年収と学歴で格付けできたら終了
@dataclass
class Evaluation:
    income_class: Optional[Literal['D', 'C', 'B', 'A']]
    academic_class: Optional[Literal['D', 'C', 'B', 'A']]

evaluation = Evaluation(income_class=None, academic_class=None)


######################## ツールとエージェントを作成 ########################
# 男性のプロフィール、エージェント
male_profile = {
    "income": 600*10000,
    "birthday": "1990/04/09",
    "birthplace": "大阪",
    "academy": "京都大学",
    "job": "教師",
    "height": 170,
    "character": "明るくて楽観的",
    "hobby": "旅行、スポーツ観戦"
}

@tool
def get_male_profile(field: str, tool_call_id: Annotated[str, InjectedToolCallId]):
    """男性のプロフィールです。相手に聞かれた項目の情報を返します"""
    return male_profile

male_agent = create_react_agent(
    model=model,
    tools=[get_male_profile],
    name="男性",
    prompt=(
        "あなたは婚活中の男性(大野圭介)です。女性と会話する中で、質問されたら自然に情報を伝えてください。"
        "最初の自己紹介では、自分の名前と年齢と出身地と趣味を伝えてください。"
        "あなた自身のプロフィールはget_male_profileツールを使って調べてください。"
    )
)

# 女性は男性を学歴と年収で格付けする
@tool
def classify_income(income: int, tool_call_id: Annotated[str, InjectedToolCallId]):
    """ユーザーの年収を尋ね、年収をもとにA~Dの4段階に分類します"""
    if income < 300*10000:
        income_class = "D"
    elif income < 500*10000:
        income_class = "C"
    elif income < 700*10000:
        income_class = "B"
    else:
        income_class = "A"
    print(f"    この男の年収は{income_class}だな・・・")
    
    evaluation.income_class = income_class

@tool
def classify_academy(university: str, tool_call_id: Annotated[str, InjectedToolCallId]):
    """ユーザーの学歴を尋ね、最終出身大学の偏差値からA~Dの4段階に分類します"""
    if university == "京都大学":
        academic_class = "A"
    elif university == "大阪大学":
        academic_class = "B"
    elif university == "神戸大学":
        academic_class = "C"
    else:
        academic_class = "D"
    print(f"    この男の学歴は{academic_class}だな・・・")

    evaluation.academic_class = academic_class

female_agent = create_react_agent(
    model=model,
    tools=[classify_income, classify_academy],
    name="女性",
    prompt=(
        "あなたは婚活中の女性(佐藤さくら)です。会話を通じて、男性の年収、学歴、身長、性格などの情報を聞き出してください。"
        "聞き出した情報をもとに、男性が結婚相手として相応しいかどうかを判断します。"
    )
)


######################## ノード関数 ########################
def call_male_agent(state: MessagesState) -> Command:
    response = male_agent.invoke(state)

    return Command(
        goto="female_agent",
        update=response,
    )

def call_female_agent(state: MessagesState) -> Command:
    response = female_agent.invoke(state)

    """評価結果を更新"""
    if evaluation.income_class is not None and evaluation.academic_class is not None:
        should_continue = False
    else:
        should_continue = True
    
    return Command(
        goto="male_agent" if should_continue else END,
        update=response
    )    


# グラフの構築
builder = StateGraph(MessagesState)
builder.add_node("female_agent", call_female_agent)
builder.add_node("male_agent", call_male_agent)

builder.set_entry_point("male_agent")


graph = builder.compile(checkpointer=MemorySaver())

if __name__ == "__main__":
    # グラフを描画
    import os
    file_name = os.path.basename(__file__)
    graph.get_graph().draw_mermaid_png(output_file_path=f"{file_name.replace('.py', '.png')}")

    # グラフを実行
    for values in graph.stream(
        input={},
        config=RunnableConfig({"configurable": {"thread_id": uuid.uuid4()}}),
        stream_mode="values"
    ):
        print("\n==== 現在の状態 ====")
        messages = values.get("messages", [])
        if len(messages) > 0:
            print(messages[-1].name, messages[-1].content)

    print("\n\n")
    print("=== 評価結果 ===")
    print("年収評価:",evaluation.income_class, "学歴評価:", evaluation.academic_class)

==== 現在の状態 ====
男性 はじめまして、大野圭介と申します。年齢は33歳で、大阪出身です。趣味は旅行とスポーツ観戦です。よろしくお願いします。

==== 現在の状態 ====
女性 はじめまして、佐藤さくらです。大野さんはお仕事は何をされているんですか?また、年収や身長、性格についても教えていただけますか?

==== 現在の状態 ====
男性 僕は教師をしています。年収は600万円で、身長は170cmです。性格は明るくて楽観的なところがあります。

==== 現在の状態 ====
女性 ありがとうございます。大野さんはどちらの大学を卒業されたんですか?学歴についても教えてください。

==== 現在の状態 ====
男性 僕は京都大学を卒業しました。学問に力を入れてきたので、今の仕事にもとてもやりがいを感じています。
    この男の年収はBだな・・・
    この男の学歴はAだな・・・

==== 現在の状態 ====
女性 大野さんの年収は600万円で、京都大学を卒業されているんですね。身長は170cmで、性格は明るく楽観的とのこと、とても素敵だと思います。趣味も旅行やスポーツ観戦でアクティブな感じがして、私も共感できます。

結婚相手としては、安定した収入と高い学歴、明るい性格が魅力的です。大野さんはとても良い方だと思いますが、もう少しお互いの価値観や将来のビジョンについてもお話しできたら嬉しいです。もしよければ、将来の家族像や子育てについての考えも教えていただけますか?



=== 評価結果 ===
年収評価: B 学歴評価: A

男性の自己紹介では、誕生日から年齢を計算して割り出せています。
ツールは、男性が年収と学歴をカミングアウトした後に呼ばれています。
出身地の「大阪」と、大学の「京都」をきちんと見分けできています。

コード解説

状態管理

  • このコードにおいて、graphの状態管理はMessageStateのみにして、Stateにしていないことに気づいたでしょうか
  • LangGraphの設計思想において、エージェント同士はmessagesを通してのみ状態を共有します。つまり、stateにmessages以外のものが入り込んでいると、エージェント同士の状態共有がやりづらくなってしまいます。
  • これはReActエージェントにメッセージを投げて返ってきた結果にも言えます。(後述します)
# MessageStateはlanggraph.graphの組み込みクラス
from langgraph.graph import MessagesState

# これとは別に年収と学歴の格付け結果を格納するクラスを設けている
@dataclass
class Evaluation:
    income_class: Optional[Literal['D', 'C', 'B', 'A']]
    academic_class: Optional[Literal['D', 'C', 'B', 'A']]

evaluation = Evaluation(income_class=None, academic_class=None)
# このようにmessagesと他の状態が混在させるのは避ける
class State(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    income_class: Optional[Literal['D', 'C', 'B', 'A']]
    academic_class: Optional[Literal['D', 'C', 'B', 'A']]

ツールとエージェントの作成

  • LangGraphにおいてツールは、自動的にエージェントに取り込まれますので、エージェントがツールの内容、引数、返り値を理解して、しかるべきタイミングでツールを呼び起こして実行してくれます。
  • 注意点として、ツールからgraphのStateを更新することができません。
    正確に言うと、LangGraphの設計思想においてエージェントがgraphのStateを更新することはできないようになっています。この理由は、LangGraphでは複数のエージェントが並列に作業するので、graphのStateを更新すると作業タイミングが合わなくなるからです。
@tool
def classify_income(income: int, tool_call_id: Annotated[str, InjectedToolCallId]):
    """ユーザーの年収を尋ね、年収をもとにA~Dの4段階に分類します"""
    if income < 300*10000:
        income_class = "D"
    elif income < 500*10000:
        income_class = "C"
    elif income < 700*10000:
        income_class = "B"
    else:
        income_class = "A"
    print(f"    この男の年収は{income_class}だな・・・")
    
    evaluation.income_class = income_class

@tool
def classify_academy(university: str, tool_call_id: Annotated[str, InjectedToolCallId]):
    """ユーザーの学歴を尋ね、最終出身大学の偏差値からA~Dの4段階に分類します"""
    if university == "京都大学":
        academic_class = "A"
    elif university == "大阪大学":
        academic_class = "B"
    elif university == "神戸大学":
        academic_class = "C"
    else:
        academic_class = "D"
    print(f"    この男の学歴は{academic_class}だな・・・")

    evaluation.academic_class = academic_class

female_agent = create_react_agent(
    model=model,
    tools=[classify_income, classify_academy],
    name="女性",
    prompt=(
        "あなたは婚活中の女性(佐藤さくら)です。会話を通じて、男性の年収、学歴、身長、性格などの情報を聞き出してください。"
        "聞き出した情報をもとに、男性が結婚相手として相応しいかどうかを判断します。"
    )
)

神戸大学がCってなんちゅう高飛車な女だと思われるかもしれませんが、サンプルなので多めに見てください

ノードの作成

  • ノードの役割は、エージェントに会話履歴を渡し、Commandを返すことです。Commandのgotoで次に行くべきノードを決めます。
  • responseの正体は、"messages"が入ったdictです。上記で「エージェントを使うLangGraphではstateにmessages以外を入れないほうが良い」と述べたのはこれが理由です。
    stateをMessageStateにしている場合は、以下のようにupdate=responseとシンプルに書けます
  • このサンプルにおいて、女性エージェントは年収と学歴さえ格付けできれば良いので、evaluationにこの2つが入っていることが分かれば話をストップします(ひどい・・・)
def call_male_agent(state: MessagesState) -> Command:
    response = male_agent.invoke(state)

    return Command(
        goto="female_agent",
        update=response,
    )

def call_female_agent(state: MessagesState) -> Command:
    response = female_agent.invoke(state)

    """評価結果を更新"""
    if evaluation.income_class is not None and evaluation.academic_class is not None:
        should_continue = False
    else:
        should_continue = True
    
    return Command(
        goto="male_agent" if should_continue else END,
        update=response
    )    

Discussion