How-To Implement Agentic RAG with Llama 3.2 on Your Own Dataset Locally

 This video is a step-by-step tutorial on implementing Agentic RAG with Llama3.2 3B model on your own dataset locally.

Code:

conda create -n ag python=3.11 -y && conda activate ag

pip install torch

pip install git+https://github.com/huggingface/transformers

pip install git+https://github.com/huggingface/accelerate

pip install huggingface_hub

pip install sentencepiece

pip install bitsandbytes

pip install haystack-ai duckduckgo-api-haystack transformers sentence-transformers datasets

from datasets import load_dataset

from haystack import Document

from haystack.document_stores.in_memory import InMemoryDocumentStore

from haystack.components.embedders import SentenceTransformersDocumentEmbedder

document_store = InMemoryDocumentStore()

dataset = load_dataset("bilgeyucel/seven-wonders", split="train")

docs = [Document(content=doc["content"], meta=doc["meta"]) for doc in dataset]

doc_embedder = SentenceTransformersDocumentEmbedder(model="sentence-transformers/all-MiniLM-L6-v2")

doc_embedder.warm_up()

docs_with_embeddings = doc_embedder.run(docs)

document_store.write_documents(docs_with_embeddings["documents"])

import torch

from haystack.components.generators import HuggingFaceLocalGenerator

generator = HuggingFaceLocalGenerator(

    model="meta-llama/Llama-3.2-3B-Instruct",

    huggingface_pipeline_kwargs={"device_map":"auto",

                                 "torch_dtype":torch.bfloat16},

    generation_kwargs={"max_new_tokens": 256})

generator.warm_up()

prompt = """<|begin_of_text|><|start_header_id|>user<|end_header_id|>

  What is the capital of Australia?<|eot_id|>

  <|start_header_id|>assistant<|end_header_id|>"""

generator.run(prompt)

from haystack.components.embedders import SentenceTransformersTextEmbedder

from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever

text_embedder = SentenceTransformersTextEmbedder(model="sentence-transformers/all-MiniLM-L6-v2")

retriever = InMemoryEmbeddingRetriever(document_store, top_k=5)

from haystack.components.builders import PromptBuilder

prompt_template = """

<|begin_of_text|><|start_header_id|>user<|end_header_id|>

Answer the following query given the documents.

If the answer is not contained within the documents reply with 'no_answer'.

If the answer is contained within the documents, start the answer with "FROM THE KNOWLEDGE BASE: ".

Documents:

{% for document in documents %}

  {{document.content}}

{% endfor %}

Query: {{query}}<|eot_id|>

<|start_header_id|>assistant<|end_header_id|>

"""

prompt_builder = PromptBuilder(template=prompt_template)

from haystack.components.routers import ConditionalRouter

routes = [

    {

        "condition": "{{'no_answer' in replies[0]}}",

        "output": "{{query}}",

        "output_name": "go_to_websearch",

        "output_type": str,

    },

    {

        "condition": "{{'no_answer' not in replies[0]}}",

        "output": "{{replies[0]}}",

        "output_name": "answer",

        "output_type": str,

    },

]

router = ConditionalRouter(routes)

from duckduckgo_api_haystack import DuckduckgoApiWebSearch

websearch = DuckduckgoApiWebSearch(top_k=5)

prompt_template_after_websearch = """

<|begin_of_text|><|start_header_id|>user<|end_header_id|>

Answer the following query given the documents retrieved from the web.

Start the answer with "FROM THE WEB: ".

Documents:

{% for document in documents %}

  {{document.content}}

{% endfor %}

Query: {{query}}<|eot_id|>

<|start_header_id|>assistant<|end_header_id|>

"""

prompt_builder_after_websearch = PromptBuilder(template=prompt_template_after_websearch)

from haystack.components.joiners import BranchJoiner

prompt_joiner  = BranchJoiner(str)

from haystack import Pipeline

pipe = Pipeline()

pipe.add_component("text_embedder", text_embedder)

pipe.add_component("retriever", retriever)

pipe.add_component("prompt_builder", prompt_builder)

pipe.add_component("prompt_joiner", prompt_joiner)

pipe.add_component("llm", generator)

pipe.add_component("router", router)

pipe.add_component("websearch", websearch)

pipe.add_component("prompt_builder_after_websearch", prompt_builder_after_websearch)

pipe.connect("text_embedder", "retriever")

pipe.connect("retriever", "prompt_builder.documents")

pipe.connect("prompt_builder", "prompt_joiner")

pipe.connect("prompt_joiner", "llm")

pipe.connect("llm.replies", "router.replies")

pipe.connect("router.go_to_websearch", "websearch.query")

pipe.connect("router.go_to_websearch", "prompt_builder_after_websearch.query")

pipe.connect("websearch.documents", "prompt_builder_after_websearch.documents")

pipe.connect("prompt_builder_after_websearch", "prompt_joiner")

def get_answer(query):

  result = pipe.run({"text_embedder": {"text": query}, "prompt_builder": {"query": query}, "router": {"query": query}})

  print(result["router"]["answer"])

query = "Why did people build Great Pyramid of Giza?"

get_answer(query)