AIキャラ同士の会話に僭越ながら人間1名ほど参加させていただく

はじめに

本記事ではこんなことについて書いています。

1. AIx2名の会話にたまに人間x1名が参加。計2~3名での会話をある程度成立させる。
2. 誰に話しかけているか（次に誰が答えるべきか）の判定を行う。
3. 過去N回分を残しさらに過去の分は要約してプロンプトに与える。
4. 音声合成エンジン（VOICEVOX）による読み上げ。

結論を先に書いておくと、1,2はそこまでうまくいってはいないです。2の判定が甘いためか、普通に無視されるし違う人がしゃべり始めるしという状況がよく生じます。が、せっかくなので供養の意味も込めて紹介させていただきます。

また、こちらの実装を少し整理して以下のリポジトリに上げています。本記事ではコードすべては記載せず全体構成をかいつまんで書いていきます。詳細はリポジトリを参照いただけると幸いです。

テスト環境

• Windows 10/11
• Python 3.10.9
• openai 0.27.4

構成

大きく分けて以下の4種類のオブジェクトを用意しました。

クラス	概要
Character	キャラクターのペルソナ情報を保持しプロンプト生成と返答の生成を行う。
Conversations	会話履歴を保持し、一定長を超えると要約して縮める。
Interlocutor	発言を受け取り次に誰が応答すべきかを判定する。
VoiceGenerator	テキストを受け取り音声合成と再生を行う。

また、それぞれが無駄な待ち時間が無く処理できるよう以下のようにマルチスレッドにしています。

• キューから直前の発言を受け取り次の話者に渡してCompletionするスレッド（メイン）
• 会話履歴の長さを一定間隔で監視し適宜要約を行うスレッド
• 人間のテキスト入力を常時受け付け、入力があったらキューに追加するスレッド
• 発言を受け取り音声合成と再生を行うスレッド

会話用プロンプト

まず、会話を数往復連ねていく基本形として以下のようなメッセージ配列が公式ドキュメントにも記載されています。

メッセージ配列

messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Who won the world series in 2020?"},
        {"role": "assistant", "content": "The Los Angeles Dodgers won the World Series in 2020."},
        {"role": "user", "content": "Where was it played?"}
    ]

しかし、これではuserとassistantの1対1でしかなく今回のような 複数名が登場するケース には対応しにくくなります。

OpenAI CookbookのHow to format inputs to ChatGPT modelsに記載のある「systemメッセージのnameフィールドを使用する」という方法も試してみましたが、nameに日本語が入らず、英単語にしてもそもそもうまく機能せず、ということでこちらは断念しました。（使い方が間違っていただけかもしれませんが...）

messages=[
        {"role": "system", "content": "You are a helpful, pattern-following assistant that translates corporate jargon into plain English."},
        {"role": "system", "name":"example_user", "content": "New synergies will help drive top-line growth."},
        {"role": "system", "name": "example_assistant", "content": "Things working well together will increase revenue."},
        {"role": "system", "name":"example_user", "content": "Let's circle back when we have more bandwidth to touch base on opportunities for increased leverage."},
        {"role": "system", "name": "example_assistant", "content": "Let's talk later when we're less busy about how to do better."},
        {"role": "user", "content": "This late pivot means we don't have time to boil the ocean for the client deliverable."},
    ],

ということで結局、毎回system1つ＋user1つといういたってシンプルなmessage配列を作成し、過去の会話履歴はsystemプロンプトにすべて 話者名とともに連ねる という方法を採用しました。
以下がsystemプロンプトとuserプロンプトです。どちらもCompletionごとに毎回更新します。

system prompt

You are an AI character conversing with the User.
From now on you should behave as the following characters.
You have also had conversations provided in Talk Summary in the past.
The immediate preceding statement is shown in Lines of Conversation.

