🧬 메모리를 활용한 AWS Strands 에이전트형 Planner-Generator-Evaluator 패턴: 간단한 블로그 포스트 작성기

@ AWS Builder Center에서 게시됨

이전 블로그 포스트에서, 저는 LangGraph를 사용하여 평가자 에이전트(Evaluator Agent)가 생성자 에이전트(Generator Agent)가 생성한 콘텐츠를 검토하고 개선하는 간단한 Generator–Evaluator 패턴을 구현했습니다. 이는 효과적이었지만, 워크플로우가 제한된 자율성을 가진 사전 정의된 시퀀스를 따랐습니다.

이번 반복에서는 다음에 수행할 작업을 동적으로 결정할 수 있는 **Planner Agent (플래너 에이전트)**를 도입하여 아키텍처를 한 단계 더 발전시켰습니다. **고정된 워크플로우 (fixed workflow)**를 따르는 대신, 플래너는 현재 상태를 평가하고, 어떤 에이전트를 호출해야 할지 결정하며, 전체 프로세스를 오케스트레이션(orchestrate)합니다. 이를 통해 **더 에이전트 중심적이고 자율적인 시스템 (more agentic and autonomous system)**을 구축하여, 워크플로우가 단순히 하드코딩된 실행 경로에 의존하는 대신 문맥에 따라 동작을 조정할 수 있게 합니다.

이 작은 개념 증명(Proof-of-Concept) 프로젝트에서, 저는 AWS Strands Agents를 사용하여 에이전트형 블로그 포스트 작성기를 구축했습니다. 워크플로우는 Planner–Generator–Evaluator 아키텍처를 따르며, 여기서 플래너 에이전트는 프로세스를 오케스트레이션하고, 생성자 에이전트는 콘텐츠를 생성하며, 평가자 에이전트는 반복적인 피드백을 통해 결과물을 검토하고 개선합니다.

아키텍처 (Architecture):

PlannerAgent: 다음에 무엇을 할지 결정 (동적으로 plan / replan)
       - @tool: generate_blog 호출  →  GeneratorAgent (리서치 + 작성)
       - @tool: evaluate_blog 호출  →  EvaluatorAgent (품질 게이트)
...

파이프라인과의 주요 차이점:

• **Planner의 추론 루프 (Strands 에이전트 루프를 통해)**는 런타임(runtime)에 도구(tool)를 선택합니다. 이는 재검색, 추가 소스 요약, 다시 쓰기, 평가 건너뛰기 또는 조기 종료를 수행할 수 있습니다.
• GeneratorAgent와 EvaluatorAgent는 그 자체로 @tool 호출 가능한 Strands 에이전트입니다 (에이전트를 도구로 사용하는 agents-as-tools 패턴).
• SummarizerAgent는 전용 저온도(low-temperature) 에이전트로, 전체 페이지를 가져온 뒤 작성자에게 전달하기 전 기사당 5~10개의 핵심 기술 불렛 포인트를 추출합니다.
• 고정된 엣지(edge)가 없습니다: Planner는 EvalResult 피드백과 반복 횟수를 기반으로 다음에 evaluate_blog를 호출할지 또는 generate_blog를 호출할지를 결정합니다.

Code GitHub Link: Project on GitHub

에이전트 설계를 탐구 중이든 자신만의 시스템을 구축 중이든, 이 내용은 명확하고 실용적인 시작점을 제공할 것입니다 😉

목차

• 의존성 및 설정 (Dependencies & Configuration)
• 메모리 구현 (Implementing Memory)
• 정보 가져오기 및 요약 에이전트 (Fetching Information & Summary Agent)
• 에이전트 상태 및 에이전트 도구 (Agents State & Agent Tools)
• Generator 및 Evaluator 에이전트 (Generator & Evaluator Agents)
• Planner 에이전트 및 메인 실행 (Planner Agent & Main Run)
• 전체 코드 및 데모 (All Code & Demo)
• Memory.MD 파일 (Memory.MD File)
• 생성된 블로그 포스트 (Generated Blog Post)
• 결론 (Conclusion)
• 참고 문헌 (References)

의존성 및 설정 (Dependencies & Configuration)

• 다음 의존성을 설치해 주세요:

python3 -m venv .venv
source .venv/bin/activate
pip install --upgrade pip
...

Requirements.txt:

strands-agents>=0.1.0
strands-agents-tools>=0.1.0
boto3>=1.34.0
...

• 사용 중인 리전(예: eu-central-1, us-east-1)에서 AWS Bedrock 모델 액세스를 활성화하세요.

  • AWS Bedrock > Bedrock Configuration > Model Access > AWS Nova-Pro 또는 Claude Sonnet, Opus

• 이 코드에서는 AWS가 다양한 리전에서 제공하는 AWS Nova-Pro를 사용하겠습니다.

