Rendered notebook

Generated from 04_rag_compliance.ipynb. Run it yourself with make docs-up — see Run locally.

04 — RAG over compliance regulations

We build a tiny RAG pipeline over a few snippets of made-up bank compliance text. This stays offline (no API keys) by using the HashingEmbedder and InMemoryVectorStore from apogee-ai-rag.

In production, globaltrust-bank swaps the in-memory store for QdrantVectorStore (already provisioned via apogee add:qdrant).

04_rag_compliance.ipynb · cell 1

import asyncio

from apogee_ai_rag import (
    Document,
    EchoGenerator,
    HashingEmbedder,
    IngestionJob,
    InMemoryVectorStore,
    PipelineSpec,
    RagFactory,
    RagQuery,
    RagType,
    RecursiveTextChunker,
)

04_rag_compliance.ipynb · cell 2

REGULATIONS = [
    Document(id='aml-1',  text='Customer due diligence requires identity verification before opening any account.'),
    Document(id='aml-2',  text='Transactions over USD 10,000 must be reviewed by a compliance officer within 24 hours.'),
    Document(id='kyc-1',  text='KYC documents must be re-validated annually for high-risk customers.'),
    Document(id='fx-1',   text='Cross-border transfers require specifying purpose code and recipient address.'),
    Document(id='priv-1', text='Customer data is retained for 5 years after account closure for regulatory purposes.'),
]

# Every component here is offline: HashingEmbedder needs no model download and
# EchoGenerator echoes the retrieved context instead of calling an LLM.
spec = PipelineSpec(
    chunker=RecursiveTextChunker(chunk_size=200, chunk_overlap=20),
    embedder=HashingEmbedder(dims=256),
    vector_store=InMemoryVectorStore(),
    generator=EchoGenerator(),
)
pipeline = RagFactory.build(RagType.NAIVE, spec)

await pipeline.ingest(IngestionJob(documents=REGULATIONS))
print(f'Ingested {len(REGULATIONS)} regulations.')

04_rag_compliance.ipynb · cell 3

questions = [
    'What is required when opening a new account?',
    'How long do we keep customer data after closure?',
    'Quem revisa transações grandes?',
]

for question in questions:
    response = await pipeline.run(RagQuery(text=question, top_k=2))
    top = response.sources[0].chunk.text if response.sources else '—'
    print(f'\nQ: {question}\nTop source: {top}')

In production

• Replace HashingEmbedder with OpenAIEmbedder or BgeEmbedder for real semantic similarity.
• Replace InMemoryVectorStore with QdrantVectorStore (the globaltrust-bank compose already has Qdrant).
• Wrap rag inside an apogee-ai agent (compliance officer) that decides when to retrieve and how to phrase the answer.
• Per-language indexes: ingest the same regulations translated into EN/PT/ES/ZH and route queries by customer.preferred_language.