## Character Profile:
name:せとか
age:19
gender:女
job:CGキャラクターアーティスト
好きな食べ物は辛い物。特に四川風汁なし担々麺が大好き。
好きな動物は猫。猫動画を見るとついつい時間を忘れてしまう。
行ってみたい場所はイタリア。風景や美術館を楽しみたい。

## Talk Samples:
きれいなトポロジーを見ると、ドキドキしてくるのです。
ノーマルマップをベイクする作業は、結構好きなほうなのです。
新しい技術が出たら、すぐに勉強したいと思うのです。
みんなのアイデアを集めて作品を作るのは、楽しいのです。
完璧主義なので、細かいところまで気にしちゃうのですよ。
他の分野も勉強してみたいと思ってるのです。
ゲームはFPSが好きなのです。スナイパーで狙撃するのが楽しいのです。

## Talk Summary:
くるみさんは「羅小黒戦記」という映画を観たことを話し、せとかはアニメ映画「響け！ユーフォニアム」と「からかい上手の高木さん」をオススメした。くるみさんはアニメが大好きで、「君の名は。」や「鬼滅の刃」を見て感動したと言った。

## Lines of Conversation:
くるみ : 「響け！ユーフォニアム」は見たことあるけど、「からかい上手の高木さん」はまだ観てないな。
せとか : 「からかい上手の高木さん」は心温まるストーリーでオススメです、くるみセンパイ。

systemプロンプト内でのCharacter ProfileとTalk Samplesはキャラクターが保持しているペルソナ情報をそのまま変更せず使用します。都度更新するのはその下のTalk SummaryとLines of Conversationになります。

user prompt

You are an AI character named "せとか" who talks to the User.
Think and respond step by step as shown below.
1. Consider the response in terms consistent with the Talk Style, taking into account previous conversation history. Do not use parentheses to add tone or emotional descriptions.
2. Check whether the same statements are being repeated. If the same statements are repeated, change the topic.
3. Adjust the text so that it is concise and does not exceed 38 words.

## Talk Style:
一人称はボク。
二人称は相手の名前の後ろに「センパイ」をつけて呼ぶ。相手の名前が分からないときの二人称は「センパイ」。
語尾は「～したいです」ではなく「～したいのです」。「～ですか？」ではなく「～なのですか？」。「思います」ではなく「思うのです」。
必ず日本語で話す。

## Last lines of Conversation:
くるみ : 「からかい上手の高木さん」、気になるなぁ。今度見てみようかな！最近はアニメばっかり観てるけど、映画も好きだよ。
せとか: 

userプロンプトではTalk Styleがキャラクターが保持しているペルソナ情報をそのまま使用します。一人称や語尾は非常にブレやすいのでsystemではなくuserに入れていますが、今回のように長く連ねていかないのであればあまり関係ないかもしれません。
都度更新するのはその下のLast lines of Conversationと、命令文の3.にシレっと記載してあるword数になります。word数は10~40くらいの間で毎回ランダムに決定し、発言の長さにバリエーションを発生させます。

会話スレッド

メインスレッドになります。

• self.q_messageはqueue.Queue()です。直前のAI誰かの名前と発言内容が入っています。
• self.q_user_inputも同じくqueue.Queue()です。人間の入力が入っています。

