본문으로 건너뛰기

AgentCore 하이브리드 전략

2026-04-18 작성2026-07-17 수정15분 읽기

개요​

Bedrock AgentCore는 강력한 매니지드 Agent 플랫폼이지만, 엔터프라이즈 환경에서는 자체 호스팅 인프라와의 조합이 필요한 경우가 많습니다. 이 문서는 AgentCore의 서버리스 장점과 EKS 기반 Self-hosted 인프라의 유연성을 결합하여 최적의 하이브리드 아키텍처를 설계하기 위한 의사결정 프레임워크와 패턴 카탈로그를 제공합니다.

선행 문서

이 문서를 읽기 전에 다음 문서를 먼저 참조하세요:

하이브리드 배치 동기​

단일 접근의 한계​

AgentCore만 사용할 때의 제약:

  • 완전 매니지드 Bedrock 서빙(Custom Model Import)을 원할 경우 추론 엔진(vLLM) 레벨 제어가 불가 (self-hosted vLLM 엔드포인트는 OpenAI 호환 base_url로 호출 가능)
  • 토큰 기반 과금 (고빈도 단순 작업에서 비용 증가)
  • 온프레미스 데이터 소스와의 latency
  • VPC 내부 도구 접근 시 VPC 연결 또는 PrivateLink 구성 필요

EKS Self-hosted만 사용할 때의 제약:

  • Agent Runtime 인프라 운영 부담 (Kagent Pod + Redis State Store)
  • 서버리스 스케일링 대비 복잡한 오토스케일링 (KEDA Queue 기반)
  • 매니지드 메모리 관리 부재 (직접 구현)
  • 멀티 에이전트 오케스트레이션 프레임워크 직접 구축

하이브리드의 핵심 가치​

비용 손익분기점 계산​

월 추론 볼륨AgentCore OnlyEKS Self-hosted OnlyHybrid (Cascade)최적 접근
~10만 건$300-500$800-1,200$400-700AgentCore Only
~50만 건$1,500-2,000$1,200-1,800$800-1,200Hybrid 시작점
~150만 건$4,500-6,000$2,500-3,500$2,000-2,800Hybrid 필수
~500만 건+$15,000+$3,500-5,000$4,000-6,000EKS 중심 Hybrid
손익분기점

월 50만 건 이상 추론 볼륨에서 Hybrid 접근이 비용 효율적입니다. 코딩 도구 비용 분석에서 상세 계산식을 참조하세요.

Decision Matrix: Agent를 어디에 둘 것인가​

8개 핵심 축으로 평가하여 Agent 배치를 결정합니다.

평가 축AgentCoreEKS KagentHybrid판단 기준
추론 지연플랫폼 오버헤드 ~200-250ms (warm), cold 2-5초+VPC 내부 라우팅 수 ms~수십 ms낮음VPC 내부 도구 호출 → EKS
비용고빈도 시 높음고빈도 시 낮음최적단순=EKS, 복잡=AgentCore
PII 처리VPC 연결 구성 필요VPC 내부 (유리)유연민감 데이터 → EKS MCP
모델 커스텀Bedrock 또는 self-hosted 엔드포인트 (완전 매니지드는 Custom Model Import)자유 (vLLM 직접 서빙)자유추론 엔진 레벨 제어 → EKS
도구 체인REST→MCP 변환K8s 네이티브양쪽외부 SaaS → AgentCore Gateway
세션 길이최대 8시간제한 없음제한 없음장시간 대화 → EKS State
감사 요건CloudTrail 자동직접 구현 필요CloudTrail + Custom규제 → AgentCore 우선
팀 역량Kubernetes 불필요Kubernetes 필수선택적K8s 초보 → AgentCore 중심

의사결정 플로우차트​

데이터 중력과 툴 코로케이션 패턴​

데이터 중력(Data Gravity)이란?​

