AI 기반 중재 서비스를 제공하기 위해 사용자 행동 분류, 감정 분석, 공감 점수 측정 등 세 가지 주요 모델을 통합하여 갈등 상황을 분석하고 과실 비율을 산출하는 프로젝트입니다. 본 문서는 AI 모델 구조, 데이터 처리 방식, 학습 과정, 그리고 모델 통합 로직을 다룹니다.
“인연을 이어주는 붉은 실”
홍연은 AI 기술을 통해 사람들 간 갈등을 분석하고, 해결 방안을 제시하는 혁신적인 중재 서비스를 제공합니다.
Aimo는 음성 및 텍스트 데이터를 AI 모델로 분석하여 갈등의 원인을 파악하고 공정한 해결 방안을 제시합니다.
서비스는 STT, OCR, 갈등 분석 AI 모델로 구성되며, 최종적으로 각 참여자의 과실 비율을 산출합니다.
- 파일:
behavior_classification.py - 기능: 사용자의 대화 속에서 행동 유형을 분류하여 갈등의 구체적인 원인을 파악
- 분류 카테고리: 경쟁형/회피형/수용형/타협형/협력형
- config:
- MAX_LENGTH: 256
- BATCH_SIZE: 16
- EPOCHS: 10
- LEARNING_RATE: 2e-5
- NUM_LABELS: 5
- PATIENCE: 5
- WEIGHT_DECAY: 0.01
- text augmentation
- EDA
- 단어 삭제, 교체, 추가, 순서 변경
- 유사 임베딩 대체
- 단어 임베딩 유사도 기반 대체
- Contetualized Embedding 대체
- 문맥 기반 단어 대체
- EDA
- 📂 데이터셋 예시:
- 총 3,869개 (경쟁형:771개/회피형:759개/수용형:786개/타협형:767개/협력형:786개) -> 증강 후 21665개
문맥 및 감정 점수 모델에서 사용된 데이터셋(
behavior_dataset.json)의 일부 예시는 다음과 같습니다:
- 총 3,869개 (경쟁형:771개/회피형:759개/수용형:786개/타협형:767개/협력형:786개) -> 증강 후 21665개
문맥 및 감정 점수 모델에서 사용된 데이터셋(
{
"behavior_examples": [
{
"category": "경쟁형",
"examples": [
{"text": "이게 다 네가 제대로 준비 안 해서 그런 거 아니야?", "label": "경쟁형"},
{"text": "네가 이렇게 할 줄 알았어", "label": "경쟁형"},
{"text": "이런 기본적인 것도 못하면서 뭘 하겠다는 거야", "label": "경쟁형"}
]
},
{
"category": "협력형",
"examples": [
{"text": "우리 이 문제를 함께 해결해 보자.", "label": "협력형"},
{"text": "다음 번엔 내가 도와줄게.", "label": "협력형"},
{"text": "우리가 힘을 합치면 잘 할 수 있을 거야.", "label": "협력형"}
]
}
]
}- 핵심 로직:
def classify_behavior(input_text: str): """ 행동 유형을 분류하는 함수 Args: input_text (str): 입력 텍스트 Returns: prediction (str): 행동 유형 예측 """ model = torch.load("Behavior_classifier.pt") tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased") tokens = tokenizer(input_text, return_tensors="pt") output = model(**tokens) prediction = torch.argmax(output.logits, dim=-1) return prediction
- 출력: 행동 분류 결과 (ex. 경쟁형)
- 파일:
BERTbasedcontext.py - 기능: 대화의 문맥과 감정을 분석하여 상황의 심각성을 점수화
- text augmentation
- Random Insertion
- 임의의 단어를 문장 내에 삽입하여 문장의 다양성 증가
- Add Noise
- 문자 수준에서 노이즈를 추가해 모델의 견고성 향상
- Random Insertion
- 📂 데이터셋 예시:
- 총 943개 -> 증강 후 2829개
문맥 및 감정 점수 모델에서 사용된 데이터셋(
BERT-based_dataset.json)의 일부 예시는 다음과 같습니다:
- 총 943개 -> 증강 후 2829개
문맥 및 감정 점수 모델에서 사용된 데이터셋(
{
"conversation_id": "conv_001",
"utterances": [
"여보세요? 방금 면접 결과 연락 받았어요!",
"헐.. 떨려서 심장이 터질 것 같아요. 근데 최종 합격이래요!!",
"아... 근데 연봉이 생각보다 너무 낮게 제시됐어요."
],
"emotions": [0.6, 0.9, 0.3],
"context_labels": {
"situation": "job_interview_result",
"emotion_flow": "anticipation->joy->disappointment"
}
}- 핵심 로직:
def analyze_emotion(input_text: str):
"""
입력 텍스트의 감정 점수를 분석
Args:
input_text (str): 입력 텍스트
Returns:
emotion_score (float): 감정 점수
"""
model = torch.load("BERTbasedemotion_model.pt")
tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased")
tokens = tokenizer(input_text, return_tensors="pt")
output = model(**tokens)
emotion_score = torch.softmax(output.logits, dim=-1)[0, 1].item()
return emotion_score- 출력: 감정 점수 (0-1 범위)
- 파일: 'empathy_data_preprocessig.py',
empathy_score.py.py - 기능: 대화에서 상대방에 대한 공감을 점수화하여 협력 가능성 평가
- text augmentation
- Random deletion
- 문장에서 임의의 단어 삭제
- Random swap
- 문장 내 임의의 두단어의 위치를 교환하여 문장의 구조 변형
- Random insertion
- 문장 내 임의의 위치에 단어 삽입하여 다양성 증가시킴
- Synonym replacement
- init에서 한국어 단어와 유사어 리스트를 사전 형태로 저장함
- 단순 사전 기반으로 단어를 유사어로 치환하여 문장의 의미 보존해 다양성 증가
- Noise Injection
- 텍스트에 노이즈 주입해 견고성 향상시킴
- Augment text
- 여러 증강 기법 조합해 텍스트 변형
- Random deletion
- 📂 데이터셋 예시:
- 원본 데이터 크기: 2,692개 대화 -> 증강 후 데이터 크기: 14787
문맥 및 감정 점수 모델에서 사용된 데이터셋(
empathy_dataset.json)의 일부 예시는 다음과 같습니다:
{
"dialogue_id": "001",
"utterances": [
{
"utterance_id": "001_1",
"speaker": "A",
"text": "자기야, 오늘 기분 어때?",
"emotion": "중립",
"empathy_score": 0.0
},
{
"utterance_id": "001_2",
"speaker": "B",
"text": "회사에서 실수해서 팀장님한테 혼났어... 많이 속상해",
"emotion": "슬픔",
"empathy_score": 0.0
},
{
"utterance_id": "001_3",
"speaker": "A",
"text": "많이 힘들었겠다... 우리 자기가 평소에 얼마나 열심히 하는데. 저녁에 맛있는 거 먹으러 갈까?",
"emotion": "공감",
"empathy_score": 0.9
}
]
}- 핵심 로직:
def compute_empathy_score(dialogues: List[str]):
"""
대화의 공감 점수를 계산
Args:
dialogues (List[str]): 대화 리스트
Returns:
empathy_score (float): 공감 점수
"""
model = torch.load("bestmodel.pt")
features = preprocess_dialogues(dialogues)
output = model(features)
empathy_score = torch.mean(output.logits).item()
return empathy_score- 출력: 공감 점수 (0-1 범위)
- 파일: 'score_multi.py'
- 기능: 3개의 모델의 결과를 통합하여 최종 과실 비율 산출
- 핵심 로직: 상황 분석 및 요약 -> 행동 분류 -> 문맥 감정 -> 공감 점수 -> 최종 판결문 출력
- 상황 분석 (SituationAnalyzer)
- 기능: 갈등 텍스트를 분석하여 주요 상황과 각 상황의 중요도를 계산하고 관련 문장 추출
- 주요 갈등 상황과 중요도 분석
- 관련 문장 빈도(frequency_score): 문장에서 키워드와 일치하는 비율
- 감정 중요도(emotion_importance): 관련 문장의 감정 점수 평균ㄴ
class SituationAnalyzer:
async def analyze_text(self, text: str) -> Dict[str, Dict[str, any]]:
"""
텍스트에서 주요 갈등 상황과 중요도를 분석합니다.
Args:
text (str): 입력 텍스트
Returns:
Dict: 상황별 중요도와 관련 문장
"""
prompt = f"""Analyze the following text:
Text: {text}
Please respond in the following JSON format:
{{
"situations": [
{{
"category": "situation category name",
"importance": importance of the situation (0 to 1),
"related_sentences": ["sentence 1", "sentence 2"]
}},
...
]
}}"""
response = await self.client.chat.completions.create(
model=Config.GPT_MODEL,
messages=[
{"role": "system", "content": "You are an expert at analyzing conflict situations."},
{"role": "user", "content": prompt}
]
)
analysis = json.loads(response.choices[0].message.content)
return {
s["category"]: {
"importance": s["importance"],
"sentences": s["related_sentences"]
} for s in analysis["situations"]
}- 행동 분류 및 점수 매핑 (Behavior_classification)
- 클래스: CustomBERTClassifier
- 행동 점수 매핑
def map_category_score(category): score_map = { "경쟁": 0, "회피": 0, "타협": 0.5, "협력": 1, "수용": 1 } return score_map.get(category, 0)
- 감정 분석 (Emotion Analysis)
- 클래스: SentenceEmotionAnalyzer
- 과실 비율 계산 (Fault Ratio Calculation)
- ConflictAnalyzer
- 기능: 감정 점수, 행동 점수, 상황 중요도를 기반으로 원고와 피고의 과실 비율을 계산
- 핵심 로직
async def analyze_content(self, content: str) -> dict: """ 텍스트 데이터를 분석하고 과실 비율을 계산합니다. Args: content (str): 분석할 텍스트 Returns: dict: 분석 결과 """ situation_data = await self.situation_analyzer.analyze_text(content) speaker_scores = {"plaintiff": 0, "defendant": 0} for situation, data in situation_data.items(): sentences = data["sentences"] importance = data["importance"] emotion_scores = self.emotion_analyzer.analyze_sentences(sentences) behavior_scores = predict_category_scores( sentences, self.behavior_model, self.tokenizer, self.label_map, self.device ) for e_score, b_score in zip(emotion_scores, behavior_scores): score = e_score * b_score * importance speaker_scores["plaintiff" if "내가" in sentences else "defendant"] += score total_score = sum(speaker_scores.values()) fault_rate = round((speaker_scores["plaintiff"] / total_score) * 100, 2) if total_score else 50.0 return {"plaintiff_fault_rate": fault_rate}
- 파일: 'audio_process.py', 'download_s3_file.py'
- 사용된 라이브러리: OpenAI Whisper, Boto3
- 엔드 포인트: /speech-to-text
- 기능: S3에 저장된 음성 파일을 다운로드하여 텍스트로 변환
- 핵심 로직:
async def download_s3_file(url):
"""
AWS S3에서 파일을 다운로드합니다.
Args:
url (str): S3 객체 URL
Returns:
str: 다운로드된 파일의 경로
"""
bucket_name, object_key, filename = await parse_s3_url(url)
temp_dir = "temp"
os.makedirs(temp_dir, exist_ok=True)
s3_client.download_file(bucket_name, object_key, temp_file_path)
if os.path.exists(temp_file_path):
logger.info(f"File downloaded successfully: {temp_file_path}")
else:
raise FileNotFoundError("Downloaded file not found")
return temp_file_path
async def transcribe_audio(file_path):
"""
음성 파일을 텍스트로 변환합니다.
Args:
file_path (str): 음성 파일 경로
Returns:
str: 변환된 텍스트
"""
client = OpenAI(api_key=settings.OPENAI_API_KEY)
with open(file_path, "rb") as audio_file:
transcription = client.audio.transcriptions.create(
model="whisper-1",
file=audio_file
)
return transcription.text- 파일: 'audio_process.py', 'download_s3_file.py'
- 사용된 라이브러리: EasyOCR, OpenCV
- 엔드 포인트: /image-to-text
- 기능: S3에 저장된 음성 파일을 다운로드하여 텍스트로 변환.
- 핵심 로직:
async def extract_text_from_image(image_path): """ 주어진 이미지에서 텍스트를 추출합니다. Args: image_path (str): 이미지 파일 경로 Returns: list[dict]: 추출된 텍스트와 확률 정보가 포함된 리스트 """ # EasyOCR 리더 객체 생성 reader = easyocr.Reader(['ko', 'en']) # 이미지 파일 읽기 image = cv2.imread(image_path) if image is None: raise FileNotFoundError(f"Image file not found: {image_path}") # 텍스트 추출 results = reader.readtext(image) # 결과 포맷팅 extracted_texts = [] for bbox, text, prob in results: extracted_texts.append({ "text": text, "probability": prob, "bounding_box": bbox }) return extracted_texts
-
RabbitMQ 기반 비동기 처리
- 기능: RabbitMQ 메시지 큐를 사용하여 AI 모델을 호출하고 결과 처리
- 핵심 로직:
def process_message(ch, method, properties, body): """ RabbitMQ 메시지 처리 콜백 함수. Args: ch: 채널 method: 메시지 전달 정보 properties: 메시지 속성 body: 메시지 본문 """ try: message = json.loads(body.decode('utf-8')) content = message.get("content") request_id = message.get("id") if not content or not request_id: raise ValueError("Invalid message: Missing 'content' or 'id'") execute_score_multi_and_callback(content, request_id) ch.basic_ack(delivery_tag=method.delivery_tag) except Exception as e: logger.error(f"Error processing message: {e}") ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
-
BackgroundTasks 처리
- 기능: FastAPI BackgroundTasks를 사용하여 비동기적으로 결과를 처리하고 콜백 URL에 POST 요청
- 핵심 로직:
@router.post("/judgement", status_code=202)
async def process_judge(request: JudgeRequest, background_tasks: BackgroundTasks):
"""
요청 데이터를 비동기적으로 처리하고 202 Accepted 응답 반환.
Args:
request (JudgeRequest): 요청 데이터
background_tasks (BackgroundTasks): 백그라운드 작업 관리 객체
Returns:
dict: 응답 상태
"""
background_tasks.add_task(execute_test_response_and_callback, request.content, request.id)
return {"status": "accepted", "message": "Judgement processing started."}- AI 및 NLP
- OpenAI GPT-4
- Hugging Face Transformers
- BERT 기반 감정 분석 모델
- BERT 기반 행동 분류 모델
- PyTorch
- 음성 및 이미지 처리
- OpenAI Whisper
- EasyOCR
- 백엔드 및 API
- FastAPI
- 고성능 비동기 API 서버
- HTTPx
- 비동기 HTTP client로 외부 API 통신
- pika
- RabbitMQ 메세지 큐 관리 및 작업 분배
- 데이터 관리 및 클라우드
- AWS S3
- 음성 및 이미지 데이터 저장 및 관리
- AWS S3
- FastAPI
- 팀 GitHub: KTB16Team
- 관련 문의는 Issues를 통해 남겨주세요.