21장. 전문 에이전트 구현

따라하기 1 — 처리 도구 만들기

수업 모델: 모든 장은 config.py의 MODEL을 사용합니다. .env의 GEMINI_MODEL=gemini-3.8-flash를 확인하세요.

출력 안내: 실행 코드는 답변·도구 결과·청크 본문을 글자 수로 자르지 않고 출력합니다. 아래의 기존 실행 예시는 일부 축약된 기록이며, 실제 실행 화면에서 전체 내용을 확인하세요.

목표

요청을 실제로 처리하는 도구를 담은 haru_actions.py 를 만듭니다. 그리고 에이전트를 붙이기 전에, LLM 없이 도구를 직접 불러 봅니다. 처리할 수 있는지 판정하고 기록을 남기는 것이 모델이 아니라 코드라는 것을 먼저 눈으로 확인합니다.

---

0. 실습 준비

이 장의 실습 고객 — C003 김도윤 고객(VIP 등급)으로 로그인한 상태라고 정해 두고 실습합니다.

VS Code에서 haru-market 폴더를 열고 터미널에서 환경을 켭니다.

$ conda activate myenv

이번 장은 새로 설치할 것이 없습니다. 앞 장에서 만든 두 가지가 폴더에 있어야 합니다.

chroma_db/ 폴더가 보이지 않으면 13장 파일(python lesson13_rag_embedding.py)을 먼저 실행합니다.

이 고객의 주문 가운데 이번 장에서 쓰는 것은 넷입니다.

---

1. 파일 만들기

haru-market 폴더 맨 위에 새 파일을 만듭니다. 이름에 lesson도 번호도 붙이지 않습니다. 8장의 haru_tools.py처럼 다른 파일이 불러 쓰는 재사용 모듈입니다.

haru_actions.py

아래 코드 전체를 복사해 붙여 넣고 저장합니다.

# -*- coding: utf-8 -*-
"""하루마켓 처리 도구 (21장 산출물) — 조회가 아니라 '처리'를 하는 도구.

8장의 haru_tools.py 는 읽기 도구가 중심이었다. 이 모듈은 고객의 요청을 실제로 처리한다.
처리는 위험도에 따라 둘로 나눈다.

  직접 처리 (되돌릴 수 있다)
    - 배송지 변경: 아직 출고되지 않은 주문만
    - 교환 접수:   배송 완료 후 기간 안이고 재고가 있을 때
  승인 후 처리 (되돌릴 수 없다 — 돈이 나간다)
    - 환불·주문 취소: 에이전트는 '승인 요청'까지만 만든다. 실행은 사람이 승인한 뒤에 한다.

가능한지 아닌지(출고 여부, 기간, 재고)는 모델이 아니라 코드가 판정한다.
처리 기록은 memory_store/actions.json 에 쌓인다. 원본 CSV 는 고치지 않는다.
"""
import json
from datetime import date, datetime
from typing import Literal, get_args

from config import ROOT
from haru_tools import ORDERS, PRODUCTS, CustomerSession

ACTIONS_PATH = ROOT / "memory_store" / "actions.json"

TODAY = date(2026, 7, 27)        # 실습 기준일 (주문 데이터가 2026년 7월까지 있다)
RETURN_DAYS = 7                  # 단순 변심 반품·교환: 배송 완료 후 7일 이내
DEFECT_DAYS = 30                 # 불량·오배송: 배송 완료 후 30일 이내 (정책 제2조)
BEFORE_SHIPPING = ("결제완료", "배송준비")   # 아직 출고되지 않은 상태


# ── 처리 기록 읽기·쓰기 ───────────────────────────────────────────────
def load_actions() -> list[dict]:
    if ACTIONS_PATH.exists():
        return json.loads(ACTIONS_PATH.read_text(encoding="utf-8"))
    return []


def _save(actions: list[dict]) -> None:
    ACTIONS_PATH.parent.mkdir(exist_ok=True)
    ACTIONS_PATH.write_text(json.dumps(actions, ensure_ascii=False, indent=2),
                            encoding="utf-8")


