8장 · 멀티에이전트 아키텍처 — 절 목차01. 한 종류로는 답이 안 나오는 질문 02. 왜 하나에 도구를 다 몰아넣지 않는가 역할 분리 03. 따라하기 네 에이전트를 한 파일로 모으기 04. 구성 방식 두 가지 한 명에게 넘기기 vs 나눠서 맡기기 05. 따라하기 팀장이 작업 목록을 만들게 하기 06. 따라하기 각 담당에게 배분하고 실행하기 07. 따라하기 여러 보고를 하나의 판단으로 합치기 08. 실행 스마트워치 Fit 5 어떻게 대응할까요 전체 데모 09. 한 담당이 실패해도 나머지는 살리기 10. 수료 무엇을 만들었고 다음에 무엇을 할 수 있나

한 담당이 실패해도 나머지는 살리기

한 줄 요약

담당이 여럿이면 한 명은 실패합니다. 그래도 나머지 보고로 답이 나와야 합니다. 실패를 예외가 아니라 보고 내용으로 다루는 것이 핵심입니다.

1. 실제로 자주 일어납니다

  [시장조사원 보고]
    검색에 실패했습니다(RatelimitException). 네트워크 상태를 확인하거나
    잠시 후 다시 시도해 주세요.

4장 9절에서 예고한 그것입니다. 웹검색은 짧은 시간에 여러 번 부르면 실패하기 쉽습니다.

담당이 넷이면 확률이 네 배입니다. 하나만 실패해도 전체가 멈추면 곤란하죠.

2. 이미 막혀 있는 부분

도구 수준에서 이미 처리했습니다.

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

# 4장 — search_web
except Exception as e:
    return (f"검색에 실패했습니다({type(e).__name__}). "
            "네트워크 상태를 확인하거나 잠시 후 다시 시도해 주세요.")

# 5장 — run_pandas
except Exception as e:
    return f"실행 오류: {type(e).__name__}: {e}"

예외를 던지지 않고 문장을 돌려줍니다. 3장부터 반복한 원칙이죠.

그래서 이렇게 흘러갑니다.

① search_web 이 실패 → "검색에 실패했습니다" 문자열을 돌려줌
② 시장조사원이 그 문장을 읽음
③ 시장조사원의 보고: "검색에 실패하여 시장 동향을 확인하지 못했습니다"
④ 종합 단계가 그 보고를 받음
⑤ 최종 답변: "...시장 동향은 확인하지 못했습니다. 다만 사내 데이터로는..."

나머지 담당의 보고로 답이 나옵니다.

3. 시스템 지시가 한몫합니다

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

"시장조사원": create_agent(
    llm, tools=[search_tool],
    system_prompt="너는 시장조사 담당자다. search_web 으로 확인한 내용만 간결히 보고한다. "
                  "검색이 실패하면 실패했다고 그대로 보고한다."),

"검색이 실패하면 실패했다고 그대로 보고한다" — 이 문장이 없으면 어떻게 될까요?

모델이 "일반적으로 스마트워치 시장은..." 하며 자기 지식으로 메울 수 있습니다. 1장의 그 환각이죠.

실패를 정직하게 보고하게 만드는 것이 더 안전합니다. 종합 단계에서 "확인하지 못했다"고 쓸 수 있으니까요.

6장 6절의 원칙과 같습니다 — 모를 때 할 말을 정해 준다. 여기서는 "실패했다고 그대로 보고하라"입니다.

4. 그래도 남는 위험 — 도구 밖의 실패

도구 안의 예외는 처리했지만, 도구 밖의 예외는 아직 처리하지 않았습니다.

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

def execute_node(state: State) -> dict:
    results = []
    for p in state["plan"]:
        agent = WORKERS[p["worker"]]
        out = agent.invoke({...})      # ← 여기서 예외가 나면 전체가 멈춘다

여기서 예외가 날 수 있는 경우들입니다.