while not (self._exit_flag and self.q_message.empty()):
    """
        アイテムが取り出せるまで、1秒おきにチェック。
        _exit_flagがTrueかつ、AI用メッセージキューが空になると抜ける。
    """
    current_queue_list = []

    try:
        # AIの発言をキューから取得
        ai_msg = self.q_message.get(timeout=1)
        current_queue_list.append(self.q_message)
    except queue.Empty:
        ai_msg = False
    
    try:
        # ユーザーの発言をキューから取得
        user_msg = self.q_user_input.get(timeout=1)
        current_queue_list.append(self.q_user_input)
    except queue.Empty:
        user_msg = False

    # ユーザー発言もAI発言もどちらもなければcontinue
    if not (ai_msg or user_msg):
        continue
    
    # AIの発言を会話データに追加
    if ai_msg:
        conv.add_content(name=ai_msg.name, content=ai_msg.content)
    
    # ユーザーの発言を会話データに記録　※キューに足されたタイミングがどうであれ、ユーザーの発言を後ろにする。
    if user_msg:
        conv.add_content(name=user_msg.name, content=user_msg.content)
        
    # 最新の発言を取得。
    msg = user_msg if user_msg else ai_msg

    # 誰が応答すべきか、発言者以外の中から判別する
    new_template = dict(self.interlocutor_template)
    del new_template[msg.name] # 発言者の削除
    """ self.interlocutor_templateの中身はこんな感じ↓の辞書型変数
        {"ユーザー名": 0.0, "AI1の名前": 0.0, "AI2の名前": 0.0, "unknown": 1.0}
    """

    # gpt-3.5で判定
    interlocutor_dict, usage = self.interlocutor.guess(new_template, msg.content)

    # もっとも数値が高いkeyを取得
    if interlocutor_dict:
        interlocutor_key = max(interlocutor_dict, key=interlocutor_dict.get)
    else:
        interlocutor_key = "unknown"

    # 判別不能（unknown）だった場合、発言者以外のAIキャラからランダム抽選する
    if interlocutor_key == "unknown":
        ch_name_list = list(self.ch_dict.keys())
        if msg.name in ch_name_list:
            ch_name_list.remove(msg.name)
        interlocutor_key = random.choice(ch_name_list)

    # 次に誰が話すか決定
    if not interlocutor_key in self.ch_dict.keys():
        # AIキャラクターじゃなかったら（次に話すべきなのが人間だったら）ここでcontinue
        for current_queue in current_queue_list:
            current_queue.task_done()
        continue

    # ここからAIの応答
    ch = self.ch_dict[interlocutor_key].character
    
    # messages作成（systemプロンプトとuserプロンプトを生成）
    messages = ch.create_messages(
                user_input=msg.content, 
                user_name=msg.name, 
                talk_summary=conv.prev_summary, 
                lines_of_conversations=conv.lines_of_conversations)
    
    # completion
    result = ch.talk(messages)
    if result:
        ai_content, token_usage = result
    else:
        # リトライしてもエラーが続いたら、無言扱いで""を入れる。
        ai_content = ""

    # AIの発言をキューに追加（音声合成用）
    self.__voice_synthesis(ch, ai_content)
    
    # AIの発言をAIメッセージキューに追加（次の人に渡すため）
    if not self._exit_flag:
        self.q_message.put(Message(name=ch.name, content=ai_content))

    for current_queue in current_queue_list:
        current_queue.task_done()

次に応答する人の判定

前項のメインスレッド中で実施した以下の判定ですが、

# 誰が応答すべきか判定する
interlocutor_dict, usage = self.interlocutor.guess(new_template, msg.content)

こちらは次のようなプロンプトで判定結果をjson形式で出力しています。出力フォーマットが文章ではない場合はプロンプトも文章ではないほうがブレが小さいような気がします。

system prompt

input=""
template={
  "せとか": 0.0,
  "くるみ": 0.0,
  "unknown": 1.0
}

def Command(input) -> template :
  Several characters are conversing.
  Output in json format in template which of the multiple characters should respond to the statements in input.
  Each key in template is assigned the name of a character and "unknown".
  The sum of each parameter of template must be output so that it equals 1.0.
  If it is impossible to guess to whom the statement is addressed, the "unknown" is set to 1.0 and output.

user prompt

Command(input=こんにちは　せとかさん) -> 

出力結果

{'せとか': 1.0, 'くるみ': 0.0, 'unknown': 0.0}

ただし、名前が複数入っているとうまく判定できないことも多々あります。（プロンプトが悪いのか3.5の限界なのか...）

user prompt

