Tech Dive

여러 에이전트를 하나의 흐름으로 엮는 기술, Agent Orchestrator

들어가며


안녕하세요, KT의 Agent 연구를 담당하는 Agentic AI Lab의 송윤성, 홍지후입니다. 저희 팀에서는 Agent를 중심으로 파생되는 다양한 기술들을 연구하고, 실제 상용화까지 적용할 수 있도록 AI Agent를 개발하고 있습니다. 이 글에서는 멀티에이전트 시스템을 프로덕션에서 운영하면서 겪는 실제 문제들과, 이를 해결하기 위해 개발한 Agent Orchestrator의 설계와 구현을 공유하려고 합니다. 

멀티에이전트하에서 웹 검색, 데이터 분석, 금융 상담, 상권 분석 등 기능별로 에이전트가 분리되면서, 복잡한 사용자 요청을 처리하기 위한 동적 라우팅과 태스크 분해가 오케스트레이터를 개발함에 있어서 필수적이 되었습니다. 이 과정에서 오케스트레이터도 여러 차례 진화했습니다. 처음에는 단일 레벨 실행기와 플랫(flat) 상태로 시작했지만, 프로덕션에서 발생하는 문제들, 예를 들어 멀티턴 대화, 병렬 파라미터 수집, 파일 전달, 대화 의도 분류와 같은 문제를 해결하면서 계층적 상태, 서브그래프 패턴, 독립 플래너 등으로 오케스트레이터 아키텍처가 발전했습니다. 

왜 오케스트레이션이 필요한가

개별 에이전트의 성능은 충분했지만, 에이전트가 증가하면서 문제는 에이전트들 간의 조율이었습니다.  


| 문제

01.jpg

최근 연구들도 이 문제를 지적하고 있습니다. Tokenomics 연구 (“Tokenomics: Quantifying Where Tokens Are Used in Agentic Software Engineering” )에서는 멀티에이전트 시스템에서 토큰이 어디에 소비되는지에 대한 실증 분석을 수행했으며, 멀티에이전트 시스템 실패 분석 연구 (“Why Do Multi-Agent LLM Systems Fail?” ) 에서는 중복된 역할이 경쟁이나 중복 작업을 유발하여 토큰과 시간을 낭비하고, 에이전트들이 동일한 논점을 무한히 논쟁하며 토큰 할당량을 소진하는 문제를 지적했습니다. 

LangChain, LangGraph, AutoGen, CrewAI 등 프레임워크는 많았지만, 저희가 마주한 문제들인 “동적 에이전트 추가, 의존성 기반 실행, 중간 상태 복구, 사용자 개입 처리”를 통합적으로 해결하는 솔루션은 없었습니다.  



멀티에이전트 시스템의 6가지 핵심 문제와 해결책


02-2.jpg


멀티에이전트 시스템에서 발생하는 6가지 핵심 문제와 이를 오케스트레이션 레벨에서 어떻게 해결했는지 설명해보도록 하겠습니다. 그리고 프로덕션 운영 과정에서 추가로 발견한 문제—대화 의도 분류—와 그 해결 방법도 함께 다루어 보도록 하겠습니다. 


1. 에이전트 추가 시 시스템 전체 수정 필요

Python 
# 일반적인 정적 라우팅 방식
def route_request(request):
    if "주식" in request:
        return stock_agent
    elif "날씨" in request:
        return weather_agent
    elif "결제" in request:  # 새 에이전트 추가마다 분기 추가
        return payment_agent
    # ... N개의 분기​
123456789

에이전트 개발 자체는 단순하지만, 라우터 로직 수정, 메시지 포맷 변경, 의존성 체인 확인, 회귀 테스트까지 고려하면 에이전트 하나를 추가할 때마다 시스템 전체를 수정해야하는 상당한 공수가 소요됩니다. 저희 시스템에서 백개 이상의 에이전트를 관리하면서, 정적 라우팅 방식은 더 이상 유지 가능하지 않았습니다.


2. 실행 순서 관리의 어려움

Python 
# 성능을 위한 병렬 실행
results = await asyncio.gather(
    websearch_agent.run(),      # 뉴스 검색 (2초)
    stock_loader_agent.run(),   # 주가 데이터 (1초)
    analysis_agent.run()        # 종합 분석 (0.5초) ← 먼저 완료
)​
123456
분석 에이전트가 가장 먼저 실행을 완료했으나, 이 시점에 뉴스 검색과 주가 데이터는 미완료 상태입니다. 결과적으로 빈 입력으로 분석을 수행하게 됩니다.

순서를 보장하면 병렬화 이점을 상실하고, 병렬로 실행하면 의존성이 깨집니다. DAG 기반 플래너 연구에서는 이 문제를 해결하기 위해 도구 선택(노드)과 의존성 모델링(엣지)의 정확도를 크게 향상시키는 DAG(Directred Acyclic Graph) 최적화 플래너를 제안합니다.

3. 실패 시 처음부터 재시작

TypeScript 
​[실행 로그]
10:00 - 금융 분석 시나리오 시작
10:15 - 웹 검색 에이전트 완료 (뉴스 ㅇㅇ건)
10:28 - 주가 데이터 에이전트 완료 (ㅇㅇ일 데이터)
10:35 - 1차 분석 에이전트 완료
10:39 - 리포트 생성 에이전트 → API 타임아웃
10:40 - 전체 워크플로우 실패 → 처음부터 재시작
1234567

체크포인트 미지원으로 인해 마지막 단계 실패 시 이미 완료된 모든 작업(뉴스 검색, 주가 조회, 1차 분석)이 유실됩니다. 


LangGraph 체크포인팅 연구에 따르면, 체크포인터는 각 실행 단계에서 그래프 상태의 스냅샷을 저장하여 장애 허용성과 오류 복구를 가능하게 합니다.


4. 사용자 추가 입력 처리 불가

TypeScript 
[시나리오: 요금제 변경]
사용자: "요금제 변경해줘"
요금제 에이전트: "5G 프리미엄(79,000원)과 5G 스탠다드(59,000원) 중 어떤 걸로 하시겠습니까?"
시스템: (사용자에게 질문을 전달할 메커니즘 없음)
       → 임의 선택 또는 실패 처리​
123456

"요청 → 응답" 단방향 흐름만 지원하는 시스템에서는 워크플로우 중간에 사용자 입력을 받을 수 없습니다. Human-in-the-Loop 연구에서 강조하듯이, HITL은 AI 에이전트의 신뢰성을 높이기 위한 필수 요소입니다.


5. 에이전트 간 정보 중복 요청

TypeScript 
[토큰 사용량 분석]
- 웹 검색 에이전트 → "A전자 최근 뉴스" 요청
- 리포트 에이전트 → "A전자 최근 뉴스" 요청 (중복)
- 요약 에이전트 → "A전자 최근 뉴스" 요청 (중복)
총 토큰: 15,000
실제 필요 토큰: 5,000
낭비율: 66%​
1234567