def _record(session: CustomerSession, kind: str, order_id: str,
            detail: str, status: str) -> dict:
    actions = load_actions()
    action = {
        "action_id": f"A{len(actions) + 1:05d}",
        "type": kind,                          # 배송지변경 / 교환접수 / 환불요청 / 주문취소요청
        "customer_id": session.customer_id,
        "order_id": order_id,
        "detail": detail,
        "status": status,                      # 완료 / 승인대기 / 승인완료 / 반려
        "created_at": datetime.now().strftime("%Y-%m-%d %H:%M"),
    }
    actions.append(action)
    _save(actions)
    return action


def _my_order(session: CustomerSession, order_id: str):
    """본인 주문이면 그 행을, 아니면 None 을 돌려준다 (8장의 본인 확인과 같은 원칙)."""
    row = ORDERS[ORDERS["order_id"] == order_id]
    if row.empty or row.iloc[0]["customer_id"] != session.customer_id:
        return None
    return row.iloc[0]


def _days_since_delivery(order) -> int | None:
    delivered = order["delivered_at"]
    if not isinstance(delivered, str) or not delivered:
        return None
    return (TODAY - date.fromisoformat(delivered)).days


ReasonCategory = Literal["단순변심", "상품하자", "오배송", "확인필요"]


def _validate_reason(reason_category: ReasonCategory) -> dict | None:
    if reason_category not in get_args(ReasonCategory):
        return {"ok": False, "reason": "유효한 사유 분류를 전달해 주세요."}
    return None


def _return_limit(reason_category: ReasonCategory) -> int | None:
    # 문장의 의미는 에이전트가 판단한다. 코드는 전달된 분류에 정책을 적용한다.
    if reason_category == "확인필요":
        return None
    return DEFECT_DAYS if reason_category in ("상품하자", "오배송") else RETURN_DAYS


NEEDS_REASON = {"ok": False, "needs_clarification": True,
                "reason": "반품·교환 사유를 확인해야 합니다. 어떤 이유인지 고객에게 질문해 주세요."}


