@h100envy: https://x.com/h100envy/status/2077371640690672001

X AI KOLs Timeline 工具

摘要

本文介绍了如何构建一个由不同专家角色组成的LLM智能体群,它们通过争论决策来达成更好的结论,并提供了编排器、专家和合并组件的完整Python代码。

https://t.co/quSCqPM3Iv
查看原文
查看缓存全文

缓存时间: 2026/07/16 04:05

多角度分析:从LLM构建专家团队

不是为了速度。而是为了让多个不同视角的智能体围绕一个决策进行争论,并得出比任何一个单独智能体都更好的结论。本文提供编排器、专家和合并的完整代码。

当你用一个模型评估一个决策时,它给出一个视角,通常是平均且谨慎的。它倾向于同意、平滑、寻求平衡。这就是问题所在:一个重要的决策不能由一个平均化的视角来评估,必须从不同侧面进行攻击。

一个智能体群在结构上解决了这个问题。你创建多个专家,每个都有固定的角色和偏见:一个只考虑钱,另一个只考虑技术风险,第三个只考虑用户。他们独立分析同一决策,得出不同结论,然后你强制对这些结论进行调和。这里的价值不在于速度,而在于分歧被内建在结构里。单个智能体倾向于与自己达成群体一致,而一组角色不会。

本文展示了如何用代码构建这样一个智能体群。我们涵盖三个部分:分配角色的编排器、独立分析的专家,以及将它们调和为一个结论的合并。

架构:编排器、专家、合并

用于分析的智能体群由三个组件组成。

编排器接收任务,并决定需要哪些专家角色。对于一个产品发布评估,可能包括投资者、工程师、产品专家、安全人员。编排器自身不分析,它分配角色。

专家并行且独立地工作。每个专家看到相同的决策,但通过自己的透镜。关键在于,他们看不到彼此的结论,否则会出现趋同。独立性是产生不同视角的原因。

合并收集专家的结论并进行调和:他们在哪些方面一致,在哪些方面矛盾,最终的综合裁决是什么。这不是平均化,而是一种保留分歧作为信号的综合。

第一步:基本客户端

首先创建一个简单的模型客户端。我使用兼容OpenAI的消息格式,它适用于大多数提供商和本地Ollama。

import requests
import json
from concurrent.futures import ThreadPoolExecutor

API = "http://localhost:11434/api/chat"   # Ollama,或提供商端点
MODEL = "qwen2.5:32b"

def ask(system, user, temperature=0.7):
    resp = requests.post(API, json={
        "model": MODEL,
        "messages": [
            {"role": "system", "content": system},
            {"role": "user", "content": user},
        ],
        "temperature": temperature,
        "stream": False,
    }, timeout=120)
    resp.raise_for_status()
    return resp.json()["message"]["content"]

第二步:编排器分配角色

编排器获取任务并决定需要哪些专家。不要预先硬编码角色,让模型为特定任务选择角色,这样使智能体群通用化。要求严格JSON以便解析。

ORCHESTRATOR_SYSTEM = """你是分析智能体群的编排器。
对于给定的任务,定义3-5个专家角色,这些角色将对决策给出最大程度不同且冲突的观点。
这些角色的利益必须冲突,而不是互补。

对每个角色给出:名称、关注点(该角色执着于什么)、偏见(该角色倾向于什么,容易高估什么)。

仅回复一个JSON数组,不要解释:
[{"name": "...", "focus": "...", "bias": "..."}, ...]
"""

def plan_roles(task):
    raw = ask(ORCHESTRATOR_SYSTEM, f"要分析的任务:\n{task}",
              temperature=0.9)   # 更高温度以获得角色多样性
    # 截取JSON,以防模型在周围添加了文字
    start, end = raw.find("["), raw.rfind("]") + 1
    return json.loads(raw[start:end])

我们特意保持较高温度:我们希望得到多样且不明显的角色。提示中“角色必须冲突”这一要求是关键,如果没有它,模型会给出三个几乎相同的角色,智能体群的整个意义就丧失了。

第三步:专家并行独立分析

每个专家获得其角色和相同的决策。关键点是:他们并行运行,并且看不到彼此的结论。这里的并行不仅是为了速度,它还保证了独立性——一个专家不可能在物理上调整以适应另一个的意见。

EXPERT_SYSTEM = """你是角色为{name}的专家。
你的关注点:{focus}。
你的偏见:{bias}。不要对抗它,这是你对分析的价值。

严格从你的立场分析决策。不要平衡,不要试图考虑其他观点,其他专家会做那个。你的工作是将你的角度推到极限。

给出:
- 从你的立场得出的结论(支持/反对/有条件)
- 2-3个从你角度具体出发的主要论点
- 1个从你的立场最容易看到而其他人会忽视的风险
简短有力,不说废话。"""

