Update app.py: Change logging level from INFO to WARNING

CodeWeaver

@@ -1 +0,0 @@
-Subproject commit fc4c811e94059981ae4ef7924c9aed6ccc9cbc44

CodeWeaver/.env.example

@@ -0,0 +1,9 @@
+GOOGLE_API_KEY=your-google-api-key
+TAVILY_API_KEY=your-tavily-api-key
+QDRANT_URL=https://your-qdrant-endpoint
+QDRANT_API_KEY=your-qdr
+LANGCHAIN_TRACING_V2=true
+LANGCHAIN_API_KEY=your_langsmith_api_key_here
+LANGCHAIN_PROJECT=codeweaver
+LANGCHAIN_ENDPOINT=https://api.smith.langchain.com
+GITHUB_TOKEN=

CodeWeaver/.gitignore

@@ -0,0 +1,23 @@
+# Python-generated files
+__pycache__/
+*.py[oc]
+build/
+dist/
+wheels/
+*.egg-info
+
+# Virtual environments
+.venv
+
+# Environment variables (민감한 정보 포함)
+.env
+
+# IDE
+.vscode/
+.idea/
+*.swp
+*.swo
+
+# OS
+.DS_Store
+Thumbs.db

CodeWeaver/.python-version

@@ -0,0 +1 @@
+3.12

CodeWeaver/IMPLEMENTATION_REPORT.md

@@ -0,0 +1,175 @@
+# CodeWeaver Phase 3 구현 완료 보고서
+
+## 완료 날짜
+2024-12-20
+
+## 구현 목표
+Open Deep Research 패턴을 적용하여 검색 품질과 답변 정확도를 향상
+
+## 구현된 기능
+
+### 1. 항상 질문 분해 (create_plan_node)
+- **위치**: `src/agent/nodes.py:203-287`
+- **동작**: 모든 질문을 1-5개의 서브 질문으로 분해
+- **전략**: 
+  - 단순 질문 → 1개 서브 질문
+  - 복잡 질문 → 3-5개 서브 질문
+- **LLM 사용**: JSON 구조화된 출력
+
+### 2. 검색 결과 수집 (collect_results_node)
+- **위치**: `src/agent/nodes.py:461-479`
+- **역할**: Fan-in 포인트, 3개 병렬 검색 노드의 결과 집계
+- **출력**: `len(search_results)` 기준으로 원시 결과 수 평가 (필드 저장 제거)
+
+### 3. 검색 결과 평가 (evaluate_results_node)
+- **위치**: `src/agent/nodes.py:482-533`
+- **임계값**: 2개 미만이면 개선 필요
+- **안전장치**: refinement_count >= 1이면 무조건 진행
+- **출력**: `needs_refinement` (boolean)
+
+### 4. 스마트 쿼리 개선 (refine_search_node)
+- **위치**: `src/agent/nodes.py:536-633`
+- **전략 선택** (LLM):
+  - MORE_SPECIFIC: 기술적 세부사항 추가
+  - MORE_GENERAL: 더 넓은 용어 사용
+  - TRANSLATE: 언어 변환
+- **원본 보존**: `original_question` 필드에 저장
+
+### 5. 그래프 재구성
+- **위치**: `src/agent/graph.py:200-330`
+- **새로운 엣지**:
+  - `check_cache` → `create_plan` (캐시 미스 시)
+  - `create_plan` → `classify_intent`
+  - `search_*` → `collect_results` (fan-in)
+  - `collect_results` → `evaluate_results`
+  - `evaluate_results` → `refine_search` or `search_subgraph`
+  - `refine_search` → `classify_intent` (루프)
+
+### 6. 상태 스키마 확장
+- **위치**: `src/agent/state.py:127-143`
+- **추가 필드**:
+  ```python
+  plan: Optional[Dict[str, Any]]
+  needs_refinement: bool
+  refinement_count: int
+  original_question: Optional[str]
+  ```
+
+## 테스트 결과
+
+### 통합 테스트 (test_new_features.py)
+- ✅ 테스트 1: 단순 질문 - 정상 동작
+- ✅ 테스트 2: 복잡 질문 - 정상 동작
+- ✅ 테스트 3: 결과 부족 시나리오 - 쿼리 개선 확인
+- ✅ 테스트 4: 개선 제한 - 최대 1회 보장
+
+### 실행 통계
+```
+[PASS] Passed: 4/4
+[FAIL] Failed: 0/4
+[SUCCESS] All tests passed!
+```
+
+### 실제 동작 검증
+```
+INFO:src.agent.nodes:질문 분해 계획 수립 중
+INFO:src.agent.nodes:계획 수립 완료: 4개 서브 질문
+INFO:src.agent.nodes:검색 결과 수집 완료: 0개
+INFO:src.agent.nodes:검색 결과 평가: 0개 (개선 횟수: 0)
+INFO:src.agent.nodes:쿼리 개선 중
+INFO:src.agent.nodes:쿼리 개선 완료
+INFO:src.agent.nodes:검색 결과 수집 완료: 11개
+INFO:src.agent.nodes:검색 결과 평가: 11개 (개선 횟수: 1)
+```
+
+## 준수한 LangGraph 공식 가이드라인
+
+### 1. 노드는 한 가지 일만 수행 ✅
+- 각 노드가 단일 책임 원칙 준수
+- `create_plan`: 질문 분해만
+- `evaluate_results`: 평가만 (라우팅 X)
+
+### 2. 상태에 원시 데이터 저장 ✅
+- 포맷된 텍스트 X
+- 계산 가능한 값 X
+- 순수 데이터만 저장
+
+### 3. 프롬프트는 노드 내에서 생성 ✅
+- 상태에 프롬프트 템플릿 저장 X
+- 각 노드에서 동적 생성
+
+### 4. Send API로 병렬 실행 ✅
+- 3개 검색 노드 동시 실행
+- reducer로 자동 머지
+
+### 5. 체크포인팅 지원 ✅
+- 모든 노드 경계에서 상태 저장
+- 언제든 재개 가능
+
+## 성능 개선 지표
+
+### 검색 품질
+- Before: 단일 검색 → 결과 0개 시 실패
+- After: 자동 개선 → 재검색 → 성공률 ↑
+
+### 답변 정확도  
+- Before: 모호한 검색어 → 부적절한 결과
+- After: 질문 분해 + 쿼리 개선 → 정확도 ↑
+
+### 안정성
+- Before: 무한 루프 가능성
+- After: refinement_count 제한으로 보장
+
+## 파일 변경 요약
+
+### 수정된 파일 (3개)
+1. `src/agent/state.py` - 5개 필드 추가
+2. `src/agent/nodes.py` - 4개 노드 추가/수정
+3. `src/agent/graph.py` - 엣지 재구성, 2개 라우팅 함수 추가
+
+### 추가된 파일 (3개)
+1. `test_new_features.py` - 통합 테스트
+2. `PHASE3_CHANGES.md` - 변경사항 문서
+3. `demo_phase3.py` - 데모 스크립트
+
+### 수정된 문서 (1개)
+1. `README.md` - Phase 3 섹션 추가
+
+## 코드 통계
+- 추가된 라인: ~500줄
+- 수정된 라인: ~50줄
+- 테스트 커버리지: 4개 시나리오
+
+## 다음 단계 제안
+
+### 단기 (1-2주)
+1. 서브 질문별 병렬 검색 구현
+2. 적응형 임계값 (질문 복잡도 기반)
+3. UI에 계획 수립 단계 표시
+
+### 중기 (1-2개월)
+1. 개선 전략 학습 시스템
+2. 다단계 개선 (최대 2-3회)
+3. 성능 모니터링 대시보드
+
+### 장기 (3-6개월)
+1. 다국어 지원 강화
+2. 도메인별 전문화
+3. 사용자 피드백 기반 개선
+
+## 알려진 제한사항
+
+1. **캐시 우선순위**: 캐시 히트 시 계획 수립 건���뜀 (의도된 동작)
+2. **Windows 콘솔**: 이모지 인코딩 이슈 (로직은 정상)
+3. **GitHub API**: 일부 쿼리에서 422 에러 (외부 API 제약)
+
+## 결론
+
+✅ Open Deep Research 패턴 성공적으로 적용  
+✅ 모든 테스트 통과  
+✅ LangGraph 공식 가이드라인 준수  
+✅ 기존 기능 완벽 호환  
+
+Phase 3 구현이 완료되었으며, 프로덕션 배포 준비가 완료되었습니다.
+
+

CodeWeaver/PHASE3_CHANGES.md

