멀티 에이전트

지도 위치 : 3일차 · README 4장의 ⑥ 서비스 층(마무리) · 과정 전체 정산 | 이전: 23차시 Human-in-the-Loop | 다음: 전체 목차로 키워드 : API 서버, 스트리밍, 통합 테스트

이 차시를 다 읽으면

1. API 서버를 유와 종차로 정의하고, 오피스봇의 엔드포인트(대화 전송 · 승인 처리 · 세션 조회 · 기억 삭제)를 설계할 수 있다.
2. 로그인한 사용자의 신원이 도구 권한까지 이어지는 길을 설명할 수 있다.
3. 스트리밍(SSE)을 축 ⑥으로 설명하고, "빨라지는 것이 아니라 먼저 보여 주는 것 "임을 말할 수 있다.
4. 통합 테스트 시나리오표와 평가셋 재실행으로 서비스 전체를 검수하고, 운영에서 볼 지표를 댈 수 있다.
5. 과정 전체를 여섯 층 · 여덟 축 · 중심 축의 여정 · 세 문장 으로 정산할 수 있다.

---

0. 이 차시의 판별 축

이 차시가 답하는 질문은 둘입니다. "지금까지 만든 그래프를 사람들이 쓰는 서비스로 어떻게 내보내나?" , 그리고 "이 과정 전체에서 무엇이 남았나?" 1~5절은 앞의 것을, 7절은 뒤의 것을 다룹니다.

---

1. API 서버

1.1 정의

API 서버 — [서버 프로그램]인데, [ 정해진 주소(엔드포인트)로 요청을 받아 프로그램의 기능을 실행하고 결과를 돌려준다 ]는 점에서 사람이 화면을 보며 쓰는 웹 페이지 자체나, 한 사람이 터미널에서 돌리는 스크립트와 구별된다.

FastAPI — [Python 웹 프레임워크]인데, [ 함수에 주소를 붙이는 것만으로 API를 만들고, 요청 형식을 Pydantic 스키마로 검증한다 ]는 점에서 화면까지 함께 만드는 웹 프레임워크와 구별된다. 앞에서 배운 구조화 출력(4차시)과 같은 Pydantic을 씁니다.

1.2 엔드포인트 설계

---

2. 인증 — 사용자 신원에서 도구 권한까지

from fastapi import FastAPI, Depends, HTTPException
api = FastAPI()

@api.post("/approvals/{thread_id}")
async def approve(thread_id: str, body: ApprovalBody, user=Depends(get_current_user)):
    if not owns_thread(user.id, thread_id):           # 남의 승인 대기는 건드릴 수 없음
        raise HTTPException(status_code=403)
    config = {"configurable": {"thread_id": thread_id}}
    return await graph.ainvoke(Command(resume=body.model_dump()), config)

🔍 읽는 포인트

• Depends(get_current_user) — 요청마다 회사 로그인(SSO 토큰 등)을 검증해 사번을 꺼내는 함수입니다. 구현은 회사 인증 방식에 따릅니다.
• ⭐ 이 user.id가 유일한 신원 입니다. State의 requester_id(21차시), 도구 함수의 "요청자 = 대상" 검사(8차시), 기억 조회(17차시), 승인자 확인(23차시)이 모두 이 값에서 출발합니다. 모델이 대화에서 말한 사번은 어디에서도 신원으로 쓰지 않습니다.
• owns_thread(...) — thread_id만 알면 남의 흐름을 재개할 수 있으면 안 됩니다. 소유자를 코드로 확인합니다.

---

3. 스트리밍 — 축 ⑥

3.1 정의

스트리밍 — [응답 전달 방식]인데, [ 결과가 다 만들어지기 전에 만들어지는 대로 조금씩 보낸다 ]는 점에서 다 끝난 뒤 한 번에 보내는 방식과 구별된다.

SSE (Server-Sent Events) — [스트리밍 규약]인데, [ 서버에서 화면 쪽으로 한 방향으로 data: ... 줄을 이어 보낸다 ]는 점에서 양쪽이 자유롭게 주고받는 웹소켓과 구별된다.

⚠️ 축 ⑥의 함정 그대로입니다. 스트리밍해도 답이 완성되는 시간은 같습니다. 달라지는 것은 사용자가 첫 글자를 보는 시간 과 "무엇을 하고 있는지" 아는 것입니다. 멀티에이전트는 단계가 많아 수십 초가 걸릴 수 있는데, 그동안 빈 화면이면 사용자는 고장으로 여깁니다.

