📌 개요
Multi-Agent 시스템에서 여러 에이전트를 조율하고 실행하는 Planning Agent를 구현했음. 이번 구현에서는 두 가지 접근 방식을 비교함:
- PlanningAgent: 사전 정의된 순차적 워크플로우를 따르는 정적 에이전트
- AutonomousPlanningAgent: OpenAI Tool Calling을 활용하여 동적으로 도구를 선택하고 실행하는 자율 에이전트
Key Concept: Autonomous Agent (자율 에이전트)
사전에 정의된 규칙 대신 LLM이 스스로 계획을 수립하고 적절한 도구(Tool)를 호출하여 목표를 달성하는 시스템임. 이는 기존의 정적 워크플로우와 달리 상황에 따라 유연하게 대응할 수 있는 장점이 있음.
🔧 구현 내용
1️⃣ PlanningAgent: 정적 워크플로우 구현
PlanningAgent는 세 가지 하위 에이전트를 순차적으로 실행하는 전통적인 접근 방식임:
class PlanningAgent(Agent):
name = "Planning Agent"
color = Agent.GREEN
DEAL_THRESHOLD = 50
def __init__(self, collection):
"""
Create instances of the 3 Agents that this planner coordinates across
"""
self.log("Planning Agent is initializing")
self.scanner = ScannerAgent()
self.ensemble = EnsembleAgent(collection)
self.messenger = MessagingAgent()
self.log("Planning Agent is ready")워크플로우 실행 로직
plan() 메서드는 다음과 같은 순서로 작업을 수행함:
def plan(self, memory: List[str] = []) -> Optional[Opportunity]:
"""
Run the full workflow:
1. Use the ScannerAgent to find deals from RSS feeds
2. Use the EnsembleAgent to estimate them
3. Use the MessagingAgent to send a notification of deals
"""
self.log("Planning Agent is kicking off a run")
selection = self.scanner.scan(memory=memory)
if selection:
opportunities = [self.run(deal) for deal in selection.deals[:5]]
opportunities.sort(key=lambda opp: opp.discount, reverse=True)
best = opportunities[0]
self.log(
f"Planning Agent has identified the best deal has discount ${best.discount:.2f}"
)
if best.discount > self.DEAL_THRESHOLD:
self.messenger.alert(best)
self.log("Planning Agent has completed a run")
return best if best.discount > self.DEAL_THRESHOLD else None
return None워크플로우 단계:
1. 스캔: ScannerAgent를 통해 RSS 피드에서 딜 정보 수집
2. 가격 평가: 각 딜에 대해 EnsembleAgent로 실제 가치 추정
3. 정렬 및 선택: 할인율 기준으로 정렬하여 최고의 딜 선택
4. 알림: 할인율이 임계값($50)을 초과하면 MessagingAgent로 사용자에게 알림
2️⃣ AutonomousPlanningAgent: Tool Calling 기반 자율 에이전트
AutonomousPlanningAgent는 OpenAI의 Tool Calling 기능을 활용하여 LLM이 스스로 도구를 선택하고 실행하도록 구현했음.
Tool 정의
세 가지 도구를 JSON Schema 형식으로 정의함:
scan_function = {
"name": "scan_the_internet_for_bargains",
"description": "Returns top bargains scraped from the internet along with the price each item is being offered for",
"parameters": {
"type": "object",
"properties": {},
"required": [],
"additionalProperties": False,
},
}
estimate_function = {
"name": "estimate_true_value",
"description": "Given the description of an item, estimate how much it is actually worth",
"parameters": {
"type": "object",
"properties": {
"description": {
"type": "string",
"description": "The description of the item to be estimated",
},
},
"required": ["description"],
"additionalProperties": False,
},
}
notify_function = {
"name": "notify_user_of_deal",
"description": "Send the user a push notification about the single most compelling deal; only call this one time",
"parameters": {
"type": "object",
"properties": {
"description": {
"type": "string",
"description": "The description of the item itself scraped from the internet",
},
"deal_price": {
"type": "number",
"description": "The price offered by this deal scraped from the internet",
},
"estimated_true_value": {
"type": "number",
"description": "The estimated actual value that this is worth",
},
"url": {
"type": "string",
"description": "The URL of this deal as scraped from the internet",
},
},
"required": ["description", "deal_price", "estimated_true_value", "url"],
"additionalProperties": False,
},
}Tool Call 처리 메커니즘
LLM이 요청한 도구 호출을 실제로 실행하는 핸들러:
def handle_tool_call(self, message):
"""
Actually call the tools associated with this message
"""
mapping = {
"scan_the_internet_for_bargains": self.scan_the_internet_for_bargains,
"estimate_true_value": self.estimate_true_value,
"notify_user_of_deal": self.notify_user_of_deal,
}
results = []
for tool_call in message.tool_calls:
tool_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
tool = mapping.get(tool_name)
result = tool(**arguments) if tool else ""
results.append(
{"role": "tool", "content": result, "tool_call_id": tool_call.id}
)
return results자율 실행 루프
LLM이 작업을 완료할 때까지 도구 호출을 반복하는 루프:
def plan(self, memory: List[str] = []) -> Optional[Opportunity]:
"""
Run the full workflow, providing the LLM with tools to surface scraped deals to the user
"""
self.log("Autonomous Planning Agent is kicking off a run")
self.memory = memory
self.opportunity = None
messages = self.messages[:]
done = False
while not done:
response = self.openai.chat.completions.create(
model=self.MODEL, messages=messages, tools=self.get_tools()
)
if response.choices[0].finish_reason == "tool_calls":
message = response.choices[0].message
results = self.handle_tool_call(message)
messages.append(message)
messages.extend(results)
else:
done = True
reply = response.choices[0].message.content
self.log(f"Autonomous Planning Agent completed with: {reply}")
return self.opportunity실행 흐름:
1. LLM에게 시스템 메시지와 사용자 메시지 전달
2. LLM이 도구 호출을 요청하면 handle_tool_call()로 실행
3. 도구 실행 결과를 메시지 히스토리에 추가
4. LLM이 "OK"를 반환할 때까지 반복
🎯 두 접근 방식의 비교
| 특성 | PlanningAgent | AutonomousPlanningAgent |
|---|---|---|
| 워크플로우 | 고정된 순차 실행 | LLM이 동적으로 결정 |
| 유연성 | 낮음 (코드 수정 필요) | 높음 (프롬프트 수정으로 조정 가능) |
| 예측 가능성 | 높음 | 중간 (LLM 판단에 의존) |
| 구현 복잡도 | 낮음 | 중간 (Tool Calling 메커니즘 필요) |
| 비용 | 낮음 | 높음 (LLM 호출 횟수 증가) |
| 확장성 | 제한적 | 우수 (새 도구 추가 용이) |
정적 vs 동적 워크플로우 선택 기준
- 정적 워크플로우: 작업 순서가 명확하고 변하지 않는 경우, 비용 효율성이 중요한 경우
- 동적 워크플로우: 상황에 따라 유연한 대응이 필요한 경우, 복잡한 의사결정이 필요한 경우
💡 핵심 학습 내용
OpenAI Tool Calling의 동작 원리
- 도구 정의: JSON Schema 형식으로 함수 시그니처와 설명 제공
- LLM 요청: 도구 목록과 함께 LLM에 요청 전송
- 도구 선택: LLM이 상황에 맞는 도구와 인자 결정
- 실행 및 피드백: 도구 실행 결과를 LLM에 다시 전달
- 반복: LLM이 작업 완료를 선언할 때까지 반복
자율 에이전트 설계 시 고려사항
system_message = "You find great deals on bargain products using your tools, and notify the user of the best bargain."
user_message = """
First, use your tool to scan the internet for bargain deals. Then for each deal, use your tool to estimate its true value.
Then pick the single most compelling deal where the price is much lower than the estimated true value, and use your tool to notify the user.
Then just reply OK to indicate success.
"""효과적인 프롬프트 작성
- 명확한 작업 순서 제시
- 종료 조건 명시 ("reply OK to indicate success")
- 도구 사용 목적과 제약사항 설명 ("only call this one time")
🧪 실험 및 검증
Tool Calling 메커니즘을 단계별로 실험했음:
- 가짜 함수로 프로토타입 구현: 실제 에이전트 없이 Tool Calling 흐름 검증
- 도구 스키마 정의 및 테스트: JSON Schema 형식의 도구 정의 검증
- 실제 에이전트 통합: ScannerAgent, EnsembleAgent, MessagingAgent와 통합
실행 결과:
Fake function to scan the internet - this returns a hardcoded set of deals
Fake function to estimating true value of The Hisense R6 Serie... - this always returns $300
Fake function to estimating true value of The Poly Studio P21 ... - this always returns $300
Fake function to estimating true value of The Lenovo IdeaPad S... - this always returns $300
Fake function to estimating true value of The Dell G15 gaming ... - this always returns $300
Fake function to notify user of The Poly Studio P21 ... which costs 30 and estimate is 300LLM이 자율적으로:
- 스캔 도구 호출
- 각 deal에 대해 가치 평가 도구 호출
- 최고의 deal 선택 및 알림 도구 호출
- "OK" 응답으로 작업 완료
📝 결론
이번 구현을 통해 정적 워크플로우와 자율 에이전트의 장단점을 명확히 이해했음.
PlanningAgent는 예측 가능하고 비용 효율적이지만 유연성이 제한적임. 반면 AutonomousPlanningAgent는 OpenAI Tool Calling을 활용하여 상황에 맞게 동적으로 대응할 수 있지만, LLM 호출 비용과 예측 불가능성이라는 트레이드오프가 존재함.
실제 프로덕션 환경에서는 두 접근 방식을 하이브리드로 결합하는 것이 효과적일 수 있음:
- 핵심 워크플로우는 정적으로 구현
- 복잡한 의사결정이 필요한 부분만 자율 에이전트 활용
자율 에이전트 구현 시 주의사항
- 무한 루프 방지를 위한 최대 반복 횟수 설정 필요
- 도구 호출 비용 모니터링 필수
- 중요한 작업은 사람의 승인 단계 추가 권장
📚 참고 자료
I Enjoy Learn-and-Run Vibe😊