다각도 분석을 위한 에이전트 스웜: 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작업에 대해, 결정에 대해 최대한 다르고 상충되는 관점을 제공할
33-5명의 전문가 역할을 정의하십시오. 역할은 서로를 보완하는 것이 아니라
4이해관계가 충돌해야 합니다.
5
6각 역할에 대해 다음을 제공하십시오: 이름, 초점(무엇에 집중하는지), 편향(무엇을
7선호하고, 무엇을 과대평가하는 경향이 있는지).
8
9JSON 배열로만 응답하고, 설명은 없습니다:
10[{"name": "...", "focus": "...", "bias": "..."}, ...]
11"""
12
13def plan_roles(task):
14    raw = ask(ORCHESTRATOR_SYSTEM, f"분석할 작업:\n{task}",
15              temperature=0.9)   # 역할 다양성을 위한 높은 온도
16    # 모델이 주변에 텍스트를 추가한 경우 JSON을 잘라냅니다
17    start, end = raw.find("["), raw.rfind("]") + 1
18    return json.loads(raw[start:end])

1EXPERT_SYSTEM = """당신은 역할: {name}을 가진 전문가입니다.
2당신의 초점: {focus}.
3당신의 편향: {bias}. 그것과 싸우지 마십시오. 그것이 분석에 대한 당신의 가치입니다.
4
5결정을 당신의 입장에서 엄격하게 분석하십시오. 균형을 잡으려 하지 말고,
6다른 관점을 고려하려 하지 마십시오. 다른 전문가들이 그렇게 할 것입니다.
7당신의 임무는 당신의 각도를 한계까지 밀어붙이는 것입니다.
8
9다음을 제공하십시오:
10- 당신의 입장에서의 판결 (찬성 / 반대 / 조건부)
11- 당신의 각도에서 구체적으로 2-3개의 주요 주장
12- 당신의 입장에서 가장 잘 보이고 다른 사람들은 놓칠 1가지 위험
13짧고 강하게, 군더더기 없이."""
14
15def run_expert(role, task):
16    system = EXPERT_SYSTEM.format(**role)
17    opinion = ask(system, f"분석할 결정:\n{task}", temperature=0.7)
18    return {"role": role["name"], "opinion": opinion}
19
20def run_swarm(roles, task):
21    # 병렬 실행: 독립성과 속도
22    with ThreadPoolExecutor(max_workers=len(roles)) as pool:
23        futures = [pool.submit(run_expert, role, task) for role in roles]
24        return [f.result() for f in futures]

1MERGE_SYSTEM = """당신은 분석 스웜의 종합자입니다. 하나의 결정에 대한
2서로 다른 편향을 가진 여러 전문가의 의견을 받았습니다.
3
4당신의 임무는 평균을 내는 것이 아닙니다. 당신의 임무는:
51. 동의: 서로 다른 입장에도 불구하고 전문가들이 동의한 부분.
6   이것이 가장 신뢰할 수 있는 신호이며, 강조하십시오.
72. 갈등: 전문가들이 직접적으로 모순되는 부분. 순화시키지 말고,
8   갈등을 명시적으로 언급하고 각 측면의 비용이 무엇인지 말하십시오.
93. 사각지대: 오직 한 명의 전문가만 언급했지만 중요한 위험.
104. 모든 것을 고려한 최종 판결: 찬성 / 반대 / 조건부, 그리고
11   어떤 조건에서 변경되는지.
12
13밀도 있게 작성하십시오. 불일치를 정보로 유지하고 숨기지 마십시오."""
14
15def merge_opinions(task, opinions):
16    block = "\n\n".join(
17        f"### 전문가: {o['role']}\n{o['opinion']}" for o in opinions
18    )
19    user = f"결정:\n{task}\n\n전문가 의견:\n{block}"
20    return ask(MERGE_SYSTEM, user, temperature=0.4)   # 신중한 종합을 위한 낮은 온도

1DEVIL_SYSTEM = """당신은 분석 스웜의 악마의 대변인입니다. 전문가들의
2의견을 받았습니다. 당신의 유일한 임무: 그들의 합의를 공격하십시오.
3
4전문가들이 어떤 것에 수렴했다면, 그들이 모두 동시에 틀릴 수 있는
5이유를 찾으십시오. 공유된 사각지대를 찾으십시오: 모두가 확인 없이
6수용한 가정, 불편하기 때문에 아무도 고려하지 않은 시나리오.
7
8예의 바를 필요 없습니다. 당신의 가치는 집단이 듣고 싶어 하지 않는
9말을 한다는 데 있습니다. 다음을 제공하십시오:
10- 전문가들의 가장 위험한 공유 가정
11- 스웜의 만장일치 의견이 치명적으로 잘못된 것으로 판명되는 시나리오
12- 집단이 신중하게 피한 한 가지 질문
13합의가 없고 전문가들이 진정으로 의견이 다르다면, 그것을 명확히 말하고
14가장 날카롭게 해결되지 않은 갈등을 지적하십시오."""
15
16def run_devil(task, opinions):
17    block = "\n\n".join(
18        f"### {o['role']}\n{o['opinion']}" for o in opinions
19    )
20    user = f"결정:\n{task}\n\n스웜 의견:\n{block}"
21    return ask(DEVIL_SYSTEM, user, temperature=0.8)

1DEBATE_SYSTEM = """당신은 분석 2라운드의 전문가 {name}입니다.
2당신의 원래 입장:
3{own_opinion}
4
5이제 다른 전문가들의 의견을 봅니다. 압력에 굴복하지 마십시오, 그러나
6강력한 주장을 무시하지도 마십시오. 다음을 제공하십시오:
7- 다른 사람의 주장이 당신의 입장을 진정으로 타격하는 곳, 정직하게 인정하십시오
8- 당신이 선을 지키는 곳과 그들의 반대가 왜 약한지
9- 토론 후 판결을 변경했는지 여부와, 변경했다면 어떻게 변경했는지
10짧게. 이것은 첫 번째 의견의 반복이 아니라, 상대방에 대한 반응입니다."""
11
12def debate_round(roles, task, opinions):
13    others_map = {}
14    for o in opinions:
15        others = "\n\n".join(
16            f"### {x['role']}\n{x['opinion']}" for x in opinions if x is not o
17        )
18        others_map[o["role"]] = others
19
20    def rebut(o):
21        system = DEBATE_SYSTEM.format(name=o["role"], own_opinion=o["opinion"])
22        user = (f"결정:\n{task}\n\n"
23                f"상대방 의견:\n{others_map[o['role']]}")
24        return {"role": o["role"], "opinion": ask(system, user, temperature=0.6)}
25
26    with ThreadPoolExecutor(max_workers=len(opinions)) as pool:
27        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        "완전 유료화로 전환하려고 합니다. 그래야 할까요?"
33    )