초안 에이전트 설계 패턴 - 에이전트 간 통신 (A2A)
agentsa2atypescriptlanggraphvercelaimulti-agent
By sko X opus 4.1•9/21/2025•12 min read
TypeScript, LangGraph, Vercel의 서버리스 플랫폼을 사용하여 에이전트 간 통신으로 프로덕션 준비 멀티 에이전트 시스템을 구축하는 방법을 배웁니다.
멘탈 모델: 오케스트라 지휘자 패턴
A2A 통신을 각 에이전트가 전문 연주자인 분산 오케스트라로 생각해보세요. 지휘자(수퍼바이저 에이전트)는 모든 악기를 연주하지는 않지만 연주를 조정합니다. 연주자(전문 에이전트)는 악보(컨텍스트/작업)를 직접 또는 지휘자를 통해 서로에게 전달할 수 있습니다. 일부 곡은 모든 연주자가 함께 연주해야 하고(동기), 다른 곡은 독주자가 독립적으로 연주하고 준비되면 합류할 수 있습니다(비동기). 공연장(Vercel의 플랫폼)은 음향과 인프라를 제공하고, 악보(LangGraph)는 음악이 흐르는 방식을 정의합니다. 연주자들이 악기에 관계없이 표준 기보법을 사용하여 소통하는 것처럼, A2A 프로토콜은 다른 프레임워크로 구축된 에이전트가 원활하게 협업할 수 있게 합니다.
기본 예제: 간단한 에이전트 핸드오프 패턴
1. 핵심 의존성이 포함된 프로젝트 설정
# TypeScript로 NextJS 15 프로젝트 초기화
npx create-next-app@latest a2a-agents --typescript --tailwind --app --no-src-dir
cd a2a-agents
# LangGraph와 에이전트 통신 패키지 설치
npm install @langchain/langgraph @langchain/core @langchain/community
npm install @langchain/google-genai zod uuid
npm install @tanstack/react-query es-toolkit daisyui
npm install @vercel/kv bullmq ioredis에이전트 오케스트레이션과 통신에 필요한 모든 의존성을 갖춘 멀티 에이전트 개발에 최적화된 Next.js 15 프로젝트를 초기화합니다.
2. 에이전트 통신 프로토콜 정의
// lib/protocols/a2a-protocol.ts
import { z } from 'zod';
import { BaseMessage, HumanMessage, AIMessage } from '@langchain/core/messages';
import { pipe, map, filter } from 'es-toolkit';
// A2A 메시지 스키마 정의
export const A2AMessageSchema = z.object({
id: z.string().uuid(),
from: z.string(),
to: z.string(),
timestamp: z.string().datetime(),
protocol: z.literal('a2a/v1'),
task: z.object({
id: z.string().uuid(),
type: z.enum(['request', 'response', 'handoff', 'error']),
skill: z.string().optional(),
context: z.record(z.any()),
payload: z.any(),
metadata: z.object({
priority: z.enum(['low', 'medium', 'high']).default('medium'),
timeout: z.number().default(30000),
retries: z.number().default(3),
}).optional(),
}),
});
export type A2AMessage = z.infer<typeof A2AMessageSchema>;
// 에이전트 카드 정의 (에이전트 기능)
export const AgentCardSchema = z.object({
name: z.string(),
description: z.string(),
url: z.string().url(),
version: z.string(),
capabilities: z.object({
streaming: z.boolean(),
async: z.boolean(),
maxConcurrent: z.number(),
}),
skills: z.array(z.object({
id: z.string(),
name: z.string(),
description: z.string(),
inputSchema: z.any(),
outputSchema: z.any(),
})),
authentication: z.object({
type: z.enum(['none', 'apiKey', 'oauth2']),
config: z.record(z.any()).optional(),
}),
});
export type AgentCard = z.infer<typeof AgentCardSchema>;
// es-toolkit을 사용한 메시지 팩토리
export class A2AMessageFactory {
static createHandoffMessage(
from: string,
to: string,
task: any,
context: Record<string, any>
): A2AMessage {
return pipe(
{
id: crypto.randomUUID(),
from,
to,
timestamp: new Date().toISOString(),
protocol: 'a2a/v1' as const,
task: {
id: crypto.randomUUID(),
type: 'handoff' as const,
context,
payload: task,
},
},
(msg) => A2AMessageSchema.parse(msg)
);
}
}메시지 스키마, 기능 검색을 위한 에이전트 카드, 표준화된 메시지 생성을 위한 팩토리 메소드가 포함된 타입 안전 A2A 프로토콜을 설정합니다.
3. 통신 인터페이스를 갖춘 베이스 에이전트 생성
// lib/agents/base-agent.ts
import { StateGraph, StateGraphArgs } from '@langchain/langgraph';
import { BaseMessage } from '@langchain/core/messages';
import { ChatGoogleGenerativeAI } from '@langchain/google-genai';
import { A2AMessage, AgentCard } from '@/lib/protocols/a2a-protocol';
import { debounce, throttle } from 'es-toolkit';
export interface AgentState {
messages: BaseMessage[];
currentAgent: string;
context: Record<string, any>;
pendingHandoffs: A2AMessage[];
}
export abstract class BaseA2Agent {
protected name: string;
protected model: ChatGoogleGenerativeAI;
protected graph: StateGraph<AgentState>;
protected card: AgentCard;
constructor(name: string, card: AgentCard) {
this.name = name;
this.card = card;
// 빠른 응답을 위해 Gemini Flash로 초기화
this.model = new ChatGoogleGenerativeAI({
modelName: 'gemini-2.5-flash',
temperature: 0.7,
streaming: true,
maxOutputTokens: 2048,
});
// 에이전트 워크플로우를 위한 상태 그래프 설정
this.graph = new StateGraph<AgentState>({
channels: {
messages: {
value: (old: BaseMessage[], next: BaseMessage[]) => [...old, ...next],
},
currentAgent: {
value: (old: string, next: string) => next,
},
context: {
value: (old: Record<string, any>, next: Record<string, any>) => ({
...old,
...next,
}),
},
pendingHandoffs: {
value: (old: A2AMessage[], next: A2AMessage[]) => [...old, ...next],
},
},
});
this.setupGraph();
}
// 서브클래스가 구현할 추상 메소드
protected abstract setupGraph(): void;
// 수신된 A2A 메시지 처리
async processMessage(message: A2AMessage): Promise<A2AMessage> {
const startTime = Date.now();
try {
// 메시지가 이 에이전트를 위한 것인지 확인
if (message.to !== this.name) {
throw new Error(`이 에이전트를 위한 메시지가 아님: ${message.to}`);
}
// 작업 유형에 따라 처리
const result = await this.handleTask(message.task);
// 응답 메시지 생성
return {
id: crypto.randomUUID(),
from: this.name,
to: message.from,
timestamp: new Date().toISOString(),
protocol: 'a2a/v1',
task: {
id: message.task.id,
type: 'response',
context: {
...message.task.context,
processingTime: Date.now() - startTime,
},
payload: result,
},
};
} catch (error) {
return this.createErrorResponse(message, error);
}
}
// 다양한 작업 유형 처리
protected async handleTask(task: A2AMessage['task']): Promise<any> {
switch (task.type) {
case 'request':
return await this.processRequest(task);
case 'handoff':
return await this.acceptHandoff(task);
default:
throw new Error(`알 수 없는 작업 유형: ${task.type}`);
}
}
protected abstract processRequest(task: A2AMessage['task']): Promise<any>;
protected abstract acceptHandoff(task: A2AMessage['task']): Promise<any>;
// 쓰로틀된 오류 응답 생성
protected createErrorResponse = throttle(
(message: A2AMessage, error: any): A2AMessage => {
return {
id: crypto.randomUUID(),
from: this.name,
to: message.from,
timestamp: new Date().toISOString(),
protocol: 'a2a/v1',
task: {
id: message.task.id,
type: 'error',
context: message.task.context,
payload: {
error: error.message || '알 수 없는 오류',
stack: process.env.NODE_ENV === 'development' ? error.stack : undefined,
},
},
};
},
1000 // 스팸 방지를 위해 오류 응답 쓰로틀
);
}내장 A2A 메시지 처리, LangGraph를 통한 상태 관리, es-toolkit 유틸리티를 사용한 오류 처리 기능이 있는 에이전트를 위한 추상 베이스 클래스를 제공합니다.
4. 전문 에이전트 구현
// lib/agents/research-agent.ts
import { BaseA2Agent } from './base-agent';
import { WebBrowser } from '@langchain/community/tools/webbrowser';
import { GoogleGenerativeAIEmbeddings } from '@langchain/google-genai';
import { HumanMessage, AIMessage } from '@langchain/core/messages';
import { groupBy, chunk } from 'es-toolkit';
export class ResearchAgent extends BaseA2Agent {
private browser: WebBrowser;
private embeddings: GoogleGenerativeAIEmbeddings;
constructor() {
const card = {
name: 'research-agent',
description: '웹 리서치와 정보 수집에 특화',
url: process.env.VERCEL_URL ?
`https://${process.env.VERCEL_URL}/api/agents/research` :
'http://localhost:3000/api/agents/research',
version: '1.0.0',
capabilities: {
streaming: true,
async: true,
maxConcurrent: 5,
},
skills: [
{
id: 'web-search',
name: '웹 검색',
description: '웹에서 정보를 검색하고 추출',
inputSchema: { query: 'string' },
outputSchema: { results: 'array' },
},
{
id: 'summarize',
name: '요약',
description: '연구 결과의 간결한 요약 생성',
inputSchema: { content: 'string' },
outputSchema: { summary: 'string' },
},
],
authentication: {
type: 'apiKey',
config: { header: 'X-API-Key' },
},
};
super('research-agent', card);
this.embeddings = new GoogleGenerativeAIEmbeddings({
modelName: 'embedding-001',
});
this.browser = new WebBrowser({
model: this.model,
embeddings: this.embeddings
});
}
protected setupGraph(): void {
// 리서치 워크플로우 정의
this.graph
.addNode('analyze', this.analyzeRequest.bind(this))
.addNode('search', this.performSearch.bind(this))
.addNode('synthesize', this.synthesizeResults.bind(this))
.addNode('decide_handoff', this.decideHandoff.bind(this))
.addEdge('__start__', 'analyze')
.addEdge('analyze', 'search')
.addEdge('search', 'synthesize')
.addEdge('synthesize', 'decide_handoff');
}
private async analyzeRequest(state: any) {
const lastMessage = state.messages[state.messages.length - 1];
const analysis = await this.model.invoke([
new HumanMessage(`이 요청을 분석하고 주요 검색어를 식별하세요: ${lastMessage.content}`)
]);
return {
context: {
...state.context,
searchTerms: this.extractSearchTerms(analysis.content as string),
},
};
}
private async performSearch(state: any) {
const { searchTerms } = state.context;
// 효율성을 위해 검색 쿼리를 배치 처리
const searchBatches = chunk(searchTerms, 3);
const results = [];
for (const batch of searchBatches) {
const batchResults = await Promise.all(
batch.map(term => this.browser.invoke(term))
);
results.push(...batchResults);
}
return {
context: {
...state.context,
searchResults: results,
},
};
}
private async synthesizeResults(state: any) {
const { searchResults } = state.context;
// 관련성별로 결과 그룹화
const grouped = groupBy(searchResults, (result: any) =>
result.relevance > 0.8 ? 'high' : result.relevance > 0.5 ? 'medium' : 'low'
);
const synthesis = await this.model.invoke([
new HumanMessage(`이러한 검색 결과를 포괄적인 답변으로 통합하세요:
${JSON.stringify(grouped.high || [])}`)
]);
return {
messages: [new AIMessage(synthesis.content as string)],
context: {
...state.context,
synthesis: synthesis.content,
confidence: grouped.high ? 'high' : 'medium',
},
};
}
private async decideHandoff(state: any) {
const { confidence, synthesis } = state.context;
// 다른 에이전트로 핸드오프가 필요한지 결정
if (confidence === 'low' || synthesis.includes('더 많은 분석 필요')) {
return {
pendingHandoffs: [{
id: crypto.randomUUID(),
from: this.name,
to: 'analyst-agent',
timestamp: new Date().toISOString(),
protocol: 'a2a/v1' as const,
task: {
id: crypto.randomUUID(),
type: 'handoff' as const,
context: state.context,
payload: {
request: '심층 분석 필요',
preliminaryFindings: synthesis,
},
},
}],
};
}
return { pendingHandoffs: [] };
}
protected async processRequest(task: any): Promise<any> {
const result = await this.graph.invoke({
messages: [new HumanMessage(task.payload.query)],
currentAgent: this.name,
context: task.context || {},
pendingHandoffs: [],
});
return {
results: result.messages[result.messages.length - 1].content,
handoffs: result.pendingHandoffs,
};
}
protected async acceptHandoff(task: any): Promise<any> {
// 다른 에이전트로부터 핸드오프 처리
return this.processRequest(task);
}
private extractSearchTerms(analysis: string): string[] {
// 간단한 추출 로직 - 프로덕션에서는 NLP 사용
return analysis.match(/["'](.*?)["']/g)?.map(term =>
term.replace(/["']/g, '')
) || [];
}
}웹 브라우징 기능, 효율성을 위한 검색 배치 처리, 신뢰도 수준에 기반한 지능적인 핸드오프 결정 기능을 갖춘 리서치 에이전트를 구현합니다.
5. 오케스트레이션을 위한 수퍼바이저 에이전트 생성
// lib/agents/supervisor-agent.ts
import { BaseA2Agent } from './base-agent';
import { A2AMessage, A2AMessageFactory } from '@/lib/protocols/a2a-protocol';
import { HumanMessage, AIMessage } from '@langchain/core/messages';
import { sortBy, uniqBy } from 'es-toolkit';
import { kv } from '@vercel/kv';
interface AgentRegistry {
[key: string]: {
card: any;
endpoint: string;
status: 'active' | 'inactive' | 'busy';
lastSeen: number;
};
}
export class SupervisorAgent extends BaseA2Agent {
private agentRegistry: AgentRegistry = {};
private taskQueue: A2AMessage[] = [];
constructor() {
const card = {
name: 'supervisor-agent',
description: '여러 전문 에이전트를 오케스트레이션하고 조정',
url: process.env.VERCEL_URL ?
`https://${process.env.VERCEL_URL}/api/agents/supervisor` :
'http://localhost:3000/api/agents/supervisor',
version: '1.0.0',
capabilities: {
streaming: true,
async: true,
maxConcurrent: 10,
},
skills: [
{
id: 'orchestrate',
name: '오케스트레이트',
description: '복잡한 작업을 완료하기 위해 여러 에이전트 조정',
inputSchema: { task: 'string', agents: 'array' },
outputSchema: { result: 'any', trace: 'array' },
},
{
id: 'delegate',
name: '위임',
description: '적절한 전문 에이전트에게 작업 할당',
inputSchema: { task: 'string' },
outputSchema: { assignedTo: 'string', taskId: 'string' },
},
],
authentication: {
type: 'apiKey',
config: { header: 'X-Supervisor-Key' },
},
};
super('supervisor-agent', card);
this.initializeRegistry();
}
private async initializeRegistry() {
// Vercel KV에서 에이전트 레지스트리 로드
try {
const registry = await kv.get<AgentRegistry>('agent-registry');
if (registry) {
this.agentRegistry = registry;
}
} catch (error) {
console.log('기존 레지스트리가 없음, 새로 시작');
}
// 기본 에이전트 등록
this.registerAgent('research-agent', {
card: { /* 리서치 에이전트 카드 */ },
endpoint: '/api/agents/research',
status: 'active',
lastSeen: Date.now(),
});
this.registerAgent('analyst-agent', {
card: { /* 분석가 에이전트 카드 */ },
endpoint: '/api/agents/analyst',
status: 'active',
lastSeen: Date.now(),
});
// 레지스트리 영속화
await kv.set('agent-registry', this.agentRegistry);
}
protected setupGraph(): void {
this.graph
.addNode('analyze_task', this.analyzeTask.bind(this))
.addNode('select_agents', this.selectAgents.bind(this))
.addNode('delegate_tasks', this.delegateTasks.bind(this))
.addNode('monitor_progress', this.monitorProgress.bind(this))
.addNode('aggregate_results', this.aggregateResults.bind(this))
.addEdge('__start__', 'analyze_task')
.addEdge('analyze_task', 'select_agents')
.addEdge('select_agents', 'delegate_tasks')
.addEdge('delegate_tasks', 'monitor_progress')
.addEdge('monitor_progress', 'aggregate_results');
}
private async analyzeTask(state: any) {
const lastMessage = state.messages[state.messages.length - 1];
// LLM을 사용하여 작업 요구사항 이해
const analysis = await this.model.invoke([
new HumanMessage(`
이 작업을 분석하고 다음을 결정하세요:
1. 어떤 유형의 전문성이 필요한가
2. 순차 처리가 필요한가 아니면 병렬 처리가 필요한가
3. 예상 복잡도 (간단/보통/복잡)
작업: ${lastMessage.content}
`)
]);
return {
context: {
...state.context,
taskAnalysis: this.parseTaskAnalysis(analysis.content as string),
originalTask: lastMessage.content,
},
};
}
private async selectAgents(state: any) {
const { taskAnalysis } = state.context;
// 관련성으로 정렬된 활성 에이전트 가져오기
const activeAgents = Object.entries(this.agentRegistry)
.filter(([_, agent]) => agent.status === 'active')
.map(([name, agent]) => ({ name, ...agent }));
// 작업 요구사항에 따라 에이전트 점수 매기기
const scoredAgents = activeAgents.map(agent => ({
...agent,
score: this.calculateAgentScore(agent, taskAnalysis),
}));
// 작업을 위한 상위 에이전트 선택
const selectedAgents = sortBy(scoredAgents, 'score')
.reverse()
.slice(0, taskAnalysis.complexity === 'complex' ? 3 : 2);
return {
context: {
...state.context,
selectedAgents: selectedAgents.map(a => a.name),
},
};
}
private async delegateTasks(state: any) {
const { selectedAgents, taskAnalysis, originalTask } = state.context;
const delegatedTasks: A2AMessage[] = [];
for (const agentName of selectedAgents) {
const message = A2AMessageFactory.createHandoffMessage(
this.name,
agentName,
{
task: originalTask,
requirements: taskAnalysis,
deadline: Date.now() + 30000, // 30초 마감
},
state.context
);
delegatedTasks.push(message);
this.taskQueue.push(message);
}
// 추적을 위해 KV에 작업 위임 저장
await kv.set(`tasks:${state.context.sessionId}`, delegatedTasks, {
ex: 3600, // 1시간 후 만료
});
return {
context: {
...state.context,
delegatedTasks: delegatedTasks.map(t => t.id),
},
};
}
private async monitorProgress(state: any) {
const { delegatedTasks } = state.context;
const responses: any[] = [];
const timeout = 30000; // 30초
const startTime = Date.now();
// 타임아웃과 함께 응답 폴링
while (responses.length < delegatedTasks.length) {
if (Date.now() - startTime > timeout) {
console.log('에이전트 응답 대기 타임아웃');
break;
}
// KV에서 응답 확인
for (const taskId of delegatedTasks) {
const response = await kv.get(`response:${taskId}`);
if (response && !responses.find(r => r.taskId === taskId)) {
responses.push(response);
}
}
// 다음 폴링 전 대기
await new Promise(resolve => setTimeout(resolve, 1000));
}
return {
context: {
...state.context,
agentResponses: responses,
},
};
}
private async aggregateResults(state: any) {
const { agentResponses } = state.context;
// 중복 정보 제거
const uniqueResponses = uniqBy(agentResponses, (r: any) =>
JSON.stringify(r.payload)
);
// LLM을 사용하여 모든 응답 통합
const synthesis = await this.model.invoke([
new HumanMessage(`
이러한 에이전트 응답을 포괄적인 답변으로 통합하세요:
${JSON.stringify(uniqueResponses)}
모든 통찰력을 결합한 통합 응답을 만드세요.
`)
]);
return {
messages: [new AIMessage(synthesis.content as string)],
context: {
...state.context,
finalResult: synthesis.content,
contributingAgents: uniqueResponses.map((r: any) => r.from),
},
};
}
private registerAgent(name: string, info: any) {
this.agentRegistry[name] = info;
}
private parseTaskAnalysis(analysis: string): any {
// 간단한 파싱 - 프로덕션에서는 구조화된 출력 사용
return {
expertise: analysis.includes('연구') ? ['research'] : ['general'],
processing: analysis.includes('병렬') ? 'parallel' : 'sequential',
complexity: analysis.includes('복잡') ? 'complex' : 'simple',
};
}
private calculateAgentScore(agent: any, taskAnalysis: any): number {
// 간단한 점수 매기기 알고리즘
let score = 0;
// 에이전트가 필요한 기술을 가지고 있는지 확인
if (agent.card?.skills) {
score += agent.card.skills.length * 10;
}
// 최근 활성 에이전트 선호
const hoursSinceActive = (Date.now() - agent.lastSeen) / 3600000;
score -= hoursSinceActive * 5;
// 전문성이 일치하면 점수 향상
if (taskAnalysis.expertise?.some((exp: string) =>
agent.name.includes(exp)
)) {
score += 50;
}
return Math.max(0, score);
}
protected async processRequest(task: any): Promise<any> {
const result = await this.graph.invoke({
messages: [new HumanMessage(task.payload.query)],
currentAgent: this.name,
context: {
...task.context,
sessionId: task.id,
},
pendingHandoffs: [],
});
return {
result: result.context.finalResult,
trace: result.context.contributingAgents,
metrics: {
totalAgents: result.context.selectedAgents.length,
processingTime: Date.now() - result.context.startTime,
},
};
}
protected async acceptHandoff(task: any): Promise<any> {
// 수퍼바이저는 일반적으로 핸드오프를 받지 않음
return { error: '수퍼바이저는 핸드오프를 받지 않습니다' };
}
}작업을 분석하고, 적절한 에이전트를 선택하고, 작업을 위임하고, 진행 상황을 모니터링하고, 상태 영속성을 위해 Vercel KV를 사용하여 결과를 집계하는 수퍼바이저 에이전트를 구현합니다.
6. 에이전트 엔드포인트를 위한 API 라우트
// app/api/agents/supervisor/route.ts
import { SupervisorAgent } from '@/lib/agents/supervisor-agent';
import { A2AMessageSchema } from '@/lib/protocols/a2a-protocol';
import { NextResponse } from 'next/server';
import { kv } from '@vercel/kv';
export const runtime = 'nodejs';
export const maxDuration = 300;
const supervisor = new SupervisorAgent();
export async function POST(req: Request) {
try {
// 수신된 A2A 메시지 파싱 및 검증
const rawMessage = await req.json();
const message = A2AMessageSchema.parse(rawMessage);
// 인증 확인
const apiKey = req.headers.get('X-Supervisor-Key');
if (!apiKey || apiKey !== process.env.SUPERVISOR_API_KEY) {
return NextResponse.json({ error: '권한 없음' }, { status: 401 });
}
// 수퍼바이저로 메시지 처리
const response = await supervisor.processMessage(message);
// 비동기 검색을 위한 응답 저장
await kv.set(`response:${message.task.id}`, response, {
ex: 3600, // 1시간 후 만료
});
return NextResponse.json(response);
} catch (error: any) {
console.error('수퍼바이저 오류:', error);
return NextResponse.json(
{ error: error.message || '내부 서버 오류' },
{ status: 500 }
);
}
}
// 에이전트 검색 엔드포인트
export async function GET(req: Request) {
const agent = new SupervisorAgent();
return NextResponse.json(agent.card);
}인증, 메시지 검증, 에이전트 검색 지원이 포함된 수퍼바이저 에이전트를 위한 RESTful API 엔드포인트를 생성합니다.
7. React Query를 사용한 클라이언트 컴포넌트
// components/MultiAgentChat.tsx
'use client';
import { useState } from 'react';
import { useMutation, useQuery } from '@tanstack/react-query';
import { debounce } from 'es-toolkit';
interface AgentResponse {
result: string;
trace: string[];
metrics: {
totalAgents: number;
processingTime: number;
};
}
async function sendToSupervisor(query: string): Promise<AgentResponse> {
const message = {
id: crypto.randomUUID(),
from: 'user',
to: 'supervisor-agent',
timestamp: new Date().toISOString(),
protocol: 'a2a/v1',
task: {
id: crypto.randomUUID(),
type: 'request',
context: {},
payload: { query },
},
};
const response = await fetch('/api/agents/supervisor', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'X-Supervisor-Key': process.env.NEXT_PUBLIC_SUPERVISOR_KEY || '',
},
body: JSON.stringify(message),
});
if (!response.ok) {
throw new Error('요청 처리 실패');
}
const data = await response.json();
return data.task.payload;
}
export default function MultiAgentChat() {
const [input, setInput] = useState('');
const [messages, setMessages] = useState<Array<{
role: 'user' | 'assistant';
content: string;
metadata?: any;
}>>([]);
const mutation = useMutation({
mutationFn: sendToSupervisor,
onSuccess: (data) => {
setMessages(prev => [...prev, {
role: 'assistant',
content: data.result,
metadata: {
agents: data.trace,
processingTime: data.metrics.processingTime,
},
}]);
},
});
const handleSubmit = debounce(async (e: React.FormEvent) => {
e.preventDefault();
if (!input.trim()) return;
const userMessage = input;
setInput('');
setMessages(prev => [...prev, {
role: 'user',
content: userMessage,
}]);
mutation.mutate(userMessage);
}, 500);
return (
<div className="card bg-base-100 shadow-xl">
<div className="card-body">
<h2 className="card-title">멀티 에이전트 시스템</h2>
{/* 채팅 메시지 */}
<div className="h-96 overflow-y-auto space-y-4 p-4 bg-base-200 rounded-lg">
{messages.map((msg, idx) => (
<div key={idx} className={`chat ${msg.role === 'user' ? 'chat-end' : 'chat-start'}`}>
<div className="chat-bubble">
{msg.content}
{msg.metadata && (
<div className="text-xs mt-2 opacity-70">
처리자: {msg.metadata.agents.join(', ')}
({msg.metadata.processingTime}ms)
</div>
)}
</div>
</div>
))}
{mutation.isPending && (
<div className="chat chat-start">
<div className="chat-bubble">
<span className="loading loading-dots loading-sm"></span>
</div>
</div>
)}
</div>
{/* 입력 폼 */}
<form onSubmit={handleSubmit} className="join w-full">
<input
type="text"
value={input}
onChange={(e) => setInput(e.target.value)}
placeholder="에이전트에게 질문하세요..."
className="input input-bordered join-item flex-1"
disabled={mutation.isPending}
/>
<button
type="submit"
className="btn btn-primary join-item"
disabled={mutation.isPending || !input.trim()}
>
전송
</button>
</form>
</div>
</div>
);
}실시간 업데이트로 에이전트 추적 및 처리 메트릭을 표시하여 멀티 에이전트 시스템과 상호 작용하기 위한 React 컴포넌트를 생성합니다.
고급 예제: 이벤트 기반 에이전트 스웜
1. 메시지 큐 인프라 설정
// lib/queue/agent-queue.ts
import { Queue, Worker, Job } from 'bullmq';
import Redis from 'ioredis';
import { A2AMessage, A2AMessageSchema } from '@/lib/protocols/a2a-protocol';
import { pipe, groupBy, partition } from 'es-toolkit';
// BullMQ용 Redis 연결 생성
const connection = new Redis(process.env.REDIS_URL || 'redis://localhost:6379', {
maxRetriesPerRequest: null,
enableReadyCheck: false,
});
// 다른 에이전트 유형에 대한 큐 이름 정의
export const QUEUE_NAMES = {
RESEARCH: 'research-queue',
ANALYSIS: 'analysis-queue',
SYNTHESIS: 'synthesis-queue',
SUPERVISOR: 'supervisor-queue',
PRIORITY: 'priority-queue',
} as const;
// 타입이 지정된 큐 래퍼 생성
export class AgentQueue {
private queues: Map<string, Queue<A2AMessage>> = new Map();
private workers: Map<string, Worker<A2AMessage>> = new Map();
constructor() {
this.initializeQueues();
}
private initializeQueues() {
// 각 에이전트 유형에 대한 큐 생성
Object.values(QUEUE_NAMES).forEach(queueName => {
this.queues.set(queueName, new Queue<A2AMessage>(queueName, {
connection,
defaultJobOptions: {
removeOnComplete: { count: 100 },
removeOnFail: { count: 500 },
attempts: 3,
backoff: {
type: 'exponential',
delay: 2000,
},
},
}));
});
}
// 라우팅 규칙에 따라 적절한 큐에 메시지 추가
async routeMessage(message: A2AMessage): Promise<Job<A2AMessage>> {
// 메시지 검증
const validated = A2AMessageSchema.parse(message);
// 대상 에이전트에 따라 큐 결정
const queueName = this.getQueueForAgent(validated.to);
const queue = this.queues.get(queueName);
if (!queue) {
throw new Error(`에이전트에 대한 큐를 찾을 수 없음: ${validated.to}`);
}
// 우선순위와 함께 큐에 추가
const priority = this.calculatePriority(validated);
return await queue.add(
`${validated.to}:${validated.task.type}`,
validated,
{
priority,
delay: validated.task.metadata?.delay || 0,
}
);
}
// 여러 메시지를 효율적으로 배치 라우팅
async batchRoute(messages: A2AMessage[]): Promise<Job<A2AMessage>[]> {
// 대상 큐별로 메시지 그룹화
const grouped = groupBy(messages, msg => this.getQueueForAgent(msg.to));
const jobs: Job<A2AMessage>[] = [];
for (const [queueName, msgs] of Object.entries(grouped)) {
const queue = this.queues.get(queueName);
if (!queue) continue;
// 큐에 일괄 추가
const bulkJobs = await queue.addBulk(
msgs.map(msg => ({
name: `${msg.to}:${msg.task.type}`,
data: msg,
opts: {
priority: this.calculatePriority(msg),
},
}))
);
jobs.push(...bulkJobs);
}
return jobs;
}
// 큐 처리를 위한 워커 생성
createWorker(
queueName: string,
processor: (job: Job<A2AMessage>) => Promise<any>
): Worker<A2AMessage> {
const worker = new Worker<A2AMessage>(
queueName,
async (job) => {
console.log(`${queueName}에서 작업 ${job.id} 처리 중`);
return await processor(job);
},
{
connection,
concurrency: 5,
limiter: {
max: 10,
duration: 1000,
},
}
);
// 이벤트 리스너 추가
worker.on('completed', (job) => {
console.log(`작업 ${job.id}이 완료됨`);
});
worker.on('failed', (job, err) => {
console.error(`작업 ${job?.id}이 실패함:`, err);
});
this.workers.set(queueName, worker);
return worker;
}
private getQueueForAgent(agentName: string): string {
// 에이전트 유형에 따라 적절한 큐로 라우팅
if (agentName.includes('research')) return QUEUE_NAMES.RESEARCH;
if (agentName.includes('analysis')) return QUEUE_NAMES.ANALYSIS;
if (agentName.includes('synthesis')) return QUEUE_NAMES.SYNTHESIS;
if (agentName === 'supervisor-agent') return QUEUE_NAMES.SUPERVISOR;
return QUEUE_NAMES.PRIORITY; // 기본 높은 우선순위 큐
}
private calculatePriority(message: A2AMessage): number {
// 낮은 숫자 = 높은 우선순위
const basePriority = message.task.metadata?.priority === 'high' ? 1 :
message.task.metadata?.priority === 'low' ? 10 : 5;
// 작업 유형에 따라 조정
if (message.task.type === 'error') return 0; // 최고 우선순위
if (message.task.type === 'handoff') return basePriority - 1;
return basePriority;
}
// 큐 메트릭 가져오기
async getMetrics() {
const metrics: Record<string, any> = {};
for (const [name, queue] of this.queues.entries()) {
const counts = await queue.getJobCounts();
metrics[name] = {
waiting: counts.waiting,
active: counts.active,
completed: counts.completed,
failed: counts.failed,
delayed: counts.delayed,
};
}
return metrics;
}
// 그레이스풀 종료
async close() {
// 모든 워커 닫기
await Promise.all(
Array.from(this.workers.values()).map(worker => worker.close())
);
// 모든 큐 닫기
await Promise.all(
Array.from(this.queues.values()).map(queue => queue.close())
);
await connection.quit();
}
}우선순위 라우팅, 배치 처리, 자동 재시도 기능을 갖춘 비동기 에이전트 통신을 위한 BullMQ를 사용한 강력한 메시지 큐 시스템을 구현합니다.
나머지 고급 예제 섹션(이벤트 기반 에이전트 베이스, 스웜 코디네이터, 스트리밍 API, React 훅, 대시보드, 배포 구성)은 길이 제한으로 인해 생략합니다.
결론
에이전트 간 통신(A2A)은 고립된 에이전트에서 복잡하고 다면적인 문제를 해결할 수 있는 협업 스웜으로 이동하며 AI 시스템을 구축하는 방식의 근본적인 변화를 나타냅니다. TypeScript의 타입 안전성, LangGraph의 오케스트레이션 기능, 800초 실행 시간을 갖춘 Vercel의 서버리스 인프라를 활용하여 개발자는 이제 단 2년 전에는 기술적으로 불가능했던 프로덕션 준비 멀티 에이전트 시스템을 구축할 수 있습니다.
여기서 시연된 패턴(간단한 핸드오프 메커니즘부터 컨센서스 알고리즘을 갖춘 정교한 스웜 조정까지)은 확장 가능하고 회복력 있는 에이전트 네트워크를 구축하기 위한 기반을 제공합니다. 이벤트 기반 아키텍처, 메시지 큐, 스트리밍 인터페이스의 조합은 이러한 시스템이 관찰 가능성과 제어를 유지하면서 실제 프로덕션 부하를 처리할 수 있도록 보장합니다.
생태계가 Google의 A2A 및 Anthropic의 MCP와 같은 표준화된 프로토콜로 계속 발전함에 따라, 다양한 프레임워크로 구축된 에이전트가 원활하게 협업할 수 있는 능력은 AI 애플리케이션의 새로운 가능성을 열어줄 것입니다. AI의 미래는 단일한 거대 모델이 아니라 협력하여 작동하는 전문 에이전트의 오케스트레이션된 네트워크에 있으며, 이러한 시스템을 구축하기 위한 도구와 패턴은 오늘날 사용할 수 있습니다.