안녕하세요.

최근 AI 에이전트 구축을 이것저것 시도하는 가운데, "어떻게 하면 인간이 수행하는 태스크를 정확무결하게 수행하는 에이전트를 구축할 수 있을까"를 고민하고 있는 요즘입니다.

AI 에이전트 구축에 자주 사용하는 SDK는 Python의 Strands Agents인데, 레퍼런스를 읽다 보면 모르는 도구가 꽤 많이 등장하곤 합니다. ("new"라는 글자가 꽤 빠른 속도로 늘어나고 있는 것 같네요...)

그러던 중, 오랫동안 궁금했지만 검증하지 못했던 workflow 도구를 검증해 보려고 합니다.

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

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

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

이때 상정하고 있는 동작은, 다음과 같은 관점으로 나누어 리뷰하고 마지막에 채택 판단이 포함된 Markdown 리포트를 반환하는 것을 목표로 합니다.

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

또한, 이번에는 "멀티 에이전트 구성 (Multi-agent configuration)"이 전제되므로, 멀티 에이전트 구성으로 에이전트를 배포해 보고 싶은 분들은 참고해 보시기 바랍니다.

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

• Strands Agents의 멀티 에이전트 구성을 시도해 보고 싶은 분
• strands_tools.workflow가 무엇을 해주는지 알고 싶은 분
• Amazon Bedrock AgentCore Runtime 상에서 Strands Agents를 구동하고 싶은 분
• AWS CDK (Python)로 AgentCore Runtime에 에이전트를 배포하고 싶은 분

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

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

• Strands Agents의 workflow 도구로 여러 태스크의 의존 관계를 정의한다.
• 부모 에이전트로부터 태스크별 자식 에이전트를 호출하여, 기술 선정 리뷰를 여러 전문 태스크로 분할하여 실행한다.
• 최종적으로 리포트를 응답(Response)으로 반환한다.

구성(Configuration)은 다음과 같습니다.

위의 플로우(Flow) 중에서 각 태스크("요구사항 정리" 이후)는 자식 에이전트에 의해 실행되도록 되어 있습니다.

mkdir strands-agents-agent-workflows
cd strands-agents-agent-workflows

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
strands-agents-tools>=0.2.0,<1.0.0
...

이번 주인공은 strands-agents-tools에 포함된 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를 고정해 두면, 가상 환경(virtual environment)을 활성화(activate)하지 않은 상태에서도 cdk synth나 cdk deploy가 동일한 Python 환경을 사용하게 됩니다.

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

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

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

필요하다면 부트스트랩(bootstrap)합니다.

cdk bootstrap

먼저, 이번 기술 선정 회의를 태스크(task)로 정의합니다.

구현은 agent/tech_selection_agent/workflow_spec.py에 두었습니다.

agent/
└── tech_selection_agent/
├── agent.py # 에이전트(agent)를 구현
...

이번 워크플로(workflow)는 다음과 같은 형태입니다.

처음에 requirements_reader가 평가 대상인 기술 선정안을 정리합니다.

그 결과를 바탕으로, 6개의 리뷰 태스크(서브 에이전트)가 병렬로 동작합니다.

frontend_reviewer
backend_reviewer
infra_reviewer
security_reviewer
cost_reviewer
operability_reviewer

마지막으로, decision_maker가 리뷰 결과를 통합하고, final_report가 Markdown 리포트를 작성합니다.

workflow 도구에 전달할 태스크는 Python의 딕셔너리(dictionary)로 정의합니다.

