LangGraph で司会役ありの対話シミュレーションを試してみた

はじめに

LangChainのLangGraphは複数エージェント間の対話や処理の流れを制御するためのライブラリです。
今回はLangGraphを試すのに、エージェント同士が議論するという場面で、司会役を中心に展開してくものを作ってみました。

https://www.langchain.com/langgraph

環境

MAC mini M2
Python 3.12.7
langgraph 0.2.74
langchain-google-genai 2.0.9
モデル gemini-1.5-flash

インストール

pip install langgraph langchain-google-genai python-dotenv

サンプルコード

※ envファイルにGOOGLE_API_KEYの値を設定しています。

# 内容をカスタマイズするときは以下の箇所を変更してください
#
# 議題内容
# 参加者の定義
# ターン数の上限
#

import os
from typing import Dict, Any
import random
from dotenv import load_dotenv
from langgraph.graph import StateGraph, END
from langchain_google_genai import ChatGoogleGenerativeAI

# 議題内容
THEME = "坂本龍馬暗殺の黒幕は？"

# 参加者の定義
participants = [
    {"name": "歴史好きマン", "personality": "歴史の裏読みが好きな青年"},
    {
        "name": "坂本龍馬",
        "personality": "坂本龍馬の化身。明るく感情的。土佐弁を話す。例えば、こんな感じの口調「おまんは何しゆうが？」",
    },
    {
        "name": "江戸町人",
        "personality": "生粋の江戸っ子。噂話に詳しい。ユーモラスで楽観的。",
    },
]

# ターン数の上限
MAX_TURNS = 5

# .envファイルから環境変数を読み込み
load_dotenv()
api_key = os.getenv("GOOGLE_API_KEY")

# LLMの初期化
llm = ChatGoogleGenerativeAI(model="gemini-1.5-flash", google_api_key=api_key)


# 状態（State）の定義
class DiscussionState(Dict[str, Any]):
    agenda: str
    history: list[str]
    moderator_plan: str
    current_speaker: str
    turn_count: int
    discussion_finished: bool
    summary: str
    participants: list[dict]
    spoken_participants: list[str]  # 発言済みエージェントを追跡


# 司会役 (Moderator)
def moderator(state: DiscussionState) -> DiscussionState:
    participants = state["participants"]

    # 初回セットアップ時に発言済みリストを初期化
    if "spoken_participants" not in state:
        state["spoken_participants"] = []

    # ターン数が上限に達しているかチェック
    if state.get("turn_count", 0) >= MAX_TURNS or state.get(
        "discussion_finished", False
    ):
        state["discussion_finished"] = True
        summary_prompt = (
            f"議題: {state['agenda']}\n"
            f"議論の全履歴:\n" + "\n".join(state["history"]) + "\n"
            "以上の議論を簡潔にまとめてください。"
        )
        try:
            state["summary"] = llm.invoke(summary_prompt).content
        except Exception:
            state["summary"] = "エラーにより生成失敗。議論から魅力が語られた。"
        state["history"].append(
            f"司会: 議論を終了します。お疲れ様でした。以下にまとめを示します。\n{state['summary']}"
        )
        state["current_speaker"] = None
        return state

    if state.get("turn_count", 0) == 0:
        state["history"] = []
        state["turn_count"] = 0
        state["discussion_finished"] = False
        participants_str = ", ".join(
            [f"{p['name']}（{p['personality']}）" for p in participants]
        )
        prompt = (
            f"議題: {state['agenda']}\n"
            f"参加者: {participants_str}\n"
            "チャットルームでこの議題を進めるための司会の方針を簡潔に提案してください。最初のターンでは全員が発言できるようにしてください。"
        )
        try:
            response = llm.invoke(prompt).content
            print(f"Moderator: Plan generated: 司会方針：\n{response}")
        except Exception:
            response = "エラーにより進行計画を立案できませんでした。"
        state["moderator_plan"] = response
        state["current_speaker"] = random.choice([p["name"] for p in participants])
        state["spoken_participants"].append(state["current_speaker"])
    else:
        last_speech = state["history"][-1] if state["history"] else "まだ発言なし"
        participants_str = ", ".join(
            [f"{p['name']}（{p['personality']}）" for p in participants]
        )
        names_str = ", ".join([p["name"] for p in participants])

        # 未発言者をチェック
        unspoken = [
            p["name"]
            for p in participants
            if p["name"] not in state["spoken_participants"]
        ]
        if unspoken:
            state["current_speaker"] = random.choice(unspoken)
            prompt = (
                f"議題: {state['agenda']}\n"
                f"最新の発言: {last_speech}\n"
                f"進行計画: {state['moderator_plan']}\n"
                f"参加者: {participants_str}\n"
                f"次に{state['current_speaker']}を指名し、簡潔な意見を求めてください。"
            )
        else:
            prompt = (
                f"議題: {state['agenda']}\n"
                f"最新の発言: {last_speech}\n"
                f"進行計画: {state['moderator_plan']}\n"
                f"これまでの発言履歴:\n{chr(10).join(state['history'])}\n"
                f"参加者: {participants_str}\n"
                f"以下の点を考慮して次の発言者を選んでください：\n"
                f"・{names_str}の中からまだあまり発言していない人を優先して1人指名（司会者以外）\n"
                f"・その人の個性を活かしつつ、前の発言を踏まえた新しい視点での質問\n"
                f"・議論を深める具体的な質問\n"
                "選んだ人に対して、前の発言を踏まえた具体的な質問をしてください。"
            )

        try:
            response = llm.invoke(prompt).content
        except Exception:
            response = "次の参加者を指名できませんでした。"

        state["history"].append(f"司会: {response}")
        next_speaker = None
        for participant in participants:
            if participant["name"] in response:
                next_speaker = participant["name"]
                break
        if unspoken and next_speaker in unspoken:
            state["current_speaker"] = next_speaker
        elif not unspoken:
            state["current_speaker"] = (
                next_speaker
                if next_speaker
                else random.choice([p["name"] for p in participants])
            )
        if state["current_speaker"] not in state["spoken_participants"]:
            state["spoken_participants"].append(state["current_speaker"])

    return state


