LangGraph + HITL

Built a multi-agent pipeline that generates a lecture plan from a 6-field input (topic, audience, duration, delivery method, tools, constraints) using LangGraph + FastAPI + Gemini API + Tavily.

The pipeline runs in 3 rounds: sequential planning → parallel web research → sequential writing → parallel review (3 perspectives simultaneously). If the combined reviewer score falls below threshold, interrupt() pauses the graph at that exact point; MemorySaver stores state per thread_id; the human POSTs a decision to /resume and execution continues from where it stopped.

Key design decisions:

Send API for dynamic parallel: the number of research subtopics is determined at runtime by the planner. Send() returns one task per subtopic and LangGraph executes them all simultaneously. Timestamps confirm same start time, different end times — true parallel. Previously implemented in n8n using a single sequential node as a workaround, because parallel branching there causes N² API calls.

Annotated[list, operator.add] + return diff only: parallel nodes must not return {**state, ...}. Multiple nodes writing the same key in one step causes InvalidUpdateError. The fix: declare accumulator fields as Annotated[list[dict], operator.add] and have each node return only its contribution {"research_results": [result]}. The reducer handles merging automatically. Both research_results and reviews need this treatment.

interrupt() as HITL primitive: one function call pauses the entire graph. No external notification system needed — the /generate response carries score and feedback directly, and the server terminal prints ready-to-copy curl commands with the actual thread_id. Resuming is a single POST to /resume with Command(resume=decision).

@log_node decorator: wraps every node to print HH:MM:SS timestamp + node name to terminal and append to subagent_log.csv. Parallel nodes show identical start times, confirming simultaneous execution.

Codebase split into 6 modules: state.py / utils.py / nodes.py / graph.py / main.py / analyze.py (standalone CLI script).

Tech: LangGraph, FastAPI, Python, Gemini 3.1 Flash Lite (free tier, fallback chain), langchain-google-genai, Tavily Search API

6개 필드(주제/대상자/일수/방식/도구/제약조건)를 입력하면 강의계획서를 자동 생성하는 멀티에이전트 파이프라인을 LangGraph + FastAPI로 만들었다.

파이프라인은 3라운드로 구성된다. 서브토픽 분해 → 서브토픽별 웹 검색(Send API, 동적 N개 진짜 병렬) → 초안 작성 → 3종 병렬 검토. 종합 점수가 기준 미달이면 interrupt()가 그래프를 정지시키고, MemorySaver가 thread_id별 상태를 저장한다. 사람이 /resume으로 결정을 보내면 이어서 실행된다.

핵심 설계 결정:

Send API (동적 병렬): 서브토픽 수는 실행 시점에 결정된다. Send()로 서브토픽 수만큼 research_node를 동시 실행한다. 이전에 n8n으로 같은 시나리오를 구현할 때는 병렬 연결이 N² 호출을 유발해서 순차 처리로 우회했다. LangGraph와 n8n의 핵심 차이다.

Annotated reducer + 변경분 반환: 병렬 노드가 {**state, ...} 전체를 반환하면 InvalidUpdateError 발생. Annotated[list[dict], operator.add]를 선언하고 각 노드가 {"research_results": [result]}만 반환하는 것이 올바른 패턴. research_results와 reviews 두 필드 모두 적용 필요.

interrupt() + MemorySaver: 코드 한 줄로 HITL 게이트 선언. 별도 알림 시스템 없이 /generate 응답에 점수와 피드백이 직접 담기고, 서버 터미널에 실제 thread_id가 포함된 curl 명령어가 자동 출력된다.

@log_node 데코레이터: 타임스탬프 + 노드명을 터미널에 출력하고 CSV에 기록. 병렬 노드의 시작 시각이 같게 찍혀 진짜 병렬임을 확인할 수 있다.

사용 기술: LangGraph, FastAPI, Python, Gemini 3.1 Flash Lite (무료 API, 폴백 체인), langchain-google-genai, Tavily Search API

Links Github: https://github.com/SangjinKO/LangGraph_HITL_lecture_planner/tree/main Velog: https://velog.io/@kosang234/LangGraph%EB%A1%9C-%EA%B0%95%EC%9D%98%EA%B3%84%ED%9A%8D%EC%84%9C%EB%A5%BC-%EC%9E%90%EB%8F%99-%EC%83%9D%EC%84%B1%ED%95%98%EB%8A%94-Multi-Agent-AI-Workflow%EB%A5%BC-%EB%A7%8C%EB%93%A4%EB%A9%B4%EC%84%9C-%EA%B9%A8%EB%8B%AC%EC%9D%80-%EA%B2%83%EB%93%A4