# ── 처리 도구 (세션에 묶는다 — 고객 ID 는 모델이 넣지 않는다) ──────────
def make_action_tools(session: CustomerSession):
    not_found = {"ok": False, "reason": "해당 주문을 찾을 수 없습니다. 주문번호를 확인해 주세요."}

    def change_shipping_address(order_id: str, new_address: str) -> dict:
        """본인 주문의 배송지를 바꾼다. 아직 출고되지 않은 주문만 바꿀 수 있다.

        Args:
            order_id: 하루마켓 주문번호. 'HR'로 시작한다.
            new_address: 새 배송지 주소. 고객이 말한 그대로 넣는다.
        """
        order = _my_order(session, order_id)
        if order is None:
            return not_found
        if order["order_status"] not in BEFORE_SHIPPING:
            return {"ok": False, "order_id": order_id, "status": order["order_status"],
                    "reason": f"이미 '{order['order_status']}' 상태라 배송지를 바꿀 수 없습니다."}
        action = _record(session, "배송지변경", order_id,
                         f"{order['product_name']} 배송지를 '{new_address}'(으)로 변경", "완료")
        return {"ok": True, "action_id": action["action_id"], "order_id": order_id,
                "product_name": order["product_name"], "new_address": new_address,
                "message": "배송지를 변경했습니다."}

    def request_exchange(order_id: str, new_option: str, reason: str, reason_category: ReasonCategory) -> dict:
        """본인 주문의 교환을 접수한다. 배송 완료 후 기간 안이고 재고가 있을 때만 접수된다.

        Args:
            order_id: 하루마켓 주문번호. 'HR'로 시작한다.
            new_option: 바꿔 받을 옵션. 예: '블랙/270'
            reason_category: 고객의 설명을 에이전트가 분류한 값. 불명확하면 확인필요.
            reason: 교환 사유. 예: '사이즈가 작음', '불량'
        """
        order = _my_order(session, order_id)
        if order is None:
            return not_found
        invalid = _validate_reason(reason_category)
        if invalid:
            return invalid
        for a in load_actions():                       # 같은 주문의 교환이 이미 접수되어 있으면 알려 준다
            if a["order_id"] == order_id and a["type"] == "교환접수":
                return {"ok": True, "action_id": a["action_id"], "order_id": order_id,
                        "message": "이미 교환이 접수되어 있습니다."}
        days = _days_since_delivery(order)
        if order["order_status"] != "배송완료" or days is None:
            return {"ok": False, "order_id": order_id, "status": order["order_status"],
                    "reason": "교환은 상품을 받은 뒤에 접수할 수 있습니다."}
        limit = _return_limit(reason_category)
        if limit is None:
            return dict(NEEDS_REASON)
        if days > limit:
            return {"ok": False, "order_id": order_id, "days_since_delivery": days,
                    "reason": f"배송 완료 후 {days}일이 지나 교환 가능 기간({limit}일)을 넘었습니다."}
        stock = int(PRODUCTS[PRODUCTS["product_id"] == order["product_id"]].iloc[0]["stock"])
        if stock <= 0:
            return {"ok": False, "order_id": order_id, "reason": "상품 재고가 없어 교환할 수 없습니다."}
        action = _record(session, "교환접수", order_id,
                         f"{order['product_name']} {order['option']} → {new_option} ({reason} · 분류: {reason_category})",
                         "완료")
        return {"ok": True, "action_id": action["action_id"], "order_id": order_id,
                "product_name": order["product_name"], "from_option": order["option"],
                "to_option": new_option, "days_since_delivery": days,
                "reason_category": reason_category,
                "message": "교환을 접수했습니다. 회수 기사가 방문해 기존 상품을 가져간 뒤 새 상품을 보냅니다."}

    def request_refund(order_id: str, reason: str, reason_category: ReasonCategory) -> dict:
        """본인 주문의 환불(반품) 또는 주문 취소를 요청한다.
        돈이 나가는 일이라 바로 실행하지 않는다. '승인 요청'을 만들고, 담당자가 승인하면 실행된다.

        Args:
            order_id: 하루마켓 주문번호. 'HR'로 시작한다.
            reason_category: 고객의 설명을 에이전트가 분류한 값. 불명확하면 확인필요.
            reason: 환불·취소 사유. 예: '단순 변심', '불량'
        """
        order = _my_order(session, order_id)
        if order is None:
            return not_found
        invalid = _validate_reason(reason_category)
        if invalid:
            return invalid
        # 같은 주문의 요청이 이미 있으면 새로 만들지 않고 그 요청을 알려 준다.
        # (승인까지 끝난 주문에 요청이 또 등록되면 환불이 두 번 나가는 모양이 된다)
        notes = {"승인대기": "이미 승인을 기다리는 요청이 있습니다.",
                 "승인완료": "이미 승인되어 처리된 요청입니다."}
        for a in load_actions():
            if (a["order_id"] == order_id and a["status"] in notes
                    and a["type"] in ("환불요청", "주문취소요청")):
                return {"ok": True, "action_id": a["action_id"], "order_id": order_id,
                        "product_name": order["product_name"],
                        "order_amount": order["order_amount"],
                        "status": a["status"], "message": notes[a["status"]]}
        status = order["order_status"]
        if status in BEFORE_SHIPPING:
            kind, note = "주문취소요청", "출고 전 주문 취소"
        elif status == "배송완료":
            days = _days_since_delivery(order)
            limit = _return_limit(reason_category)
            if limit is None:
                return dict(NEEDS_REASON)
            if days is None or days > limit:
                return {"ok": False, "order_id": order_id, "days_since_delivery": days,
                        "reason": f"배송 완료 후 {days}일이 지나 반품 가능 기간({limit}일)을 넘었습니다."}
            kind, note = "환불요청", f"배송 완료 후 {days}일, 기간({limit}일) 이내"
        else:
            return {"ok": False, "order_id": order_id, "status": status,
                    "reason": f"'{status}' 상태에서는 취소할 수 없습니다. 상품을 받은 뒤 반품으로 진행합니다."}
        action = _record(session, kind, order_id,
                         f"{order['product_name']} {order['option']} · 결제 {order['order_amount']}원"
                         f" · 사유: {reason} · 분류: {reason_category} · {note}", "승인대기")
        return {"ok": True, "action_id": action["action_id"], "order_id": order_id,
                "product_name": order["product_name"], "order_amount": order["order_amount"],
                "status": "승인대기", "reason_category": reason_category,
                "message": "승인 요청을 등록했습니다. 담당자가 승인하면 처리됩니다."}

    def get_my_requests() -> list[dict]:
        """이 고객이 요청한 처리 내역(배송지 변경, 교환 접수, 환불·취소 요청)과 현재 상태를 조회한다."""
        mine = [a for a in load_actions() if a["customer_id"] == session.customer_id]
        return mine or [{"message": "처리 내역이 없습니다."}]

    return {
        "change_shipping_address": change_shipping_address,
        "request_exchange": request_exchange,
        "request_refund": request_refund,
        "get_my_requests": get_my_requests,
    }


