따라하기 — 앞 장의 에이전트를 한 파일로 모으기

한 줄 요약

code/ch08_multi_agent.py 를 만듭니다. 3·4·5·6장에서 만든 도구를 import 로 가져와 담당자 넷을 조립합니다. 새로 만드는 도구는 하나도 없습니다.

---

1. 파일 만들기

먼저 전체 그림

220줄을 붙여 넣기 전에 완성된 프로그램이 어떤 구조인지 한 장으로 봅니다.

위가 그래프(우리가 이 파일에서 새로 짜는 부분), 아래가 앞 장에서 가져오는 부분입니다. 회사 업무로 옮길 때 갈아 끼우는 것은 아래 네 칸입니다.

붙여 넣기

code 폴더에 ch08_multi_agent.py 를 만들고 아래를 전부 붙여 넣습니다. 이 과정의 마지막 파일입니다.

새 파일을 만드세요 — agentic_ai/code/ch08_multi_agent.py
VS Code 왼쪽 탐색기에서 code 폴더를 오른쪽 클릭 → New File → 파일명 ch08_multi_agent.py 입력 → 아래를 전부 복사해 붙여 넣고 저장(Ctrl+S / Cmd+S).

# -*- coding: utf-8 -*-
"""8장 — 멀티에이전트 완성: Planner–Worker

실행:  python code/ch08_multi_agent.py

지금까지 네 종류의 에이전트를 만들었습니다.
  3장  주문 조회 에이전트  — 정해진 값 하나
  4장  웹검색 에이전트     — 바깥세상 정보
  5장  CSV 분석 에이전트   — 사내 숫자
  6장  RAG 문서 에이전트   — 사내 규정·매뉴얼
7장에서는 여기에 기억을 붙였습니다.

이제 하나의 팀으로 묶습니다. 아래 질문 하나에 여러 담당이 필요합니다.
  "승승 스마트워치 Fit 5, 지금 어떻게 대응해야 할까요?"
    → 재고와 판매 추이는 CSV 에,  제품 사양은 매뉴얼에,  시장 반응은 웹에 있다.

핵심 개념 — Planner–Worker
  플래너(planner) : 질문을 읽고 '누구에게 무엇을 물을지' 계획만 세운다. 도구는 없다.
  워커(worker)    : 자기 도구 하나만 잘 쓰는 전문가. 다른 일은 모른다.
  종합(synthesis) : 워커들이 가져온 근거를 모아 하나의 답으로 정리한다.

  왜 나누는가? 도구 열 개를 한 에이전트에 다 물리면 무엇을 쓸지 헷갈립니다.
  나눠 두면 각자는 판단이 쉬워지고, 우리는 어디서 틀렸는지 찾기 쉬워집니다.
"""
import sys
import pathlib

sys.path.append(str(pathlib.Path(__file__).resolve().parent))

from typing import Annotated, Literal, TypedDict

from langchain.agents import create_agent
from langchain_core.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from pydantic import BaseModel, Field

from common import get_chat
from ch03_order_agent import get_order_status  # 3장 — 주문 조회 함수
from ch04_web_agent import search_web          # 4장 — 웹검색 함수
from ch05_csv_agent import run_pandas          # 5장 — pandas 계산 도구
from ch06_rag_agent import policy_tool         # 6장 — 사내 문서 검색 도구

llm = get_chat(temperature=0)
# 평범한 함수를 LangChain 도구로 등록한다(@tool 데코레이터 없이도 이렇게 감쌀 수 있다)
order_tool = tool(get_order_status)
search_tool = tool(search_web)


