[NOTICKET] Tuning prompt generate audio text
Browse files- src/agents/chatbot.py +59 -9
src/agents/chatbot.py
CHANGED
|
@@ -6,7 +6,6 @@ from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
|
|
| 6 |
from langchain_core.output_parsers import StrOutputParser
|
| 7 |
from src.config.settings import settings
|
| 8 |
from src.middlewares.logging import get_logger
|
| 9 |
-
from langchain_core.messages import HumanMessage, AIMessage
|
| 10 |
|
| 11 |
logger = get_logger("chatbot")
|
| 12 |
|
|
@@ -77,21 +76,72 @@ class ChatbotAgent:
|
|
| 77 |
logger.error("Response generation failed", error=str(e))
|
| 78 |
raise
|
| 79 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 80 |
async def generate_audio_text(self, full_response: str) -> str:
|
| 81 |
"""Generate a 2-3 sentence TTS-friendly summary of the assistant response."""
|
| 82 |
try:
|
|
|
|
| 83 |
prompt = (
|
| 84 |
-
"You are a text
|
| 85 |
-
"write a plain
|
| 86 |
-
"
|
| 87 |
-
"
|
| 88 |
-
"
|
| 89 |
f"Response:\n{full_response}\n\n"
|
| 90 |
-
"Summary
|
| 91 |
)
|
|
|
|
| 92 |
result = await self.llm.ainvoke(prompt)
|
| 93 |
-
|
| 94 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 95 |
except Exception as e:
|
| 96 |
logger.error("Audio text generation failed", error=str(e))
|
| 97 |
return ""
|
|
|
|
| 6 |
from langchain_core.output_parsers import StrOutputParser
|
| 7 |
from src.config.settings import settings
|
| 8 |
from src.middlewares.logging import get_logger
|
|
|
|
| 9 |
|
| 10 |
logger = get_logger("chatbot")
|
| 11 |
|
|
|
|
| 76 |
logger.error("Response generation failed", error=str(e))
|
| 77 |
raise
|
| 78 |
|
| 79 |
+
def language_hint(self, full_response: str) -> str:
|
| 80 |
+
text = full_response.lower()
|
| 81 |
+
words = set(re.findall(r"\b[\w']+\b", text, flags=re.UNICODE))
|
| 82 |
+
|
| 83 |
+
indo_markers = {
|
| 84 |
+
"yang", "dan", "untuk", "tidak", "akan", "saya", "kamu", "kita",
|
| 85 |
+
"mereka", "adalah", "ini", "itu", "dengan", "karena", "sebagai",
|
| 86 |
+
"oleh", "pada", "dari", "ke", "di"
|
| 87 |
+
}
|
| 88 |
+
|
| 89 |
+
eng_markers = {
|
| 90 |
+
"the", "and", "for", "you", "your", "i", "is", "are", "will",
|
| 91 |
+
"not", "this", "that", "with", "because", "as", "from", "to",
|
| 92 |
+
"in", "of"
|
| 93 |
+
}
|
| 94 |
+
|
| 95 |
+
indo_count = sum(1 for w in words if w in indo_markers)
|
| 96 |
+
eng_count = sum(1 for w in words if w in eng_markers)
|
| 97 |
+
|
| 98 |
+
if indo_count > eng_count and indo_count >= 2:
|
| 99 |
+
return "Indonesian"
|
| 100 |
+
if eng_count > indo_count and eng_count >= 2:
|
| 101 |
+
return "English"
|
| 102 |
+
|
| 103 |
+
if indo_count > 0 and eng_count == 0:
|
| 104 |
+
return "Indonesian"
|
| 105 |
+
if eng_count > 0 and indo_count == 0:
|
| 106 |
+
return "English"
|
| 107 |
+
|
| 108 |
+
return "the same language as the response"
|
| 109 |
+
|
| 110 |
async def generate_audio_text(self, full_response: str) -> str:
|
| 111 |
"""Generate a 2-3 sentence TTS-friendly summary of the assistant response."""
|
| 112 |
try:
|
| 113 |
+
lang = self.language_hint(full_response)
|
| 114 |
prompt = (
|
| 115 |
+
"You are a text to speech assistant. Given the following AI response, "
|
| 116 |
+
"write a plain language summary in exactly 2 or 3 sentences. "
|
| 117 |
+
"Output only the summary text. Allowed characters are letters numbers spaces and periods only. "
|
| 118 |
+
"Do not output any other characters. Do not name symbols. "
|
| 119 |
+
f"The response language is {lang}. Write the summary in {lang} only. Do not translate.\n\n"
|
| 120 |
f"Response:\n{full_response}\n\n"
|
| 121 |
+
"Summary:"
|
| 122 |
)
|
| 123 |
+
|
| 124 |
result = await self.llm.ainvoke(prompt)
|
| 125 |
+
logger.info(f"Generated audio text: {str(result)[:250]}...")
|
| 126 |
+
|
| 127 |
+
text = result.content if hasattr(result, "content") else str(result)
|
| 128 |
+
text = text.replace("!", ".").replace("?", ".").replace(";", ".").replace("*", "")
|
| 129 |
+
|
| 130 |
+
sentences = [s.strip() for s in text.split(".") if s.strip()][:3]
|
| 131 |
+
|
| 132 |
+
def sanitize(sentence: str) -> str:
|
| 133 |
+
sentence = re.sub(r"[^A-Za-z0-9 .]", " ", sentence)
|
| 134 |
+
sentence = re.sub(r"\s+", " ", sentence).strip()
|
| 135 |
+
return sentence
|
| 136 |
+
|
| 137 |
+
sanitized = [sanitize(s) for s in sentences if s]
|
| 138 |
+
if not sanitized:
|
| 139 |
+
return ""
|
| 140 |
+
|
| 141 |
+
output = ". ".join(sanitized).strip()
|
| 142 |
+
if output and not output.endswith("."):
|
| 143 |
+
output += "."
|
| 144 |
+
return output
|
| 145 |
except Exception as e:
|
| 146 |
logger.error("Audio text generation failed", error=str(e))
|
| 147 |
return ""
|