20장. 복합 요청 분해 처리

따라하기 — 복합 문의 분해 처리

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

목표

한 문장에 세 가지 일이 든 문의를 계획 → 실행 → 취합으로 처리합니다. 모델이 세운 계획을 눈으로 보고, 각 Worker의 보고가 최종 답변에 어떻게 들어갔는지 확인합니다.

---

0. 실습 준비

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

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

$ conda activate myenv

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

실행했을 때 아래처럼 나오면 이렇게 합니다.

---

1. 파일 만들기

haru-market 폴더 맨 위에 새 파일을 만듭니다.

lesson20_planner_worker.py

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

# -*- coding: utf-8 -*-
"""[20장] Planner-Worker — 복합 문의 분해

문제 상황: "지난주에 산 운동화 반품하고, 같은 걸로 검정색 270 다시 주문하고 싶어요."
  한 문장에 [주문 찾기 + 반품 정책 확인 + 재고 확인] 세 작업이 들어 있다.
  19장 라우터는 '하나의 갈래'로만 보낸다 — 복합 문의는 못 푼다.

해결: Planner-Worker 패턴
  Planner: 문의를 작업 목록(JSON)으로 분해한다 — 구조화 출력(4장) 재사용
  Worker : 작업 유형별 전담 함수가 하나씩 실행한다
  취합   : 작업 결과들을 모아 하나의 답변으로 합성한다

실행:  python lesson20_planner_worker.py  (13장 선행 필요)

temperature 는 넣지 않는다. Gemini 3 계열은 기본값을 그대로 쓰라는 것이 공식 권장이다.
"""
import enum
import os

from google.genai import types
from langchain_chroma import Chroma
from langchain_google_genai import GoogleGenerativeAIEmbeddings
from pydantic import BaseModel, Field

from config import EMBEDDING_MODEL, MODEL, ROOT, get_client
from haru_tools import CustomerSession, make_tools

client = get_client()
session = CustomerSession("C003")   # 표준 질문셋의 고객 (쿠션 운동화 주문이 있다)
tools = make_tools(session)

# 13장에서 만든 벡터 저장소가 있는지 먼저 확인한다. 폴더가 없는데 Chroma(...) 를
# 부르면 빈 폴더를 새로 만들어 버려서, 오류 없이 검색 결과만 0건이 된다.
CHROMA_DIR = ROOT / "chroma_db"
NEED_13 = "13장 lesson13_rag_embedding.py 를 먼저 실행하세요."
if not CHROMA_DIR.exists():
    raise SystemExit(f"[준비 필요] chroma_db/ 폴더가 없습니다. {NEED_13}")

vectorstore = Chroma(
    collection_name="haru_policy",
    embedding_function=GoogleGenerativeAIEmbeddings(
        model=EMBEDDING_MODEL, google_api_key=os.getenv("GOOGLE_API_KEY")),
    persist_directory=str(CHROMA_DIR),
)
if not vectorstore.get(limit=1)["ids"]:
    raise SystemExit(f"[준비 필요] chroma_db/ 에 정책 문서가 없습니다. {NEED_13}")


# ── 1. Planner — 문의를 작업 목록으로 분해 ────────────────────────────
class TaskType(str, enum.Enum):
    ORDER_LOOKUP = "주문조회"
    POLICY_CHECK = "정책확인"
    STOCK_CHECK = "재고확인"
    TICKET = "티켓생성"


class PlanTask(BaseModel):
    task_type: TaskType
    instruction: str = Field(description="이 작업이 알아내야 할 것 (한 문장)")
    depends_on_previous: bool = Field(
        description="이전 작업 결과가 있어야 실행 가능한가")


class Plan(BaseModel):
    tasks: list[PlanTask] = Field(description="실행 순서대로 정렬된 작업 목록 (1~4개)")


def make_plan(question: str) -> Plan:
    r = client.models.generate_content(
        model=MODEL,
        contents=f"고객 문의를 처리하기 위한 작업 계획을 세워라.\n문의: {question}",
        config=types.GenerateContentConfig(
            system_instruction="하루마켓 상담 Planner. 문의를 최소 작업으로 분해한다. "
                               "불필요한 작업을 만들지 않는다.",
            response_mime_type="application/json",
            response_schema=Plan,
        ),
    )
    return r.parsed