# ══════════════════════════════════════════════════════════════
# 1. 워커 — 앞에서 만든 도구를 하나씩 나눠 가진 전문가 넷
# ══════════════════════════════════════════════════════════════
WORKERS = {
    "주문담당": create_agent(
        llm, tools=[order_tool],
        system_prompt="너는 승승장구몰의 CS 상담원이다. get_order_status 로 조회한 결과만 근거로 답한다. "
                      "주문번호가 없으면 조회할 수 없다고 보고한다. 추측하지 않는다."),
    "데이터분석가": create_agent(
        llm, tools=[run_pandas],
        system_prompt="너는 승승장구몰의 데이터 분석가다. 숫자는 반드시 run_pandas 로 계산한다. "
                      "매출 집계에서는 취소·환불을 제외한다. "
                      "리뷰 평점은 reviews 표의 rating 으로 계산한다(products 의 rating 은 카탈로그 표시값이다). "
                      "리뷰 반응을 물으면 review_text 를 몇 건 직접 읽어 어떤 불만인지 함께 보고한다. "
                      "사실과 숫자만 간결히 보고한다."),
    "규정담당": create_agent(
        llm, tools=[policy_tool],
        system_prompt="너는 승승장구몰의 규정 담당자다. policy_search 로 찾은 사내 문서 내용만 근거로 답한다. "
                      "문서에 없으면 '문서에 없음'이라고 보고한다. 추측하지 않는다."),
    "시장조사원": create_agent(
        llm, tools=[search_tool],
        system_prompt="너는 시장조사 담당자다. search_web 으로 확인한 내용만 간결히 보고한다. "
                      "검색이 실패하면 실패했다고 그대로 보고한다."),
}

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


# ══════════════════════════════════════════════════════════════
# 2. 계획의 모양을 미리 정해 둔다
# ══════════════════════════════════════════════════════════════
class Task(BaseModel):
    """워커 한 명에게 맡길 일 하나."""
    worker: Literal["주문담당", "데이터분석가", "규정담당", "시장조사원"] = Field(description="맡길 워커")
    task: str = Field(description="그 워커에게 그대로 전달할 질문 한 문장")


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


planner = llm.with_structured_output(Plan)      # 정해 둔 모양으로만 답하게 한다

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"
    "  - 너는 답을 만들지 않는다. 계획만 세운다.")

SYNTH_SYSTEM = (
    "너는 승승장구몰 대응팀의 팀장이다. 팀원들이 조사해 온 내용만 근거로 최종 답을 쓴다.\n"
    "형식: ① 확인된 사실을 근거와 함께 정리 → ② 그래서 무엇을 해야 하는지 권고 2~3개.\n"
    "조사되지 않은 내용은 지어내지 않는다. 한국어로 간결하게 쓴다."
)


# ══════════════════════════════════════════════════════════════
# 3. 그래프 — 계획 → 실행 → 종합
# ══════════════════════════════════════════════════════════════
class State(TypedDict):
    messages: Annotated[list, add_messages]   # 대화 기록 (7장과 같은 방식)
    plan: list                                # 이번 턴의 계획
    results: list                             # 워커들이 가져온 결과


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": []}       # results 는 이번 턴 것만 새로 담는다


def execute_node(state: State) -> dict:
    """계획에 적힌 워커를 차례로 부른다."""
    results = []
    for p in state["plan"]:
        agent = WORKERS[p["worker"]]
        out = agent.invoke({"messages": [{"role": "user", "content": p["task"]}]})
        report = out["messages"][-1].text.strip()
        print(f"\n  [{p['worker']} 보고]\n    {report[:160]}{'…' if len(report) > 160 else ''}")
        results.append({"worker": p["worker"], "task": p["task"], "report": report})
    return {"results": results}


def synthesize_node(state: State) -> dict:
    """보고를 모아 하나의 답으로 정리한다."""
    evidence = "\n\n".join(
        f"[{r['worker']}] {r['task']}\n{r['report']}" for r in state["results"])
    question = state["messages"][-1].text
    answer = llm.invoke([
        {"role": "system", "content": SYNTH_SYSTEM},
        {"role": "user", "content": f"[질문]\n{question}\n\n[팀원 보고]\n{evidence}"},
    ]).text.strip()
    return {"messages": [{"role": "assistant", "content": answer}]}


