chat-agent-flow

Chat Agent Flow Guide

Chat 서비스 E2E 테스트 및 이벤트 파이프라인 가이드

Quick Reference

┌─────────────────────────────────────────────────────────────────────────────┐
│                           Chat Agent E2E Flow                                │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                              │
│  [1] POST /api/v1/chat              →  Create session (chat_id)             │
│  [2] POST /api/v1/chat/{id}/messages →  Send message (job_id)               │
│  [3] GET  /api/v1/chat/{job_id}/events →  Subscribe SSE (via chat-vs)       │
│                                                                              │
└─────────────────────────────────────────────────────────────────────────────┘

API Endpoints

API Endpoint 역할 상세

⚠️ 중요: POST /chat만 호출하면 세션만 생성되고 메시지는 처리되지 않습니다. 메시지를 Worker로 전송하려면 반드시 POST /chat/{chat_id}/messages를 호출해야 합니다.

Istio VirtualService Routing

┌─────────────────────────────────────────────────────────────────────────────┐
│                      Istio Routing Priority (chat-vs)                        │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                              │
│  순서가 중요! Regex는 Prefix보다 우선순위 낮음 → 같은 VS 내에서 순서로 제어   │
│                                                                              │
│  1. [regex]  /api/v1/chat/[^/]+/events  →  sse-gateway (SSE)                │
│  2. [prefix] /api/v1/chat               →  chat-api (REST)                  │
│                                                                              │
│  ⚠️ 순서 바뀌면 SSE 요청이 chat-api로 라우팅됨!                               │
│                                                                              │
└─────────────────────────────────────────────────────────────────────────────┘

파일: workloads/routing/chat/base/virtual-service.yaml

E2E Test Commands

TOKEN="<JWT_TOKEN>"

# 1. Create chat session
CHAT_RESPONSE=$(curl -s -X POST "https://api.dev.growbin.app/api/v1/chat" \
  -H "Content-Type: application/json" \
  -H "Cookie: s_access=$TOKEN" \
  -d '{"title": "E2E Test"}')
CHAT_ID=$(echo $CHAT_RESPONSE | jq -r '.id')

# 2. Send message
MSG_RESPONSE=$(curl -s -X POST "https://api.dev.growbin.app/api/v1/chat/${CHAT_ID}/messages" \
  -H "Content-Type: application/json" \
  -H "Cookie: s_access=$TOKEN" \
  -d '{"message": "플라스틱 분리배출 방법 알려줘"}')
JOB_ID=$(echo $MSG_RESPONSE | jq -r '.job_id')

# 3. Subscribe SSE
curl -sN --max-time 60 \
  "https://api.dev.growbin.app/api/v1/chat/${JOB_ID}/events" \
  -H "Accept: text/event-stream" \
  -H "Cookie: s_access=$TOKEN"

Intent Classification (10 types)

Dynamic Routing (Send API)

# Multi-Intent: "종이 버리는 법이랑 수거함도 알려줘"
→ intent = "waste"
→ additional_intents = ["collection_point"]
→ Sends:
   - Send("waste_rag", state)         # 주 intent
   - Send("collection_point", state)  # multi-intent
   - Send("weather", state)           # enrichment rule

# 3개 노드 병렬 실행!

Redis Streams Event Flow

┌─────────────────────────────────────────────────────────────────────────────┐
│                    Redis Streams Event Pipeline                              │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                              │
│  [1] chat-worker                                                             │
│       │                                                                      │
│       │ XADD chat:events:{shard} (Redis Streams)                            │
│       ▼                                                                      │
│  [2] rfr-streams-redis (Redis Sentinel)                                     │
│       │                                                                      │
│       │ XREADGROUP (Consumer Group)                                         │
│       ▼                                                                      │
│  [3] event-router                                                            │
│       │                                                                      │
│       │ PUBLISH sse:events:{job_id} (Redis Pub/Sub)                         │
│       ▼                                                                      │
│  [4] rfr-pubsub-redis (Redis Sentinel)                                      │
│       │                                                                      │
│       │ SUBSCRIBE (SSE Gateway)                                              │
│       ▼                                                                      │
│  [5] sse-gateway → SSE Stream → Client                                      │
│                                                                              │
└─────────────────────────────────────────────────────────────────────────────┘

Redis Instances

Message Persistence Architecture

┌─────────────────────────────────────────────────────────────────────────────┐
│                    Message Persistence Pipeline                              │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                              │
│  [1] chat-worker                                                             │
│       │                                                                      │
│       │ XADD chat:events:{shard} (done event 포함)                           │
│       ▼                                                                      │
│  [2] Redis Streams                                                           │
│       │                                                                      │
│       ├──► Consumer Group: eventrouter → event-router → SSE                 │
│       │                                                                      │
│       └──► Consumer Group: chat-persistence → chat-persistence-consumer                 │
│                   │                                                          │
│                   ▼                                                          │
│  [3] chat-persistence-consumer (ChatPersistenceConsumer)                                │
│       │                                                                      │
│       │ INSERT INTO chat.messages                                            │
│       ▼                                                                      │
│  [4] PostgreSQL (chat.messages)                                             │
│                                                                              │
└─────────────────────────────────────────────────────────────────────────────┘

Persistence Components

Consumer Group 확인