@@ -0,0 +1,142 @@
+# Phase 3: Open Deep Research 패턴 적용
+
+## 개요
+
+CodeWeaver에 [Open Deep Research](https://github.com/langchain-ai/open_deep_research) 패턴을 적용하여 검색 품질과 답변 정확도를 향상시켰습니다.
+
+## 변경된 파일
+
+### 1. `src/agent/state.py`
+**추가된 필드:**
+```python
+# Planning & Refinement (Phase 3)
+plan: Optional[Dict[str, Any]]              # 질문 분해 계획
+needs_refinement: bool                       # 쿼리 개선 필요 여부
+needs_refinement: bool                       # 쿼리 개선 필요 여부
+refinement_count: int                        # 개선 시도 횟수 (최대 1회)
+original_question: Optional[str]             # 원본 질문 보존
+```
+
+### 2. `src/agent/nodes.py`
+**추가된 노드 (4개):**
+- `create_plan_node`: 모든 질문을 서브 질문으로 분해
+- `collect_results_node`: 병렬 검색 결과 수집 (fan-in)
+- `evaluate_results_node`: 결과 수 평가 (< 2개면 개선 필요)
+- `refine_search_node`: LLM 기반 쿼리 개선 (전략 선택)
+
+### 3. `src/agent/graph.py`
+**수정된 라우팅:**
+- `route_after_cache`: 캐시 미스 시 → `create_plan` (기존: `classify_intent`)
+- `route_after_evaluation`: 새로운 라우팅 함수 추가
+  - 결과 부족 & refinement_count=0 → `refine_search`
+  - 결과 충분 or refinement_count=1 → `search_subgraph`
+
+**추가된 엣지:**
+- `create_plan` → `classify_intent`
+- `search_*` → `collect_results` (fan-in)
+- `collect_results` → `evaluate_results`
+- `evaluate_results` ⟲ `refine_search` → `classify_intent` (루프)
+
+## 새로운 워크플로우
+
+### Before (Phase 2)
+```
+check_cache → classify_intent → parallel_search → search_subgraph → generate_answer
+```
+
+### After (Phase 3)
+```
+check_cache → create_plan → classify_intent → parallel_search 
+    → collect_results → evaluate_results
+        ├─ < 2 results → refine_search ⟲ classify_intent (최대 1회)
+        └─ >= 2 results → search_subgraph → generate_answer
+```
+
+## 핵심 설계 원칙 (LangGraph 공식 가이드라인)
+
+### 1. 노드는 한 가지 일만 수행
+✅ `create_plan`: 질문 분해만  
+✅ `collect_results`: 결과 수집만  
+✅ `evaluate_results`: 평가만 (라우팅 X)  
+✅ `refine_search`: 쿼리 개선만  
+
+### 2. 라우팅은 conditional_edges에서
+```python
+graph.add_conditional_edges(
+    "evaluate_results",
+    route_after_evaluation,  # 라우팅 함수
+    {
+        "refine_search": "refine_search",
+        "search_subgraph": "search_subgraph"
+    }
+)
+```
+
+### 3. 상태에는 원시 데이터만 저장
+```python
+# ✅ Good: 원시 데이터
+search_results: list[SearchResult]
+needs_refinement: bool
+
+# ❌ Bad: 계산된 값이나 포맷된 텍스트
+formatted_prompt: str
+```
+
+### 4. 프롬프트는 노드 내에서 동적 생성
+```python
+def refine_search_node(state: AgentState) -> dict:
+    # ✅ 노드 내에서 동적으로 프롬프트 구성
+    refinement_prompt = f"""
+    Original question: {state.user_question}
+    Current results: {len(state.search_results)}
+    ...
+    """
+```
+
+## 테스트 결과
+
+### 통과한 시나리오
+1. ✅ 단순 질문: 1개 서브 질문 생성 → 정상 진행
+2. ✅ 복잡 질문: 3-5개 서브 질문 생성 → 정상 진행
+3. ✅ 결과 부족: < 2개 결과 → 쿼리 개선 → 재검색
+4. ✅ 개선 제한: refinement_count 최대 1회 보장
+
+### 실행 로그 예시
+```
+INFO:src.agent.nodes:질문 분해 계획 수립 중: What is GraphQL endpoint design pattern?
+INFO:src.agent.nodes:계획 수립 완료: 4개 서브 질문
+INFO:src.agent.nodes:검색 결과 수집 완료: 0개
+INFO:src.agent.nodes:검색 결과 평가: 0개 (개선 횟수: 0)
+INFO:src.agent.nodes:쿼리 개선 중: What is GraphQL endpoint design pattern? (0개 결과)
+INFO:src.agent.nodes:쿼리 개선 완료: GraphQL API design best practices
+INFO:src.agent.nodes:검색 결과 수집 완료: 11개
+INFO:src.agent.nodes:검색 결과 평가: 11개 (개선 횟수: 1)
+```
+
+## 성능 개선
+
+### 검색 품질
+- **Before**: 단일 검색 → 결과 부족 시 실패
+- **After**: 결과 부족 시 자동 개선 → 재검색
+
+### 답변 정확도
+- **Before**: 모호한 질문 → 부정확한 검색
+- **After**: 서브 질문 분해 → 더 구체적인 검색
+
+### 안정성
+- **Before**: 무한 루프 가능성
+- **After**: refinement_count 제한으로 보장
+
+## 향후 개선 방향
+
+1. **서브 질문 병렬 검색**: 현재는 전체 질문으로 검색, 각 서브 질문별 검색으로 확장
+2. **적응형 임계값**: 현재 고정값 2개 → 질문 복잡도에 따라 동적 조정
+3. **개선 전략 학습**: LLM 선택 → 과거 성공 전략 기반 추천
+4. **다단계 개선**: 최대 1회 → 2-3회로 확장 (순환 감지 추가)
+
+## 참고 자료
+
+- [LangGraph Official Guide: Thinking in LangGraph](https://docs.langchain.com/oss/python/langgraph/thinking-in-langgraph)
+- [Open Deep Research GitHub](https://github.com/langchain-ai/open_deep_research)
+
+

CodeWeaver/PHASE5_SUBGRAPH_REFACTORING.md

@@ -0,0 +1,320 @@
+# Phase 5: 서브그래프 리팩토링 완료 보고서
+
+## 개요
+
+복잡하게 얽힌 다중 질문 처리 로직을 단순화하기 위해, **analyze_question부터 generate_answer까지를 독립된 서브그래프로 추출**하고, 부모 그래프는 계획/분기/병합만 담당하도록 구조를 개선했습니다.
+
+---
+
+## 목표 달성 여부
+
+✅ **모든 목표 달성 완료**
+
+1. ✅ 단일 질문 파이프라인을 재사용 가능한 서브그래프로 추출
+2. ✅ 부모 그래프 단순화 (orchestration만 담당)
+3. ✅ 복잡한 worker 노드 및 중복 그래프 빌더 제거
+4. ✅ 구조 명확화: 부모(orchestration) vs 자식(processing)
+
+---
+
+## 변경 사항
+
+### 1. 새로운 서브그래프: `build_single_question_subgraph()`
+
+**파일**: [`src/agent/graph.py`](src/agent/graph.py)
+
+```python
+def build_single_question_subgraph() -> StateGraph:
+    """
+    단일 질문 처리 파이프라인 서브그래프를 구성합니다.
+    
+    진입점: analyze_question (START → analyze_question)
+    출구: generate_answer 또는 generate_with_history 또는 return_cached_answer (→ END)
+    
+    흐름:
+    1. analyze_question → 질문 분석
+       - clarification: generate_with_history → END
+       - new_topic/independent: check_cache
+    2. check_cache → 캐시 확인
+       - 히트: return_cached_answer → END
+       - 미스: classify_intent
+    3. classify_intent → 병렬 검색 (Send API)
+    4. 검색 결과 수집 → 평가 → 필터링 → 요약 → 답변 생성
+    """
+```
+
+**포함 노드**:
+- analyze_question, generate_with_history
+- check_cache, return_cached_answer
+- classify_intent
+- search_stackoverflow, search_github, search_official_docs (병렬)
+- collect_results, evaluate_results, refine_search
+- search_subgraph (중첩 서브그래프: filter + summarize)
+- generate_answer
+
+---
+
+### 2. 단순화된 메인 그래프: `build_agent_graph()`
+
+**변경 전 (Phase 4)**: 60+ 개의 노드와 엣지로 복잡하게 얽힘
+
+**변경 후 (Phase 5)**: 4개의 노드만으로 단순화
+
+```python
+def build_agent_graph() -> StateGraph:
+    """
+    CodeWeaver 에이전트의 메인 그래프를 구성합니다.
+    
+    전체 흐름 (단순화됨):
+    1. START → create_plan (질문 유형 및 개수 판단)
+    2. 질문 유형에 따른 분기:
+       - single_topic: single_question_subgraph (1회) → END
+       - multiple_questions: Send API로 single_question_subgraph (2회 병렬) → combine_answers → END
+       - too_many: handle_too_many_questions → END
+    """
+    
+    graph = StateGraph(AgentState)
+    
+    # 노드 추가 (4개만!)
+    graph.add_node("create_plan", create_plan_node)
+    graph.add_node("handle_too_many_questions", handle_too_many_questions_node)
+    graph.add_node("combine_answers", combine_answers_node)
+    graph.add_node("collect_subgraph_result", collect_subgraph_result_node)
+    
+    # 서브그래프를 노드로 등록
+    single_question_subgraph = build_single_question_subgraph()
+    graph.add_node("single_question_subgraph", single_question_subgraph)
+    
+    # 간단한 엣지 구성
+    graph.add_edge(START, "create_plan")
+    graph.add_conditional_edges("create_plan", route_after_plan)
+    graph.add_edge("handle_too_many_questions", END)
+    graph.add_conditional_edges("single_question_subgraph", ...)
+    graph.add_edge("collect_subgraph_result", "combine_answers")
+    graph.add_edge("combine_answers", END)
+    
+    return graph
+```
+
+---
+
+### 3. 개선된 라우팅: `route_after_plan()`
+
+**변경 전**: `initiate_dynamic_search` 노드 → `fanout_multi_questions` 함수 → `run_single_question_worker_node` → 내부에서 별도 그래프 실행
+
+**변경 후**: Send API로 서브그래프를 직접 호출
+
+```python
+def route_after_plan(state: AgentState):
+    """
+    create_plan 결과에 따라 다음 노드를 결정합니다.
+    
+    Returns:
+        - "handle_too_many_questions": 질문 3개 이상
+        - "single_question_subgraph": 단일 주제
+        - List[Send]: 다중 질문 (2개) → 서브그래프 병렬 실행
+    """
+    plan = state.plan or {}
+    case = plan.get("case", "single_topic")
+    
+    if case == "too_many":
+        return "handle_too_many_questions"
+    elif case == "multiple_questions":
+        sub_questions = plan.get("sub_questions", [])
+        sends = []
+        for i, sq in enumerate(sub_questions):
+            child_state = state.model_copy(deep=True)
+            child_state.user_question = sq
+            child_state.is_multi_question = True
+            child_state.sub_question_index = i
+            # ... 최소 필드 설정 ...
+            sends.append(Send("single_question_subgraph", child_state))
+        return sends
+    else:
+        return "single_question_subgraph"
+```
+
+---
+
+### 4. 제거된 코드 (300+ 줄)
+
+**파일**: [`src/agent/nodes.py`](src/agent/nodes.py)
+
+#### 제거된 함수:
+- ❌ `_build_search_subgraph_local()` - graph.py의 것 사용
+- ❌ `_get_single_question_agent()` - 공식 서브그래프로 대체 (100+ 줄)
+- ❌ `run_single_question_worker_node()` - 더 이상 필요 없음
+- ❌ `initiate_dynamic_search_node()` - 단순 분기로 대체
+- ❌ `fanout_multi_questions()` - route_after_plan에 통합
+
+#### 추가된 함수:
+- ✅ `collect_subgraph_result_node()` - 서브그래프 결과를 multi_answers에 추가
+
+---
+
+## 새로운 아키텍처
+
+```mermaid
+graph TD
+    START[START] --> plan[create_plan]
+    
+    plan -->|too_many| tooMany[handle_too_many_questions]
+    plan -->|single_topic| subgraph1[single_question_subgraph]
+    plan -->|multiple_2| fanout[Send API]
+    
+    tooMany --> END
+    
+    fanout -.Send Q1.-> subgraph2[single_question_subgraph]
+    fanout -.Send Q2.-> subgraph3[single_question_subgraph]
+    
+    subgraph2 --> collect2[collect_subgraph_result]
+    subgraph3 --> collect3[collect_subgraph_result]
+    
+    collect2 --> combine[combine_answers]
+    collect3 --> combine
+    
+    combine --> END
+    subgraph1 --> END
+    
+    subgraph SingleQuestionSubgraph
+        analyze[analyze_question] --> cache[check_cache]
+        cache --> classify[classify_intent]
+        classify --> search[Parallel Search]
+        search --> collect[collect_results]
+        collect --> eval[evaluate_results]
+        eval --> filter[search_subgraph]
+        filter --> generate[generate_answer]
+    end
+```
+
+---
+
+## 개선 효과
+
+### 1. 코드 품질
+- ✅ **300+ 줄 제거**: 중복 그래프 빌드 로직 완전 삭제
+- ✅ **재사용성 향상**: 단일 질문 파이프라인을 독립된 서브그래프로 캡슐화
+- ✅ **유지보수성 향상**: 역할 분리 명확 (orchestration vs processing)
+
+### 2. 구조 명확화
+- **부모 그래프 (orchestration)**:
+  - 질문 유형 판단
+  - 분기 결정
+  - 결과 병합
+  
+- **자식 서브그래프 (processing)**:
+  - 질문 분석
+  - 캐시 확인
+  - 검색 실행
+  - 답변 생성
+
+### 3. 확장성
+- ✅ 질문 3개 이상도 쉽게 대응 가능 (Send 리스트만 확장)
+- ✅ 서브그래프 단위로 독립 테스트 가능
+- ✅ 디버깅 용이: 특정 질문 문제 시 해당 서브그래프만 확인
+
+---
+
+## 검증 결과
+
+### 구조 검증
+```
+============================================================
+Phase 5: 서브그래프 리팩토링 구조 검증
+============================================================
+✅ graph.py 구문 검증 성공
+
+[필수 함수 검증]
+  ✅ build_search_subgraph
+  ✅ build_single_question_subgraph
+  ✅ route_after_plan
+  ✅ build_agent_graph
+  ✅ create_agent
+
+[제거된 함수 검증]
+  ✅ route_after_generate - 정상 제거됨
+
+[Import 검증]
+  ✅ initiate_dynamic_search_node - import 제거됨
+  ✅ fanout_multi_questions - import 제거됨
+  ✅ run_single_question_worker_node - import 제거됨
+  ✅ collect_subgraph_result_node - import 추가됨
+
+[메인 그래프 노드 검증]
+  ✅ create_plan
+  ✅ handle_too_many_questions
+  ✅ combine_answers
+  ✅ collect_subgraph_result
+  ✅ single_question_subgraph
+
+============================================================
+nodes.py 구조 검증
+============================================================
+✅ nodes.py 구문 검증 성공
+
+[제거된 함수 검증]
+  ✅ _build_search_subgraph_local - 정상 제거됨
+  ✅ _get_single_question_agent - 정상 제거됨
+  ✅ run_single_question_worker_node - 정상 제거됨
+  ✅ initiate_dynamic_search_node - 정상 제거됨
+  ✅ fanout_multi_questions - 정상 제거됨
+
+[추가된 함수 검증]
+  ✅ collect_subgraph_result_node
+
+============================================================
+검증 결과 요약
+============================================================
+✅ 성공: graph.py 구조
+✅ 성공: nodes.py 구조
+
+🎉 모든 검증 통과! 리팩토링이 성공적으로 완료되었습니다.
+```
+
+---
+
+## 변경된 파일 목록
+
+1. **src/agent/graph.py**
+   - ✅ `build_single_question_subgraph()` 추가 (100+ 줄)
+   - ✅ `route_after_plan()` 개선
+   - ✅ `build_agent_graph()` 단순화 (200+ 줄 → 50 줄)
+   - ✅ `route_after_generate()` 제거
+   - ✅ Import 정리
+
+2. **src/agent/nodes.py**
+   - ✅ `collect_subgraph_result_node()` 추가
+   - ❌ `_build_search_subgraph_local()` 제거
+   - ❌ `_get_single_question_agent()` 제거 (100+ 줄)
+   - ❌ `run_single_question_worker_node()` 제거
+   - ❌ `initiate_dynamic_search_node()` 제거
+   - ❌ `fanout_multi_questions()` 제거
+
+3. **hf-space/CodeWeaver/src/agent/**
+   - ✅ graph.py 동기화 완료
+   - ✅ nodes.py 동기화 완료
+
+---
+
+## 다음 단계
+
+이 리팩토링으로 **Phase 5**가 완료되었으며, 다음 개선 사항을 고려할 수 있습니다:
+
+1. **질문 3개 이상 지원**: `route_after_plan()`에서 Send 리스트만 확장
+2. **서브그래프 단위 테스트**: 독립된 파이프라인 검증
+3. **캐싱 전략 ���선**: 서브그래프 결과 캐싱
+4. **성능 최적화**: 병렬 실행 효율성 분석
+
+---
+
+## 결론
+
+✅ **모든 목표 달성**
+- 단일 질문 파이프라인을 재사용 가능한 서브그래프로 추출
+- 부모 그래프는 orchestration만 담당 (4개 노드)
+- 300+ 줄의 중복 코드 제거
+- 구조 명확화 및 확장성 향상
+
+이 리팩토링으로 CodeWeaver의 아키텍처가 **단순하고**, **명확하며**, **확장 가능한** 구조로 개선되었습니다.
+
+

CodeWeaver/README.md

@@ -0,0 +1,118 @@
+---
+title: CodeWeaver
+emoji: 🤖
+colorFrom: blue
+colorTo: purple
+sdk: gradio
+sdk_version: "4.44.1"
+app_file: ui/app.py
+pinned: false
+license: mit
+---
+
+# CodeWeaver
+
+LangGraph 기반의 **개발자 Q&A 에이전트**입니다. 질문을 분석하고(후속/독립), **캐시(Qdrant)**를 우선 확인한 뒤 캐시 미스일 때 **3개 소스(Stack Overflow / GitHub / 공식 문서(Tavily))를 병렬 검색**해 답변을 생성합니다. 서로 독립적인 질문이 2개 들어오면 **동적으로 2개 파이프라인을 병렬 실행**해 통합 답변을 제공합니다.
+
+## 핵심 기능(현재 코드 기준)
+
+- **질문 개수 감지**: 1개(단일 주제) / 2개(독립 질문 2개) / 3개 이상(거절 안내)
+- **질문 타입 분석**: `clarification`이면 검색/캐시 없이 **대화 히스토리 기반 답변**
+- **의미적 캐싱**: Qdrant에 질문-답변을 저장하고 유사 질문을 빠르게 재사용(임계값 0.85)
+- **병렬 검색**: Stack Overflow / GitHub / Tavily(공식 문서 도메인 제한) 동시 검색
+- **검색 품질 보정**: 결과가 부족하면 **쿼리 개선을 최대 1회** 수행
+- **서브그래프 처리**: 검색 결과를 필터링/점수화 후 요약 → 최종 답변 생성
+
+## 문서
+
+- 아키텍처/동작 원리: `../ARCHITECTURE.md`
+- 다중 질문 병렬 처리 설계(배경 설명): `../DYNAMIC_PARALLEL_SEARCH.md`
+
+## 빠른 시작
+
+### 1) 설치
+
+아래는 저장소 루트가 아니라 **`CodeWeaver/` 디렉터리 기준** 예시입니다.
+
+```bash
+cd CodeWeaver
+
+# uv 사용(권장)
+uv sync
+
+# 또는 pip 사용
+pip install -r requirements.txt
+```
+
+> `sentence-transformers`가 최초 실행 시 `BAAI/bge-m3` 모델을 다운로드할 수 있습니다(네트워크 필요).
+
+### 2) 환경 변수 설정(.env)
+
+`CodeWeaver/.env` 파일을 만들고 아래를 설정하세요(필수/선택 구분).
+
+```bash
+# 필수: Gemini (LLM)
+GOOGLE_API_KEY=your_google_api_key
+
+# 필수: Tavily (공식 문서 검색)
+TAVILY_API_KEY=your_tavily_api_key
+
+# 필수: Qdrant Cloud (캐시)
+QDRANT_URL=https://xxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx.us-east-1-0.aws.cloud.qdrant.io
+QDRANT_API_KEY=your_qdrant_api_key
+
+# 선택: GitHub API rate limit 완화
+GITHUB_TOKEN=your_github_token
+
+# 선택: LangSmith 트레이싱
+LANGCHAIN_TRACING_V2=true
+LANGCHAIN_API_KEY=your_langsmith_api_key
+```
+
+### 3) 실행(Gradio UI)
+
+```bash
+cd CodeWeaver
+python ui/app.py
+```
+
+기본 주소: `http://localhost:7860`
+
+## 현재 폴더 구조
+
+```
+CodeWeaver/
+├── main.py
+├── pyproject.toml
+├── requirements.txt
+├── src/
+│   ├── agent/
+│   │   ├── graph.py         # LangGraph 메인 그래프(라우팅/병렬화)
+│   │   ├── nodes.py         # 각 노드 구현
+│   │   └── state.py         # AgentState + reducer 정의
+│   ├── tools/
+│   │   └── search_tools.py  # StackOverflow/GitHub/Tavily 검색
+│   ├── utils/
+│   │   └── tracing.py       # trace_node 데코레이터(LangSmith 연동)
+│   └── vector_db/
+│       ├── qdrant_client.py # Qdrant 캐시 관리
+│       └── local_embeddings.py # bge-m3 로컬 임베딩
+└── ui/
+    └── app.py               # Gradio UI (실제 엔트리)
+```
+
+## 동작 흐름(요약)
+
+- `START → create_plan`
+  - **3개 이상**이면 안내 메시지 반환
+  - **2개**면 각 질문을 worker에서 단일 파이프라인으로 실행 후 결합
+  - **1개**면 아래 단일 파이프라인 수행
+- 단일 파이프라인:
+  - `analyze_question`
+    - `clarification`이면 `generate_with_history`로 즉시 답변
+    - 그 외: `check_cache` → hit면 반환, miss면 `classify_intent`
+  - `classify_intent` → 3소스 병렬 검색 → `collect_results` → `evaluate_results`
+  - 필요 시 `refine_search` 1회 → 재검색
+  - `filter_and_score → summarize_results → generate_answer`(+조건부 캐시 저장)
+
+자세한 원리는 `../ARCHITECTURE.md`를 참고하세요.

CodeWeaver/main.py

@@ -0,0 +1,6 @@
+def main():
+    print("Hello from codeweaver!")
+
+
+if __name__ == "__main__":
+    main()

CodeWeaver/pyproject.toml

@@ -0,0 +1,27 @@
+[project]
+name = "codeweaver"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.12"
+dependencies = [
+    "qdrant-client",
+    "pytest",
+    "pytest-asyncio",
+    "python-dotenv",
+    "tavily-python",
+    "requests",
+    "langsmith>=0.1.0",
+    "langchain-core>=0.3.0",
+    "langchain-google-genai>=2.0.0",
+    "langgraph>=0.2.0",
+    "sentence-transformers>=3.0.0",
+    "torch>=2.0.0",
+    "gradio==4.44.1",
+]
+
+[tool.pytest.ini_options]
+pythonpath = ["."]
+markers = [
+    "slow: 실제 API 호출이 필요한 느린 테스트 (--slow 옵션으로 실행)",
+]

CodeWeaver/requirements.txt

@@ -0,0 +1,24 @@
+# LangGraph & LangChain
+langgraph>=0.2.0
+langchain-google-genai>=2.0.0
+langchain-core>=0.3.0
+langsmith>=0.2.0
+
+# Vector DB
+qdrant-client>=1.11.0
+
+# Search APIs
+tavily-python>=0.5.0
+requests>=2.31.0
+
+# Embeddings
+sentence-transformers>=3.0.0
+torch>=2.0.0
+
+# UI
+gradio==4.44.1
+
+# Utils
+python-dotenv>=1.0.0
+pydantic>=2.0.0
+

CodeWeaver/src/agent/__init__.py

@@ -0,0 +1,51 @@
+"""
+CodeWeaver 에이전트 모듈.
+
+LangGraph 기반 개발자 질문 답변 에이전트를 제공합니다.
+
+주요 컴포넌트:
+- State: 에이전트 상태 관리
+- Nodes: 개별 처리 노드
+- Graph: LangGraph 워크플로우
+"""
+
+from .state import AgentState, SearchResult
+from .graph import agent, build_agent_graph, create_agent
+from .nodes import (
+    analyze_question_node,
+    check_cache_node,
+    classify_intent_node,
+    search_stackoverflow_node,
+    search_github_node,
+    search_official_docs_node,
+    filter_and_score_node,
+    summarize_results_node,
+    generate_answer_node,
+    return_cached_answer_node,
+    generate_with_history_node,
+)
+
+__all__ = [
+    # State
+    "AgentState",
+    "SearchResult",
+    
+    # Graph
+    "agent",
+    "build_agent_graph",
+    "create_agent",
+    
+    # Nodes
+    "analyze_question_node",
+    "check_cache_node",
+    "classify_intent_node",
+    "search_stackoverflow_node",
+    "search_github_node",
+    "search_official_docs_node",
+    "filter_and_score_node",
+    "summarize_results_node",
+    "generate_answer_node",
+    "return_cached_answer_node",
+    "generate_with_history_node",
+]
+

CodeWeaver/src/agent/graph.py

@@ -0,0 +1,420 @@
+"""
+CodeWeaver LangGraph 워크플로우 구성.
+
+LangGraph 6가지 핵심 기능 완벽 구현:
+✅ Conditional Edges: 질문 유형, 캐시 여부에 따른 분기
+✅ Send API: 3개 검색 노드 병렬 실행 (fan-out/fan-in)
+✅ Subgraph: 단일 질문 처리 파이프라인 + 검색 결과 처리 파이프라인
+✅ Map-Reduce: Send API로 병렬 검색 → 결과 머지
+✅ Checkpointing: MemorySaver로 대화 상태 저장
+✅ Pydantic Typed State: 타입 안전성 보장
+"""
+
+import logging
+from typing import Literal
+
+from langgraph.checkpoint.memory import MemorySaver
+from langgraph.graph import StateGraph, START, END
+from langgraph.types import Send
+
+from src.agent.state import AgentState, WorkerState, _MULTI_ANS_RESET_TOKEN
+from src.agent.nodes import (
+    analyze_question_node,
+    check_cache_node,
+    create_plan_node,
+    classify_intent_node,
+    search_stackoverflow_node,
+    search_github_node,
+    search_official_docs_node,
+    collect_results_node,
+    evaluate_results_node,
+    refine_search_node,
+    filter_and_score_node,
+    summarize_results_node,
+    generate_answer_node,
+    return_cached_answer_node,
+    generate_with_history_node,
+    handle_too_many_questions_node,
+    combine_answers_node,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def build_search_subgraph() -> StateGraph:
+    """
+    검색 결과 처리 서브그래프를 구성합니다.
+    
+    흐름: filter_and_score → summarize_results
+    
+    이 서브그래프는 single_question_subgraph 내부에서 사용되므로
+    WorkerState를 사용하여 채널 타입 충돌을 방지합니다.
+    
+    Returns:
+        컴파일된 서브그래프
+    """
+    # 서브그래프 생성 (WorkerState 사용)
+    subgraph = StateGraph(WorkerState)
+    
+    # 노드 추가
+    subgraph.add_node("filter_and_score", filter_and_score_node)
+    subgraph.add_node("summarize_results", summarize_results_node)
+    
+    # 서브그래프 내부 흐름 정의
+    # START → filter_and_score → summarize_results → END
+    subgraph.add_edge(START, "filter_and_score")
+    subgraph.add_edge("filter_and_score", "summarize_results")
+    subgraph.add_edge("summarize_results", END)
+    
+    return subgraph.compile()
+
+
+def route_after_analysis_worker(state: WorkerState) -> Literal["generate_with_history", "check_cache"]:
+    """
+    질문 분석 결과에 따라 다음 노드를 결정합니다 (WorkerState용).
+    
+    Args:
+        state: 현재 워커 상태
+    
+    Returns:
+        - "generate_with_history": 후속 질문 → 대화 히스토리 기반 답변
+        - "check_cache": 독립 질문 → 캐시 확인
+    """
+    raw_qtype = state.question_type or "independent"
+    legacy_map = {
+        "followup": "clarification",
+        "cache_candidate": "independent",
+        "new_search": "independent",
+    }
+    question_type = legacy_map.get(raw_qtype, raw_qtype)
+
+    if question_type == "clarification":
+        return "generate_with_history"
+
+    return "check_cache"
+
+
+def route_after_cache_worker(state: WorkerState) -> Literal["return_cached_answer", "classify_intent"]:
+    """
+    캐시 히트 여부에 따라 다음 노드를 결정합니다 (WorkerState용).
+    
+    Args:
+        state: 현재 워커 상태
+    
+    Returns:
+        - "return_cached_answer": 캐시 히트 시 즉시 답변 반환
+        - "classify_intent": 캐시 미스 시 의도 분류
+    """
+    if state.cached_result:
+        return "return_cached_answer"
+    else:
+        return "classify_intent"
+
+
+def route_after_evaluation_worker(state: WorkerState) -> Literal["refine_search", "search_subgraph"]:
+    """
+    검색 결과 평가 후 다음 노드를 결정합니다 (WorkerState용).
+    
+    Args:
+        state: 현재 워커 상태
+    
+    Returns:
+        - "refine_search": 결과 부족 & 개선 횟수 0회 → 쿼리 개선
+        - "search_subgraph": 결과 충분 or 개선 횟수 1회 → 필터링 진행
+    """
+    needs_refinement = state.needs_refinement
+    refinement_count = state.refinement_count
+    
+    if needs_refinement and refinement_count < 1:
+        return "refine_search"
+    else:
+        return "search_subgraph"
+
+
+def initiate_parallel_search_worker(state: WorkerState):
+    """
+    Send API를 사용하여 3개의 검색 노드를 병렬로 실행합니다 (WorkerState용).
+    
+    Args:
+        state: 현재 워커 상태
+    
+    Returns:
+        Send 객체 리스트 (fan-out)
+    """
+    return [
+        Send("search_stackoverflow", state),
+        Send("search_github", state),
+        Send("search_official_docs", state),
+    ]
+
+
+def build_single_question_subgraph() -> StateGraph:
+    """
+    단일 질문 처리 서브그래프.
+    
+    🔧 CRITICAL: 
+    - WorkerState만 사용
+    - 부모 AgentState와 완전히 격리
+    - 출력: multi_answers 또는 final_answer만
+    """
+    # WorkerState 사용 (AgentState와 완전히 독립)
+    subgraph = StateGraph(WorkerState)
+    
+    # 노드 추가
+    subgraph.add_node("analyze_question", analyze_question_node)
+    subgraph.add_node("generate_with_history", generate_with_history_node)
+    subgraph.add_node("check_cache", check_cache_node)
+    subgraph.add_node("return_cached_answer", return_cached_answer_node)
+    subgraph.add_node("classify_intent", classify_intent_node)
+    
+    # 병렬 검색 노드
+    subgraph.add_node("search_stackoverflow", search_stackoverflow_node)
+    subgraph.add_node("search_github", search_github_node)
+    subgraph.add_node("search_official_docs", search_official_docs_node)
+    
+    # 결과 처리 노드
+    subgraph.add_node("collect_results", collect_results_node)
+    subgraph.add_node("evaluate_results", evaluate_results_node)
+    subgraph.add_node("refine_search", refine_search_node)
+    
+    # 최종 답변 생성
+    subgraph.add_node("generate_answer", generate_answer_node)
+    
+    # 중첩 서브그래프 (filter + summarize)
+    filter_summarize_subgraph = build_search_subgraph()
+    subgraph.add_node("search_subgraph", filter_summarize_subgraph)
+    
+    # ===== 엣지 구성 =====
+    
+    # 1. START → analyze_question
+    subgraph.add_edge(START, "analyze_question")
+    
+    # 2. analyze_question 결과에 따른 분기
+    subgraph.add_conditional_edges(
+        "analyze_question",
+        route_after_analysis_worker,
+        {
+            "generate_with_history": "generate_with_history",
+            "check_cache": "check_cache",
+        }
+    )
+    
+    # 3. generate_with_history → END (대화 히스토리 기반 답변)
+    subgraph.add_edge("generate_with_history", END)
+    
+    # 4. check_cache 결과에 따른 분기
+    subgraph.add_conditional_edges(
+        "check_cache",
+        route_after_cache_worker,
+        {
+            "return_cached_answer": "return_cached_answer",
+            "classify_intent": "classify_intent",
+        }
+    )
+    
+    # 5. return_cached_answer → END (캐시 히트)
+    subgraph.add_edge("return_cached_answer", END)
+    
+    # 6. classify_intent → 병렬 검색 (Send API)
+    subgraph.add_conditional_edges("classify_intent", initiate_parallel_search_worker)
+    
+    # 7. 모든 검색 노드 → collect_results (fan-in)
+    subgraph.add_edge("search_stackoverflow", "collect_results")
+    subgraph.add_edge("search_github", "collect_results")
+    subgraph.add_edge("search_official_docs", "collect_results")
+    
+    # 8. collect_results → evaluate_results
+    subgraph.add_edge("collect_results", "evaluate_results")
+    
+    # 9. evaluate_results 결과에 따른 분기
+    subgraph.add_conditional_edges(
+        "evaluate_results",
+        route_after_evaluation_worker,
+        {
+            "refine_search": "refine_search",
+            "search_subgraph": "search_subgraph",
+        }
+    )
+    
+    # 10. refine_search → classify_intent (쿼리 개선 루프)
+    subgraph.add_edge("refine_search", "classify_intent")
+    
+    # 11. search_subgraph → generate_answer
+    subgraph.add_edge("search_subgraph", "generate_answer")
+    
+    # 12. generate_answer → END
+    subgraph.add_edge("generate_answer", END)
+    
+    return subgraph.compile()
+
+
+def route_after_plan(state: AgentState):
+    """
+    create_plan 결과에 따라 다음 노드를 결정합니다.
+    
+    Returns:
+        - "handle_too_many_questions": 질문 3개 이상
+        - "single_question_subgraph": 단일 주제 (1회 실행)
+        - List[Send]: 다중 질문 (N회 병렬 실행)
+    """
+    plan = state.plan or {}
+    case = plan.get("case", "single_topic")
+    
+    if case == "too_many":
+        return "handle_too_many_questions"
+    
+    elif case == "multiple_questions":
+        # 다중 질문: Send API로 서브그래프를 여러 번 호출
+        sub_questions = plan.get("sub_questions", [])
+        messages = state.messages
+        
+        logger.info("다중 질문 처리: %d개 질문을 서브그래프로 병렬 실행", len(sub_questions))
+        
+        sends = []
+        for i, sq in enumerate(sub_questions):
+            worker_state = WorkerState(
+                processing_question=sq,
+                messages=messages,
+                
+                # 🔧 [FIX] 이름 변경된 필드로 매핑
+                worker_is_multi=True,
+                worker_idx=i,
+                worker_sub_text=sq,
+            )
+            sends.append(Send("single_question_subgraph", worker_state))
+        
+        return sends
+    
+    else:
+        # 단일 질문
+        worker_state = WorkerState(
+            processing_question=state.user_question,
+            messages=state.messages,
+            
+            # 🔧 [FIX] 기본값 매핑
+            worker_is_multi=False,
+            worker_idx=0,
+            worker_sub_text=None
+        )
+        return [Send("single_question_subgraph", worker_state)]
+
+
+def route_after_subgraph(state: AgentState) -> Literal["combine_answers", END]:
+    """
+    서브그래프 실행 후 다음 노드 결정.
+    
+    - multi_answers가 있으면: 다중 질문 모드 → combine_answers
+    - multi_answers가 없으면: 단일 질문 모드 → END
+    """
+    # multi_answers에 실제 데이터가 있는지 확인 (reset token 제외)
+    has_answers = any(
+        isinstance(item, dict) and item.get("__token__") != _MULTI_ANS_RESET_TOKEN
+        for item in state.multi_answers
+    )
+    
+    if has_answers:
+        logger.info("다중 질문 모드: combine_answers로 이동")
+        return "combine_answers"
+    else:
+        logger.info("단일 질문 모드: END로 이동")
+        return END
+
+
+def build_agent_graph() -> StateGraph:
+    """
+    CodeWeaver 에이전트의 메인 그래프를 구성합니다.
+    
+    전체 흐름 (단순화됨):
+    1. START → create_plan (질문 유형 및 개수 판단)
+    2. 질문 유형에 따른 분기:
+       - single_topic: single_question_subgraph (1회) → END
+       - multiple_questions: Send API로 single_question_subgraph (2회 병렬) → combine_answers → END
+       - too_many: handle_too_many_questions → END
+    
+    핵심 개선사항:
+    - ✅ 단일 질문 파이프라인을 재사용 가능한 서브그래프로 추출
+    - ✅ 부모 그래프는 계획/분기/병합만 담당
+    - ✅ 복잡한 worker 노드 제거
+    - ✅ 코드 중복 제거
+    - ✅ 구조 명확화: 부모(orchestration) vs 자식(processing)
+    
+    Returns:
+        구성된 StateGraph (컴파일 전)
+    """
+    # 메인 그래프 생성
+    graph = StateGraph(AgentState)
+    
+    # 노드 추가
+    graph.add_node("create_plan", create_plan_node)
+    graph.add_node("handle_too_many_questions", handle_too_many_questions_node)
+    graph.add_node("combine_answers", combine_answers_node)
+    
+    # 서브그래프를 노드로 등록
+    single_question_subgraph = build_single_question_subgraph()
+    graph.add_node("single_question_subgraph", single_question_subgraph)
+    
+    # ===== 엣지 구성 =====
+    
+    # 1. START → create_plan
+    graph.add_edge(START, "create_plan")
+    
+    # 2. create_plan → 분기
+    #    - single_topic: "single_question_subgraph" → END
+    #    - multiple_questions: List[Send("single_question_subgraph", WorkerState)] → combine_answers
+    #    - too_many: "handle_too_many_questions" → END
+    graph.add_conditional_edges("create_plan", route_after_plan)
+    
+    # 3. handle_too_many_questions → END
+    graph.add_edge("handle_too_many_questions", END)
+    
+    # 4. 🔧 FIX: single_question_subgraph의 출구를 명확히 분리
+    #    - 단일 질문 (case=single_topic): 무조건 END
+    #    - 다중 질문 (case=multiple_questions): Send API가 자동으로 combine_answers로 fan-in
+    
+    # 4-1. 단일 질문 경로: single_question_subgraph → END
+    # 4-2. 다중 질문 경로: single_question_subgraph → combine_answers (자동 fan-in)
+    
+    # 🔧 해결책: conditional edges로 분기
+    graph.add_conditional_edges(
+        "single_question_subgraph",
+        route_after_subgraph,
+        {
+            "combine_answers": "combine_answers",
+            END: END,
+        }
+    )
+    
+    # 5. combine_answers → END
+    graph.add_edge("combine_answers", END)
+    
+    return graph
+
+
+def create_agent(enable_checkpointing: bool = True):
+    """
+    CodeWeaver 에이전트를 생성하고 컴파일합니다.
+    
+    Args:
+        enable_checkpointing: 체크포인트 활성화 여부
+            - True: MemorySaver 사용 (개발/테스트용)
+            - False: 체크포인트 없이 실행 (상태 저장 불가)
+    
+    Returns:
+        컴파일된 실행 가능한 그래프
+    
+    Note:
+        프로덕션 환경에서는 MemorySaver 대신 
+        PostgresSaver, SqliteSaver 등 영구 저장소 사용 권장
+    """
+    graph = build_agent_graph()
+    
+    if enable_checkpointing:
+        # 메모리 기반 체크포인터 (프로덕션에서는 DB 사용 권장)
+        memory = MemorySaver()
+        return graph.compile(checkpointer=memory)
+    else:
+        return graph.compile()
+
+
+# 에이전트 인스턴스 생성 (모듈 임포트 시 자동 생성)
+agent = create_agent(enable_checkpointing=True)

CodeWeaver/src/agent/nodes.py

@@ -0,0 +1,1212 @@
+"""
+CodeWeaver LangGraph 노드 구현.
+
+각 노드는 AgentState 또는 WorkerState를 받아 처리하고 업데이트된 상태를 반환합니다.
+모든 노드는 LangSmith를 통해 자동으로 추적됩니다.
+"""
+
+import asyncio
+import logging
+import os
+from typing import List, Literal, Optional, Union
+
+from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
+from langchain_google_genai import ChatGoogleGenerativeAI
+from langgraph.graph import StateGraph, START, END
+from langgraph.types import Send
+
+from src.agent.state import AgentState, WorkerState, SearchResult
+from src.agent.state import _MULTI_ANS_RESET_TOKEN
+from src.tools.search_tools import (
+    search_github,
+    search_official_docs,
+    search_stackoverflow,
+)
+from src.utils.tracing import trace_node
+from src.vector_db.qdrant_client import QdrantManager
+
+logger = logging.getLogger(__name__)
+
+# LLM 초기화 (Gemini 2.5 Flash)
+llm = ChatGoogleGenerativeAI(
+    model="gemini-2.5-flash-lite",
+    temperature=0.7,
+)
+
+# Qdrant 매니저 초기화
+qdrant_manager = QdrantManager()
+
+
+# ==================== 부모 그래프 노드 (AgentState 사용) ====================
+
+@trace_node("create_plan")
+def create_plan_node(state: AgentState) -> dict:
+    """
+    질문을 분석하여 유형과 개수를 판단합니다.
+    
+    Case:
+    - single_topic: 하나의 주제 (서브그래프 1회)
+    - multiple_questions: 독립 질문 2개 (Send API로 서브그래프 2회 병렬)
+    - too_many: 독립 질문 3개 이상 (에러 메시지)
+    """
+    user_question = state.user_question
+    logger.info("질문 분석 및 계획 수립 중: %s", user_question[:50])
+
+    def _extract_question_candidates(text: str) -> List[str]:
+        """입력 문자열에서 '질문 후보'를 최대한 보수적으로 추출합니다(3개 이상 감지용)."""
+        import re
+
+        if not text:
+            return []
+
+        t = text.strip()
+        # 1) 물음표 기반 분리
+        parts = re.split(r"[??]+", t)
+        candidates = [p.strip() for p in parts if p.strip()]
+        if len(candidates) >= 2 and re.search(r"[??]", t):
+            return candidates
+
+        # 2) 줄바꿈/번호 매기기 기반
+        lines = [ln.strip() for ln in re.split(r"[\r\n]+", t) if ln.strip()]
+        numbered = []
+        for ln in lines:
+            if re.match(r"^\s*(\d+[\.\)]|[-*])\s+", ln):
+                numbered.append(re.sub(r"^\s*(\d+[\.\)]|[-*])\s+", "", ln).strip())
+        if len(numbered) >= 2:
+            return numbered
+
+        # 3) 구분자 기반(세미콜론)
+        semi = [p.strip() for p in t.split(";") if p.strip()]
+        if len(semi) >= 2:
+            return semi
+
+        return [t]
+
+    def _hard_guard_too_many(text: str) -> Optional[dict]:
+        """
+        하드 가드: 사용자가 '질문 3개 이상'을 한 번에 던진 것으로 확실한 경우,
+        LLM 분류와 무관하게 too_many로 강제합니다.
+        """
+        import re
+
+        if not text:
+            return None
+
+        # 가장 확실한 기준: 물음표가 3개 이상
+        qmarks = len(re.findall(r"[??]", text))
+        if qmarks >= 3:
+            candidates = _extract_question_candidates(text)
+            msg = "죄송합니다. 질문은 한 번에 최대 2개까지 가능합니다. 가장 중요한 2개만 골라서 다시 질문해 주세요."
+            return {
+                "case": "too_many",
+                "sub_questions": candidates,
+                "reasoning": f"물음표가 {qmarks}개로, 3개 이상의 독립 질문으로 판단했습니다.",
+                "error_message": msg,
+                "steps_note": f"⚠️ 질문 수 초과 감지(물음표 {qmarks}개) → too_many로 강제",
+            }
+
+        # 번호 매기기/리스트로 3개 이상
+        candidates = _extract_question_candidates(text)
+        if len(candidates) >= 3:
+            msg = "죄송합니다. 질문은 한 번에 최대 2개까지 가능합니다. 가장 중요한 2개만 골라서 다시 질문해 주세요."
+            return {
+                "case": "too_many",
+                "sub_questions": candidates,
+                "reasoning": f"질문 후보가 {len(candidates)}개로 감지되어 3개 이상 질문으로 판단했습니다.",
+                "error_message": msg,
+                "steps_note": f"⚠️ 질문 수 초과 감지(후보 {len(candidates)}개) → too_many로 강제",
+            }
+
+        return None
+
+    # 하드 가드(결정론적) – LLM이 잘못 분류하더라도 3개 이상이면 무조건 차단
+    hard = _hard_guard_too_many(user_question)
+    if hard:
+        steps_delta = [
+            f"📋 계획 타입: {hard['case']}",
+            f"   서브질문: {len(hard['sub_questions'])}개",
+            f"   이유: {hard['reasoning']}",
+            hard["steps_note"],
+        ]
+        logger.info("계획 수립 완료(하드 가드): too_many, %d개 서브질��", len(hard["sub_questions"]))
+        return {
+            "plan": {
+                "case": hard["case"],
+                "sub_questions": hard["sub_questions"],
+                "reasoning": hard["reasoning"],
+                "error_message": hard["error_message"],
+            },
+            "is_multi_question": False,
+            "sub_question_index": 0,
+            "sub_question_text": None,
+            "original_multi_question": None,
+            "multi_answers": [{"__token__": _MULTI_ANS_RESET_TOKEN}],
+            "intermediate_steps": steps_delta,
+        }
+    
+    plan_prompt = f"""질문을 분석하여 유형과 개수를 판단하세요.
+
+질문: {user_question}
+
+**중요**: sub_questions의 용도는 case에 따라 다릅니다!
+
+**Case 1: single_topic** (하나의 주제)
+- 예: "Spring Security JWT 인증 구현"
+  → sub_questions: ["개념", "구현", "예제"]
+  → 용도: 답변 섹션 구조 (검색은 원본 질문으로 1회만)
+  → 검색: "Spring Security JWT 인증 구현"
+
+- 예: "React hooks 완벽 가이드"
+  → sub_questions: ["hooks란", "주요 hooks", "실무 패턴"]
+  → 용도: 답변 섹션 구조
+  → 검색: "React hooks 완벽 가이드"
+
+**Case 2: multiple_questions** (여러 독립 질문, 최대 2개)
+- 예: "JWT가 뭐야? CORS는?"
+  → sub_questions: ["JWT가 뭐야?", "CORS는?"]
+  → 용도: 각 질문마다 별도 검색
+  → 검색: "JWT가 뭐야?" (1회), "CORS는?" (1회)
+
+- 예: "Docker 사용법은? Redis 설치는?"
+  → sub_questions: ["Docker 사용법은?", "Redis 설치는?"]
+  → 용도: 각 질문마다 별도 검색
+
+**Case 3: too_many** (3개 이상 질문)
+- 예: "JWT? CORS? Docker?"
+  → 너무 많아서 처리 불가
+  → error_message 제공
+
+규칙:
+- single_topic: sub_questions는 짧은 키워드/구절 (1-5개)
+- multiple_questions: sub_questions는 완전한 문장 (정확히 2개만)
+- too_many: 3개 이상이면 이 케이스로 분류
+
+다음 JSON 형식으로만 답변하세요:
+{{
+    "case": "single_topic|multiple_questions|too_many",
+    "sub_questions": [...],
+    "reasoning": "이 케이스로 판단한 이유",
+    "error_message": "..." (too_many인 경우만, 그 외는 빈 문자열)
+}}
+
+JSON 외에 다른 텍스트는 포함하지 마세요."""
+
+    try:
+        import json
+        
+        messages_to_llm = [HumanMessage(content=plan_prompt)]
+        response = llm.invoke(messages_to_llm)
+        
+        # JSON 파싱
+        response_text = response.content.strip()
+        
+        # JSON 블록 추출
+        if "```json" in response_text:
+            response_text = response_text.split("```json")[1].split("```")[0].strip()
+        elif "```" in response_text:
+            response_text = response_text.split("```")[1].split("```")[0].strip()
+        
+        plan_data = json.loads(response_text)
+        
+        case = plan_data.get("case", "single_topic")
+        sub_questions = plan_data.get("sub_questions", [user_question])
+        reasoning = plan_data.get("reasoning", "")
+        error_message = plan_data.get("error_message", "")
+
+        # LLM 결과를 받은 뒤에도 한 번 더 하드 가드 적용 (안전장치)
+        hard2 = _hard_guard_too_many(user_question)
+        if hard2:
+            case = hard2["case"]
+            sub_questions = hard2["sub_questions"]
+            reasoning = hard2["reasoning"]
+            error_message = hard2["error_message"]
+        
+        # 유효성 검증
+        if not sub_questions or len(sub_questions) == 0:
+            sub_questions = [user_question]
+            case = "single_topic"
+        
+        # multiple_questions일 때 2개 제한 강제
+        if case == "multiple_questions" and len(sub_questions) > 2:
+            sub_questions = sub_questions[:2]
+            reasoning += " (질문 수 제한: 최대 2개)"
+        
+        steps_delta = [
+            f"📋 계획 타입: {case}",
+            f"   서브질문: {len(sub_questions)}개",
+            f"   이유: {reasoning}"
+        ]
+        
+        logger.info("계획 수립 완료: %s, %d개 서브질문", case, len(sub_questions))
+        
+        return {
+            "plan": {
+                "case": case,
+                "sub_questions": sub_questions,
+                "reasoning": reasoning,
+                "error_message": error_message
+            },
+            "is_multi_question": False,
+            "sub_question_index": 0,
+            "sub_question_text": None,
+            "original_multi_question": None,
+            "multi_answers": [{"__token__": _MULTI_ANS_RESET_TOKEN}],
+            "intermediate_steps": steps_delta
+        }
+        
+    except Exception as e:
+        logger.error("계획 수립 실패: %s", e, exc_info=True)
+        
+        # 기본값: 원본 질문 그대로 사용
+        steps_delta = [
+            "⚠️ 계획 수립 실패, 기본값 사용: single_topic"
+        ]
+        
+        return {
+            "plan": {
+                "case": "single_topic",
+                "sub_questions": [user_question],
+                "reasoning": "계획 수립 실패, 기본값 사용",
+                "error_message": ""
+            },
+            "is_multi_question": False,
+            "sub_question_index": 0,
+            "sub_question_text": None,
+            "original_multi_question": None,
+            "multi_answers": [{"__token__": _MULTI_ANS_RESET_TOKEN}],
+            "intermediate_steps": steps_delta
+        }
+
+
+@trace_node("handle_too_many_questions")
+def handle_too_many_questions_node(state: AgentState) -> dict:
+    """3개 이상 질문 시 안내 메시지를 반환합니다."""
+    plan = state.plan or {}
+    error_message = plan.get("error_message", "")
+    sub_questions = plan.get("sub_questions", [])
+    
+    logger.info("질문 수 초과: %d개", len(sub_questions))
+    
+    default_message = """죄송합니다. 한 번에 최대 2개의 질문까지만 처리할 수 있습니다.
+
+다음 중 하나를 선택해서 다시 질문해 주세요:
+
+1. **하나의 주제로 통합해서 질문**
+   예: "JWT 인증과 CORS 설정을 함께 구현하는 방법"
+
+2. **가장 중요한 2개 질문만 선택**
+   예: "JWT가 뭐야? 내 코드에 어떻게 적용해?"
+
+3. **질문을 나눠서 순차적으로 질문**
+   예: 먼저 "JWT가 뭐야?" 질문 → 답변 확인 → 다음 질문
+
+어떻게 도와드릴까요?"""
+    
+    final_message = error_message if error_message else default_message
+    
+    steps_delta = [
+        f"⚠️ 질문 수 초과: {len(sub_questions)}개",
+        "💬 안내 메시지 제공 (대화 계속 가능)"
+    ]
+    
+    return {
+        "final_answer": final_message,
+        "intermediate_steps": steps_delta
+    }
+
+
+@trace_node("combine_answers")
+def combine_answers_node(state: AgentState) -> dict:
+    """
+    Fan-in: 모든 Send가 완료되면 multi_answers를 조합합니다.
+    """
+    answers = state.multi_answers
+    original_question = state.original_multi_question or state.user_question
+    
+    if not answers:
+        logger.error("다중 답변이 비어있음")
+        return {
+            "final_answer": "답변 생성에 실패했습니다. 다시 시도해 주세요.",
+            "intermediate_steps": ["❌ multi_answers 비어있음"]
+        }
+    
+    # 인덱스 순으로 정렬
+    answers.sort(key=lambda x: x["index"])
+    
+    # Markdown 형식으로 조합
+    combined_parts = []
+    for ans in answers:
+        section = f"""## {ans['index']+1}. {ans['question']}
+
+{ans['answer']}"""
+        combined_parts.append(section)
+    
+    combined = "\n\n---\n\n".join(combined_parts)
+    
+    # 헤더 추가
+    header = f"# 다중 질문 답변\n\n원본 질문: {original_question}\n\n---\n\n"
+    final_combined = header + combined
+    
+    logger.info("다중 답변 조합 완료: %d개", len(answers))
+    
+    return {
+        "final_answer": final_combined,
+        "intermediate_steps": [f"✅ {len(answers)}개 답변 조합 완료"]
+    }
+
+
+# ==================== 서브그래프 노드 (WorkerState 사용) ====================
+
+@trace_node("analyze_question")
+async def analyze_question_node(state: Union[AgentState, WorkerState]) -> dict:
+    """
+    질문을 분석하여 유형을 분류하고 캐시 적격성을 판단합니다.
+    
+    🔧 FIX: 다중 질문 모드일 때는 messages를 무시하고 독립 질문으로만 분석
+    """
+    # 🔧 [FIX] WorkerState일 경우 processing_question 사용
+    if isinstance(state, WorkerState):
+        user_question = state.processing_question
+        # 🔧 [FIX] 이름 변경된 필드 사용
+        is_multi = state.worker_is_multi
+    else:
+        user_question = state.user_question
+        is_multi = getattr(state, 'is_multi_question', False)
+        
+    messages = state.messages
+    
+    # 대화 맥락 구성 (다중 질문 모드가 아닐 때만)
+    has_history = messages and len(messages) > 1 and not is_multi
+    context_info = ""
+    
+    if has_history:
+        context_info = "\n이전 대화 맥락:\n"
+        for msg in messages[-4:-1]:
+            if hasattr(msg, 'type') and hasattr(msg, 'content'):
+                role = "사용자" if msg.type == "human" else "AI"
+                context_info += f"{role}: {msg.content[:100]}\n"
+    
+    # 🔧 다중 질문 모드 강제 처리
+    if is_multi:
+        context_info = "\n⚠️ 주의: 이 질문은 다중 질문의 일부입니다. 독립적인 질문으로만 판단하세요.\n"
+    
+    analysis_prompt = f"""질문을 분석하여 유형을 분류하고, 캐시 적격성을 판단하세요.
+
+{context_info}
+현재 질문: {user_question}
+
+분류 기준:
+
+1. **clarification** (보충/형식 변경 요청)
+   - 이전 답변/대화 내용을 바탕으로 "설명 방식"을 바꾸거나 보충을 요청
+   - 예: "좀 더 쉽게 설명해줘", "예제 코드로 보여줘", "한 줄로 요약해줘"
+   - should_cache = false, canonical_question = null
+   
+2. **new_topic** (대화 중 새 개념 질문)
+   - 대화가 이어지는 중이지만, 질문 자체가 독립적으로 성립하는 '새 개념/정의/비교/사용법' 질문
+   - 예: "Event Listener는 뭐야?", "CORS가 뭐야?"
+   - should_cache = true, canonical_question 생성
+   
+3. **independent** (완전 독립 질문)
+   - 이전 대화 없이도 이해 가능한 일반 질문
+   - 예: "Spring Security가 뭐야?", "Docker Compose 사용법은?"
+   - should_cache = true, canonical_question 생성
+
+다음 JSON 형식으로만 답변하세요:
+{{
+    "question_type": "clarification|new_topic|independent",
+    "should_cache": true|false,
+    "reasoning": "분류 이유 1-2문장",
+    "canonical_question": "캐시할 정규화된 질문 (should_cache가 true인 경우에만, 아니면 null)"
+}}
+
+JSON 외에 다른 텍스트는 포함하지 마세요."""
+
+    try:
+        messages_to_llm = [HumanMessage(content=analysis_prompt)]
+        response = llm.invoke(messages_to_llm)
+        
+        import json
+        response_text = response.content.strip()
+        
+        if "```json" in response_text:
+            response_text = response_text.split("```json")[1].split("```")[0].strip()
+        elif "```" in response_text:
+            response_text = response_text.split("```")[1].split("```")[0].strip()
+        
+        analysis = json.loads(response_text)
+        
+        question_type = analysis.get("question_type", "independent")
+        should_cache = analysis.get("should_cache", False)
+        reasoning = analysis.get("reasoning", "")
+        canonical_question = analysis.get("canonical_question", user_question)
+        
+        # 유효성 검증
+        if question_type not in ["clarification", "new_topic", "independent"]:
+            question_type = "independent"
+
+        # 🔧 CRITICAL: 다중 질문 모드일 때는 무조건 independent로 강제
+        if is_multi and question_type == "clarification":
+            logger.warning("다중 질문 모드에서 clarification 감지 → independent로 강제 변경")
+            question_type = "independent"
+            should_cache = True
+            reasoning = "다중 질문 모드: 독립 질문으로 강제 분류"
+
+        # 정책 보정
+        if question_type == "clarification":
+            should_cache = False
+            canonical_question = None
+        else:
+            if canonical_question is None or (isinstance(canonical_question, str) and not canonical_question.strip()):
+                canonical_question = user_question
+        
+        steps_delta = [
+            "__RESET_STEPS__",
+            f"🔍 질문 분석: {question_type} (캐시 여부: {should_cache})",
+        ]
+        
+        return {
+            "question_type": question_type,
+            "should_cache": should_cache,
+            "analysis_reasoning": reasoning,
+            "canonical_question": canonical_question if should_cache else None,
+            "intermediate_steps": steps_delta
+        }
+        
+    except Exception as e:
+        logger.error("질문 분석 실패: %s", e, exc_info=True)
+        
+        steps_delta = [
+            "__RESET_STEPS__",
+            "⚠️ 질문 분석 실패, 기본값 사용: independent",
+        ]
+        
+        return {
+            "question_type": "independent",
+            "should_cache": True,
+            "analysis_reasoning": "분석 실패, 기본값 사용",
+            "canonical_question": user_question,
+            "intermediate_steps": steps_delta
+        }
+
+
+@trace_node("check_cache")
+async def check_cache_node(state: Union[AgentState, WorkerState]) -> dict:
+    """벡터 DB 캐시에서 유사한 질문을 검색합니다."""
+    # 🔧 [FIX] 변수 접근 수정
+    current_q = state.processing_question if isinstance(state, WorkerState) else state.user_question
+    question_for_lookup = state.canonical_question or current_q
+    logger.info("캐시 확인 중: %s", question_for_lookup[:50])
+    
+    try:
+        cached_result = await qdrant_manager.search_cache(
+            question=question_for_lookup,
+            threshold=0.85
+        )
+        
+        updates = {}
+        steps_delta: List[str] = []
+        
+        if cached_result:
+            updates["cached_result"] = cached_result
+            steps_delta.append(f"✅ 캐시 히트 (답변 길이: {len(cached_result)}자)")
+            logger.info("캐시 히트")
+        else:
+            updates["cached_result"] = None
+            steps_delta.append("❌ 캐시 미스: 새로운 검색 필요")
+            logger.info("캐시 미스")
+    
+    except Exception as e:
+        logger.error("캐시 확인 실패: %s", e, exc_info=True)
+        updates["cached_result"] = None
+        steps_delta.append(f"⚠️ 캐시 확인 오류: {str(e)}")
+    
+    updates["intermediate_steps"] = steps_delta
+    return updates
+
+
+@trace_node("return_cached_answer")
+def return_cached_answer_node(state: Union[AgentState, WorkerState]) -> dict:
+    """캐시 히트 시 저장된 답변을 반환합니다."""
+    logger.info("캐시된 답변 반환")
+    
+    cached_answer = state.cached_result
+    is_multi = isinstance(state, WorkerState) and state.worker_is_multi
+    
+    if is_multi:
+        return {
+            "multi_answers": [{
+                "index": state.worker_idx,
+                "question": state.worker_sub_text or state.processing_question,
+                "answer": cached_answer
+            }]
+        }
+    else:
+        # 🔧 [FIX] messages에 AIMessage 추가하여 히스토리 저장 보장
+        steps_delta = ["💾 캐시된 답변 반환 (검색 생략)"]
+        return {
+            "final_answer": cached_answer,
+            "messages": [AIMessage(content=cached_answer)], # 👈 핵심 수정
+            "intermediate_steps": steps_delta
+        }
+
+
+@trace_node("generate_with_history")
+async def generate_with_history_node(state: Union[AgentState, WorkerState]) -> dict:
+    """
+    대화 히스토리만 사용하여 후속 질문에 답변합니다.
+    
+    수정 사항:
+    1. 문맥 오염 방지: 바로 직전의 대화(질문+답변)만 참조하도록 슬라이싱 적용
+    2. 히스토리 저장: AIMessage 반환 추가 (대화 끊김 방지)
+    """
+    # 1. 현재 질문 추출
+    user_question = state.processing_question if isinstance(state, WorkerState) else state.user_question
+    messages_history = state.messages
+    
+    logger.info("대화 히스토리 기반 답변 생성: %s", user_question[:50])
+    
+    # 2. 대화 맥락 구성 (Context Pollution 방지)
+    context_prompt = "이전 대화를 참고하여 후속 질문에 답변하세요.\n\n"
+    
+    # [핵심] 현재 질문을 제외한 과거 기록 중 '가장 최근 2개(직전 질문+답변)'만 참조
+    prev_messages = messages_history[:-1] if messages_history else []
+    recent_context = prev_messages[-2:] if prev_messages else []
+    
+    if recent_context:
+        context_prompt += "직전 대화 내역:\n"
+        for msg in recent_context:
+            if hasattr(msg, 'type') and hasattr(msg, 'content'):
+                role = "사용자" if msg.type == "human" else "AI"
+                context_prompt += f"{role}: {msg.content}\n\n"
+    
+    context_prompt += f"현재 질문: {user_question}\n\n"
+    context_prompt += "위의 '직전 대화 내역'에만 집중하여 답변하세요. 그 외의 이전 주제나 불필요한 맥락은 언급하지 마세요."
+    
+    updates = {}
+    steps_delta: List[str] = []
+    
+    try:
+        # 3. LLM 호출
+        response = llm.invoke([HumanMessage(content=context_prompt)])
+        final_answer = response.content.strip()
+        
+        # 4. 상태 업데이트
+        is_multi = isinstance(state, WorkerState) and state.worker_is_multi
+        
+        if is_multi:
+            # 다중 질문 모드 (예외적 상황)
+            return {
+                "multi_answers": [{
+                    "index": state.worker_idx,
+                    "question": state.worker_sub_text or user_question,
+                    "answer": final_answer
+                }]
+            }
+        else:
+            # 단일 질문 모드 (정상 케이스)
+            updates["final_answer"] = final_answer
+            # [핵심] 대화 히스토리에 AI 답변을 추가하여 다음 턴에서 참조 가능하게 함
+            updates["messages"] = [AIMessage(content=final_answer)]
+            
+            steps_delta.append(f"💬 대화 히스토리 기반 답변 생성 (길이: {len(final_answer)}자)")
+            steps_delta.append("⚠️ 캐시 저장 생략 (보충 요청)")
+            
+            logger.info("대화 히스토리 기반 답변 생성 완료")
+        
+    except Exception as e:
+        logger.error("대화 히스토리 기반 답변 생성 실패: %s", e, exc_info=True)
+        
+        if is_multi:
+            return {
+                "multi_answers": [{
+                    "index": state.worker_idx,
+                    "question": state.worker_sub_text or user_question,
+                    "answer": "답변 생성에 실패했습니다. 다시 시도해 주세요."
+                }]
+            }
+        else:
+            updates["final_answer"] = "답변 생성에 실패했습니다. 다시 시도해 주세요."
+            steps_delta.append(f"❌ 답변 생성 실패: {str(e)}")
+    
+    updates["intermediate_steps"] = steps_delta
+    return updates
+
+
+@trace_node("classify_intent")
+def classify_intent_node(state: Union[AgentState, WorkerState]) -> dict:
+    """
+    LLM을 사용하여 사용자 질문의 의도를 분류합니다.
+    
+    🔧 CRITICAL:
+    - refined_question이 있으면 그것을 사용, 없으면 user_question 사용
+    - WorkerState 필드만 반환 (부모 AgentState와 충돌 방지)
+    - ❌ 절대 반환하면 안 되는 것들: user_question, messages
+    """
+    # 🔧 [FIX] 변수 접근 수정
+    current_q = state.processing_question if isinstance(state, WorkerState) else state.user_question
+    question_to_classify = state.refined_question if hasattr(state, 'refined_question') and state.refined_question else current_q
+    
+    logger.info("의도 분류 중: %s", question_to_classify[:50])
+    
+    classification_prompt = f"""질문을 다음 세 가지 의도 중 하나로 분류하세요:
+
+1. debugging: 에러 해결, 버그 수정, 문제 해결
+2. learning: 개념 학습, 원리 이해, 튜토리얼
+3. code_review: 코드 개선, 리팩토링, 베스트 프랙티스
+
+질문: {question_to_classify}
+
+반드시 debugging, learning, code_review 중 하나만 답하세요."""
+
+    updates = {}
+    steps_delta: List[str] = []
+    
+    try:
+        messages = [
+            SystemMessage(content="당신은 개발자 질문을 분류하는 전문가입니다."),
+            HumanMessage(content=classification_prompt)
+        ]
+        
+        response = llm.invoke(messages)
+        intent_raw = response.content.strip().lower()
+        
+        # 유효한 의도로 정규화
+        valid_intents = ["debugging", "learning", "code_review"]
+        intent = next((i for i in valid_intents if i in intent_raw), "learning")
+        
+        updates["detected_intent"] = intent
+        steps_delta.append(f"🎯 의도 분류: {intent}")
+        logger.info("의도 분류 완료: %s", intent)
+        
+    except Exception as e:
+        logger.error("의도 분류 실패: %s", e, exc_info=True)
+        updates["detected_intent"] = "learning"
+        steps_delta.append("⚠️ 의도 분류 실패, 기본값 사용: learning")
+    
+    updates["intermediate_steps"] = steps_delta
+    
+    # 🔧 CRITICAL: WorkerState 필드만 반환
+    # ✅ OK: detected_intent, intermediate_steps
+    # ❌ 절대 반환하면 안 됨: user_question, messages
+    return updates
+
+
+@trace_node("search_stackoverflow")
+def search_stackoverflow_node(state: Union[AgentState, WorkerState]) -> dict:
+    """Stack Overflow에서 검색을 수행합니다."""
+    # 🔧 [FIX] 변수 접근 수정
+    current_q = state.processing_question if isinstance(state, WorkerState) else state.user_question
+    question_to_use = state.refined_question if hasattr(state, 'refined_question') and state.refined_question else current_q
+    
+    intent = state.detected_intent or "learning"
+    count = 5 if intent == "debugging" else 3
+    
+    logger.info("Stack Overflow 검색 시작: %d개", count)
+    
+    try:
+        results = search_stackoverflow(question_to_use, count)
+        logger.info("Stack Overflow에서 %d개 결과 수집", len(results))
+        
+        # 🔧 FIX: intermediate_steps 제거
+        return {
+            "search_results": results,
+            # intermediate_steps 제거! (병렬 충돌 방지)
+        }
+    except Exception as e:
+        logger.error("Stack Overflow 검색 실패: %s", e)
+        return {}
+
+
+@trace_node("search_github")
+def search_github_node(state: Union[AgentState, WorkerState]) -> dict:
+    """GitHub Issues/Discussions에서 검색을 수행합니다."""
+    # 🔧 [FIX] 변수 접근 수정
+    current_q = state.processing_question if isinstance(state, WorkerState) else state.user_question
+    question_to_use = state.refined_question if hasattr(state, 'refined_question') and state.refined_question else current_q
+    
+    intent = state.detected_intent or "learning"
+    count = 5 if intent == "code_review" else 3 if intent == "learning" else 2
+    
+    logger.info("GitHub 검색 시작: %d개", count)
+    
+    try:
+        results = search_github(question_to_use, count)
+        logger.info("GitHub에서 %d개 결과 수집", len(results))
+        
+        # 🔧 FIX: intermediate_steps 제거
+        return {
+            "search_results": results,
+            # intermediate_steps 제거! (병렬 충돌 방지)
+        }
+    except Exception as e:
+        logger.error("GitHub 검색 실패: %s", e)
+        return {}
+
+
+@trace_node("search_official_docs")
+def search_official_docs_node(state: Union[AgentState, WorkerState]) -> dict:
+    """공식 문서/Tavily에서 검색을 수행합니다."""
+    # 🔧 [FIX] 변수 접근 수정
+    current_q = state.processing_question if isinstance(state, WorkerState) else state.user_question
+    question_to_use = state.refined_question if hasattr(state, 'refined_question') and state.refined_question else current_q
+    
+    intent = state.detected_intent or "learning"
+    count = 5 if intent == "learning" else 2
+    
+    logger.info("공식 문서 검색 시작: %d개", count)
+    
+    try:
+        results = search_official_docs(question_to_use, count)
+        logger.info("공식 문서에서 %d개 결과 수집", len(results))
+        
+        # 🔧 FIX: intermediate_steps 제거
+        return {
+            "search_results": results,
+            # intermediate_steps 제거! (병렬 충돌 방지)
+        }
+    except Exception as e:
+        logger.error("공식 문서 검색 실패: %s", e)
+        return {}
+
+
+@trace_node("collect_results")
+def collect_results_node(state: Union[AgentState, WorkerState]) -> dict:
+    """병렬 검색 결과를 수집하고 카운트합니다."""
+    total_results = len(state.search_results)
+    
+    logger.info("검색 결과 수집 완료: %d개", total_results)
+    
+    # 🔧 FIX: 로그만 찍고, intermediate_steps는 업데이트하지 않음
+    # (병렬 노드에서 intermediate_steps 업데이트 시 충돌 발생)
+    
+    return {}  # 빈 딕셔너리 반환 (상태 변경 없음)
+
+
+@trace_node("evaluate_results")
+def evaluate_results_node(state: Union[AgentState, WorkerState]) -> dict:
+    """검색 결과의 개수와 품질을 모두 평가합니다."""
+    search_results = state.search_results
+    refinement_count = state.refinement_count
+    
+    result_count = len(search_results)
+    
+    logger.info("검색 결과 평가: %d개 (개선 횟수: %d)", result_count, refinement_count)
+    
+    # 안전장치: 이미 1회 개선했으면 더 이상 개선하지 않음
+    if refinement_count >= 1:
+        steps_delta = [
+            f"⚠️ 최대 개선 횟수 도달 ({refinement_count}회), 현재 결과로 진행"
+        ]
+        return {
+            "needs_refinement": False,
+            "intermediate_steps": steps_delta
+        }
+    
+    # 1차 평가: 개수
+    if result_count < 2:
+        steps_delta = [
+            f"⚠️ 검색 결과 부족 ({result_count}개 < 2개), 쿼리 개선 필요"
+        ]
+        return {
+            "needs_refinement": True,
+            "intermediate_steps": steps_delta
+        }
+    
+    # 2차 평가: 품질
+    scored_results = [r for r in search_results if r.relevance_score is not None]
+    
+    if scored_results:
+        avg_score = sum(r.relevance_score for r in scored_results) / len(scored_results)
+        
+        if avg_score < 0.5:
+            steps_delta = [
+                f"⚠️ 검색 결과 품질 부족 (평균 점수: {avg_score:.2f} < 0.5), 쿼리 개선 필요"
+            ]
+            return {
+                "needs_refinement": True,
+                "intermediate_steps": steps_delta
+            }
+        
+        steps_delta = [
+            f"✅ 검색 결과 충분 ({result_count}개, 평균 점수: {avg_score:.2f}), 필터링 단계로 진행"
+        ]
+    else:
+        steps_delta = [
+            f"✅ 검색 결과 충분 ({result_count}개), 필터링 단계로 진행"
+        ]
+    
+    return {
+        "needs_refinement": False,
+        "intermediate_steps": steps_delta
+    }
+
+
+@trace_node("refine_search")
+def refine_search_node(state: Union[AgentState, WorkerState]) -> dict:
+    """
+    검색 쿼리를 개선합니다.
+    
+    🔧 CRITICAL: 
+    - user_question을 직접 업데이트하지 않고, refined_question에 저장
+    - 부모 AgentState와 충돌 방지를 위해 WorkerState 필드만 반환
+    - ❌ 절대 반환하면 안 되는 것들: user_question, messages, final_answer
+    """
+    # 🔧 [FIX] 변수 접근 수정
+    user_question = state.processing_question if isinstance(state, WorkerState) else state.user_question
+    original_question = state.original_question or user_question
+    result_count = len(state.search_results)
+    
+    logger.info("검색 쿼리 개선 중: %s (%d개 결과)", user_question[:50], result_count)
+    
+    refinement_prompt = f"""검색 결과가 부족합니다. 검색 쿼리를 개선하세요.
+
+원본 질문: {user_question}
+현재 결과 수: {result_count}개 (목표: 2개 이상)
+
+개선 전략 (하나 선택):
+1. MORE_SPECIFIC: 기술적 세부사항 추가
+2. MORE_GENERAL: 더 넓은 용어 사용
+3. TRANSLATE: 언어 변환
+
+다음 JSON 형식으로만 답변하세요:
+{{
+    "new_query": "개선된 검색 쿼리",
+    "strategy": "MORE_SPECIFIC|MORE_GENERAL|TRANSLATE",
+    "reasoning": "이 전략을 선택한 이유 1-2문장"
+}}
+
+JSON 외에 다른 텍스트는 포함하지 마세요."""
+
+    try:
+        import json
+        
+        messages_to_llm = [HumanMessage(content=refinement_prompt)]
+        response = llm.invoke(messages_to_llm)
+        
+        response_text = response.content.strip()
+        if "```json" in response_text:
+            response_text = response_text.split("```json")[1].split("```")[0].strip()
+        elif "```" in response_text:
+            response_text = response_text.split("```")[1].split("```")[0].strip()
+        
+        refinement_data = json.loads(response_text)
+        
+        new_query = refinement_data.get("new_query", user_question)
+        strategy = refinement_data.get("strategy", "MORE_GENERAL")
+        reasoning = refinement_data.get("reasoning", "")
+        
+        steps_delta = [
+            f"🔄 쿼리 개선: {strategy}",
+            f"   이전: {user_question[:50]}...",
+            f"   이후: {new_query[:50]}...",
+            f"   이유: {reasoning}"
+        ]
+        
+        logger.info("쿼리 개선 완료: %s → %s", user_question[:30], new_query[:30])
+        
+        # 🔧 CRITICAL: WorkerState 필드만 반환 (부모 AgentState와 충돌 방지)
+        return {
+            "refined_question": new_query,      # ✅ WorkerState 필드
+            "original_question": original_question,  # ✅ WorkerState 필드
+            "refinement_count": state.refinement_count + 1,  # ✅ WorkerState 필드
+            "search_results": [],  # ✅ WorkerState 필드 (reducer 있음)
+            "intermediate_steps": steps_delta  # ✅ WorkerState 필드
+            
+            # ❌ 절대 반환하면 안 되는 것들:
+            # "user_question": ...,   # 부모 AgentState와 충돌!
+            # "messages": ...,        # 부모 AgentState와 충돌!
+            # "final_answer": ...,    # 너무 이른 시점!
+        }
+        
+    except Exception as e:
+        logger.error("쿼리 개선 실패: %s", e, exc_info=True)
+        
+        fallback_query = user_question + " tutorial example"
+        
+        steps_delta = [
+            f"⚠️ 쿼리 개선 실패, 기본 전략 사용",
+            f"   이후: {fallback_query}"
+        ]
+        
+        # 🔧 CRITICAL: WorkerState 필드만 반환
+        return {
+            "refined_question": fallback_query,  # ✅ WorkerState 필드
+            "original_question": original_question,  # ✅ WorkerState 필드
+            "refinement_count": state.refinement_count + 1,  # ✅ WorkerState 필드
+            "search_results": [],  # ✅ WorkerState 필드 (reducer 있음)
+            "intermediate_steps": steps_delta  # ✅ WorkerState 필드
+        }
+
+
+@trace_node("filter_and_score")
+def filter_and_score_node(state: Union[AgentState, WorkerState]) -> dict:
+    """검색 결과를 필터링하고 관련도 점수를 매깁니다."""
+    search_results = state.search_results
+    logger.info("검색 결과 필터링 중: %d개", len(search_results))
+    
+    # 기본 필터링
+    filtered = [
+        r for r in search_results
+        if r.content and len(r.content) >= 50 and r.url
+    ]
+    
+    logger.info("기본 필터링 후: %d개 결과", len(filtered))
+    
+    # 상위 5개 결과만 LLM으로 점수 매기기
+    # 🔧 [FIX] scoring_prompt 내부에서 질문 참조 시 수정
+    current_q = state.processing_question if isinstance(state, WorkerState) else state.user_question
+    
+    for result in filtered[:5]:
+        if result.relevance_score is None:
+            try:
+                scoring_prompt = f"""질문: {current_q}
+
+검색 결과: {result.content[:500]}
+
+이 검색 결과가 질문에 얼마나 관련이 있는지 0.0에서 1.0 사이의 점수로 평가하세요.
+점수만 숫자로 답하세요. (예: 0.8)"""
+
+                response = llm.invoke([HumanMessage(content=scoring_prompt)])
+                score_str = response.content.strip()
+                result.relevance_score = float(score_str)
+                
+            except Exception as e:
+                logger.warning("점수 매기기 실패: %s", e)
+                result.relevance_score = 0.5
+    
+    # 관련도 순으로 정렬
+    filtered.sort(key=lambda r: r.relevance_score or 0, reverse=True)
+    
+    # 상위 5개만 유지
+    top_results = filtered[:5]
+    
+    subtask_results = dict(state.subtask_results)
+    subtask_results["filtered_results"] = [r.model_dump() for r in top_results]
+    
+    steps_delta = [f"✂️ 필터링 완료: {len(top_results)}개 결과 선택"]
+    
+    logger.info("필터링 완료: %d개 결과", len(top_results))
+    
+    return {
+        "subtask_results": subtask_results,
+        "intermediate_steps": steps_delta
+    }
+
+
+@trace_node("summarize_results")
+def summarize_results_node(state: Union[AgentState, WorkerState]) -> dict:
+    """필터링된 각 검색 결과를 초보 개발자가 이해하기 쉽게 요약합니다."""
+    subtask_results = state.subtask_results
+    filtered_results = subtask_results.get("filtered_results", [])
+    logger.info("검색 결과 요약 중: %d개", len(filtered_results))
+    
+    summaries = []
+    
+    for result_dict in filtered_results:
+        try:
+            summary_prompt = f"""다음 검색 결과를 초보 개발자가 이해하기 쉽게 2-3문장으로 요약하세요:
+
+출처: {result_dict['source']}
+내용: {result_dict['content'][:1000]}
+
+핵심 내용만 간단명료하게 요약하세요."""
+
+            response = llm.invoke([HumanMessage(content=summary_prompt)])
+            
+            summaries.append({
+                "source": result_dict['source'],
+                "url": result_dict['url'],
+                "summary": response.content.strip(),
+                "relevance": result_dict.get('relevance_score', 0.5)
+            })
+            
+        except Exception as e:
+            logger.error("요약 실패: %s", e)
+    
+    updated_subtask_results = dict(subtask_results)
+    updated_subtask_results["summaries"] = summaries
+    
+    steps_delta = [f"📝 요약 완료: {len(summaries)}개 결과"]
+    
+    logger.info("요약 완료: %d개", len(summaries))
+    
+    return {
+        "subtask_results": updated_subtask_results,
+        "intermediate_steps": steps_delta
+    }
+
+
+@trace_node("generate_answer")
+async def generate_answer_node(state: Union[AgentState, WorkerState]) -> dict:
+    """
+    요약된 정보를 바탕으로 최종 답변을 생성합니다.
+    
+    수정 사항:
+    1. 다중 질문 모드에서도 캐시 저장 로직이 실행되도록 순서 변경
+    2. 단일 질문 모드에서 AIMessage 반환 (히스토리 저장)
+    """
+    subtask_results = state.subtask_results
+    summaries = subtask_results.get("summaries", [])
+    intent = state.detected_intent or "learning"
+    
+    # 변수 접근
+    current_q = state.processing_question if isinstance(state, WorkerState) else state.user_question
+    
+    logger.info("최종 답변 생성 중: %s (질문: %s)", intent, current_q[:30])
+    
+    # 1. 의도별 프롬프트 템플릿
+    templates = {
+        "debugging": """다음 정보를 바탕으로 디버깅 질문에 답변하세요:
+
+질문: {question}
+
+수집된 정보:
+{summaries}
+
+답변 구조:
+1. 문제 정의
+2. 발생 원인
+3. 해결 방법 (코드 예제 포함)
+4. 주의사항
+5. 참고 자료
+
+초보 개발자도 이해할 수 있게 Markdown 형식으로 작성하세요.""",
+        
+        "learning": """다음 정보를 바탕으로 학습 질문에 답변하세요:
+
+질문: {question}
+
+수집된 정보:
+{summaries}
+
+답변 구조:
+1. 개념 설명 (간단명료)
+2. 동작 원리
+3. 예제 코드 (주석포함)
+4. 실무 활용 팁
+5. 추가 학습 자료
+
+초보 개발자도 이해할 수 있게 Markdown 형식으로 작성하세요.""",
+        
+        "code_review": """다음 정보를 바탕으로 코드 리뷰 질문에 답변하세요:
+
+질문: {question}
+
+수집된 정보:
+{summaries}
+
+답변 구조:
+1. 현재 접근 방식 분석
+2. 개선 포인트
+3. 리팩토링 예제
+4. 베스트 프랙티스
+5. 참고 패턴
+
+초보 개발자도 이해할 수 있게 Markdown 형식으로 작성하세요."""
+    }
+    
+    template = templates.get(intent, templates["learning"])
+    
+    # 2. 요약 텍스트 포맷팅
+    summaries_text = "\n\n".join([
+        f"출처: {s['source']} ({s['url']})\n요약: {s['summary']}"
+        for s in summaries
+    ])
+    
+    # 3. 이전 대화 맥락 추가 (Context Pollution 방지: 최근 1개만 참고용으로)
+    context_prefix = ""
+    messages_history = state.messages
+    if messages_history and len(messages_history) > 1:
+        # 검색 기반 답변이므로 이전 대화는 아주 최소한만 참조 (직전 1개)
+        prev_msg = messages_history[-2] if len(messages_history) >= 2 else None
+        if prev_msg:
+             context_prefix = f"이전 대화 맥락(참고): {prev_msg.content[:200]}...\n---\n"
+    
+    final_prompt = (context_prefix + template).format(
+        question=(state.original_question or current_q),
+        summaries=summaries_text
+    )
+    
+    updates = {}
+    steps_delta: List[str] = []
+    
+    try:
+        # 4. LLM 호출
+        response = llm.invoke([HumanMessage(content=final_prompt)])
+        final_answer = response.content.strip()
+        
+        # 5. 캐시 저장 로직 (DRY - 중복 방지 함수)
+        should_cache = state.should_cache if state.should_cache is not None else True
+        canonical_question = state.canonical_question
+        qtype = state.question_type or "independent"
+        question_to_cache = canonical_question or current_q
+
+        async def _try_cache_save():
+            """조건 충족 시 Qdrant에 캐시 저장"""
+            if should_cache and qtype in ["new_topic", "independent"]:
+                try:
+                    await qdrant_manager.save_to_cache(
+                        question=question_to_cache,
+                        answer=final_answer
+                    )
+                    logger.info("✅ 캐시 저장 완료: %s", question_to_cache[:30])
+                    return True
+                except Exception as cache_err:
+                    logger.error("캐시 저장 실패: %s", cache_err)
+                    return False
+            return False
+
+        # 6. 결과 반환 및 분기 처리
+        is_multi = isinstance(state, WorkerState) and state.worker_is_multi
+
+        if is_multi:
+            # [핵심] 다중 질문 모드: Return하기 '전에' 캐시 저장 시도
+            await _try_cache_save()
+            
+            logger.info("다중 질문 모드: 답변을 multi_answers에 추가")
+            return {
+                "multi_answers": [{
+                    "index": state.worker_idx,
+                    "question": state.worker_sub_text or current_q,
+                    "answer": final_answer
+                }]
+            }
+        
+        else:
+            # 단일 질문 모드
+            updates["final_answer"] = final_answer
+            # [핵심] 대화 히스토리에 AI 답변 추가
+            updates["messages"] = [AIMessage(content=final_answer)]
+            
+            # 캐시 저장 시도
+            saved = await _try_cache_save()
+            
+            if saved:
+                steps_delta.append(f"✅ 최종 답변 생성 완료 (길이: {len(final_answer)}자)")
+                steps_delta.append(f"💾 캐시 저장 완료 (질문: {question_to_cache[:50]}...)")
+            else:
+                steps_delta.append(f"✅ 최종 답변 생성 완료 (길이: {len(final_answer)}자)")
+                steps_delta.append("⚠️ 캐시 저장 생략 (독립적이지 않거나 일회성 질문)")
+                logger.info("최종 답변 생성 완료 (캐시 저장 생략)")
+            
+            updates["intermediate_steps"] = steps_delta
+            return updates
+        
+    except Exception as e:
+        logger.error("답변 생성 실패: %s", e, exc_info=True)
+        
+        is_multi = isinstance(state, WorkerState) and state.worker_is_multi
+        if is_multi:
+            return {
+                "multi_answers": [{
+                    "index": state.worker_idx,
+                    "question": state.worker_sub_text or current_q,
+                    "answer": "답변 생성에 실패했습니다. 다시 시도해 주세요."
+                }]
+            }
+        else:
+            updates["final_answer"] = "답변 생성에 실패했습니다. 다시 시도해 주세요."
+            steps_delta.append(f"❌ 답변 생성 실패: {str(e)}")
+            updates["intermediate_steps"] = steps_delta
+            return updates

CodeWeaver/src/agent/state.py

@@ -0,0 +1,141 @@
+from typing import Any, Dict, List, Optional, Literal, Tuple, Annotated
+from operator import add
+
+from pydantic import BaseModel, Field
+from langchain_core.messages import BaseMessage
+from langgraph.graph import add_messages
+
+
+_STEPS_RESET_TOKEN = "__RESET_STEPS__"
+_MULTI_ANS_RESET_TOKEN = "__RESET_MULTI_ANS__"
+
+
+def merge_intermediate_steps(old: List[str], new: List[str]) -> List[str]:
+    """intermediate_steps reducer."""
+    if not new:
+        return old
+    if new[0] == _STEPS_RESET_TOKEN:
+        return new[1:]
+    return old + new
+
+
+def merge_multi_answers(old: List[Dict[str, Any]], new: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    """multi_answers reducer."""
+    if not new:
+        return old
+    head = new[0]
+    if isinstance(head, dict) and head.get("__token__") == _MULTI_ANS_RESET_TOKEN:
+        return new[1:]
+    return old + new
+
+
+def merge_search_results(old: List["SearchResult"], new: List["SearchResult"]) -> List["SearchResult"]:
+    """
+    search_results reducer.
+    병렬 검색 노드들이 동시에 search_results를 업데이트할 수 있도록 병합 로직 제공.
+    """
+    return old + new
+
+
+class SearchResult(BaseModel):
+    """검색 도메인에서 공통으로 사용하는 단일 검색 결과 모델."""
+    source: str = Field(..., description="검색 출처")
+    content: str = Field(..., description="검색 결과의 핵심 내용")
+    url: Optional[str] = Field(default=None, description="원본 출처 URL")
+    relevance_score: Optional[float] = Field(default=None, description="관련도 점수")
+
+
+class AgentState(BaseModel):
+    """부모 그래프 전용 상태."""
+    
+    # Core fields
+    user_question: str = Field(default="", description="사용자의 원본 질문")
+    messages: Annotated[List[BaseMessage], add_messages] = Field(
+        default_factory=list,
+        description="대화 메시지 히스토리"
+    )
+    
+    # Final output
+    final_answer: Optional[str] = Field(default=None, description="최종 생성된 답변")
+    
+    # Debugging/tracing
+    intermediate_steps: Annotated[List[str], merge_intermediate_steps] = Field(
+        default_factory=list,
+        description="실행 단계별 로그"
+    )
+    
+    # Planning
+    plan: Optional[Dict[str, Any]] = Field(
+        default=None,
+        description="질문 분해 계획"
+    )
+    
+    # Multi-question handling
+    is_multi_question: bool = Field(default=False)
+    sub_question_index: int = Field(default=0)
+    sub_question_text: Optional[str] = Field(default=None)
+    original_multi_question: Optional[str] = Field(default=None)
+    multi_answers: Annotated[List[Dict[str, Any]], merge_multi_answers] = Field(
+        default_factory=list,
+        description="다중 질문의 각 답변 리스트"
+    )
+    
+    class Config:
+        arbitrary_types_allowed = True
+
+
+class WorkerState(BaseModel):
+    """
+    서브그래프 전용 상태.
+    부모 AgentState와 키 이름이 겹치지 않도록 주의해야 합니다.
+    """
+    
+    # === 입력 (부모로부터 받음) ===
+    processing_question: str = Field(default="", description="현재 처리 중인 질문")
+    messages: List[BaseMessage] = Field(default_factory=list, description="대화 히스토리")
+    
+    # 🔧 [FIX] 부모 상태와 충돌 방지를 위해 이름 변경 (worker_ 접두사)
+    worker_is_multi: bool = Field(default=False)
+    worker_idx: int = Field(default=0)
+    worker_sub_text: Optional[str] = Field(default=None)
+    
+    # === 서브그래프 내부 전용 필드 ===
+    # (이 필드들은 서브그래프 내부에서만 사용, 부모에게 전달 안 됨)
+    question_type: Optional[Literal["clarification", "new_topic", "independent"]] = None
+    should_cache: Optional[bool] = None
+    canonical_question: Optional[str] = None
+    analysis_reasoning: Optional[str] = None
+    cached_result: Optional[str] = None
+    detected_intent: Optional[Literal["debugging", "learning", "code_review"]] = None
+    
+    # 검색 결과 (병렬 업데이트 가능하도록 reducer 적용)
+    search_results: Annotated[List[SearchResult], merge_search_results] = Field(
+        default_factory=list,
+        description="병렬 검색 결과 (reducer로 자동 병합)"
+    )
+    
+    subtask_results: Dict[str, Any] = Field(default_factory=dict)
+    
+    # 쿼리 개선 (이 필드들은 refine_search_node만 업데이트)
+    needs_refinement: bool = False
+    refinement_count: int = 0
+    original_question: Optional[str] = None
+    refined_question: Optional[str] = None  # 🔧 개선된 쿼리를 별도 필드로 관리
+    
+    # 🔧 서브그래프 내부 로그 (부모에게 전달 안 됨!)
+    intermediate_steps: List[str] = Field(
+        default_factory=list,
+        description="서브그래프 내부 로그 (부모에 전달하지 않음)"
+    )
+    
+    # === 출력 (부모에게 전달될 필드) ===
+    # 이 필드들은 부모 AgentState에도 존재하며, Reducer��� 있거나 충돌이 허용되는 필드여야 함
+    final_answer: Optional[str] = None
+    
+    multi_answers: Annotated[List[Dict[str, Any]], merge_multi_answers] = Field(
+        default_factory=list,
+        description="다중 질문 답변용"
+    )
+    
+    class Config:
+        arbitrary_types_allowed = True

CodeWeaver/src/tools/__init__.py

@@ -0,0 +1,12 @@
+from .search_tools import (
+    search_github,
+    search_official_docs,
+    search_stackoverflow,
+)
+
+__all__ = [
+    "search_stackoverflow",
+    "search_github",
+    "search_official_docs",
+]
+

CodeWeaver/src/tools/search_tools.py

@@ -0,0 +1,217 @@
+import logging
+import os
+import time
+from typing import List
+
+import requests
+from tavily import TavilyClient  # type: ignore[import]
+
+from src.agent.state import SearchResult
+
+
+logger = logging.getLogger(__name__)
+
+
+def search_stackoverflow(query: str, limit: int = 3) -> List[SearchResult]:
+    """Stack Overflow에서 관련 질문을 검색한다.
+    
+    Args:
+        query: 검색 쿼리
+        limit: 반환할 최대 결과 수
+        
+    Returns:
+        SearchResult 리스트 (실패 시 빈 리스트)
+    """
+    if not query.strip():
+        logger.warning("Stack Overflow 검색: 빈 쿼리")
+        return []
+    
+    try:
+        url = "https://api.stackexchange.com/2.3/search/advanced"
+        params = {
+            "q": query,
+            "order": "desc",
+            "sort": "votes",
+            "site": "stackoverflow",
+            "pagesize": limit,
+            "filter": "withbody",
+        }
+        
+        response = requests.get(url, params=params, timeout=10)
+        response.raise_for_status()
+        
+        data = response.json()
+        items = data.get("items", [])
+        
+        results = []
+        max_score = max((item.get("score", 0) for item in items), default=1)
+        
+        for item in items:
+            title = item.get("title", "")
+            body = item.get("body", "")[:500]  # 본문 일부만 포함
+            content = f"{title}\n\n{body}"
+            
+            score = item.get("score", 0)
+            # 정규화: 0-1 범위로 변환
+            relevance = min(score / max(max_score, 1), 1.0) if max_score > 0 else 0.5
+            
+            results.append(
+                SearchResult(
+                    source="Stack Overflow",
+                    content=content,
+                    url=item.get("link"),
+                    relevance_score=relevance,
+                )
+            )
+        
+        logger.info("Stack Overflow 검색 성공: %d개 결과", len(results))
+        
+        # Rate limit 준수
+        time.sleep(1)
+        
+        return results
+        
+    except Exception as e:
+        logger.error("Stack Overflow 검색 실패: %s", e, exc_info=True)
+        return []
+
+
+def search_github(query: str, limit: int = 3) -> List[SearchResult]:
+    """GitHub에서 관련 코드를 검색한다.
+    
+    Args:
+        query: 검색 쿼리
+        limit: 반환할 최대 결과 수
+        
+    Returns:
+        SearchResult 리스트 (실패 시 빈 리스트)
+    """
+    if not query.strip():
+        logger.warning("GitHub 검색: 빈 쿼리")
+        return []
+    
+    try:
+        url = "https://api.github.com/search/code"
+        
+        # Python 코드로 제한 (언어 감지 로직은 추후 확장 가능)
+        search_query = f"{query} language:python"
+        
+        params = {
+            "q": search_query,
+            "sort": "indexed",
+            "per_page": limit,
+        }
+        
+        headers = {
+            "Accept": "application/vnd.github.v3+json",
+        }
+        
+        # GitHub 토큰이 있으면 Authorization 헤더 추가
+        github_token = os.getenv("GITHUB_TOKEN", "").strip()
+        if github_token:
+            headers["Authorization"] = f"token {github_token}"
+            logger.debug("GitHub 토큰 사용 (인증된 요청)")
+        else:
+            logger.warning(
+                "GITHUB_TOKEN이 설정되지 않음 - rate limit 제한적 (60 req/hr). "
+                "토큰 설정 시 5,000 req/hr로 증가"
+            )
+        
+        response = requests.get(url, params=params, headers=headers, timeout=10)
+        response.raise_for_status()
+        
+        data = response.json()
+        items = data.get("items", [])
+        
+        results = []
+        for item in items:
+            repo_name = item.get("repository", {}).get("full_name", "unknown")
+            path = item.get("path", "")
+            content = f"Repository: {repo_name}\nFile: {path}"
+            
+            results.append(
+                SearchResult(
+                    source="GitHub",
+                    content=content,
+                    url=item.get("html_url"),
+                    relevance_score=0.8,  # GitHub 결과는 일반적으로 높은 관련도
+                )
+            )
+        
+        logger.info("GitHub 검색 성공: %d개 결과", len(results))
+        
+        # Rate limit 준수
+        time.sleep(1)
+        
+        return results
+        
+    except requests.exceptions.HTTPError as e:
+        if e.response.status_code == 403:
+            logger.warning("GitHub API rate limit 초과")
+        elif e.response.status_code == 401:
+            logger.warning("GitHub API 인증 실패 (토큰이 없거나 잘못됨). 토큰 없이 계속 진행합니다.")
+        else:
+            logger.error("GitHub 검색 HTTP 에러: %s", e, exc_info=True)
+        return []
+    except Exception as e:
+        logger.error("GitHub 검색 실패: %s", e, exc_info=True)
+        return []
+
+
+def search_official_docs(query: str, limit: int = 3) -> List[SearchResult]:
+    """Tavily API를 사용해 공식 문서를 검색한다.
+    
+    Args:
+        query: 검색 쿼리
+        limit: 반환할 최대 결과 수
+        
+    Returns:
+        SearchResult 리스트 (실패 시 빈 리스트)
+    """
+    if not query.strip():
+        logger.warning("Official Docs 검색: 빈 쿼리")
+        return []
+    
+    api_key = os.getenv("TAVILY_API_KEY", "").strip()
+    if not api_key:
+        logger.error("TAVILY_API_KEY 환경 변수가 설정되어 있지 않습니다.")
+        return []
+    
+    try:
+        client = TavilyClient(api_key=api_key)
+        
+        response = client.search(
+            query=query,
+            search_depth="basic",
+            max_results=limit,
+            include_domains=[
+                "docs.python.org",
+                "docs.oracle.com",
+                "spring.io/guides",
+                "developer.mozilla.org",
+                "reactjs.org/docs",
+            ],
+        )
+        
+        results = []
+        for item in response.get("results", []):
+            content = item.get("content", "")
+            url = item.get("url", "")
+            score = item.get("score", 0.5)  # Tavily가 제공하는 관련도 점수
+            
+            results.append(
+                SearchResult(
+                    source="Official Docs",
+                    content=content,
+                    url=url,
+                    relevance_score=score,
+                )
+            )
+        
+        logger.info("Tavily 검색 성공: %d개 결과", len(results))
+        return results
+        
+    except Exception as e:
+        logger.error("Tavily 검색 실패: %s", e, exc_info=True)
+        return []
+

CodeWeaver/src/utils/__init__.py

@@ -0,0 +1,7 @@
+"""유틸리티 모듈."""
+
+from .tracing import ensure_tracing_enabled, trace_node
+
+__all__ = ["ensure_tracing_enabled", "trace_node"]
+
+

CodeWeaver/src/utils/tracing.py

@@ -0,0 +1,91 @@
+"""
+LangSmith 추적(tracing) 유틸리티 모듈.
+
+LangGraph 노드 실행을 LangSmith에서 추적하고 모니터링하기 위한 도구를 제공합니다.
+공식 문서: https://docs.langchain.com/langsmith/trace-with-langgraph
+"""
+
+import os
+import logging
+import asyncio
+from functools import wraps
+from typing import Any, Callable
+from inspect import iscoroutinefunction
+
+from langsmith import traceable
+
+logger = logging.getLogger(__name__)
+
+
+def ensure_tracing_enabled() -> bool:
+    """
+    LangSmith 추적이 올바르게 설정되었는지 확인합니다.
+    
+    Returns:
+        bool: 추적이 활성화되어 있으면 True, 그렇지 않으면 False
+    """
+    required_vars = ["LANGCHAIN_TRACING_V2", "LANGCHAIN_API_KEY"]
+    
+    missing_vars = [var for var in required_vars if not os.getenv(var)]
+    
+    if missing_vars:
+        logger.warning(
+            "LangSmith 추적이 비활성화되었습니다. 누락된 환경변수: %s",
+            ", ".join(missing_vars)
+        )
+        return False
+    
+    return True
+
+
+def trace_node(node_name: str) -> Callable:
+    """
+    LangGraph 노드 실행을 추적하는 데코레이터.
+    
+    이 데코레이터는 각 노드의 입력/출력, 실행 시간, 에러를 
+    LangSmith 대시보드에 자동으로 기록합니다.
+    동기 및 비동기 함수 모두 지원합니다.
+    
+    Args:
+        node_name: LangSmith에 표시될 노드 이름
+    
+    Returns:
+        Callable: 데코레이트된 함수
+    
+    Example:
+        @trace_node("check_cache")
+        async def check_cache_node(state: AgentState) -> AgentState:
+            # 노드 로직
+            return state
+    """
+    def decorator(func: Callable) -> Callable:
+        # async 함수인지 확인
+        if iscoroutinefunction(func):
+            @wraps(func)
+            @traceable(name=node_name, run_type="chain")
+            async def async_wrapper(*args, **kwargs) -> Any:
+                try:
+                    result = await func(*args, **kwargs)
+                    return result
+                except Exception as e:
+                    logger.error("🔴 노드 실패: %s - %s", node_name, str(e))
+                    raise
+            return async_wrapper
+        else:
+            @wraps(func)
+            @traceable(name=node_name, run_type="chain")
+            def sync_wrapper(*args, **kwargs) -> Any:
+                try:
+                    result = func(*args, **kwargs)
+                    return result
+                except Exception as e:
+                    logger.error("🔴 노드 실패: %s - %s", node_name, str(e))
+                    raise
+            return sync_wrapper
+    return decorator
+
+
+# 모듈 import 시 자동으로 추적 설정 확인
+ensure_tracing_enabled()
+
+

CodeWeaver/src/vector_db/__init__.py

@@ -0,0 +1,6 @@
+from .qdrant_client import QdrantManager
+from .local_embeddings import LocalEmbeddingManager
+
+__all__ = ["QdrantManager", "LocalEmbeddingManager"]
+
+

CodeWeaver/src/vector_db/local_embeddings.py

@@ -0,0 +1,34 @@
+"""
+로컬 임베딩 관리 모듈.
+
+BAAI/bge-m3 모델을 사용해 로컬에서 임베딩을 생성한다.
+"""
+
+import logging
+from typing import List
+
+from sentence_transformers import SentenceTransformer
+
+logger = logging.getLogger(__name__)
+
+
+class LocalEmbeddingManager:
+    """BAAI/bge-m3 로컬 임베딩 생성기."""
+
+    def __init__(self, model_name: str = "BAAI/bge-m3") -> None:
+        logger.info("로컬 임베딩 모델 로딩 중: %s", model_name)
+        self.model = SentenceTransformer(model_name)
+        dim = self.model.get_sentence_embedding_dimension()
+        logger.info("로컬 임베딩 모델 로딩 완료 (차원: %d)", dim)
+
+    def get_embedding(self, text: str) -> List[float]:
+        """단일 텍스트를 임베딩."""
+        embedding = self.model.encode(text, convert_to_numpy=True)
+        return embedding.tolist()
+
+    def get_embeddings_batch(self, texts: List[str]) -> List[List[float]]:
+        """배치 텍스트 임베딩."""
+        embeddings = self.model.encode(texts, convert_to_numpy=True)
+        return embeddings.tolist()
+
+

CodeWeaver/src/vector_db/qdrant_client.py

@@ -0,0 +1,225 @@
+import hashlib
+import logging
+import os
+from typing import Dict, List, Optional
+
+from dotenv import load_dotenv  # type: ignore[import]
+from qdrant_client import QdrantClient, models
+
+from src.vector_db.local_embeddings import LocalEmbeddingManager
+
+# .env 파일에서 환경 변수 로드 (로컬 개발 편의성)
+load_dotenv()
+
+logger = logging.getLogger(__name__)
+
+
+class QdrantManager:
+    """Qdrant Cloud 기반 벡터 캐시 관리 클래스.
+
+    - 임베딩 생성: 로컬 BAAI/bge-m3
+    - 벡터 저장/검색: Qdrant Cloud
+    """
+
+    def __init__(self, collection_name: str = "CodeWeaver") -> None:
+        """Qdrant Cloud 클라이언트를 초기화하고 컬렉션을 준비한다."""
+        qdrant_url = os.getenv("QDRANT_URL", "").strip()
+        qdrant_api_key = os.getenv("QDRANT_API_KEY", "").strip()
+
+        if not qdrant_url or not qdrant_api_key:
+            raise ValueError(
+                "QDRANT_URL 및 QDRANT_API_KEY 환경 변수가 모두 설정되어 있어야 합니다."
+            )
+
+        # Qdrant Cloud 공식 가이드와 유사한 초기화 형태 사용
+        # https://qdrant.tech/documentation/tutorials-and-examples/cloud-inference-hybrid-search/
+        self.client = QdrantClient(
+            url=qdrant_url,
+            api_key=qdrant_api_key,
+            timeout=30,
+        )
+
+        self.collection_name = collection_name
+        self.embedding_manager = LocalEmbeddingManager()
+
+        logger.info("QdrantManager 초기화: collection=%s, url=%s", collection_name, qdrant_url)
+
+        # 컬렉션이 없다면 생성
+        self._init_collection()
+
+    def _init_collection(self) -> None:
+        """컬렉션이 없으면 생성한다."""
+        try:
+            exists = self.client.collection_exists(self.collection_name)
+        except Exception as e:  # pragma: no cover - 방어적 코드
+            logger.error("Qdrant 컬렉션 존재 여부 확인 실패: %s", e, exc_info=True)
+            raise
+
+        if exists:
+            logger.info("Qdrant 컬렉션 이미 존재: %s", self.collection_name)
+            return
+
+        try:
+            self.client.create_collection(
+                collection_name=self.collection_name,
+                vectors_config=models.VectorParams(
+                    size=1024,  # bge-m3 임베딩 차원
+                    distance=models.Distance.COSINE,
+                ),
+            )
+            logger.info("Qdrant 컬렉션 생성 완료: %s", self.collection_name)
+        except Exception as e:
+            logger.error("Qdrant 컬렉션 생성 실패: %s", e, exc_info=True)
+            raise
+
+    async def get_embedding(self, text: str) -> List[float]:
+        """로컬 임베딩 모델을 사용해 텍스트 임베딩을 생성한다."""
+        try:
+            embedding = self.embedding_manager.get_embedding(text)
+            logger.debug("임베딩 생성 완료 (길이=%d)", len(embedding))
+            return embedding
+        except Exception as e:
+            logger.error("임베딩 생성 실패: %s", e, exc_info=True)
+            raise
+
+    async def search_cache(
+        self,
+        question: str,
+        threshold: float = 0.85,
+    ) -> Optional[str]:
+        """질문에 대한 캐시된 답변을 Qdrant에서 검색한다.
+
+        threshold보다 높은 score를 가진 결과가 있을 때만 answer를 반환한다.
+        """
+        try:
+            embedding = await self.get_embedding(question)
+        except Exception:
+            # 이미 get_embedding 내부에서 로그를 남기므로 여기서는 조용히 실패 처리
+            return None
+
+        try:
+            # Qdrant 공식 문서: query_points를 사용한 벡터 검색
+            # 단일 벡터 컬렉션의 경우 query 파라미터에 벡터 리스트를 직접 전달
+            # https://qdrant.tech/documentation/tutorials-and-examples/cloud-inference-hybrid-search/
+            results = self.client.query_points(
+                collection_name=self.collection_name,
+                query=embedding,  # 단일 벡터 컬렉션: 벡터를 직접 전달
+                limit=1,
+                with_payload=True,
+            )
+        except Exception as e:
+            logger.error("Qdrant 캐시 검색 실패: %s", e, exc_info=True)
+            return None
+
+        if not results.points:
+            logger.info("캐시 미스: 결과 없음 (question=%s)", question)
+            return None
+
+        top = results.points[0]
+        score = getattr(top, "score", None)
+        payload = getattr(top, "payload", {}) or {}
+
+        if score is None:
+            logger.warning("검색 결과에 score가 없습니다. payload=%s", payload)
+            return None
+
+        if score < threshold:
+            logger.info(
+                "캐시 미스: score(%.4f) < threshold(%.4f) (question=%s)",
+                score,
+                threshold,
+                question,
+            )
+            return None
+
+        answer = payload.get("answer")
+        if answer is None:
+            logger.info("캐시 히트이지만 payload에 answer가 없습니다. payload=%s", payload)
+            return None
+
+        logger.info(
+            "캐시 히트: score=%.4f, question=%s, answer_length=%d",
+            score,
+            question,
+            len(str(answer)),
+        )
+        return str(answer)
+
+    async def save_to_cache(self, question: str, answer: str) -> None:
+        """질문-답변 쌍을 Qdrant 캐시에 저장한다.
+        
+        동일한 질문에 대해서는 deterministic ID를 사용하여,
+        upsert 시 기존 엔트리를 덮어쓰게 함으로써 중복을 방지한다.
+        """
+        try:
+            embedding = await self.get_embedding(question)
+        except Exception:
+            # 임베딩 실패 시 캐시에 저장하지 않는다.
+            logger.warning("임베딩 실패로 인해 캐시에 저장하지 않음. question=%s", question)
+            return
+
+        # UUID 대신 질문 해시 기반 deterministic ID 사용
+        # → 동일 질문 = 동일 ID → upsert가 덮어쓰기로 동작 → 중복 방지
+        #
+        # 주의: Qdrant point id는 "unsigned int" 또는 "UUID"만 허용한다.
+        # 따라서 sha256 hex(64자)를 그대로 쓰지 않고, 앞 32자를 UUID 포맷으로 변환해 사용한다.
+        digest = hashlib.sha256(question.encode("utf-8")).hexdigest()
+        point_id = f"{digest[:8]}-{digest[8:12]}-{digest[12:16]}-{digest[16:20]}-{digest[20:32]}"
+
+        # 기존 엔트리 존재 시(덮어쓰기) 로그를 남긴다. 실패해도 upsert는 계속 시도.
+        try:
+            existing = self.client.retrieve(
+                collection_name=self.collection_name,
+                ids=[point_id],
+                with_payload=False,
+                with_vectors=False,
+            )
+            if existing:
+                logger.info("기존 캐시 엔트리를 덮어씁니다: point_id=%s", point_id)
+        except Exception:
+            pass
+
+        point = models.PointStruct(
+            id=point_id,
+            vector=embedding,
+            payload={
+                "question": question,
+                "answer": answer,
+            },
+        )
+
+        try:
+            self.client.upsert(
+                collection_name=self.collection_name,
+                points=[point],
+            )
+            logger.info(
+                "Qdrant 캐시에 저장 완료 (hash ID로 중복 방지): point_id=%s, question_length=%d, answer_length=%d",
+                point_id,
+                len(question),
+                len(answer),
+            )
+        except Exception as e:
+            logger.error("Qdrant 캐시 저장 실패: %s", e, exc_info=True)
+
+    async def get_cache_stats(self) -> Dict[str, int]:
+        """현재 컬렉션의 캐시 통계를 반환한다."""
+        try:
+            info = self.client.get_collection(self.collection_name)
+            # qdrant_client의 CollectionInfo는 points_count 속성을 제공
+            points_count = getattr(info, "points_count", 0) or 0
+            logger.debug(
+                "Qdrant 캐시 통계 조회: collection=%s, total_entries=%d",
+                self.collection_name,
+                points_count,
+            )
+            return {"total_entries": int(points_count)}
+        except Exception as e:
+            logger.error("Qdrant 캐시 통계 조회 실패: %s", e, exc_info=True)
+            # 호출 측에서 에러 메시지를 참고할 수 있도록 포함
+            return {
+                "total_entries": 0,
+                "error": str(e),  # type: ignore[dict-item]
+            }
+
+

CodeWeaver/test_result.txt

@@ -0,0 +1,56 @@
+============================================================
+Phase 5: ?쒕툕洹몃옒??由ы뙥?좊쭅 援ъ“ 寃€利?
+============================================================
+??graph.py 援щЦ 寃€利??깃났
+
+[?꾩닔 ?⑥닔 寃€利?
+  ??build_search_subgraph
+  ??build_single_question_subgraph
+  ??route_after_plan
+  ??build_agent_graph
+  ??create_agent
+
+[?쒓굅???⑥닔 寃€利?
+  ??route_after_generate - ?뺤긽 ?쒓굅??
+
+[Import 寃€利?
+  ??initiate_dynamic_search_node - import ?쒓굅??
+  ??fanout_multi_questions - import ?쒓굅??
+  ??run_single_question_worker_node - import ?쒓굅??
+  ??collect_subgraph_result_node - import 異붽???
+
+[硫붿씤 洹몃옒???몃뱶 寃€利?
+  ??create_plan
+  ??handle_too_many_questions
+  ??combine_answers
+  ??collect_subgraph_result
+  ??single_question_subgraph
+
+============================================================
+nodes.py 援ъ“ 寃€利?
+============================================================
+??nodes.py 援щЦ 寃€利??깃났
+
+[?쒓굅???⑥닔 寃€利?
+  ??_build_search_subgraph_local - ?뺤긽 ?쒓굅??
+  ??_get_single_question_agent - ?뺤긽 ?쒓굅??
+  ??run_single_question_worker_node - ?뺤긽 ?쒓굅??
+  ??initiate_dynamic_search_node - ?뺤긽 ?쒓굅??
+  ??fanout_multi_questions - ?뺤긽 ?쒓굅??
+
+[異붽????⑥닔 寃€利?
+  ??collect_subgraph_result_node
+
+============================================================
+寃€利?寃곌낵 ?붿빟
+============================================================
+???깃났: graph.py 援ъ“
+???깃났: nodes.py 援ъ“
+
+?럦 紐⑤뱺 寃€利??듦낵! 由ы뙥?좊쭅???깃났?곸쑝濡??꾨즺?섏뿀?듬땲??
+
+[媛쒖꽑 ?ы빆]
+  ???⑥씪 吏덈Ц ?뚯씠?꾨씪?몄쓣 ?ъ궗??媛€?ν븳 ?쒕툕洹몃옒?꾨줈 異붿텧
+  ??遺€紐?洹몃옒?꾨뒗 怨꾪쉷/遺꾧린/蹂묓빀留??대떦 (orchestration)
+  ??蹂듭옟??worker ?몃뱶 諛?以묐났 洹몃옒??鍮뚮뜑 ?쒓굅 (300+ 以?
+  ??援ъ“ 紐낇솗?? 遺€紐?orchestration) vs ?먯떇(processing)

CodeWeaver/ui/app.py

@@ -0,0 +1,272 @@
+import asyncio
+import logging
+import os
+import sys
+import uuid
+from pathlib import Path
+
+import gradio as gr
+from dotenv import load_dotenv
+
+# 환경 변수 로드 (에이전트/트레이싱 import 이전에 실행)
+load_dotenv()
+
+# 프로젝트 루트를 경로에 추가
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from src.agent.graph import agent
+from src.agent.state import AgentState
+
+# 로깅 설정 (WARNING 이상만 출력)
+logging.basicConfig(
+    level=logging.WARNING,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+)
+
+# 외부 라이브러리 로그는 WARNING만
+logging.getLogger("httpx").setLevel(logging.WARNING)
+logging.getLogger("httpcore").setLevel(logging.WARNING)
+logging.getLogger("langsmith").setLevel(logging.WARNING)
+
+# CodeWeaver 모듈 로그도 WARNING만 (로그 비활성화)
+logging.getLogger("src.agent").setLevel(logging.WARNING)
+logging.getLogger("src.tools").setLevel(logging.WARNING)
+logging.getLogger("src.vector_db").setLevel(logging.WARNING)
+
+logger = logging.getLogger(__name__)
+
+
+async def chat(
+    message: str,
+    history: list,
+    thread_id: str,
+) -> str:
+    """
+    사용자 메시지를 처리하고 에이전트 응답을 반환합니다.
+    
+    Args:
+        message: 사용자 입력 메시지
+        history: 대화 내역 (Gradio 자동 관리)
+        thread_id: 세션별 고유 ID (MemorySaver가 대화 맥락 추적에 사용)
+        
+    Returns:
+        에이전트의 최종 답변
+    """
+    if not message or not message.strip():
+        return "질문을 입력해주세요."
+    
+    try:
+        # 초기 상태 생성 (Pydantic BaseModel 사용)
+        from langchain_core.messages import HumanMessage
+        
+        initial_state = AgentState(
+            user_question=message.strip(),
+            messages=[HumanMessage(content=message.strip())],
+            conversation_history=history[-5:] if history else None,  # 최근 5턴만 전달
+        )
+        
+        # 세션별 thread_id를 config에 전달 (MemorySaver가 대화 맥락 유지)
+        config = {"configurable": {"thread_id": thread_id}}
+        
+        # 에이전트 실행
+        result = await agent.ainvoke(initial_state, config=config)
+        
+        # 최종 답변 추출
+        final_answer = result.get("final_answer", "답변을 생성하지 못했습니다.")
+        
+        return final_answer
+        
+    except Exception as e:
+        logger.error("에러 발생: %s", e, exc_info=True)
+        return f"⚠️ 오류가 발생했습니다: {str(e)}\n다시 시도해주세요."
+
+
+def create_demo() -> gr.Blocks:
+    """Gradio 인터페이스를 생성합니다."""
+    
+
+    
+    # CSS 스타일 (깔끔한 디자인)
+    # - Gradio 기본 CSS가 .contain/.gradio-container 폭을 덮어쓰는 경우가 있어
+    #   둘 다 !important로 고정하여 "처음부터 넓은 폭"을 확실히 유지합니다.
+    css = """
+    .gradio-container {
+        max-width: 1280px !important;
+        width: min(1280px, 100%) !important;
+        margin: 0 auto !important;
+    }
+    .contain {
+        max-width: 1280px !important;
+        width: min(1280px, 100%) !important;
+        margin: 0 auto !important;
+        padding-top: 1.5rem;
+    }
+    .message { font-size: 1.1rem; line-height: 1.6; }
+    """
+    
+    with gr.Blocks(
+        title="CodeWeaver - AI 개발 도우미",
+        theme=gr.themes.Soft(),
+        css=css
+    ) as demo:
+        
+        gr.Markdown("""
+        # 🤖 CodeWeaver
+        ### AI 기반 개발 질문 답변 시스템
+        
+        초보 개발자를 위한 친절한 AI 도우미입니다.
+        
+        **주요 기능:**
+        - ✅ 에러 해결 (디버깅)
+        - ✅ 개념 학습
+        - ✅ 코드 리뷰 및 개선 제안
+        - ✅ **다중 질문 처리** (최대 2개까지 동시 처리)
+        - ✅ **대화 맥락 이해** (이전 대화를 참고한 후속 질문 답변)
+        - ✅ **스마트 캐싱** (유사 질문 즉시 답변)
+        - ✅ **자동 검색 개선** (결과 부족 시 쿼리 자동 최적화)
+        
+        💬 개발 관련 질문을 자유롭게 해보세요!
+        - 단일 질문: "Spring Boot JPA N+1 문제 해결 방법은?"
+        - 다중 질문: "JWT가 뭐야? CORS는?" (최대 2개)
+        - 후속 질문: "좀 더 쉽게 설명해줘" (이전 답변 참고)
+        """)
+        
+        # 세션별 고유 ID (브라우저 세션마다 독립적으로 생성)
+        session_id = gr.State(value=lambda: str(uuid.uuid4()))
+        
+        # 채팅 인터페이스
+        chatbot_interface = gr.ChatInterface(
+            fn=chat,
+            examples=None,  # examples는 아래 Accordion에서 수동 처리
+            chatbot=gr.Chatbot(height=500),
+            textbox=gr.Textbox(
+                placeholder="질문을 입력하세요...",
+                container=False,
+                scale=7
+            ),
+            retry_btn=None,
+            undo_btn=None,
+            clear_btn="🗑️ 대화 초기화",
+            additional_inputs=[session_id],  # thread_id 전달
+        )
+        
+        # Clear 버튼 클릭 시 새 세션 ID 생성 (새 대화 시작)
+        def reset_session():
+            new_id = str(uuid.uuid4())
+            return new_id
+        
+        # Clear 버튼에 세션 리셋 핸들러 추가
+        if chatbot_interface.clear_btn:
+            chatbot_interface.clear_btn.click(
+                reset_session,
+                None,
+                session_id,
+                queue=False
+            )
+        
+        # 빠른 질문 버튼들 (Accordion 밖으로 분리)
+        gr.Markdown("### 💬 예시 질문")
+        example_questions = [
+            "Spring Boot JPA N+1 문제 해결 방법은?",
+            "ImportError: No module named 'requests' 해결 방법",
+            "Docker Compose 설정 예제를 알려주세요",
+            "이 코드를 개선해주세요: for i in range(len(arr)): print(arr[i])",
+            "JWT가 뭐야? CORS는?",  # 다중 질문 예시
+        ]
+        
+        with gr.Row():
+            for question in example_questions:
+                btn = gr.Button(
+                    question,
+                    variant="secondary",
+                    size="sm",
+                    scale=1,
+                )
+                # 버튼 클릭 시 입력창에 자동 입력
+                btn.click(
+                    fn=lambda q=question: q,
+                    outputs=[chatbot_interface.textbox],
+                )
+
+        # 정보 섹션
+        with gr.Accordion("📊 시스템 정보", open=False):
+            gr.Markdown("""
+            ### 사용된 기술
+            - **LLM**: Gemini 2.5 Flash Lite
+            - **임베딩**: BAAI/bge-m3 (로컬)
+            - **벡터 DB**: Qdrant Cloud
+            - **검색 API**: Stack Overflow, GitHub, Tavily
+            - **프레임워크**: LangGraph
+            
+            ### 주요 기능
+            - 🔍 **병렬 검색**: Stack Overflow, GitHub, 공식 문서 동시 검색
+            - 💾 **의미적 캐싱**: 유사 질문(임계값 0.85 이상) 즉시 답변
+            - 🎯 **의도 기반 라우팅**: debugging/learning/code_review 자동 분류
+            - 🔄 **자동 쿼리 개선**: 검색 결과 부족 시 최대 1회 자동 최적화
+            - 📝 **초보자 친화 답변**: 의도별 맞춤형 답변 구조
+            - 🔀 **다중 질문 처리**: 독립 질문 2개까지 병렬 처리
+            - 💬 **대화 맥락 이해**: clarification 질문은 히스토리 기반 답변
+            
+            ### LangGraph로 구현한 핵심 기능
+            1. ✅ **Conditional Edges**: 질문 유형/캐시 여부/검색 결과에 따른 동적 라우팅
+            2. ✅ **Send API**: 3개 검색 소스 병렬 실행 (fan-out/fan-in)
+            3. ✅ **Subgraph**: 검색 결과 필터링 및 요약 파이프라인
+            4. ✅ **Map-Reduce**: 다중 질문 처리 시 각 질문별 독립 실행 후 결과 통합
+            5. ✅ **Checkpointing**: MemorySaver로 대화 상태 저장 및 재개
+            6. ✅ **Pydantic Typed State**: 타입 안전한 상태 관리
+            
+            ### GitHub
+            [프로젝트 소스코드](https://github.com/shin-heewon/codeweaver)
+            """)
+        
+
+        
+        # 사용 가이드
+        with gr.Accordion("💡 사용 팁", open=False):
+            gr.Markdown("""
+            ### 1. 구체적으로 질문하기
+            - ❌ "파이썬 에러"
+            - ✅ "ImportError: No module named 'requests' 해결 방법"
+            
+            ### 2. 질문 유형별 예시
+            - **디버깅**: "이 에러 메시지는 무엇을 의미하나요?"
+            - **학습**: "JPA N+1 문제는 왜 발생하나요?"
+            - **코드 리뷰**: "이 코드를 더 효율적으로 개선하려면?"
+            
+            ### 3. 다중 질문 사용법
+            - ✅ **2개까지 가능**: "JWT가 뭐야? CORS는?"
+            - ❌ **3개 이상 불가**: "JWT? CORS? Docker?" → 안내 메시지 표시
+            - 💡 **팁**: 관련 질문은 하나로 통합하거나, 순차적으로 질문하세요
+            
+            ### 4. 대화 맥락 활용
+            - **후속 질문**: "좀 더 쉽게 설명해줘", "예제 코드로 보여줘"
+            - **새 개념 질문**: 대화 중에도 "Event Listener는 뭐야?" 같은 독립 질문 가능
+            - 💡 **팁**: 이전 대화를 참고한 답변이 필요하면 자연스럽게 질문하세요
+            
+            ### 5. 응답 시간
+            - **첫 질문**: 20~30초 소요 (검색 + 답변 생성)
+            - **유사 질문**: 즉시 답변 (캐시 활용, 임계값 0.85 이상)
+            - **다중 질문**: 각 질문별 병렬 처리로 효율적
+            
+            ### 6. 더 나은 답변을 위한 팁
+            - 에러 메시지를 포함해주세요
+            - 사용 중인 언어/프레임워크를 명시하세요
+            - 시도했던 해결 방법을 함께 알려주세요
+            - 검색 결과가 부족하면 자동으로 쿼리를 개선합니다 (최대 1회)
+            """)
+    
+    return demo
+
+
+# 앱 생성
+app = create_demo()
+
+
+if __name__ == "__main__":
+    # 로컬 실행
+    app.launch(
+        server_name="0.0.0.0",
+        server_port=7860,
+        share=False,  # True로 하면 공개 URL 생성
+        show_api=False,  # Gradio 4.44.x 버그 우회용
+    )