# ── 승인 처리 (사람이 한다 — 23장의 승인 게이트와 24장의 상담원 화면이 부른다) ──
def pending_approvals() -> list[dict]:
    """승인을 기다리는 요청 목록."""
    return [a for a in load_actions() if a["status"] == "승인대기"]


def decide(action_id: str, approve: bool, by: str = "상담원") -> dict:
    """승인 대기 요청을 승인하거나 반려한다. 승인하면 그때 실행(환불·취소)된 것으로 기록한다."""
    actions = load_actions()
    for a in actions:
        if a["action_id"] == action_id and a["status"] == "승인대기":
            a["status"] = "승인완료" if approve else "반려"
            a["decided_by"] = by
            a["decided_at"] = datetime.now().strftime("%Y-%m-%d %H:%M")
            _save(actions)
            return a
    return {"ok": False, "reason": f"승인 대기 중인 요청 {action_id} 이(가) 없습니다."}

이 파일의 이름과 함수 이름을 바꾸지 마세요. 뒤의 장들이 from haru_actions import make_action_tools처럼 이 이름 그대로 불러 씁니다.

---

2. 코드에서 볼 곳

파일은 세 덩어리입니다.

(1) 도구는 세션에 묶여 있다

def make_action_tools(session: CustomerSession):
    not_found = {"ok": False, "reason": "해당 주문을 찾을 수 없습니다. 주문번호를 확인해 주세요."}

    def change_shipping_address(order_id: str, new_address: str) -> dict:
        ...

8장의 make_tools(session)과 같은 모양입니다. 도구의 인자는 order_id와 new_address뿐이고 고객 ID는 없습니다. 누구의 주문인지는 모델이 넣는 값이 아니라 바깥의 session이 정합니다.

(2) 판정을 통과해야 기록이 남는다

네 도구 가운데 처리하는 셋은 모두 같은 순서로 움직입니다.

본인 주문인가  →  조건을 채웠는가  →  기록을 남기고 처리번호를 돌려준다
     아니면              아니면
  ok=False            ok=False + 이유

_record(...)는 맨 마지막에 한 번만 불립니다. 앞의 판정에서 걸리면 return으로 빠져나가므로, 기록 파일에 줄이 생겼다는 것은 판정을 통과했다는 뜻입니다.

(3) 도구는 예외로 끝나지 않는다

처리하지 못한 경우에도 도구는 오류를 내지 않고 {"ok": False, "reason": "…"}을 돌려줍니다. 8장의 원칙 그대로입니다. 모델은 이 값을 읽고 "왜 안 됐는지"를 보고에 옮길 수 있습니다.

---

3. 도구를 직접 불러 보기

haru_actions.py는 함수를 정의만 하므로 실행해도 아무것도 출력하지 않습니다. 작은 확인 파일을 하나 만들어 도구를 직접 불러 봅니다. 폴더 맨 위에 새 파일을 만듭니다.

check21_actions.py

아래 코드 전체를 복사해 붙여 넣고 저장합니다.

from haru_actions import ACTIONS_PATH, make_action_tools
from haru_tools import CustomerSession

act = make_action_tools(CustomerSession("C003"))      # C003 고객에게 묶인 처리 도구

print("1) 아직 출고되지 않은 주문 (콜드브루 원액, 결제완료)")
print("  ", act["change_shipping_address"]("HR20260726004", "서울 중구 세종대로 110"))

