[TIL]Agent Safety와 Plan-Execute-Replan Workflow

1. 학습 주제

Agent가 Tool을 사용할 때 필요한 안전 제어와, 복잡한 작업을 계획 -> 실행 -> 검증 -> 재계획 흐름으로 처리하는 Workflow 구조를 학습했다.

핵심 내용은 다음과 같다.

• Agent Memory 보완
• thread_id 기반 대화 기억 분리
• Agent Safety
• 재시도, Fallback, Handoff 처리
• Graph의 종료 조건과 recursion_limit
• Planner
• Executor
• Validator
• Replanner
• Pydantic 기반 실행 계약 정의

---

2. Agent Memory 보완

Agent Memory는 이전 대화를 기억해 다음 요청에서 활용하는 구조이다.
이번 실습에서는 Checkpointer를 사용해 같은 thread_id의 대화 메시지를 이어가도록 구성했다.

checkpointer=InMemorySaver()

대화는 thread_id를 기준으로 분리된다.

config = {
    "configurable": {
        "thread_id": "student-1"
    }
}

같은 thread에서는 이전에 말한 이름, 동행인, 예산을 기억해 다음 Tool 호출 인자로 사용할 수 있다.

1. 내 이름은 수진이고 아이와 여행해. 전체 예산은 20만원이야.
2. 내 조건에 맞는 장소를 추천해줘.

두 번째 질문에는 이름, 동행인, 예산이 직접 들어있지 않지만, 같은 thread에 이전 대화가 저장되어 있다면 Agent가 그 정보를 찾아 Tool 인자를 만들 수 있다.

---

3. thread_id가 중요한 이유

Agent Memory에서 중요한 점은 사용자별 기억이 섞이면 안 된다는 것이다.

student-1
-> 수진, 아이 동반, 예산 정보 있음

student-2
-> 이전 정보 없음

student-1에서 장소 추천을 요청하면 이전 정보를 활용할 수 있다.
하지만 student-2에서 같은 질문을 하면 정보가 없으므로 Tool을 호출하지 않고 추가 질문을 해야 한다.

즉, thread_id는 “이 대화가 누구의 흐름인지” 구분하는 기준이다.

---

4. Agent Safety

Agent는 Tool을 사용할 수 있지만, 모든 판단을 모델에게만 맡기면 위험할 수 있다.
그래서 실패 결과에 따라 재시도할지, 중단할지, 담당자에게 넘길지 결정하는 안전 제어가 필요하다.

이번 실습에서는 Tool 결과를 다음 상태로 나누었다.

ResultCode	의미	처리
FOUND	검색 성공	성공 종료
NOT_FOUND	결과 없음	제한 횟수 안에서 재시도
TIMEOUT	일시적 실패	제한 횟수 안에서 재시도
AUTH_ERROR	권한 문제	재시도 없이 Handoff
NOT_RUN	아직 실행 안 함	초기 상태

---

5. Retry, Fallback, Handoff

실패라고 해서 모두 같은 방식으로 처리하면 안 된다.

NOT_FOUND / TIMEOUT
-> 다시 시도 가능

AUTH_ERROR
-> 다시 시도해도 해결되지 않음
-> 담당자 확인 필요

처리 흐름은 다음과 같다.

검색 실행
-> 결과 확인
-> FOUND면 성공 종료
-> NOT_FOUND 또는 TIMEOUT이면 재시도
-> 최대 횟수 초과 시 Fallback
-> AUTH_ERROR면 Handoff

Fallback은 자동 처리를 더 이상 하지 않고 사용자에게 조건 변경을 요청하는 흐름이다.

조건에 맞는 장소를 찾지 못했습니다.
실내, 무료, 산책 중 하나로 조건을 변경해 주세요.

Handoff는 자동 처리로 해결할 수 없는 상황을 담당자 확인으로 넘기는 흐름이다.

인증 문제로 자동 검색을 계속할 수 없습니다.
담당자 확인이 필요합니다.

---

6. 안전 규칙은 모델이 아니라 Graph가 강제

중요한 점은 재시도 여부와 최대 횟수를 모델의 판단에만 맡기지 않는다는 것이다.
모델은 답변을 만들고 Tool을 고를 수 있지만, 반복 제어와 종료 조건은 Graph State와 Edge에서 강제한다.