에이전트 간 직접 통신 구조에서는 "다른 에이전트가 이미 조회한 정보"를 알 방법이 없습니다. 동일 정보를 반복 요청하여 토큰을 낭비합니다. 정적 오케스트레이션 연구에 따르면 원칙적인 동적 조율의 부재는 중복 계산과 비효율성을 초래합니다.


6. 다중 결과 통합의 어려움

TypeScript 
[개별 에이전트 결과]
웹 검색: "A전자 관련 뉴스 ㅇㅇ건 수집"
주가 분석: "최근 ㅇㅇ일 5.2% 상승, RSI 67"
실적 분석: "전분기 대비 영업이익 XX% 증가"
감성 분석: "긍정 뉴스 62%, 부정 뉴스 38%"
→ 사용자 기대: "그래서 투자 판단은?"​
123456

산발적인 개별 결과를 사용자에게 그대로 전달하면 혼란을 야기합니다. 맥락 이해, 모순 조율, 일관된 메시지 정리가 필요합니다.


저희는 각 문제를 심도있게 연구하여, 이 문제들을 오케스트레이션 레벨에서 해결하고자 했습니다. 개별 에이전트의 로직을 건드리지 않고, 에이전트들을 조율하는 중앙 레이어에서 문제를 해결하는 접근입니다.

| Agent Orchestrator로의 접근


03.jpg


각 해결책은 서로 독립적으로 동작하지 않고, 긴밀하게 연결되어 있습니다. 예를 들어 HITL로 워크플로우가 중단되면 State Manager가 현재 상태를 저장하고, 사용자 입력 후 재개할 때 Executor가 저장된 상태에서 이어서 실행합니다.




Agent Orchestrator 아키텍처


설계 철학

이 문제들을 오케스트레이션 레벨에서 해결하고자 했습니다. 개별 에이전트의 로직을 건드리지 않고, 에이전트들을 조율하는 중앙 레이어에서 문제를 해결하는 접근입니다.

핵심 아이디어는 사용자 요청을 DAG(Directed Acyclic Graph, 방향성 비순환 그래프) 형태의 워크플로우로 변환하는 것입니다. DAG란 작업들을 노드로, 작업 간 의존성을 화살표로 표현한 그래프입니다. "비순환"이므로 A→B→C→A처럼 순환하는 구조가 없어서, 어떤 순서로 실행해야 하는지 명확하게 결정할 수 있습니다.

예를 들어 "잔고 조회 → 투자 배분 결정 → 주식/채권/암호화폐 추천(병렬) → 매매 실행"이라는 흐름은 DAG로 표현하면 의존성이 명시되고, 병렬 실행 가능한 부분도 자동으로 파악됩니다. Agentic Tool-Orchestration 연구에서도 이 접근을 핵심으로 제시합니다.

전체 아키텍처

04.jpg


핵심 컴포넌트 역할

Agent orchestrator의 각 컴포넌트들은 의도 분류, 계획수립, 그룹 실행, 테스크 실행, 파라미터 수집, 결과 통합, 상태 관리, 에이전트 검색 역할을 수행합니다.


컴포넌트

역할

주요 책임

Intent Classifier

의도 분류

사용자 메시지가 새 요청인지, 파라미터 응답인지, 일반 대화인지 판별

Planner

계획 수립

독립 A2A 에이전트로 동작. Agent Hub 검색 + DAG 실행 계획 생성

Group Executor

그룹 실행

서브그래프로 동작. Send API로 그룹 내 태스크 병렬 dispatch

Task Executor

태스크 실행

서브그래프로 동작. 개별 에이전트 호출 + A2A 검증

Parameter Handler

파라미터 수집

서브그래프로 동작. 집계 윈도우로 복수 요청 통합

Finalizer

결과 통합

여러 에이전트의 결과를 수집하여 일관된 응답 정리

State Manager

상태 관리

상태를 체크포인트로 관리

Agent Hub

에이전트 검색

백개 이상의 에이전트 중 요청에 적합한 에이전트를 하이브리드 검색으로 탐색


서브 에이전트들은 A2A(Agent-to-Agent) 프로토콜로 오케스트레이터와 통신합니다. A2A는 Google이 제안한 에이전트 간 통신 표준으로, 태스크 생성/상태 조회/결과 수신을 스트리밍 방식으로 처리합니다.




6가지 핵심 기술 상세


05.jpg


1. 동적 플래닝: Agent Card 기반 자동 계획 수립


문제 상황

백개 이상의 에이전트 중에서 사용자 요청에 적합한 에이전트를 어떻게 찾을 것인가? 그리고 찾은 에이전트들을 어떤 순서로 실행할 것인가?


해결 방식

06.jpg


이 흐름은 세 단계로 동작합니다.

사용자 요청 입력 : "잔고 조회해서 투자 포트폴리오 구성해줘" →  Agent Hub 검색 : 요청과 관련된 에이전트들을 하이브리드 검색으로 탐색 → LLM Planner : 검색된 에이전트들로 실행 계획(DAG) 생성

Agent Card는 에이전트의 역량을 JSON으로 명세한 것입니다. 사람으로 치면 이력서와 같습니다. 이 에이전트가 무엇을 할 수 있는지, 어떤 상황에서 호출해야 하는지를 기술합니다.

JSON 
​{
  "name": "WealthManagementAgent",
  "description": "개인 자산배분 전략 컨설턴트입니다. 잔고나 자산 정보를 바탕으로
                  주식, 암호화폐, 부동산, 채권 등 자산 클래스 간 최적 배분 비율을 제안합니다.",
  "url": "http://localhost:10330/",
  "version": "1.0.0",
  "skills": [
    {
      "id": "asset_allocation",
      "name": "자산 배분 전략",
      "description": "투자 성향과 자산 규모에 따른 최적 포트폴리오 배분 비율 제안",
      "tags": ["allocation", "portfolio", "investment", "배분", "투자"],
      "examples": [
        "5천만원으로 분산투자 하려면 어떻게 배분해야 할까?",
        "위험 성향이 보수적인데 자산 배분 추천해줘"
      ]
    }
  ]
}
12345678910111213141516171819


Agent Card의 각 필드 역할은 아래와 같습니다.

  • name/description: 에이전트의 정체성과 전문 분야

  • url: 에이전트가 실행되는 서버 주소 (A2A 통신용)

  • skills: 에이전트가 수행할 수 있는 구체적인 기능 목록

  • tags: 검색 시 매칭에 사용되는 키워드

  • examples: LLM이 이 에이전트를 언제 선택해야 하는지 판단하는 예시


새 에이전트를 추가할 때는 이 JSON 파일만 Agent Hub에 등록하면 됩니다. 코드 수정이 필요 없습니다.


Agent Hub 하이브리드 검색


백개 이상의 에이전트에서 적합한 에이전트를 찾으려면 단순 키워드 매칭으로는 부족합니다. "잔고 조회"라는 요청에 "계좌 조회" 에이전트가 매칭되어야 하는데, 키워드가 다르면 놓칠 수 있습니다.


