Arag / app /services /formatter.py
AuthorBot
Simplify upsell stack and align tests with current prompt wiring.
a7271b7
Raw
History Blame Contribute Delete
3.75 kB
"""Author RAG Chatbot SaaS β€” Response Formatter & Link Injector.
Formats final responses and injects purchase links.
RULE: Max 2 links per response β€” never spam.
RULE: Max 2 short paragraphs per response (~75 words total).
RULE: Links rendered as buttons in the widget β€” never paste raw URLs in text.
"""
import re
import structlog
logger = structlog.get_logger(__name__)
class ResponseFormatter:
"""Formats responses and injects structured link data."""
MAX_LINKS = 2
MAX_PARAGRAPHS = 2
MAX_RESPONSE_CHARS = 380
PURCHASE_LABEL = "Buy Book"
PREVIEW_LABEL = "Read Preview"
def format(
self,
response_text: str,
upsell_hook: str | None = None,
purchase_url: str | None = None,
preview_url: str | None = None,
show_link: bool = False,
) -> dict:
"""Format a raw response into the final structured output."""
text = self._clean_response(response_text)
if upsell_hook and upsell_hook.strip() not in text:
text = text.rstrip() + "\n\n" + upsell_hook.strip()
links = []
if show_link:
if purchase_url:
links.append({
"label": self.PURCHASE_LABEL,
"url": purchase_url,
"type": "purchase",
"icon": "πŸ›’",
})
if preview_url and len(links) < self.MAX_LINKS:
links.append({
"label": self.PREVIEW_LABEL,
"url": preview_url,
"type": "preview",
"icon": "πŸ“–",
})
return {
"text": text,
"links": links[:self.MAX_LINKS],
"has_links": len(links) > 0,
}
def _clean_response(self, text: str) -> str:
"""Trim and clean a response to meet brevity guidelines."""
text = text.strip()
# Strip markdown artifacts the model sometimes emits
text = re.sub(r"\*\*(.+?)\*\*", r"\1", text)
text = re.sub(r"^[-*]\s+", "", text, flags=re.MULTILINE)
paragraphs = [p.strip() for p in re.split(r"\n{2,}", text) if p.strip()]
if len(paragraphs) > self.MAX_PARAGRAPHS:
paragraphs = paragraphs[:self.MAX_PARAGRAPHS]
logger.debug("Response trimmed to max paragraphs", count=self.MAX_PARAGRAPHS)
text = "\n\n".join(paragraphs)
if len(text) > self.MAX_RESPONSE_CHARS:
text = text[: self.MAX_RESPONSE_CHARS].rsplit(".", 1)[0].strip()
if text and not text.endswith("."):
text += "."
logger.debug("Response truncated to max chars")
return text
def format_book_selector(
self,
books: list[dict],
intro: str = "Select a book below to ask about it.",
) -> dict:
"""Format a book selector prompt response."""
return {
"text": intro,
"book_selector": [
{
"id": book["id"],
"title": book["title"],
"tagline": book.get("tagline") or "",
"cover_url": book.get("cover_path") or "",
}
for book in books
],
"has_links": False,
"links": [],
}
def purchase_only(
self,
text: str,
purchase_url: str | None,
preview_url: str | None = None,
) -> dict:
"""Format a short message with a mandatory purchase button."""
return self.format(
response_text=text,
purchase_url=purchase_url,
preview_url=preview_url,
show_link=bool(purchase_url),
)