人の話を最後まで聞かないAIを作った

「あれって人の話を最後まで聞かない人じゃない?」
「わー本当だ。人の話を最後まで聞かない人だ」
「まだ相手が話終えていないのに、自分が何話そうか考えてるんだよな」
「そうそう。で、自分の話す番が来たら、待ってましたとばかりに話し始めるんだよな」
「わかる。しかも話の流れガン無視で、自分の言いたいことだけ話すから会話が成立してないんだよ」
「で、こっちが『あ、でもさ』って言おうとすると、もう次の話題に行ってるんだよな」
「そうそう。で、最終的に『え、今なんの話してたっけ？』ってなるやつ」
「あるある。で、結局そいつだけがスッキリして終わるんだよな」
「そういうやつに限って『話聞くのうまいね』とか言われたがるんだよな」
「わかるわー。聞くどころか、ほぼ一人で喋ってるのにな」

https://www.youtube.com/@ore.fiction

そんな感じのAIを作りました

最後までユーザーの話を聞かないクラス

ユーザーの話はgeneratorとして、細切れのチャンク（テキスト）を渡すようにします。
チャンクは_process_messages内において、textに集積されます。
このプロセスをt1としています。

一方で、LLMがユーザーの話（self.text）を受け取り、model_classの項目を受け取ったか判断します。
そして、その結果はstructured_outputに格納しています
これらはt2というプロセスで進めています。

structured_outputがすべて埋まれば、_stop_processingフラグを立てて、上記のプロセスを終了させます。

from pydantic import BaseModel, Field
import threading
import time
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, AIMessage, HumanMessage
from typing import Callable, Generator

class Structurer():
    def __init__(self, base_model: type[BaseModel], generator: Callable[[], Generator[str | None, None, None]]):
        self.model_class = base_model
        self.structured_output = base_model()
        self.generator = generator
        
        self.text = ""
        self.llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
        
        self._stop_processing = threading.Event()
        self.t1 = threading.Thread(target=self._process_messages)
        self.t2 = threading.Thread(target=self._repeat_analyze)

    def __enter__(self):
        self.t1.start()
        self.t2.start()
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self._stop_processing.set()
        self.t1.join()
        self.t2.join()
        return self
    
    def result(self):
        """構造体のデータを整形して返す"""

        return self.llm.invoke([
            SystemMessage(content="構造体のデータを渡しますので、そのデータを整形してください。"),
            HumanMessage(content=self.structured_output.model_dump_json())
        ])
    
    def _process_messages(self):
        """メッセージ処理用のスレッド（内部使用）"""
        try:
            for chunk in self.generator():
                if self._stop_processing.is_set():
                    print("最後まで話を聞かずに終了します。受け取ったテキスト：",self.text[-30:])
                    break

                if chunk:
                    self.text += chunk
                else:
                    print("Noneを受け取ったため終了します。受け取ったテキスト：",self.text[-30:])
                    break

            # ジェネレータが終了したらストップフラグを立てる
            if not self._stop_processing.is_set():
                self._stop_processing.set()

        except Exception as e:
            print(f"メッセージ処理中にエラーが発生: {e}")
            self._stop_processing.set()
    
    def analyze(self):
        """現在のテキストを分析"""
        # テキストが空ならスキップ
        if not self.text:
            return
        
        system_prompt = """
            あなたは面接官です。面接官として、与えられた文章を分析し、以下の構造体（model_class）に定めた情報を抽出してください。
            ユーザーの答えはストリームで流れてくるので、話はまだ途中の可能性があります。
            そのため、ユーザーの答えの中に明確な情報がない場合は、その項目にNoneを返してください。

            model_class:{model_class}
            """
        
        try:
            result = self.llm.with_structured_output(self.model_class).invoke([
                SystemMessage(content=system_prompt),
                AIMessage(content="自己PRをしてください"),
                HumanMessage(content=self.text)
            ])

            self.structured_output = self.model_class.model_validate(result)

        except Exception as e:
            print(f"分析中にエラーが発生: {e}")
    
    def _repeat_analyze(self):
        """分析を繰り返し実行するスレッド（内部使用）"""
        while not self._stop_processing.is_set():
            self.analyze()
            
            # 分析結果をチェック
            values = [getattr(self.structured_output, field) for field in self.structured_output.model_fields]
            print(f"分析状況: {values}\n")
            if all(values):
                self._stop_processing.set()
                break
            
            # 少し待ってから再分析
            time.sleep(0.1)

ユーザーの話

就活の面接を想定しています。
読んでの通り、途中から全く関係のない話を入れています。

本当は音声で話するとこまで実装すればいいのですが、今回はgeneratorで仮実装です。

# サンプルデータ
sample_messages = """私はリーダーシップを発揮できる人材です。
    学生時代にサークル長として運営に携わった際に、リーダーシップを養うことができました。
    サークル長を務めていたフットボールサークルでは、練習場所や時間が取れないことや、連携を取り切れていないことが問題でした。
    そこで、大学生側に掛け合い週に2回の練習場所を確保し、時間を決め活動するようにメンバーに声掛けを行いました。
    さらに週末明けに今週の活動の詳細をメンバーに配信することで連携強化に努めた結果、サークル加入率を前年度の3倍まで伸ばすことができました。
　　日常のドラマを演じるYouTuberグループ「俺フィク」。半年ほど前から現在のメンバー4人での活動を発表し、早くもチャンネル登録者数8万人を突破している
    サブチャンネルは「裏フィク」で検索！
    メンバーシップもあります！
    未公開動画多数！要チェック！
    """

sample_messages = sample_messages.replace(" ","").replace("\n","")

# 小さなチャンクに分割
chunks = [sample_messages[i:i+5] for i in range(0, len(sample_messages), 5)]

def generate_messages():
    for chunk in chunks:
        yield chunk
        time.sleep(0.2)
    yield None

聞きたい話

BaseModelを継承したクラスで定義しています。

今回はこの3つ（issue、solution、effect）を聞き出したところで、ユーザーの話を聞くのをやめます

class Info(BaseModel):
    issue: str | None = Field(default=None, description="難しかったことを記述する")
    solution: str | None = Field(default=None, description="その時の解決法を記述する")
    effect: str | None = Field(default=None, description="その時の成果を記述する")

実行手順

上記のコードと、以下のコードを一つのファイルに書いて実行してみてください

if __name__ == "__main__":
    with Structurer(Info, generate_messages) as s:

        while not s._stop_processing.is_set():
            time.sleep(0.2)

    print(s.result().content)

結果

最初らへんはまだユーザーの話がissue、solution、effectの内容に触れていないため、
分析状況: [None, None, None]ですが、
最後は分析状況にデータが入っています。

分析状況: [None, None, None]

分析状況: [None, None, None]

分析状況: [None, None, None]

分析状況: ['練習場所や時間が取れなかったこと', 'サークルのメンバーと話し合い、練習スケジュールを調整した', 'メンバー全員が参加できる練習が実現し、チームの士気が向上した']

この時点で受け取ったテキストを見てみると、最後まで話を聞いていないことがわかります

最後まで話を聞かずに終了します。
受け取ったテキスト： 取り切れていないことが問題でした。そこで、大学生側に掛け合い