이 문제를 해결하기 위해 BM25(키워드 매칭) + Vector(의미 유사도) 앙상블 검색을 구현했습니다.

Python 
​# Agent Hub 하이브리드 검색 과정

# 1. BM25 키워드 검색 (40% 가중치)
# 에이전트의 이름, 역할, 설명, 스킬을 하나의 텍스트로 합침
searchable_text = f"{name} {role} {description} {skills}"
# 한글/영문/숫자를 토큰으로 분리
tokens = re.findall(r'[가-힣]+|[a-zA-Z]+|[0-9]+', searchable_text)
# BM25 알고리즘으로 키워드 매칭 점수 계산
bm25_scores = bm25.get_scores(query_tokens)
normalized_bm25 = scores / max(scores)  # 0~1 정규화

# 2. Vector 의미 검색 (60% 가중치)
# 쿼리를 임베딩 벡터로 변환
query_embedding = await openai.embeddings.create(
    model="text-embedding-3-large",
    input=query
)
# ChromaDB에서 가장 유사한 에이전트 검색
results = chroma_collection.query(
    query_embeddings=[query_embedding],
    n_results=k
)
# 거리를 유사도 점수로 변환 (거리가 가까울수록 점수가 높음)
vector_score = 1.0 / (1.0 + distance)

# 3. 앙상블 스코어
# BM25와 Vector 점수를 가중 평균하여 최종 점수 산출
ensemble_score = (bm25_score × 0.4) + (vector_score × 0.6)
12345678910111213141516171819202122232425262728


왜 40:60 비율인가?

BM25(40%) → 정확한 키워드가 일치할 때 강점. "A전자"라는 단어가 명시적으로 있으면 높은 점수.

Vector (60%) → 의미적 유사성 파악. "주식 투자"와 "증권 매매"가 같은 의미임을 이해.

실험 결과, Vector 검색에 더 높은 가중치를 줄 때 사용자 의도에 맞는 에이전트를 더 잘 찾았습니다.


실제 검색 결과 예시 (KT 멤버십 혜택 질의) ▾ 

TypeScript 
[VectorStore] 최종 선택된 에이전트 (BM25=0.4, Vector=0.6):

  1. KT혜택추천에이전트 (BM25=0.992, Vector=0.685, Ensemble=0.808)
  2. KT영화에이전트 (BM25=1.000, Vector=0.619, Ensemble=0.771)
  3. KTGeneralAssistantAgent (BM25=0.780, Vector=0.645, Ensemble=0.699)
  4. Mem0MockAgent (BM25=0.801, Vector=0.601, Ensemble=0.681)
  ...​
1234567


Diversity Reranker

하이브리드 검색만으로는 문제가 있습니다. 점수가 높은 에이전트들이 비슷한 기능을 가질 수 있습니다. 예를 들어 "주식 투자" 검색 시 StockAgent, EquityAgent, SecuritiesAgent가 모두 상위에 올 수 있는데, 이들은 사실상 같은 역할입니다.

이를 방지하기 위해 다양성 리랭킹(Diversity Reranking)을 적용했습니다. 이미 선택된 에이전트들과 너무 유사한 에이전트는 점수를 낮춰서 다양한 기능의 에이전트가 선택되도록 합니다.

Python 
​# 다양성 점수 계산
for candidate in candidates:
    # 이미 선택된 에이전트들과의 유사도 계산
    similarities = [calculate_similarity(candidate, s) for s in selected]
    avg_similarity = mean(similarities) if similarities else 0

    # 유사도가 낮을수록 다양성 점수가 높음
    diversity_score = 1 - avg_similarity

    # 최종 점수: 원래 점수 75% + 다양성 점수 25%
    final_score = original_score × 0.75 + diversity_score × 0.25

    # 유사도 70% 미만인 에이전트만 선택 (너무 비슷하면 제외)
    if avg_similarity < 0.7:
        selected.append(candidate)
123456789101112131415


실제 적용 결과

TypeScript 
🎯 Query: '내 KT멤버십으로 어떤 혜택을 받을 수 있어?'
🌈 Selected 20 agents (from 000 total)
🏷️ Top skill tags: {'추천': 4, 'lifestyle': 4, 'recommendation': 4}
📈 Skill diversity ratio: 17.95
1
2
3
4

100개 이상의 에이전트 중 20개가 선택되었고, 다양한 스킬 태그(추천, lifestyle, recommendation)가 고르게 분포되어 있습니다.


Planner의 독립 에이전트화

초기 버전에서 Planner는 오케스트레이터 그래프 내부의 노드였습니다. 기능적으로는 문제가 없었지만, 프로덕션 운영에서 몇 가지 한계가 드러났습니다:


  • 관심사 결합: 계획 로직 수정이 오케스트레이터 전체 배포를 요구

  • 스케일링: 계획 수립 부하가 높을 때 오케스트레이터 전체가 영향 받음

  • 테스트: 계획 로직을 오케스트레이터와 독립적으로 테스트하기 어려움


이를 해결하기 위해 Planner를 독립 A2A 에이전트로 분리했습니다. 


Planner Agent는 자체 StateGraph를 갖습니다. analyze_request → find_agents → create_plan → analyze_dependencies 4단계 파이프라인으로 동작하며, 각 단계가 독립적으로 테스트 가능합니다.


특히 주목할 점은 에이전트 검색을 MCP를 통해 수행​한다는 것입니다. Planner가 Agent Hub MCP Server에 find_top_k_agents를 호출하면, Agent Hub가 하이브리드 검색 + 다양성 리랭킹을 수행하고 결과를 반환합니다.

Python 
# Planner Agent의 에이전트 검색 (MCP 기반)
class FindRelevantAgentsTool:
    async def run(self, query: str, top_k: int = 20) -> List[AgentCard]:
        # MCP Server Session을 통해 Agent Hub에 검색 요청
        result = await mcp_session.call_tool(
            "find_top_k_agents",
            arguments={"query": query, "top_k": top_k}
        )
        return result
1
2
3
4
5
6
7
8
9

의존성 분석은 계획 생성 후 별도 LLM 호출로 수행합니다. 이 단계에서 각 태스크에 execution_group(실행 순서)을 할당하고, 중복 태스크를 통합(consolidation)합니다. 또한 찾은 에이전트 카드 정보(URL, 인증 등)를 계획에 직접 주입하여, 오케스트레이터가 별도의 에이전트 카드 디렉터리를 조회할 필요가 없습니다.



효과

07-2.jpg




2. 의존성 기반 실행 : 서브그래프와 Send API로 병렬 실행

문제 상황

여러 에이전트를 실행할 때 두 가지 상반된 요구사항이 있습니다.

  1. 순서 보장 : 잔고 조회가 완료된 후에야 투자 배분을 결정할 수 있음
  2. 병렬 실행 : 주식/채권/암호화폐 추천은 서로 독립적이므로 동시에 실행 가능


