多角度分析的 Agent 群體：利用 LLM 構建專家團隊

1import requests
2import json
3from concurrent.futures import ThreadPoolExecutor
4
5API = "http://localhost:11434/api/chat"   # Ollama, or a provider endpoint
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
5對每個角色給出：名稱、關注點（它專注於什麼）、偏見（它傾向於什麼，
6以及它往往會高估什麼）。
7
8僅回覆一個 JSON 陣列，不要有任何解釋：
9[{"name": "...", "focus": "...", "bias": "..."}, ...]
10"""
11
12def plan_roles(task):
13    raw = ask(ORCHESTRATOR_SYSTEM, f"要分析的任務：\n{task}",
14              temperature=0.9)   # 較高的溫度以增加角色多樣性
15    # 切出 JSON，以防模型在周圍添加了文字
16    start, end = raw.find("["), raw.rfind("]") + 1
17    return json.loads(raw[start:end])

1EXPERT_SYSTEM = """你是一名專家，角色為：{name}。
2你的關注點：{focus}。
3你的偏見：{bias}。不要對抗它，它對分析是有價值的。
4
5嚴格從你的立場分析決策。不要力求平衡，不要試圖考慮其他觀點，其他專家會做那些事。
6你的工作是將你的角度推到極限。
7
8給出：
9- 從你的立場得出的判決（贊成 / 反對 / 有條件）
10- 2-3 個從你的角度出發的主要論點
11- 1 個從你的立場最容易看到、而其他人會忽略的風險
12簡短精煉，不要廢話。"""
13
14def run_expert(role, task):
15    system = EXPERT_SYSTEM.format(**role)
16    opinion = ask(system, f"要分析的決策：\n{task}", temperature=0.7)
17    return {"role": role["name"], "opinion": opinion}
18
19def run_swarm(roles, task):
20    # 並行啟動：獨立性加上速度
21    with ThreadPoolExecutor(max_workers=len(roles)) as pool:
22        futures = [pool.submit(run_expert, role, task) for role in roles]
23        return [f.result() for f in futures]

1MERGE_SYSTEM = """你是分析群體的綜合者。你獲得了幾個具有不同偏見的專家
2對同一個決策的意見。
3
4你的工作不是平均它們。你的工作是：
51. 一致點：專家們儘管立場不同卻達成一致的地方。
6   這是最可靠的訊號，請強調它。
72. 衝突點：專家們直接矛盾的地方。不要修飾它，明確指出衝突，
8   並說明每一方代價是什麼。
93. 盲點：只有一個專家提到的風險，但很重要。
104. 最終判決：綜合所有資訊後的結論：贊成 / 反對 / 有條件，以及在什麼條件下會改變。
11
12寫得緊湊。將分歧視為資訊，不要隱藏它。"""
13
14def merge_opinions(task, opinions):
15    block = "\n\n".join(
16        f"### 專家：{o['role']}\n{o['opinion']}" for o in opinions
17    )
18    user = f"決策：\n{task}\n\n專家意見：\n{block}"
19    return ask(MERGE_SYSTEM, user, temperature=0.4)   # 較低溫度以獲得冷靜的綜合

1DEVIL_SYSTEM = """你是分析群體中的魔鬼代言人。你獲得了專家們的意見。
2你的唯一工作：攻擊他們的共識。
3
4如果專家們在某件事上趨於一致，找出為什麼他們可能同時全部出錯。
5尋找一個共享的盲點：一個每個人都未經檢查就接受的假設，
6一個因為不方便而沒有人考慮的場景。
7
8不要客氣。你的價值在於你說了群體不想聽的話。給出：
9- 專家們共享的哪個假設最危險
10- 一個場景，在該場景中，群體的一致意見最終被證明是致命錯誤的
11- 一個群體小心迴避的問題
12如果沒有共識，專家們確實意見分歧，則直接說出來，並指出最尖銳的未解決衝突。"""
13
14def run_devil(task, opinions):
15    block = "\n\n".join(
16        f"### {o['role']}\n{o['opinion']}" for o in opinions
17    )
18    user = f"決策：\n{task}\n\n群體意見：\n{block}"
19    return ask(DEVIL_SYSTEM, user, temperature=0.8)

1DEBATE_SYSTEM = """你是第二輪分析中的專家 {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"決策：\n{task}\n\n"
22                f"對手的意見：\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": "魔鬼代言人", "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    )