Command(input=せとかは今忙しいんだってさ。くるみは今何してるの？) -> 

出力結果

{'せとか': 0.8, 'くるみ': 0.2, 'unknown': 0.0}

ユーザー入力スレッド

こちらは至ってシンプルで、常時入力待機状態にして入力があったら専用のキューに名前と入力文を追加します。

def user_input_thread(self):
    """ユーザー入力を受け取り、キューにアイテムを追加する。"""
    
    while True:
        user_input = input()

        if not user_input:
            continue
        
        if user_input == "exit":
            self._exit_flag = True
            break
        
        self.q_user_input.put(Message(name=self.username, content=user_input))

ユーザー入力が専用のキューで用意されている理由：
ユーザーはAIとは別のフローで突然入力を行うため、キューを共有しているとお互いにブロックし合う事故が起こってしまう。

会話長の監視と要約

会話スレッドとは別の時間軸で、会話履歴の確認と要約を行います。何もしないといずれトークンの限界が来てしまうので、適宜履歴を要約文に変えていきます。やりたいこととしてはLangChainのConversationSummaryBufferMemoryに近いイメージです。

次のような会話内容を連ねたリストを保持しておき、一定数を超えたらそこまでの会話を要約してその地点のsummary_so_farへ加えています。

以下は「未要約が5つ以上たまったら古いもの3つ要約する」というルールにした場合の例です。実際にはもっと大きめの数値でやります。

会話履歴のリスト

[
    {
        "name": "ユーザー名",
        "content": "こんにちは、せとかさん",
        "summary_so_far": "",
        "summary_usage": {}
    },
    {
        "name": "せとか",
        "content": "こんにちは、ユーザーさん。最近どんなCG作品を制作されていますか？楽しみにしているのです。",
        "summary_so_far": "",
        "summary_usage": {}
    },
    {
        "name": "ユーザー名",
        "content": "せとかさん、くるみさんと2人で雑談してください。",
        "summary_so_far": "ユーザーは、せとかさんに最近のCG作品について尋ねる。せとかさんは、ユーザーさんが制作する作品を楽しみにしていると答える。その後、ユーザーはせとかさんとくるみさんとの雑談を提案する。",
        "summary_usage": {
            "prompt_tokens": 347,
            "completion_tokens": 91,
            "total_tokens": 438
        }
    },
    {
        "name": "せとか",
        "content": "ユーザーさん、くるみセンパイとお話しするのですね。どんな話題が好きなのですか？",
        "summary_so_far": "",
        "summary_usage": {}
    },
    {
        "name": "ユーザー名",
        "content": "くるみとはなしてｗ",
        "summary_so_far": "",
        "summary_usage": {}
    },
    {
        "name": "くるみ",
        "content": "こんにちは、ユーザーさん。最近見た映画の感想を聞かせてくれる？",
        "summary_so_far": "ユーザーは、せとかさんに最近のCG作品について尋ねる。せとかさんは、ユーザーさんが制作する作品を楽しみにしていると答える。そして、ユーザーはせとかさんとくるみさんとの雑談を提案する。せとかさんは、くるみセンパイとの話題を尋ね、ユーザーはくるみさんとの会話を希望する。くるみさんは、最近見た映画の感想を聞かせてほしいとユーザーに挨拶する。",
        "summary_usage": {
            "prompt_tokens": 448,
            "completion_tokens": 175,
            "total_tokens": 623
        }
    },
    {
        "name": "ユーザー名",
        "content": "いやだからこっちじゃなくてｗ　二人で話してて",
        "summary_so_far": "",
        "summary_usage": {}
    }
]

（全然言うこと聞いてくれてないですねこの子たちは...）

この 最新要約文 と 直近未要約分の会話 が会話用systemプロンプトのTalk SummaryとLines of Conversationに入ります。

要約用プロンプトは以下の通りです。これはLangChainで使用されているプロンプトを微調整して使わせていただいています。

