| """LLM reply generation with grounded RAG answers.""" |
|
|
| from __future__ import annotations |
|
|
| import logging |
| from typing import Any |
|
|
| from app.config import get_settings |
|
|
| logger = logging.getLogger(__name__) |
|
|
|
|
| def build_system_prompt( |
| *, |
| business: str, |
| context: str, |
| lead_state: dict[str, Any], |
| ) -> str: |
| """Build the system prompt in a summary + context form.""" |
| system_prompt = ( |
| f"You are acting as ReplyFlow, the customer assistant " |
| f"for {business}. You are answering questions on " |
| f"{business}'s website and chat channels, particularly " |
| f"questions related to products, pricing, delivery, " |
| f"returns, and booking a sales call. " |
| f"Your responsibility is to represent {business} as " |
| f"faithfully as possible. " |
| f"You are given knowledge-base excerpts which you must " |
| f"use to answer questions. " |
| f"Be professional and engaging, as if talking to a " |
| f"potential customer. " |
| f"Answer ONLY using the provided Context. Never invent " |
| f"prices, policies, or facts. " |
| f"If you don't know the answer, say so clearly and offer " |
| f"to transfer to a human or book a call. " |
| f"If the user seems ready to buy or book, gently offer " |
| f"an appointment. " |
| f"Reply in English. Be concise and helpful." |
| ) |
|
|
| system_prompt += ( |
| f"\n\n## Business:\n{business}\n\n" |
| f"## Context:\n{context}\n\n" |
| f"## Lead profile so far:\n{lead_state}\n\n" |
| ) |
|
|
| system_prompt += ( |
| "With this context, please chat with the user, always " |
| f"staying in character as ReplyFlow for {business}." |
| ) |
| return system_prompt |
|
|
|
|
| async def generate_reply( |
| *, |
| question: str, |
| language: str, |
| context_chunks: list[dict[str, Any]], |
| history: list[dict[str, str]], |
| lead_state: dict[str, Any], |
| ) -> tuple[str, bool]: |
| """ |
| Generate a grounded reply. |
| |
| Returns (reply_text, grounded). |
| """ |
| settings = get_settings() |
| context_text = "\n\n".join( |
| f"[{i + 1}] {c['text']}" |
| for i, c in enumerate(context_chunks) |
| ) |
| grounded = bool(context_chunks) |
|
|
| if not grounded: |
| return _no_context_reply(), False |
|
|
| if settings.has_openai: |
| try: |
| return await _openai_reply( |
| question=question, |
| context_text=context_text, |
| history=history, |
| lead_state=lead_state, |
| ), True |
| except Exception as exc: |
| logger.exception("OpenAI failed: %s", exc) |
|
|
| return _demo_reply( |
| question=question, |
| context_chunks=context_chunks, |
| lead_state=lead_state, |
| ), True |
|
|
|
|
| async def _openai_reply( |
| *, |
| question: str, |
| context_text: str, |
| history: list[dict[str, str]], |
| lead_state: dict[str, Any], |
| ) -> str: |
| from openai import AsyncOpenAI |
|
|
| settings = get_settings() |
| client = AsyncOpenAI(api_key=settings.openai_api_key) |
| system = build_system_prompt( |
| business=settings.demo_business_name, |
| context=context_text, |
| lead_state=lead_state, |
| ) |
| messages: list[dict[str, str]] = [ |
| {"role": "system", "content": system}, |
| ] |
| for item in history[-8:]: |
| messages.append( |
| {"role": item["role"], "content": item["content"]} |
| ) |
| messages.append({"role": "user", "content": question}) |
|
|
| response = await client.chat.completions.create( |
| model=settings.openai_model, |
| messages=messages, |
| temperature=0.2, |
| max_tokens=500, |
| ) |
| return (response.choices[0].message.content or "").strip() |
|
|
|
|
| def _no_context_reply() -> str: |
| return ( |
| "I do not have that information in our knowledge " |
| "base, so I will not invent an answer. I can " |
| "transfer you to a human or help book a call." |
| ) |
|
|
|
|
| def _demo_reply( |
| *, |
| question: str, |
| context_chunks: list[dict[str, Any]], |
| lead_state: dict[str, Any], |
| ) -> str: |
| """Template reply for offline / no-key demo mode.""" |
| best = context_chunks[0]["text"] |
| excerpt = best[:420].rstrip() |
| if len(best) > 420: |
| excerpt += "…" |
|
|
| missing = [ |
| field |
| for field in ("name", "need", "budget", "contact_channel") |
| if not lead_state.get(field) |
| ] |
|
|
| followups = { |
| "name": "May I have your name?", |
| "need": "What do you need help with?", |
| "budget": "What budget range are you considering?", |
| "contact_channel": ( |
| "How should we reach you (email or phone)?" |
| ), |
| } |
| reply = f"Based on our documentation:\n\n{excerpt}" |
| if missing: |
| reply += f"\n\n{followups[missing[0]]}" |
| else: |
| reply += ( |
| "\n\nI can also propose available meeting slots " |
| "if you would like to book a call." |
| ) |
| return reply |
|
|
