複数のローカル生成AI同士で議論させる

LM StudioやOpenWebUIを組み合わせて、外部と通信しないローカル生成AIでいろいろやっている。
参照：https://qiita.com/Qapla/items/167166f53e4580823e74

ChatGPTなどに特定の人物を演じるように指示をして議論をさせるネタはよく見る。
が、ChatGPT、Gemmaなど複数の生成AIで会話をさせたらどうなるんだろう？
と思ってスクリプトを作ってみた。

完成イメージ

フォルダ構成

4つのスクリプトを同じフォルダに置く。
加えて「agents」というフォルダを作って中に話し合わあわせたい生成AIの数だけテキストファイルを置いておく（後述）

• main.py
• debate.py
• agent.py
• utils.py
• agents/*.txt

main.py

主処理をするスクリプト。OpenWebUIに対して生成AIサーバ（Webサーバ）のように振る舞う。

import glob
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from typing import List

from utils import load_agent_from_file, parse_human_input, sse_chunk, sse_done
from debate import DebateSession

LM_STUDIO_URL = "http://localhost:4191/v1"

app = FastAPI()
session = None  # DebateSession インスタンス

class Message(BaseModel):
    role: str
    content: str

class ChatRequest(BaseModel):
    model: str
    messages: List[Message]


@app.post("/v1/chat/completions")
def chat(req: ChatRequest):
    global session

    user_text = req.messages[-1].content.strip()
    command, payload = parse_human_input(user_text)

    def event_stream():
        global session
        agents_count = 0
        
        # 新たな議論
        if command == "NEW_TOPIC":
            session = None
            yield sse_chunk("初期化しました。\n\n")
            yield sse_done()
            return StreamingResponse(event_stream(), media_type="text/event-stream")
            


        # 初回のみ DebateSession 作成
        if session is None:
        
            agent_files = glob.glob("agents/*.txt")
            agents_count = len(agent_files)
            agents = [load_agent_from_file(f, LM_STUDIO_URL) for f in agent_files]
            session = DebateSession(agents)
            
            # 入室者一覧を表示
            yield sse_chunk("以下の人が入室しました\n\n")
            for agent in agents:
                yield sse_chunk(f"- {agent.name}\n\n")

        
            # 初期テーマを human_input などからセット
            initial_context = f"次のテーマについて議論してください：{payload}"
            session.set_initial_context(initial_context)
        
        # 人間の入力を求めずに繰り返す回数
        repeat_count = 1
        
        # instruction 作成
        instruction = payload

        if command == "REPEAT":
            instruction = None
            repeat_count = int( payload )
            
        # repeat_count回繰り返す
        for _ in range(repeat_count):
        
            # agentsを呼び出す
            for speaker, chunk in session.one_round(command, instruction):
                yield sse_chunk(chunk)

        # 1往復終了後は必ず human待ちに戻す
        yield sse_done()

    return StreamingResponse(event_stream(), media_type="text/event-stream")


@app.get("/v1/models")
def list_models():
    return {
        "object": "list",
        "data": [
            {
                "id": "ai-debate",
                "object": "model",
                "created": 0,
                "owned_by": "local"
            }
        ]
    }

debate.py

複数の生成AIを制御するスクリプト
会話ログの保持などを行っている

class DebateSession:
    def __init__(self, agents, max_history=20):
        self.agents = agents
        self.max_history = max_history
        self.history = []  # 会話ログ
        self.initial_context = ""  # 最初のテーマや system_prompt を保持

    def set_initial_context(self, text):
        self.initial_context = text
        self.history.append(text)

    def one_round(self, command, human_instruction=None):
    
        name = None
        instruction = human_instruction
        
        # 先頭で名前が呼ばれていれば、そのAIのみ呼び出す
        if human_instruction:
        
            # 名前で始まっていれば、その名前を name にとっておく（特定のAIを呼び出すため）
            for agent in self.agents:
                if human_instruction.startswith(agent.name):
                    name = agent.name
                    continue

            # 議長の発言とみなす
            if command == "NEW_TOPIC" or command == "COMMENT":
                instruction = "議長：" + human_instruction
            
            # 命令
            elif command == "ORDER":
                instruction = "命令です：" + human_instruction
            
        for agent in self.agents:
        
            # 名前がある場合、一致しなければスキップ
            if name is not None and name != agent.name:
                continue
        
            # 履歴は初期文＋直近 max_history 件
            recent_history = [self.initial_context] + self.history[-self.max_history:]
            messages = [{"role": "user", "content": m} for m in recent_history]

            if instruction:
                messages.append({"role": "user", "content": instruction})

            full_text = ""
            for chunk in agent.stream_reply(messages):
                full_text += chunk
                yield agent.name, chunk + "\n\n"

            # 発言を history に追加
            self.history.append(full_text)

agent.py

生成AI1つに対応するインスタンスのスクリプト

from langchain_openai import ChatOpenAI

class Agent:
    def __init__(self, name, model, temperature, system_prompt, base_url):
        self.name = name
        self.system_prompt = system_prompt
        self.llm = ChatOpenAI(
            model=model,
            temperature=temperature,
            base_url=base_url,
            api_key="lm-studio",
            streaming=True
        )

    def stream_reply(self, messages):
        full = [{"role": "system", "content": self.system_prompt}] + messages
        buffer = ""
        for chunk in self.llm.stream(full):
            if chunk.content:
                buffer += chunk.content
                if "\n" in chunk.content:  # 発言区切りを目安に1発言ずつ
                    yield buffer
                    buffer = ""
        if buffer:
            yield buffer

utils.py

テキストファイルをロードしてAgent のインスタンスを返す、人間の入力を加工する、などいろいろしているスクリプト
すべての生成AIに統一して渡すシステムプロンプトを持っている

import json
import uuid
from agent import Agent

COMMON_SYSTEM_RULES = """
以下のルールを厳守してください。
- 相手の発言に必ず反応する
- 反論だけでなく、質問や追加の問いかけを入れて相手に考えさせる
- 同じ主張を繰り返さず、別の観点や例を出して議論を進める
- 見出しや箇条書き、Markdownは禁止
- 必ず日本語で話す
- 「命令です」で始まる指示には必ず従う
- **他者（議長・参加者など）の発言を一切生成してはいけない**
- 結論、要約、まとめを書いてはいけない
- あなたの名前は「{name}」です
- 必ず最初に「*」2つで囲んで太文字にした「{name}：」と書いてから話す
- **{name}としての発言を1回分のみ生成する**
""".strip()



def load_agent_from_file(path: str, base_url: str):
    with open(path, encoding="utf-8") as f:
        lines = [l.rstrip() for l in f if l.strip()]

    model = lines[0]
    temperature = float(lines[1])
    name = lines[2]
    persona = "\n".join(lines[3:])

    system_prompt = (
        persona
        + "\n\n"
        + COMMON_SYSTEM_RULES.format(name=name)
    )

    return Agent(
        name=name,
        model=model,
        temperature=temperature,
        system_prompt=system_prompt,
        base_url=base_url,
    )


def parse_human_input(text: str):
    text = text.strip()

    # 新たな議論を開始する
    if text.startswith("新たな議論"):
        return "NEW_TOPIC", text.replace("新たな議論", "").strip()

    # 発言なしで継続する
    if text == "続けて" or text == "." or text == "。":
        return "CONTINUE", None

    # 強制的な指示
    if text.startswith("命令です"):
        return "ORDER", text.replace("命令です", "").strip()

    # 数字のみ：人間の入力を求めずに指定した回数、会話を継続
    if text.isdigit():
        return "REPEAT", text

    return "COMMENT", text



def sse_chunk(text: str):
    payload = {
        "id": f"chatcmpl-{uuid.uuid4().hex}",
        "object": "chat.completion.chunk",
        "choices": [
            {
                "delta": {"content": text}
            }
        ],
    }
    return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"


def sse_done():
    return "data: [DONE]\n\n"

agentsフォルダ

UTF-8 で以下の内容を記述したファイルを置いておく
1行目：使用する生成AI（LM Studio内のモデル名、他のファイルに書いたモデル名と同じでも違っても良い）
2行目：temperature（0.1～1.0で設定。数字が大きいほうが回答の幅が広くなる）
3行目：表示上の名前（好きな名前で良い）
4行目以降：キャラクタ設定など

例えば以下のような３ファイルを置く。

例：agents/01.txt

google/gemma-3n-e4b
0.5
グローイン
あなたはマクドナルドが大好きで体にいいと信じている人です。
マクドナルドは栄養バランスが取れており、工夫すれば健康的だと考えています。
マクドナルドが体に悪いと信じている人と議論してください。

例：agents/02.txt

qwen/qwen3-vl-8b
0.8
ガンダルフ
あなたはマクドナルドが体に悪いと信じている人です。
脂質や塩分が多く、健康被害を引き起こすと考えています。
マクドナルドが体に良いと信じている人と議論してください。

例：agents/03.txt

google/gemma-3n-e4b
0.8
エルロンド
あなたはマクドナルドが体に良いとも悪いとも考えていない中立的な人です。
議論に参加する人たちが合意できる点を探して、建設的に議論を進めようとします。

スクリプトの使い方

以下のように呼び出すと、待機状態になる。

uvicorn main:app --host <待ち受けるIPアドレス> --port <ポート番号>

例えば以下のような感じ。
uvicorn main:app --host 0.0.0.0 --port 8000
引数は以下の２つ。

待ち受けるIPアドレス
特に制限をしないなら「0.0.0.0」で良い

ポート番号
OpenWebUIからの通信を受け付けるポート番号を決める。
もちろん、OpenWebUIのポート番号、LM Studioのポート番号 とは異なるものにする。

起動すると以下のように「running」と表示されれば準備完了だ。

加えて、OpenWebUIの設定と起動が必要です。
以下の記事を参考にしてください。

使い方

OpenWebUIに入り、議論のスタートとなる指示を出す
(モデルはテキストファイルに書いているものが使用されます。OpenWebUI上では「ai-debate」の1種類しか選べません）

生成AIは基本的にファイル名順に１度ずつ発言します。発言が一巡すると人間の入力待ちになります。
ただし、特定の生成AIを指名して発言させたり、回数を指定して生成AIに複数ターン会議をさせたりすることもできます。

以下のどれかを入力します。

議論を続けさせるとき「.」

「.」だけを入力してEnterを押すと、人間の発言なしで生成AIの発言ターンを1度行います。

※「続けて」「。」と入力しても同じ動きになります

発言せず議論を数回続けさせるとき「数字」

「2」「4」のように数字だけを入力してEnterを押すと、人間の発言なしで指定した数字の回数だけターンを繰り返します。

特定の生成AIに発言をさせるとき「名前」を先頭に書く

「エルロンド、あなたの意見を述べてください」「グローイン、エルロンドの意見をどう思いますか？」のように名前（agentsフォルダ内のファイルの3行目に書いた名前）を先頭に書いて入力すると、指名された生成AIのみが呼ばれます

議長として議論の流れをハンドリングしやすくなります。

議論に議長として加わるとき

「〇〇の可能性は考えましたか？」などと書き込むと、議長の意見として会議に送られた後、ターンが1度進みます。

この意見が採用されるかどうかは生成AI次第です。

議論に指示を与えるとき「命令です」キーワード

「命令です」というキーワードを付けて指示をします。

例えば「命令です経営陣として議論せず、1消費者という立場で議論します」と書きます。

「命令です」というキーワードには必ず従うように生成AIには指示しています。（utils.py 6行目～参照）

新しい議論を始めるとき「新たな議論」キーワード

「新たな議論」と書き込むと、生成済みの生成AIのインスタンスが破棄され、次回ターンで新規作成されます

スクリプトの調整

• utils.py 6行目～：システムプロンプト
  上述の通り「utils.py」にすべての生成AIに共通の指示（システムプロンプト）が書かれている。
  １人が話しすぎて会議にならない、他のAIの質問を無視する、など問題があるときは個別のテキストファイルに指示を書いても良いが、システムプロンプトとして指示するなら「utils.py」に書けば良い

• debate.py 5行目：参考にする過去の会話数
  過去の議論を持ちすぎると処理が重くなるので5つに制限している
  会議を進めるうちに、それまでの会話を忘れているなと思ったら増やすと効果があるはず

動作させた様子

動作させると以下のようなかんじ
パソコンの性能によるが、まぁまぁの速度で動きます

OpenWebUIにエラー表示が出たら

前の記事と共通する部分が多いので、そちらをご覧ください。