マルチアングル分析のためのエージェント群：LLM を活用した専門家チームの構築

1import requests
2import json
3from concurrent.futures import ThreadPoolExecutor
4
5API = "http://localhost:11434/api/chat"   # Ollama、またはプロバイダーのエンドポイント
6MODEL = "qwen2.5:32b"
7
8def ask(system, user, temperature=0.7):
9    resp = requests.post(API, json={
10        "model": MODEL,
11        "messages": [
12            {"role": "system", "content": system},
13            {"role": "user", "content": user},
14        ],
15        "temperature": temperature,
16        "stream": False,
17    }, timeout=120)
18    resp.raise_for_status()
19    return resp.json()["message"]["content"]

1ORCHESTRATOR_SYSTEM = """あなたは分析スワームのオーケストレーターです。
2タスクに対して、決定に対して最大限に異なり、矛盾した見解を与える 3 〜 5 のエキスパート役割を定義してください。
3役割は互いの利益で衝突する必要があり、補完し合ってはいけません。
4各役割について以下を提供してください：name（名前）、focus（何に固執するか）、bias（何に偏っているか、何を過大評価しがちか）。
5説明なしで JSON 配列のみを返してください：
6[{"name": "...", "focus": "...", "bias": "..."}, ...]
7"""
8
9def plan_roles(task):
10    raw = ask(ORCHESTRATOR_SYSTEM, f"Task to analyze:\n{task}",
11              temperature=0.9)   # 役割の多様性のために高い温度を設定
12    # モデルが周囲にテキストを追加した場合に備えて JSON を切り出す
13    start, end = raw.find("["), raw.rfind("]") + 1
14    return json.loads(raw[start:end])

1EXPERT_SYSTEM = """あなたは役割 {name} を持つエキスパートです。
2あなたの焦点：{focus}。
3あなたのバイアス：{bias}。それと戦わないでください。それが分析におけるあなたの価値です。
4
5決定を厳密にあなたの立場から分析してください。バランスを取ろうとせず、他の視点を考慮しようとしないでください。他のエキスパートがそれを行います。あなたの仕事は自分の角度を限界まで押し進めることです。
6
7以下を提供してください：
8- あなたの立場からの評決（賛成 / 反対 / 条件付き）
9- あなたの角度から具体的に 2 〜 3 の主な論点
10- あなたの立場から最も見えやすく、他の人が見逃すリスク 1 つ
11簡潔に、厳しく、無駄なものはなし。"""
12
13def run_expert(role, task):
14    system = EXPERT_SYSTEM.format(**role)
15    opinion = ask(system, f"Decision to analyze:\n{task}", temperature=0.7)
16    return {"role": role["name"], "opinion": opinion}
17
18def run_swarm(roles, task):
19    # 並列起動：独立性と速度の両立
20    with ThreadPoolExecutor(max_workers=len(roles)) as pool:
21        futures = [pool.submit(run_expert, role, task) for role in roles]
22        return [f.result() for f in futures]

1MERGE_SYSTEM = """あなたは分析スワームの統合者です。異なるバイアスを持つ複数のエキスパートの意見が一つの決定について与えられています。
2
3あなたの仕事は平均化することではありません。あなたの仕事は：
41. 一致：異なる立場にもかかわらずエキスパートが同意した点。これが最も信頼できるシグナルなので、強調すること。
52. 対立：エキスパートが直接矛盾する点。それを曖昧にせず、対立を明示し、各立場のコストを述べること。
63. 盲点：一人のエキスパートだけが挙げたが、重要なリスク。
74. すべてを考慮した最終評決：賛成 / 反対 / 条件付き、および条件が変わる場合。
8
9密度高く書け。不一致を情報として保持し、隠すな。"""
10
11def merge_opinions(task, opinions):
12    block = "\n\n".join(
13        f"### Expert: {o['role']}\n{o['opinion']}" for o in opinions
14    )
15    user = f"Decision:\n{task}\n\nExpert opinions:\n{block}"
16    return ask(MERGE_SYSTEM, user, temperature=0.4)   # 冷静な統合のために低い温度を設定

