Chat

chat_stream(request, chat_service, assistant_service, user) async

Stream chat response for a session.

Supports SSE (Server-Sent Events) via NDJSON or text/event-stream.

Parameters:

• request (ChatRequest) –

  The chat request payload.

• chat_service (Annotated[ChatService, Depends(get_chat_service)]) –

  The chat service instance.

• assistant_service (Annotated[AssistantService, Depends(get_assistant_service)]) –

  The assistant service instance.

• user (Annotated[Optional[User], Depends(get_optional_user)]) –

  The optional authenticated user.

Returns:

• StreamingResponse –

  A StreamingResponse yielding the chat stream.

Raises:

• EntityNotFound –

  If the assistant is not found.

• HTTPException –

  If authentication is required but user is not authenticated.

• TechnicalError –

  If an unexpected error occurs during initialization.

Source code in app/api/v1/endpoints/chat.py

@router.post("/stream")
async def chat_stream(
    request: ChatRequest,
    chat_service: Annotated[ChatService, Depends(get_chat_service)],
    assistant_service: Annotated[AssistantService, Depends(get_assistant_service)],
    user: Annotated[Optional[User], Depends(get_optional_user)],
) -> StreamingResponse:
    """
    Stream chat response for a session.

    Supports SSE (Server-Sent Events) via NDJSON or text/event-stream.

    Args:
        request: The chat request payload.
        chat_service: The chat service instance.
        assistant_service: The assistant service instance.
        user: The optional authenticated user.

    Returns:
        A StreamingResponse yielding the chat stream.

    Raises:
        EntityNotFound: If the assistant is not found.
        HTTPException: If authentication is required but user is not authenticated.
        TechnicalError: If an unexpected error occurs during initialization.
    """
    try:
        # 1. Fetch Assistant (raw SQLAlchemy model for ChatService)
        assistant_uuid = request.assistant_id
        assistant = await assistant_service.get_assistant_model(assistant_uuid)
        if not assistant:
            raise EntityNotFound("Assistant not found")

        # 2. Authorization Check
        user_id = str(user.id) if user else None
        if assistant.user_authentication and not user_id:
            logger.warning(f"Access denied for assistant {assistant.id}: User Authentication Required")
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Authentication required for this assistant",
            )

        # 3. Stream
        logger.info(f"Init Chat Stream | Session: {request.session_id} | User: {user_id}")

        # Standard RAG Stream (with Vanna routing via SQL connector type)
        return StreamingResponse(
            chat_service.stream_chat(
                request.message,
                assistant,
                request.session_id,
                language=request.language,
                history=request.history,
                user_id=user_id,
            ),
            media_type="application/x-ndjson",
        )

    except (FunctionalError, EntityNotFound, HTTPException):
        raise
    except Exception as e:
        logger.error(f"Chat stream initialization failed: {e}", exc_info=True)
        raise TechnicalError(f"Chat stream initialization failed: {e}")

get_chat_history(session_id, chat_service, user) async

Retrieve conversation history for a specific session ID.

Parameters:

• session_id (str) –

  The session ID to retrieve history for.

• chat_service (Annotated[ChatService, Depends(get_chat_service)]) –

  The chat service instance.

• user (Annotated[Optional[User], Depends(get_optional_user)]) –

  The optional authenticated user.

Returns:

• Dict[str, List[Dict[str, Any]]] –

  A list of formatted chat messages.

Source code in app/api/v1/endpoints/chat.py

