LangGraph, Claude, AWS를 활용한 AI 리서치 에이전트 구축
요약
LangGraph, Claude, AWS를 활용하여 자율적으로 웹과 학술 논문을 검색하고 보고서를 작성하는 AI 리서치 에이전트 구축 사례를 소개합니다. ReAct 패턴을 기반으로 추론과 도구 호출을 반복하며 구조화된 결과물을 생성하는 아키텍처를 다룹니다.
핵심 포인트
- LangGraph의 StateGraph를 이용한 ReAct 에이전트 구현
- Claude의 추론 능력과 LangGraph의 루프 제어 결합
- AWS ECS Fargate를 활용한 서버리스 컨테이너 호스팅
- 실시간 추론 과정 스트리밍 및 대화 메모리 관리
내가 만든 것
웹, Wikipedia, 학술 논문을 검색하여 어떤 질문에도 답하고, 그 추론 과정을 브라우저에 실시간으로 스트리밍하는 AI 리서치 에이전트입니다.
사용자 흐름 (User flow):
- 자연어로 어떤 리서치 질문이든 질문하기
- 에이전트가 어떤 도구 (tools)를 호출할지 결정하는 과정을 실시간으로 지켜보기
- 구조화된 보고서 읽기: 요약 (Summary), 주요 발견 사항 (Key Findings), 학술 연구 (Academic Research), 출처 (Sources), 결론 (Conclusion)
- 후속 질문하기 — 에이전트가 전체 대화 내용을 기억합니다
단순한 챗봇과의 핵심적인 차이점은 이 에이전트가 **자율적 (autonomous)**이라는 점입니다. 에이전트는 언제 검색할지, 무엇을 검색할지, 어떤 출처를 사용할지를 스스로 결정합니다. Claude가 추론 (reasoning)을 수행하고, LangGraph가 루프 (loop)를 제어합니다.
아키텍처 (Architecture)
Browser
│
▼
...
저의 이전 AI 이력서 분석기 (AI Resume Analyzer) 포스트와 동일한 CloudFront 패턴을 사용합니다 — /api/v1/*가 ALB로 프록시되어 프론트엔드가 CORS 문제 없이 단일 HTTPS 엔드포인트를 가질 수 있도록 합니다.
기술 스택 (Tech Stack)
| 계층 (Layer) | 기술 (Technology) |
|---|---|
| 프론트엔드 (Frontend) | React 18 + TypeScript + Vite + TailwindCSS |
| ... |
사용된 AWS 서비스 (AWS Services Used)
| 서비스 (Service) | 용도 (Purpose) |
|---|---|
| ECS Fargate | FastAPI를 위한 서버리스 컨테이너 호스팅 |
| ... |
핵심: LangGraph ReAct 에이전트
이 프로젝트에서 가장 흥미로운 부분은 에이전트 자체입니다. 고정된 파이프라인 (step 1 → step 2 → step 3) 대신, ReAct 에이전트는 답변을 향해 추론하며 나아갑니다:
- 추론 (Reason) — 어떤 정보가 필요한지 생각하기
- 행동 (Act) — 도구 호출 (웹 검색, Wikipedia, arXiv)
- 관찰 (Observe) — 결과 읽기
- 반복 (Repeat) — 답변하기에 충분한 정보를 얻을 때까지
LangGraph는 이를 StateGraph로 모델링합니다. 이는 각 노드 (node)가 Claude 모델이거나 도구 실행기 (tool executor)인 유향 그래프 (directed graph)이며, 엣지 (edge)는 도구를 호출할 시점과 최종 답변을 반환할 시점을 정의합니다.
# agent/graph.py
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
...
should_continue 함수는 핵심적인 결정 지점입니다. Claude의 응답에 도구 호출 (tool calls)이 포함되어 있으면 tools 노드로 라우팅하고, 그렇지 않으면 답변이 준비된 것으로 간주하여 그래프를 종료합니다.
AgentState 및 대화 메모리 (Conversation Memory)
AgentState는 LangGraph의 add_messages 리듀서 (reducer)를 사용하여 전체 메시지 기록을 보유합니다. 즉, 새로운 메시지는 기존 리스트를 대체하는 대신 리스트에 추가됩니다.
# agent/state.py
from langgraph.graph import add_messages
from typing import Annotated
...
이 단일 필드만 있으면 멀티턴 메모리 (multi-turn memory)를 구현하는 데 충분합니다. 각 대화는 DynamoDB 기록으로부터 시작되므로, 후속 질문은 이전에 말한 모든 내용에 대한 전체 문맥 (context)을 갖게 됩니다.
세 가지 리서치 도구
에이전트는 세 가지 도구를 가지고 있습니다. Claude는 질문을 바탕으로 어떤 도구를 호출할지 자율적으로 결정합니다.
Tavily — 웹 검색 (Web Search)
Tavily는 LLM 에이전트를 위해 특화되어 설계되었습니다. 일반적인 검색 엔진의 노이즈 없이 깔끔하고 구조화된 결과를 반환합니다.
# agent/tools.py
from langchain_community.tools.tavily_search import TavilySearchResults
...
Wikipedia — 배경 지식 (Background Facts)
이미 잘 정립된 주제의 경우, Wikipedia는 Tavily 할당량 (quota)을 소모하지 않고도 신뢰할 수 있는 배경 지식을 제공합니다.
@tool
def wikipedia_search(query: str) -> str:
"""주제에 대한 배경 정보를 찾기 위해 Wikipedia를 검색합니다."""
...
arXiv — 학술 논문 (Academic Papers)
과학적 또는 기술적인 질문의 경우, 에이전트는 동료 검토 (peer-reviewed)를 거친 연구를 찾기 위해 arXiv를 검색합니다. 이는 결과물을 일반적인 웹 검색과 차별화합니다.
@tool
def arxiv_search(query: str) -> str:
"""주제에 대한 학술 논문 및 연구를 찾기 위해 arXiv를 검색합니다."""
...
SSE 스트리밍 (SSE Streaming) — 토큰 + 실시간 에이전트 추적 (Live Agent Trace)
이 부분은 UI가 살아있는 것처럼 느껴지게 만드는 요소입니다. 완전한 답변을 위해 20초 동안 기다리는 대신, 브라우저는 두 가지 유형의 이벤트를 동시에 수신합니다.
- 추적 이벤트 (Trace events) — 에이전트가 수행하는 모든 도구 호출을 실시간 패널에 표시
- 토큰 이벤트 (Token events) — 답변을 단어 단위로 스트리밍
SSE 이벤트 프로토콜:
data: {"type": "conversation_id", "conversation_id": "abc-123"}
data: {"type": "trace", "step": "tool_start", "tool": "web_search", "input": "LangGraph tutorial"}
...
스트리밍 엔드포인트(streaming endpoint)는 LangGraph의 astream_events를 사용합니다. 이는 에이전트의 실행 그래프(execution graph) 내 모든 이벤트를 생성(yield)하는 비동기 제너레이터(async generator)입니다.
# routers/chat.py
async def stream_agent_response(message: str, conversation_id: str):
history = await dynamodb_service.get_messages(conversation_id)
...
FastAPI 엔드포인트는 media_type="text/event-stream"을 설정하고 StreamingResponse를 반환합니다. 따라서 WebSockets는 필요하지 않습니다.
멀티턴 대화 (Multi-turn Conversations)
모든 대화는 DynamoDB에 저장됩니다. 사용자가 후속 질문을 보내면 전체 대화 기록(history)이 로드되어 에이전트에게 다시 전달되므로, 에이전트는 논의된 모든 내용에 대한 전체 문맥(context)을 파악할 수 있습니다.
DynamoDB 스키마 (Schema)
테이블은 단일 테이블 디자인(single-table design)을 사용하며, 최신순으로 대화 목록을 나열하기 위해 GSI(Global Secondary Index)를 활용합니다.
| 키 (Key) | 값 (Value) | 용도 (Purpose) |
|---|---|---|
pk | CONV#{conversation_id} | 파티션 키 (Partition key) |
| ... |
GSI (entity-type-index)를 사용하면 전체 테이블을 스캔하지 않고도 가장 최근에 업데이트된 순서대로 모든 대화 목록을 효율적으로 나열할 수 있습니다.
# services/dynamodb_service.py
async def save_messages(conversation_id: str, messages: list[BaseMessage]) -> None:
with table.batch_writer() as batch:
...
API 엔드포인트 (Endpoints)
| 메서드 (Method) | 엔드포인트 (Endpoint) | 설명 (Description) |
|---|---|---|
| GET | /health | 상태 확인 (Health check) |
| ... |
AWS CDK를 활용한 인프라 (Infrastructure with AWS CDK)
두 개의 스택(stack)으로 구성됩니다. 프론트엔드가 서버 사이드 렌더링(server-side rendering)이 없는 순수 정적 사이트이기 때문에, 이전에 살펴본 이력서 분석기(resume analyzer)보다 구조가 더 단순합니다.
BackendStack
2개의 가용 영역(AZs)을 가진 VPC, 1개의 NAT 게이트웨이(NAT Gateway), ECS Fargate 서비스 및 ALB(Application Load Balancer)로 구성됩니다. 두 API 키는 컨테이너 시작 시 SSM Parameter Store에서 주입됩니다.
# stacks/backend_stack.py (simplified)
fargate_service = ecs_patterns.ApplicationLoadBalancedFargateService(
self, "BackendService",
...
FrontendStack
SSE(Server-Sent Events)를 위한 두 가지 중요한 타임아웃 설정이 적용된 CloudFront:
# stacks/frontend_stack.py (간략화 버전)
api_behaviour = cloudfront.BehaviorOptions(
origin=alb_origin,
...
SSE를 위해서는 CloudFront read timeout (기본값 30초)을 반드시 연장해야 합니다. REST 엔드포인트와 달리, SSE 연결은 에이전트가 추론(reasoning)하고 내용을 작성하는 동안 계속 열려 있어야 합니다. 복잡한 질문의 경우 이 시간이 30초를 쉽게 초과할 수 있습니다.
테스트 전략 (Testing Strategy)
Backend (백엔드) — astream_events와 DynamoDB를 모킹(mocked)한 pytest를 사용합니다. 실제 API 호출은 필요하지 않습니다.
cd backend
pytest # 모든 테스트
pytest tests/agent/ # ReAct 에이전트 + 도구(tools)
...
| 테스트 파일 | 커버리지 내용 |
|---|---|
test_state.py | add_messages가 메시지를 올바르게 추가하고 이력을 보존하는지 확인 |
| ... |
Frontend (프론트엔드) — Vitest + Testing Library를 사용합니다.
cd frontend
npm test
npm run test:coverage
교훈 (Lessons Learned)
1. ReAct 에이전트가 결정합니다 — 로직을 하드코딩할 필요가 없습니다
고정된 파이프라인(fixed pipeline)을 사용한다면 "만약 과학 질문이라면 → arXiv를 사용하라"와 같은 명시적인 코드가 필요했을 것입니다. 하지만 ReAct 루프를 사용하면 Claude가 문맥(context)에 따라 스스로 결정할 수 있습니다. 에이전트는 종종 단일 질문에 대해 세 가지 도구를 모두 조합하여 사용하기도 합니다.
2. SSE를 위해 CloudFront read timeout을 반드시 늘려야 합니다
CloudFront의 기본 read timeout은 30초입니다. 복잡한 연구 질문을 처리하는 SSE 연결은 이보다 오래 지속됩니다. 이 값을 120초로 설정하지 않으면, 스트림이 응답 중간에 조용히 끊겨버립니다. 에러가 발생하는 것이 아니라, 브라우저가 그냥 멈춰버리는 현상이 발생합니다. 이는 진단하기 가장 까다로웠던 버그였습니다.
3. ALB idle timeout 또한 연장해야 합니다
CloudFront(read timeout)와 ALB(idle timeout) 모두 120초로 설정해야 합니다. 하나만 수정해서는 긴 연결을 유지할 수 없습니다. ALB의 기본값인 60초는 변경하지 않을 경우 CloudFront보다 먼저 작동하여 연결을 끊어버립니다.
4. 프라이빗 ECS 태스크에는 NAT Gateway가 필요합니다
프라이빗 서브넷(private subnet)에 있는 ECS 태스크가 Anthropic, Tavily, Wikipedia, arXiv에 접속하려면 NAT Gateway가 필요합니다. NAT Gateway가 없으면 모든 외부 API 호출이 조용히 타임아웃됩니다. DNS 에러가 발생하는 것이 아니라, 요청이 그냥 멈춰버린 상태(hanging request)가 됩니다.
5. 대화 메모리(conversation memory)를 위해 필요한 것은 add_messages뿐입니다
LangGraph의 add_messages 리듀서(reducer)는 상태(state) 리스트에 새로운 메시지를 자동으로 추가합니다. DynamoDB의 전체 히스토리를 초기 상태(initial state)로 전달하면, 추가적인 로직 없이도 에이전트가 과거 대화 턴(turn)에 대한 완전한 문맥(context)을 가질 수 있습니다.
6. Apple Silicon에서 linux/amd64용 Docker 이미지 빌드하기
ECS Fargate는 기본적으로 x86에서 실행됩니다. 플랫폼을 지정하지 않고 M 시리즈 Mac에서 빌드하면 Fargate에서 시작되지 않는 arm64 이미지가 생성됩니다. Docker 빌드 시 항상 --platform linux/amd64를 추가하거나, Dockerfile에 이를 설정하십시오.
7. 두 개의 SSM 파라미터 — 모두 컨테이너 시작 시 주입됨
이 프로젝트에는 Anthropic과 Tavily라는 두 개의 API 키가 필요합니다. 두 키 모두 SSM SecureString으로 저장되며 컨테이너 시작 시 환경 변수(environment variables)로 주입됩니다. 키는 CDK 출력, CloudFormation 템플릿 또는 Docker 이미지에 절대 나타나지 않습니다.
GitHub
전체 소스 코드는 GitHub에서 확인할 수 있습니다:
👉 github.com/sanjaypatoliya/ai-research-agent
저자 소개
저는 Sanjay Patoliya입니다. AWS에서 프로덕션급(production-ready) AI 시스템을 구축하고 있는 7개의 AWS 자격증을 보유한 AWS 인증 엔지니어입니다.
- LinkedIn: linkedin.com/in/sanjaykumar-patoliya-b234a287
- GitHub: github.com/sanjaypatoliya
- Email: sbpatoliya@gmail.com
원문은 sanjaypatoliya.com에 게시되었습니다.
AI 자동 생성 콘텐츠
본 콘텐츠는 Dev.to AI tag의 원문을 AI가 자동으로 요약·번역·분석한 것입니다. 원 저작권은 원저작자에게 있으며, 정확한 내용은 반드시 원문을 확인해 주세요.
원문 바로가기