LangGraph · A2A로 멀티 에이전트 오케스트레이션 구현하기 — 공유 메모리 설계부터 실전 코드까지

[에이전트 A] ──A2A──▶ [에이전트 B]
     │                      │
    MCP                    MCP
     │                      │
     ▼                      ▼
[도구/DB/API]         [도구/DB/API]

# Claude Code 멀티 에이전트 — 오케스트레이터 + 팀원 에이전트 구성 예시
import anthropic
import concurrent.futures
 
client = anthropic.Anthropic()
 
# 오케스트레이터: 전체 작업을 분석하고 서브태스크로 분해
orchestrator_prompt = """
당신은 소프트웨어 엔지니어링 팀의 오케스트레이터입니다.
주어진 코드베이스 변경 요청을 분석하고, 다음 세 에이전트에게 작업을 분배합니다:
- test_agent: 테스트 케이스 작성 담당
- refactor_agent: 코드 리팩토링 담당
- doc_agent: 문서화 담당
 
각 에이전트는 독립된 컨텍스트를 가지며, blockers 리스트를 통해 이슈를 공유합니다.
"""
 
def spawn_agent(role: str, task: str, shared_context: dict) -> str:
    agent_prompts = {
        "test_agent": "당신은 테스트 전문가입니다. 주어진 코드의 엣지 케이스를 중심으로 테스트를 작성해 주세요.",
        "refactor_agent": "당신은 리팩토링 전문가입니다. 가독성과 성능을 함께 고려해 코드를 개선해 주세요.",
        "doc_agent": "당신은 기술 문서 전문가입니다. 개발자가 바로 이해할 수 있는 문서를 작성해 주세요.",
    }
 
    response = client.messages.create(
        model="claude-opus-4-7",
        max_tokens=4096,
        system=agent_prompts[role],
        messages=[
            {
                "role": "user",
                "content": f"작업: {task}\n\n공유 컨텍스트:\n{shared_context}"
            }
        ]
    )
    return response.content[0].text
 
# 공유 컨텍스트 (블랙보드 역할)
shared_context = {
    "codebase_summary": "Python FastAPI 기반 REST API 서버",
    "changed_files": ["src/user_service.py", "src/auth.py"],
    "blockers": []  # 에이전트들이 공유하는 블로커 리스트
}
 
tasks = {
    "test_agent": "user_service.py의 유닛 테스트 작성",
    "refactor_agent": "auth.py의 토큰 검증 로직 리팩토링",
    "doc_agent": "변경된 API 엔드포인트 문서화"
}
 
# 세 에이전트를 병렬로 실행
results = {}
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    futures = {
        executor.submit(spawn_agent, role, task, shared_context): role
        for role, task in tasks.items()
    }
    for future in concurrent.futures.as_completed(futures):
        role = futures[future]
        results[role] = future.result()

# LangGraph 기반 공유 벡터 메모리 멀티 에이전트 예시
# 필요 패키지: pip install langgraph qdrant-client sentence-transformers
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict, Annotated
import operator
 
# 공유 상태 스키마 정의 — 모든 에이전트가 읽고 쓰는 블랙보드
class SharedState(TypedDict):
    query: str
    facts: Annotated[list, operator.add]           # 리듀서: 여러 에이전트가 추가 가능
    analysis_results: Annotated[list, operator.add]
    final_answer: str
    active_agent: str
 
# ── 아래 함수들은 실제 구현이 필요한 인터페이스입니다 ──
# search_shared_vector_db(query): Qdrant 컬렉션에서 유사 팩트 조회
# store_to_shared_vector_db(facts): 새 팩트를 Qdrant에 저장
# collect_facts(query): LLM 또는 외부 API로 신규 팩트 수집
# load_private_memory(agent_id): 에이전트별 전용 벡터 저장소에서 도메인 지식 로드
# run_analysis(facts, knowledge): 수집된 팩트 + 도메인 지식으로 분석 실행
# synthesize(results): 복수 분석 결과를 최종 답변으로 합성
# ─────────────────────────────────────────────────────────
 
