diff --git a/ai-python/app/services/eventhub_service.py b/ai-python/app/services/eventhub_service.py index c6109e6..7e63350 100644 --- a/ai-python/app/services/eventhub_service.py +++ b/ai-python/app/services/eventhub_service.py @@ -3,7 +3,6 @@ import asyncio import logging import json from azure.eventhub.aio import EventHubConsumerClient -from azure.eventhub.extensions.checkpointstoreblobaio import BlobCheckpointStore from app.config import get_settings from app.services.redis_service import RedisService @@ -87,8 +86,8 @@ class EventHubService: logger.debug(f"Redis 저장 완료 - meetingId: {meeting_id}") - # 체크포인트 업데이트 - await partition_context.update_checkpoint(event) + # MVP 개발: checkpoint 업데이트 제거 (InMemory 모드) + # await partition_context.update_checkpoint(event) except Exception as e: logger.error(f"이벤트 처리 오류: {e}", exc_info=True) diff --git a/ai-python/main.py b/ai-python/main.py index 98d1617..73169f3 100644 --- a/ai-python/main.py +++ b/ai-python/main.py @@ -1,5 +1,6 @@ """AI Service - FastAPI 애플리케이션""" import logging +import asyncio import uvicorn from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware @@ -7,6 +8,7 @@ from contextlib import asynccontextmanager from app.config import get_settings from app.api.v1 import suggestions +from app.services.eventhub_service import start_eventhub_listener # 로깅 설정 logging.basicConfig( @@ -27,8 +29,9 @@ async def lifespan(app: FastAPI): logger.info(f"Redis: {settings.redis_host}:{settings.redis_port}") logger.info("=" * 60) - # TODO: Event Hub 리스너 시작 (별도 백그라운드 태스크) - # asyncio.create_task(start_eventhub_listener()) + # Event Hub 리스너 시작 (백그라운드 태스크) + logger.info("Event Hub 리스너 백그라운드 시작...") + asyncio.create_task(start_eventhub_listener()) yield @@ -43,11 +46,11 @@ app = FastAPI( lifespan=lifespan ) -# CORS 설정 +# CORS 설정 (개발 환경: 모든 origin 허용) app.add_middleware( CORSMiddleware, - allow_origins=settings.cors_origins, - allow_credentials=True, + allow_origins=["*"], # 개발 환경에서 모든 origin 허용 + allow_credentials=False, # allow_origins=["*"]일 때는 False여야 함 allow_methods=["*"], allow_headers=["*"], ) diff --git a/ai/.run/ai-service.run.xml b/ai/.run/ai-service.run.xml index afc1de4..854373b 100644 --- a/ai/.run/ai-service.run.xml +++ b/ai/.run/ai-service.run.xml @@ -57,7 +57,7 @@ - + diff --git a/ai/src/main/java/com/unicorn/hgzero/ai/infra/config/EventHubConfig.java b/ai/src/main/java/com/unicorn/hgzero/ai/infra/config/EventHubConfig.java index b1123bd..efb9dc0 100644 --- a/ai/src/main/java/com/unicorn/hgzero/ai/infra/config/EventHubConfig.java +++ b/ai/src/main/java/com/unicorn/hgzero/ai/infra/config/EventHubConfig.java @@ -50,31 +50,29 @@ public class EventHubConfig { @PostConstruct public void startEventProcessor() { - // Checkpoint Storage가 설정되지 않은 경우 Event Hub 기능 비활성화 - if (checkpointStorageConnectionString == null || checkpointStorageConnectionString.isEmpty()) { - log.warn("Event Hub Processor 비활성화 - checkpoint storage 설정이 없습니다. " + - "개발 환경에서는 Event Hub 없이 실행 가능하며, 운영 환경에서는 AZURE_CHECKPOINT_STORAGE_CONNECTION_STRING 환경 변수를 설정해야 합니다."); - return; - } - log.info("Event Hub Processor 시작 - eventhub: {}, consumerGroup: {}", eventHubName, consumerGroup); - // Blob Checkpoint Store 생성 (체크포인트 저장소) - BlobContainerAsyncClient blobContainerAsyncClient = new BlobContainerClientBuilder() - .connectionString(checkpointStorageConnectionString) - .containerName(checkpointContainer) - .buildAsyncClient(); - - // Event Processor Client 빌드 - eventProcessorClient = new EventProcessorClientBuilder() + EventProcessorClientBuilder builder = new EventProcessorClientBuilder() .connectionString(connectionString, eventHubName) .consumerGroup(consumerGroup) - .checkpointStore(new BlobCheckpointStore(blobContainerAsyncClient)) .processEvent(this::processEvent) - .processError(this::processError) - .buildEventProcessorClient(); + .processError(this::processError); + // Checkpoint Storage 설정 + if (checkpointStorageConnectionString != null && !checkpointStorageConnectionString.isEmpty()) { + log.info("Checkpoint Storage 활성화 (Azure Blob) - container: {}", checkpointContainer); + BlobContainerAsyncClient blobContainerAsyncClient = new BlobContainerClientBuilder() + .connectionString(checkpointStorageConnectionString) + .containerName(checkpointContainer) + .buildAsyncClient(); + builder.checkpointStore(new BlobCheckpointStore(blobContainerAsyncClient)); + } else { + log.warn("Checkpoint Storage 미설정 - InMemory 모드 사용 (MVP 개발용, 재시작 시 처음부터 읽음)"); + builder.checkpointStore(new InMemoryCheckpointStore()); + } + + eventProcessorClient = builder.buildEventProcessorClient(); eventProcessorClient.start(); log.info("Event Hub Processor 시작 완료"); diff --git a/build.gradle b/build.gradle index 26e3bcb..17b6d3c 100644 --- a/build.gradle +++ b/build.gradle @@ -44,7 +44,7 @@ subprojects { hypersistenceVersion = '3.7.3' openaiVersion = '0.18.2' feignJacksonVersion = '13.1' - azureSpeechVersion = '1.37.0' + azureSpeechVersion = '1.44.0' azureBlobVersion = '12.25.3' azureEventHubsVersion = '5.18.2' azureEventHubsCheckpointVersion = '1.19.2' diff --git a/stt/.run/stt-service.run.xml b/stt/.run/stt-service.run.xml index b2f5316..5514542 100644 --- a/stt/.run/stt-service.run.xml +++ b/stt/.run/stt-service.run.xml @@ -33,7 +33,7 @@ - + diff --git a/stt/.run/stt.run.xml b/stt/.run/stt.run.xml index 99c5c9a..598caeb 100644 --- a/stt/.run/stt.run.xml +++ b/stt/.run/stt.run.xml @@ -34,8 +34,8 @@ - - + + diff --git a/stt/build.gradle b/stt/build.gradle index 42d3f2c..29a4d73 100644 --- a/stt/build.gradle +++ b/stt/build.gradle @@ -15,8 +15,14 @@ dependencies { // Database runtimeOnly 'org.postgresql:postgresql' - // Azure Speech SDK - implementation "com.microsoft.cognitiveservices.speech:client-sdk:${azureSpeechVersion}" + // Azure Speech SDK (macOS/Linux/Windows용) + implementation("com.microsoft.cognitiveservices.speech:client-sdk:${azureSpeechVersion}") { + artifact { + name = 'client-sdk' + extension = 'jar' + type = 'jar' + } + } // Azure Blob Storage implementation "com.azure:azure-storage-blob:${azureBlobVersion}" diff --git a/stt/src/main/java/com/unicorn/hgzero/stt/config/RedisStreamConfig.java b/stt/src/main/java/com/unicorn/hgzero/stt/config/RedisStreamConfig.java index fa0ba51..3174945 100644 --- a/stt/src/main/java/com/unicorn/hgzero/stt/config/RedisStreamConfig.java +++ b/stt/src/main/java/com/unicorn/hgzero/stt/config/RedisStreamConfig.java @@ -42,4 +42,24 @@ public class RedisStreamConfig { public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory connectionFactory) { return new StringRedisTemplate(connectionFactory); } + + /** + * 범용 Object 저장용 RedisTemplate + */ + @Bean + public RedisTemplate redisTemplate(RedisConnectionFactory connectionFactory) { + RedisTemplate template = new RedisTemplate<>(); + template.setConnectionFactory(connectionFactory); + + // Key Serializer + template.setKeySerializer(new StringRedisSerializer()); + template.setHashKeySerializer(new StringRedisSerializer()); + + // Value Serializer + template.setValueSerializer(new GenericJackson2JsonRedisSerializer()); + template.setHashValueSerializer(new GenericJackson2JsonRedisSerializer()); + + template.afterPropertiesSet(); + return template; + } } diff --git a/stt/src/main/java/com/unicorn/hgzero/stt/config/WebSocketConfig.java b/stt/src/main/java/com/unicorn/hgzero/stt/config/WebSocketConfig.java index 20b3df0..5750e5e 100644 --- a/stt/src/main/java/com/unicorn/hgzero/stt/config/WebSocketConfig.java +++ b/stt/src/main/java/com/unicorn/hgzero/stt/config/WebSocketConfig.java @@ -2,10 +2,12 @@ package com.unicorn.hgzero.stt.config; import com.unicorn.hgzero.stt.controller.AudioWebSocketHandler; import lombok.RequiredArgsConstructor; +import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.web.socket.config.annotation.EnableWebSocket; import org.springframework.web.socket.config.annotation.WebSocketConfigurer; import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry; +import org.springframework.web.socket.server.standard.ServletServerContainerFactoryBean; /** * WebSocket 설정 @@ -24,4 +26,16 @@ public class WebSocketConfig implements WebSocketConfigurer { registry.addHandler(audioWebSocketHandler, "/ws/audio") .setAllowedOrigins("*"); // 실제 운영 환경에서는 특정 도메인으로 제한 } + + /** + * WebSocket 메시지 버퍼 크기 설정 + * 오디오 청크 전송을 위해 충분한 버퍼 크기 확보 (10MB) + */ + @Bean + public ServletServerContainerFactoryBean createWebSocketContainer() { + ServletServerContainerFactoryBean container = new ServletServerContainerFactoryBean(); + container.setMaxTextMessageBufferSize(10 * 1024 * 1024); // 10MB + container.setMaxBinaryMessageBufferSize(10 * 1024 * 1024); // 10MB + return container; + } } \ No newline at end of file diff --git a/stt/src/main/java/com/unicorn/hgzero/stt/controller/AudioWebSocketHandler.java b/stt/src/main/java/com/unicorn/hgzero/stt/controller/AudioWebSocketHandler.java index ac3d580..3d1438e 100644 --- a/stt/src/main/java/com/unicorn/hgzero/stt/controller/AudioWebSocketHandler.java +++ b/stt/src/main/java/com/unicorn/hgzero/stt/controller/AudioWebSocketHandler.java @@ -12,13 +12,12 @@ import org.springframework.web.socket.TextMessage; import org.springframework.web.socket.WebSocketSession; import org.springframework.web.socket.handler.AbstractWebSocketHandler; -import java.util.Base64; -import java.util.Map; +import java.util.*; import java.util.concurrent.ConcurrentHashMap; /** * 오디오 WebSocket 핸들러 - * 프론트엔드에서 실시간 오디오 스트림을 수신 + * 프론트엔드에서 실시간 오디오 스트림을 수신하고 STT 결과를 전송 */ @Slf4j @Component @@ -31,6 +30,9 @@ public class AudioWebSocketHandler extends AbstractWebSocketHandler { // 세션별 회의 ID 매핑 private final Map sessionMeetingMap = new ConcurrentHashMap<>(); + // 회의 ID별 세션 목록 (결과 브로드캐스트용) + private final Map> meetingSessionsMap = new ConcurrentHashMap<>(); + @Override public void afterConnectionEstablished(WebSocketSession session) throws Exception { log.info("WebSocket 연결 성공 - sessionId: {}", session.getId()); @@ -50,6 +52,11 @@ public class AudioWebSocketHandler extends AbstractWebSocketHandler { // 녹음 시작 String meetingId = (String) data.get("meetingId"); sessionMeetingMap.put(session.getId(), meetingId); + + // 세션을 회의별 목록에 추가 + meetingSessionsMap.computeIfAbsent(meetingId, k -> ConcurrentHashMap.newKeySet()) + .add(session); + log.info("녹음 시작 - sessionId: {}, meetingId: {}", session.getId(), meetingId); // 응답 전송 @@ -147,9 +154,66 @@ public class AudioWebSocketHandler extends AbstractWebSocketHandler { } } + /** + * STT 결과를 특정 회의의 모든 클라이언트에게 전송 + */ + public void sendTranscriptToMeeting(String meetingId, String text, double confidence) { + Set sessions = meetingSessionsMap.get(meetingId); + + if (sessions == null || sessions.isEmpty()) { + log.debug("전송할 세션 없음 - meetingId: {}", meetingId); + return; + } + + try { + Map result = new HashMap<>(); + result.put("transcript", text); + result.put("confidence", confidence); + result.put("timestamp", System.currentTimeMillis()); + result.put("speaker", "참석자"); + + String jsonMessage = objectMapper.writeValueAsString(result); + TextMessage message = new TextMessage(jsonMessage); + + // 모든 세션에 브로드캐스트 + Iterator iterator = sessions.iterator(); + while (iterator.hasNext()) { + WebSocketSession session = iterator.next(); + try { + if (session.isOpen()) { + session.sendMessage(message); + } else { + iterator.remove(); + } + } catch (Exception e) { + log.error("메시지 전송 실패 - sessionId: {}", session.getId(), e); + iterator.remove(); + } + } + + log.info("STT 결과 전송 완료 - meetingId: {}, sessions: {}개, text: {}", + meetingId, sessions.size(), text); + + } catch (Exception e) { + log.error("STT 결과 전송 실패 - meetingId: {}", meetingId, e); + } + } + @Override public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception { String meetingId = sessionMeetingMap.remove(session.getId()); + + // 회의별 세션 목록에서도 제거 + if (meetingId != null) { + Set sessions = meetingSessionsMap.get(meetingId); + if (sessions != null) { + sessions.remove(session); + if (sessions.isEmpty()) { + meetingSessionsMap.remove(meetingId); + } + } + } + log.info("WebSocket 연결 종료 - sessionId: {}, meetingId: {}, status: {}", session.getId(), meetingId, status); } diff --git a/stt/src/main/java/com/unicorn/hgzero/stt/service/AudioBatchProcessor.java b/stt/src/main/java/com/unicorn/hgzero/stt/service/AudioBatchProcessor.java index ae0d9d3..78f68fd 100644 --- a/stt/src/main/java/com/unicorn/hgzero/stt/service/AudioBatchProcessor.java +++ b/stt/src/main/java/com/unicorn/hgzero/stt/service/AudioBatchProcessor.java @@ -1,15 +1,13 @@ package com.unicorn.hgzero.stt.service; +import com.unicorn.hgzero.stt.controller.AudioWebSocketHandler; import com.unicorn.hgzero.stt.dto.AudioChunkDto; import com.unicorn.hgzero.stt.event.TranscriptionEvent; import com.unicorn.hgzero.stt.event.publisher.EventPublisher; -import com.unicorn.hgzero.stt.repository.entity.TranscriptSegmentEntity; -import com.unicorn.hgzero.stt.repository.jpa.TranscriptSegmentRepository; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; -import org.springframework.transaction.annotation.Transactional; import java.time.LocalDateTime; import java.util.List; @@ -19,6 +17,9 @@ import java.util.UUID; /** * 오디오 배치 프로세서 * 5초마다 Redis에 축적된 오디오를 처리하여 텍스트로 변환 + * + * Note: STT 결과는 DB에 저장하지 않고, Event Hub와 WebSocket으로만 전송 + * 최종 회의록은 AI 서비스에서 저장 */ @Slf4j @Service @@ -27,18 +28,17 @@ public class AudioBatchProcessor { private final AudioBufferService audioBufferService; private final AzureSpeechService azureSpeechService; - private final TranscriptSegmentRepository segmentRepository; private final EventPublisher eventPublisher; + private final AudioWebSocketHandler webSocketHandler; /** * 5초마다 오디오 배치 처리 * - Redis에서 오디오 청크 조회 * - Azure Speech로 텍스트 변환 - * - DB 저장 - * - Event Hub 이벤트 발행 + * - Event Hub 이벤트 발행 (AI 서비스로 전송) + * - WebSocket 실시간 전송 (클라이언트 표시) */ @Scheduled(fixedDelay = 5000, initialDelay = 10000) // 5초마다 실행, 최초 10초 후 시작 - @Transactional public void processAudioBatch() { try { // 활성 회의 목록 조회 @@ -96,12 +96,12 @@ public class AudioBatchProcessor { return; } - // 텍스트 세그먼트 DB 저장 - saveTranscriptSegment(meetingId, result); - - // Event Hub 이벤트 발행 + // Event Hub 이벤트 발행 (AI 서비스로 전송) publishTranscriptionEvent(meetingId, result); + // WebSocket으로 실시간 결과 전송 (클라이언트 표시) + sendTranscriptToClients(meetingId, result); + // Redis 정리 audioBufferService.clearProcessedChunks(meetingId); @@ -112,35 +112,9 @@ public class AudioBatchProcessor { } } - /** - * 텍스트 세그먼트 DB 저장 - */ - private void saveTranscriptSegment(String meetingId, AzureSpeechService.RecognitionResult result) { - String segmentId = UUID.randomUUID().toString(); - long timestamp = System.currentTimeMillis(); - boolean warningFlag = result.getConfidence() < 0.6; - - TranscriptSegmentEntity segment = TranscriptSegmentEntity.builder() - .segmentId(segmentId) - .recordingId(meetingId) // 간소화: recordingId = meetingId - .text(result.getText()) - .speakerId("UNKNOWN") // 화자 식별 제거 - .speakerName("참석자") - .timestamp(timestamp) - .duration(5.0) // 5초 분량 - .confidence(result.getConfidence()) - .warningFlag(warningFlag) - .chunkIndex(0) - .build(); - - segmentRepository.save(segment); - - log.debug("텍스트 세그먼트 저장 완료 - segmentId: {}, text: {}", - segmentId, result.getText()); - } - /** * Event Hub 이벤트 발행 (AI 서비스로 전송) + * AI 서비스에서 Claude API로 제안사항 분석 후 처리 */ private void publishTranscriptionEvent(String meetingId, AzureSpeechService.RecognitionResult result) { try { @@ -167,4 +141,16 @@ public class AudioBatchProcessor { log.error("Event Hub 이벤트 발행 실패 - meetingId: {}", meetingId, e); } } + + /** + * WebSocket으로 STT 결과를 클라이언트에게 실시간 전송 + */ + private void sendTranscriptToClients(String meetingId, AzureSpeechService.RecognitionResult result) { + try { + webSocketHandler.sendTranscriptToMeeting(meetingId, result.getText(), result.getConfidence()); + log.debug("WebSocket 결과 전송 완료 - meetingId: {}, text: {}", meetingId, result.getText()); + } catch (Exception e) { + log.error("WebSocket 결과 전송 실패 - meetingId: {}", meetingId, e); + } + } } diff --git a/stt/src/main/java/com/unicorn/hgzero/stt/service/AudioBufferService.java b/stt/src/main/java/com/unicorn/hgzero/stt/service/AudioBufferService.java index 8d9955f..35c149f 100644 --- a/stt/src/main/java/com/unicorn/hgzero/stt/service/AudioBufferService.java +++ b/stt/src/main/java/com/unicorn/hgzero/stt/service/AudioBufferService.java @@ -8,6 +8,7 @@ import org.springframework.data.redis.core.RedisTemplate; import org.springframework.stereotype.Service; import java.util.ArrayList; +import java.util.Base64; import java.util.List; import java.util.Map; import java.util.Set; @@ -38,9 +39,12 @@ public class AudioBufferService { try { String streamKey = getStreamKey(chunk.getMeetingId()); + // 바이트 배열을 Base64로 인코딩 + String encodedAudioData = Base64.getEncoder().encodeToString(chunk.getAudioData()); + // Hash 형태로 저장 Map data = Map.of( - "audioData", chunk.getAudioData(), + "audioData", encodedAudioData, "timestamp", chunk.getTimestamp(), "chunkIndex", chunk.getChunkIndex(), "format", chunk.getFormat() != null ? chunk.getFormat() : "audio/webm", @@ -87,9 +91,13 @@ public class AudioBufferService { for (MapRecord record : records) { Map value = record.getValue(); + // Base64로 인코딩된 문자열을 바이트 배열로 디코딩 + String encodedAudioData = (String) value.get("audioData"); + byte[] audioData = Base64.getDecoder().decode(encodedAudioData); + AudioChunkDto chunk = AudioChunkDto.builder() .meetingId(meetingId) - .audioData((byte[]) value.get("audioData")) + .audioData(audioData) .timestamp(Long.valueOf(value.get("timestamp").toString())) .chunkIndex(Integer.valueOf(value.get("chunkIndex").toString())) .format((String) value.get("format")) diff --git a/test-audio/stt-test-ai.html b/test-audio/stt-test-ai.html new file mode 100644 index 0000000..7466d40 --- /dev/null +++ b/test-audio/stt-test-ai.html @@ -0,0 +1,471 @@ + + + + + + HGZero AI 제안사항 실시간 테스트 + + + +
+

