diff --git a/backend/app/alembic/versions/088_add_chatbot_session.py b/backend/app/alembic/versions/088_add_chatbot_session.py new file mode 100644 index 000000000..367a924ce --- /dev/null +++ b/backend/app/alembic/versions/088_add_chatbot_session.py @@ -0,0 +1,161 @@ +"""add chatbot session and message tables + +Revision ID: 088 +Revises: 087 +Create Date: 2026-10-20 00:00:00.000000 +""" + +import sqlalchemy as sa +import sqlmodel.sql.sqltypes +from alembic import op +from sqlalchemy.dialects import postgresql + +revision = "088" +down_revision = "087" +branch_labels = None +depends_on = None + + +def upgrade(): + op.create_table( + "chatbot_session", + sa.Column( + "id", + sa.Uuid(), + nullable=False, + comment="Session identifier returned to the caller", + ), + sa.Column( + "contact_ref", + sqlmodel.sql.sqltypes.AutoString(length=500), + nullable=False, + comment="Caller-side identifier of the end user (e.g. a contact id); opaque to Kaapi", + ), + sa.Column( + "config_id", + sa.Uuid(), + nullable=False, + comment="Saved Anthropic text config that supplies the chat model", + ), + sa.Column( + "config_version", + sa.Integer(), + nullable=True, + comment="Pinned config version; NULL resolves to the latest version on every turn", + ), + sa.Column( + "system_prompt", + sqlmodel.sql.sqltypes.AutoString(), + nullable=False, + comment="Admin-authored flow prompt captured at session creation", + ), + sa.Column( + "status", + sa.Enum( + "active", + "inactive", + name="chatbotsessionstatusenum", + ), + nullable=False, + comment="Conversation lifecycle: active or inactive", + ), + sa.Column( + "collected_fields", + postgresql.JSON(astext_type=sa.Text()), + nullable=False, + comment="Flat map of field name -> value gathered from the user so far", + ), + sa.Column( + "organization_id", + sa.Integer(), + nullable=False, + comment="Reference to the organization", + ), + sa.Column( + "project_id", + sa.Integer(), + nullable=False, + comment="Reference to the project", + ), + sa.Column( + "inserted_at", + sa.DateTime(), + nullable=False, + comment="Timestamp when the session was created", + ), + sa.Column( + "updated_at", + sa.DateTime(), + nullable=False, + comment="Timestamp when the session was last updated", + ), + sa.ForeignKeyConstraint(["config_id"], ["config.id"], ondelete="RESTRICT"), + sa.ForeignKeyConstraint( + ["organization_id"], ["organization.id"], ondelete="CASCADE" + ), + sa.ForeignKeyConstraint(["project_id"], ["project.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index( + op.f("ix_chatbot_session_config_id"), + "chatbot_session", + ["config_id"], + unique=False, + ) + op.create_index( + op.f("ix_chatbot_session_organization_id"), + "chatbot_session", + ["organization_id"], + unique=False, + ) + op.create_index( + op.f("ix_chatbot_session_project_id"), + "chatbot_session", + ["project_id"], + unique=False, + ) + + op.create_table( + "chatbot_session_message", + sa.Column("id", sa.Uuid(), nullable=False, comment="Unique identifier"), + sa.Column( + "session_id", + sa.Uuid(), + nullable=False, + comment="Reference to the owning chatbot session", + ), + sa.Column( + "role", + sa.Enum("system", "user", "assistant", name="chatbotmessageroleenum"), + nullable=False, + comment="Speaker: user or assistant (system is reserved; the prompt lives on the session)", + ), + sa.Column( + "content", + sqlmodel.sql.sqltypes.AutoString(), + nullable=False, + comment="Message text", + ), + sa.Column( + "inserted_at", + sa.DateTime(), + nullable=False, + comment="Timestamp when the message was stored; defines transcript order", + ), + sa.ForeignKeyConstraint( + ["session_id"], ["chatbot_session.id"], ondelete="CASCADE" + ), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index( + op.f("ix_chatbot_session_message_session_id"), + "chatbot_session_message", + ["session_id"], + unique=False, + ) + + +def downgrade(): + raise NotImplementedError( + "not reversible: dropping chatbot_session tables discards stored conversations" + ) diff --git a/backend/app/api/docs/chatbot/create_session.md b/backend/app/api/docs/chatbot/create_session.md new file mode 100644 index 000000000..543966e2d --- /dev/null +++ b/backend/app/api/docs/chatbot/create_session.md @@ -0,0 +1,5 @@ +Start a server-side chatbot conversation and return its `session_id`. + +The `system_prompt` and the `config_id` / `config_version` pin are stored on the session, so later turns only send the new user message. `config_id` must be a config in the caller's project whose completion is `provider: "anthropic"`, `type: "text"` with `params.model` set; when `config_version` is omitted, each turn uses the config's latest version at that time. + +`contact_ref` is the caller's own identifier for the end user and is stored as-is. Returns 404 if the config or version isn't found in the project and 422 if it isn't a usable Anthropic text config. diff --git a/backend/app/api/docs/chatbot/list_sessions.md b/backend/app/api/docs/chatbot/list_sessions.md new file mode 100644 index 000000000..202480359 --- /dev/null +++ b/backend/app/api/docs/chatbot/list_sessions.md @@ -0,0 +1,3 @@ +List every chatbot session for the given `contact_ref` in the caller's project, most recent first. + +`contact_ref` is matched exactly against the value supplied when the session was created. Each entry carries the session's current `status` and the `collected_fields` gathered so far. Returns an empty list when the contact has no sessions in this project. diff --git a/backend/app/api/docs/chatbot/send_session_message.md b/backend/app/api/docs/chatbot/send_session_message.md new file mode 100644 index 000000000..237e0fa97 --- /dev/null +++ b/backend/app/api/docs/chatbot/send_session_message.md @@ -0,0 +1,5 @@ +Queue one turn of a chatbot session; the bot reply is delivered asynchronously to `webhook_url`. + +The user `message` is stored immediately and the response is `202` with `status: "queued"`. Once the turn completes, Kaapi POSTs `{"success": true, "data": {"session_id", "reply", "status", "collected_fields"}}` to `webhook_url` (HMAC-signed when the project has a webhook secret). On a model or config failure it POSTs `{"success": false, "error": "...", "data": {"session_id"}}` instead. + +On the first call of a session, omit `message` to get the bot's opening greeting; after that `message` is required. `webhook_url` must be a public HTTPS URL. Returns 404 if the session isn't in the caller's project and 422 for an invalid `webhook_url` or a missing `message` on a started session. diff --git a/backend/app/api/routes/chatbot/__init__.py b/backend/app/api/routes/chatbot/__init__.py index d427b0256..151b59066 100644 --- a/backend/app/api/routes/chatbot/__init__.py +++ b/backend/app/api/routes/chatbot/__init__.py @@ -1,6 +1,8 @@ from fastapi import APIRouter from app.api.routes.chatbot.message import router as message_router +from app.api.routes.chatbot.session import router as session_router router = APIRouter(prefix="/chatbot", tags=["Chatbot"]) router.include_router(message_router) +router.include_router(session_router) diff --git a/backend/app/api/routes/chatbot/session.py b/backend/app/api/routes/chatbot/session.py new file mode 100644 index 000000000..0ea7d60f1 --- /dev/null +++ b/backend/app/api/routes/chatbot/session.py @@ -0,0 +1,165 @@ +from typing import Annotated +from uuid import UUID + +from asgi_correlation_id import correlation_id +from fastapi import APIRouter, Depends, HTTPException, Path, Query + +from app.api.deps import AuthContextDep, SessionDep +from app.api.permissions import Permission, require_permission +from app.celery.utils import start_chatbot_session_message +from app.crud.chatbot import ( + create_chatbot_session, + create_chatbot_session_message, + get_chatbot_session, + list_chatbot_session_messages, + list_chatbot_sessions_by_contact_ref, +) +from app.models import ( + CONTACT_REF_MAX_LENGTH, + ChatbotMessageRoleEnum, + ChatbotSessionSummary, + CreateChatbotSessionRequest, + CreateChatbotSessionResponse, + SendChatbotSessionMessageRequest, + SendChatbotSessionMessageResponse, +) +from app.services.chatbot.config import resolve_chatbot_model_settings +from app.services.chatbot.exceptions import ChatbotInputError +from app.utils import APIResponse, load_description, validate_callback_url + +DEFAULT_TRACE_ID = "N/A" + +router = APIRouter() + + +@router.post( + "/sessions", + status_code=201, + description=load_description("chatbot/create_session.md"), + response_model=APIResponse[CreateChatbotSessionResponse], + dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))], +) +def start_chatbot_session( + request: CreateChatbotSessionRequest, + session: SessionDep, + auth_context: AuthContextDep, +) -> APIResponse[CreateChatbotSessionResponse]: + project_id = auth_context.project_.id + try: + resolve_chatbot_model_settings( + session=session, + project_id=project_id, + config_id=request.config_id, + config_version=request.config_version, + max_tokens=None, + history_token_limit=None, + ) + except ChatbotInputError as e: + raise HTTPException(status_code=422, detail=str(e)) + + chatbot_session = create_chatbot_session( + session, + organization_id=auth_context.organization_.id, + project_id=project_id, + contact_ref=request.contact_ref, + config_id=request.config_id, + config_version=request.config_version, + system_prompt=request.system_prompt, + ) + return APIResponse.success_response( + CreateChatbotSessionResponse(session_id=chatbot_session.id) + ) + + +@router.get( + "/sessions", + description=load_description("chatbot/list_sessions.md"), + response_model=APIResponse[list[ChatbotSessionSummary]], + dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))], +) +def list_chatbot_sessions( + contact_ref: Annotated[ + str, + Query( + min_length=1, + max_length=CONTACT_REF_MAX_LENGTH, + description="Caller-side identifier of the end user whose sessions to list", + ), + ], + session: SessionDep, + auth_context: AuthContextDep, +) -> APIResponse[list[ChatbotSessionSummary]]: + chatbot_sessions = list_chatbot_sessions_by_contact_ref( + session, + contact_ref=contact_ref, + organization_id=auth_context.organization_.id, + project_id=auth_context.project_.id, + ) + summaries: list[ChatbotSessionSummary] = [] + for chatbot_session in chatbot_sessions: + summary = ChatbotSessionSummary( + session_id=chatbot_session.id, + contact_ref=chatbot_session.contact_ref, + status=chatbot_session.status, + collected_fields=chatbot_session.collected_fields, + config_id=chatbot_session.config_id, + config_version=chatbot_session.config_version, + inserted_at=chatbot_session.inserted_at, + updated_at=chatbot_session.updated_at, + ) + summaries.append(summary) + return APIResponse.success_response(summaries) + + +@router.post( + "/sessions/{session_id}/message", + status_code=202, + description=load_description("chatbot/send_session_message.md"), + response_model=APIResponse[SendChatbotSessionMessageResponse], + dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))], +) +def send_chatbot_session_message( + session_id: Annotated[UUID, Path(description="Chatbot session identifier")], + request: SendChatbotSessionMessageRequest, + session: SessionDep, + auth_context: AuthContextDep, +) -> APIResponse[SendChatbotSessionMessageResponse]: + organization_id = auth_context.organization_.id + project_id = auth_context.project_.id + + try: + validate_callback_url(request.webhook_url) + except ValueError as e: + raise HTTPException(status_code=422, detail=f"Invalid webhook_url: {e}") + + chatbot_session = get_chatbot_session( + session, + session_id=session_id, + organization_id=organization_id, + project_id=project_id, + ) + if chatbot_session is None: + raise HTTPException(status_code=404, detail="Chatbot session not found") + + if request.message is None: + if list_chatbot_session_messages(session, session_id=session_id): + raise HTTPException( + status_code=422, + detail="`message` is required once the session has started; omit it only on the first call.", + ) + else: + create_chatbot_session_message( + session, + session_id=session_id, + role=ChatbotMessageRoleEnum.USER, + content=request.message, + ) + + start_chatbot_session_message( + session_id=str(session_id), + organization_id=organization_id, + project_id=project_id, + webhook_url=request.webhook_url, + trace_id=correlation_id.get() or DEFAULT_TRACE_ID, + ) + return APIResponse.success_response(SendChatbotSessionMessageResponse()) diff --git a/backend/app/celery/tasks/job_execution.py b/backend/app/celery/tasks/job_execution.py index 1288876a8..968b88d5b 100644 --- a/backend/app/celery/tasks/job_execution.py +++ b/backend/app/celery/tasks/job_execution.py @@ -3,7 +3,8 @@ All tasks share one queue (`default`, declared with `x-max-priority=10`) and are ordered by the per-task `priority`: - 9 LLM call + LLM chain (run_llm_job, run_llm_chain_job, run_response_job) + 9 LLM call + LLM chain (run_llm_job, run_llm_chain_job, run_response_job, + run_chatbot_session_message) 6 Fast evaluation (run_evaluation_fast_chunk, run_evaluation_fast_aggregate, run_prompt_improvement, run_evaluation_iteration_graph_step) 2 Everything else (doctransform, collections, STT/TTS evaluation, assessment) @@ -15,6 +16,7 @@ import logging from collections.abc import Callable from typing import TYPE_CHECKING, TypeVar +from uuid import UUID from asgi_correlation_id import correlation_id from celery import Task, current_task @@ -567,3 +569,29 @@ def send_eval_completion_callback( ) return execute_eval_completion_callback(evaluation_id=evaluation_id) + + +@celery_app.task(bind=True, queue="default", priority=9) +@gevent_timeout(settings.CELERY_TASK_SOFT_TIME_LIMIT, "run_chatbot_session_message") +def run_chatbot_session_message( + self: Task, + session_id: str, + organization_id: int, + project_id: int, + webhook_url: str, + trace_id: str, +) -> None: + """Idempotency: the service skips the turn if the transcript already ends + on an assistant reply, so a redelivery doesn't re-bill or double-reply.""" + from app.services.chatbot.session import process_chatbot_session_message + + _set_trace(trace_id) + return _run_with_otel_parent( + self, + lambda: process_chatbot_session_message( + session_id=UUID(session_id), + organization_id=organization_id, + project_id=project_id, + webhook_url=webhook_url, + ), + ) diff --git a/backend/app/celery/utils.py b/backend/app/celery/utils.py index 3bd205d22..88285c0bb 100644 --- a/backend/app/celery/utils.py +++ b/backend/app/celery/utils.py @@ -337,6 +337,30 @@ def start_evaluation_iteration_round( return task_id +def start_chatbot_session_message( + session_id: str, + organization_id: int, + project_id: int, + webhook_url: str, + trace_id: str = "N/A", +) -> str: + from app.celery.tasks.job_execution import run_chatbot_session_message + + task_id = _enqueue_with_trace_context( + run_chatbot_session_message, + session_id=session_id, + organization_id=organization_id, + project_id=project_id, + webhook_url=webhook_url, + trace_id=trace_id, + ) + logger.info( + f"[start_chatbot_session_message] Enqueued chatbot turn | " + f"session_id: {session_id}, task_id: {task_id}" + ) + return task_id + + def get_task_status(task_id: str) -> Dict[str, Any]: result = AsyncResult(task_id) return { diff --git a/backend/app/crud/chatbot.py b/backend/app/crud/chatbot.py new file mode 100644 index 000000000..3b1d10304 --- /dev/null +++ b/backend/app/crud/chatbot.py @@ -0,0 +1,114 @@ +import logging +from uuid import UUID + +from sqlmodel import Session, col, select + +from app.core.util import now +from app.models import ( + ChatbotMessageRoleEnum, + ChatbotSession, + ChatbotSessionMessage, +) + +logger = logging.getLogger(__name__) + + +def create_chatbot_session( + session: Session, + *, + organization_id: int, + project_id: int, + contact_ref: str, + config_id: UUID, + config_version: int | None, + system_prompt: str, +) -> ChatbotSession: + timestamp = now() + chatbot_session = ChatbotSession( + organization_id=organization_id, + project_id=project_id, + contact_ref=contact_ref, + config_id=config_id, + config_version=config_version, + system_prompt=system_prompt, + inserted_at=timestamp, + updated_at=timestamp, + ) + session.add(chatbot_session) + session.commit() + session.refresh(chatbot_session) + logger.info( + f"[create_chatbot_session] Chatbot session created | " + f"session_id: {chatbot_session.id}, project_id: {project_id}, config_id: {config_id}" + ) + return chatbot_session + + +def get_chatbot_session( + session: Session, + *, + session_id: UUID, + organization_id: int, + project_id: int, +) -> ChatbotSession | None: + statement = select(ChatbotSession).where( + ChatbotSession.id == session_id, + ChatbotSession.organization_id == organization_id, + ChatbotSession.project_id == project_id, + ) + return session.exec(statement).first() + + +def create_chatbot_session_message( + session: Session, + *, + session_id: UUID, + role: ChatbotMessageRoleEnum, + content: str, +) -> ChatbotSessionMessage: + message = ChatbotSessionMessage( + session_id=session_id, + role=role, + content=content, + inserted_at=now(), + ) + session.add(message) + session.commit() + session.refresh(message) + logger.info( + f"[create_chatbot_session_message] Chatbot message stored | " + f"session_id: {session_id}, role: {role.value}, message_id: {message.id}" + ) + return message + + +def list_chatbot_session_messages( + session: Session, *, session_id: UUID +) -> list[ChatbotSessionMessage]: + # Unpaginated on purpose: the engine needs the whole transcript each turn + # and summarizes long histories itself. + statement = ( + select(ChatbotSessionMessage) + .where(ChatbotSessionMessage.session_id == session_id) + .order_by(col(ChatbotSessionMessage.inserted_at).asc()) + ) + return list(session.exec(statement).all()) + + +def list_chatbot_sessions_by_contact_ref( + session: Session, + *, + contact_ref: str, + organization_id: int, + project_id: int, +) -> list[ChatbotSession]: + statement = ( + select(ChatbotSession) + .where( + ChatbotSession.contact_ref == contact_ref, + ChatbotSession.organization_id == organization_id, + ChatbotSession.project_id == project_id, + ) + .order_by(col(ChatbotSession.inserted_at).desc()) + ) + return list(session.exec(statement).all()) diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index e13fe062a..dc7f2e3be 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -36,10 +36,19 @@ BatchJobUpdate, ) from .chatbot import ( + CONTACT_REF_MAX_LENGTH, ChatbotMessage, ChatbotMessageRoleEnum, + ChatbotSession, + ChatbotSessionMessage, + ChatbotSessionStatusEnum, + ChatbotSessionSummary, ChatbotTurnRequest, ChatbotTurnResponse, + CreateChatbotSessionRequest, + CreateChatbotSessionResponse, + SendChatbotSessionMessageRequest, + SendChatbotSessionMessageResponse, ) from .collection import ( Collection, diff --git a/backend/app/models/chatbot.py b/backend/app/models/chatbot.py index 3797fe43e..a5cfe381b 100644 --- a/backend/app/models/chatbot.py +++ b/backend/app/models/chatbot.py @@ -1,8 +1,14 @@ +from datetime import datetime from enum import StrEnum -from uuid import UUID +from typing import Any +from uuid import UUID, uuid4 +from sqlalchemy import JSON, Column +from sqlalchemy.dialects.postgresql import ENUM from sqlmodel import Field, SQLModel +from app.core.util import now + class ChatbotMessageRoleEnum(StrEnum): SYSTEM = "system" @@ -53,3 +59,189 @@ class ChatbotTurnResponse(SQLModel): messages: list[ChatbotMessage] = Field( description="Full updated transcript (system prompt + every turn, including the new reply); resend it as `history` next call.", ) + + +class ChatbotSessionStatusEnum(StrEnum): + ACTIVE = "active" + INACTIVE = "inactive" + + +_SESSION_STATUS_PG_ENUM = ENUM( + ChatbotSessionStatusEnum, + name="chatbotsessionstatusenum", + values_callable=lambda enum_cls: [member.value for member in enum_cls], + create_type=False, +) +_MESSAGE_ROLE_PG_ENUM = ENUM( + ChatbotMessageRoleEnum, + name="chatbotmessageroleenum", + values_callable=lambda enum_cls: [member.value for member in enum_cls], + create_type=False, +) + +CONTACT_REF_MAX_LENGTH = 500 + + +class ChatbotSession(SQLModel, table=True): + __tablename__ = "chatbot_session" + + id: UUID = Field( + default_factory=uuid4, + primary_key=True, + sa_column_kwargs={"comment": "Session identifier returned to the caller"}, + ) + contact_ref: str = Field( + max_length=CONTACT_REF_MAX_LENGTH, + sa_column_kwargs={ + "comment": "Caller-side identifier of the end user (e.g. a Glific contact id); opaque to Kaapi" + }, + ) + config_id: UUID = Field( + foreign_key="config.id", + nullable=False, + index=True, + ondelete="RESTRICT", + sa_column_kwargs={ + "comment": "Saved Anthropic text config that supplies the chat model" + }, + ) + config_version: int | None = Field( + default=None, + ge=1, + sa_column_kwargs={ + "comment": "Pinned config version; NULL resolves to the latest version on every turn" + }, + ) + system_prompt: str = Field( + sa_column_kwargs={ + "comment": "Admin-authored flow prompt captured at session creation" + }, + ) + status: ChatbotSessionStatusEnum = Field( + default=ChatbotSessionStatusEnum.ACTIVE, + sa_column=Column( + _SESSION_STATUS_PG_ENUM, + nullable=False, + comment="Conversation lifecycle: active or inactive", + ), + ) + collected_fields: dict[str, Any] = Field( + default_factory=dict, + sa_column=Column( + JSON, + nullable=False, + comment="Flat map of field name -> value gathered from the user so far", + ), + ) + organization_id: int = Field( + foreign_key="organization.id", + nullable=False, + index=True, + ondelete="CASCADE", + sa_column_kwargs={"comment": "Reference to the organization"}, + ) + project_id: int = Field( + foreign_key="project.id", + nullable=False, + index=True, + ondelete="CASCADE", + sa_column_kwargs={"comment": "Reference to the project"}, + ) + inserted_at: datetime = Field( + default_factory=now, + nullable=False, + sa_column_kwargs={"comment": "Timestamp when the session was created"}, + ) + updated_at: datetime = Field( + default_factory=now, + nullable=False, + sa_column_kwargs={ + "comment": "Timestamp when the session was last updated", + "onupdate": now, + }, + ) + + +class ChatbotSessionMessage(SQLModel, table=True): + __tablename__ = "chatbot_session_message" + + id: UUID = Field( + default_factory=uuid4, + primary_key=True, + sa_column_kwargs={"comment": "Unique identifier"}, + ) + session_id: UUID = Field( + foreign_key="chatbot_session.id", + nullable=False, + index=True, + ondelete="CASCADE", + sa_column_kwargs={"comment": "Reference to the owning chatbot session"}, + ) + role: ChatbotMessageRoleEnum = Field( + sa_column=Column( + _MESSAGE_ROLE_PG_ENUM, + nullable=False, + comment="Speaker: user or assistant (system is reserved; the prompt lives on the session)", + ), + ) + content: str = Field( + sa_column_kwargs={"comment": "Message text"}, + ) + inserted_at: datetime = Field( + default_factory=now, + nullable=False, + sa_column_kwargs={ + "comment": "Timestamp when the message was stored; defines transcript order" + }, + ) + + +class CreateChatbotSessionRequest(SQLModel): + config_id: UUID = Field( + description="Saved Kaapi config (Anthropic text completion) that supplies the model and reasoning settings.", + ) + config_version: int | None = Field( + default=None, + ge=1, + description="Version of `config_id` to pin; omit to use the latest version on every turn.", + ) + contact_ref: str = Field( + min_length=1, + max_length=CONTACT_REF_MAX_LENGTH, + description="Caller-side identifier of the end user this conversation is with.", + ) + system_prompt: str = Field( + min_length=1, + description="Admin-authored flow prompt (goal, fields to collect, rules); stored for the session's lifetime.", + ) + + +class CreateChatbotSessionResponse(SQLModel): + session_id: UUID + + +class ChatbotSessionSummary(SQLModel): + session_id: UUID + contact_ref: str + status: ChatbotSessionStatusEnum + collected_fields: dict[str, Any] + config_id: UUID + config_version: int | None + inserted_at: datetime + updated_at: datetime + + +class SendChatbotSessionMessageRequest(SQLModel): + message: str | None = Field( + default=None, + min_length=1, + description="The new user message; omit only on the first call to get the opening greeting.", + ) + webhook_url: str = Field( + min_length=1, + description="HTTPS URL the bot reply is POSTed to once the turn completes.", + ) + + +class SendChatbotSessionMessageResponse(SQLModel): + status: str = "queued" diff --git a/backend/app/services/chatbot/session.py b/backend/app/services/chatbot/session.py new file mode 100644 index 000000000..06f899308 --- /dev/null +++ b/backend/app/services/chatbot/session.py @@ -0,0 +1,199 @@ +import asyncio +import logging +from typing import Any +from uuid import UUID + +from fastapi import HTTPException +from sqlmodel import Session + +from app.core.db import engine +from app.crud.chatbot import ( + create_chatbot_session_message, + get_chatbot_session, + list_chatbot_session_messages, +) +from app.models import ChatbotMessageRoleEnum, ChatbotSessionMessage +from app.services.chatbot.config import resolve_chatbot_model_settings +from app.services.chatbot.engine import run_chatbot_turn +from app.services.chatbot.exceptions import ChatbotInputError, ChatbotProviderError +from app.services.chatbot.types import ChatbotMessageDict +from app.services.chatbot.utils import ChatbotModelSettings +from app.utils import get_webhook_secret, send_callback + +logger = logging.getLogger(__name__) + + +def _split_pending_turn( + stored_messages: list[ChatbotSessionMessage], +) -> tuple[list[ChatbotMessageDict], str | None]: + """Split the stored transcript into (prior history, pending user message). + + The route persists the user's message before enqueueing, so it is the last + stored row; the engine wants it separately from the history before it. + """ + history: list[ChatbotMessageDict] = [] + for stored in stored_messages: + # The system prompt lives on the session row and is passed separately. + if stored.role == ChatbotMessageRoleEnum.SYSTEM: + continue + history.append({"role": stored.role.value, "content": stored.content}) + + if not history: + return history, None + if history[-1]["role"] != ChatbotMessageRoleEnum.USER: + # No pending user turn; the engine rejects this with ChatbotInputError. + return history, None + pending = history.pop() + return history, pending["content"] + + +def _extract_reply(transcript: list[ChatbotMessageDict]) -> str: + for entry in reversed(transcript): + if entry["role"] == ChatbotMessageRoleEnum.ASSISTANT: + return entry["content"] + raise ChatbotInputError("[KAAPI] Chat model returned no assistant reply.") + + +def _send_failure_webhook( + *, + session_id: UUID, + webhook_url: str, + webhook_secret: str | None, + error: str, +) -> None: + payload: dict[str, Any] = { + "success": False, + "error": error, + "data": {"session_id": str(session_id)}, + } + delivered = send_callback(webhook_url, payload, webhook_secret) + logger.info( + f"[_send_failure_webhook] Failure webhook dispatched | " + f"session_id: {session_id}, delivered: {delivered}" + ) + + +def process_chatbot_session_message( + *, + session_id: UUID, + organization_id: int, + project_id: int, + webhook_url: str, +) -> None: + """Run the pending chatbot turn for a session and POST the reply to `webhook_url`. + + Idempotent under Celery redelivery: if the transcript already ends on an + assistant turn, the pending turn was answered and nothing is re-run. + """ + webhook_secret = get_webhook_secret(project_id, organization_id) + + # The DB session is closed before the LLM call so a slow provider doesn't + # pin a pooled connection for the whole turn. + with Session(engine) as session: + chatbot_session = get_chatbot_session( + session, + session_id=session_id, + organization_id=organization_id, + project_id=project_id, + ) + if chatbot_session is None: + logger.error( + f"[process_chatbot_session_message] Chatbot session not found | " + f"session_id: {session_id}, project_id: {project_id}" + ) + return + + stored_messages = list_chatbot_session_messages(session, session_id=session_id) + if ( + stored_messages + and stored_messages[-1].role == ChatbotMessageRoleEnum.ASSISTANT + ): + logger.warning( + f"[process_chatbot_session_message] Turn already answered, skipping | " + f"session_id: {session_id}" + ) + return + + system_prompt = chatbot_session.system_prompt + session_status = chatbot_session.status.value + collected_fields = dict(chatbot_session.collected_fields) + + try: + model_settings: ChatbotModelSettings = resolve_chatbot_model_settings( + session=session, + project_id=project_id, + config_id=chatbot_session.config_id, + config_version=chatbot_session.config_version, + max_tokens=None, + history_token_limit=None, + ) + except HTTPException as e: + # The config was resolvable at session creation; it has since been + # deleted or lost its version, and the caller only hears via webhook. + logger.warning( + f"[process_chatbot_session_message] Config no longer resolvable | " + f"session_id: {session_id}, detail: {e.detail}" + ) + _send_failure_webhook( + session_id=session_id, + webhook_url=webhook_url, + webhook_secret=webhook_secret, + error=str(e.detail), + ) + return + except ChatbotInputError as e: + _send_failure_webhook( + session_id=session_id, + webhook_url=webhook_url, + webhook_secret=webhook_secret, + error=str(e), + ) + return + + history, user_message = _split_pending_turn(stored_messages) + logger.info( + f"[process_chatbot_session_message] Running turn | session_id: {session_id}, " + f"history_count: {len(history)}, has_user_message: {user_message is not None}" + ) + + try: + transcript = asyncio.run( + run_chatbot_turn( + system_prompt=system_prompt, + history=history, + user_message=user_message, + model_settings=model_settings, + ) + ) + reply = _extract_reply(transcript) + except (ChatbotInputError, ChatbotProviderError) as e: + _send_failure_webhook( + session_id=session_id, + webhook_url=webhook_url, + webhook_secret=webhook_secret, + error=str(e), + ) + return + + with Session(engine) as session: + create_chatbot_session_message( + session, + session_id=session_id, + role=ChatbotMessageRoleEnum.ASSISTANT, + content=reply, + ) + + payload: dict[str, Any] = { + "success": True, + "data": { + "session_id": str(session_id), + "reply": reply, + "status": session_status, + "collected_fields": collected_fields, + }, + } + delivered = send_callback(webhook_url, payload, webhook_secret) + logger.info( + f"[process_chatbot_session_message] Turn completed | session_id: {session_id}, " + f"reply_length: {len(reply)}, webhook_delivered: {delivered}" + ) diff --git a/docs/wiki/modules/chatbot.md b/docs/wiki/modules/chatbot.md index 261046755..8023c2521 100644 --- a/docs/wiki/modules/chatbot.md +++ b/docs/wiki/modules/chatbot.md @@ -6,6 +6,7 @@ All paths relative to `backend/app/`. ## Routes - `api/routes/chatbot.py` — `POST /chatbot/message` (one synchronous turn; requires project scope) +- `api/routes/chatbot/session.py` — `GET /chatbot/sessions?contact_ref=` (all sessions for a contact in the caller's project, newest first → `list[ChatbotSessionSummary]`) ## Tables (SQLModel) None of its own. Request/response shapes only: `models/chatbot.py` (`ChatbotMessage`, `ChatbotMessageRoleEnum`, `ChatbotTurnRequest`, `ChatbotTurnResponse`). Reads `config` / `config_version` (see the config module) to resolve the model. @@ -14,6 +15,7 @@ None of its own. Request/response shapes only: `models/chatbot.py` (`ChatbotMess - `services/chatbot/config.py` — `resolve_chatbot_model_settings` (saved config + request overrides → `ChatbotModelSettings`; 404 via `ConfigVersionCrud`, 422 for non-Anthropic/non-text/no-model configs) - `services/chatbot/engine.py` — `run_chatbot_turn` (dict <-> LangChain message conversion, error mapping), `build_chatbot_graph`, history summarization - `services/chatbot/utils.py` — `ChatbotModelSettings`, defaults (`DEFAULT_MAX_OUTPUT_TOKENS`, `DEFAULT_HISTORY_TOKEN_LIMIT`, `FLOW_KEEP_RECENT_MESSAGES`), `get_chat_model` +- `crud/chatbot.py` — `list_chatbot_sessions_by_contact_ref` (org + project scoped, `inserted_at` desc, unpaginated) - `crud/config/version.py` — `ConfigVersionCrud.read_latest` (used when `config_version` is omitted) ## Async