1DEVIL_SYSTEM = """あなたは分析スワームの悪魔の代弁者です。エキスパートの意見が与えられています。あなたの唯一の仕事：彼らの合意を攻撃すること。
2
3もしエキスパートが何かに収束した場合、なぜ彼ら全員が同時に間違っている可能性があるかを見つけてください。共通の盲点を探してください：誰も確認せずに受け入れた前提、誰も都合が悪いので考慮しなかったシナリオ。
4
5礼儀正しくする必要はありません。あなたの価値は、グループが聞きたくないことを言うことです。以下を提供してください：
6- エキスパートの最も危険な共通の前提
7- スワームの全会一致の意見が致命的に間違っていることが判明するシナリオ
8- グループが注意深く避けた質問
9もし合意がなく、エキスパートが本当に意見を異にしている場合は、それを率直に述べ、最も鋭い未解決の対立を指摘してください。"""
10
11def run_devil(task, opinions):
12    block = "\n\n".join(
13        f"### {o['role']}\n{o['opinion']}" for o in opinions
14    )
15    user = f"Decision:\n{task}\n\nSwarm opinions:\n{block}"
16    return ask(DEVIL_SYSTEM, user, temperature=0.8)

1DEBATE_SYSTEM = """あなたは分析の第 2 ラウンドのエキスパート {name} です。
2あなたの元の立場：
3{own_opinion}
4
5今、他のエキスパートの意見を見ています。圧力に屈してはいけませんが、強力な議論を無視してもいけません。以下を提供してください：
6- 他人の議論があなたの立場に本当に影響を与えた場合、正直に認めること
7- どこで自分の立場を守り、なぜ彼らの反論が弱いか
8- 討論後に評決を変更したかどうか、変更した場合はどのように変更したか
9簡潔に。これは最初の意見の繰り返しではなく、反対者への反応です。"""
10
11def debate_round(roles, task, opinions):
12    others_map = {}
13    for o in opinions:
14        others = "\n\n".join(
15            f"### {x['role']}\n{x['opinion']}" for x in opinions if x is not o
16        )
17        others_map[o["role"]] = others
18
19    def rebut(o):
20        system = DEBATE_SYSTEM.format(name=o["role"], own_opinion=o["opinion"])
21        user = (f"Decision:\n{task}\n\n"
22                f"Opponents' opinions:\n{others_map[o['role']]}")
23        return {"role": o["role"], "opinion": ask(system, user, temperature=0.6)}
24
25    with ThreadPoolExecutor(max_workers=len(opinions)) as pool:
26        return list(pool.map(rebut, opinions))

1def analyze(task, debate=True):
2    print("オーケストレーターが役割を選んでいます...")
3    roles = plan_roles(task)
4    for r in roles:
5        print(f"  - {r['name']}: {r['focus']}")
6
7    print(f"\n{len(roles)} 人のエキスパートを並行して起動しています...")
8    opinions = run_swarm(roles, task)
9    for o in opinions:
10        print(f"\n[{o['role']}]\n{o['opinion']}")
11
12    # オプションの討論ラウンド：エキスパートが互いに反論
13    if debate:
14        print("\n討論ラウンド：エキスパートが互いに反論しています...")
15        opinions = debate_round(roles, task, opinions)
16
17    # 悪魔の代弁者がスワームの合意を攻撃
18    print("\n悪魔の代弁者が合意の亀裂を探しています...")
19    devil = run_devil(task, opinions)
20    print(f"\n[悪魔の代弁者]\n{devil}")
21
22    # マージが結論と代弁者の攻撃を統合
23    print("\nマージが結論を調整しています...")
24    opinions_plus = opinions + [{"role": "Devil's advocate", "opinion": devil}]
25    verdict = merge_opinions(task, opinions_plus)
26    print(f"\n=== 最終評決 ===\n{verdict}")
27    return verdict
28
29if __name__ == "__main__":
30    analyze(
31        "無料プランを廃止し、製品を完全有料化（14 日間のトライアル付き）にしたい。"
32        "実行すべきか？"
33    )