안녕하세요.

이전에 「Strands Agents의 Workflow 툴을 시도해 보았다」라는 제목의 기사를 작성했습니다.

여기에서는, 태스크를 분해한 워크플로우 (Workflow)에 따라 서브 에이전트 (Sub-agent)가 순차적으로 실행하는 검증을 수행했습니다.

이번에는 Strands Agents의 GraphBuilder를 사용하여, 여러 전문 에이전트에게 기술 선정 리뷰를 분담시키는 검증을 해보았습니다.

지난번에는 strands_tools.workflow를 사용했지만, 이번에는 Strands SDK 본체에 있는 Graph를 사용합니다.

솔직히 문서를 읽을 때, Workflow와 Graph의 차이점이 명확히 와닿지 않았기에 저에게는 좋은 검증이 되었습니다.

주제는 마찬가지로 「기술 선정 & 조사 에이전트」입니다.

무언가 새로운 서비스를 개발할 때 조사를 의뢰하거나, 자신의 머릿속에 떠오르는 구성을 평가받기 위한 「기술 선정 & 조사 에이전트」입니다.

예를 들어, 다음과 같은 입력을 전달합니다.

{
"prompt": "신규 B2B SaaS를 Next.js + FastAPI + Aurora PostgreSQL + ECS Fargate + Bedrock로 만드는 안을 평가해줘"
}

이때 상정하고 있는 동작으로는, 다음과 같은 관점으로 나누어 리뷰하고 마지막에 채택 판단이 포함된 Markdown 리포트를 반환하는 것을 상정하고 있습니다. (지난번 기사와 마찬가지입니다.)

• 요구사항 정리
• 프론트엔드 (Frontend)
• 백엔드 (Backend)
• 인프라 (Infrastructure)
• 보안 (Security)
• 비용 (Cost)
• 운영 (Operation)

또한, 이번에도 「멀티 에이전트 (Multi-agent) 구성」이 전제 조건이 됩니다.

이 기사는 다음과 같은 분들을 위해 작성되었습니다.

• Strands Agents의 Graph를 시도해 보고 싶은 분
• Strands Agents로 결정적인 멀티 에이전트 구성을 만들고 싶은 분
• workflow와 graph의 차이를 알고 싶은 분
• Amazon Bedrock AgentCore Runtime 위에서 Strands Agents를 구동하고 싶은 분
• AWS CDK (Python)로 AgentCore Runtime에 에이전트를 배포하고 싶은 분

실제로 구동하고 싶은 분은 아래를 참조해 주세요.

이번 목표는 다음 3가지입니다.

• Strands Agents의 GraphBuilder로 복수 Agent의 실행 그래프를 정의한다
• 기술 선정 리뷰를 복수의 전문 Agent 노드로 분할하여 실행한다
• AgentCore Runtime에서 Graph를 기동하고, 최종 리포트를 응답(Response)으로 반환한다

구성은 다음과 같습니다.

솔직히 이것만 봐서는 워크플로우와 다를 바 없습니다. 여기서는 구성을 이해해 주시면 될 것 같습니다.

Graph는 Strands Agents의 멀티 에이전트 패턴 중 하나입니다.

GraphBuilder라는 툴을 사용하여, Agent 또는 다른 MultiAgent를 노드로 추가하고, 에지 (Edge)로 실행 순서나 의존 관계를 정의합니다.

구현 이미지는 이런 느낌입니다.

from strands.multiagent import GraphBuilder
builder = GraphBuilder()
builder.add_node(requirements_agent, node_id="requirements_reader")
...

Graph는 전 단계 노드의 출력을 다음 단계 노드의 입력에 자동으로 포함해 주기 때문에, 워크플로우 때와 같은 dependencies 지정이 필요 없다는 점은 개발자에게 고마운 부분이라고 생각합니다.

여기서 이해해 두어야 할 것은 Workflow와 Graph의 비교입니다.

둘 다 「복수의 전문 Agent를 조합하여 하나의 결과물을 만드는」 메커니즘이지만, 설계의 주어가 조금 다릅니다.

저는, 「태스크끼리 연결하는 것에 중점을 두고 있는 것」이 Workflow이고, 「에이전트끼리 연결하는 것에 중점을 두고 있는 것」이 Graph라고 이해했습니다.

mkdir strands-agents-graph
cd strands-agents-graph

CDK 측의 의존 관계는 다음과 같습니다.

aws-cdk-lib==2.255.0
constructs>=10.0.0,<11.0.0
cdk-ecr-deployment>=4.0.0,<5.0.0

