따라하기 — 팀장이 작업 목록을 만들게 하기

한 줄 요약

플래너는 도구가 없습니다. 계획만 세웁니다. 그런데 계획이 정해진 모양으로 나와야 프로그램이 쓸 수 있죠. 구조화 출력(structured output) 으로 모양을 강제합니다.

1. 문제 — 계획을 어떻게 받나

플래너에게 이렇게 물었다고 합시다.

"승승 스마트워치 Fit 5, 지금 어떻게 대응해야 할까요?"
누구에게 무엇을 물을지 계획을 세워 주세요.

모델이 이렇게 답할 수 있습니다.

먼저 데이터분석가에게 재고와 판매 추이를 물어보고,
규정담당에게는 제품 사양을 확인하는 게 좋겠습니다.
시장 상황도 봐야 하니 시장조사원에게도...

사람이 읽기엔 좋은데, 프로그램이 쓸 수가 없습니다. 여기서 "데이터분석가"와 "재고와 판매 추이"를 뽑아내려면 문자열을 파싱해야 하는데, 답이 매번 다르게 나오니 안정적으로 안 됩니다.

정해진 모양으로 답하게 강제해야 합니다.

2. 계획의 모양을 정의합니다

아래는 code/ch08_multi_agent.py 의 일부입니다 — 위에서 이미 붙여 넣었으니 다시 넣지 마세요.

class Task(BaseModel):
    """워커 한 명에게 맡길 일 하나."""
    worker: Literal["주문담당", "데이터분석가", "규정담당", "시장조사원"] = Field(description="맡길 워커")
    task: str = Field(description="그 워커에게 그대로 전달할 질문 한 문장")


class Plan(BaseModel):
    tasks: list[Task] = Field(description="필요한 작업만 담는다. 필요 없는 워커는 넣지 않는다.")

파일 맨 위의 from typing import Annotated, Literal, TypedDict 와 from pydantic import BaseModel, Field 가 이 두 클래스를 위한 것입니다.

이 클래스를 읽는 법

7장 2절에서 class State(TypedDict): 를 읽는 법을 봤죠. 여기 나오는 BaseModel 도 같은 종류입니다 — 클래스가 아니라 모양 정의입니다. 상속도, 메서드도, self 도 안 나옵니다.

BaseModel 은 pydantic(파이댄틱)이라는 라이브러리의 것으로, "이런 모양의 데이터" 를 정의할 때 씁니다. 이 과정에서 pydantic이 나오는 곳은 여기 한 군데뿐입니다.

읽어 보면 이렇습니다.

계획 = 작업들의 목록
작업 = { worker: 네 이름 중 하나, task: 문자열 }

정의한 모양 ↔ 모델이 채워 준 값

클래스로 적어 두면 모델이 실제로 무엇을 출력하는가 — 좌우로 놓고 보면 분명해집니다.

왼쪽을 적어 두는 것이 우리 일이고, 오른쪽을 채우는 것이 모델의 일입니다.

Literal 이 중요합니다

아래는 개념 설명용 코드입니다 — 실습 파일에 넣지 않습니다.

worker: Literal["주문담당", "데이터분석가", "규정담당", "시장조사원"]

이 넷 중 하나만 올 수 있다는 뜻입니다. 모델이 "마케팅담당"이라고 답할 수 없습니다.

왜 중요한가 — 우리 코드가 이렇게 되어 있으니까요.

아래는 개념 설명용 코드입니다 — 실습 파일에 넣지 않습니다.

agent = WORKERS[p["worker"]]        # 딕셔너리에 없는 이름이면 오류

Literal 이 애초에 잘못된 이름이 나오지 않게 막아 줍니다.

Field(description=...) 은 설명입니다

아래는 code/ch08_multi_agent.py 의 일부입니다 — 위에서 이미 붙여 넣었으니 다시 넣지 마세요.

task: str = Field(description="그 워커에게 그대로 전달할 질문 한 문장")

이 설명도 모델에게 전달됩니다. 3장에서 배운 독스트링과 같은 역할이죠. "한 문장으로 쓰라"는 지시가 여기 들어 있습니다.

3. 모양을 강제하기

아래는 code/ch08_multi_agent.py 의 일부입니다 — 위에서 이미 붙여 넣었으니 다시 넣지 마세요.

planner = llm.with_structured_output(Plan)

