diff --git a/journey/plans/251218_langgraph_1.0_upgrade_plan.md b/journey/plans/251218_langgraph_1.0_upgrade_plan.md index 5d4fbbb..c938cf3 100644 --- a/journey/plans/251218_langgraph_1.0_upgrade_plan.md +++ b/journey/plans/251218_langgraph_1.0_upgrade_plan.md @@ -1,55 +1,197 @@ # LangGraph 1.0 업그레이드 계획 **날짜**: 2025-12-18 +**수정일**: 2026-01-17 **작성자**: admin -**관련 파일**: `rb8001/app/services/workflows/coldmail_workflow.py`, `rb8001/requirements.txt` +**관련 파일**: +- `rb8001/app/services/workflows/coldmail_workflow.py` +- `rb8001/app/router/ir_deck.py` +- `rb8001/app/services/ir_deck_analyzer.py` +- `rb8001/requirements.txt` --- ## 현재 상황 -- **현재 버전**: `langgraph==0.6.10` -- **사용 위치**: 콜드메일 워크플로우 (`coldmail_workflow.py`) -- **체크포인터**: `AsyncSqliteSaver` 사용 (수동 설정) +- **현재 버전**: `langgraph==0.6.10` (1.0은 2025-10-22 출시 완료, 업그레이드 가능) +- **사용 위치**: + - 콜드메일 워크플로우 (`coldmail_workflow.py`) - `interrupt_after=["process"]` 사용 중 + - IR Deck 평가 (`ir_deck.py`) - `asyncio.create_task`로 비동기 실행 (LangGraph 미사용) +- **체크포인터**: `AsyncSqliteSaver` 사용 (`coldmail_briefing.py:103`, `coldmail_service.py:121`) + +**참고**: LangGraph 1.0은 2025-10-22 출시 완료. 현재 0.6.10에서 1.0으로 업그레이드 진행 가능. `interrupt_after`는 1.0에서도 지원되지만, `interrupt()` 동적 인터럽트 사용 권장. ## LangGraph 1.0 주요 변경사항 -1. **지속적인 상태 관리 (Durable State)**: 서버 재시작/중단 시 자동 재개 -2. **내장된 지속성**: 워크플로우 저장/재개 기능 강화 -3. **휴먼 인 더 루프 (HITL) 패턴**: 실행 일시 중지 후 인간 승인 API 제공 -4. **그래프 기반 실행 모델 개선**: 결정론적/에이전트 컴포넌트 혼합 제어 강화 +1. **Durable State**: `AsyncSqliteSaver` 같은 외부 체크포인터 사용 시 서버 재시작/중단 시 자동 재개 지원 (0.6.10에서도 지원, 1.0에서 강화) +2. **HITL 패턴 개선**: `interrupt_after` 정적 인터럽트는 유지되지만, 동적 `interrupt()` 사용 권장 (조건부 일시 중지 가능) +3. **API 변경**: `Command(resume=...)`로 재개, `interrupt()` 함수 사용 권장 -## Phase 1: 콜드메일 워크플로우 HITL 패턴 적용 +--- + +## Phase 1: 콜드메일 워크플로우 HITL 패턴 마이그레이션 + +**목표**: `interrupt_after=["process"]` 정적 인터럽트 → 동적 `interrupt()` 사용 **파일**: `rb8001/app/services/workflows/coldmail_workflow.py` -- `process_node`에서 HITL 패턴으로 일시 중지 → 사용자 승인 대기 → 재개 -- "이 기업을 분석해 드릴까요?" 버튼을 워크플로우 내부에서 처리 + +### 1.1 마이그레이션 방법 + +**현재 코드** (0.6.10): +```python +# coldmail_workflow.py:281 +if checkpointer is not None: + return workflow.compile(checkpointer=checkpointer, interrupt_after=["process"]) +``` + +**변경 후** (1.0 권장 방식): +```python +from langgraph.types import interrupt, Command + +async def process_node(state: ColdmailState) -> ColdmailState: + """콜드메일 처리 노드 (동적 인터럽트 사용)""" + # ... 기존 로직 ... + + # 신뢰도 95% 미만: 동적 인터럽트로 일시 중지 + if confidence < 0.95: + interrupt({ + "type": "confirmation", + "email_id": email.get("id"), + "subject": email.get("subject"), + "confidence": confidence + }) + + # ... 나머지 로직 ... +``` + +**재개 방법** (`coldmail_service.py` 수정): +```python +# 기존: workflow.ainvoke(state, config) - 상태 업데이트로 재개 +# 변경: Command(resume=True)로 명시적 재개 +result = await workflow.ainvoke(Command(resume=True), config=config) +``` + +### 1.2 작업 내용 + +1. `process_node`: 신뢰도 95% 미만일 때 `interrupt()` 호출 (기존 `waiting_confirmation` 로직 유지) +2. `coldmail_workflow.py:281`: `interrupt_after` 제거, `interrupt()` 동적 사용으로 전환 +3. `coldmail_service.py:158`: 재개 로직을 `Command(resume=True)` 사용으로 변경 + +### 1.3 테스트 + +- 신뢰도 95% 이상: 자동 처리 (인터럽트 없음) +- 신뢰도 95% 미만: `interrupt()` 호출 → 확인 버튼 클릭 → `Command(resume=True)` 재개 + +**참고**: `interrupt_after`는 1.0에서도 지원되지만, 디버깅용으로만 권장. 프로덕션 HITL은 `interrupt()` 사용. + +--- ## Phase 2: 프론트엔드 IR 평가 워크플로우화 -**파일**: `rb8001/app/router/ir_deck.py` -- 프론트엔드 요청을 LangGraph 워크플로우로 처리 -- 신규 워크플로우 생성 (HITL 없음) -- 공통 함수: `extract_ir_metrics()`, `evaluate_ir_deck()`, `save_evaluation()` +**목표**: `asyncio.create_task` 비동기 실행 → LangGraph StateGraph 워크플로우 -## Phase 3: 자동 상태 복구 +**파일**: +- 신규: `rb8001/app/services/workflows/ir_deck_workflow.py` +- 수정: `rb8001/app/router/ir_deck.py` -**파일**: `rb8001/app/services/workflows/coldmail_workflow.py` -- LangGraph 1.0 자동 상태 관리 활용 -- 체크포인터 수동 설정 제거 +### 2.1 워크플로우 설계 -## 필요 작업 +**상태 모델** (`IRDeckState`): +```python +class IRDeckState(TypedDict): + document_id: str + team_id: str + evaluation_id: Optional[str] + full_text: Optional[str] # _get_full_text 결과 + page_count: Optional[int] + evaluation_result: Optional[Dict] # _evaluate_comprehensive 결과 + page_evaluations: List[Dict] # _analyze_pages 결과 + question: Optional[str] # chat 요청 시 + answer: Optional[str] # chat 응답 +``` -1. `requirements.txt`: `langgraph==0.6.10` → `langgraph>=1.0.0` -2. `coldmail_workflow.py` API 호환성 테스트 -3. HITL 패턴 적용 (콜드메일 "분석할까요?" 버튼) -4. 프론트엔드 IR 평가 워크플로우화 -5. 체크포인트 데이터 마이그레이션 스크립트 작성 +**노드 구성**: +1. `extract_node`: `_get_full_text()` 호출 → `full_text`, `page_count` 저장 +2. `evaluate_node`: `_evaluate_comprehensive()` 호출 → `evaluation_result` 저장 +3. `analyze_pages_node`: `_analyze_pages()` 호출 → `page_evaluations` 저장 +4. `save_node`: `repository.create_evaluation()` 호출 → `evaluation_id` 저장 +5. `answer_node` (조건부): `question` 있을 때 RAG 검색 + LLM 답변 → `answer` 저장 +**라우팅**: +- `extract` → `evaluate` → `analyze_pages` → `save` → `answer` (question 있으면) → `END` -## 참고 +### 2.2 작업 내용 + +1. `ir_deck_workflow.py` 신규 생성: + - `IRDeckState` 정의 + - 5개 노드 함수 구현 (`extract_node`, `evaluate_node`, `analyze_pages_node`, `save_node`, `answer_node`) + - `create_ir_deck_workflow()` 함수로 그래프 생성 (체크포인터 옵션) +2. `ir_deck.py:216-250` 수정: + - `asyncio.create_task` 제거 + - `create_ir_deck_workflow()` 호출 → `workflow.ainvoke()` 실행 + - 체크포인터 사용 (서버 재시작 시 재개 지원) +3. `ir_deck_analyzer.py` 수정: + - `analyze()` 메서드는 워크플로우로 이동, 기존 메서드는 호환성 유지용으로 유지 (deprecated) + +### 2.3 테스트 + +- `/api/ir-deck/evaluate` 호출 → 워크플로우 실행 → `evaluation_id` 반환 +- `question` 파라미터 있을 때: `answer_node` 실행되어 `answer` 반환 +- 체크포인터 재개 테스트: 서버 재시작 후 동일 `thread_id`로 재개 + +--- + +## Phase 3: Durable State 자동 재개 강화 + +**목표**: 서버 재시작 시 워크플로우 자동 재개 로직 추가 + +**현재 상태**: +- `AsyncSqliteSaver` 사용 중 (Durable State 지원) +- 서버 재시작 시 체크포인트는 유지되나, 재개 로직은 수동 (`coldmail_service.py`에서 재개) + +### 3.1 작업 내용 + +**"수동 설정 제거" 의미 명확화**: +- 체크포인터는 여전히 설정 필요 (`AsyncSqliteSaver.from_conn_string()`) +- 변경 사항: 서버 시작 시 미완료 워크플로우 자동 감지 및 재개 로직 추가 + +**구현 방법**: +1. `rb8001/app/services/workflow_recovery_service.py` 신규 생성: + - `recover_incomplete_workflows()`: 체크포인터 DB에서 `waiting_confirmation` 상태인 워크플로우 조회 + - `resume_workflows()`: 조회된 워크플로우 재개 (선택적 - 사용자 확인 대기 중인 것만) +2. `main.py` 수정: + - 서버 시작 시 `recover_incomplete_workflows()` 호출 (선택적 - 콜드메일은 사용자 확인 대기 중이므로 자동 재개 불필요, IR 평가는 가능) + +**참고**: 콜드메일은 사용자 확인 대기 중이므로 자동 재개하지 않음. IR 평가는 백그라운드 작업이므로 재개 가능. + +### 3.2 체크포인터 설정 정리 + +**현재**: +- `coldmail_briefing.py:103`: `AsyncSqliteSaver.from_conn_string(db_path)` +- `coldmail_service.py:121`: 동일 + +**변경 없음**: 체크포인터 경로는 환경변수 (`LANGGRAPH_STATE_DIR`, `LANGGRAPH_SQLITE_FILE`)로 관리되므로 수정 불필요. + +--- + +## 필요 작업 요약 + +1. **버전 업그레이드**: `requirements.txt` - `langgraph==0.6.10` → `langgraph>=1.0.0` +2. **Phase 1**: `coldmail_workflow.py` - `interrupt_after` → `interrupt()` 동적 인터럽트 마이그레이션 +3. **Phase 1**: `coldmail_service.py` - `Command(resume=True)` 재개 로직 추가 +4. **Phase 2**: `ir_deck_workflow.py` 신규 생성 - IR 평가 워크플로우 구현 +5. **Phase 2**: `ir_deck.py` - `asyncio.create_task` → LangGraph 워크플로우 전환 +6. **Phase 3**: `workflow_recovery_service.py` 신규 생성 - 서버 재시작 시 미완료 워크플로우 자동 재개 (IR 평가만) +7. **테스트**: 각 Phase별 단위 테스트 및 통합 테스트 + +--- + +## 참고 문서 - LangGraph 공식 문서: https://github.com/langchain-ai/langgraph +- LangGraph 1.0 마이그레이션 가이드: https://docs.langchain.com/oss/python/migrate/langgraph-v1 - 현재 구현: `rb8001/app/services/workflows/coldmail_workflow.py` -- 관련 문서: `troubleshooting/251015_claude_coldmail_workflow_langgraph_test.md` +- 관련 troubleshooting: + - `troubleshooting/251015_claude_coldmail_workflow_langgraph_test.md` + - `troubleshooting/251223_coldmail_langgraph_interrupt_and_checkpoint_fix.md`