단순히 모든 태스크를 순차 실행하면 느리고, 모든 태스크를 병렬 실행하면 의존성이 깨집니다.


해결 방식 : 서브그래프 아키텍처

오케스트레이터의 실행 엔진은 3단계 중첩 서브그래프로 구성됩니다. 처음에는 단일 레벨 executor로 시작했지만, 병렬 실행 시 상태 격리 문제와 에러 전파 문제가 발생하면서 이 구조로 발전했습니다.


08.jpg


각 서브그래프의 역할

서브그래프

역할

상태 스코프

Group Executor

실행 그룹 단위 조율

GroupExecutorState

Task Executor

개별 에이전트 호출

TaskExecutorState

Parameter Handler

파라미터 수집/라우팅

ParameterHandlerState



| 왜 서브그래프인가?

서브그래프를 사용하면 각 실행 단위가 독립적인 상태 스코프를 갖습니다. Group Executor의 상태 변경이 다른 그룹에 영향을 주지 않고, Task Executor에서 에러가 발생해도 같은 그룹의 다른 태스크에는 영향이 없습니다.
| LangGraph Send API: 동적 병렬 실행

그룹 내 태스크들을 병렬로 실행하는 핵심은 LangGraph Send API입니다. Send API는 런타임에 동적으로 결정된 수의 노드를 동시에 실행할 수 있게 해줍니다.

Python 
# Group Executor의 dispatch 단계
from langgraph.types import Send

def create_send_tasks(state: GroupExecutorState) -> List[Send]:
    """그룹 내 각 태스크를 병렬 실행할 Send 객체 생성"""
    sends = []
    for task in state["tasks"]:
        task_state = TaskExecutorState(
            task_id=task["task_id"],
            agent_name=task["agent_name"],
            task_description=task["task_description"],
            group_id=state["group_id"],
            # ... 컨텍스트 주입
        )
        # 각 태스크를 독립적인 task_executor 서브그래프로 dispatch
        sends.append(Send("execute_task", task_state))
    return sends
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

Send API의 핵심은 sends.append(Send("execute_task", task_state))입니다. 이 한 줄로 각 태스크가 독립적인 task_executor 서브그래프 인스턴스에서 실행됩니다. 3개의 태스크가 있으면 3개의 task_executor가 동시에 실행됩니다.


병렬 실행된 결과는 operator.add 리듀서로 자동 수집됩니다.

Python 
class GroupExecutorState(TypedDict):
    # operator.add: 병렬 실행된 task_executor들의 결과를 리스트에 누적
    task_states: Annotated[List[TaskExecutorState], operator.add]
    group_results: Annotated[List[Dict], operator.add]
1
2
3
4

operator.add 리듀서 덕분에 각 병렬 task_executor가 반환한 결과가 자동으로 리스트에 누적됩니다. 순서 보장이 필요 없는 수집 작업에 적합합니다.



| A2A Task State Machine

각 태스크는 A2A 프로토콜의 Task State Machine을 따릅니다.

09.jpg

상태

유형

의미

submitted

초기

태스크가 제출됨

working

진행

에이전트가 처리 중

input-required

재개 가능

사용자 입력 필요
(→ working으로 복귀)

auth-required

재개 가능

인증 필요
(→ working으로 복귀)

completed

종료

성공

failed

종료

실패

canceled

종료

취소

rejected

종료

에이전트가 거부


task_executor는 A2A 스트리밍 응답을 받으면서 이 상태 전이를 추적합니다. completed면 결과를 저장하고, input-required면 파라미터 핸들러로 넘깁니다.



| 의존성 충족 확인과 실행 그룹

Planner가 생성하는 DAG 예시 (금융 시나리오) ▾ 

JSON 
{
  "tasks": [
    {"id": "task_1", "description": "통장 잔고 조회", "execution_group": 0, "agent_name": "BankAgent"},
    {"id": "task_2", "description": "자산 배분 결정", "execution_group": 1, "agent_name": "WealthManagementAgent"},
    {"id": "task_3", "description": "주식 투자 추천", "execution_group": 2, "agent_name": "StockAgent"},
    {"id": "task_4", "description": "암호화폐 투자 추천", "execution_group": 2, "agent_name": "CryptoAgent"},
    {"id": "task_5", "description": "채권 투자 추천", "execution_group": 2, "agent_name": "BondAgent"},
    {"id": "task_6", "description": "매매 실행", "execution_group": 3, "agent_name": "TradingAgent"}
  ]
}
1
2
3
4
5
6
7
8
9
10

execution_group이 같은 태스크들은 동시에 실행됩니다. 그룹 번호가 순차적으로 진행되므로, 앞 그룹이 모두 완료된 후 다음 그룹이 실행됩니다:

TypeScript 
Group 0: [BankAgent] - 단독 실행
    ↓ (그룹 완료 후)
Group 1: [WealthManagementAgent] - 단독 실행
    ↓ (그룹 완료 후)
Group 2: [StockAgent, CryptoAgent, BondAgent] - Send API로 병렬 실행
    ↓ (그룹 완료 후)
Group 3: [TradingAgent] - 단독 실행
1
2
3
4
5
6
7


| 병렬 실행 효과 (실측)

TypeScript 
Group 2 실행 시간:
  - 순차 실행 시: ~15초 (5초 × 3개)
  - 병렬 실행 시: ~5초 (max)
  → 약 70% 시간 단축
1
2
3
4


효과
10.jpg



3. 상태 관리 & 복구: 계층적 상태와 Checkpointing


문제 상황

멀티에이전트 워크플로우는 수 분에서 수십 분이 걸릴 수 있습니다. 이 과정에서 네트워크 오류로 특정 에이전트 호출이 실패할 수 있고, 서버가 재시작될 수 있고, HITL로 워크플로우가 중단되고, 사용자가 나중에 응답할 수 있습니다. 

체크포인트가 없으면 이런 상황에서 처음부터 다시 시작해야 합니다. 앞서 완료한 모든 작업이 유실됩니다.

해결 방식 : Flat에서 Hierarchical로 초기에는 모든 상태를 하나의 플랫(flat) 딕셔너리로 관리했습니다. 

Python 
# 초기 버전: Flat State
class OrchestratorState(TypedDict, total=False):
    user_query: str
    plan: List[Dict]
    tasks: Annotated[Dict[str, TaskRuntime], merge_tasks]
    input_slots: Annotated[Dict[str, InputSlot], merge_input_slots]
    final_answer: Optional[dict]
1
2
3
4
5
6
7

이 구조는 단순한 시나리오에서는 잘 동작했지만, 프로덕션에서 다음 문제들이 발생했습니다. 


  • 멀티세션: 한 사용자가 연속으로 여러 요청을 보낼 때 이전 세션과 새 세션의 상태가 충돌

  • 상태 폭발: 에이전트 수가 늘어나면서 단일 딕셔너리에 수십 개의 필드가 혼재

  • 파생 뷰 불일치: agent_results, group_summaries, detailed_results 등이 각각 독립적으로 업데이트되면서 불일치 발생