graph = StateGraph(State)
graph.add_node("plan", plan_node)
graph.add_node("execute", execute_node)
graph.add_node("synthesize", synthesize_node)
graph.add_edge(START, "plan")
graph.add_edge("plan", "execute")
graph.add_edge("execute", "synthesize")
graph.add_edge("synthesize", END)

# 7장에서 배운 기억을 그대로 붙인다 → 멀티에이전트 + 멀티턴
team = graph.compile(checkpointer=InMemorySaver())


def ask(question: str, thread_id: str = "대응팀-1") -> str:
    config = {"configurable": {"thread_id": thread_id}}
    result = team.invoke({"messages": [{"role": "user", "content": question}]}, config)
    return result["messages"][-1].text.strip()


if __name__ == "__main__":
    print("=" * 62)
    print("[1] 팀 구성 — 앞 장에서 만든 것을 그대로 가져왔습니다")
    for name, desc in ROLE_DESC.items():
        print(f"   {name:<8} {desc}")

    print("\n" + "=" * 62)
    print("[2] 여러 담당이 필요한 질문")
    q1 = "승승 스마트워치 Fit 5(P0003), 지금 어떻게 대응해야 할까요?"
    print(f"  질문: {q1}")
    answer1 = ask(q1)
    print(f"\n  [최종 답변]\n{answer1}")

    print("\n" + "=" * 62)
    print("[3] 이어서 묻기 — 계획이 달라지고, 앞 대화를 기억합니다")
    q2 = "그 상품을 산 주문 O001203은 지금 어떤 상태인가요? 그리고 재주문 기준을 채우려면 몇 개를 발주해야 하나요?"
    print(f"  질문: {q2}")
    answer2 = ask(q2)
    print(f"\n  [최종 답변]\n{answer2}")

    print("\n" + "=" * 62)
    print("관찰: 질문 하나에 여러 종류의 에이전트가 각자 다른 곳을 뒤졌습니다.")
    print("      우리가 짠 것은 '누가 무엇을 맡는가'뿐입니다. 나누는 것은 플래너가 정했습니다.")
    print("      두 질문의 계획이 다릅니다 — 부르는 담당도, 작업 수도 질문에 따라 달라집니다.")

처음 보는 두 줄이 있을 겁니다 — class Task(BaseModel): 과 class Plan(BaseModel):. 지금 이해하지 않아도 됩니다. 5절에서 한 줄씩 읽는 법을 다룹니다. 지금은 "플래너가 답할 형식을 미리 적어 둔 것" 정도로 넘어가세요.

---

2. 실행 — [1] 부분

터미널에서 실행 — 프로젝트 폴더 agentic_ai 에서, (agentic) 표시를 확인한 뒤.

python code/ch08_multi_agent.py

1분 안팎 걸립니다. 작업 수에 따라 호출이 10~20번 일어나니까요.

  기존 인덱스를 불러옵니다 (policy_index) — 임베딩을 다시 하지 않습니다
==============================================================
[1] 팀 구성 — 앞 장에서 만든 것을 그대로 가져왔습니다
   주문담당     주문번호 하나의 상태를 조회한다 — 고객·상품·금액·주문일·배송상태. ...
   데이터분석가   사내 CSV 를 계산한다 — 매출·판매 추이, 재고 수량과 재주문 기준(reorder_level), ...
   규정담당     사내 문서를 검색한다 — 환불·교환 기간, 멤버십 등급, 제품 사양(배터리·방수 등). ...
   시장조사원    인터넷을 검색한다 — 경쟁 제품, 시장 동향, 바깥 사용자 반응.

네 줄이 찍힙니다. ROLE_DESC 딕셔너리를 그대로 훑어 출력하기 때문이죠. 담당은 넷입니다.

팀은 넷이지만, 질문 하나에 넷이 다 불리는 것은 아닙니다. 대표 질문("지금 어떻게 대응해야 할까요?")에는 데이터분석가·규정담당·시장조사원 셋이 나섭니다. 주문담당은 주문번호가 있는 질문에서만 불립니다 — 8절의 이어 묻기에서 등장합니다.