def route_after_guard(state):
    if result == "AUTH_ERROR":
        return "finalize_handoff"
    if result == "FOUND":
        return "agent"
    if result in RETRYABLE_RESULTS and state["attempts"] < state["max_attempts"]:
        return "agent"
    return "finalize_fallback"

즉, Agent가 계속 Tool을 호출하려 해도 Graph의 안전 규칙이 막을 수 있다.

---

7. recursion_limit

Graph에서 종료 조건을 잘못 만들면 무한 반복이 생길 수 있다.

search_destination
-> search_destination
-> search_destination
-> ...

이런 경우 recursion_limit가 마지막 안전장치 역할을 한다.

graph.invoke(state, config={"recursion_limit": 5})

하지만 recursion_limit는 최후의 방어선일 뿐이다.
정상적인 Graph라면 업무 조건으로 먼저 END에 도달해야 한다.

---

8. Plan-Execute-Replan Workflow

복잡한 작업은 한 번에 처리하기보다 단계별로 나누는 것이 좋다.

이번 실습에서는 여행 일정 생성을 다음 흐름으로 구성했다.

Planner
-> Executor
-> Validator
-> Replanner

각 단계의 역할은 다음과 같다.

단계	역할
Planner	목표를 실행 가능한 Task 목록으로 분해
Executor	Plan에 적힌 Task를 순서대로 실행
Validator	실행 결과가 완료 기준을 만족하는지 검증
Replanner	실패 원인을 보고 재시도할지 중단할지 판단

---

9. Plan

Plan은 Planner와 Executor가 공유하는 실행 계약이다.

class Plan(BaseModel):
    goal: str
    tasks: list[TaskName]
    success_criteria: list[str]

Task는 정해진 값만 사용할 수 있다.

TaskName = Literal[
    "requirements",
    "weather",
    "destination",
    "budget",
    "compose",
]

각 Task의 의미는 다음과 같다.

Task	역할
requirements	인원, 날씨, 아이 동반, 예산 조건 확인
weather	날씨를 실내외 장소 선택 정책으로 변환
destination	조건에 맞는 장소 후보 검색
budget	후보별 예상 비용 계산
compose	조사와 계산 결과로 일정 초안 작성

Plan에는 같은 Task가 중복되면 안 된다.

if len(tasks) != len(set(tasks)):
    raise ValueError("Plan에는 같은 Task를 중복할 수 없습니다.")

이렇게 하면 불필요한 반복 실행을 막을 수 있다.

---

10. Planner

Planner는 사용자의 목표를 보고 필요한 Task를 선택한다.

예를 들어 목표가 다음과 같다면:

아이와 비 오는 날 20만원 이내의 송파 하루 일정을 만들어줘.

필요한 Task는 다음처럼 구성될 수 있다.

requirements
-> weather
-> destination
-> budget
-> compose

반면 “예상 비용만 계산해줘” 같은 요청이라면 모든 Task가 필요하지 않을 수 있다.

즉, Planner는 목표를 작은 실행 단계로 분해하는 역할이다.

---

11. Executor

Executor는 Plan에 들어 있는 Task를 순서대로 실행한다.

for task in plan.tasks:
    TASK_HANDLERS[task](request, context)
    path.append(task)

TASK_HANDLERS는 Task 이름과 실제 실행 함수를 연결한다.

TASK_HANDLERS = {
    "requirements": execute_requirements,
    "weather": execute_weather,
    "destination": execute_destination,
    "budget": execute_budget,
    "compose": execute_compose,
}

Executor는 실행 중간 결과를 context에 저장한다.

requirements 결과
weather 결과
destinations 결과
budget_options 결과
draft 결과

---

12. Context

context는 Workflow 안에서 Task들이 공유하는 중간 결과 저장소이다.

예를 들어 destination Task는 장소 후보를 저장한다.

context["destinations"] = search_destinations(request)

budget Task는 앞에서 저장된 장소 후보를 읽어 비용을 계산한다.

places = context.get("destinations", [])
context["budget_options"] = calculate_budget_options(request, places)

즉, Task들은 독립적으로 실행되는 것이 아니라 이전 Task가 만든 context를 읽고 다음 결과를 만든다.

---

13. Validator

Validator는 실행 결과가 목표 조건을 만족하는지 확인한다.

