2nd/10_Wiki/Topic_Programming/From_Other/Multi-agent-System.md

---
id: wiki-2026-0508-multi-agent-system
title: Multi-agent System
category: 10_Wiki/Topics
status: verified
canonical_id: self
aliases: [MAS, 멀티에이전트, Agent Swarm, Agentic Systems]
duplicate_of: none
source_trust_level: A
confidence_score: 0.9
verification_status: applied
tags: [ai, agents, llm, orchestration, distributed]
raw_sources: []
last_reinforced: 2026-05-10
github_commit: pending
tech_stack:
  language: python
  framework: 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)
```python
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
```python
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
```python
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
```python
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
```python
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
```python
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|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 패턴 |