맨 위 줄이 6장의 인덱스입니다. from ch06_rag_agent import policy_tool 을 하는 순간 build_index() 가 실행되기 때문이죠. 6장에서 저장해 뒀으므로 즉시 불러옵니다. 6장 5절에서 예고한 그대로입니다.

---

3. 새로 만든 도구가 없습니다

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

from ch03_order_agent import get_order_status  # 3장
from ch04_web_agent import search_web          # 4장
from ch05_csv_agent import run_pandas          # 5장
from ch06_rag_agent import policy_tool         # 6장

네 줄로 도구를 다 가져왔습니다. 이 파일에서 새로 만든 도구는 하나도 없습니다. 1장에서 "에이전트 네 종류를 만들어 마지막에 하나의 팀으로 묶는다"고 한 약속이 여기서 지켜집니다.

7장 5절에서 본 그 방식입니다 — if __name__ == "__main__": 덕분에 필요한 것만 조용히 가져올 수 있습니다.

평범한 함수 둘은 한 번 감쌉니다

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

order_tool  = tool(get_order_status)
search_tool = tool(search_web)

run_pandas(@tool 을 붙였죠)와 policy_tool(create_retriever_tool 로 만들었죠)은 이미 LangChain 도구입니다. get_order_status 와 search_web 만 평범한 파이썬 함수라 감싸 줍니다.

5장 3절에서 배운 것이죠 — @tool 데코레이터 없이도 tool(함수) 로 감쌀 수 있습니다.

---

4. 워커 넷 — 전부 create_agent

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

WORKERS = {
    "주문담당":     create_agent(llm, tools=[order_tool],  system_prompt="..."),
    "데이터분석가": create_agent(llm, tools=[run_pandas],  system_prompt="..."),
    "규정담당":     create_agent(llm, tools=[policy_tool], system_prompt="..."),
    "시장조사원":   create_agent(llm, tools=[search_tool], system_prompt="..."),
}

5장에서 만든 것과 똑같은 구조입니다. 도구를 하나씩만 준 것이 다를 뿐이죠.

주문담당과 데이터분석가는 왜 따로인가 — 둘 다 CSV를 보지만 하는 일이 다릅니다. 1장 표에서 나눠 둔 그대로죠. 주문담당은 주문번호 하나로 정해진 조회를 하고, 데이터분석가는 매번 다른 계산을 합니다.

4장 4절의 원칙 — 겹치지 않게, 경계선을 적어라 — 을 따라 역할 설명에 "'O001203' 같은 주문번호가 질문에 있을 때만 부른다. 집계나 통계는 못 한다" 를 적었습니다.

딕셔너리로 묶은 이유는 4장의 TOOLS 와 같습니다 — 플래너가 이름 문자열로 지목하기 때문입니다.

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

agent = WORKERS[p["worker"]]        # "데이터분석가" 라는 문자열로 꺼낸다

---

5. 시스템 지시가 담당마다 다릅니다

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

"주문담당":    "... get_order_status 로 조회한 결과만 근거로 답한다.
              주문번호가 없으면 조회할 수 없다고 보고한다. 추측하지 않는다."

"데이터분석가": "... 숫자는 반드시 run_pandas 로 계산한다. 매출 집계에서는 취소·환불을 제외한다.
              리뷰 평점은 reviews 표의 rating 으로 계산한다(products 의 rating 은 카탈로그 표시값이다).
              리뷰 반응을 물으면 review_text 를 몇 건 직접 읽어 어떤 불만인지 함께 보고한다. ..."

"규정담당":    "... policy_search 로 찾은 사내 문서 내용만 근거로 답한다.
              문서에 없으면 '문서에 없음'이라고 보고한다. 추측하지 않는다."

"시장조사원":  "... search_web 으로 확인한 내용만 간결히 보고한다.
              검색이 실패하면 실패했다고 그대로 보고한다."

각 담당의 업무 규칙입니다. 4장 4절에서 정리한 구분 그대로죠.