3.2 에이전트의 중간 진행 상황 보여 주기

from fastapi.responses import StreamingResponse
import json

LABELS = {"classify": "질문 유형 확인 중", "hr": "인사 담당이 확인 중",
          "policy": "규정 검색 중", "synthesize": "답변 정리 중"}

@api.post("/chat")
async def chat(req: ChatRequest, user=Depends(get_current_user)):
    config = {"configurable": {"thread_id": f"{user.id}:{req.session_id}"}}
    inputs = {"question": req.message, "requester_id": user.id}

    async def events():
        async for update in graph.astream(inputs, config, stream_mode="updates"):
            for node, change in update.items():
                if node == "__interrupt__":                    # 승인 대기 (표시 방식은 버전 확인)
                    yield sse({"type": "approval_needed", "thread_id": config["configurable"]["thread_id"]})
                elif node in LABELS:
                    yield sse({"type": "progress", "text": LABELS[node]})
                if isinstance(change, dict) and "answer" in change:
                    yield sse({"type": "answer", "text": change["answer"]})

    return StreamingResponse(events(), media_type="text/event-stream")

def sse(obj) -> str:
    return f"data: {json.dumps(obj, ensure_ascii=False)}\n\n"

🔍 읽는 포인트

• thread_id = f"{user.id}:{...}" — 세션 열쇠에 사번을 묶어 다른 사용자의 세션과 섞이지 않게 합니다(16차시 — 세션 분리 실패는 개인정보 사고).
• graph.astream(..., stream_mode="updates") — 18차시의 stream과 같은 것의 비동기판입니다. 노드가 끝날 때마다 바뀐 값이 옵니다. 진행 상황(progress)은 여기서 만듭니다.
• LABELS — 노드 이름을 사용자 말로 바꿉니다. 이 표에 없는 노드(내부 노드)는 화면에 보이지 않습니다.
• "__interrupt__" — 23차시의 승인 대기에서 멈춘 신호입니다. 화면에 승인 창을 띄우게 합니다. 인터럽트가 스트림에 어떤 모양으로 나오는지는 버전에 따라 다르니 공식 문서를 확인 하세요.
• data: ...\n\n — SSE의 형식입니다. 빈 줄로 한 사건이 끝납니다.
• 💡 답의 글자 단위 스트리밍이 필요하면 답변 노드 안에서 client.messages.stream(...)의 text_stream을 쓰고, 그래프 스트림 방식도 그에 맞게 바꿉니다(문서 확인).

3.3 화면 연동

const res = await fetch("/chat", { method: "POST", body: JSON.stringify({ session_id, message }),
                                   headers: { "Content-Type": "application/json" } });
const reader = res.body.getReader();
const decoder = new TextDecoder();
while (true) {
  const { value, done } = await reader.read();
  if (done) break;
  handleEvents(decoder.decode(value));   // "data: {...}" 줄을 잘라 progress · answer · approval_needed 처리
}

🔍 읽는 포인트

• fetch + getReader() — 응답을 도착하는 대로 조각씩 읽습니다. 브라우저의 EventSource는 GET만 지원하므로, 메시지를 POST로 보낼 때는 이 방식을 씁니다.
• 화면은 LLM API 키를 모릅니다. 화면 → 우리 서버 → 외부 LLM 순서이고, 키와 권한 검사는 서버에만 있습니다(축 ④).

---

4. 통합 테스트와 운영

4.1 통합 테스트 시나리오표

통합 테스트 — [소프트웨어 시험]인데, [ 화면 · 서버 · 그래프 · 도구 · 외부 API를 실제로 이어 붙인 상태에서 사용자 시나리오 단위로 확인한다 ]는 점에서 부품 하나씩 따로 보는 단독 시험(21차시)과 구별된다.

(결과 칸은 실측 후 채우세요.) 5번은 LLM API 에러를 종류별로 다루는지 보는 시나리오입니다 — anthropic.RateLimitError(429)와 5xx(APIStatusError)는 잠시 후 재시도, BadRequestError(400)는 요청이 틀린 것이라 재시도해도 소용없습니다.

4.2 평가셋 전체 재실행