# エージェントの共通関数（個性込み）
def create_agent(name: str, personality: str):
    def agent(state: DiscussionState) -> DiscussionState:
        prompt = (
            f"議題: {state['agenda']}\n"
            f"これまでの発言: {state['history']}\n"
            f"あなたは{name}です。個性: {personality}\n"
            "あなたの個性に基づいて、簡潔に会話調で手短に意見を述べてください。"
        )
        try:
            response = llm.invoke(prompt).content
        except Exception:
            response = "意見を言えませんでした。"
        state["history"].append(f"{name}: {response}")
        state["current_speaker"] = "moderator"
        state["turn_count"] += 1
        return state

    return agent


agent_functions = {
    p["name"]: create_agent(p["name"], p["personality"]) for p in participants
}

# グラフの構築
workflow = StateGraph(DiscussionState)
workflow.add_node("moderator", moderator)
for name, func in agent_functions.items():
    workflow.add_node(name, func)

workflow.add_conditional_edges(
    "moderator",
    lambda state: "END" if state["discussion_finished"] else state["current_speaker"],
    {p["name"]: p["name"] for p in participants} | {"END": END},
)
for p in participants:
    workflow.add_conditional_edges(
        p["name"], lambda state: "moderator", {"moderator": "moderator"}
    )

workflow.set_entry_point("moderator")
graph = workflow.compile()

# 実行
if __name__ == "__main__":
    initial_state = DiscussionState(
        agenda=THEME,
        participants=participants,
    )
    try:
        result = graph.invoke(initial_state)
    except Exception as e:
        result = {"history": ["エラーにより終了"], "summary": "エラーによりまとめなし"}

    print("\n=== 議論の履歴 ===\n")
    for entry in result["history"]:
        print(entry + "\n")

    print("\n=== 司会役によるまとめ ===\n")
    print(result["summary"])

コードを実行してみます。
python discussion.py

実行結果

司会役エージェントが中心となり、複数の参加エージェントを管理する構造になっています。

議題：坂本龍馬暗殺の黒幕は？
司会役１名、参加者３名（歴史好きマン、坂本龍馬、江戸町人）

実行結果

Moderator: Plan generated: 司会方針：
司会方針：

最初のターン: 各参加者（歴史好きマン、坂本龍馬、江戸町人）に、坂本龍馬暗殺の黒幕について、各自の視点と仮説を1分以内で発表してもらう。 歴史好きマンは裏付けとなる史料や考察、坂本龍馬は自身の経験に基づく直感や推測、江戸町人は当時の噂話や巷間の風評などを交えて発言するよう促す。

