File size: 5,440 Bytes
5889992 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 |
import textwrap
import asyncio
from typing import Dict, Any
import logging
from .base_agent import BaseAgent
from src.llm.models.schemas import EmotionalAnalysis
class EmotionAgent(BaseAgent):
def process(self, text: str) -> EmotionalAnalysis:
"""Process text for emotional content"""
prompt = self._construct_emotion_prompt(text)
response = self.llm.generate(prompt)
analysis = self._parse_emotion_response(response.content)
self._log_action(action="emotion_analysis", metadata={"text": text, "analysis": analysis}, level=logging.INFO)
return EmotionalAnalysis(
primary_emotion=analysis['primary_emotion'],
intensity=analysis['intensity'],
secondary_emotions=analysis['secondary_emotions'],
triggers=analysis['emotional_triggers'],
coping_strategies=analysis['coping_strategies'],
confidence_score=analysis['confidence_score']
)
def _construct_emotion_prompt(self, text: str) -> str:
emotion_prompt = f"""
Analyze the emotional content in the following text:
Text: {text}
Provide analysis in the following format:
1. Primary emotion: [single emotion]
2. Intensity: [number between 1 and 10]
3. Secondary emotions: [comma-separated list of emotions]
4. Emotional triggers: [comma-separated list of triggers]
5. Suggested coping strategies: [comma-separated list of strategies]
6. Confidence score: [number between 0 and 1]
Example:
1. Primary emotion: Anxiety
2. Intensity: 7
3. Secondary emotions: Fear, Worry
4. Emotional triggers: Work deadline, Family conflict
5. Suggested coping strategies: Deep breathing, Journaling, Talking to a friend
6. Confidence score: 0.8
"""
return textwrap.dedent(emotion_prompt).strip()
def _parse_emotion_response(self, response: str) -> dict:
try:
analysis = {
'primary_emotion': '',
'intensity': 0,
'secondary_emotions': [],
'emotional_triggers': [],
'coping_strategies': [],
'confidence_score': 0.0
}
for line in response.split('\n'):
# Convert the line to string explicitly in case it's not
line = str(line).strip()
if not line:
continue
# Split on first colon only
parts = line.split(':', 1)
if len(parts) != 2:
continue
self._log_action(action="emotion_analysis_debug", metadata={"line":line}, level=logging.DEBUG)
# Ensure key is a string before calling lower()
key = str(parts[0]).strip().lower() # Explicitly convert to string
value = str(parts[1]).strip()
self._log_action(action="emotion_analysis_debug", metadata={"line":line, "key": key, "value": value}, level=logging.DEBUG)
if 'primary emotion' in key:
analysis['primary_emotion'] = value
elif 'intensity' in key:
# Convert intensity to integer safely
try:
analysis['intensity'] = int(value.strip('[]'))
except ValueError:
analysis['intensity'] = 5 # default value
elif 'secondary emotions' in key:
analysis['secondary_emotions'] = [
s.strip() for s in value.split(',') if s.strip()
]
elif 'emotional triggers' in key:
analysis['emotional_triggers'] = [
t.strip() for t in value.split(',') if t.strip()
]
elif 'suggested coping strategies' in key:
analysis['coping_strategies'] = [
c.strip() for c in value.split(',') if c.strip()
]
elif 'confidence score' in key:
# Convert confidence score to float safely
try:
analysis['confidence_score'] = float(value.strip('[]'))
except ValueError:
analysis['confidence_score'] = 0.5
if not analysis['primary_emotion']:
raise ValueError("Primary emotion not found in response")
self._log_action(action="emotion_analysis_success", metadata={"response": response, "analysis": analysis}, level=logging.INFO)
return analysis
except Exception as e:
self._log_action(
action="emotion_analysis_error",
metadata={"response": str(response), "error": str(e)},
level=logging.ERROR
)
raise ValueError(f"Failed to parse emotion response: {str(e)}")
async def process_async(self, text: str) -> EmotionalAnalysis:
return await asyncio.get_event_loop().run_in_executor(
None,
lambda: self.process(text)
) |