# ── 2. Workers — 작업 유형별 전담 실행 함수 ───────────────────────────
def worker_order(instruction: str, context: str) -> str:
    r = client.models.generate_content(
        model=MODEL,
        contents=f"{context}\n\n작업: {instruction}",
        config=types.GenerateContentConfig(
            system_instruction="주문 조회 담당. 도구로 조회한 사실만 간결히 보고. "
                               "고객 등급·포인트는 get_my_membership 으로 확인.",
            tools=[tools["get_my_orders"], tools["get_order_status"],
                   tools["get_my_membership"]],
            automatic_function_calling=types.AutomaticFunctionCallingConfig(
                maximum_remote_calls=4)))
    return r.text.strip()


def worker_policy(instruction: str, context: str) -> str:
    docs = vectorstore.similarity_search(instruction, k=3)
    ctx = "\n\n".join(d.page_content for d in docs)
    r = client.models.generate_content(
        model=MODEL,
        contents=f"[정책 발췌]\n{ctx}\n\n{context}\n\n작업: {instruction}",
        config=types.GenerateContentConfig(
            system_instruction="정책 확인 담당. 발췌 내용으로만, 조항 번호와 함께 보고."))
    return r.text.strip()


def worker_stock(instruction: str, context: str) -> str:
    r = client.models.generate_content(
        model=MODEL,
        contents=f"{context}\n\n작업: {instruction}",
        config=types.GenerateContentConfig(
            system_instruction="재고 확인 담당. 반드시 search_products 로 상품을 찾고 "
                               "check_stock 을 호출한 결과로만 보고한다.\n"
                               "재고 데이터는 상품 단위다. 색상·사이즈 등 옵션별 재고는 "
                               "시스템이 관리하지 않으므로, '상품 전체 재고 N개, 요청하신 "
                               "옵션의 재고는 별도 확인 필요' 형식으로 보고한다. "
                               "조회 없이 있다/없다를 단정하지 않는다.",
            tools=[tools["search_products"], tools["check_stock"]],
            automatic_function_calling=types.AutomaticFunctionCallingConfig(
                maximum_remote_calls=4)))
    return r.text.strip()


def worker_ticket(instruction: str, context: str) -> str:
    t = tools["create_ticket"](category="기타", summary=instruction[:80])
    return f"티켓 {t['ticket_id']} 접수 완료"


WORKERS = {
    TaskType.ORDER_LOOKUP: worker_order,
    TaskType.POLICY_CHECK: worker_policy,
    TaskType.STOCK_CHECK: worker_stock,
    TaskType.TICKET: worker_ticket,
}


# ── 3. 실행 + 취합 ───────────────────────────────────────────────────
def plan_and_execute(question: str) -> str:
    plan = make_plan(question)
    print("  [Plan]")
    for i, t in enumerate(plan.tasks, 1):
        dep = " (이전 결과 필요)" if t.depends_on_previous else ""
        print(f"    {i}. [{t.task_type.value}] {t.instruction}{dep}")

    # 순차 실행 — 이전 작업 결과를 다음 작업의 context 로 넘긴다.
    # (의존성이 없는 작업들은 병렬 실행도 가능. 수업에서는 순차로 단순화)
    results, context = [], f"[고객 문의] {question}"
    for i, task in enumerate(plan.tasks, 1):
        out = WORKERS[task.task_type](task.instruction, context)
        print(f"  [Worker {i} 완료] {out}")
        results.append(f"({task.task_type.value}) {out}")
        context += f"\n[작업{i} 결과] {out}"

    # 취합: Worker 들의 보고를 고객 답변으로 합성
    r = client.models.generate_content(
        model=MODEL,
        contents=f"고객 문의: {question}\n\n작업 결과:\n" + "\n".join(results)
                 + "\n\n위 결과를 종합해 고객에게 하나의 답변을 작성하라.",
        config=types.GenerateContentConfig(
            system_instruction="하루마켓 상담원 '하루'. 존댓말, 5문장 이내.\n"
                               "- 작업 결과에 없는 사실을 만들지 않는다. "
                               "'확인 불가'로 보고된 것은 없음으로 바꾸지 말고 "
                               "확인이 필요하다고 그대로 안내한다.\n"
                               "- 확인된 주문(상품명·옵션)과 작업 결과의 숫자(금액·"
                               "재고 수량)는 빠뜨리지 않고 그대로 옮긴다.\n"
                               "- 접수·신청을 고객에게 떠넘기지 않는다. "
                               "원하시면 채팅에서 바로 접수를 도와드리겠다고 안내한다."))
    return r.text