원인	언제
API 한도 (`429`)	호출이 몰릴 때 — 8장에서 가장 흔함
네트워크 끊김	인터넷 문제
없는 담당자 이름	플래너가 이상한 이름을 낼 때 (`Literal` 로 막았지만)
응답 시간 초과	서버가 느릴 때

하나라도 나면 그때까지 받은 보고가 전부 날아갑니다.

5. 담당별로 감싸기

한 담당의 실패가 전체를 무너뜨리지 않게 만듭니다.

개선안 — 원하면 고쳐 보세요 · agentic_ai/code/ch08_multi_agent.py
실습 기본 코드에는 없는 내용입니다. 지금 안 고쳐도 다음 절로 넘어갈 수 있습니다.

def execute_node(state: State) -> dict:
    """계획에 적힌 워커를 차례로 부른다. 한 명이 실패해도 나머지는 계속한다."""
    results = []
    for p in state["plan"]:
        try:
            agent = WORKERS[p["worker"]]
            out = agent.invoke({"messages": [{"role": "user", "content": p["task"]}]})
            report = out["messages"][-1].text.strip()
        except Exception as e:
            # [중요] 담당 한 명이 실패해도 나머지 보고로 답을 만든다.
            report = f"조사에 실패했습니다({type(e).__name__}). 이 항목은 확인되지 않았습니다."
        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}

try/except 로 감싸고, 실패도 보고로 만듭니다.

도구 안에서 한 것과 똑같은 처리를 한 단계 위에서 한 번 더 하는 것이죠.

[도구 수준]     search_web 이 실패 → "검색에 실패했습니다" 문자열
[담당 수준]     담당자 호출이 실패 → "조사에 실패했습니다" 보고
[종합 수준]     받은 보고로 답을 만든다

층마다 방어가 있습니다.

6. 종합이 실패를 다루게 하기

종합의 시스템 지시에도 한 줄 넣어 두면 좋습니다.

개선안 — 원하면 고쳐 보세요 · agentic_ai/code/ch08_multi_agent.py
실습 기본 코드에는 없는 내용입니다. 지금 안 고쳐도 다음 절로 넘어갈 수 있습니다.

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

그러면 이런 답이 나옵니다.

① 확인된 사실
  * 재고 3개, 재주문 기준 30개 (데이터분석가)
  * 배터리 일반 7일 / GPS 18시간 (규정담당)
  * 시장 동향: 확인되지 않음 (시장조사원 — 검색 실패)

② 권고
  1. 즉시 27개 발주
  2. 상세페이지에 GPS 배터리 표기 보완
  (시장 상황은 확인되지 않아 경쟁 대응 방안은 제외했습니다)

무엇이 빠졌는지 사람이 알 수 있습니다.

이게 중요합니다. 실패를 숨기면 사람이 불완전한 답을 완전한 답으로 착각합니다. "확인되지 않음"을 드러내는 것이 훨씬 안전합니다.

7. 재시도는 어디에 넣나

한 번 더 시도하면 되는 실패도 있습니다. 특히 429 는 잠깐 기다리면 풀립니다.

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

import time

def run_worker(p, attempts=2):
    for i in range(attempts):
        try:
            out = WORKERS[p["worker"]].invoke({"messages": [{"role": "user", "content": p["task"]}]})
            return out["messages"][-1].text.strip()
        except Exception as e:
            if i < attempts - 1:
                print(f"    (재시도 {i+1}/{attempts-1}: {type(e).__name__})")
                time.sleep(3)
            else:
                return f"조사에 실패했습니다({type(e).__name__})."

주의할 점 둘.

무한 재시도는 안 됩니다. 4장 8절의 MAX_STEPS 와 같은 이야기죠. 횟수를 정해 둡니다.
기다렸다 다시 합니다. 바로 재시도하면 같은 이유로 또 실패합니다.

모든 실패에 재시도가 답은 아닙니다. 429(한도)나 일시적 네트워크 문제는 재시도가 통하지만, 잘못된 인자나 없는 데이터는 몇 번을 해도 같습니다. 실패의 종류를 보고 정해야 합니다.