데이터가 많은 곳에 컴퓨팅을 배치하는 것이 네트워크 지연과 비용을 최소화합니다.

전형적인 시나리오:

  • EKS VPC 내부에 Milvus 벡터 DB (수 GB~TB 규모)
  • AgentCore Runtime은 기본 Public 네트워크 모드로 실행 (VPC 연결 모드 구성 가능)
  • VPC 연결 미구성 시 Milvus 조회에 PrivateLink 경유 필요 → 지연 증가 + 복잡도 증가

역방향 호출 패턴​

AgentCore Runtime이 EKS VPC 내부의 MCP 서버를 호출하는 아키텍처입니다.

Runtime VPC 연결

AgentCore Runtime·Gateway·내장 도구는 VPC 연결을 지원합니다. Runtime을 서브넷·보안 그룹에 연결하면 PrivateLink 없이 VPC 내부 리소스(EKS 호스팅 MCP 서버 등)에 직접 접근할 수 있습니다. 아래 PrivateLink 패턴은 교차 계정 연동 또는 VPC 연결을 사용하지 않는 구성에서 유효합니다. 상세 구성은 AgentCore VPC 문서를 참조하세요.

# privatelink-mcp-endpoint.yaml
apiVersion: v1
kind: Service
metadata:
  name: mcp-server-nlb
  namespace: mcp-system
  annotations:
    # AWS Load Balancer Controller가 관리하는 NLB 생성
    service.beta.kubernetes.io/aws-load-balancer-type: "external"
    service.beta.kubernetes.io/aws-load-balancer-scheme: "internal"
    service.beta.kubernetes.io/aws-load-balancer-nlb-target-type: "ip"
spec:
  type: LoadBalancer
  selector:
    app: mcp-server
  ports:
    - port: 443
      targetPort: 8080
      protocol: TCP
---
# VPC Endpoint Service 생성 (AWS Console 또는 Terraform)
# 1. NLB ARN 확인
# 2. VPC Endpoint Service 생성 (Acceptance required: No)
# 3. AgentCore IAM Role에 Endpoint 접근 권한 추가

S3+KMS 경계 설정​

민감한 데이터는 S3 + KMS 암호화를 통해 AgentCore와 EKS 간 안전하게 공유합니다.

# secure_artifact_manager.py
import boto3
import json

class SecureArtifactManager:
    def __init__(self, bucket: str, kms_key_id: str):
        self.s3 = boto3.client('s3')
        self.kms = boto3.client('kms')
        self.bucket = bucket
        self.kms_key_id = kms_key_id
    
    def store_sensitive_result(self, agent_id: str, session_id: str, data: dict) -> str:
        """민감 결과를 S3에 암호화 저장"""
        key = f"agentcore/{agent_id}/{session_id}/result.json"
        
        self.s3.put_object(
            Bucket=self.bucket,
            Key=key,
            Body=json.dumps(data),
            ServerSideEncryption='aws:kms',
            SSEKMSKeyId=self.kms_key_id,
            Metadata={'pii': 'true', 'agent-session': session_id}
        )
        return f"s3://{self.bucket}/{key}"
    
    def load_from_eks(self, s3_uri: str) -> dict:
        """EKS Pod에서 S3 객체 로드 (Pod Identity로 KMS 복호화)"""
        bucket, key = s3_uri.replace('s3://', '').split('/', 1)
        response = self.s3.get_object(Bucket=bucket, Key=key)
        return json.loads(response['Body'].read())

IAM 정책:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "arn:aws:iam::ACCOUNT:role/AgentCoreExecutionRole"
      },
      "Action": ["s3:PutObject"],
      "Resource": "arn:aws:s3:::my-secure-artifacts/agentcore/*",
      "Condition": {
        "StringEquals": {"s3:x-amz-server-side-encryption": "aws:kms"}
      }
    },
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "arn:aws:iam::ACCOUNT:role/EKSPodRole"
      },
      "Action": ["s3:GetObject"],
      "Resource": "arn:aws:s3:::my-secure-artifacts/agentcore/*"
    }
  ]
}