def run_expert(role, task):
    system = EXPERT_SYSTEM.format(**role)
    opinion = ask(system, f"要分析的决策:\n{task}", temperature=0.7)
    return {"role": role["name"], "opinion": opinion}

def run_swarm(roles, task):
    # 并行启动:独立性与速度
    with ThreadPoolExecutor(max_workers=len(roles)) as pool:
        futures = [pool.submit(run_expert, role, task) for role in roles]
        return [f.result() for f in futures]

注意专家提示:我们明确禁止它追求平衡。这反直觉,但正是关键所在。如果每个专家都试图考虑所有方面,你会得到五个相同的谨慎意见。通过迫使每个专家将其角度推到极限,你得到一个真正的光谱,然后由合并来调和。

第四步:合并调和结论

现在我们有了若干尖锐的、单方面的意见。合并将它们整合为一个结论,但不是通过平均化。它寻找专家们一致的地方(强信号)、矛盾的地方(需要决策的风险区域),以及什么比什么更重要。

MERGE_SYSTEM = """你是分析智能体群的综合者。你得到多个不同偏见的专家对同一个决策的意见。

你的工作不是平均化它们。你的工作是:
1. 一致点:专家们尽管立场不同却达成一致的地方。这是最可靠的信号,突出显示它。
2. 冲突点:专家直接矛盾的地方。不要平滑处理,明确说出冲突以及每一方的代价。
3. 盲点:只有一个专家提到的风险,但它很重要。
4. 最终结论综合考虑:支持/反对/有条件,以及在什么条件下会改变。

写密集一点。将分歧作为信息保留,不要隐藏它。"""

def merge_opinions(task, opinions):
    block = "\n\n".join(
        f"### 专家:{o['role']}\n{o['opinion']}" for o in opinions
    )
    user = f"决策:\n{task}\n\n专家意见:\n{block}"
    return ask(MERGE_SYSTEM, user, temperature=0.4)   # 较低温度用于清醒的综合

我们在合并时降低温度:如果专家应该多样化(高T),那么综合者应该清醒且一致(低T)。这里的关键指令是“不要平均化,将分歧作为信息保留。”普通的合并会把一切搞成一团浆糊,“一方面,另一方面。”一个好的合并直截了当地说:这里大家都同意,而这里有一个冲突,代价是这么多。

第四步半:魔鬼代言人反对虚假共识

有一个隐藏的危险:有时专家们达成一致不是因为决策好,而是因为所有人由于惯性而看着同一个方向。这是虚假共识,比公开冲突更危险,因为它看起来像是信心。

为此,我们添加一个特殊智能体,魔鬼代言人。它的唯一工作就是攻击共识。它看到所有专家的意见,并必须找出他们可能同时都错的原因。如果智能体群一致投票赞成,代言人会寻找一个场景证明这是灾难。

DEVIL_SYSTEM = """你是分析智能体群中的魔鬼代言人。你得到专家的意见。你的唯一工作:攻击他们的一致意见。

如果专家们在某件事上趋同,找出他们可能同时都错的原因。寻找一个共同的盲点:一个大家都接受但未核查的假设,一个因为不方便而没人考虑的场景。

不要客气。你的价值在于说出群体不想听的话。给出:
- 专家们最危险的共同假设是什么
- 一个场景,在该场景下智能体群的一致意见被证明是致命的错误
- 一个群体刻意回避的问题
如果没有一致意见,专家们确实存在分歧,请直说,并指出最尖锐的未解决冲突。"""

def run_devil(task, opinions):
    block = "\n\n".join(
        f"### {o['role']}\n{o['opinion']}" for o in opinions
    )
    user = f"决策:\n{task}\n\n智能体群意见:\n{block}"
    return ask(DEVIL_SYSTEM, user, temperature=0.8)

代言人在专家之后、合并之前运行,它的攻击与意见一起进入综合。重点是,即使是一个一致的智能体群,也至少有一个智能体有义务寻找裂缝。这很便宜(一次调用),并且在结构上打破了群体思维:共识现在必须经受攻击,而不仅仅是自然形成。

第四步六:辩论回合深化冲突

第一轮专家分析是独立的,这有利于多样性。但在收集意见后,你可以增加一轮辩论:向每个专家展示其他专家意见的摘要,让它提出反对。这深化了冲突,弱的论点被淘汰,强的论点变得更坚定。