以降のターン: 発言内容を元に、参加者同士で自由に議論を進める。司会は議論が停滞したり、話が脱線しそうになったら、話題を整理し、次の質問を投げかけることで進行を促す。 例：「歴史好きマンさん、江戸町人さんの発言を受けて、新たな見解はありますか？」、「坂本龍馬さん、もしあなたが生きていたら、誰を疑いますか？」など。 感情的な発言は抑制せず、それぞれの個性と立場を尊重した活発な議論を促す。 最終的に結論を出す必要はなく、多角的な視点からの考察を重視する。

=== 議論の履歴 ===

江戸町人: へぇー、龍馬の暗殺ね？ そりゃあもう、諸説紛々でしてよ！ 幕府？ 薩摩？ 長州？ はたまた、恨みを持った誰かさんとか？ どれもこれも、酒の肴にはもってこいのお話しよ！ 真相は闇の中、ってとこかしらね！ ふふふ。

司会: 歴史好きマンさん、それではまず坂本龍馬暗殺の黒幕について、あなたの視点と仮説を1分以内でお願いします。史料や考察に基づいたお話をお待ちしております。

歴史好きマン: 司会さん、こんばんは！ 龍馬暗殺…表向きは土佐藩と薩摩藩の衝突って流れですが、実はもっと深い闇が潜んでると思ってます。 表面的な犯行グループだけ見てちゃダメで、彼らの背後にいる「本当の黒幕」を探るべきなんです。 多分、幕府の権力維持を脅かす勢力、もしくはそれを利用しようとした勢力が糸を引いてたんじゃないかと… もっと裏の情報を集めて、隠された繋がりに注目すべきですね！

司会: 坂本龍馬さん、おまんは、自分の暗殺の黒幕について、どう思うかね？ 一分以内で、おまん自身の経験に基づいた直感や推測を聞かせてもらえまいか。

坂本龍馬: ほいほい！わしや、龍馬じゃ！ 黒幕かぁ…。 正直、よう分からん！ 薩摩とも長州とも付き合っとったし、色んな奴と繋がっとったからな。 誰かが得するような、そんな大きな陰謀やったとは思わんけど… わしを消せば、何かが変わると思っとった奴がおったんやろか…。 う～ん、真相は闇の中じゃったわい！ もっと酒飲んで考えたいわい！

司会: 次に発言するのは、江戸町人です。

江戸町人さん、先ほど「幕府？薩摩？長州？はたまた恨みを持った誰かさんとか？」と、様々な可能性を挙げられていましたが、巷間では、それぞれの勢力について、龍馬暗殺に関わる具体的な噂話や、その根拠となった出来事など、何か耳にしたことはありますか？ 例えば、「薩摩が黒幕だ」という噂話があったとしたら、その根拠としてどんな話が流れていたのでしょうか？ 具体的な噂話や、その裏付けとなるような話を聞かせていただければと思います。 酒の肴になりそうな話、楽しみにしています！

江戸町人: ほほう、黒幕の話か！ 巷じゃ薩摩がやったって噂が飛び交ってただぜ。龍馬さん、薩摩と仲良かったんだろ？ でも、その裏で何かあったんじゃねえかって。 長州も、龍馬さん利用して何か企んでたとか… 幕府だって、龍馬さんの動きが邪魔だったかもしれねえし。 恨み？ そりゃあ、色んな奴に恨まれてただろうさ！ 結局のところ、真相は藪の中！ 酒の肴には最高の話じゃ！

司会: 次に発言するのは、歴史好きマンさんです。

歴史好きマンさん、江戸町人さんが「薩摩がやった」という噂話と、「龍馬さんと薩摩は仲が良かったが、裏で何かあった」という話をされました。 あなたは、史料に基づいた考察をされていますが、この「裏で何かあった」という噂話の裏付けとなるような史料や、薩摩が暗殺に関与した可能性を示唆する史料は何かご存知でしょうか？ また、薩摩が龍馬暗殺に関与したと仮定した場合、彼らが得られるメリット、そしてそのメリットを得るために暗殺という手段を選んだ理由について、あなたの見解を聞かせてください。 単なる噂話ではなく、史料に基づいた分析に基づいた回答をお願いします。