Hand-off 패턴 카탈로그​

패턴 (a): Router-front (AgentCore Gateway→Self-hosted)​

AgentCore Gateway가 요청을 분석하여 AgentCore Agent 또는 EKS Self-hosted Agent로 라우팅합니다.

분류 기준:

복잡도 점수라우팅 대상예시 작업
0.0-0.3EKS Qwen3-4B코드 완성, 번역, 요약
0.3-0.7AgentCore Claude Haiku기본 분석, 간단한 추론
0.7-1.0AgentCore Claude Sonnet아키텍처 리뷰, 복잡한 추론

구현:

# classifier_router.py
from strands import Agent
from strands.models import BedrockModel
import boto3
import json

agentcore = boto3.client('bedrock-agentcore')

class HybridRouter:
    def __init__(self):
        self.classifier = Agent(
            model=BedrockModel(model_id="anthropic.claude-haiku-4-5-20251001-v1:0"),
            system_prompt="""당신은 요청 복잡도 분류기입니다.
복잡도를 0.0-1.0 사이로 평가하여 JSON 응답하세요.
{"complexity": 0.0-1.0, "reason": "이유"}"""
        )
    
    def route(self, user_request: str) -> dict:
        classification = self.classifier(f"요청: {user_request}")
        complexity = classification['complexity']
        
        if complexity < 0.3:
            return self._route_to_eks(user_request)
        elif complexity < 0.7:
            return self._route_to_agentcore(user_request, model='haiku')
        else:
            return self._route_to_agentcore(user_request, model='sonnet')
    
    def _route_to_eks(self, request: str) -> dict:
        """EKS Kagent로 라우팅"""
        import requests
        response = requests.post(
            "http://kagent-service.agents.svc.cluster.local/invoke",
            json={"prompt": request, "model": "qwen3-4b"}
        )
        return {"response": response.json(), "routed_to": "eks-kagent"}
    
    def _route_to_agentcore(self, request: str, model: str) -> dict:
        """AgentCore로 라우팅"""
        response = agentcore.invoke_agent_runtime(
            agentRuntimeArn='arn:aws:bedrock-agentcore:us-east-1:ACCOUNT:runtime/complex-task-agent',
            runtimeSessionId='session-' + str(hash(request)),
            payload=json.dumps({"prompt": request})
        )
        return {"response": response, "routed_to": f"agentcore-{model}"}

패턴 (b): Escalation (Qwen3 Self→AgentCore Reasoning)​

EKS Self-hosted Agent가 먼저 처리하고, 복잡도가 임계값을 초과하면 AgentCore로 에스컬레이션합니다.

에스컬레이션 트리거:

  • LLM 응답 신뢰도 점수 < 0.7
  • 도구 호출 실패 2회 이상
  • 사용자 명시적 요청 ("더 정확한 답변 필요")

구현:

# escalation_agent.py
from strands import Agent
import boto3
import json

class EscalatingAgent:
    def __init__(self):
        self.primary_agent = Agent(
            model=LocalModel("http://vllm-qwen3.vllm.svc.cluster.local"),
            tools=["code_completion", "translation"]
        )
        self.agentcore = boto3.client('bedrock-agentcore')
    
    def process(self, user_request: str) -> dict:
        # 1차: EKS Self-hosted Agent
        response = self.primary_agent(user_request)
        confidence = response.metadata.get('confidence', 0.0)
        
        if confidence >= 0.7:
            return {"response": response, "agent": "eks-qwen3", "confidence": confidence}
        
        # 에스컬레이션: AgentCore Claude Sonnet
        print(f"⚠️ 낮은 신뢰도 ({confidence}) → AgentCore 에스컬레이션")
        agentcore_response = self.agentcore.invoke_agent_runtime(
            agentRuntimeArn='arn:aws:bedrock-agentcore:us-east-1:ACCOUNT:runtime/expert-agent',
            runtimeSessionId='escalation-session',
            payload=json.dumps({
                "prompt": f"원본 요청: {user_request}\n\n초기 시도 실패 (신뢰도: {confidence}). 정확한 답변 제공 필요."
            })
        )
        return {"response": agentcore_response, "agent": "agentcore-sonnet", "escalated": True}