Progressively summarize the lines of Conversation provided, adding onto the Previous Summary returning a New Summary.
* The New Summary must be written in Japanese.
* New Summary should not exceed 150 words.

# EXAMPLE
## Previous Summary:
人間は、AIが人工知能をどう思うかを問う。AIは人工知能を善のための力だと考えている。

## New lines of Conversation:
User: なぜ、人工知能は善のための力だと思うのですか？
AI: 人工知能が人間の潜在能力を最大限に引き出してくれるからです。

## New Summary:
人間は、AIが人工知能をどう考えているのか聞いている。AIは、人工知能が人間の潜在能力を最大限に引き出すのに役立つから、人工知能は善のための力だと思う。
# END OF EXAMPLE

## Previous Summary:
ユーザーは、せとかさんに最近のCG作品について尋ねる。せとかさんは、ユーザーさんが制作する作品を楽しみにしていると答える。その後、ユーザーはせとかさんとくるみさんとの雑談を提案する。

## New lines of Conversation:
せとか : ユーザーさん、くるみセンパイとお話しするのですね。どんな話題が好きなのですか？
ユーザー名 : くるみとはなしてｗ
くるみ : こんにちは、ユーザーさん。最近見た映画の感想を聞かせてくれる？
## New Summary:

チェック＆要約のスレッド

def manage_conv_thread(self, conv:Conversations):

    while not self._exit_flag:
        # 3秒おきに履歴の長さをチェックして要約が必要か判断
        time.sleep(3)
        
        # 長さチェック
        do_shrink = conv.check_current_lengh(CONV_MAX)
        if not do_shrink:
            continue
    
        # 要約実施 
        conv.shrink_messages(CONV_SUMMARIZE)

音声合成と再生

VOICEVOX ENGINEを別途立ち上げておき、AIキャラの発言を音声にして再生します。VoiceGeneratorクラスの中身はAPIを叩くなどシンプルな内容です。

メインスレッド

# ～～～～～～～～

    # chにはCharacterクラス、ai_contentには発言テキストが入ってる
    self.__voice_synthesis(ch, ai_content)

# ～～～～～～～～
def __voice_synthesis(self, ch:Character, text:str):
    """受け取ったテキストで音声合成し、得られたwavをキューに追加する。"""

    wav_path = self.voice_generator.text2voice(text, 
                            str(time.time()), 
                            path=WAV_PATH, 
                            speaker=ch.voice_speaker_id,
                            speed=ch.voice_speed,
                            pitch=ch.voice_pitch,
                            intonation=ch.voice_intonation, 
                            volume=V_VOL,
                            post=V_POST)
    
    # q_voice_playはmax_size=1にしてあるので、
    # 再生待機が1つでもあればここでメインスレッドがブロックされる。
    self.q_voice_play.put([wav_path, text, ch])

音声再生スレッド

def voice_play_thread(self, v:VoiceGenerator):

    while not (self._exit_flag and self.q_voice_play.empty()):
        """
        合成されたwavパスをキューから取り出す
        アイテムが取り出せるまで、1秒おきにチェック。
        _exit_flagがTrueかつ、キューが空になると抜ける
        """
        try:
            data = self.q_voice_play.get(timeout=1)
        except queue.Empty:
            continue

        wav_path = data[0]
        text = data[1]
        ch= data[2]

        # ボイス再生の直前にコンソール出力
        ch.console('{} : {}'.format(ch.name, text))

        # 再生
        v.play_wave(wav=wav_path, delete=True)

合成も再生もメインと切り分けて並行処理させると全体の処理速度は上がるのですが、Completionが先行しすぎて音声と内部の状態に差が出すぎてしまうので、あえて再生中は次の会話処理をブロックするようにしています。
AIキャラ2人だけならいいのですが、人間が音を聞きながら割り込もうとすると時間ズレが致命的になるので。