8. 계획 단계가 실패하면

가장 치명적인 실패입니다.

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

def plan_node(state: State) -> dict:
    result = planner.invoke([...])      # ← 여기서 실패하면 아무것도 못 한다

계획이 없으면 실행할 것도 없습니다.

가장 단순한 대비는 기본 계획을 두는 것입니다.

개선안 — 원하면 고쳐 보세요 · agentic_ai/code/ch08_multi_agent.py
실습 기본 코드에는 없는 내용입니다. 지금 안 고쳐도 다음 절로 넘어갈 수 있습니다.

def plan_node(state: State) -> dict:
    try:
        result = planner.invoke([{"role": "system", "content": planner_prompt()}] + state["messages"])
        plan = [{"worker": t.worker, "task": t.task} for t in result.tasks]
    except Exception as e:
        # 계획을 못 세우면 질문 그대로 담당 전원에게 물어본다
        print(f"  (계획 실패: {type(e).__name__} — 전체 조사로 대체)")
        question = state["messages"][-1].text
        plan = [{"worker": w, "task": question} for w in WORKERS]

    print("\n  [계획]")
    for p in plan:
        print(f"    → {p['worker']}: {p['task']}")
    return {"plan": plan, "results": []}

for w in WORKERS 는 딕셔너리의 키를 훑는 것이라 담당 넷 전부가 들어갑니다. 계획을 못 세우면 넷에게 다 물어보는 것입니다. 비효율적이지만 아무 답도 못 내는 것보다는 낫습니다.

9. 정리 — 방어의 층

실패를 세 단계에서 처리한다 — 도구 · 담당 · 종합

각 층에서 같은 원칙을 씁니다.

예외를 던지지 말고, 상황을 설명하는 문장을 돌려준다.

3장 4절에서 처음 나온 이 원칙이 8장까지 그대로 이어집니다.

스스로 해 보기

search_web 을 일부러 실패하게 만들고 전체를 실행해 보세요.
python def search_web(query: str) -> str: """인터넷에서 최신 정보를 검색한다. ...""" raise RuntimeError("일부러 실패")
답이 나오나요? 무엇이 빠졌다고 하나요?
5번의 담당별 try/except 를 넣고, 워커 하나를 일부러 고장 내 보세요.
python WORKERS["시장조사원"] = None # invoke 에서 예외가 난다
6번의 종합 지시를 추가하고, 실패 상황에서 답이 어떻게 달라지는지 보세요.
8번의 기본 계획을 넣고, planner 를 고장 내 보세요.

1번은 아무것도 안 고쳐도 답이 나옵니다. 도구 안에서 이미 실패를 처리했기 때문이죠. 2번부터가 이 절에서 추가하는 방어입니다.

핵심 정리

담당이 여럿이면 한 명은 실패합니다. 특히 웹검색과 API 한도(429)가 그렇습니다.
도구 안에서는 이미 실패를 처리하고 있습니다 — 3~6장에서 try/except 로 문장을 돌려주게 했습니다.
시스템 지시의 "검색이 실패하면 실패했다고 그대로 보고한다" 가 지어내기를 막습니다.
도구 밖의 실패(API 한도, 네트워크)는 아직 처리하지 않았습니다. 담당별로 try/except 를 씌웁니다.
종합 지시에 "조사되지 않은 항목은 '확인되지 않음'이라고 명시" 를 넣습니다. 실패를 숨기면 사람이 착각합니다.
재시도는 횟수를 정하고 기다렸다 합니다. 모든 실패에 재시도가 답은 아닙니다.
계획 단계 실패가 가장 치명적입니다. 기본 계획(담당 전원에게 질문 그대로)을 두면 최소한 답은 나옵니다.
모든 층에서 같은 원칙 — 예외를 던지지 말고 상황을 설명하는 문장을 돌려준다.