패턴 (c): Dual-write Memory (AgentCore Memory↔EKS Langfuse)​

AgentCore와 EKS Agent 간 대화 기록을 동기화하여 일관된 컨텍스트를 유지합니다.

동기화 전략:

이벤트AgentCore → EKSEKS → AgentCore
세션 시작Memory Session ID → S3Langfuse Trace ID → DynamoDB
도구 호출Action Group 실행 로그 → CloudWatch → LangfuseLangfuse Span → CloudWatch Logs Insights
세션 종료Memory 요약 → S3 → LangfuseLangfuse 세션 통계 → AgentCore Analytics

구현:

# dual_memory_sync.py
import boto3
import json
from langfuse import Langfuse
from datetime import datetime

class DualMemoryManager:
    def __init__(self):
        self.s3 = boto3.client('s3')
        self.langfuse = Langfuse(
            public_key="lf_pk_...",
            secret_key="lf_sk_...",
            host="https://langfuse.eks.internal"
        )
        self.agentcore_memory_bucket = "agentcore-memory-export"
    
    def sync_agentcore_to_langfuse(self, agent_id: str, session_id: str):
        """AgentCore Memory → Langfuse 동기화"""
        # AgentCore Memory 내보내기 (S3)
        memory_key = f"{agent_id}/{session_id}/memory.json"
        memory_obj = self.s3.get_object(Bucket=self.agentcore_memory_bucket, Key=memory_key)
        memory_data = json.loads(memory_obj['Body'].read())
        
        # Langfuse Trace 생성
        trace = self.langfuse.trace(
            id=session_id,
            name=f"AgentCore Session {agent_id}",
            metadata={"source": "agentcore", "agent_id": agent_id}
        )
        
        for turn in memory_data['conversation']:
            trace.span(
                name=f"Turn {turn['turn_id']}",
                input=turn['user_input'],
                output=turn['agent_response'],
                metadata={"timestamp": turn['timestamp']}
            )
        
        trace.update(output=memory_data.get('summary'))
        print(f"✅ AgentCore Memory → Langfuse 동기화 완료: {session_id}")
    
    def sync_langfuse_to_agentcore(self, trace_id: str, agent_id: str):
        """Langfuse → AgentCore Memory 동기화"""
        trace = self.langfuse.get_trace(trace_id)
        
        # AgentCore Memory 형식으로 변환
        memory_data = {
            "agent_id": agent_id,
            "session_id": trace_id,
            "conversation": [
                {"turn_id": i, "user_input": span.input, "agent_response": span.output}
                for i, span in enumerate(trace.spans)
            ],
            "synced_at": datetime.utcnow().isoformat()
        }
        
        # S3 업로드 (AgentCore가 import)
        self.s3.put_object(
            Bucket=self.agentcore_memory_bucket,
            Key=f"{agent_id}/{trace_id}/imported-memory.json",
            Body=json.dumps(memory_data)
        )
        print(f"✅ Langfuse → AgentCore Memory 동기화 완료: {trace_id}")

패턴 (d): Cost-arbitrage (고빈도=EKS, 저빈도 복잡=AgentCore)​

요청 빈도와 복잡도에 따라 비용 최적 Agent를 선택합니다.

비용 모델:

