이코에코(Eco²) Agent Chat Worker E2E LangGraph 트러블슈팅

결과: Multi-Agent 병렬 실행 성공 (Aggregation 전 단계)

기간: 2026-01-17 ~ 2026-01-18
관련 PR: #400 ~ #414
목적: Chat Worker E2E 테스트 중 발생한 인프라 레벨 오류 해결

---

개요

Chat Worker E2E 테스트 진행 중 발생한 LangGraph 및 인프라 정합성 이슈들을 순차적으로 해결한 기록.

테스트 플로우:
POST /api/v1/chat (세션 생성)
    → POST /api/v1/chat/{session_id} (메시지 전송)
        → GET /api/v1/chat/{job_id}/events (SSE 스트림)
            → chat-worker 처리
                → SSE 이벤트 수신

---

Issue #1: SSE 라우팅 오류

PR #400

증상

GET /api/v1/chat/{job_id}/events
# 예상: sse-gateway로 라우팅
# 실제: chat-api로 라우팅 (404 또는 잘못된 응답)

원인

Istio VirtualService 매칭 우선순위:

1. Exact match
2. Prefix match (우선)
3. Regex match

chat-vs의 prefix: /api/v1/chat이 sse-gateway-external의 regex: /api/v1/[^/]+/[^/]+/events보다 우선 적용.

해결

chat-vs 내에 SSE events 라우팅 규칙을 prefix 규칙보다 앞에 배치:

# workloads/routing/chat/base/virtual-service.yaml
http:
  # SSE Events → sse-gateway (먼저!)
  - name: chat-sse-events
    match:
    - uri:
        regex: /api/v1/chat/[^/]+/events
      method:
        exact: GET
    route:
    - destination:
        host: sse-gateway.sse-consumer.svc.cluster.local

  # Chat API (prefix - 나중에)
  - match:
    - uri:
        prefix: /api/v1/chat
    route:
    - destination:
        host: chat-api.chat.svc.cluster.local

교훈

Istio VirtualService에서 regex는 prefix보다 우선순위가 낮음
같은 VirtualService 내에서 순서로 우선순위를 제어 필요

---

Issue #2: TaskIQ 메시지 형식 불일치

PR #401

증상

chat-worker 로그:
Cannot parse message: b'{"args": [], "kwargs": {...}}'
ValidationError: 3 validation errors for TaskiqMessage
  task_id: Field required
  task_name: Field required
  labels: Field required

원인

job_submitter.py에서 BrokerMessage.message에 {"args": [], "kwargs": {...}}만 전송.
Worker의 broker.formatter.loads()는 전체 TaskiqMessage 형식을 기대.

# Before (잘못됨)
message = {"args": [], "kwargs": {...}}

# After (정상)
message = {
    "task_id": job_id,
    "task_name": "chat.process",
    "labels": {},
    "args": [],
    "kwargs": {...},
}

해결

# apps/chat/infrastructure/messaging/job_submitter.py
taskiq_message = {
    "task_id": job_id,
    "task_name": "chat.process",
    "labels": {},
    "args": [],
    "kwargs": {
        "job_id": job_id,
        "user_id": user_id,
        "message": message,
        # ...
    },
}

교훈

TaskIQ의 BrokerMessage와 TaskiqMessage는 다른 형식
API 측에서 직접 큐에 publish할 때는 Worker가 기대하는 형식을 맞춰야 함

---

Issue #3: LangGraph Checkpointer 타입 오류

PR #408, #409, #410

증상

TypeError: Invalid checkpointer provided.
Expected an instance of BaseCheckpointSaver,
got <class 'AsyncGeneratorContextManager'>

원인

1. CachedPostgresSaver가 BaseCheckpointSaver를 상속하지 않음
2. AsyncRedisSaver가 async context manager를 반환하여 싱글톤 패턴과 호환 안 됨

해결

# apps/chat_worker/infrastructure/orchestration/langgraph/checkpointer.py

