AGENT DESIGN PATTERNS
먼저 네 가지 기본 형태를 보자
상자 안에는 누가 무엇을 하는지, 화살표에는 작업과 결과가 어디로 흐르는지를 표시했다.
단일 에이전트 + 도구
한 에이전트가 판단하고, 필요한 도구를 호출한다.
검색·계산은 도구가 하고, 다음 행동과 최종 답변은 같은 에이전트가 결정한다. 필요한 만큼 호출과 관찰을 반복한다.
순차 워크플로
작업 1 → 2 → 3 → 4를 마친 뒤 결과를 종합한다.
앞 단계의 결과를 다음 단계에 넘긴다. 종합 단계는 누적된 작업 결과를 받아 하나의 답변으로 정리한다.
병렬 워크플로
독립적인 작업을 동시에 시작하고, 결과를 모은다.
각 에이전트는 다른 에이전트의 결과를 기다리지 않고 일한다. 종합 단계에서 세 결과를 함께 읽는다.
계층적 작업 분해
상위 에이전트가 목표를 나누고, 하위에서 다시 나눈다.
위에서 아래로 일을 맡기고, 아래에서 위로 결과를 보고한다. 담당 에이전트가 하위 결과를 정리하고 총괄이 최종 답변을 만든다.
LLM에 도구를 붙이면 처음에는 아주 작은 그림으로 충분해 보인다.
User → Model → Tool → Observation → Model → Final answer모델은 질문을 읽고, 검색 도구를 한 번 호출하고, 결과를 보고 답한다. 시연에서는 이 흐름이 곧바로 “Agent”처럼 느껴진다.
하지만 서비스를 만들기 시작하면 그림 바깥의 질문이 곧 설계가 된다. 언제 멈추는가? 서로 독립적인 조사와 데이터 조회는 무엇을 병렬로 돌리는가? 외부 메일이나 티켓 생성은 누가 승인하는가? 실패는 재시도할 것인가, 다른 경로로 전환할 것인가? 초안의 품질이 낮다면 얼마나 다시 고칠 것인가?
이 글의 관점은 단순하다. “계속 생각해”는 아키텍처가 아니다. 실행 경로마다 상태, 허용 행동, 종료 조건, 실패 처리가 있어야 한다. 이후 글에서 async, worker, 상태, 스트리밍을 다루기 전에 먼저 이 여섯 패턴을 공통 어휘로 잡아두려는 이유다.
먼저 큰 그림을 보자: 여섯 가지 패턴
| 패턴 | 푸는 문제 | 런타임이 강제할 것 |
|---|---|---|
| Tool Use / 제한된 도구 호출 루프 | 다음 행동과 관측 결과를 어떻게 연결할까? | 도구 목록, 입력 형식, 단계 예산 |
| Orchestrator / Parallel | 어떤 전문 역할을 부르고 무엇을 동시에 돌릴까? | 경로 선택, 동시 실행 한도, 결과 병합 |
| Planner / Executor | 의존성이 있는 여러 단계를 어떻게 실행할까? | DAG 검증, 실행 가능한 단계 선택 |
| Human in the Loop | 부수 효과 직전에 어떻게 사람을 넣을까? | 정확한 호출의 승인 관문, 감사 기록 |
| Retry / Fallback | 어떤 실패를 다시 시도하고 언제 기능을 축소할까? | 오류 분류, 시도 횟수와 대기 시간 한도 |
| Evaluator / Refiner | 답변 품질을 어떻게 제한된 루프로 개선할까? | 평가자 검증, 개선 예산 |
각 칸은 특정 제품이나 프레임워크의 기능명이 아니다. 여러 구현에서 반복되는 제어 흐름이다. 모든 요청에 여섯 가지가 필요한 것도 아니다. 작은 읽기 전용 검색 답변에는 제한된 도구 루프 하나면 충분할 수 있다. 복잡해질 때 필요한 패턴만 명시적으로 더하는 편이, 처음부터 Agent를 여러 개 만드는 것보다 낫다.
패턴 1 — 도구 호출: 언제 멈출지 먼저 정한다
가장 작은 Agent 런타임은 모델 정책이 다음 행동을 반환한다고 가정한다. 행동은 도구 호출이거나 최종 답변이다. 핵심은 모델 호출이 아니라 상태 머신의 경계다.
from dataclasses import dataclass, field
from typing import Any, Callable
Tool = Callable[..., Any]
@dataclass
class AgentState:
query: str
history: list[dict[str, Any]] = field(default_factory=list)
def run_agent(query: str, policy, tools: dict[str, Tool], max_steps: int = 8) -> str:
state = AgentState(query=query)
for _ in range(max_steps):
action = policy(state)
if action.get("type") == "final":
return action["answer"]
if action.get("type") != "tool":
raise ValueError("unknown action type")
name, args = action.get("name"), action.get("args")
if name not in tools:
raise ValueError(f"unknown tool: {name!r}")
if not isinstance(args, dict):
raise TypeError("tool args must be a dict")
observation = tools[name](**args)
state.history.append({"action": action, "observation": observation})
raise RuntimeError("agent exceeded max_steps without a final answer")여기에는 의도적으로 명시적인 상태, 도구 레지스트리, 인자 검증, 관측 이력, 최종 답변, max_steps가 있다. 모델이 검색을 제안해도 레지스트리 밖의 도구를 부를 수 없다. 관측 결과는 다음 정책 판단에 남는다. 그리고 정책이 계속 도구를 고르더라도 런타임은 무한히 따르지 않는다.
도구 레지스트리는 권한 부여 장치는 아니지만 첫 번째 허용 목록이다. 실제 서비스라면 사용자 권한과 테넌트 정책을 이 레지스트리보다 더 안쪽의 실행 경계에서 확인해야 한다.
READ_ONLY_TOOLS = {"search", "get_customer"}
def execute_tool(name: str, args: dict, tools: dict, user_permissions: set[str]):
if name not in tools:
raise ValueError(f"unknown tool: {name}")
if name not in user_permissions:
raise PermissionError(f"not allowed to invoke {name}")
if not isinstance(args, dict):
raise TypeError("arguments must be an object")
return tools[name](**args)max_steps만으로 모든 예산을 표현할 수는 없다. 뒤에서는 마감 시간, 토큰, 비용 같은 독립 제한도 필요하다. 하지만 첫 번째 질문은 늘 같다. 이 루프가 종료한다는 사실을 코드로 설명할 수 있는가?
패턴 2 — 작업 분배와 병렬 실행: 역할보다 독립성을 먼저 본다
“NVIDIA의 최근 실적과 기업 가치를 분석해줘”라는 요청을 생각해 보자. 뉴스와 공시를 읽는 조사, 구조화된 수치를 조회하는 데이터 분석, 외부 시스템을 바꾸는 행동은 같은 런타임 안에서 서로 다른 책임을 가질 수 있다.
조사와 데이터 분석처럼 서로의 결과를 기다리지 않는 작업은 병렬 실행 후보가 된다. Python의 asyncio.gather는 awaitable들을 동시에 실행하고 입력 순서에 맞춰 결과를 모은다. 실패를 개별 결과로 다루려면 return_exceptions=True처럼 정책을 명시해야 한다. Python asyncio 문서는 이러한 동시성과 예외 동작을 구분해 설명한다.
import asyncio
async def run_specialists(query: str) -> dict[str, object]:
names = ["research", "data"]
calls = [research_agent(query), data_agent(query)]
outcomes = await asyncio.gather(*calls, return_exceptions=True)
results, failures = {}, {}
for name, outcome in zip(names, outcomes):
if isinstance(outcome, Exception):
failures[name] = str(outcome)
else:
results[name] = outcome
return {"results": results, "failures": failures}여러 Agent로 분리하는 것이 의미 있는 이유는 Agent 수 자체가 아니다.
- 전문화가 서로 다를 때
- 각 역할이 접근할 수 있는 데이터와 도구 권한이 다를 때
- 지연 시간 특성이 달라 독립 작업을 겹칠 수 있을 때
- 하나의 실패가 다른 결과를 버리게 하지 않아야 할 때
반대로 경로 선택은 충분히 결정론적일 수 있다. 키워드, 요청 유형, 권한, 데이터 최신성 규칙으로 분기할 수 있다면 그 편이 더 저렴하고 테스트하기 쉽다.
def route(request: str) -> list[str]:
text = request.lower()
selected = []
if any(word in text for word in ("revenue", "margin", "valuation")):
selected.append("data")
if any(word in text for word in ("news", "why", "recent")):
selected.append("research")
return selected or ["research"]모델이 경로 후보를 제안할 수는 있다. 하지만 “Agent가 많을수록 똑똑하다”는 법칙은 없다. 필요 없는 역할 하나는 맥락 전달, 재시도 정책, 권한, 추적 기록, 비용이라는 새 경계를 하나 더 만든다.
패턴 3 — 계획과 실행: 계획은 만들고, 그래프는 검증한다
복잡한 요청은 계획을 먼저 만들면 읽기 쉬워진다. 예를 들어 기업 가치 평가 보고서는 research → valuation → report와 research → peer_analysis → report라는 의존성을 가질 수 있다.
LLM이 계획을 생성할 수 있다. 그러나 계획이 유효한지 LLM에게 다시 물어보는 것은 검증이 아니다. 중복 단계, 존재하지 않는 의존성, 자기 의존성, 순환은 코드가 실행 전에 막아야 한다.
from dataclasses import dataclass
@dataclass(frozen=True)
class PlanStep:
id: str
instruction: str
depends_on: tuple[str, ...] = ()
def ready_steps(plan: list[PlanStep], completed: set[str], terminal: set[str]):
return [
step for step in plan
if step.id not in terminal
and all(dep in completed for dep in step.depends_on)
]실행 가능 집합은 “아직 실행하지 않았고, 모든 의존성이 성공적으로 끝난 단계”다. 이 기준이면 완료된 의존성 결과만 Executor에게 전달할 수 있다. 실패한 분기에 매달린 후속 단계는 failed가 아니라 skipped로 남기는 편이 실행 궤적을 더 정확하게 설명한다. 실행 가능한 여러 단계는 병렬로 실행할 수 있지만, 계획 순서와 결과 병합 규칙은 결정론적으로 유지하는 것이 좋다.
검증의 핵심을 작게 쓰면 다음과 같다.
def validate_plan(plan: list[PlanStep]) -> None:
ids = [step.id for step in plan]
if len(ids) != len(set(ids)):
raise ValueError("duplicate step id")
known = set(ids)
for step in plan:
if step.id in step.depends_on:
raise ValueError("self dependency")
if not set(step.depends_on) <= known:
raise ValueError("missing dependency")
# Production code must also run a cycle check before execution.모든 문제에 Planner가 필요한 것은 아니다. 분기가 적고 규칙이 안정적인 청구서 승인, ETL, 정해진 온보딩 같은 작업은 고정 워크플로가 낫다. 계획을 생성하는 비용, 오류, 검증 책임을 추가하지 않고도 더 쉽게 테스트할 수 있기 때문이다.
패턴 4 — 사람의 승인: 프롬프트가 아니라 실행 관문
HITL(Human in the Loop)은 모델에게 “필요하면 사용자에게 물어봐”라고 적는 일이 아니다. 읽기 전용 도구는 자동 실행할 수 있어도, 외부 메일 발송·티켓 생성·금융 거래 같은 부수 효과는 도구 실행 직전에 결정론적 관문을 지나야 한다.
from dataclasses import dataclass
from typing import Any
@dataclass(frozen=True)
class ToolSpec:
handler: Any
side_effecting: bool = False
async def execute_action(action, tools: dict[str, ToolSpec], request_approval):
name, args = action["name"], action["args"]
spec = tools[name]
if spec.side_effecting:
try:
approved = await request_approval(tool_name=name, args=args)
except Exception:
approved = False # approval service failure is denial
if not approved:
return {"status": "rejected", "tool": name}
return {"status": "executed", "result": await spec.handler(**args)}여기서 실패 시 차단은 편의가 아니라 안전성의 성질이다. 승인 콜백이 실패했을 때 실행해 버리면 승인은 경계가 아니다. 또한 승인에는 정확한 도구 이름과 인자, 사용자 요청, 승인 사유, 결정 시각을 감사 로그에 남길 이유가 있다.
Authorization과 approval도 구분해야 한다.
| 통제 | 질문 |
|---|---|
| 권한 부여 | 이 사용자는 이 도구와 행동을 수행할 권한이 있는가? |
| 승인 | 권한은 있지만, 이 행동을 정확히 그대로 지금 실행해도 되는가? |
승인 뒤 인자가 바뀌면 예전 승인을 재사용해서는 안 된다. 운영 환경에서는 도구와 정규화한 인자의 해시를 승인 기록에 묶고, 실행 시 다시 비교하는 방식이 필요하다. HITL과 재시도가 만나는 설계는 Agent Tool은 정확히 한 번 실행될 수 있을까에서 이어진다.
패턴 5 — 재시도와 대체 경로: 실패도 분류해야 한다
재시도는 “오류가 나면 한 번 더”가 아니다. 시간 초과, 일시적 네트워크 실패, 일부 호출량 제한은 재시도 후보일 수 있다. 잘못된 인자, 권한 거부, 스키마 위반은 보통 즉시 실패해야 한다. 뒤의 종류를 재시도하면 돈과 시간만 쓰고 잘못된 요청을 숨긴다.
import asyncio
RETRYABLE = (TimeoutError, ConnectionError)
async def call_with_retry(primary, fallback, max_attempts: int = 3):
errors = []
for attempt in range(1, max_attempts + 1):
try:
return {"source": "primary", "value": await primary(), "errors": errors}
except RETRYABLE as exc:
errors.append({"attempt": attempt, "type": type(exc).__name__})
if attempt < max_attempts:
await asyncio.sleep(2 ** (attempt - 1))
except Exception:
raise # invalid input, permission, schema: do not fallback
return {"source": "fallback", "value": await fallback(), "errors": errors}여기서 max_attempts에는 첫 호출도 포함된다. 대기에도 상한이 있어야 하며, 실제 운영에서는 지터, 서버의 Retry-After, 전체 마감 시간과 비용 예산을 함께 고려한다. Fallback도 모든 오류를 가리는 장치가 아니다. 주 경로가 유효하지 않은 요청을 거절했다면 예비 경로에 같은 잘못을 보내지 않는 편이 맞다.
부수 효과와 재시도가 만나면 더 어려워진다. 클라이언트에서 시간 초과가 발생했다고 원격 행동이 실패한 것은 아니다. 이미 티켓이 만들어진 뒤 응답만 잃었을 수 있다. 같은 멱등성 키 없이 재시도하면 중복 실행이 가능하다. 승인·멱등성·재시도 경계는 Agent Tool은 정확히 한 번 실행될 수 있을까에서 다룬다.
패턴 6 — 평가와 수정: 품질 개선에도 횟수 제한이 있다
Reflection은 보통 다음 흐름이다.
평가자가 피드백을 주고 Refiner가 초안을 고친다. 여기서 Evaluator가 정답 기준은 아니다. 같은 모델 계열의 판정자는 같은 맹점을 공유할 수 있다. 평가 기준의 보정, 사람이 만든 별도 검증 데이터셋, 오프라인 평가가 있어야 점수를 품질 자체로 오해하지 않을 수 있다.
from dataclasses import asdict, dataclass
@dataclass(frozen=True)
class Evaluation:
passed: bool
score: float
feedback: list[str]
def refine_until_done(request, generate, evaluate, refine, max_refinements: int = 2):
draft, history = generate(request), []
for refinement in range(max_refinements + 1):
verdict = evaluate(request, draft)
if not isinstance(verdict, Evaluation) or not 0.0 <= verdict.score <= 1.0:
raise ValueError("invalid evaluator output")
if not all(isinstance(item, str) for item in verdict.feedback):
raise ValueError("feedback must be strings")
history.append(asdict(verdict))
if verdict.passed:
return {"answer": draft, "stop_reason": "passed", "evaluations": history}
if refinement == max_refinements:
return {"answer": draft, "stop_reason": "max_refinements", "evaluations": history}
draft = refine(request, draft, verdict)max_refinements는 평가 횟수가 아니라 수정 횟수다. 두 번 고칠 수 있다면 최초 초안을 포함해 최대 세 번 평가한다. 마지막 초안도 평가해야 한다. 이 규칙과 stop_reason이 있어야 비용·지연 시간·품질 사이의 상충 관계를 나중에 측정할 수 있다.
여섯 패턴은 실제로 이렇게 조합된다
실제 요청에서는 패턴이 차례로 겹친다.
서로 독립적인 작업은 동시에 실행
여기까지 오면 이제 backend 문제가 시작된다. 독립 작업은 어떤 동시 실행 한도 아래 병렬로 돌릴까? 30초짜리 run을 HTTP 요청이 계속 소유해도 될까? 진행 상태는 어디에 저장할까? 브라우저에는 SSE로 무엇을 보내야 할까? 승인을 기다리는 동안 worker가 죽으면? 같은 세션에 run이 두 개면 누가 어떤 상태를 바꾸는가?
이 질문들이 Agent Engineering 시리즈의 출발점이다. 패턴은 “무슨 제어 흐름이 필요한가”를 알려준다. backend는 그 제어 흐름을 여러 사용자와 실패 속에서도 실제로 소유하는 방법을 다룬다.
LLM과 코드의 책임 경계
LLM은 유용한 정책 구성 요소다. 도구 후보나 계획을 제안하고, 관측 결과를 해석하고, 초안을 평가할 수 있다. 그러나 그 사실이 런타임의 안전 속성을 모델에게 위임해야 한다는 뜻은 아니다.
| 결정 | LLM이 할 수 있는 일 | 코드가 강제할 일 |
|---|---|---|
| 도구 선택 | 후보 제안 | 허용된 레지스트리와 사용자 권한 |
| 도구 인자 | 구조화된 인자 생성 | 스키마 검증, 크기와 범위 제한 |
| 계획 생성 | 단계와 의존성 제안 | 중복·누락·자기 의존·순환 검증 |
| 병렬 실행 후보 | 독립 작업 제안 | 동시 실행 한도, 취소와 병합 정책 |
| HITL 필요성 | 위험 신호 설명 | 부수 효과 전 최종 승인 관문 |
| 재시도 | 맥락과 대안 제안 | 오류 분류, 최대 시도 횟수, 대기 시간 증가 |
| Reflection | 피드백 생성 | Evaluator 스키마와 루프 예산 |
| 권한 | 설명할 수는 있음 | 권한 부여 강제 |
| 종료 | 최종 답변 제안 | 단계·시간·비용의 강제 상한 |
LLM은 정책의 일부가 될 수 있지만, 안전성과 실행 상한을 보장하는 런타임 자체가 되어서는 안 된다. 이 표는 모델이 나쁘다는 말이 아니다. 모델은 확률적으로 다음 행동을 선택하고, 런타임은 그 행동이 시스템 계약 안에 있는지 검증한다. 둘을 분리해야 각자의 실패를 관찰하고 바꿀 수 있다.
Python 직접 구현을 이해한 뒤 LangGraph에 연결하기
이 글은 LangGraph 사용 안내서가 아니다. 위 패턴을 Python으로 직접 먼저 보인 이유는 프레임워크 API보다 구조를 먼저 소유하기 위해서다. 그다음 대응은 자연스럽다.
| Python 직접 구현 패턴 | LangGraph에서의 대응 개념 |
|---|---|
| 도구 루프 | 그래프 루프와 조건부 간선 |
| Orchestrator | 병렬 분기와 fan-out/fan-in |
| Planner / Executor | 동적 작업 그래프 또는 상태 저장 워크플로 |
| HITL | 중단, 체크포인트, 재개 |
| Retry / Fallback | 재시도 정책과 조건부 경로 선택 |
| Evaluator / Refiner | 예산이 있는 순환 |
현재 LangGraph 문서도 결정론적 단계와 LLM 주도 단계를 같은 그래프에 섞는 오케스트레이션 런타임으로 설명하며, 영속 실행·스트리밍·HITL를 주요 기능으로 든다. 개요와 중단 문서는 일시 중지 지점, 영속 체크포인터, 동일 thread ID를 사용한 재개 같은 구체적 요구를 설명한다. 그것은 프레임워크를 선택할 좋은 이유가 될 수 있다. 다만 API가 패턴을 대신 설계해 주지는 않는다.
다음: 패턴 뒤에는 백엔드가 있다
이제 Agent를 어떤 패턴으로 구성할지 안다. 그다음은 그 패턴이 어디에서, 누가 소유하며, 어떤 저장소와 연결로 실행되는지다.
다음 글 Agent Backend 한 장 지도에서는 API, queue, worker, PostgreSQL, Redis, SSE가 각각 어떤 책임과 실행 수명을 갖는지 전체 구조를 먼저 본다. 그다음 에이전트는 왜 Async가 필요한가에서 LLM·DB·검색 API를 기다리는 동안 backend가 어떻게 동시에 진행하는지 execution 모델부터 깊게 다룬다.
Discussion
토론
질문, 정정, 반례, 다른 해석이 있다면 남겨 주세요.