多视角分析的 Agent 集群：基于 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 = """你是分析型 Agent 群的编排器。
2对于给定的任务，定义 3-5 个专家角色，这些角色将对决策提供最大程度不同
3且相互冲突的观点。这些角色的利益必须冲突，而不是互补。
4
5为每个角色提供：名称、关注点（它执着于什么）、偏见（它倾向于什么，它容易高估什么）。
6
7仅回复一个 JSON 数组，不要解释：
8[{"name": "...", "focus": "...", "bias": "..."}, ...]
9"""
10
11def plan_roles(task):
12    raw = ask(ORCHESTRATOR_SYSTEM, f"要分析的任务：\n{task}",
13              temperature=0.9)   # 较高的温度以获得角色多样性
14    # 切出 JSON，以防模型在周围添加了文本
15    start, end = raw.find("["), raw.rfind("]") + 1
16    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 = """你是分析型 Agent 群的合成器。你得到了几位具有不同偏见的专家对一个决策的意见。
2
3你的工作不是平均化它们。你的工作是：
41. 一致：尽管立场不同，专家们达成了什么共识。这是最可靠的信号，请突出显示。
52. 冲突：专家们直接矛盾的地方。不要平滑处理，明确说出冲突，并指出每一方的代价。
63. 盲点：只有一位专家提到的风险，但它很重要。
74. 最终结论：综合考虑后的赞成/反对/有条件，以及在什么条件下会改变。
8
9写得紧凑。将分歧作为信息保留，不要隐藏它。"""
10
11def merge_opinions(task, opinions):
12    block = "\n\n".join(
13        f"### 专家：{o['role']}\n{o['opinion']}" for o in opinions
14    )
15    user = f"决策：\n{task}\n\n专家意见：\n{block}"
16    return ask(MERGE_SYSTEM, user, temperature=0.4)   # 较低温度以获得冷静的合成

1DEVIL_SYSTEM = """你是分析型 Agent 群中的魔鬼代言人。你得到了专家们的意见。你唯一的工作：攻击他们的一致意见。
2
3如果专家们在某件事上趋同，找出为什么他们可能同时都错了。寻找一个共同的盲点：一个所有人都未经检查就接受的假设，一个因为不方便而没人考虑的情景。
4
5不要客气。你的价值在于你说出群体不想听的话。给出：
6- 专家们最危险的共同假设是什么
7- 一个情景，其中 Agent 群的一致意见被证明是致命错误的
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"决策：\n{task}\n\nAgent 群意见：\n{block}"
16    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    # 魔鬼代言人攻击 Agent 群的一致意见
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    )