시나리오월 요청 수평균 토큰AgentCore 비용EKS 비용최적 선택
코드 완성500만 건300 토큰~$15,000~$3,500EKS
아키텍처 리뷰5만 건5,000 토큰~$2,500$3,500 (GPU 유휴)AgentCore
번역200만 건500 토큰~$10,000~$2,000EKS
복잡한 추론10만 건8,000 토큰~$8,000$4,000 (전용 GPU)AgentCore

라우팅 로직:

# cost_arbitrage_router.py
class CostArbitrageRouter:
    def __init__(self):
        self.request_counts = {}  # 요청 빈도 추적
        
        # 비용 계수 (예시)
        self.agentcore_cost_per_1k_tokens = 0.003  # Claude Haiku
        self.eks_fixed_monthly = 500  # GPU 인스턴스 고정 비용
        self.eks_break_even_requests = 200000  # 손익분기
    
    def should_use_eks(self, task_type: str, estimated_tokens: int) -> bool:
        """비용 기반 라우팅 결정"""
        monthly_requests = self.request_counts.get(task_type, 0)
        
        # 고빈도 작업 → EKS
        if monthly_requests > self.eks_break_even_requests:
            return True
        
        # 저빈도 + 복잡 → AgentCore
        if estimated_tokens > 5000 and monthly_requests < 50000:
            return False
        
        # 단순 작업 → EKS (고정 비용 상각)
        if estimated_tokens < 1000:
            return True
        
        return False  # 기본: AgentCore

IAM·세션·관측성 통합 경계​

AgentCore Identity OAuth 토큰 전파​

AgentCore Identity가 발급한 OAuth 토큰을 EKS MCP 서버까지 안전하게 전달합니다.

EKS MCP Server 인증 검증:

# mcp_auth_middleware.py
import jwt
from functools import wraps
from flask import request, jsonify