with_structured_output(모양) 한 줄입니다.

이렇게 만든 플래너는 반드시 그 모양으로만 답합니다.

아래는 개념 설명용 코드입니다 — 실습 파일에 넣지 않습니다.

result = planner.invoke([...])
print(type(result))          # <class '계획'>
print(result.tasks)          # [작업(worker='데이터분석가', task='...'), ...]

문자열이 아니라 객체가 돌아옵니다. 파싱할 필요가 없죠.

아래는 code/ch08_multi_agent.py 의 일부입니다 — 위에서 이미 붙여 넣었으니 다시 넣지 마세요.

plan = [{"worker": t.worker, "task": t.task} for t in result.tasks]

딕셔너리 목록으로 바꿔 상태에 넣습니다.

어떻게 강제되나 — 내부적으로는 3장에서 본 Function Calling과 비슷한 방식을 씁니다. 모델에게 "이 모양의 함수를 불러라" 라고 시키는 것이죠. 스키마 자체가 모델에게 전달되므로 어긋날 확률이 크게 낮아집니다. 우리가 문자열을 파싱하는 것보다 훨씬 안정적입니다.

다만 자동 재시도는 없습니다. 그래도 실패하면 예외가 그대로 올라옵니다. 9절에서 그 대비를 다룹니다.

4. 플래너의 시스템 지시

아래는 code/ch08_multi_agent.py 의 일부입니다 — 위에서 이미 붙여 넣었으니 다시 넣지 마세요.

def planner_prompt() -> str:
    """팀장에게 줄 시스템 지시. 역할설명이 바뀌면 여기도 자동으로 따라온다."""
    return (
    "너는 승승장구몰 대응팀의 팀장이다. 질문에 답하는 데 필요한 작업만 골라 계획을 세운다.\n"
    "팀원:\n"
    + "\n".join(f"  - {k}: {v}" for k, v in ROLE_DESC.items()) + "\n"
    "규칙:\n"
    "  - 필요 없는 팀원은 부르지 않는다. 한 명이면 충분하면 한 명만 부른다.\n"
    "  - 각 task 는 그 팀원이 앞뒤 맥락 없이 읽어도 알아들을 수 있게 완결된 문장으로 쓴다.\n"
    "  - 한 task 에는 한 가지 주제만 담는다. 물어볼 것이 둘이면 task 를 둘로 나눈다.\n"
    "  - 손님이 '그거'처럼 앞말을 가리키면 대화 기록에서 찾아 실제 이름으로 바꿔 쓴다.\n"
    "  - 너는 답을 만들지 않는다. 계획만 세운다.")

왜 상수가 아니라 함수인가 — ROLE_DESC 에 담당을 추가하면 부를 때마다 새로 만들어지므로 팀원 목록이 자동으로 갱신됩니다. 상수(PLANNER_SYSTEM = ...)로 두면 파일이 처음 읽힐 때 딱 한 번 만들어져서, 나중에 담당을 추가해도 이미 만들어진 문자열은 안 바뀝니다. 담당을 늘릴 일이 있다면 함수 쪽이 안전합니다. 10절 5번에서 다시 나옵니다.

규칙 다섯 개를 하나씩 보겠습니다.

① "필요 없는 팀원은 부르지 않는다"

호출 수를 줄이는 규칙입니다. 이게 없으면 "재고 몇 개?" 같은 질문에도 담당 넷을 다 부릅니다.

실제로 두 번째 질문에서 이 규칙이 적용됩니다.

"그 상품을 산 주문 O001203은 지금 어떤 상태인가요?
 그리고 재주문 기준을 채우려면 몇 개를 발주해야 하나요?"
  [계획]
    → 주문담당: ...                  ← 넷 중 둘만
    → 데이터분석가: ...

규정담당·시장조사원은 물어볼 것이 없으니 빠졌습니다.

② "앞뒤 맥락 없이 읽어도 알아들을 수 있게"

워커에는 기억이 없기 때문입니다. 3절에서 본 그 이야기죠.

✗ "그거 재고 알려주세요"                       ← 워커가 못 알아듣는다
✓ "승승 스마트워치 Fit 5(P0003)의 재고 수량을 알려주세요"

③ "한 task 에는 한 가지 주제만"

앞 절에서 본 그 규칙입니다. 두 주제를 한 번에 물으면 RAG 검색이 흐려집니다.