if __name__ == "__main__":
    q = ("지난주에 산 운동화가 작아서 반품하고 싶은데 배송비 얼마인지, "
         "그리고 검정 270 재고 있으면 그걸로 교환 가능한지 알려주세요.")
    print(f"고객: {q}\n")
    print(f"\n하루: {plan_and_execute(q)}")

    print("""
──────────────────────────────────────────────────────────
Planner-Worker 정리
  - Planner 출력은 '자유 문장'이 아니라 스키마가 강제된 JSON (4장 기술)
  - Worker 는 자기 작업에 필요한 도구만 가진다 (19장 원칙)
  - depends_on_previous=False 인 작업들은 병렬 실행으로 시간 단축 가능
남은 문제: Worker 가 아직 '함수' 수준이다. 각자 판단하는 '에이전트'로
  승격시키면? → 21장 전문 에이전트 3종
──────────────────────────────────────────────────────────""")

---

2. 코드에서 볼 곳

파일은 세 덩어리입니다.

전체 흐름은 plan_and_execute 안의 세 줄로 줄일 수 있습니다.

plan = make_plan(question)                                    # 계획 (LLM 1번)
for i, task in enumerate(plan.tasks, 1):
    out = WORKERS[task.task_type](task.instruction, context)  # 실행 (작업 수만큼)
r = client.models.generate_content(...)                       # 취합 (LLM 1번)

로그인한 고객은 C003(김도윤) 입니다. 표준 질문셋이 전제하는 그 고객이고, 최근 주문에 쿠션 운동화(화이트/250)가 있습니다.

---

3. 실행하기

실행하기 전에 짐작해 보세요.

• 아래 문의를 여러분이 직접 쪼갠다면 작업은 몇 개이고, 어떤 순서일까요?
• 상품 데이터에는 쿠션 운동화의 재고가 70 하나로만 적혀 있습니다. 최종 답변은 "검정 270 재고"를 뭐라고 안내할까요?

$ python lesson20_planner_worker.py

LLM을 여덟 번쯤 호출하므로 10~15초 걸립니다.

기존 실행 예시 (아래 기록은 Worker 보고가 축약된 이전 실행입니다. 현재 코드는 보고 전체를 출력하므로 실제 화면에서 내용을 끝까지 확인합니다.)

고객: 지난주에 산 운동화가 작아서 반품하고 싶은데 배송비 얼마인지, 그리고 검정 270 재고 있으면 그걸로 교환 가능한지 알려주세요.

  [Plan]
    1. [주문조회] 고객의 이전 구매 내역과 주문 상태를 확인한다.
    2. [정책확인] 반품 및 교환 시 발생하는 배송비 규정을 확인한다.
    3. [재고확인] 해당 운동화 검정색 270 사이즈의 재고를 확인한다.
  [Worker 1 완료] - 주문번호: HR20260721001
- 상품명: 쿠션 운동화 (화이트/250)
- 주문일시: 2026-0...
  [Worker 2 완료] 발췌된 정책 내용에 따른 반품 및 교환 시 배송비 규정은 다음과 같습니다.