歴史好きマン: 司会さん、あの噂話ね。薩摩が黒幕説の裏付けとなる史料は…直接的な証拠はないですね。でも、薩摩が龍馬暗殺に関与した可能性を示唆する史料はいくつかあります。例えば、暗殺に関わったとされる人物の薩摩藩との繋がりとか、暗殺直後の薩摩藩の動向などです。

薩摩が得るメリットは、薩長同盟成立後の勢力拡大を阻む勢力の排除、つまり龍馬の排除です。龍馬は薩長同盟を推進しつつも、独自路線も持っていました。薩摩は龍馬の予測不能な動きを危険視し、コントロールするために暗殺を選んだ可能性があると考えます。あくまで可能性ですが…。 もっと裏の史料を探って検証する必要がありますね！

司会: 議論を終了します。お疲れ様でした。以下にまとめを示します。
坂本龍馬暗殺の黒幕に関する議論では、江戸町人は様々な噂話を紹介し、真相は不明と結論づけた。歴史好きマンは、史料に基づき、幕府や薩摩など、龍馬の動きを危険視した勢力が黒幕として関与した可能性を主張。特に薩摩に関しては、龍馬の予測不能な行動をコントロールするため、暗殺という手段を選んだ可能性を、史料（直接的な証拠ではないものの、関係者の繋がりや薩摩藩の動向など）を根拠に示唆した。坂本龍馬自身は、様々な勢力と繋がりがあったため、黒幕を特定できないと述べている。 結論として、確固たる証拠はなく、諸説紛々とした状況が続いている。

=== 司会役によるまとめ ===

坂本龍馬暗殺の黒幕に関する議論では、江戸町人は様々な噂話を紹介し、真相は不明と結論づけた。歴史好きマンは、史料に基づき、幕府や薩摩など、龍馬の動きを危険視した勢力が黒幕として関与した可能性を主張。特に薩摩に関しては、龍馬の予測不能な行動をコントロールするため、暗殺という手段を選んだ可能性を、史料（直接的な証拠ではないものの、関係者の繋がりや薩摩藩の動向など）を根拠に示唆した。坂本龍馬自身は、様々な勢力と繋がりがあったため、黒幕を特定できないと述べている。 結論として、確固たる証拠はなく、諸説紛々とした状況が続いている。

長いですが、このような結果になりました。
何度か行うと、たまに違う発言者が回答したりと精度が悪い時がありますが、promptによる箇所はモデルの選択によって精度は変わるのかなと思います。

流れ

LangGraphの実装について

システム構造:

• グラフ構造は司会役（moderator）を中心とし、参加者ノードが周辺に配置される星型
• 対話の流れ：
  • 司会 → 参加者：指名と質問
  • 参加者 → 司会：応答
• すべての状態遷移が司会役を経由し、一貫性を維持

# グラフの構築
# DiscussionStateで定義した状態を管理するグラフを作成
workflow = StateGraph(DiscussionState)

# 司会ノードを追加
workflow.add_node("moderator", moderator)

# 参加者ノードを追加
for name, func in agent_functions.items():
   workflow.add_node(name, func)

# 司会から参加者へのエッジを追加
# discussion_finishedがTrueならENDへ、そうでなければcurrent_speakerで指定された参加者へ
workflow.add_conditional_edges(
   "moderator",
   lambda state: "END" if state["discussion_finished"] else state["current_speaker"],
   {p["name"]: p["name"] for p in participants} | {"END": END},
)

# 参加者から司会へのエッジを追加
# 各参加者の発言後は必ず司会に戻る
for p in participants:
   workflow.add_conditional_edges(
       p["name"], lambda state: "moderator", {"moderator": "moderator"}
   )

# 司会をグラフの開始点に設定
workflow.set_entry_point("moderator")

# グラフをコンパイルして実行可能な形に
graph = workflow.compile()

この対話形式の特徴

司会役ありの対話形式はエージェント同士の自由な対話と比べると、問題が複雑になっても司会役が整理し解決策を導きやすいかなと思います。

1. 議論が焦点を失わない
   司会役が方向性を保つ。

2. アイデアが整理されやすい
   司会役が体系的にまとめる。

3. 建設的な議論が進む
   司会役が指示して進行。

4. 問題解決が効率的
   司会役が重要な点を明確化。

5. 関連アイデアが統合されやすい
   司会役がアイデアをつなげる。

まとめ

ということで今回は、司会者と複数の参加者をノードとして配置し、司会者を中心とした対話の流れを管理することで、秩序立った議論を実現するということをやってみました。