from langgraph.checkpoint.base import BaseCheckpointSaver  # 상속 추가

class CachedPostgresSaver(BaseCheckpointSaver):  # 상속!
    """PostgreSQL 체크포인터 with 캐싱."""
    ...

# Redis checkpointer: AsyncRedisSaver 대신 MemorySaver fallback
def create_redis_checkpointer():
    # AsyncRedisSaver는 context manager라 싱글톤 불가
    # MemorySaver로 fallback (개발 환경)
    return MemorySaver()

교훈

LangGraph 1.0+에서 커스텀 체크포인터는 반드시 BaseCheckpointSaver를 상속
AsyncRedisSaver는 async context manager로 설계되어 DI 컨테이너의 싱글톤 패턴과 호환되지 않음

 

---

Issue #4: LangGraph State Access 오류

PR #413

증상

KeyError: 'job_id'

원인

astream_events 사용 시 일부 이벤트에서 state 필드가 누락될 수 있음.
state["job_id"] 직접 접근 시 KeyError 발생.

해결

모든 노드에서 안전한 접근 방식으로 변경:

# AS-IS (위험)
job_id = state["job_id"]

# TO-BE (안전)
job_id = state.get("job_id", "")

수정된 노드:

• answer_node.py
• feedback_node.py
• intent_node.py
• rag_node.py
• web_search_node.py

교훈

LangGraph의 astream_events는 다양한 이벤트 타입을 발생시키며,
state가 포함되지 않는 Event가 존재할 수 있어 항상 .get() 메서드로 안전하게 접근해야 한다.

---

Issue #5: LangGraph Send API 병렬 실행 충돌

PR #414

증상

langgraph.errors.InvalidUpdateError:
At key '__root__': Can receive only one value per step.
Use an Annotated key with a reducer.

원인

# StateGraph(dict) 사용
graph = StateGraph(dict)  # Untyped state

# Send API로 병렬 실행
sends = [
    Send("waste_rag", state),
    Send("weather", state),
    Send("collection_point", state),
]

# 각 노드가 {**state, "my_field": value} 반환
# → 병렬로 __root__ 업데이트 시도 → 충돌!

임시 해결

Dynamic routing 비활성화:

# apps/chat_worker/setup/dependencies.py
return create_chat_graph(
    ...
    enable_dynamic_routing=False,  # 임시 비활성화
)

근본 해결 (TODO)

1. StateGraph(ChatState) - Typed State 사용
2. 각 subagent별 전용 채널 정의 with Annotated reducer
3. 노드는 자기 채널만 반환 ({**state, ...} 금지)

상세 설계: docs/plans/langgraph-channel-separation-adr.md

교훈

LangGraph Send API로 병렬 실행 시 반드시 Typed State + Annotated Reducer 필요.
StateGraph(dict)는 단일 노드 순차 실행에만 안전.

---

요약

Issue	PR	레이어	원인	해결
SSE 라우팅	#400	Istio	VirtualService 우선순위	규칙 순서 조정
TaskIQ 메시지	#401	Messaging	메시지 형식 불일치	전체 TaskiqMessage 형식 사용
Checkpointer	#408-410	LangGraph	BaseCheckpointSaver 미상속	상속 추가, MemorySaver fallback
State Access	#413	LangGraph	astream_events 동작	.get() 안전 접근
Send API 충돌	#414	LangGraph	Untyped State + 병렬	Typed State + Reducer (TODO)

---

레이어별 교훈

Istio/Routing

• VirtualService에서 regex < prefix 우선순위
• 같은 서비스 내에서 규칙 순서로 제어

Messaging (TaskIQ/RabbitMQ)

• BrokerMessage와 TaskiqMessage 형식 구분
• API → Worker 직접 publish 시 전체 메시지 형식 준수

LangGraph

• 커스텀 Checkpointer는 BaseCheckpointSaver 상속 필수
• astream_events 시 state 필드 안전 접근
• Send API 병렬 실행 시 Typed State + Reducer 필수