개발용으로는 pytest도 넣습니다.

-r requirements.txt
pytest>=8.0.0,<9.0.0

cdk-ecr-deployment는 CDK가 빌드한 Docker image asset을 Runtime용으로 생성한 전용 ECR Repository로 복사하기 위해 사용합니다.

Runtime 컨테이너 측의 의존 관계는 다음과 같습니다.

bedrock-agentcore>=1.3.0,<2.0.0
strands-agents>=1.0.0,<2.0.0
boto3>=1.42.0

Graph는 strands-agents 본체에 포함되어 있기 때문에, workflow 버전처럼 strands-agents-tools는 필요하지 않습니다.

app.py는 일반적인 CDK 엔트리 포인트 (Entry Point)입니다.

#!/usr/bin/env python3
import os
import aws_cdk as cdk
...

cdk.json은 CDK CLI가 항상 프로젝트 내의 가상 환경을 사용하도록 설정합니다.

{
"app": ".venv/bin/python app.py",
"context": {
...

.venv/bin/python을 고정해 두면, 가상 환경을 활성화(activate)하지 않은 상태에서도 cdk synth나 cdk deploy가 동일한 Python 환경을 사용합니다.

python3.12 -m venv .venv
source .venv/bin/activate
pip install -r requirements-dev.txt -r agent/requirements.txt

AWS 인증과 리전(Region)을 확인합니다.

aws sts get-caller-identity
export AWS_REGION=ap-northeast-1

필요하다면 bootstrap 합니다.

cdk bootstrap

이번 에이전트 구현은 agent/tech_selection_graph_agent/에 둡니다.

agent/
└── tech_selection_graph_agent/
├── agent.py
...

노드 ID는 graph_spec.py에 정의합니다.

앞서 언급했듯이, 에이전트 구성은 '요건 정리 에이전트 (requirements_reader_agent)'가 먼저 동작하고, 그 후에 병렬 리뷰 에이전트들이 동작하는 형태입니다. 병렬 리뷰 에이전트들이 완료되면 채택 에이전트가 동작하여 최종 리포트가 생성되는 흐름입니다.

다음과 같이 상수 GRAPH_NODE_IDS에 에이전트 플로우(agent flow)에 따라 노드 ID를 정의해 둡시다. 병렬로 동작하는 에이전트는 가변 길이 변수(variable length variable)를 사용하도록 합니다. (스플랫 연산자(splat operator)를 상수(변수) 이름 앞에 두면 가변 길이 상수(변수)가 됩니다.)

Python에서는 가변 길이 상수를 뭐라고 안 하나...? 스플랫 연산자라고 안 하나...? Ruby에만 있는 건가...?

GRAPH_NODE_IDS = [
"requirements_reader",
*REVIEW_NODE_IDS,
...

다음으로, 병렬로 동작하는 리뷰 에이전트들의 노드 ID를 정의합니다. 위의 가변 길이 변수에 노드 ID를 저장하면 됩니다.

REVIEW_NODE_IDS = [
"frontend_reviewer",
"backend_reviewer",
...

저는 가변 길이 상수(변수)에 병렬 에이전트를 저장하여 나누어 정의했지만, GRAPH_NODE_IDS에 한꺼번에 정의할 수도 있습니다.

또한, 병렬로 동작하는 에이전트들 중에서 다시 병렬로 동작하는 에이전트를 정의할 수도 있습니다.

↓대략 이런 느낌으로 정의할 수 있습니다.↓

SUB_SUB_PARA_AGENTS = [
"sub_sub_sub_1",
"sub_sub_sub_2"
...

각 노드는 독립된 Agent이므로, 각각 별도의 시스템 프롬프트 (system prompt)를 갖게 합니다.

def build_system_prompt(role: str, instruction: str) -> str:
    return (
        f"당신은 {role}입니다.\n"
        ...
    )

예를 들어, 프론트엔드 리뷰용 Agent는 다음과 같이 정의하고 있습니다.

NODE_SYSTEM_PROMPTS = {
"frontend_reviewer": build_system_prompt(
"B2B SaaS의 프론트엔드 아키텍트",
...

workflow 도구에서는 태스크 딕셔너리(task dictionary)의 system_prompt에 역할을 작성했습니다.

Graph 버전에서는 각 노드에 전달할 Agent의 system prompt로서 역할을 부여합니다.

Graph 본체는 agent/tech_selection_graph_agent/agent.py에 구현합니다.

각 노드에는 독립된 Agent 인스턴스(instance)를 전달합니다.

def create_node_agent(node_id: str) -> Agent:
    return Agent(
        model=create_model(),
        ...
    )

여기서 한 가지 주의할 점이 있습니다.

Graph는 동일한 Agent 객체를 여러 노드에서 재사용하는 것을 허용하지 않습니다.

따라서 다음과 같이 노드마다 Agent 인스턴스를 생성하고 있습니다.

for node_id in GRAPH_NODE_IDS:
    builder.add_node(create_node_agent(node_id), node_id=node_id)

def build_graph():
    builder = GraphBuilder()
    for node_id in GRAPH_NODE_IDS:
        ...

requirements_reader를 엔트리 포인트 (entry point)로 지정합니다.

그 후, requirements_reader에서 각 리뷰어로 에지 (edge)를 연결합니다.

builder.add_edge("requirements_reader", "frontend_reviewer")
builder.add_edge("requirements_reader", "backend_reviewer")

이를 통해 requirements_reader가 완료된 후 여러 리뷰어를 실행할 수 있게 됩니다.

이번 구현에서 가장 주의했던 부분이 바로 여기입니다.

Graph의 Python 구현에서는, 다중 입력 노드(multiple input nodes)는 "입력 에지(edge) 중 하나라도 ready 상태가 되면 실행 가능"하다고 판정됩니다.

즉, 아무 생각 없이 다음과 같이 작성하면, frontend_reviewer만 완료된 시점에 decision_maker가 움직이기 시작할 가능성이 있습니다.

builder.add_edge("frontend_reviewer", "decision_maker")
builder.add_edge("backend_reviewer", "decision_maker")
builder.add_edge("infra_reviewer", "decision_maker")

이번에는 모든 리뷰 결과를 확인한 후 의사결정을 하고 싶기 때문에, decision_maker가 움직이기 위한 에지(edge) 조건을 설정했습니다.

def all_review_nodes_completed(state: Any) -> bool:
    completed_node_ids = {node.node_id for node in state.completed_nodes}
    # 모든 리뷰 에이전트의 처리가 완료되면 True를 반환
    ...

각 리뷰어에서 decision_maker로 가는 에지에 동일한 조건을 부여합니다.

builder.add_edge(
    review_node_id,
    "decision_maker",
    ...

이렇게 하면, 6개의 리뷰어가 모두 완료된 후에 decision_maker가 실행됩니다.

final_report 또한 마찬가지로, decision_maker 완료 후에만 실행되도록 설정했습니다.

def decision_completed(state: Any) -> bool:
    completed_node_ids = {node.node_id for node in state.completed_nodes}
    return "decision_maker" in completed_node_ids

Graph는 동기적(synchronously)으로 호출할 수 있습니다.

def execute_tech_selection_graph(prompt: str) -> dict[str, Any]:
    graph = build_graph()
    result = graph(prompt)
    ...

result.results에는 각 노드의 결과가 들어갑니다.

이번에 필요한 것은 마지막 final_report이므로, 그것만 추출하여 Runtime의 response로 만듭니다.

final_report = extract_node_text(result, "final_report")

실패 시나 final_report를 가져오지 못한 경우를 대비하여, 실행된 노드의 결과를 정리하는 폴백(fallback)도 준비해 두었습니다.

if not final_report:
    final_report = extract_graph_text(result)

AgentCore Runtime에서 호출되는 엔트리포인트(entrypoint)는 agent/tech_selection_graph_agent/runtime_app.py에 둡니다.

from bedrock_agentcore.runtime import BedrockAgentCoreApp, PingStatus
from .agent import execute_tech_selection_graph
app = BedrockAgentCoreApp()

입력은 {"prompt": "..."} 형식에 맞추었습니다.

def invoke_agent(payload: dict) -> dict:
    prompt = payload.get("prompt")
    if not isinstance(prompt, str) or not prompt.strip():
        ...

응답(response)에는 최종 리포트 본문과 Graph의 실행 정보를 포함합니다.

응답 이미지는 다음과 같습니다.

{
"response": "# 기술 선정 리뷰\n\n## 결론\n조건부 채택...",
"evidence": {
...

AgentCore Runtime의 entrypoint와 health check는 다음과 같습니다.

@app.entrypoint
def invoke(payload: dict) -> dict:
return invoke_agent(payload)
...

AgentCore Runtime에 올리기 위해, 에이전트를 Docker화합니다.

FROM public.ecr.aws/docker/library/python:3.12-slim
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1
...

Graph 버전에서는 workflow 버전과 같은 STRANDS_WORKFLOW_DIR는 지정하지 않았습니다.

workflow는 상태 파일(state file)을 저장하지만, Graph는 이번 구성에서 파일 영속화(file persistence)를 사용하지 않기 때문입니다.

CDK 스택은 infra/agentcore_tech_selection_graph_stack.py에 구현했습니다.

이번에 만드는 리소스는 주로 다음과 같습니다.

• ECR Repository
• DockerImageAsset
• ECRDeployment
• AgentCore Runtime
• Runtime용 IAM Role

image_repository = ecr.Repository(
self,
"AgentImageRepository",
...

검증 용도이므로, cdk destroy로 삭제할 수 있도록 구성했습니다.

CDK의 Docker image asset은 CDK가 관리하는 asset repository에 push됩니다.

이번에는 AgentCore Runtime용 전용 ECR Repository를 만들었기 때문에, cdk-ecr-deployment로 복사합니다.

image_asset = ecr_assets.DockerImageAsset(
self,
"AgentImageAsset",
...

AgentCore Runtime의 컨테이너는 ARM64로 빌드합니다.

Runtime용 역할(Role)에는 다음 권한을 부여합니다.

• ECR pull
• ECR authorization token
• CloudWatch Logs
• Bedrock model invoke

runtime_role.add_to_policy(
iam.PolicyStatement(
actions=[
...

model_id가 ARN으로 전달된 경우에는 해당 ARN을 사용하고, 그 외에는 foundation model과 inference profile의 ARN을 허용하는 형태로 구성했습니다.

def _model_resource_arns(self, model_id: str) -> list[str]:
if model_id.startswith("arn:"):
return [model_id]
...

Runtime은 AWS::BedrockAgentCore::Runtime을 생성합니다.

runtime = bedrock_agentcore.CfnRuntime(
self,
"AgentRuntime",
...

Graph 버전에서 Runtime 환경 변수는 AWS_REGION과 BEDROCK_MODEL_ID뿐입니다.

이번에도 최소한의 유닛 테스트(unit test)를 작성했습니다.

tests/
├── test_agent.py
├── test_agentcore_tech_selection_graph_stack.py
...

test_graph_spec.py에서는 노드 구성과 조건 함수를 확인합니다.

def test_all_review_nodes_completed_requires_every_review_node() -> None:
assert all_review_nodes_completed(state_with_completed(*REVIEW_NODE_IDS))
assert not all_review_nodes_completed(state_with_completed("frontend_reviewer"))

test_runtime_app.py

이제 AgentCore Runtime의 entrypoint가 기대한 응답 형식을 반환하는지 확인하고 있습니다.

CDK 테스트에서는 Runtime, ECR Repository, IAM Policy, Output이 생성되는 것을 확인합니다.

로컬 환경에서는 다음과 같이 테스트했습니다.

env JSII_RUNTIME_PACKAGE_CACHE_ROOT=/tmp/jsii-package-cache \
JSII_SILENCE_WARNING_UNTESTED_NODE_VERSION=1 \
.venv/bin/pytest

결과는 다음과 같습니다.

11 passed

배포는 평소와 동일하게 진행합니다.

pytest
cdk synth
cdk deploy

다른 Bedrock 모델 ID를 사용하는 경우에는 CDK context를 통해 지정할 수 있습니다.

cdk deploy -c modelId=your-model-or-inference-profile-id

배포가 완료되면, AWS Management Console에 접속하여 Amazon Bedrock AgentCore Runtime 콘솔로 이동합니다.

그러면 배포된 런타임이 표시되어 있을 것이므로, 이를 클릭한 뒤 화면 우측 상단의 「테스트 (Test)」를 클릭합니다.

그다음, 「입력 (Input)」 에디터에 JSON 형식으로 입력 문장을 넣습니다.

{
"prompt": "新規B2B SaaSを、Next.js + FastAPI + Aurora PostgreSQL + ECS Fargate + Bedrockで作る案を評価して"
}

입력한 후, 「실행 (Run)」을 클릭합니다.

여러 개의 서브 에이전트 (Sub-agent)를 통한 태스크를 실행하므로, 응답까지는 어느 정도 시간이 소요됩니다.

처리가 완료되면, 출력(output)의 response에 기술 선정 보고서가 포함되어 있는 것을 확인할 수 있습니다.

evidence.workflow 에 workflow 실행 정보가 저장됩니다.