• 모델 액세스 후, IAM에서 AWS Bedrock 서비스에 접근할 수 있도록 AmazonBedrockFullAccess 권한을 부여하세요.

• AWS 계정을 사용하여 AWS Bedrock 모델에 도달하는 두 가지 옵션:

  • AWS Config: aws configure를 사용하여 config 및 credentials 파일을 생성합니다.
  • .env 파일을 사용하여 변수 가져오기: .env 파일을 추가합니다:

AWS_ACCESS_KEY_ID=여기에_액세스_키_ID를_붙여넣으세요
AWS_SECRET_ACCESS_KEY=여기에_비밀_액세스_키를_붙여넣으세요

• Bedrock API 키를 생성하는 또 다른 대안적인 방법 (장기 또는 단기).

메모리(Memory) 구현

• 메모리 함수 (init, read, append):

import re, sys, json, datetime, requests
from pathlib import Path
from dotenv import load_dotenv
...

정보 검색 및 요약 에이전트 (Fetching Information & Summary Agent)

• 리서치 함수 (키워드 생성, DuckDuckGo 검색, 가져오기 및 수집):

from strands import Agent, tool
from strands.models.bedrock import BedrockModel

...

에이전트 상태(Agents State) 및 에이전트 도구(Agent Tools)

• 상태(State)는 에이전트 간에 통신할 필요가 있습니다 (EvalResult). Planner 에이전트는 에이전트 도구를 호출합니다.

from strands import Agent, tool
from typing import Optional
from pydantic import BaseModel, Field
...

생성(Generator) 및 평가(Evaluator) 에이전트

• Generator는 키워드를 생성하고, 리서치를 수행하며, 데이터(링크, 요약)를 메모리(Memory)에 추가한 다음, 블로그 포스트를 생성합니다.

from strands import Agent, tool
from strands.models.bedrock import BedrockModel

...

• Evaluator는 특정 기준(깊이, 최신성, 구조, 문체)에 따라 포스트를 평가하고, 그 결과를 메모리(Memory)에 추가합니다.

from strands import Agent, tool
from strands.models.bedrock import BedrockModel

...

Planner 에이전트 및 메인 실행 (Planner Agent & Main Run)

Planner 에이전트 및 메인 실행 (Planner Agent & Main Run)

[이번 청크]

• 각 에이전트에 대해 ChatBedrockConverse를 할당합니다.

from strands import Agent, tool
from strands.models.bedrock import BedrockModel

# Planner Agent 
_plan_model    = BedrockModel(model_id="us.amazon.nova-pro-v1:0", temperature=0.3)

def run(topic: str, max_iter: int = MAX_ITER) -> None:
    print(f"\n{'='*60}\n  TOPIC : {topic}\n  MAX ITER: {max_iter}\n{'='*60}")
    _state_init(topic, max_iter)
    _mem_init(topic)

    planner = Agent(
        model=_plan_model,
        tools=[generate_blog, evaluate_blog, search_web, fetch_page,
               summarize_article, read_memory, write_memory],
        system_prompt=(
            "You are an autonomous blog-pipeline orchestrator for senior AI/ML engineering content.\n\n"
            "Goal: produce a high-quality technical blog post (all eval scores ≥ 4) "
            f"within the allowed iterations.\n\n"
            "Tools available:\n"
            "  • generate_blog(extra_instructions)  — research + summarise + write / rewrite\n"
            "  • evaluate_blog()                    — score the current draft\n"
            "  • search_web(query)                  — ad-hoc article search\n"
            "  • fetch_page(url)                    — full raw text of a page\n"
            "  • summarize_article(url, title)      — LLM-distilled key points from a page\n"
            "  • read_memory()                      — inspect research & critiques\n"
            "  • write_memory(section, content)     — log planning notes\n\n"
            "Flow:\n"
            "  1. generate_blog() → 2. evaluate_blog() → 3. if rejected, generate_blog(extra_instructions=<fix>) → repeat.\n"
            "  Stop when accepted or iteration budget exhausted.\n"
            "  Use summarize_article between steps when you need deeper insight on a specific source."
        ),
    )

print("
[PlannerAgent] starting autonomous pipeline...\n")
    planner(
        f"Topic: **{topic}**\n"
        f"Max iterations allowed: {max_iter}\n\n"
        "전체 파이프라인을 자율적으로 실행합니다. 게시물이 승인되거나 예산이 소진될 때까지 생성하고, 평가하며, 반복합니다."
    )

    print(f"\n{'='*60}")
    print(f"  완료 — 반복 {_state['iteration']} | 승인됨={_state['done']}")
    print(f"  출력: {BLOG}  |  메모리: {MEM}")
    print(f"{'='*60}\n")

if __name__ == "__main__":