# Consumer Group 목록 조회
kubectl exec -n redis rfr-streams-redis-0 -c redis -- \
  redis-cli XINFO GROUPS chat:events:0

# Expected:
# 1) name: eventrouter
# 2) name: chat-persistence

# 메시지 수 확인
kubectl exec -n postgres deploy/postgresql -- \
  psql -U sesacthon -d ecoeco -c "SELECT COUNT(*) FROM chat.messages;"

Persistence 관련 테이블

Hop-by-Hop Address Reference

Stream Keys & Patterns

Environment Variables

Debug Commands (Redis)

# Redis Streams 확인 (rfr-streams-redis)
kubectl exec -n redis rfr-streams-redis-0 -- redis-cli XLEN chat:events:0
kubectl exec -n redis rfr-streams-redis-0 -- redis-cli XRANGE chat:events:0 - + COUNT 5

# Pub/Sub 모니터링 (rfr-pubsub-redis)
kubectl exec -n redis rfr-pubsub-redis-0 -- redis-cli PSUBSCRIBE "sse:events:*"

# Consumer Group 상태 확인
kubectl exec -n redis rfr-streams-redis-0 -- redis-cli XINFO GROUPS chat:events:0

Troubleshooting

Issue 1: SSE 요청이 404

증상: GET /api/v1/chat/{job_id}/events → 404

원인: VirtualService에서 SSE 규칙이 prefix 뒤에 배치

해결: chat-vs에서 SSE events 규칙을 prefix 규칙 앞에 배치

Issue 2: TaskIQ 메시지 파싱 오류

증상: Cannot parse message: b'{"args": [], "kwargs": {...}}'

원인: Worker가 TaskiqMessage 전체 형식을 기대

해결: task_id, task_name, labels 필드 포함

message = {
    "task_id": job_id,
    "task_name": "chat.process",
    "labels": {},
    "args": [],
    "kwargs": {...},
}

Issue 3: Redis READONLY 오류

증상: -READONLY You can't write against a read only replica

원인: Headless service가 replica로 연결

해결: Master pod DNS 직접 사용

rfr-pubsub-redis-0.rfr-pubsub-redis.redis.svc.cluster.local:6379

Issue 4: LangGraph State Access 오류

증상: KeyError: 'job_id'

원인: astream_events에서 일부 이벤트에 state 누락

해결: .get() 메서드로 안전한 접근

# Bad
job_id = state["job_id"]

# Good
job_id = state.get("job_id", "")

Issue 5: Send API 병렬 실행 충돌

증상: InvalidUpdateError: Can receive only one value per step

원인: StateGraph(dict) + Send API 병렬 실행

해결: Typed State + Annotated Reducer (진행 중)

# 임시: dynamic_routing 비활성화
enable_dynamic_routing=False

# 근본: StateGraph(ChatState) + Reducer

Issue 6: Message Persistence 실패 (chat-persistence-consumer)

증상: 대화는 정상 진행되지만 chat.messages 테이블에 메시지가 저장되지 않음

진단:

# 1. chat.messages 레코드 수 확인
kubectl exec -n postgres deploy/postgresql -- \
  psql -U sesacthon -d ecoeco -c "SELECT COUNT(*) FROM chat.messages;"

# 2. chat-persistence Consumer Group 확인
kubectl exec -n redis rfr-streams-redis-0 -c redis -- \
  redis-cli XINFO GROUPS chat:events:0 | grep -A5 chat-persistence

# 3. chat-persistence-consumer Pod 상태 확인
kubectl get pods -n chat -l app=chat-persistence-consumer

가능한 원인:

참조: docs/troubleshooting/chat-messages-not-persisted.md

Issue 7: SSE 이벤트 수신 불가 (Redis 불일치)

SSE Event Types (검증됨 - 2026-01-19):

상세 형식: references/sse-event-format.md 참조

Redis Streams Pipeline

chat-worker → XADD chat:events:{0-3} → event-router → PUBLISH sse:events:{job_id} → sse-gateway

Debug Commands

# Worker logs
kubectl logs -n chat -l app=chat-worker -f --tail=100

# Event Router logs
kubectl logs -n event-router -l app=event-router -f --tail=50

# Redis Streams check
kubectl exec -n redis rfr-streams-redis-0 -c redis -- redis-cli XLEN chat:events:0
kubectl exec -n redis rfr-streams-redis-0 -c redis -- redis-cli XINFO GROUPS chat:events:0

# Pub/Sub monitoring
kubectl exec -n redis rfr-pubsub-redis-0 -c redis -- redis-cli PSUBSCRIBE "sse:events:*"

Reference Files

• SSE Event Format: references/sse-event-format.md ← Token streaming 이벤트 형식 상세
• E2E Test Plan: docs/reports/e2e-intent-test-plan.md
• Troubleshooting (E2E): docs/troubleshooting/chat-worker-e2e-infra-fixes.md
• Troubleshooting (Persistence): docs/troubleshooting/chat-messages-not-persisted.md
• VirtualService (Chat): workloads/routing/chat/base/virtual-service.yaml
• VirtualService (SSE): workloads/domains/sse-gateway/base/virtualservice.yaml
• Frontend Spec: docs/specs/chat-agent-frontend-spec.md
• Dynamic Routing: docs/blogs/applied/30-langgraph-dynamic-routing-send-api.md
• Native Streaming: docs/blogs/applied/32-langgraph-native-streaming.md

Related Documents