1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

# 전형적인 Multi-Agent 통신 코드
from langchain.agents import AgentExecutor

class ManagerAgent:
    def __init__(self, llm):
        self.llm = llm
        self.workers = {}
    
    def delegate_task(self, task: str, worker_id: str, context: dict):
        # 문제: context가 클수록 압축이 필요함
        compressed_context = self._compress_context(context)
        
        # 여기서 정보 손실 발생
        prompt = f"""
        Task: {task}
        Context: {compressed_context}  # 핵심 정보가 이미 누락되었을 수 있음
        """
        
        return self.workers[worker_id].execute(prompt)
    
    def _compress_context(self, context: dict, max_tokens: int = 4000):
        # LLM으로 요약 -> 정보 손실 가능성
        summary = self.llm.predict(f"Summarize: {context}")
        return summary

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

# 비용 계산 예시
class MultiAgentCostAnalyzer:
    def __init__(self):
        self.call_log = []
    
    def estimate_cost(self, num_agents: int, avg_tokens_per_call: int = 500):
        # 각 에이전트 간 통신이 평균 2번 발생한다고 가정
        communication_pairs = num_agents * (num_agents - 1) / 2
        total_calls = num_agents + communication_pairs * 2
        
        # GPT-4 기준 ($0.03/1K input, $0.06/1K output)
        input_cost = total_calls * avg_tokens_per_call * 0.03 / 1000
        output_cost = total_calls * avg_tokens_per_call * 0.06 / 1000
        
        return {
            "total_calls": total_calls,
            "estimated_cost_usd": input_cost + output_cost
        }

# 실험
analyzer = MultiAgentCostAnalyzer()
print(analyzer.estimate_cost(num_agents=5))
# Output: {'total_calls': 25.0, 'estimated_cost_usd': 2.25}

print(analyzer.estimate_cost(num_agents=10))
# Output: {'total_calls': 100.0, 'estimated_cost_usd': 9.0}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

# Bad: 과도한 역할 분담
agents = {
    "researcher": ResearchAgent(),
    "analyst": AnalystAgent(),
    "writer": WriterAgent(),
    "editor": EditorAgent(),
    "reviewer": ReviewAgent(),
    "formatter": FormatAgent()
}

# Good: 역할 통합
agents = {
    "research_analyst": ResearchAnalystAgent(),  # Research + Analysis
    "content_creator": ContentCreatorAgent()      # Write + Edit + Format
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

from pydantic import BaseModel
from typing import List, Optional

class TaskContext(BaseModel):
    """에이전트 간 전달되는 표준 컨텍스트 형식"""
    original_request: str
    constraints: List[str]
    previous_decisions: List[str]
    must_preserve: List[str]  # 절대 손실되면 안 되는 정보
    confidence_score: float

class AgentInterface:
    def execute(self, context: TaskContext) -> TaskContext:
        # 입력/출력이 동일한 스키마를 가짐
        result = self._process(context)
        return TaskContext(
            original_request=context.original_request,
            constraints=context.constraints,
            previous_decisions=context.previous_decisions + [result.decision],
            must_preserve=context.must_preserve,
            confidence_score=self._calculate_confidence(result)
        )

1
2
3
4
5

class OrchestrationConfig:
    max_iterations: int = 3  # 에이전트 간 피드백 최대 횟수
    convergence_threshold: float = 0.95  # 결과 품질 임계값
    timeout_seconds: int = 300  # 전체 타임아웃
    fallback_to_single_agent: bool = True  # 실패 시 단일 에이전트로 폴백

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

import wandb

class ObservableAgent:
    def __init__(self, name: str):
        self.name = name
        wandb.init(project="multi-agent-debug")
    
    def log_communication(self, sender: str, receiver: str, message: str):
        wandb.log({
            "sender": sender,
            "receiver": receiver,
            "message_length": len(message),
            "timestamp": time.time()
        })
    
    def log_decision(self, decision: str, confidence: float):
        wandb.log({
            "agent": self.name,
            "decision": decision[:100],  # Truncate
            "confidence": confidence
        })