④ "'그거'처럼 앞말을 가리키면 실제 이름으로 바꿔 쓴다"

②와 짝입니다. 손님이 "그 상품"이라고 하면, 플래너가 대화 기록에서 찾아 실제 이름으로 바꿔야 합니다.

여기가 멀티턴이 동작하는 지점입니다.

아래는 code/ch08_multi_agent.py 의 일부입니다 — 위에서 이미 붙여 넣었으니 다시 넣지 마세요.

result = planner.invoke([{"role": "system", "content": planner_prompt()}]
                     + state["messages"])
                       └─ 대화 기록 전체

플래너만 대화 기록을 봅니다. 워커는 안 봅니다.

⑤ "너는 답을 만들지 않는다"

역할을 못 박는 것입니다. 플래너가 "제가 아는 바로는..." 하며 답을 만들어 버리면 곤란하죠.

5. 팀원 목록이 자동으로 들어갑니다

아래는 code/ch08_multi_agent.py 의 일부입니다 — 위에서 이미 붙여 넣었으니 다시 넣지 마세요.

+ "\n".join(f"  - {k}: {v}" for k, v in ROLE_DESC.items()) + "\n"

ROLE_DESC 딕셔너리를 문자열로 펼쳐 시스템 지시에 넣습니다.

팀원:
  - 주문담당: 주문번호 하나의 상태를 조회한다 — 고객·상품·금액·주문일·배송상태. 'O001203' 같은 주문번호가 질문에 있을 때만 부른다. 집계나 통계는 못 한다.
  - 데이터분석가: 사내 CSV 를 계산한다 — 매출·판매 추이, 재고 수량과 재주문 기준(reorder_level), ...
  - 규정담당: 사내 문서를 검색한다 — 환불·교환 기간, 멤버십 등급, 제품 사양(배터리·방수 등). 재고나 매출 같은 수치는 여기 없다.
  - 시장조사원: 인터넷을 검색한다 — 경쟁 제품, 시장 동향, 바깥 사용자 반응.

담당 넷이 그대로 들어갑니다.

ROLE_DESC 만 고치면 시스템 지시가 따라 바뀝니다. 두 곳을 따로 고칠 필요가 없죠.

2절에서 본 사고를 고칠 때도 ROLE_DESC 딕셔너리 한 곳만 고쳤습니다.

6. 계획 노드

아래는 code/ch08_multi_agent.py 의 일부입니다 — 위에서 이미 붙여 넣었으니 다시 넣지 마세요.

def plan_node(state: State) -> dict:
    """질문을 읽고 누구에게 무엇을 물을지 정한다."""
    result = planner.invoke([{"role": "system", "content": planner_prompt()}] + state["messages"])
    plan = [{"worker": t.worker, "task": t.task} for t in result.tasks]
    print("\n  [계획]")
    for p in plan:
        print(f"    → {p['worker']}: {p['task']}")
    return {"plan": plan, "results": []}

7장에서 배운 노드 그대로입니다 — 상태를 받아, 일을 하고, 바뀐 부분을 돌려줍니다.

results: [] 를 함께 돌려주는 이유

아래는 code/ch08_multi_agent.py 의 일부입니다 — 위에서 이미 붙여 넣었으니 다시 넣지 마세요.

return {"plan": plan, "results": []}       # results 는 이번 턴 것만 새로 담는다

State 정의를 다시 보세요.

아래는 개념 설명용 코드입니다 — 실습 파일에 넣지 않습니다.

class State(TypedDict):
    messages: Annotated[list, add_messages]   # 이어 붙임
    plan: list                                # 덮어씀
    results: list                             # 덮어씀

plan 과 results 에는 리듀서가 없어서 덮어씁니다. 7장 2절에서 본 그것이죠.

그런데 사실 없어도 동작합니다. 바로 다음 노드인 execute_node 가 return {"results": results} 로 항상 통째로 덮어쓰기 때문입니다.

그럼에도 적어 두는 이유 — "이 값은 턴마다 새로 담는 것"이라는 의도를 코드에 남기기 위해서입니다. 나중에 results 에 리듀서(operator.add 같은 것)를 붙여 누적하도록 바꾸면, 그때는 이 한 줄이 없으면 진짜로 섞입니다. 미리 대비해 둔 셈입니다.

세 필드를 누가 읽고 누가 쓰나