이를 해결하기 위해 Session > Group > Task 3계층 상태로 재설계했습니다.

Python 
# 현재 버전: Hierarchical State
class HierarchicalOrchestratorState(TypedDict):
    # 세션 레벨: 멀티턴 대화의 논리적 단위
    active_session: Annotated[Optional[Dict], Reducers.merge_session_state]

    # 결과 계층: Session > Group > Task
    hierarchical_results: Annotated[Dict, Reducers.merge_hierarchical_results]

    # 파일 라우팅: 에이전트 간 파일 전달
    file_routing: Annotated[Optional[FileRoutingState], merge_file_routing_state]f
1
2
3
4
5
6
7
8
9
10


세션 내부 구조  ▾ 
11.jpg



| Single Source of Truth 패턴

계층적 상태에서 가장 중요한 설계 결정은 **task_results_by_group을 유일한 정보 원천(Single Source of Truth)**으로 설정한 것입니다. agent_results, group_summaries, detailed_results_by_group은 모두 이 정보 원천에서 파생되는 읽기 전용 뷰입니다.

Python 
# 파생 뷰 계산: 정보 원천 → 캐시용 뷰 생성
def derive_views_from_task_results(
    task_results_by_group: Dict[int, List[Dict]]
) -> Dict[str, Dict]:
    """
    task_results_by_group에서 모든 파생 뷰를 계산.
    이 함수만이 파생 뷰를 생성하며, 직접 업데이트는 금지.
    """
    group_summaries = {}      # 그룹별 요약 텍스트
    detailed_results = {}     # 에이전트별 상세 결과
    agent_results = {}        # 에이전트별 상태 스냅샷

    for gid, tasks in task_results_by_group.items():
        for task_entry in tasks:
            agent_name = task_entry.get("agent_name")
            # A2A artifact에서 텍스트 추출
            texts = _artifact_texts_from_result(task_entry.get("result"))
            # 각 파생 뷰에 반영
            ...

    return {
        "group_summaries": group_summaries,
        "detailed_results_by_group": detailed_results,
        "agent_results": agent_results,
    }
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

이 패턴의 이점은 명확합니다. 


  • 불일치 제거: 모든 뷰가 동일한 원천에서 파생되므로 상태 불일치가 원천적으로 불가능

  • 디버깅 용이: 문제 발생 시 task_results_by_group만 확인하면 됨

  • 확장 용이: 새로운 뷰가 필요하면 파생 함수만 추가
| 커스텀 Reducer 시스템

LangGraph에서 리듀서(reducer)는 여러 노드가 같은 필드를 업데이트할 때 어떻게 병합할지 정의하는 함수입니다. 계층적 상태에서는 다양한 병합 전략이 필요하여 전용 Reducers 클래스를 구현했습니다. 

Python 
class Reducers:
    """12개 이상의 전용 리듀서를 중앙 관리"""

    @staticmethod
    def merge_session_state(old, new):
        """세션 상태 병합 - 초기값 전달과 리셋 경로 지원"""
        # _reset 플래그: 새 요청 시 세션 상태 초기화
        if isinstance(new, dict) and new.get("_reset"):
            reset_fields = new.get("_reset_fields", [])
            result = dict(old or {})
            for field in reset_fields:
                result.pop(field, None)
            return result
        # 일반 병합: 변경된 필드만 덮어씀
        result = dict(old or {})
        result.update(new or {})
        return result

    @staticmethod
    def merge_task_results_by_group(old, new):
        """그룹별 태스크 결과 병합 - 그룹 레벨 중복 제거"""
        result = dict(old or {})
        for gid, tasks in (new or {}).items():
            existing = result.get(gid, [])
            # task_id 기준 중복 제거
            existing_ids = {t.get("task_id") for t in existing}
            for task in tasks:
                if task.get("task_id") not in existing_ids:
                    existing.append(task)
            result[gid] = existing
        return result
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
27
28
29
30
31


| 각 리듀서의 역할

리듀서

대상

전략

merge_session_state

세션 상태

패치 적용 + 리셋 경로

merge_task_results_by_group

태스크 결과

그룹별 중복 제거

merge_hierarchical_results

계층 결과

Session > Group > Task 계층 병합

merge_dicts_shallow

딕셔너리

얕은 병합

append_unique_efficient

리스트

중복 없는 추가


| Checkpointing

LangGraph 체크포인팅 을 활용합니다. 그래프의 각 노드가 실행될 때마다 전체 상태의 스냅샷을 저장합니다. 이를 슈퍼스텝(superstep)이라고 합니다.

TypeScript 
from langgraph.checkpoint.memory import MemorySaver

def build_orchestrator_graph():
    builder = StateGraph(HierarchicalOrchestratorState)
    # ... 노드 추가 ...

    # Checkpointer 연결
    saver = MemorySaver()  # 메모리 저장소 (프로덕션에서는 Redis/PostgreSQL 사용)
    return builder.compile(checkpointer=saver)
1
2
3
4
5
6
7
8
9

체크포인터를 연결하면 각 노드 실행 전후로 상태 스냅샷 저장하고, thread_id로 워크플로우 식별하며, 동일 thread_id로 재개하면 마지막 체크포인트에서 이어서 실행이 가능합니다. 


효과
12.jpg




4. HITL: 파라미터 집계 윈도우와 Human-in-the-Loop

문제 상황

워크플로우 실행 중 사용자 입력이 필요한 경우가 있습니다. 


  • "어떤 계좌를 조회할까요?" (계좌 선택)

  • "5G 프리미엄과 5G 스탠다드 중 어떤 요금제로 변경할까요?" (옵션 선택)

  • "결제를 진행할까요?" (확인)


단방향 "요청→응답" 구조에서는 이런 상호작용이 불가능합니다.


여기에 추가로, 병렬 실행되는 여러 에이전트가 동시에 파라미터를 요청하는 경우가 있습니다. 각각을 별도로 사용자에게 물어보면 UX가 떨어집니다.


해결 방식

LangGraph의 interrupt() 메커니즘을 활용합니다. interrupt()를 호출하면 워크플로우가 즉시 중단되고, 현재 상태가 체크포인트로 저장됩니다. 사용자가 응답하면 Command(resume=...)로 중단 지점에서 이어서 실행합니다.

13.jpg


| 파라미터 집계 윈도우

병렬 실행에서 여러 에이전트가 동시에 input-required 상태를 반환하면, 각각을 별도의 interrupt로 처리하면 사용자가 여러 번 응답해야 합니다. 이를 해결하기 위해 800ms 집계 윈도우를 도입했습니다.



| 동작 원리

  1. 첫 번째 에이전트가 input-required를 반환하면 타이머 시작 (param_window_started_at)

  2. 800ms 동안 같은 그룹의 다른 에이전트들의 파라미터 요청을 수집

  3. 윈도우 종료 후 모든 요청을 하나의 통합 프롬프트로 합쳐 Parameter Handler에 전달

  4. Parameter Handler가 interrupt()를 호출하여 사용자에게 한 번에 질문

  5. 사용자 응답을 각 에이전트별로 분배 (parameter routing)