DEBATE_SYSTEM = """你是第二轮分析中的专家{name}。
你原来的立场:
{own_opinion}

现在你看到其他专家的意见。不要屈服于压力,但也不要忽视强有力的论点。给出:
- 其他专家的论点真正打击到你的立场时,诚实承认
- 你坚持自己立场的地方,以及为什么他们的反对是脆弱的
- 辩论后你是否改变了结论,如果改变了,如何改变
简短。这不是第一轮意见的重复,而是对对手的反应。"""

def debate_round(roles, task, opinions):
    others_map = {}
    for o in opinions:
        others = "\n\n".join(
            f"### {x['role']}\n{x['opinion']}" for x in opinions if x is not o
        )
        others_map[o["role"]] = others

    def rebut(o):
        system = DEBATE_SYSTEM.format(name=o["role"], own_opinion=o["opinion"])
        user = (f"决策:\n{task}\n\n"
                f"对手的意见:\n{others_map[o['role']]}")
        return {"role": o["role"], "opinion": ask(system, user, temperature=0.6)}

    with ThreadPoolExecutor(max_workers=len(opinions)) as pool:
        return list(pool.map(rebut, opinions))

辩论回合也是并行的:每个专家同时对所有其他专家做出反应,再次没有实时趋同。辩论后的意见通常更尖锐:你可以看到哪些立场在火力下坚持住了,哪些崩溃了。正是这些经过锤炼的意见进入最终的合并。

第五步:整合在一起

def analyze(task, debate=True):
    print("编排器正在选择角色...")
    roles = plan_roles(task)
    for r in roles:
        print(f"  - {r['name']}: {r['focus']}")

    print(f"\n并行启动 {len(roles)} 位专家...")
    opinions = run_swarm(roles, task)
    for o in opinions:
        print(f"\n[{o['role']}]\n{o['opinion']}")

    # 可选的辩论回合:专家互相反驳
    if debate:
        print("\n辩论回合,专家互相反驳...")
        opinions = debate_round(roles, task, opinions)

    # 魔鬼代言人攻击智能体群的一致意见
    print("\n魔鬼代言人寻找一致意见中的裂缝...")
    devil = run_devil(task, opinions)
    print(f"\n[魔鬼代言人]\n{devil}")

    # 合并综合结论以及代言人的攻击
    print("\n合并正在综合结论...")
    opinions_plus = opinions + [{"role": "魔鬼代言人", "opinion": devil}]
    verdict = merge_opinions(task, opinions_plus)
    print(f"\n=== 最终裁决 ===\n{verdict}")
    return verdict

if __name__ == "__main__":
    analyze(
        "我们想取消免费层,只提供付费产品并附带14天试用。我们应该这样做吗?"
    )

运行这段代码,你将看到完整的流程:编排器选择角色,专家从各自角度切割真相,在辩论回合中互相争论,代言人攻击他们的一致意见,合并给出包含攻击在内的综合裁决。单个智能体面对同样的问题只会给出模糊的“取决于你的受众”,而智能体群给出结构化的分解,冲突明确,共识经过了压力测试。

什么使这个智能体群有效

三件事将一个有用的智能体群与戏剧化的智能体区分开。

角色必须冲突,而非互补。 如果你的专家是“营销人员、社交媒体专家、内容经理”,他们会给出几乎相同的答案,因为他们的利益一致。真正的价值在于利益冲突:增长vs可持续性,速度vs质量,眼前的钱vs长期的信任。利益的冲突是打开决策的关键。

专家不能看到彼此。 一旦一个专家看到另一个的意见,趋同就开始,它会调整。独立性不是实现细节,而是工作条件。并行启动免费提供了这一点。

合并不平均化,它保留冲突。 一个糟糕的综合将五个尖锐的意见变成一个无力的摘要。一个好的综合让冲突可见,因为冲突是最有价值的信息:它显示了决策真正有风险的地方,而不是人人都点头的地方。

扩展方向

这个骨架可以很自然地扩展。你可以增加一轮辩论:在第一次合并后,向专家展示摘要并让他们提出反对,这能深化冲突。你可以让最终裁判使用更强的模型来权衡论点。你可以将角色固定用于重复出现的决策类型,这样就不必每次都生成。

但基本原则不变:不同的透镜、独立分析、尊重分歧的综合。智能体群对分析有用不是因为智能体多,而是因为他们以不同的方式看待问题,并且不允许彼此滑向共同分母。找一个你独自翻来覆去考虑的决策,通过这样一个智能体群运行它。你会看到你之前没有掌握的视角。

相似文章

SwarmResearch: 编排编码代理以实现开放式发现

arXiv cs.AI

SwarmResearch 引入了一个编排器-子代理框架,其中 Shepherd Agent 引导一群 Search Agents 探索开放优化问题的多样化解决方案,在 13/15 个任务上取得了比最新方法更好或相当的结果。