어디에	무엇
도구의 독스트링	그 도구 자체의 설명 (여러 담당이 공유 가능)
담당의 시스템 지시	그 담당의 업무 규칙

데이터분석가에 두 줄이 더 있는 이유

"리뷰 평점은 reviews 표의 rating 으로 계산한다(products 의 rating 은 카탈로그 표시값이다)."
"리뷰 반응을 물으면 review_text 를 몇 건 직접 읽어 어떤 불만인지 함께 보고한다."

첫 줄 — products.csv 에도 rating 열이 있습니다(카탈로그 표시용 4.8). reviews.csv 의 실제 평균은 3.8이고요. 둘이 다릅니다. 안 알려 주면 4.8을 보고할 수 있습니다.

둘째 줄 — 리뷰 내용을 읽으라고 시킨 것입니다. 이게 없으면 "평균 3.8점"만 보고하고, "GPS 켜면 하루도 못 간다" 는 결정적인 단서가 안 나옵니다.

이 두 줄이 8장의 최종 판단을 이끌어 냅니다. 데이터를 아는 사람이 아니면 쓸 수 없는 문장이죠. 5장 3절에서 말한 "모델이 알 수 없는 것을 알려 주는 것이 사람의 일" 이 여기서 결정적입니다.

---

6. 워커는 기억이 없습니다

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

create_agent(llm, tools=[run_pandas], system_prompt="...")   # checkpointer 없음

워커에는 checkpointer 를 안 붙였습니다. 왜일까요?

워커가 받는 것: "P0003의 재고 수량을 알려주세요"   ← 완결된 문장

플래너가 완결된 문장으로 만들어 주기 때문입니다. 워커는 앞뒤 맥락이 필요 없습니다.

기억은 그래프 전체에 한 번만 붙어 있습니다.

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

team = graph.compile(checkpointer=InMemorySaver())

이것도 설계 판단입니다. 워커마다 기억을 주면 각자 대화 이력이 쌓여 복잡해지고 비용도 늡니다. 맥락 해석은 플래너가 하고, 워커는 매번 새로 시작 — 이 편이 단순합니다.

---

스스로 해 보기

1. 워커를 하나만 따로 불러 보세요.
   python out = WORKERS["데이터분석가"].invoke({"messages": [ {"role": "user", "content": "P0003 스마트워치의 재고와 재주문 기준을 알려주세요."}]}) print(out["messages"][-1].text)
   5장 에이전트와 같은 구조임을 확인하세요.
2. 규정담당에게 CSV 질문을 던져 보세요. 어떻게 답하나요?
   python out = WORKERS["규정담당"].invoke({"messages": [ {"role": "user", "content": "P0003의 재고가 몇 개인가요?"}]})
3. 데이터분석가의 시스템 지시에서 리뷰 관련 두 줄을 지우고 "P0003 리뷰 반응을 알려주세요" 를 물어 보세요. 무엇이 빠지나요?

3번이 이 절의 핵심 실험입니다. 결정적인 단서가 사라지는 것을 직접 보세요.

---

핵심 정리

• 이 파일에서 새로 만든 도구는 하나도 없습니다. 3·4·5·6장 것을 import 로 가져옵니다.
• get_order_status 와 search_web 둘을 tool() 로 감쌉니다. 평범한 파이썬 함수이기 때문입니다.
• 워커 넷은 전부 create_agent — 5장에서 만든 것과 똑같은 구조입니다. 도구를 하나씩만 줬을 뿐입니다.
• 딕셔너리로 묶습니다. 플래너가 이름 문자열로 지목하기 때문입니다.
• 시스템 지시는 담당마다 다릅니다. 도구의 독스트링과는 다른 자리입니다.
• 데이터분석가의 리뷰 관련 두 줄이 8장의 최종 판단을 이끌어 냅니다. 데이터를 아는 사람만 쓸 수 있는 문장입니다.
• 워커에는 기억이 없습니다. 플래너가 완결된 문장을 만들어 주기 때문입니다. 기억은 그래프 전체에 한 번만.