| Parameter Handler 서브그래프

파라미터 수집은 전용 서브그래프에서 처리됩니다.

Python 
# Parameter Handler 서브그래프: collect → validate → route

async def collect_parameters(state: ParameterHandlerState):
    """사용자로부터 파라미터 수집"""
    pending = state.get("pending_requests", {})

    # 이미 사용자 응답이 있으면 (Command(resume)으로 전달됨) interrupt 건너뜀
    if state.get("user_response"):
        return {"status": "completed"}

    # ★ interrupt() 호출: 워크플로우 중단
    user_answer = interrupt({
        "namespace": "parameter_collection",
        "payload": {
            "requests": pending,  # 집계된 전체 파라미터 요청
            "message": "다음 정보가 필요합니다..."
        }
    })
    # ↑ 사용자가 응답하면 여기서부터 재개
    return {"user_response": user_answer}

async def route_parameters(state: ParameterHandlerState):
    """사용자 응답을 각 에이전트에게 분배"""
    extracted = state.get("extracted_params", {})
    pending = state.get("pending_requests", {})
    agent_params = {}

    for agent_name, request in pending.items():
        # 에이전트별로 필요한 파라미터만 추출하여 전달
        agent_params[agent_name] = {
            "parameters": extract_for_agent(extracted, request),
            "a2a_task_id": request.get("a2a_task_id"),      # 동일 세션 유지
            "a2a_context_id": request.get("a2a_context_id"), # A2A 컨텍스트 유지
        }

    return {"agent_params": agent_params}
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
27
28
29
30
31
32
33
34
35
36

파라미터 라우팅 후, 각 에이전트는 기존 A2A taskId/contextId로 **재구독(resubscribe)**하여 세션을 이어갑니다. 에이전트 입장에서는 마치 "잠깐 사용자 입력을 기다렸다가 이어서 처리"하는 것과 같습니다.
| Executor의 input_required 감지

서브에이전트가 A2A 프로토콜로 input_required 상태를 반환하면, Task Executor가 이를 감지합니다.

Python 
# Task Executor: A2A 응답에서 input_required 감지

if status == "input-required":
    # A2A Task의 메시지에서 질문 텍스트와 필수 파라미터 추출
    status_obj = result.get("status", {})
    message = status_obj.get("message", {})
    metadata = message.get("metadata", {})
    required_params = metadata.get("requiredParameters", [])

    return TaskExecutorState(
        status="input-required",
        required_params=required_params,
        param_message=extract_question_text(message),
        # A2A 세션 유지를 위한 ID 보존
        a2a_task_id=result.get("id"),
        a2a_context_id=result.get("contextId"),
    )
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
실제 HITL 동작 예시 (금융 시나리오) ▾ 
TypeScript 
[1] BankAgent → ORCHESTRATOR: Parameter Request
    Status: input-required
    Required Parameters:
      - account_type: "조회할 계좌 종류 (예: 일반예금, 적금)"
      - currency: "통화 단위 (예: 원, 달러)"

[2] Parameter Handler → interrupt() 호출
    → 워크플로우 중단, 사용자 입력 대기

[3] 사용자 응답: "일반예금, 원"

[4] Command(resume) → BankAgent에 A2A resubscribe
    → 동일 taskId/contextId로 파라미터 전달
    → 잔고 조회 완료
1
2
3
4
5
6
7
8
9
10
11
12
13
14


효과
14.jpg



5. 컨텍스트 조율: A2A Context Builder와 파일 라우팅
문제 상황

에이전트 간 직접 통신 구조에서는 아래와 같은 문제가 있습니다. 


  • Agent C가 Agent A의 결과를 필요로 하면, Agent A에게 직접 요청해야 함

  • Agent D도 같은 정보가 필요하면 또 요청해야 함

  • 텍스트뿐만 아니라 **파일(CSV, Excel 등)**도 에이전트 간 전달해야 함


결과적으로 동일 정보를 여러 번 요청하여 토큰이 낭비되고, 파일 전달 메커니즘이 없어 수동으로 처리해야 합니다.

해결 방식 : 구조화된 컨텍스트 빌더

오케스트레이터가 이전 그룹의 결과를 구조화된 마크다운 프롬프트로 변환하여 다음 에이전트에게 전달합니다. 이때 A2A 프로토콜의 Opaque Execution 원칙(내부 상태가 아닌 출력만 공유)을 준수합니다.


15.jpg



| 컨텍스트 빌더 구현

Python 
class A2AContextPromptBuilder:
    @staticmethod
    def build_task_with_context(
        current_task_description: str,
        previous_groups: List[Dict],     # 이전 그룹 결과
        session_parameters: Dict,        # 세션 파라미터
        memory_context: Optional[str],   # MCP 메모리
        max_history_groups: int = 5      # 토큰 절약: 최근 5개만
    ) -> str:
        """
        구조화된 마크다운 프롬프트 생성.
        A2A TextPart로만 전달 (커스텀 필드 없이 표준 준수)
        """
        sections = []

        # 1. 현재 태스크 설명
        sections.append(f"# Current Task\n{current_task_description}")

        # 2. 이전 그룹 결과 (최근 N개만, 토큰 최적화)
        for group in previous_groups[-max_history_groups:]:
            sections.append(format_group_results(group))

        # 3. 세션 파라미터 (사용자가 이전에 제공한 정보)
        if session_parameters:
            sections.append(format_session_params(session_parameters))

        # 4. 메모리 컨텍스트 (MCP 메모리에서 조회한 사용자 정보)
        if memory_context:
            sections.append(f"# Additional Context\n{memory_context}")

        return "\n\n".join(sections)

    @staticmethod
    def build_lightweight_context(previous_groups) -> str:
        """경량 모드: 그룹 요약만 포함하여 토큰 대폭 절약"""
        summaries = [f"Group {g['id']}: {g['summary']}" for g in previous_groups]
        return "\n".join(summaries)
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
27
28
29
30
31
32
33
34
35
36
37


| 토큰 최적화 전략

  • max_history_groups=5: 오래된 그룹은 요약만 포함

  • _strip_duplicate_sections(): 재귀적으로 포함된 이전 컨텍스트 제거 (에이전트 A의 응답에 이미 포함된 컨텍스트가 에이전트 B에 중복 전달되는 것 방지)

  • build_lightweight_context(): 토큰 예산이 적을 때 요약 모드 사용
| 파일 라우팅 시스템 (FileRouter)

텍스트뿐만 아니라 **파일(CSV, Excel, 이미지 등)**도 에이전트 간 전달해야 합니다. A2A Protocol의 FilePart를 활용한 파일 라우팅 시스템을 구현했습니다.

16.jpg