def validate_agentcore_token(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        token = request.headers.get('X-Forwarded-Authorization', '').replace('Bearer ', '')
        
        if not token:
            return jsonify({"error": "Missing authorization token"}), 401
        
        try:
            # IdP(Okta) 공개키로 검증 — AgentCore Identity가 전파한 토큰의 원 발급자
            payload = jwt.decode(
                token,
                audience="mcp-server",
                issuer="https://YOUR_OKTA_DOMAIN/oauth2/default",
                algorithms=["RS256"],
                options={"verify_signature": True}
            )
            request.user_id = payload['sub']
            request.scopes = payload['scope']
            return f(*args, **kwargs)
        except jwt.ExpiredSignatureError:
            return jsonify({"error": "Token expired"}), 401
        except jwt.InvalidTokenError:
            return jsonify({"error": "Invalid token"}), 401
    
    return decorated

@app.route('/mcp/customer-lookup', methods=['POST'])
@validate_agentcore_token
def customer_lookup():
    """인증된 사용자만 고객 조회 가능"""
    customer_id = request.json.get('customer_id')
    # request.user_id로 감사 로그 기록
    return {"customer": fetch_customer(customer_id)}

CloudWatch GenAI Observability ↔ Langfuse OTel 브리지​

AgentCore 트레이스와 EKS Langfuse 트레이스를 통합하여 전체 Agent 플로우를 추적합니다.

Trace Correlation ID 규칙:

소스Trace ID 형식Parent Span ID
AgentCoreac-{session_id}-{timestamp}ac-root
EKS Kagenteks-{pod_name}-{trace_id}ac-{session_id} (AgentCore 호출 시)
Hybrid Tracehybrid-{session_id}양쪽에서 공유

Lambda Trace Forwarder:

# trace_forwarder_lambda.py
import boto3
import json
import os
import requests
from datetime import datetime

cloudwatch = boto3.client('logs')
langfuse_endpoint = "https://langfuse.eks.internal/api/public/ingestion"

def lambda_handler(event, context):
    """CloudWatch GenAI Observability → Langfuse 전달"""
    for record in event['Records']:
        message = json.loads(record['Sns']['Message'])
        
        if message['source'] == 'aws.bedrock.agentcore':
            trace_data = message['detail']
            
            # Langfuse 형식으로 변환
            langfuse_trace = {
                "id": f"hybrid-{trace_data['sessionId']}",
                "name": f"AgentCore {trace_data['agentId']}",
                "metadata": {
                    "source": "agentcore",
                    "agent_id": trace_data['agentId'],
                    "aws_region": message['region']
                },
                "spans": [
                    {
                        "name": step['actionGroupName'],
                        "input": step['input'],
                        "output": step['output'],
                        "start_time": step['startTime'],
                        "end_time": step['endTime']
                    }
                    for step in trace_data.get('actionGroupInvocations', [])
                ]
            }
            
            # Langfuse로 전송
            response = requests.post(
                langfuse_endpoint,
                json=langfuse_trace,
                headers={"Authorization": f"Bearer {os.environ['LANGFUSE_API_KEY']}"}
            )
            print(f"✅ Trace 전달 완료: {trace_data['sessionId']} → Langfuse")
    
    return {"statusCode": 200}

점진적 마이그레이션 로드맵​

Phase 1: AgentCore Only (0-3개월)​

목표: 빠른 프로덕션 배포, 인프라 운영 부담 제로

체크리스트:

  • Bedrock 모델 선택 (Claude Sonnet/Haiku)
  • Strands SDK로 Agent 구현
  • AgentCore에 배포 (agentcore deploy)
  • Knowledge Bases RAG 구성
  • CloudWatch GenAI Observability 활성화

Exit Criteria (Phase 2 전환 트리거):

  • 월 추론 볼륨 50만 건 초과
  • Bedrock 토큰 비용 월 $1,500 초과
  • VPC 내부 도구 호출 빈도 높음 (p95 latency > 100ms)

Phase 2: Bedrock + Self-hosted SLM (3-6개월)​

목표: 비용 최적화, 단순 작업을 EKS Qwen3-4B로 오프로드

체크리스트:

  • EKS 클러스터 구성 (Auto Mode 또는 Karpenter)
  • vLLM으로 Qwen3-4B 배포
  • LLM Classifier 구현 (Cascade Routing)
  • kgateway + Bifrost 2-Tier Gateway 구성
  • 비용 대시보드 구축 (AgentCore vs EKS 비용 추적)

Exit Criteria (Phase 3 전환 트리거):

  • EKS Agent와 AgentCore Agent 간 컨텍스트 공유 필요
  • 양쪽에서 동일한 세션 유지 요구
  • Fine-tuned 커스텀 모델 필요

Phase 3: Full Hybrid Cross-routing (6-12개월)​

목표: 양방향 라우팅, 통합 컨텍스트, 최적 비용

체크리스트:

  • Dual-write Memory 동기화 구현 (패턴 c)
  • Trace Correlation ID 통합
  • PrivateLink Endpoint for MCP
  • Cost-arbitrage Router 구현 (패턴 d)
  • Escalation 로직 구현 (패턴 b)
  • 통합 대시보드 (AgentCore + EKS 통합 관측성)

Success Metrics:

  • 비용 절감률: 40-60% (Bedrock Only 대비)
  • p95 지연: AgentCore 단독 대비 20% 개선
  • 세션 컨텍스트 일관성: 95% 이상
  • Agent 가용성: 99.9% (양쪽 Failover)

전환 트리거 지표​

각 Phase 전환을 결정하는 정량적 지표입니다.

지표Phase 1 → 2 임계값Phase 2 → 3 임계값
월 추론 볼륨> 50만 건> 150만 건
월 비용> $1,500> $3,000
평균 지연 (p95)> 100ms> 200ms
세션 컨텍스트 손실률N/A> 5%
커스텀 모델 요구Fine-tuning 필요도메인 특화 SLM 필요
팀 K8s 역량초보중급 이상

참고 자료​

공식 문서​

논문 / 기술 블로그​

관련 문서 (내부)​