| """Core chat orchestration service.""" |
|
|
| from __future__ import annotations |
|
|
| from typing import Any, Optional |
|
|
| from sqlalchemy import select |
| from sqlalchemy.ext.asyncio import AsyncSession |
| from sqlalchemy.orm import selectinload |
|
|
| from app.models import ( |
| Conversation, |
| Lead, |
| Message, |
| UnansweredQuestion, |
| ) |
| from app.services import calcom, handoff, leads, llm |
| from app.services.language import detect_language |
| from app.services.rag import get_knowledge_store |
|
|
| BOOKING_HINTS = ( |
| "book", |
| "appointment", |
| "schedule", |
| "meeting", |
| "slot", |
| "rendez-vous", |
| "rdv", |
| "créneau", |
| "запис", |
| "встреч", |
| "слот", |
| ) |
|
|
|
|
| GREETINGS = { |
| "en": ( |
| "Hello! I am ReplyFlow for Nordic Paper Solutions. " |
| "Ask about products, pricing, delivery, or book a call." |
| ), |
| "fr": ( |
| "Bonjour ! Je suis ReplyFlow pour Nordic Paper " |
| "Solutions. Posez vos questions sur les produits, " |
| "tarifs, livraison, ou réservez un appel." |
| ), |
| "ru": ( |
| "Здравствуйте! Я ReplyFlow для Nordic Paper " |
| "Solutions. Спросите о продуктах, ценах, доставке " |
| "или запишитесь на звонок." |
| ), |
| } |
|
|
|
|
| async def start_conversation( |
| db: AsyncSession, |
| *, |
| channel: str = "web", |
| ui_language: str = "en", |
| external_id: Optional[str] = None, |
| ) -> tuple[Conversation, str]: |
| """Create conversation + empty lead + greeting.""" |
| lang = ui_language if ui_language in GREETINGS else "en" |
| conv = Conversation( |
| channel=channel, |
| language=lang, |
| external_id=external_id, |
| ) |
| db.add(conv) |
| await db.flush() |
| db.add(Lead(conversation_id=conv.id)) |
| greeting = GREETINGS[lang] |
| db.add( |
| Message( |
| conversation_id=conv.id, |
| role="assistant", |
| content=greeting, |
| language=lang, |
| grounded=True, |
| ) |
| ) |
| await db.commit() |
| await db.refresh(conv) |
| return conv, greeting |
|
|
|
|
| async def handle_user_message( |
| db: AsyncSession, |
| *, |
| conversation_id: int, |
| message: str, |
| ui_language: str = "en", |
| ) -> dict[str, Any]: |
| """Process one user turn end-to-end.""" |
| result = await db.execute( |
| select(Conversation) |
| .where(Conversation.id == conversation_id) |
| .options( |
| selectinload(Conversation.messages), |
| selectinload(Conversation.lead), |
| ) |
| ) |
| conv = result.scalar_one() |
| language = detect_language(message, default=ui_language) |
| conv.language = language |
|
|
| previous_bot = None |
| for msg in reversed(conv.messages): |
| if msg.role == "assistant": |
| previous_bot = msg.content |
| break |
|
|
| lead = conv.lead or Lead(conversation_id=conv.id) |
| if conv.lead is None: |
| db.add(lead) |
| await db.flush() |
| leads.update_lead_from_message(lead, message, previous_bot) |
|
|
| db.add( |
| Message( |
| conversation_id=conv.id, |
| role="user", |
| content=message, |
| language=language, |
| ) |
| ) |
|
|
| wants_booking = any( |
| h in message.lower() for h in BOOKING_HINTS |
| ) |
| store = get_knowledge_store() |
| chunks = await store.search(message, top_k=4) |
| grounded = bool(chunks) |
|
|
| if not grounded: |
| conv.failure_streak += 1 |
| db.add( |
| UnansweredQuestion( |
| conversation_id=conv.id, |
| question=message, |
| ) |
| ) |
| else: |
| conv.failure_streak = 0 |
|
|
| history = [ |
| {"role": m.role, "content": m.content} |
| for m in conv.messages |
| if m.role in {"user", "assistant"} |
| ] |
| lead_state = leads.lead_to_dict(lead) |
|
|
| do_handoff, reason = handoff.should_handoff( |
| message=message, |
| failure_streak=conv.failure_streak, |
| grounded=grounded, |
| ) |
|
|
| suggested_slots = None |
| reply: str |
| if do_handoff: |
| conv.handed_off = True |
| conv.status = "handed_off" |
| reply = handoff.handoff_message(language, reason) |
| await handoff.notify_handoff( |
| conversation_id=conv.id, |
| reason=reason, |
| history=history + [{"role": "user", "content": message}], |
| lead=lead_state, |
| ) |
| else: |
| reply, grounded = await llm.generate_reply( |
| question=message, |
| language=language, |
| context_chunks=chunks, |
| history=history, |
| lead_state=lead_state, |
| ) |
| if wants_booking or ( |
| lead.complete and "book" in reply.lower() |
| ): |
| suggested_slots = await calcom.get_available_slots(3) |
| offer = { |
| "en": "Here are 3 available slots:", |
| "fr": "Voici 3 créneaux disponibles :", |
| "ru": "Вот 3 свободных слота:", |
| } |
| reply += "\n\n" + offer.get(language, offer["en"]) |
| for i, slot in enumerate(suggested_slots, start=1): |
| reply += f"\n{i}. {slot['label']}" |
|
|
| if grounded and not conv.handed_off: |
| conv.resolved_without_human = True |
|
|
| db.add( |
| Message( |
| conversation_id=conv.id, |
| role="assistant", |
| content=reply, |
| language=language, |
| grounded=grounded if not do_handoff else None, |
| ) |
| ) |
| conv.updated_at = conv.updated_at |
| await db.commit() |
|
|
| return { |
| "conversation_id": conv.id, |
| "reply": reply, |
| "language": language, |
| "grounded": grounded and not do_handoff, |
| "lead": leads.lead_to_dict(lead), |
| "suggested_slots": suggested_slots, |
| "handed_off": conv.handed_off, |
| "appointment_booked": conv.appointment_booked, |
| } |
|
|