| 5단계 라이프사이클

  1. 등록 : 사용자가 첨부한 파일을 file_registry에 등록
  2. 계획 : Planner가 어떤 태스크에 어떤 파일을 전달할지 결정 (file_plan)
  3. 라우팅 : Send API로 태스크를 dispatch할 때 필요한 파일을 A2A FilePart로 주입
  4. 소비 : 태스크가 파일을 처리하면 소비 완료로 마킹
  5. 정리 : 모든 소비자가 처리한 파일은 메모리에서 제거



| Send API와 경쟁 조건 방지

병렬 dispatch에서 파일 라우팅의 까다로운 점은 경쟁 조건입니다. 여러 태스크가 동시에 같은 파일을 요청하면 중복 전달되거나, 한 태스크가 소비를 마킹한 후 다른 태스크가 파일을 찾지 못할 수 있습니다.


이를 해결하기 위해 Send 객체를 생성하기 전에 모든 파일 배정을 사전 계산합니다.

Python 
# Group Executor: dispatch 전에 파일 배정 사전 계산
def create_send_tasks(state: GroupExecutorState) -> List[Send]:
    # ★ 핵심: Send 전에 모든 파일 배정을 원자적으로 계산
    prepared_files = file_router.prepare_files_for_parallel_dispatch(
        tasks=state["tasks"],
        file_plan=state.get("file_routing", {}).get("file_plan", {})
    )

    sends = []
    for task in state["tasks"]:
        task_state = TaskExecutorState(
            # ... 기본 정보 ...
            fileparts=prepared_files.get(task["task_id"], []),  # 사전 계산된 파일
        )
        sends.append(Send("execute_task", task_state))

    # 전달된 파일들을 일괄 소비 마킹 (원자적)
    file_router.mark_files_consumed_batch(prepared_files)
    return sends
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

prepare_files_for_parallel_dispatch()가 모든 파일 배정을 한 번에 계산하고, mark_files_consumed_batch()가 원자적으로 마킹하여 경쟁 조건을 방지합니다.


실제 컨텍스트 전파 예시 (상권분석 시나리오) ▾ 

TypeScript 
[Group 0] CommercialDistrictAgent
    │
    │  컨텍스트 빌더 → 구조화된 마크다운 생성:
    │  "# Previous Results
    │   ## Group 0 - CommercialDistrictAgent
    │   **Result**: 추천 지역: 건대입구, 월 임대료: 300-500만원, 유동인구: 10만명/일"
    ▼
[Group 1] SmallBusinessConsultingAgent
    │
    │  "# Previous Results
    │   ## Group 0, Group 1 결과 포함
    │   총 창업비용: 1.4억원, 월 운영비: 1,500만원, 손익분기점: 6개월"
    ▼
[Group 2] B금융상품추천에이전트
    │
    │  계산: 대출 필요액 = 1.4억원 - 0.1억원 = 1.3억원
    ▼
[최종 응답]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18


효과
17.jpg



6. 결과 종합: Finalizer로 일관된 응답 생성

문제 상황

여러 에이전트가 각각 결과를 반환합니다. 


  • BankAgent: "잔고 5,234만원"

  • StockAgent: "A전자 50주 추천"

  • CryptoAgent: "비트코인 0.03BTC 추천"


이 결과들을 그대로 나열하면 사용자에게 혼란을 줍니다. "그래서 어떻게 하라는 건가요?"

해결 방식 : Finalizer가 모든 태스크 결과를 수집하고, 상황에 따라 적절히 처리합니다. 

  • 단일 태스크: 결과를 그대로 반환 (LLM 호출 불필요)

  • 복수 태스크: LLM이 결과를 통합하여 자연스러운 응답 생성

  • 에러 발생: 성공한 결과와 에러 정보를 함께 전달

TypeScript 
async def finalizer(state: OrchestratorState) -> dict:
    tasks = dict(state.get("tasks") or {})

    # Case 1: 단일 태스크 완료, 에러 없음 → 그대로 반환
    # 불필요한 LLM 호출을 피해 비용과 지연 시간 절약
    if len(completed_set) == 1 and not errored_map:
        artifact = extract_single_artifact(tasks)
        if artifact:
            return {"final_answer": artifact}

    # Case 2: 복수 태스크 또는 에러 발생 → LLM으로 통합
    # 각 태스크의 결과를 수집하여 LLM에 전달
    llm_context = build_context_for_llm(tasks, plan)

    # LLM이 결과를 분석하고 일관된 응답 생성
    # "잔고 5,234만원 중 투자 가능 금액을 분석하여..."
    final_text = await llm.ainvoke({
        "input": {"ctx": llm_context}
    })

    return {
        "final_answer": {
            "response_type": "text",
            "is_task_complete": True,
            "content": final_text,
        }
    }
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
27


 효과

상황

처리 방식

이점

단일 태스크

결과 직접 반환

LLM 호출 비용/시간 절약

복수 태스크

LLM이 통합

산발적 결과를 일관된 응답으로

에러 발생

성공 결과 + 에러 설명

부분 실패해도 유용한 정보 제공



의도 분류: 불필요한 에이전트 위임 방지

지금까지 설명한 6가지 기능은 "에이전트를 어떻게 잘 실행하느냐"에 대한 것이었습니다. 하지만 프로덕션 운영에서 발견한 또 다른 중요한 문제가 있습니다: 모든 사용자 메시지가 에이전트 위임을 필요로 하지는 않는다는 것입니다.

문제 상황

TypeScript 
사용자: "안녕하세요"
→ Planner 호출 → Agent Hub 검색 → 에이전트 0개 매칭 → 빈 계획 → 실패
→ 불필요한 LLM 호출 3회, 약 5초 소요
​
사용자: "일반예금, 원" (이전 파라미터 질문에 대한 답변)
→ Planner 호출 → "일반예금" 관련 에이전트 검색 → 잘못된 계획 생성
→ 기존 워크플로우와 관계없는 새 계획 수립
1
2
3
4
5
6
7

"안녕하세요" 같은 인사에 백개가 넘는 에이전트를 검색할 필요가 없고, 파라미터 응답을 새로운 요청으로 처리하면 기존 워크플로우가 끊깁니다.

해결 방식 : 4-way Intent Classification

사용자 메시지를 먼저 의도 분류하여 적절한 처리 경로로 라우팅합니다. 

18.jpg

의도

설명

처리 경로

예시

➀ new_request

새로운 에이전트 작업 필요

Planner → DAG 생성

"주식 포트폴리오 추천해줘"

➁ parameter_response

이전 파라미터 질문에 대한 답변

Parameter Handler 재개

"일반예금, 원"

➂ continuation_request

기존 작업의 연속

기존 워크플로우 이어서
실행

"좀 더 보수적으로 바꿔줘"

➃ general_answer

에이전트 위임 불필요

일반 LLM 직접 응답

"안녕하세요", "감사합니다"


| 단축 경로 (Short-circuit)

모든 의도 분류에 LLM을 호출하면 지연이 발생합니다. 명확한 경우에는 LLM 호출 없이 즉시 결정합니다

