Hello RAG!
Environment Setup
Start by creating your virtual environment using uv:
uv init
uv sync
git add .
git commit -m "Initial commit"
uv add openai langchain langchain-openai langchain-chroma langchain-community langchain-experimental cryptography pypdf chromadb
And make sure your terminal is using it:
which python
/usr/bin/python
source .venv/bin/activate
which python
~/Projects/llm_basics/02_basic_chains/.venv/bin/python
Basic OpenAI API Call
Let's test the environment by prompting a local Ollama model:
from openai import OpenAI
client = OpenAI(
base_url="http://localhost:11434/v1",
api_key="ollama",
)
response = client.responses.create(
model="qwen3.8:latest",
input="Tell me a joke that places Buffy Summers of the television show Buffy the Vampire Slayer in Quarks, the Ferengi bar on Deep Space Nine during the Cardasian occupation when Wesley Snipes dressed as Blade breaks through the door Katana in hand looking haunted and out of place.",
)
print(response.output_text)
.venv/bin/python -m llm_basics.api_calls
# A Quarks Incident (Classified: Bajoran Occupation, Year 2)
It's 1993. Buffy Summers is wiping down the bar at Quark's because she needed the tips to restock her stake inventory and no one in the BSGA would authorize a purchase order. A Cardassian occupation officer is in the corner trying, with great bureaucratic patience, to conscript her as a "tactical scented-broom logistics coordinator."
The door detonates inward.
Wesley Snipes in full Blade leather is standing in the smoke, Kitana cradled under one arm like a very confused bouquet of lilies, looking like a man who missed his scene, got lost in the studio parking lot, and is now three movies past his cue. His eyes are haunted. His soul is *out of place.* He does not know why there is a Ferengi in a tank-top.
He snarls, "Where is the vampire?"
Buffy points at Quark. Quark polishes a glass and goes, "That's *me.* I mean—*I* am the one with the little— the fangy— the..."
Blade looks at Kitana. Kitana looks at Buffy. Buffy looks at the Cardassian, who is already filling out Form 7-C: *"Unscheduled Cinematic Incursion, Week 3, Second One This Shift."*
Blade says, very quietly, to no one: "I'm in the wrong one. I think I'm in the wrong one."
He backpedals. The door he broke stays broken. Quark slaps a 40-credit tab on the counter for "psychological infrastructure damages." The Cardassian adds eleven for the scented-broom.
Buffy goes back to wiping the bar and mutters, "I used to slay the monster of the week. Now I just slay the *vibes.*"
And Kitana, still tucked under his arm as Blade shuffles out into the corridor, says nothing.
She says nothing.
She's from *Mortal Kombat.* She just *is.*
Retrieval Augmented Generation
Text Embedding
I want to use a PDF as my source of truth. To use it in a RAG pipeline it first need to be embedded using a specialized embedding model, e.g.:
ollama pull qwen3-embedding:0.6b
import os
from pathlib import Path
from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
asset_path = Path("src/rag_intro/assets/IMX415-AAQR_Flyer.pdf")
asset_path.parent.mkdir(parents=True, exist_ok=True)
# Load the text source from pdf file
loader = PyPDFLoader(asset_path)
text = loader.load()
splitter = RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=20)
chunks = splitter.split_documents(text)
# Create vector embeddings from pages
embed_model = OpenAIEmbeddings(
model="qwen3-embedding:0.6b",
base_url="http://localhost:11434/v1",
api_key="ollama",
check_embedding_ctx_length=False
)
embedded_chunks = embed_model.embed_documents([str(chunk) for chunk in chunks])
Though it seems that this part is actually already handled by the Vector Database I will have to use for the RAG pipeline. But just as as a PoC - this already works like a charm!
Vector Database
To get started I will try ChromaDB to store my embeddings. And I was right - the embedding creation is now fully automatic. I just need to point the Chroma function to my PDF asset chunks and to a db location and it database is automatically created containing the embeddings:
from pathlib import Path
from langchain_chroma import Chroma
from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
asset_path = Path("src/rag_intro/assets/IMX415-AAQR_Flyer.pdf")
asset_path.parent.mkdir(parents=True, exist_ok=True)
db_path = Path("src/rag_intro/db/")
db_path.parent.mkdir(parents=True, exist_ok=True)
# Load the text source from pdf file
loader = PyPDFLoader(asset_path)
text = loader.load()
splitter = RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=20)
chunks = splitter.split_documents(text)
# Create vector embeddings from pages
embed_model = OpenAIEmbeddings(
model="qwen3-embedding:0.6b",
base_url="http://localhost:11434/v1",
api_key="ollama",
check_embedding_ctx_length=False
)
# Create db and store embeddings
db = Chroma.from_documents(chunks, embed_model, persist_directory=str(db_path))
❯ python -m rag_intro.main
❯ ls -lah src/rag_intro/db/
Permissions Size User Date Modified Name
drwxr-xr-x - user 14 Oct 17:00 47e4a561-ab4e-4f8d-bfb0-ad097ce09eab
.rw-r--r-- 586k user 14 Oct 17:00 chroma.sqlite3
An now I am able to query my vector embedding and return the corresponding text!
from pathlib import Path
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
db_path = Path("src/rag_intro/db/")
db_path.parent.mkdir(parents=True, exist_ok=True)
# DB connection and retrieval
embed_model = OpenAIEmbeddings(
model="qwen3-embedding:0.6b",
base_url="http://localhost:11434/v1",__
api_key="ollama",
check_embedding_ctx_length=False
)
db_con = Chroma(persist_directory=str(db_path), embedding_function=embed_model)
print(db_con.similarity_search("What is STARVIS?", k=3))
print(db_con.similarity_search("How many effective pixels does the IMX415-AAQR CMOS have?", k=3))
python -m rag_intro.main
[Document(id='a8e86394-9afe-4dbe-b386-559852d50fd0', metadata={'sourcemodified': 'D:20190611050147', 'total_pages': 2, 'title': '', 'subject': '', 'creationdate': '2019-06-11T14:01:56+09:00', 'moddate': '2022-05-13T11:34:40+09:00', 'company': 'Sony', 'keywords': '', 'page_label': '1', 'source': 'src/rag_intro/assets/IMX415-AAQR_Flyer.pdf', 'producer': 'Adobe PDF Library 15.0', 'creator': 'Word 用 Acrobat PDFMaker 15', 'page': 0}, page_content='its affiliates. The STARVIS is back-illuminated pixel'), Document(id='edb9c695-74ac-40fb-9cc7-6decf5735eb3', metadata={'producer': 'Adobe PDF Library 15.0', 'keywords': '', 'moddate': '2022-05-13T11:34:40+09:00', 'creationdate': '2019-06-11T14:01:56+09:00', 'sourcemodified': 'D:20190611050147', 'page': 0, 'page_label': '1', 'company': 'Sony', 'source': 'src/rag_intro/assets/IMX415-AAQR_Flyer.pdf', 'creator': 'Word 用 Acrobat PDFMaker 15', 'subject': '', 'title': '', 'total_pages': 2}, page_content='* STARVIS is a registered trademark or trademark of Sony Group Corporation or its affiliates. The'), Document(id='6102eab5-80ed-438d-a837-80801c617d03', metadata={'page': 1, 'page_label': '2', 'company': 'Sony', 'title': '', 'source': 'src/rag_intro/assets/IMX415-AAQR_Flyer.pdf', 'creator': 'Word 用 Acrobat PDFMaker 15', 'producer': 'Adobe PDF Library 15.0', 'moddate': '2022-05-13T11:34:40+09:00', 'sourcemodified': 'D:20190611050147', 'subject': '', 'total_pages': 2, 'creationdate': '2019-06-11T14:01:56+09:00', 'keywords': ''}, page_content='Saturation signal Min. 3895 Digit 12 bit converted value \nBasic Drive Mode')]
[Document(id='f4b01653-0c25-49da-bc95-61725f3407d2', metadata={'company': 'Sony', 'total_pages': 2, 'source': 'src/rag_intro/assets/IMX415-AAQR_Flyer.pdf', 'title': '', 'creator': 'Word 用 Acrobat PDFMaker 15', 'subject': '', 'keywords': '', 'producer': 'Adobe PDF Library 15.0', 'creationdate': '2019-06-11T14:01:56+09:00', 'sourcemodified': 'D:20190611050147', 'page': 0, 'page_label': '1', 'moddate': '2022-05-13T11:34:40+09:00'}, page_content='The IMX415-AAQR is a diagonal 6.4 mm (Type 1/2.8) CMOS active pixel type solid-state image sensor'), Document(id='1fa08814-2b96-4f80-9161-1a9fc5375cec', metadata={'sourcemodified': 'D:20190611050147', 'creator': 'Word 用 Acrobat PDFMaker 15', 'source': 'src/rag_intro/assets/IMX415-AAQR_Flyer.pdf', 'moddate': '2022-05-13T11:34:40+09:00', 'company': 'Sony', 'page_label': '1', 'subject': '', 'page': 0, 'title': '', 'total_pages': 2, 'creationdate': '2019-06-11T14:01:56+09:00', 'keywords': '', 'producer': 'Adobe PDF Library 15.0'}, page_content='1 \nCopyr\night 2019 Sony Semiconductor Solutions Corporation \n \nVer.1.0 IMX415-AAQR'), Document(id='9b852b1e-5004-45b8-95f2-f9ea9300ff0f', metadata={'keywords': '', 'sourcemodified': 'D:20190611050147', 'page': 1, 'creationdate': '2019-06-11T14:01:56+09:00', 'title': '', 'total_pages': 2, 'producer': 'Adobe PDF Library 15.0', 'moddate': '2022-05-13T11:34:40+09:00', 'subject': '', 'source': 'src/rag_intro/assets/IMX415-AAQR_Flyer.pdf', 'company': 'Sony', 'page_label': '2', 'creator': 'Word 用 Acrobat PDFMaker 15'}, page_content='IMX415-AAQR \n2 \nDevice Structure \n◆ CMOS image sensor')]
And now I can feed the retrieved chunks directly into an LLM as user query:
# Text generation from retrieved text
user_query = "What is the diagonal of the SONY IMX415-AAQR STARVIS sensor?"
retrieved_chunks = db_con.similarity_search(user_query, k=3)
content = []
for i, chunk in enumerate(retrieved_chunks):
# # DEBUG
# print(chunk.page_content)
content.append(chunk.page_content)
client = ChatOpenAI(
model="qwen3.8:latest",
temperature=0.6,
top_p=0.95,
base_url="http://localhost:11434/v1",
api_key="ollama"
)
print(client.invoke(f"{user_query}. Use the following context to answer the question: {str(content)}").content)
python -m rag_intro.main
Based on the provided context, the diagonal of the SONY IMX415-AAQR STARVIS sensor is **6.43 mm** (Type 1/2.8). This is described as a CMOS Solid-state Image Sensor with Square Pixel for Color Cameras.
RAG Agents
I previously already build a LangChain ReAct agent based on a RAG loop from an SQL database. Let's do the same thing now with the ChromaDB connection - I will also add a mocked SQL query for alarm events so that I will have 2 tools to work with and can test the model selecting the correct one based on the user prompt:
from pathlib import Path
from langchain.agents import create_agent
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.tools import tool
client = ChatOpenAI(
model="qwen3.8:latest",
temperature=0.6,
top_p=0.95,
base_url="http://localhost:11434/v1",
api_key="ollama",
)
# --- 1. Vector DB (RAG over your Chroma embeddings) -------------------------
db_path = Path(__file__).resolve().parent.parent / "rag-intro" / "db"
db_path.parent.mkdir(parents=True, exist_ok=True)
embed_model = OpenAIEmbeddings(
model="qwen3-embedding:0.6b",
base_url="http://localhost:11434/v1",
api_key="ollama",
check_embedding_ctx_length=False
)
db_con = Chroma(persist_directory=str(db_path), embedding_function=embed_model)
# --- 2. Tools ----------------------------------------------------------------
@tool
def license_plate_list(plate: str) -> str:
'''Use this tool when you need to answer a question about when a
specified registration number / license plate was last detected.
Do NOT use it for questions about camera or sensor specifications.
Args:
plate (str): The license plate number of the car to look up.
Returns:
str: A string containing information of the corresponding alarm
event - containing the time stamp of the event and the alarm trigger.
'''
# Hard-coded info - simulating SQL retrieved data
records = [{
"plate": "K:FG:123",
"timestamp": "Wed Oct 14 08:17:05 PM HKT 2026",
"trigger": "Alarm triggered by Motion area 1"
},{
"plate": "K:SX:712",
"timestamp": "Wed Oct 14 12:01:45 PM HKT 2026",
"trigger": "Alarm triggered by Motion area 3"
}]
for record in records:
if record["plate"].lower() == plate.lower():
return (
f"(License plate {record['plate']} last seen at "
f"{record['timestamp']}. {record['trigger']}.)"
)
known = ", ".join(r["plate"] for r in records)
return (
f"(License plate {plate} was not found in recent alarm events. "
f"Plates on file: {known}.)"
)
@tool
def sensor_spec_search(query: str) -> str:
'''Use this tool when a question is asked about the technical
specifications of the SONY IMX415-AAQR STARVIS sensor, e.g. its size
or diagonal, resolution, readout, frame rate, sensitivity, or lens
mount. It searches the sensor datasheet in the Chroma vector database.
Do NOT use it for questions about license plates or alarm events.
Args:
query (str): The specification question or the property the user
is asking about.
Returns:
str: The relevant datasheet passages retrieved from the vector DB.
'''
chunks = db_con.similarity_search(query, k=3)
if not chunks:
return "(No relevant passages found in the sensor datasheet database.)"
passages = [
f"Passage {i + 1}:\n{chunk.page_content}"
for i, chunk in enumerate(chunks)
]
return "(Datasheet passages retrieved from the vector DB:\n\n" + "\n\n".join(passages) + ")"
toolkit = [
license_plate_list,
sensor_spec_search,
]
# --- 3. Agent loop: execute tool calls -----------------------------
agent = create_agent(client, toolkit)
def ask(question: str) -> str:
result = agent.invoke({"messages": [("user", question)]})
print("\nModel Response:")
print(result["messages"][-1].content)
print("=" * 72)
return result["messages"][-1].content
print("Q1 (routes to Chroma tool): What is the diagonal of the SONY IMX415-AAQR STARVIS sensor?\n")
ask("What is the diagonal of the SONY IMX415-AAQR STARVIS sensor?")
print("\nQ2 (routes to plate tool): When was the car with license plate K:SX:712 last seen?\n")
ask("When was the car with license plate K:SX:712 last seen?")
And both queries are routed to the correct tool and return the correct answer - awesome!
python -m rag-intro.main
Q1 (routes to Chroma tool): What is the diagonal of the SONY IMX415-AAQR STARVIS sensor?
Model Response:
According to the datasheet, the SONY IMX415-AAQR STARVIS sensor has a **diagonal of 6.43 mm** (Type 1/2.8). It is a CMOS solid-state image sensor with square pixels designed for color cameras.
========================================================================
Q2 (routes to plate tool): When was the car with license plate K:SX:712 last seen?
Model Response:
The car with license plate **K:SX:712** was last seen on **Wednesday, October 14, 2026 at 12:01:45 PM HKT**. The alarm was triggered by **Motion area 3**.
========================================================================
Improving the RAG Data Quality with Symantic Chunking
The data retrieved from ChromaDB was often cut off / incomplete because of the fixed chunking. The solution is to have the process check which part of the text belongs together and should be retrieved in an single chunk. LangChain just added a text splitter to it's experimental package that says it is doing just that. So let's rebuild the database using the semantic splitter:
from pathlib import Path
from langchain_chroma import Chroma
from langchain_community.document_loaders import PyPDFLoader
from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings
asset_path = Path(__file__).resolve().parent.parent / "rag_intro" / "assets"
asset_path.parent.mkdir(parents=True, exist_ok=True)
db_path = Path(__file__).resolve().parent.parent / "rag_intro" / "db"
db_path.parent.mkdir(parents=True, exist_ok=True)
# Load the text source from pdf file
loader = PyPDFLoader(str(asset_path) + "/IMX415-AAQR_Flyer.pdf")
docs = loader.load()
full_text = "\n".join([page.page_content for page in docs]) # the pages have to be merged into a single string
# Create vector embeddings from pages
embed_model = OpenAIEmbeddings(
model="qwen3-embedding:0.6b",
base_url="http://localhost:11434/v1",
api_key="ollama",
check_embedding_ctx_length=False
)
splitter = SemanticChunker(embed_model, breakpoint_threshold_type="percentile", breakpoint_threshold_amount=0.6)
semantic_chunks = splitter.create_documents([full_text])
# # DEBUG
# print(len(semantic_chunks)) # returns 19 -> much larger chunks now
# print(semantic_chunks[0])
# # -> The IMX415-AAQR is a diagonal 6.4 mm (Type 1/2.8) CMOS active pixel type solid-state image sensor with a square pixel array and 8.46 M effective pixels. This chip operates with analog 2.9 V, digital 1.1 V, and interface 1.8 V triple power supply, and has low power consumption.
# Create db with embeddings
db = Chroma.from_documents(semantic_chunks, embed_model, persist_directory=str(db_path))
So it is still working - but the result is pretty same-ish. Meaning the chunking before already contained all the essential information for the reply ~ though it was only by chance:
python -m rag_intro.main
Q1 (routes to Chroma tool): What is the diagonal of the SONY IMX415-AAQR STARVIS sensor?
Model Response:
The SONY IMX415-AAQR STARVIS sensor has a **diagonal of 6.43 mm (Type 1/2.8)**. It is a CMOS solid-state image sensor with a square pixel array and 8.46 M effective pixels.
========================================================================
Agent Memory
Having your agent recall the current conversation. The easiest way to save the information in a variable and simply pass it back to the agent for every call that requires the context:
agent = create_agent(client, toolkit)
agent_memory = []
def ask(question: str) -> str:
result = agent.invoke({"messages": [("user", question)]})
message = result["messages"][-1].content
agent_memory.append(message)
print("\nModel Response:")
print(message)
print("=" * 72)
return message
print("Q1 (routes to Chroma tool): What is the diagonal of the SONY IMX415-AAQR STARVIS sensor?\n")
ask("What is the diagonal of the SONY IMX415-AAQR STARVIS sensor?")
print("\nQ2 (routes to plate tool): When was the car with license plate K:SX:712 last seen?\n")
ask("When was the car with license plate K:SX:712 last seen?")
print("\nQ3 (agent memory): What was the license plate of the car that triggered an alarm on Wednesday, October 14, 2026, at 12:01:45 PM HKT?\n")
ask(f"What was the license plate of the car that triggered an alarm on Wednesday, October 14, 2026, at 12:01:45 PM HKT? You can refer to the previous conversation: {agent_memory}")
print("\nQ4 (agent memory): Can you summarize the conversation?\n")
ask(" ".join(str(memory) for memory in agent_memory))
python -m rag_intro.main
Q1 (routes to Chroma tool): What is the diagonal of the SONY IMX415-AAQR STARVIS sensor?
Model Response:
The diagonal of the SONY IMX415-AAQR STARVIS sensor is **6.43 mm** (Type 1/2.8). It is a CMOS solid-state image sensor with a square pixel array and 8.46 M effective pixels, designed for color cameras.
========================================================================
Q2 (routes to plate tool): When was the car with license plate K:SX:712 last seen?
Model Response:
The car with license plate **K:SX:712** was last seen on **Wednesday, October 14, 2026 at 12:01:45 PM HKT**. The alarm was triggered by **Motion area 3**.
========================================================================
Q3 (agent memory): What was the license plate of the car that triggered an alarm on Wednesday, October 14, 2026, at 12:01:45 PM HKT?
Model Response:
Based on the previous conversation, the license plate of the car that triggered an alarm on **Wednesday, October 14, 2026, at 12:01:45 PM HKT** was **K:SX:712**. The alarm was triggered by **Motion area 3**.
========================================================================
Q4 (agent memory): Can you summarize the conversation?
Model Response:
Thank you for the summary! Here's a recap of the key information:
**SONY IMX415-AAQR STARVIS Sensor:**
- **Diagonal:** 6.43 mm (Type 1/2.8)
- **Type:** CMOS solid-state image sensor
- **Pixel Array:** Square
- **Effective Pixels:** 8.46 M
- **Design:** Color cameras
**Alarm Event:**
- **License Plate:** K:SX:712
- **Last Detected:** Wednesday, October 14, 2026 at 12:01:45 PM HKT
- **Alarm Trigger:** Motion area 3
Is there anything else you'd like to look up — whether it's additional sensor specifications or details about another license plate / alarm event?
========================================================================