따라하기 — 각 담당에게 배분하고 실행하기
한 줄 요약
실행 노드는 for 문 하나입니다. 계획에 적힌 담당을 차례로 부르고 보고를 모읍니다. 각 담당 안에서는 4장의 ReAct 루프가 돕니다.
1. 실행 노드
아래는 code/ch08_multi_agent.py 의 일부입니다 — 위에서 이미 붙여 넣었으니 다시 넣지 마세요.
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}
핵심은 세 줄입니다.
아래는 개념 설명용 코드입니다 — 실습 파일에 넣지 않습니다.
agent = WORKERS[p["worker"]] # ① 담당자를 찾고
out = agent.invoke({"messages": [{"role": "user", "content": p["task"]}]}) # ② 일을 시키고
report = out["messages"][-1].text.strip() # ③ 보고를 받는다
2. 4장과 똑같은 구조
4장의 도구 실행 부분과 나란히 놓아 보세요.
아래는 개념 설명용 코드입니다 — 실습 파일에 넣지 않습니다. (4장 파일과 8장 파일을 나란히 놓은 것입니다.)
# 4장 — 도구를 부른다
for fc in resp.function_calls:
result = TOOLS[fc.name](**dict(fc.args))
# 8장 — 담당자를 부른다
for p in state["plan"]:
agent = WORKERS[p["worker"]]
out = agent.invoke({"messages": [{"role": "user", "content": p["task"]}]})
TOOLS[name] 이 WORKERS[name] 이 됐을 뿐입니다.
| 4장 | 8장 |
|---|---|
TOOLS 딕셔너리 |
WORKERS 딕셔너리 |
fc.name (문자열) |
p["worker"] (문자열) |
fc.args (인자) |
p["task"] (지시 문장) |
| 함수를 실행 | 에이전트를 실행 |
이게 멀티에이전트의 정체입니다. 도구를 부르던 자리에 에이전트를 부르는 것. 그래서 "에이전트를 도구처럼 쓴다" 고 표현하기도 합니다.
3. 각 담당 안에서 벌어지는 일
아래는 code/ch08_multi_agent.py 의 일부입니다 — 위에서 이미 붙여 넣었으니 다시 넣지 마세요.
out = agent.invoke({"messages": [{"role": "user", "content": p["task"]}]})
이 한 줄 안에서 4장의 ReAct 루프가 돕니다.
우리는 이 안을 안 봅니다. create_agent 가 다 하니까요. 5장에서 배운 그대로입니다.
보고만 받아 갑니다.
아래는 개념 설명용 코드입니다 — 실습 파일에 넣지 않습니다.
report = out["messages"][-1].text.strip() # 마지막 메시지가 최종 보고
4. 순서대로 부릅니다
아래는 개념 설명용 코드입니다 — 실습 파일에 넣지 않습니다.
for p in state["plan"]: # 하나씩 차례로
계획에 작업이 4개면 4번을 차례로 실행합니다. 앞이 끝나야 뒤가 시작되죠.
이게 8장이 1분쯤 걸리는 이유입니다. 동시에 부르지 않습니다 — 이 절 아래에서 이유를 밝힙니다.
동시에 부를 수는 없나요
있습니다. 각 담당이 서로 독립이므로 동시에 실행해도 됩니다.
아래는 개념 설명용 코드입니다 — 실습 파일에 넣지 않습니다.
# 개념만 — 이 과정에서는 안 씁니다
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor() as pool:
reports = list(pool.map(run_worker, state["plan"]))
작업이 4개라면 시간이 4분의 1 가까이로 줄어듭니다.
그런데 이 과정에서는 순서대로 합니다. 이유가 둘입니다.
- 화면에 보고가 순서대로 찍혀야 배우기 좋습니다. 동시에 실행하면 출력이 뒤섞입니다.
- API 요청이 한꺼번에 몰려 한도(
429)에 걸리기 쉽습니다.
Planner–Worker의 장점 중 하나가 "병렬화 가능" 이라고 앞 절에서 말했죠. 실제 서비스에서는 이걸 씁니다. 구조가 이미 그렇게 되어 있으므로 나중에 바꾸기 쉽습니다.
5. 보고를 모으는 형태
아래는 code/ch08_multi_agent.py 의 일부입니다 — 위에서 이미 붙여 넣었으니 다시 넣지 마세요.
results.append({"worker": p["worker"], "task": p["task"], "report": report})
세 가지를 함께 담습니다.
| 키 | 무엇 | 왜 필요한가 |
|---|---|---|
worker |
누가 | 종합 단계에서 출처 표시 |
task |
무엇을 물었는지 | 보고의 맥락 |
report |
보고 내용 | 실제 근거 |
task 를 함께 담는 이유가 있습니다. 종합 단계에서 이렇게 쓰이거든요.
아래는 code/ch08_multi_agent.py 의 일부입니다 — 위에서 이미 붙여 넣었으니 다시 넣지 마세요.
evidence = "\n\n".join(
f"[{r['worker']}] {r['task']}\n{r['report']}" for r in state["results"])
[데이터분석가] 최근 판매 추이, 재고 수량, 리뷰 평점과 부정 리뷰 내용을 분석해주세요.
재고 3개, 평균 3.8점, 주요 불만은 GPS 사용 시 배터리...
[규정담당] 제품 사양(배터리·방수 등)을 알려주세요.
배터리 일반 7일 / GPS 18시간, 방수 5ATM
무엇을 물어서 나온 답인지 함께 보여 주면, 종합 단계의 모델이 맥락을 정확히 잡습니다.
3장 3절의 원칙이 또 나옵니다 — 도구(여기서는 담당자)의 반환값은 "다음 사람이 읽을 보고서" 입니다.
6. 실행 결과
[데이터분석가 보고]
승승 스마트워치 Fit 5(P0003)의 정보는 다음과 같습니다.
* 재고 수량: 3개
* 리뷰 평균 평점: 3.8점
* 부정 리뷰(2점 이하) 주요 내용: "GPS를 켜면 하루도 못 갑니다. 일주일 간다는…
[규정담당 보고]
승승 스마트워치 Fit 5(P0003)의 제품 사양은 다음과 같습니다.
* 배터리(일반): 약 7일
* 배터리(GPS): 약 18시간
* 방수 등급: 5ATM
[규정담당 보고]
승승 스마트워치 Fit 5(P0003)의 환불 및 교환 기간 규정은 다음과 같습니다.
* 단순 변심: 상품 수령일로부터 7일 이내
* 상품 불량·하자: 상품 수령일로부터 30일 이내…
[시장조사원 보고]
승승 스마트워치 Fit 5는 가성비 스마트워치 시장에서 경쟁하는 것으로 보입니다.
주요 경쟁 제품으로는 화웨이, 아마즈핏 등의 브랜드에서 출시하는 20만원…
네 개의 보고가 순서대로 나왔습니다. 계획의 작업 4개에 대응하죠.
여기서 이미 결말이 보입니다.
[데이터분석가] "GPS를 켜면 하루도 못 갑니다. 일주일 간다는 설명과 너무 달라요"
[규정담당] 배터리 일반 7일 / GPS 18시간
두 보고를 나란히 놓으면 1절에서 본 그 판단이 나옵니다.
7. 보고를 잘라서 출력하는 이유
아래는 code/ch08_multi_agent.py 의 일부입니다 — 위에서 이미 붙여 넣었으니 다시 넣지 마세요.
print(f"\n [{p['worker']} 보고]\n {report[:160]}{'…' if len(report) > 160 else ''}")
160자만 찍습니다. 화면이 너무 길어지지 않게요.
종합 단계에는 전체가 넘어갑니다. 화면 출력만 자른 것입니다.
아래는 개념 설명용 코드입니다 — 실습 파일에 넣지 않습니다.
results.append({..., "report": report}) # 자르지 않은 전체
전체를 보고 싶다면 이렇게 바꿔 보세요.
아래는 개념 설명용 코드입니다 — 실습 파일에 넣지 않습니다.
print(f"\n [{p['worker']} 보고]\n{report}")
8. 담당이 실패하면
지금 코드에는 실패 처리가 없습니다.
아래는 개념 설명용 코드입니다 — 실습 파일에 넣지 않습니다.
out = agent.invoke(...) # 여기서 예외가 나면 전체가 멈춘다
시장조사원의 웹검색이 실패하면 어떻게 될까요?
괜찮습니다. 4장 9절에서 search_web 을 try/except 로 감쌌기 때문입니다.
아래는 개념 설명용 코드입니다 — 실습 파일에 넣지 않습니다.
except Exception as e:
return (f"검색에 실패했습니다({type(e).__name__}). ...")
도구 안에서 이미 실패를 처리하고 있습니다. 시장조사원은 "검색에 실패했습니다"라는 보고를 하고, 나머지 담당의 보고로 답이 나옵니다.
그런데 도구가 아닌 곳에서 예외가 나면 전체가 멈춥니다. 9절에서 이 문제를 다룹니다.
스스로 해 보기
- 보고 전체를 출력하도록 바꿔 실행해 보세요. 각 담당이 얼마나 자세히 보고하나요?
- 담당자 한 명을 직접 불러 보세요.
python out = WORKERS["규정담당"].invoke({"messages": [ {"role": "user", "content": "승승 스마트워치 Fit 5의 배터리 사양을 알려주세요."}]}) for m in out["messages"]: print(f" {type(m).__name__:14s} {str(m.content)[:60]}")
안에서 ReAct 루프가 도는 것을 확인하세요. execute_node에 각 담당의 실행 시간을 재는 코드를 넣어 보세요.
python import time start = time.time() out = agent.invoke(...) print(f" ({time.time() - start:.1f}초)")results에서task를 빼고 종합해 보세요. 최종 답변이 달라지나요?
핵심 정리
- 실행 노드는
for문 하나입니다. 담당자를 찾고, 일을 시키고, 보고를 받습니다. - 4장의 도구 실행과 구조가 같습니다.
TOOLS[name]이WORKERS[name]이 된 것뿐입니다. - 도구를 부르던 자리에 에이전트를 부르는 것 — 이것이 멀티에이전트의 정체입니다.
- 각 담당의
invoke한 줄 안에서 4장의 ReAct 루프가 돕니다. 우리는 안 봅니다. - 순서대로 부릅니다. 동시에 할 수도 있지만, 출력이 뒤섞이고 API 한도에 걸리기 쉬워서 안 합니다.
- 보고는 누가·무엇을 물어서·무엇이라 답했는지 셋을 함께 담습니다.
- 화면에는 160자만 찍고, 종합 단계에는 전체가 넘어갑니다.
- 도구 안에서 난 실패는 이미 처리하고 있어 나머지 보고로 답이 나옵니다. 도구 밖의 실패(API 한도
429, 네트워크)는 아직 처리하지 않았습니다 — 9절에서 처리합니다.