mirror of
https://github.com/hwanny1128/HGZero.git
synced 2025-12-06 17:16:25 +00:00
b5159ef74e
주요 변경사항:
1. AI 서비스 설정
- claude_max_tokens: 8192 → 25000으로 증가 (회의록 통합을 위한 충분한 토큰 확보)
- AI 서비스 타임아웃: 30초 → 60초로 증가
2. 프롬프트 개선 (consolidate_prompt.py)
- JSON 생성 전문가 역할 추가
- JSON 이스케이프 규칙 명시 (큰따옴표, 줄바꿈, 역슬래시)
- Markdown 볼드체(**) 제거하여 JSON 파싱 오류 방지
- 문자열 검증 지시사항 추가
3. JSON 파싱 개선 (claude_service.py)
- 4단계 재시도 전략 구현:
* 이스케이프되지 않은 개행 문자 자동 수정
* strict=False 옵션으로 파싱
* 잘린 응답 복구 시도
* 제어 문자 제거 후 재시도
- 디버깅 로깅 강화 (Input/Output Tokens, Stop Reason)
- 파싱 실패 시 전체 응답을 파일로 저장
4. 회의 종료 로직 개선 (EndMeetingService.java)
- 통합 회의록 생성 또는 조회 로직 추가 (userId=NULL)
- Minutes 테이블에 전체 결정사항 저장
- AgendaSection에 minutesId 정확히 매핑
5. 테스트 데이터 추가
- AI 회의록 요약 테스트용 SQL 스크립트 작성
- 3명 참석자, 3개 안건의 현실적인 회의 시나리오
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-Authored-By: Claude <noreply@anthropic.com>
315 lines
12 KiB
Python
315 lines
12 KiB
Python
"""Claude API Service"""
|
|
import anthropic
|
|
import json
|
|
import logging
|
|
from typing import Dict, Any
|
|
from app.config import get_settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
settings = get_settings()
|
|
|
|
|
|
class ClaudeService:
|
|
"""Claude API 호출 서비스"""
|
|
|
|
def __init__(self):
|
|
self.client = anthropic.Anthropic(api_key=settings.claude_api_key)
|
|
self.model = settings.claude_model
|
|
self.max_tokens = settings.claude_max_tokens
|
|
self.temperature = settings.claude_temperature
|
|
|
|
async def generate_completion(
|
|
self,
|
|
prompt: str,
|
|
system_prompt: str = None
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Claude API 호출하여 응답 생성
|
|
|
|
Args:
|
|
prompt: 사용자 프롬프트
|
|
system_prompt: 시스템 프롬프트 (선택)
|
|
|
|
Returns:
|
|
Claude API 응답 (JSON 파싱)
|
|
"""
|
|
try:
|
|
# 메시지 구성
|
|
messages = [
|
|
{
|
|
"role": "user",
|
|
"content": prompt
|
|
}
|
|
]
|
|
|
|
# API 호출
|
|
logger.info(f"Claude API 호출 시작 - Model: {self.model}, Max Tokens: {self.max_tokens}")
|
|
|
|
if system_prompt:
|
|
response = self.client.messages.create(
|
|
model=self.model,
|
|
max_tokens=self.max_tokens,
|
|
temperature=self.temperature,
|
|
system=system_prompt,
|
|
messages=messages
|
|
)
|
|
else:
|
|
response = self.client.messages.create(
|
|
model=self.model,
|
|
max_tokens=self.max_tokens,
|
|
temperature=self.temperature,
|
|
messages=messages
|
|
)
|
|
|
|
# 응답 텍스트 추출
|
|
response_text = response.content[0].text
|
|
logger.info(
|
|
f"Claude API 응답 수신 완료 - "
|
|
f"Input Tokens: {response.usage.input_tokens}, "
|
|
f"Output Tokens: {response.usage.output_tokens}, "
|
|
f"Stop Reason: {response.stop_reason}"
|
|
)
|
|
|
|
# JSON 파싱
|
|
# ```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()
|
|
|
|
# JSON 파싱 전 전처리: 제어 문자 및 문제 문자 정리
|
|
import re
|
|
# 탭 문자를 공백으로 변환
|
|
response_text = response_text.replace('\t', ' ')
|
|
# 연속된 공백을 하나로 축소 (JSON 문자열 내부는 제외)
|
|
# response_text = re.sub(r'\s+', ' ', response_text)
|
|
|
|
result = json.loads(response_text)
|
|
|
|
return result
|
|
|
|
except json.JSONDecodeError as e:
|
|
logger.error(f"JSON 파싱 실패: {e}")
|
|
logger.error(f"응답 텍스트 전체 길이: {len(response_text)}")
|
|
logger.error(f"응답 텍스트 (처음 1000자): {response_text[:1000]}")
|
|
logger.error(f"응답 텍스트 (마지막 1000자): {response_text[-1000:]}")
|
|
|
|
# 전체 응답을 파일로 저장하여 디버깅
|
|
import os
|
|
from datetime import datetime
|
|
debug_dir = "/Users/jominseo/HGZero/ai-python/logs/debug"
|
|
os.makedirs(debug_dir, exist_ok=True)
|
|
debug_file = f"{debug_dir}/claude_response_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt"
|
|
with open(debug_file, 'w', encoding='utf-8') as f:
|
|
f.write(response_text)
|
|
logger.error(f"전체 응답을 파일로 저장: {debug_file}")
|
|
|
|
# JSON 파싱 재시도 전략
|
|
|
|
# 방법 1: 이스케이프되지 않은 개행 문자 처리
|
|
try:
|
|
import re
|
|
# JSON 문자열 내부의 이스케이프되지 않은 개행을 찾아 \\n으로 변환
|
|
# 패턴: 큰따옴표 내부에서 "\"로 시작하지 않는 개행
|
|
def fix_unescaped_newlines(text):
|
|
# 간단한 접근: 모든 실제 개행을 \\n으로 변환
|
|
# 단, JSON 구조의 개행 (객체/배열 사이)은 유지
|
|
in_string = False
|
|
escape_next = False
|
|
result = []
|
|
|
|
for char in text:
|
|
if escape_next:
|
|
result.append(char)
|
|
escape_next = False
|
|
continue
|
|
|
|
if char == '\\':
|
|
escape_next = True
|
|
result.append(char)
|
|
continue
|
|
|
|
if char == '"':
|
|
in_string = not in_string
|
|
result.append(char)
|
|
continue
|
|
|
|
if char == '\n':
|
|
if in_string:
|
|
# 문자열 내부의 개행은 \\n으로 변환
|
|
result.append('\\n')
|
|
else:
|
|
# JSON 구조의 개행은 유지
|
|
result.append(char)
|
|
else:
|
|
result.append(char)
|
|
|
|
return ''.join(result)
|
|
|
|
fixed_text = fix_unescaped_newlines(response_text)
|
|
result = json.loads(fixed_text)
|
|
logger.info("JSON 파싱 재시도 성공 (개행 문자 수정)")
|
|
return result
|
|
except Exception as e1:
|
|
logger.warning(f"개행 문자 수정 실패: {e1}")
|
|
|
|
# 방법 2: strict=False 옵션으로 파싱
|
|
try:
|
|
result = json.loads(response_text, strict=False)
|
|
logger.info("JSON 파싱 재시도 성공 (strict=False)")
|
|
return result
|
|
except Exception as e2:
|
|
logger.warning(f"strict=False 파싱 실패: {e2}")
|
|
|
|
# 방법 3: 마지막 닫는 괄호까지만 파싱 시도
|
|
try:
|
|
last_brace_idx = response_text.rfind('}')
|
|
if last_brace_idx > 0:
|
|
truncated_text = response_text[:last_brace_idx+1]
|
|
# 개행 수정도 적용
|
|
truncated_text = fix_unescaped_newlines(truncated_text)
|
|
result = json.loads(truncated_text, strict=False)
|
|
logger.info("JSON 파싱 재시도 성공 (잘린 부분 복구)")
|
|
return result
|
|
except Exception as e3:
|
|
logger.warning(f"잘린 부분 복구 실패: {e3}")
|
|
|
|
# 방법 4: 정규식으로 문제 문자 제거 후 재시도
|
|
try:
|
|
# 제어 문자 제거 (줄바꿈, 탭 제외)
|
|
cleaned = re.sub(r'[\x00-\x08\x0B-\x0C\x0E-\x1F\x7F]', '', response_text)
|
|
result = json.loads(cleaned, strict=False)
|
|
logger.info("JSON 파싱 재시도 성공 (제어 문자 제거)")
|
|
return result
|
|
except Exception as e4:
|
|
logger.warning(f"제어 문자 제거 실패: {e4}")
|
|
|
|
raise ValueError(f"Claude API 응답을 JSON으로 파싱할 수 없습니다: {str(e)}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Claude API 호출 실패: {e}")
|
|
raise
|
|
|
|
async def analyze_suggestions(self, transcript_text: str):
|
|
"""
|
|
회의 텍스트에서 AI 제안사항 추출
|
|
|
|
Args:
|
|
transcript_text: 회의 텍스트
|
|
|
|
Returns:
|
|
RealtimeSuggestionsResponse 객체
|
|
"""
|
|
from app.models import RealtimeSuggestionsResponse, SimpleSuggestion
|
|
from app.prompts.suggestions_prompt import get_suggestions_prompt
|
|
from datetime import datetime
|
|
import uuid
|
|
|
|
try:
|
|
# 프롬프트 생성
|
|
system_prompt, user_prompt = get_suggestions_prompt(transcript_text)
|
|
|
|
# Claude API 호출
|
|
result = await self.generate_completion(
|
|
prompt=user_prompt,
|
|
system_prompt=system_prompt
|
|
)
|
|
|
|
# 응답 파싱
|
|
suggestions_data = result.get("suggestions", [])
|
|
|
|
# SimpleSuggestion 객체로 변환
|
|
suggestions = [
|
|
SimpleSuggestion(
|
|
id=str(uuid.uuid4()),
|
|
content=s["content"],
|
|
timestamp=datetime.now().strftime("%H:%M:%S"),
|
|
confidence=s.get("confidence", 0.85)
|
|
)
|
|
for s in suggestions_data
|
|
if s.get("confidence", 0) >= 0.65 # 신뢰도 0.65 이상 (0.7 → 0.65 낮춤)
|
|
]
|
|
|
|
logger.info(f"AI 제안사항 {len(suggestions)}개 추출 완료")
|
|
|
|
return RealtimeSuggestionsResponse(suggestions=suggestions)
|
|
|
|
except Exception as e:
|
|
logger.error(f"제안사항 분석 실패: {e}", exc_info=True)
|
|
# 빈 응답 반환
|
|
return RealtimeSuggestionsResponse(suggestions=[])
|
|
|
|
async def generate_summary(
|
|
self,
|
|
text: str,
|
|
language: str = "ko",
|
|
style: str = "bullet",
|
|
max_length: int = None
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
텍스트 요약 생성
|
|
|
|
Args:
|
|
text: 요약할 텍스트
|
|
language: 요약 언어 (ko/en)
|
|
style: 요약 스타일 (bullet/paragraph)
|
|
max_length: 최대 요약 길이
|
|
|
|
Returns:
|
|
요약 결과 딕셔너리
|
|
"""
|
|
from app.models.summary import SummaryResponse
|
|
from app.prompts.summary_prompt import get_summary_prompt
|
|
|
|
try:
|
|
# 프롬프트 생성
|
|
system_prompt, user_prompt = get_summary_prompt(
|
|
text=text,
|
|
language=language,
|
|
style=style,
|
|
max_length=max_length
|
|
)
|
|
|
|
# Claude API 호출
|
|
result = await self.generate_completion(
|
|
prompt=user_prompt,
|
|
system_prompt=system_prompt
|
|
)
|
|
|
|
# 단어 수 계산
|
|
summary_text = result.get("summary", "")
|
|
key_points = result.get("key_points", [])
|
|
|
|
# 한국어와 영어의 단어 수 계산 방식 다르게 처리
|
|
if language == "ko":
|
|
# 한국어: 공백으로 구분된 어절 수
|
|
original_word_count = len(text.split())
|
|
summary_word_count = len(summary_text.split())
|
|
else:
|
|
# 영어: 공백으로 구분된 단어 수
|
|
original_word_count = len(text.split())
|
|
summary_word_count = len(summary_text.split())
|
|
|
|
compression_ratio = summary_word_count / original_word_count if original_word_count > 0 else 0
|
|
|
|
# 응답 생성
|
|
response = SummaryResponse(
|
|
summary=summary_text,
|
|
key_points=key_points,
|
|
word_count=summary_word_count,
|
|
original_word_count=original_word_count,
|
|
compression_ratio=round(compression_ratio, 2)
|
|
)
|
|
|
|
logger.info(f"요약 생성 완료 - 원본: {original_word_count}단어, 요약: {summary_word_count}단어")
|
|
|
|
return response.model_dump()
|
|
|
|
except Exception as e:
|
|
logger.error(f"요약 생성 실패: {e}", exc_info=True)
|
|
raise
|
|
|
|
|
|
# 싱글톤 인스턴스
|
|
claude_service = ClaudeService()
|