@router.get("/{session_id}/history")
async def get_chat_history(
    session_id: str,
    chat_service: Annotated[ChatService, Depends(get_chat_service)],
    user: Annotated[Optional[User], Depends(get_optional_user)],
) -> Dict[str, List[Dict[str, Any]]]:
    """
    Retrieve conversation history for a specific session ID.

    Args:
        session_id: The session ID to retrieve history for.
        chat_service: The chat service instance.
        user: The optional authenticated user.

    Returns:
        A list of formatted chat messages.
    """
    try:
        # Get history from the chat repository (Postgres) instead of Redis for full persistence
        messages = await chat_service.chat_repository.get_messages(session_id)

        # Transform to frontend-compatible format
        formatted_messages = []
        for msg in messages:
            m = {
                "id": str(msg.id),  # Use actual message UUID
                "text": msg.content,
                "sender": "user" if msg.role == "user" else "bot",
            }
            # Include metadata (visualization, sources, etc.)
            if msg.metadata_:
                meta = msg.metadata_.copy()

                # Format Sources for Frontend (Source[] interface)
                if "sources" in meta and isinstance(meta["sources"], list):
                    meta["sources"] = [_format_source(s) for s in meta["sources"]]

                # Format Steps (Ensure labels are present)
                if "steps" in meta and isinstance(meta["steps"], list):
                    meta["steps"] = [_format_step(step) for step in meta["steps"]]
                    meta["steps"] = _rebuild_step_hierarchy(meta["steps"])

                # Format Visualization (Repair types if stringified by sanitization)
                if "visualization" in meta and meta["visualization"]:
                    meta["visualization"] = _format_visualization(meta["visualization"])

                m.update(meta)

            formatted_messages.append(m)

        return {"messages": formatted_messages}
    except Exception as e:
        logger.error(f"Failed to retrieve history for session {session_id}: {e}", exc_info=True)
        if isinstance(e, (TechnicalError, FunctionalError)):
            raise
        raise TechnicalError(f"Failed to retrieve history: {e}")

get_optional_user(request, db) async

Resolves user from Bearer token if present, otherwise returns None.

Does not raise 401 for invalid tokens, just returns None (for public access if allowed).

Parameters:

• request (Request) –

  The FastAPI request object.

• db (Annotated[AsyncSession, Depends(get_db)]) –

  The database session.

Returns:

• Optional[User] –

  The User object if found and valid, otherwise None.

Source code in app/api/v1/endpoints/chat.py

async def get_optional_user(request: Request, db: Annotated[AsyncSession, Depends(get_db)]) -> Optional[User]:
    """
    Resolves user from Bearer token if present, otherwise returns None.

    Does not raise 401 for invalid tokens, just returns None (for public access if allowed).

    Args:
        request: The FastAPI request object.
        db: The database session.

    Returns:
        The User object if found and valid, otherwise None.
    """
    auth_header = request.headers.get("Authorization")
    if auth_header and auth_header.startswith("Bearer "):
        parts = auth_header.split(" ")
        if len(parts) != 2:
            return None
        token = parts[1]
        try:
            return await get_current_user(token=token, db=db)
        except Exception:
            return None
    return None

reset_chat_session(session_id, chat_service) async

Resets the conversation history for a specific session ID.

Parameters:

• session_id (str) –

  The session ID to reset.

• chat_service (Annotated[ChatService, Depends(get_chat_service)]) –

  The chat service instance.

Returns:

• Dict[str, str] –

  A success message.

Raises:

• TechnicalError –

  If resetting the conversation fails.

Source code in app/api/v1/endpoints/chat.py

@router.delete("/{session_id}")
async def reset_chat_session(
    session_id: str, chat_service: Annotated[ChatService, Depends(get_chat_service)]
) -> Dict[str, str]:
    """
    Resets the conversation history for a specific session ID.

    Args:
        session_id: The session ID to reset.
        chat_service: The chat service instance.

    Returns:
        A success message.

    Raises:
        TechnicalError: If resetting the conversation fails.
    """
    try:
        await chat_service.reset_conversation(session_id)
        return {"message": "Conversation history reset successfully"}
    except Exception as e:
        logger.error(f"Failed to reset conversation {session_id}: {e}", exc_info=True)
        if isinstance(e, (TechnicalError, FunctionalError)):
            raise
        raise TechnicalError(f"Failed to reset conversation: {e}")