노드 셋을 가로로 놓고, 필드 셋을 가로줄로 그리면 한 장에 들어옵니다.

점선(┄)으로 지나가는 칸은 "그 노드는 이 필드를 건드리지 않는다"는 뜻입니다. 실선 화살표(─▶)는 값이 전달되는 방향입니다.

읽어 보면 이렇습니다.

7. 실행 결과

  [계획]
    → 데이터분석가: 승승 스마트워치 Fit 5(P0003)의 최근 판매 추이, 재고 수량,
                    리뷰 평점과 부정 리뷰 내용을 분석해주세요.
    → 규정담당: 승승 스마트워치 Fit 5(P0003)의 제품 사양(배터리·방수 등)을 알려주세요.
    → 규정담당: 승승 스마트워치 Fit 5(P0003)의 환불·교환 기간 규정을 알려주세요.
    → 시장조사원: 승승 스마트워치 Fit 5(P0003)의 경쟁 제품 동향과 외부 사용자 반응을 조사해주세요.

확인할 것 세 가지.

1. 작업이 4개 — 규정담당을 두 번 부릅니다. 규칙 ③이 적용된 결과입니다.
2. 각 작업이 완결된 문장 — "승승 스마트워치 Fit 5(P0003)의..." 로 시작합니다. 규칙 ②죠.
3. 아직 아무것도 조사 안 했습니다 — 계획만 세웠습니다.

작업 수는 실행할 때마다 다릅니다. 위는 4개로 나뉜 한 번의 실행입니다. 같은 질문에 8개가 나오기도 합니다(8절 전체 데모가 그런 경우입니다). "한 task 에 한 주제" 규칙 때문에 플래너가 잘게 나누는 정도가 매번 조금씩 달라지기 때문이죠. 개수가 아니라 담당 배분이 맞으면 성공입니다.

스스로 해 보기

1. 플래너만 따로 불러 보세요.
   python result = planner.invoke([ {"role": "system", "content": planner_prompt()}, {"role": "user", "content": "P0003 재고가 몇 개인가요?"}]) for t in result.tasks: print(f" {t.worker}: {t.task}")
   몇 명을 부르나요?
2. 질문을 여러 개 넣어 계획이 어떻게 달라지는지 보세요.
   python for q in ["P0003 재고가 몇 개인가요?", "불량 반품 기간이 며칠인가요?", "요즘 스마트워치 시장은 어떤가요?", "P0003을 지금 어떻게 해야 하나요?"]:
3. planner_prompt() 안의 규칙을 하나씩 지우고 계획이 어떻게 나빠지는지 보세요.
4. Literal 을 str 로 바꾸고, 시스템 지시에서 팀원 목록을 지운 뒤 실행해 보세요. 어떤 이름이 나오나요?

4번이 Literal 의 역할을 확인하는 실험입니다. 없는 담당자 이름이 나오면 WORKERS[…] 에서 오류가 납니다.

핵심 정리

• 플래너는 도구가 없습니다. 계획만 세웁니다.
• 계획을 프로그램이 쓰려면 정해진 모양으로 나와야 합니다.
• pydantic BaseModel 로 모양을 정의하고, with_structured_output(모양) 으로 강제합니다.
• Literal 은 정해진 값만 허용합니다. 없는 담당자 이름이 나오는 사고를 방지합니다.
• Field(description=...) 도 모델에게 전달됩니다. 독스트링과 같은 역할입니다.
• 시스템 지시는 planner_prompt() 함수가 만듭니다. 규칙 다섯 — 필요한 만큼만 · 완결된 문장 · 한 주제씩 · 대명사 해석 · 답은 만들지 않기.
• 팀원 목록은 ROLE_DESC 에서 자동으로 들어갑니다. 한 곳만 고치면 됩니다. 시스템 지시를 상수가 아니라 함수로 둔 이유가 이것입니다.
• 플래너만 대화 기록을 봅니다. 여기가 멀티턴이 동작하는 지점입니다.
• plan 과 results 는 리듀서가 없어 덮어씁니다. 계획 단계에서 results 를 비우지 않으면 지난 턴 결과가 섞입니다.
• 다음 턴까지 남는 것은 messages 뿐입니다. plan·results 는 턴마다 새로 담깁니다.
• 작업 수는 실행할 때마다 다릅니다. 개수가 아니라 담당 배분이 맞으면 성공입니다.