검증 기준은 다음과 같다.

• 필요한 Task 결과가 context에 존재하는가
• 비 오는 날이면 실내 장소인가
• 아이 동반이면 아이 친화 장소인가
• 예상 비용이 예산을 넘지 않는가
• 일정 초안이 실제로 선택되었는가

if request.weather == "rain" and not draft.indoor:
    failures.append("비 오는 날에는 실내 장소가 필요합니다.")

검증 결과는 ValidationReport로 관리한다.

class ValidationReport(BaseModel):
    passed: bool
    failed_conditions: list[str]

성공하면 Workflow는 종료하고, 실패하면 Replanner로 넘어간다.

---

14. Replanner

Replanner는 검증 실패 원인을 보고 다시 실행할지 중단할지 판단한다.

class ReplanDecision(BaseModel):
    action: Literal["retry", "stop"]
    reason: str
    plan: Plan | None = None

중요한 규칙은 retry를 선택했다면 수정된 Plan이 반드시 있어야 한다는 점이다.

if self.action == "retry" and self.plan is None:
    raise ValueError("retry에는 수정 Plan이 필요합니다.")

실패가 누락된 Task 때문에 발생했다면 Plan을 수정해 다시 실행할 수 있다.
하지만 예산이 너무 낮아서 어떤 장소도 조건을 만족하지 못한다면 Plan을 바꿔도 해결되지 않을 수 있다. 이 경우 stop이 더 적절하다.

---

15. 전체 Workflow 흐름

전체 Graph 흐름은 다음과 같다.

START
-> planner
-> executor
-> validator
-> 검증 성공이면 END
-> 검증 실패이고 재시도 가능하면 replanner
-> retry면 executor
-> stop이면 END

이 구조의 장점은 문제 발생 지점을 분리해서 확인할 수 있다는 점이다.

Plan이 이상한가?
Task 실행 결과가 빠졌는가?
Draft가 조건을 만족하지 못했는가?
재계획으로 해결 가능한 문제인가?

---

16. Agent Safety와 Workflow의 차이

Agent Safety는 Tool 실행 중 실패를 안전하게 제어하는 데 집중한다.

Tool 실패
-> 재시도 가능 여부 판단
-> Fallback 또는 Handoff

Plan-Execute-Replan Workflow는 복잡한 목표를 단계별 작업으로 나누고, 결과를 검증한 뒤 필요하면 계획을 수정하는 데 집중한다.

목표
-> 계획
-> 실행
-> 검증
-> 재계획

정리하면 다음과 같다.

구분	목적
Agent Safety	Tool 실행 실패를 안전하게 제어
Workflow	복잡한 작업을 계획, 실행, 검증, 재계획으로 처리

---

17. 핵심 정리

1. Agent Memory는 같은 thread_id의 대화 메시지를 이어준다.
2. 다른 thread_id끼리는 기억이 섞이면 안 된다.
3. Agent가 이전 대화를 바탕으로 Tool 인자를 만들 수 있다.
4. 사용자가 정보를 수정하면 가장 최근 값을 기준으로 판단해야 한다.
5. Agent Safety는 실패 종류에 따라 재시도, Fallback, Handoff를 결정한다.
6. NOT_FOUND, TIMEOUT은 재시도 가능한 실패로 볼 수 있다.
7. AUTH_ERROR는 재시도하지 않고 Handoff하는 것이 적절하다.
8. 재시도 횟수는 max_attempts로 제한해야 한다.
9. Graph의 반복은 반드시 종료 조건을 가져야 한다.
10. recursion_limit는 무한 반복을 막는 최후의 안전장치이다.
11. Planner는 목표를 실행 가능한 Task 목록으로 분해한다.
12. Executor는 Plan의 Task를 순서대로 실행한다.
13. Context는 Task 사이의 중간 결과를 저장한다.
14. Validator는 실행 결과가 조건을 만족하는지 검증한다.
15. Replanner는 실패 원인을 보고 retry 또는 stop을 결정한다.
16. Pydantic 모델을 사용하면 Planner, Executor, Validator 사이의 데이터 계약을 명확히 만들 수 있다.
17. 복잡한 Agentic 작업은 “모델의 판단”과 “코드가 강제하는 규칙”을 분리해야 더 안전하게 운영할 수 있다.