2nd/10_Wiki/Topics/Other/Multi-agent-System.md

id, title, category, status, canonical_id, aliases, duplicate_of, source_trust_level, confidence_score, verification_status, tags, raw_sources, last_reinforced, github_commit, tech_stack

id	title	category	status	canonical_id	aliases	duplicate_of	source_trust_level	confidence_score	verification_status	tags	raw_sources	last_reinforced	github_commit	tech_stack
wiki-2026-0508-multi-agent-system	Multi-agent System	10_Wiki/Topics	verified	self	MAS 멀티에이전트 Agent Swarm Agentic Systems	none	A	0.9	applied	ai agents llm orchestration distributed		2026-05-10	pending	language framework python claude-agent-sdk

Multi-agent System

매 한 줄

"매 specialization × coordination > monolith". Multi-agent system 은 여러 autonomous agent 가 message passing / shared state 로 협업해 single agent 보다 큰 task 해결. 2026 LLM 시대에 Claude Agent SDK, OpenAI Swarm, LangGraph, AutoGen 등이 표준 framework.

매 핵심

매 architecture pattern

• Orchestrator-worker: 1 lead agent + N specialist worker. 매 Anthropic 의 research agent 패턴.
• Peer-to-peer: 모든 agent equal, message bus 로 통신.
• Hierarchical: layered supervisor tree.
• Blackboard: shared memory 기반 indirect coordination.

매 communication

• Function calling / tool use.
• Structured message (JSON schema).
• Shared filesystem / vector DB.
• A2A (Agent-to-Agent) protocol (2025 Anthropic spec).

매 응용

1. Research / report generation (parallel search + synthesis).
2. Software engineering (planner + coder + tester).
3. Customer support routing.
4. Game NPC behavior.

💻 패턴

Orchestrator-worker (Claude Agent SDK)

from anthropic import Anthropic

client = Anthropic()

def spawn_worker(task: str, system: str) -> str:
    resp = client.messages.create(
        model="claude-opus-4-7",
        max_tokens=4096,
        system=system,
        messages=[{"role": "user", "content": task}],
    )
    return resp.content[0].text

def orchestrate(query: str):
    plan = spawn_worker(
        f"Decompose into 3 sub-tasks: {query}",
        "You are a research planner. Output JSON list.",
    )
    subtasks = parse_plan(plan)
    results = [spawn_worker(t, "You are a domain expert.") for t in subtasks]
    return spawn_worker(
        f"Synthesize: {results}",
        "You are an editor. Merge into a coherent report.",
    )

Tool-use loop

def agent_loop(messages, tools, max_iter=10):
    for _ in range(max_iter):
        resp = client.messages.create(
            model="claude-opus-4-7",
            tools=tools,
            messages=messages,
            max_tokens=4096,
        )
        messages.append({"role": "assistant", "content": resp.content})
        if resp.stop_reason == "end_turn":
            return resp
        for block in resp.content:
            if block.type == "tool_use":
                result = execute_tool(block.name, block.input)
                messages.append({
                    "role": "user",
                    "content": [{
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result,
                    }],
                })

Shared state via filesystem

import json, fcntl
from pathlib import Path

def shared_write(path: Path, key: str, value):
    with open(path, "r+") as f:
        fcntl.flock(f, fcntl.LOCK_EX)
        state = json.load(f)
        state[key] = value
        f.seek(0); f.truncate()
        json.dump(state, f)
        fcntl.flock(f, fcntl.LOCK_UN)

LangGraph state machine

from langgraph.graph import StateGraph, END

def planner(state): return {"plan": llm_plan(state["query"])}
def executor(state): return {"result": run_steps(state["plan"])}
def critic(state):
    if quality_score(state["result"]) < 0.7:
        return {"next": "planner"}
    return {"next": END}

g = StateGraph(dict)
g.add_node("plan", planner)
g.add_node("exec", executor)
g.add_node("crit", critic)
g.add_edge("plan", "exec")
g.add_edge("exec", "crit")
g.add_conditional_edges("crit", lambda s: s["next"])

Parallel agent fan-out

import asyncio

async def parallel_search(queries: list[str]) -> list[str]:
    tasks = [asyncio.to_thread(spawn_worker, q, "Researcher") for q in queries]
    return await asyncio.gather(*tasks)

Critic-actor consensus

def consensus(question: str, n_agents=3) -> str:
    answers = [spawn_worker(question, f"Expert #{i}") for i in range(n_agents)]
    return spawn_worker(
        f"Q: {question}\nAnswers:\n" + "\n".join(answers) +
        "\nReturn consensus + dissent.",
        "You are a meta-reviewer.",
    )

매 결정 기준

상황	Approach
Clear task decomposition	Orchestrator-worker
Open-ended exploration	Peer-to-peer + blackboard
Quality-critical	Critic-actor + consensus
Latency-critical	Parallel fan-out
Stateful workflow	LangGraph / state machine

기본값: Orchestrator-worker + tool use loop.

🔗 Graph

• 부모: Distributed Systems
• 변형: Agent Orchestration · Swarm_Intelligence
• 응용: LangGraph
• Adjacent: Tool Use · Function Calling

🤖 LLM 활용

언제: complex task decomposition, parallel research, multi-step pipeline. 언제 X: simple single-shot Q&A — overhead 만 추가.

❌ 안티패턴

• Over-decomposition: 너무 많은 agent → coordination overhead 폭증.
• No termination condition: infinite loop 위험.
• Shared mutable state without lock: race condition.
• Tool sprawl: 한 agent 에 50+ tools — selection 정확도 폭락.

🧪 검증 / 중복

• Verified (Anthropic Multi-agent Research 2024, OpenAI Swarm, LangGraph 0.3).
• 신뢰도 A.

🕓 Changelog

날짜	변경
2026-05-08	Phase 1
2026-05-10	Manual cleanup — orchestrator/tool-use/LangGraph 패턴