gration page. Some of the other examples used are:Question URL Ending How do I get going with the Label Studio integration? What are the first steps? stacks-and-components/component-guide/annotators/label-studio How can I write my own custom materializer? user-guide/advanced-guide/data-management/handle-custom-data-types How do I generate embeddings as part of a RAG pipeline when using ZenML? user-guide/llmops-guide/rag-with-zenml/embeddings-generation How do I use failure hooks in my ZenML pipeline? user-guide/advanced-guide/pipelining-features/use-failure-success-hooks Can I deploy ZenML self-hosted with Helm? How do I do it? deploying-zenml/zenml-self-hosted/deploy-with-helm

For the retrieval pipeline, all we have to do is encode the query as a vector and then query the PostgreSQL database for the most similar vectors. We then check whether the URL for the document we thought must show up is actually present in the top n results.

def query_similar_docs(question: str, url_ending: str) -> tuple:

embedded_question = get_embeddings(question)

db_conn = get_db_conn()

top_similar_docs_urls = get_topn_similar_docs(

embedded_question, db_conn, n=5, only_urls=True

urls = [url[0] for url in top_similar_docs_urls]  # Unpacking URLs from tuples

return (question, url_ending, urls)

def test_retrieved_docs_retrieve_best_url(question_doc_pairs: list) -> float:

total_tests = len(question_doc_pairs)

failures = 0

for pair in question_doc_pairs:

question, url_ending, urls = query_similar_docs(

pair["question"], pair["url_ending"]

if all(url_ending not in url for url in urls):

logging.error(

f"Failed for question: {question}. Expected URL ending: {url_ending}. Got: {urls}"

failures += 1

logging.info(f"Total tests: {total_tests}. Failures: {failures}")

failure_rate = (failures / total_tests) * 100

return round(failure_rate, 2)

We include some logging so that when running the pipeline locally we can get some immediate feedback logged to the console.