# 리서치 에이전트 — 사실 수집 전담
def research_agent(state: SharedState) -> dict:
    # 공유 벡터 DB에서 기존 팩트 조회 (의미론적 잠금 패턴)
    existing_facts = search_shared_vector_db(state["query"])
 
    if existing_facts:
        # 이미 알려진 사실이 있으면 중복 분석 건너뜀
        return {"facts": existing_facts, "active_agent": "analysis_agent"}
 
    # 새 팩트 수집 후 공유 벡터 DB에 저장
    new_facts = collect_facts(state["query"])
    store_to_shared_vector_db(new_facts)
    return {"facts": new_facts, "active_agent": "analysis_agent"}
 
# 분석 에이전트 — 사실 기반 분석 전담 (독립 장기 메모리 활용)
def analysis_agent(state: SharedState) -> dict:
    # 에이전트 자신의 Private Memory에서 도메인 지식 로드
    domain_knowledge = load_private_memory("analysis_agent")
    analysis = run_analysis(state["facts"], domain_knowledge)
    return {"analysis_results": [analysis], "active_agent": "synthesis_agent"}
 
# 종합 에이전트 — 최종 답변 합성
def synthesis_agent(state: SharedState) -> dict:
    final = synthesize(state["analysis_results"])
    return {"final_answer": final, "active_agent": "end"}
 
# 라우팅 로직 — 오케스트레이터가 블랙보드 상태를 보고 다음 에이전트 결정
def route_next(state: SharedState) -> str:
    routing_map = {
        "analysis_agent": "analysis_agent",
        "synthesis_agent": "synthesis_agent",
        "end": END
    }
    return routing_map.get(state["active_agent"], END)
 
# 그래프 구성
from langgraph.graph import END  # END는 langgraph.graph에서 임포트
 
workflow = StateGraph(SharedState)
workflow.add_node("research_agent", research_agent)
workflow.add_node("analysis_agent", analysis_agent)
workflow.add_node("synthesis_agent", synthesis_agent)
 
workflow.set_entry_point("research_agent")
workflow.add_conditional_edges("research_agent", route_next)
workflow.add_conditional_edges("analysis_agent", route_next)
workflow.add_conditional_edges("synthesis_agent", route_next)
 
# 체크포인트 — 중단 후 재개 가능, 디버깅에도 활용
checkpointer = MemorySaver()
app = workflow.compile(checkpointer=checkpointer)

// A2A 프로토콜 — Agent Card (능력 광고) 예시
// 실행 환경: Node.js 18+ (crypto.randomUUID, fetch 내장)
 
const inventoryAgentCard = {
  name: "inventory-optimizer-agent",
  version: "1.0.0",
  description: "실시간 재고 최적화 및 발주 예측 전담 에이전트",
  capabilities: {
    streaming: true,           // SSE 기반 스트리밍 지원
    pushNotifications: true,   // 비동기 알림 지원
    stateTransitionHistory: true
  },
  skills: [
    {
      id: "optimize-stock-level",
      name: "재고 수준 최적화",
      description: "현재 재고 데이터를 바탕으로 최적 발주량 계산",
      inputModes: ["application/json"],
      outputModes: ["application/json", "text/plain"]
    },
    {
      id: "predict-demand",
      name: "수요 예측",
      description: "과거 판매 데이터 기반 수요 예측",
      inputModes: ["application/json"],
      outputModes: ["application/json"]
    }
  ]
}
 
interface ProductData {
  productId: string
  currentStock: number
  salesHistory: number[]
}
 
interface OptimizationResult {
  recommendedOrderQuantity: number
  predictedDemand: number
}
 
// A2A 클라이언트 — 다른 에이전트에게 작업 위임
async function delegateToInventoryAgent(
  productData: ProductData,
  agentEndpoint: string
): Promise<OptimizationResult> {
  const taskRequest = {
    id: crypto.randomUUID(),
    message: {
      role: "user",
      parts: [{
        type: "data",
        data: productData
      }]
    }
  }
 
  // A2A는 내부 메모리·도구를 노출하지 않고 표준 인터페이스로만 협력
  const response = await fetch(`${agentEndpoint}/tasks/send`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify(taskRequest)
  })
 
  return await response.json() as OptimizationResult
}