💡 HGZero AI 제안사항 실시간 테스트

+

STT + Claude AI 기반 실시간 회의 제안사항

+ +
+

📋 테스트 정보

+

STT Service: ws://localhost:8084/ws/audio

+

AI Service: http://localhost:8086/api/v1/ai/suggestions

+

Meeting ID: test-meeting-001

+
+ +
+ 🔴 준비 중 +
+ +
+ + +
+ +
+
+

🎙️ 회의를 시작하면 AI가 분석한 제안사항이 여기에 표시됩니다.

+

명확하게 회의 내용을 말씀해주세요.

+
+
+ +
+
시스템 로그...
+
+
+ + + + diff --git a/test-audio/stt-test-wav.html b/test-audio/stt-test-wav.html new file mode 100644 index 0000000..04b995a --- /dev/null +++ b/test-audio/stt-test-wav.html @@ -0,0 +1,560 @@ + + + + + + HGZero STT 실시간 테스트 (WAV) + + + +
+

🎤 HGZero 실시간 STT 테스트 (WAV)

+

WebSocket 기반 실시간 음성-텍스트 변환 (PCM WAV 16kHz)

+ +
+

📋 테스트 정보

+

WebSocket URL: ws://localhost:8084/ws/audio

+

Meeting ID: test-meeting-001

+

Audio Format: PCM WAV, 16kHz, Mono, 16-bit

+
+ +
+ 🔴 연결 끊김 +
+ +
+ + +
+ +
+

여기에 실시간 STT 결과가 5초마다 표시됩니다...

+
+ +

💡 실시간 AI 제안사항

+
+

AI 제안사항이 여기에 표시됩니다...

+
+ +
+
시스템 로그...
+
+
+ + + + diff --git a/test-audio/stt-test.html b/test-audio/stt-test.html new file mode 100644 index 0000000..5ffaba6 --- /dev/null +++ b/test-audio/stt-test.html @@ -0,0 +1,405 @@ + + + + + + HGZero STT 실시간 테스트 + + + +
+

🎤 HGZero 실시간 STT 테스트

+

WebSocket 기반 실시간 음성-텍스트 변환

+ +
+

📋 테스트 정보

+

WebSocket URL: ws://localhost:8084/ws/audio

+

Meeting ID: test-meeting-001

+

Sample Rate: 16000 Hz

+
+ +
+ 🔴 연결 끊김 +
+ +
+ + +
+ +
+

여기에 실시간 STT 결과가 표시됩니다...

+
+ +
+
시스템 로그...
+
+
+ + + +