print("2) 이미 출고된 주문 (무선 블루투스 이어폰, 배송중)")
print("  ", act["change_shipping_address"]("HR20260725002", "서울 중구 세종대로 110"))

print("3) 처리 기록 파일 memory_store/actions.json")
print(ACTIONS_PATH.read_text(encoding="utf-8"))

ACTIONS_PATH.unlink()                                 # 연습 기록을 지운다
print("4) 연습 기록을 지웠습니다. 다음 처리번호는 A00001 부터 다시 시작합니다.")

이 파일에는 LLM 호출이 한 줄도 없습니다. 같은 도구 change_shipping_address를 주문만 바꿔 두 번 부릅니다.

실행하기 전에 짐작해 보세요. 두 번의 호출 가운데 기록 파일에 남는 것은 몇 건일까요?

$ python check21_actions.py

LLM을 부르지 않으므로 바로 끝납니다.

실행 결과 (created_at의 시각은 실행한 때로 나옵니다)

1) 아직 출고되지 않은 주문 (콜드브루 원액, 결제완료)
   {'ok': True, 'action_id': 'A00001', 'order_id': 'HR20260726004', 'product_name': '콜드브루 원액 500ml 2병', 'new_address': '서울 중구 세종대로 110', 'message': '배송지를 변경했습니다.'}
2) 이미 출고된 주문 (무선 블루투스 이어폰, 배송중)
   {'ok': False, 'order_id': 'HR20260725002', 'status': '배송중', 'reason': "이미 '배송중' 상태라 배송지를 바꿀 수 없습니다."}
3) 처리 기록 파일 memory_store/actions.json
[
  {
    "action_id": "A00001",
    "type": "배송지변경",
    "customer_id": "C003",
    "order_id": "HR20260726004",
    "detail": "콜드브루 원액 500ml 2병 배송지를 '서울 중구 세종대로 110'(으)로 변경",
    "status": "완료",
    "created_at": "2026-10-08 01:37"
  }
]
4) 연습 기록을 지웠습니다. 다음 처리번호는 A00001 부터 다시 시작합니다.

다른 것이 나오면 아래 표를 봅니다.

---

4. 무엇을 관찰했나

같은 도구가 주문에 따라 다르게 답했다

기록 파일에는 한 건만 있습니다. 2번 호출은 if order["order_status"] not in BEFORE_SHIPPING:에서 걸려 _record까지 가지 못했습니다. 이 판단에 모델은 끼어 있지 않습니다. 이 파일은 LLM을 부르지 않았습니다.

기록의 값은 어디서 왔나

에이전트를 붙이면 마지막 줄의 새 주소만 모델이 채웁니다. 나머지는 지금과 똑같이 코드가 넣습니다.

연습 기록은 지웠다

확인 파일의 마지막 두 줄이 기록 파일을 지웁니다. 지우지 않으면 다음 실습의 처리번호가 A00002부터 이어져, 이 교재에 실린 번호와 한 칸씩 어긋납니다. 지금 memory_store 폴더에는 actions.json이 없습니다.

처리 기록을 지우고 처음 상태로 돌아가고 싶을 때는 언제든 터미널에서 아래 한 줄을 실행하면 됩니다. 원본 데이터(data/)는 처리 도구가 고치지 않으므로 그대로입니다.

$ python -c "from haru_actions import ACTIONS_PATH; ACTIONS_PATH.unlink(missing_ok=True)"

---

핵심 정리

• haru_actions.py는 처리 도구를 담은 재사용 모듈입니다. 파일 이름과 함수 이름을 그대로 둡니다.
• 처리 도구는 본인 확인 → 조건 판정 → 기록 순서로 움직입니다. 판정을 통과해야 기록이 남습니다.
• 출고된 주문의 배송지 변경은 LLM 없이도 거절됐습니다. 판정은 코드가 합니다.
• 처리번호와 고객 ID는 코드와 세션이 넣습니다. 모델이 채우는 것은 새 주소 같은 요청 내용뿐입니다.
• 처리 기록은 memory_store/actions.json에 쌓이고, 이 파일을 지우면 처리번호가 A00001부터 다시 시작합니다.