TypeScript 
async def determine_intent(state: HierarchicalOrchestratorState):
    session = state.get("active_session", {})

    # ★ 단축 경로 1: 파라미터 대기 중이면 무조건 parameter_response
    if session.get("pending_parameter_requests"):
        return {"intent": "parameter_response"}

    # ★ 단축 경로 2: 일반 대화 패턴 감지
    message = state.get("user_message", "").strip()
    if is_general_conversation(message):  # 인사, 감사, 작별 등
        return {"intent": "general_answer"}

    # 그 외: LLM으로 의도 분류
    intent = await classify_with_llm(message, session)
    return {"intent": intent}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15


| 단축 경로의 효과:

  • 파라미터 대기 상태에서는 LLM 호출 0회 (즉시 parameter_response)

  • 일반 인사에는 에이전트 검색 0회 (즉시 general_answer)

  • LLM 비용 절약 + 응답 속도 향상
| 대화 경계 탐지

continuation_request와 new_request를 구분하는 것은 까다롭습니다. "좀 더 보수적으로 바꿔줘"는 이전 투자 포트폴리오를 수정하라는 연속 요청인데, 문맥을 모르면 새 요청으로 오인할 수 있습니다.


이를 해결하기 위해 유사도 스코어링 기반 대화 경계 탐지를 적용합니다. 현재 메시지와 이전 세션의 주제 사이 유사도를 계산하여, 높으면 연속(continuation), 낮으면 새 세션(new_request)으로 판단합니다.


효과
19.jpg




현재 한계점

Agent Orchestrator가 많은 문제를 해결했지만, 여전히 과제가 있습니다.


1. 서브에이전트 내부 행동 검증

TypeScript 
오케스트레이터: "뉴스 검색해줘"
서브에이전트: "네, 검색했습니다" (실제로 했는지?)

→ 서브에이전트가 "했다"고 말하면, 실제로 했는지 확인할 방법이 없음
1
2
3
4

원인 : 서브에이전트는 A2A 프로토콜을 통해 통신하는 블랙박스입니다. 내부 동작을 관찰하거나 강제할 수 없습니다.


2. 내용적 오류 검증

분석 에이전트: "A전자 주가가 10만원입니다" → 이 정보의 정확성을 검증하는 메커니즘 부재


원인: 도메인별로 검증 로직이 다르고, 범용 검증 시스템을 만들기 어렵습니다. "팩트체크 에이전트" 추가 방안 검토 중입니다.


3. 에러 처리: Graceful Degradation

프로덕션에서 에이전트 실패는 불가피합니다. 이를 위해 여러 안전장치를 구현했지만, 완벽하지는 않습니다.


  • A2A Strict Validation : Task Executor에서 A2A 프로토콜 준수를 강제 검증합니다. 비표준 상태값이나 잘못된 응답 형식은 즉시 에러로 처리됩니다.

  • 크리티컬 에러 카운터 : 특정 워크플로우에서 에러가 3회 누적되면 전체를 Graceful abort합니다. 무한 재시도로 리소스가 낭비되는 것을 방지합니다.


# 에러 누적 → Graceful Abort if critical_error_count >= 3: return {"status": "aborted", "reason": "Critical error threshold exceeded"}


  • General Answer Fallback : 에이전트 위임이 실패하면 일반 LLM 응답으로 폴백하여 사용자에게 최소한의 응답을 제공합니다.


하지만 이 안전장치들은 "에이전트가 실패했을 때 다른 에이전트로 대체"하는 자동 복구 메커니즘은 아직 없습니다. 현재는 실패한 태스크를 그대로 보고하고, 성공한 태스크의 결과만으로 최선의 응답을 생성합니다.


4. 중앙화의 구조적 한계

20.jpg

구조적 문제

  • SPOF: 오케스트레이터가 죽으면 전체 시스템 마비

  • 확장성 병목: 에이전트가 많아지면 오케스트레이터 부하 증가

  • 정보 집중: 모든 컨텍스트가 오케스트레이터를 통과


Multi-Agent Collaboration via Evolving Orchestration 연구에서도 이러한 정적 오케스트레이션의 한계를 지적하며, 동적 조율의 필요성을 강조합니다.




다음 연구: AgentNet과 탈중앙화

중앙 오케스트레이터의 한계를 넘어서기 위해, 분산형 멀티에이전트 시스템도 연구하고 있습니다.


AgentNet 개요

AgentNet은 중앙 오케스트레이터 없이 에이전트들이 자율적으로 협업하는 분산형 시스템입니다.
(NeurIPS 2025 Accepted, 상하이 자오퉁 대학)

21.jpg


AgentNet의 핵심 혁신

AgentNet GitHub에서 제시하는 세 가지 핵심 혁신

23.jpg


Self-Evolving 4대 요소

요소

설명

역할

RAG 메모리

각 에이전트가 과거 경험 저장

유사 태스크 처리 방법 참조

연결 가중치

협업 성공률 기반 신뢰도 점수 (0~1)

협업 대상 선택 기준

능력 벡터

각 에이전트의 전문성 수치화

적합한 에이전트 매칭

동적 토폴로지

가중치 기반 연결 자동 조정

네트워크 자동 최적화


오케스트레이터 vs AgentNet 비교

관점

오케스트레이터

AgentNet

장애 대응

SPOF 위험

다른 경로로 우회

확장성

중앙 병목 가능

수평 확장 용이

프라이버시

중앙에 정보 집중

각자 로컬 저장

전문화

정적 역할 배정

경험 기반 자동 전문화

성능

일관된 품질

단일/중앙 집중 시스템보다 높은
태스크 정확도


AgentNet의 Trade-off

AgentNet도 만능은 아닙니다. 


  • 콜드 스타트: 초기에 연결 가중치가 없어서 비효율적

  • 분산 합의: 복잡한 작업에서 합의 도달이 어려움

  • 디버깅: 중앙 로그가 없어서 문제 추적이 어려움


오케스트레이터와 AgentNet의 장점을 결합하는 하이브리드 구조도 검토하고 있습니다.




정리

멀티에이전트 시스템을 만드는 건 쉬워졌습니다. 하지만 프로덕션에서 안정적으로 운영하는 건 여전히 어렵습니다.


| Agent Orchestrator 개발을 통해 정리한 핵심 사항

22.jpg


한계도 인정합니다. 

  • 서브에이전트 내부 행동 검증은 아직 해결하지 못했습니다

  • 중앙화 구조의 SPOF 문제가 있습니다

  • 에이전트 실패 시 자동 대체 메커니즘이 없습니다

  • 그래서 AgentNet 같은 분산형 접근도 연구하고 있습니다


멀티에이전트 시스템은 아직 초기 단계이고, 저희도 계속 실험하고 배우고 있습니다. 이 글이 비슷한 고민을 하시는 분들께 참고가 되었으면 합니다.




참고 자료

핵심 프레임워크

학술 논문

실무 가이드

홍지후, 송윤성

LLM 기반 AI Agent 기술 연구 및 개발을 담당하고 있습니다.

←이전 글
다음 글→