Lesson 5 — knowledge retrieval you build¶
What you will build¶
The part of retrieval the framework does not ship, because no amount of framework can decide it for
you: where documents are cut, what a passage is, where vectors live, and what a score means. The
package gives embed and nothing else — no loader, no chunker, no index, no vector store.
This lesson is deliberately the same material as the AI path's lessons 9 and 10, applied to the desk's own articles. If you have read those, the only new thing here is that the passages live in the same database as the tickets, in a table this application owns.
What you need¶
Nothing new.
examples/support_desk/knowledge.py
"""Retrieval the application owns: passages, vectors, and a similarity query.
The package gives an application ``embed`` and nothing else. There is no document loader, no
chunker, no index, and no vector store, so this module is what "you build it" means in practice: the
rows, the cut, the arithmetic, and the honesty about what the arithmetic proves.
"""
from __future__ import annotations
import json
import math
from typing import Annotated
from sqlalchemy import insert, select
from support_desk.tables import passages
from xyberos import Inject
from xyberos.ai import AIFacade, EmbeddingRequest
from xyberos.database import DatabaseProvider
#: What the desk knows, a passage at a time. Cutting on blank lines keeps a paragraph whole, which
#: is the simplest rule that does not split the sentence that made a passage true.
ARTICLES: dict[str, str] = {
"refunds.md": (
"Refunds are issued to the original payment method.\n\n"
"A refund takes three business days to appear on a statement.\n\n"
"Orders older than ninety days are handled by the billing team."
),
"badges.md": (
"A badge that stops working is usually a battery, not an account problem.\n\n"
"Replacement badges are shipped within two business days of a request."
),
"accounts.md": (
"Seats can be added at any time during the term of an enterprise plan.\n\n"
"An account is closed when its last administrator asks for it to be closed."
),
}
def cosine(left: tuple[float, ...], right: tuple[float, ...]) -> float:
"""Return the cosine similarity of two vectors, or 0.0 when either has no length."""
dot = sum(a * b for a, b in zip(left, right, strict=True))
left_length = math.sqrt(sum(a * a for a in left))
right_length = math.sqrt(sum(b * b for b in right))
if not left_length or not right_length:
return 0.0
return dot / (left_length * right_length)
def split(text: str) -> list[str]:
"""Split a document into passages, one paragraph each."""
return [" ".join(block.split()) for block in text.split("\n\n") if block.strip()]
class KnowledgeBase:
"""The articles the desk may quote, stored with their vectors."""
def __init__(
self,
database: Annotated[DatabaseProvider, Inject()],
assistant: Annotated[AIFacade, Inject()],
) -> None:
self._database = database
self._assistant = assistant
async def embed(self, texts: tuple[str, ...]) -> tuple[tuple[float, ...], ...]:
"""Embed every text in one request, which is the cheapest way to use a provider."""
response = await self._assistant.embed(EmbeddingRequest(inputs=texts))
return response.embeddings
async def index(self, articles: dict[str, str] | None = None) -> int:
"""Embed and store the articles, and return how many passages that made."""
documents = ARTICLES if articles is None else articles
stored = 0
for source, text in documents.items():
chunks = split(text)
vectors = await self.embed(tuple(chunks))
async with self._database.transaction() as connection:
for ordinal, (chunk, vector) in enumerate(zip(chunks, vectors, strict=True)):
await connection.execute(
insert(passages).values(
source=source,
ordinal=ordinal,
text=chunk,
vector=json.dumps(list(vector)),
)
)
stored += 1
return stored
async def search(self, question: str, *, limit: int = 2) -> list[dict[str, object]]:
"""Return the passages closest to a question, best first.
The similarity is arithmetic over what the provider returned, so what it proves depends on
the provider: with a real embedding model the refund passage wins for a refund question, and
with the deterministic one every vector is a hash of its text and the ranking is procedural.
That difference is the reason lesson 16 of the AI path exists.
"""
(question_vector,) = await self.embed((question,))
async with self._database.transaction() as connection:
rows = (
(await connection.execute(select(passages).order_by(passages.c.id)))
.mappings()
.all()
)
ranked = sorted(
(
{
"score": cosine(question_vector, tuple(json.loads(str(row["vector"])))),
"source": row["source"],
"ordinal": row["ordinal"],
"text": row["text"],
}
for row in rows
),
key=lambda hit: (-float(hit["score"]), str(hit["source"]), int(hit["ordinal"])),
)
return ranked[:limit]
examples/learn_hybrid/hybrid05_retrieval.py
"""Lesson 5 — retrieval you build: passages, vectors, and a similarity query.
Run it with:
python -m learn_hybrid.hybrid05_retrieval
"""
from __future__ import annotations
import asyncio
import json
import tempfile
from pathlib import Path
from support_desk.app import create_app
from support_desk.knowledge import KnowledgeBase
from support_desk.migrate import migrate
from xyberos.database import SqliteSettings
SECRET = "a-secret-the-deployment-keeps"
QUESTION = "How long does a refund take?"
async def main() -> None:
"""Index the desk's articles, then ask one question and read the ranking."""
with tempfile.TemporaryDirectory() as directory:
settings = SqliteSettings(path=Path(directory) / "support.db")
migrate(settings)
app = create_app(settings=settings, secret=SECRET)
async with app.lifespan():
knowledge = await app.resolve(KnowledgeBase)
print("passages indexed:", await knowledge.index())
print("question:", QUESTION)
for hit in await knowledge.search(QUESTION, limit=3):
print(f" {hit['score']:.4f} {hit['source']}#{hit['ordinal']} {hit['text']}")
(vector,) = await knowledge.embed((QUESTION,))
print("vector width:", len(vector))
print("one vector:", len(json.dumps(list(vector))), "characters of JSON")
print("every score above is the same because the deterministic provider hashes text")
print("instead of representing meaning, so the ranking is procedural. A real embedding")
print(
"model is what makes the refund passage win - and lesson 16 of the AI path is how"
)
print("you measure that instead of assuming it.")
if __name__ == "__main__":
asyncio.run(main())
How to run it¶
What it prints¶
passages indexed: 7
question: How long does a refund take?
1.0000 accounts.md#0 Seats can be added at any time during the term of an enterprise plan.
1.0000 refunds.md#1 A refund takes three business days to appear on a statement.
1.0000 accounts.md#1 An account is closed when its last administrator asks for it to be closed.
vector width: 4
one vector: 24 characters of JSON
every score above is the same because the deterministic provider hashes text
instead of representing meaning, so the ranking is procedural. A real embedding
model is what makes the refund passage win - and lesson 16 of the AI path is how
you measure that instead of assuming it.
Every score is 1.0000, and the refund passage is not first. That is the truth about the deterministic
provider, not a bug: it hashes text into a vector, so every vector points the same way and the ordering
you see is the tie-break in search. A retrieval test that passes against this provider proves the
plumbing and nothing about relevance.
That is why the transcript says so out loud, and why the AI path's lesson 16 exists: measure retrieval against labelled questions, with the model you will actually run, and treat the score as evidence only then.
The one thing this lesson does prove is the arithmetic and the cut: seven passages from three documents, four-number vectors, cosine similarity over them, and a ranking function whose tie-breaks are deterministic so a test can read it.
Common mistakes¶
- Treating a similarity score as relevance. It is a distance in whatever space the model learned. Order by it, then measure whether the order is any good.
- One passage per document. A whole article dilutes what it matches; a sentence loses the context that made it true. Cut on structure (paragraphs, headings) and keep the source and ordinal.
- A vector store before you have an index problem. A text column and a Python sort is honest for hundreds of passages. Reach for the store when the arithmetic says so.
- Forgetting
limit. Retrieval that returns everything is a context window waiting to overflow, and a bill to match. - Indexing inside a request. Indexing is a deployment step with its own cadence; the route here exists so the desk has one, not so a request can trigger one.
Next¶
Lesson 6 drafts a reply that quotes a passage — and sends nothing.