def build_tasks(prompt: str) -> list[dict[str, Any]]:
proposal = prompt.strip()
base_context = (
...

dependencies에 의존 태스크를 지정하면, 해당 태스크가 완료될 때까지 실행되지 않습니다.

이 예시에서는 frontend_reviewer가 requirements_reader가 끝난 후에 실행됩니다.

이번에 조금 주의를 기울인 부분은 tools 지정입니다.

strands_tools.workflow는 태스크에 tools를 지정하지 않을 경우, 부모 에이전트(parent Agent)가 가지고 있는 도구를 자식 에이전트(child Agent)에게 상속합니다.

이번 부모 에이전트는 workflow 도구를 가지고 있습니다. 즉, 아무것도 하지 않으면 자식 에이전트도 workflow를 갖게 됩니다.

이것이 반드시 나쁜 것은 아니지만, 자식 에이전트가 새롭게 워크플로 도구를 실행해 버리면 불필요한 루프(loop) 처리가 발생할 수 있기 때문에, 이를 피하고자 다음과 같이 존재하지 않는 도구 이름을 지정하여 자식 에이전트에게 도구를 전달하지 않도록 했습니다.

# 처음에 `agent/tech_selection_agent/workflow_spec.py`에 상수로 정의해 둠
NO_TASK_TOOLS = ["__no_task_tools__"]
# 태스크(서브 에이전트)의 인자로 `NO_TASK_TOOLS`를 전달해 둠
...

이렇게 하면 각 태스크는 도구가 없는 전문 리뷰 에이전트(Agent)로서 동작합니다.

다음으로, workflow 도구를 가진 부모 에이전트를 생성합니다.

구현은 agent/tech_selection_agent/agent.py입니다.

strands_tools.workflow는 워크플로의 상태를 JSON 파일로 저장합니다.

기본값은 ~/.strands/workflows이지만, AgentCore Runtime 컨테이너 상에서는 /tmp 하위로 맞춰 두었습니다.

이것은 취향입니다!

# JSON 파일의 저장 위치를 지정
DEFAULT_WORKFLOW_DIR = "/tmp/strands-workflows"
os.environ.setdefault("STRANDS_WORKFLOW_DIR", DEFAULT_WORKFLOW_DIR)
...

workflow

는 import 시에 STRANDS_WORKFLOW_DIR을 읽기 때문에, from strands_tools import workflow 보다 먼저 환경 변수를 설정하고 있습니다.

def create_agent() -> Agent:
region = get_region()
model = BedrockModel(
...

이 부모 Agent (Parent Agent)는 기술 선정 리뷰 본문을 직접 작성하는 역할이 아닙니다.

역할은 workflow 도구를 호출하는 것입니다.

각 리뷰 담당 Agent는 workflow 도구 내부에서 태스크(Task)별로 생성됩니다.

workflow 도구에는 action이라는 인자(Argument)가 존재합니다.

먼저 create로 태스크 정의를 저장하고, 다음으로 start로 실행합니다.

def execute_tech_selection_workflow(prompt: str, workflow_id: str) -> dict[str, Any]:
# 3-2에서 작성한 부모 에이전트를 인자로 저장한다
agent = create_agent()
...

create는 상태 파일(State file)을 만들 뿐이며, 아직 LLM 실행은 시작되지 않습니다.

실제로 각 태스크가 동작하는 것은 start입니다. start는 동기(Synchronous) 실행이므로, final_report가 끝날 때까지 반환되지 않습니다.

workflow start의 반환값은 "workflow가 완료되었습니다"라는 메시지가 중심입니다.

이번에 필요한 것은 마지막 final_report 태스크가 생성한 Markdown 본문입니다. 따라서 workflow가 저장한 JSON을 읽습니다.

workflow_data = load_workflow_data(workflow_id, get_workflow_dir())
final_report = extract_task_text(workflow_data, "final_report")
if not final_report:
...

상태 파일은 다음 위치에 저장됩니다.

/tmp/strands-workflows/<workflow_id>.json

이 안의 task_results.final_report.result에서 text 블록(Block)만 추출하여 최종 응답으로 사용합니다.

부모 에이전트가 최종 리포트를 읽고 사용자에게 응답을 반환하는 방식은 아닙니다. 최종 리포트의 내용을 그대로 출력하도록 하고 있습니다.

AgentCore Runtime으로부터 호출되는 엔트리포인트 (Entrypoint)는 agent/tech_selection_agent/runtime_app.py에 둡니다.

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

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

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

workflow_id는 상태 파일명이기도 하므로, 요청(Request)마다 uuid를 붙여 충돌하지 않도록 하고 있습니다.

AgentCore Runtime의 엔트리포인트와 헬스 체크 (Health check)는 다음과 같습니다.

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

응답에는 최종 리포트 본문과, 어떤 workflow가 동작했는지 확인하기 위한 증거(Evidence)를 포함합니다.

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

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

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

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

여기에서도 STRANDS_WORKFLOW_DIR=/tmp/strands-workflows를 지정하고 있습니다.

agent.py 측에서도 기본값(Default value)을 설정하고 있지만, Runtime 컨테이너로서는 Dockerfile과 CDK의 환경 변수(Environment variable)로도 명시해 두는 것이 동작을 추적하기 쉬울 것이라고 생각합니다. (차이가 발생하지 않도록 주의가 필요하겠지만 말입니다.)

CDK 스택(Stack)은 infra/agentcore_tech_selection_stack.py에 구현했습니다.

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

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

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

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",
...

여기서 STRANDS_WORKFLOW_DIR를 Runtime 환경 변수로 전달하고 있습니다.

이 검증에서는 workflow의 상태를 /tmp에 저장합니다. 장기적으로 저장되는 것이 아니므로, 장기 보관이나 재실행 관리가 필요한 경우에는 최소한 DynamoDB가 필요할 것이라고 생각합니다.

(DynamoDB에서 workflow_id를 키(Key)로 하여, 태스크 정의(Task definition)를 저장해 두는 모습을 상상하고 있습니다.)

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

tests/
├── test_agentcore_tech_selection_stack.py
├── test_runtime_app.py
...

test_workflow_spec.py에서는 태스크의 의존 관계(Dependency)가 예상대로인지 확인하고 있습니다.

def test_build_tasks_creates_expected_dependency_shape() -> None:
    tasks = build_tasks("Next.js + FastAPI + Aurora PostgreSQL 를 평가해줘")
    task_by_id = {task["task_id"]: task for task in tasks}
    ...

test_runtime_app.py에서는 AgentCore Runtime의 엔트리포인트(entrypoint)가 기대한 응답 형식(response format)을 반환하는지 확인하고 있습니다.

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

결과는 다음과 같습니다.

8 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 콘솔로 이동합니다.

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

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

이번에는 샘플로 다음과 같이 입력해 보겠습니다.

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

입력했다면, 「실행(Run)」을 클릭합시다.

여러 서브 에이전트(sub-agent)에 의한 태스크를 실행하고 있으므로, 응답(response)까지는 시간이 다소 걸립니다.

서두르지 마세요... 서두르지 마세요...

처리가 완료되면, 출력의 response에 기술 선정 보고서가 들어있는 것을 확인할 수 있습니다. evidence.workflow에는 workflow 실행 정보가 들어갑니다.