* **제4조 (교환) ②** ...
  [Worker 3 완료] 쿠션 운동화(상품ID: P006) 상품 전체 재고는 70개입니다. (색상·사이즈 등 옵션별 재고는 시스템에서...

하루: 고객님, 쿠션 운동화(화이트/250) 주문 건에 대해 문의해 주셨습니다. 단순 변심 반품 시 편도 배송비 3,000원이 환불 금액에서 차감되며, 교환의 왕복 배송비는 6,000원입니다. 말씀하신 검정 270 사이즈의 재고는 시스템에서 옵션별로 별도 관리되지 않아 정확한 재고 확인이 필요합니다. 원하시면 채팅에서 바로 반품 및 교환 접수를 도와드리겠습니다.
(이하 생략)

---

4. 무엇을 관찰했나

계획 — 사람이 쪼갠 것과 같았다

[Plan]에 작업 셋이 주문조회 → 정책확인 → 재고확인 순서로 나왔습니다. 앞 절에서 사람이 쪼갠 세 가지와 같습니다. 이 교재를 준비하며 다섯 번 실행했을 때 작업의 종류와 순서는 다섯 번 모두 같았고, 지시문의 문장만 조금씩 달랐습니다.

눈여겨볼 것은 이 계획이 Worker가 하나도 실행되기 전에 찍혔다는 점입니다. 무엇을 할지 먼저 보고, 그다음에 실행했습니다.

Worker 1 — 주문을 찾았다

Worker 1은 고객에게 주문번호를 되묻지 않고 get_my_orders로 목록을 조회해 HR20260721001 쿠션 운동화(화이트/250) 를 찾았습니다. "지난주에 산 운동화"가 어떤 주문인지가 여기서 정해졌고, 이 보고가 context에 붙어 뒤의 Worker에게 넘어갔습니다.

Worker 3 — 상품을 알고 재고를 봤다

계획에 적힌 Worker 3의 지시문은 "해당 운동화 검정색 270 사이즈의 재고를 확인한다"입니다. 어떤 운동화인지는 지시문에 없습니다. 그런데 보고는 "쿠션 운동화(상품ID: P006) 상품 전체 재고는 70개입니다"로 시작합니다. Worker 1의 보고가 context로 넘어왔기 때문에 상품을 알고 조회한 것입니다. 앞의 보고를 뒤에 이어 주는 코드 한 줄이 한 일입니다.

보고의 모양도 봅니다. "상품 전체 재고는 70개"라고 하고, 곧바로 옵션별 재고는 따로 관리되지 않는다는 단서를 붙였습니다. 프롬프트에 적어 둔 보고 형식 그대로입니다.

취합 — 세 가지가 모두 답에 들어갔다

최종 답변에서 세 가지를 확인합니다.

19장의 라우터에서는 정책 노드 하나가 배송비를 답했습니다. 이번에는 세 곳을 모두 보고 세 가지를 모두 답했습니다. 다섯 번의 실행 모두 답변에 주문(쿠션 운동화 화이트/250), 배송비 금액, 재고 안내가 들어 있었습니다.

재고 안내는 "있다"도 "없다"도 아닌 "확인이 필요하다" 입니다. Worker 3의 보고와 취합 프롬프트의 "'확인 불가'로 보고된 것은 없음으로 바꾸지 말고"가 함께 만든 문장입니다.

마지막 문장 "원하시면 채팅에서 바로 반품 및 교환 접수를 도와드리겠습니다"는 모델이 생각해 낸 말이 아니라 취합 프롬프트에 그렇게 안내하라고 적어 둔 문장입니다.

호출 횟수 — 문의 하나에 여덟 번쯤

계획 1번, Worker 3개, 취합 1번입니다. 도구를 쓰는 Worker는 "도구 요청 → 결과를 받아 보고 작성"으로 호출이 두세 번씩 들어가므로, 모두 합치면 여덟 번 안팎입니다. 19장의 라우터는 같은 문의에 두 번이었습니다.

복합 문의를 빠짐없이 처리하는 대가는 호출 횟수입니다. 그래서 문의의 성격에 맞는 구조를 고릅니다. 「실습문제」에서 단순한 문의의 계획이 어떻게 나오는지 봅니다.

---

5. 지금 폴더의 모습

haru-market/
├── config.py
├── haru_tools.py                    (8장)
├── …                                (2~18장의 파일)
├── lesson19_router.py
├── lesson20_planner_worker.py       ← 이번 장
├── chroma_db/                       (13장이 만든 정책 문서 검색용 폴더)
└── memory_store/

---

핵심 정리

• 모델이 세운 계획은 주문조회 → 정책확인 → 재고확인 세 작업이었고, 실행 전에 먼저 출력되었습니다.
• 주문·배송비·재고 세 가지가 모두 답변에 들어갔습니다.
• Worker 3은 Worker 1이 찾은 상품을 context로 넘겨받아 재고를 조회했습니다.
• 재고는 "있다"도 "없다"도 아닌 "확인이 필요하다" 로 안내되었습니다. 프롬프트에 적어 둔 경계가 지켜졌습니다.
• 호출은 여덟 번 안팎입니다. 빠짐없는 처리의 대가는 호출 횟수입니다.