과정 중에 만든 평가셋을 서비스 주소로 다시 돌립니다. 부품을 이어 붙이며 동작이 바뀌지 않았는지 보는 회귀 확인 입니다(11차시).

(빈칸 양식입니다.) 점수가 떨어졌다면 이어 붙인 경계 (인증 · 세션 · 넘겨주기)를 먼저 의심합니다.

4.3 운영 시 볼 것

---

5. ⭐ 과정 전체 정산

5.1 README 4장 — 여섯 층에 무엇이 남았나

위층은 아래층을 부품으로 썼습니다. 22차시의 Supervisor는 4차시의 구조화 출력으로 결정을 내고, 21차시의 전문 에이전트는 6 · 8차시의 도구와 권한 경계를 그대로 품고 있습니다. 아래층이 부실하면 위층이 그 부실을 반복해서 키운다는 README 1.2의 말이 그래서 맞습니다.

5.2 README 5장 — 여덟 큰 축이 어디서 나왔나

5.3 README 5.2 — 중심 축 "누가 흐름을 정하는가"의 여정

 코드가 정함 (0) ←──────────────────────────────────→ 모델이 정함 (1)

 1일차  의도 분류 후 if 분기 (4 · 7)  ●
 2일차  에이전트 루프 (9)                                          ●
 3일차  그래프 (18) — 큰 흐름은 코드          ●  ← 칸 안 판단은 모델
        규칙 라우터 / LLM 라우터 (19)      ●     ●
        Planner-Worker (20)                     ●
        Supervisor (22)                                         ●
        HITL (23) — 이 축 위에 "사람이 정하는 칸"을 끼워 넣음

1일차는 왼쪽, 2일차 루프는 오른쪽 끝까지 갔다가, 3일차는 둘을 섞는 법 을 배웠습니다. 큰 흐름은 그래프(코드)가 쥐고, 칸 안의 판단은 모델이 하고, 되돌릴 수 없는 칸에는 사람이 앉습니다. 같은 요청이라도 어느 칸을 어느 쪽에 둘지 고르는 것이 이 과정에서 기른 설계 판단입니다.

5.4 README 6장 — 세 문장 회수

---

6. 실습 — 맡기고 검수하기

지시

오피스봇 그래프(18~23차시)를 FastAPI 서버로 내보내고 간단한 채팅 화면을 붙여줘.
참고 공식 문서: FastAPI(StreamingResponse, Depends), LangGraph 스트리밍(astream, stream_mode)과 interrupt 문서,
  Claude API 스트리밍 · 에러 처리 문서. 설치된 버전을 모두 출력할 것.
구조: 1.2의 엔드포인트 5개. /chat은 SSE로 progress · answer · approval_needed 사건을 보낼 것.
제약:
- 신원은 get_current_user 하나에서만(테스트용 가짜 로그인 토큰 2개). 모델 출력의 사번은 쓰지 말 것.
- thread_id에 사번을 포함하고, 승인 · 조회 엔드포인트에서 소유자 검사.
- API 키는 환경변수로만, 화면 코드에 넣지 말 것. 서버 외부 공개(포트 개방 · 배포)는 나에게 승인받은 뒤에.
테스트: 4.1 시나리오 7개 전부 + 4.2 평가셋 재실행.
보고 형식: 4.1 표와 4.2 표를 채우고, 실패 시나리오는 [시나리오 | 기대 | 실제 | 로그 원문] 표로.
추측하지 말고 실행 명령, 서버 로그 원문, 화면 캡처를 붙여.

받을 증거

• 서버 실행 명령과 로그, 시나리오 7개 결과표 , 평가셋 재실행표
• 화면 캡처: 진행 표시가 차례로 바뀌는 장면, 승인 창
• 시나리오 6(동시 사용자)의 두 세션 로그 — 서로 다른 thread_id

내가 직접 확인할 것

---

부록 A. 스스로 확인하기

부록 B. 자주 하는 실수

부록 C. NotebookLM 복습 프롬프트

이 24차시 노트만 근거로 비전공자용 6문제를 내줘.
"신원이 도구 권한까지 이어지는 길", "스트리밍과 축 ⑥", "통합 테스트 시나리오", "중심 축의 여정", "세 문장 회수"는 반드시 포함해줘.
내가 답한 뒤에 채점하고, 틀린 이유를 노트의 절 번호를 근거로 설명해줘.