Super RAG Basics

Name: SHIPIT Agent
Author: SHIPIT

This notebook covers the standalone RAG subsystem at shipit_agent.rag — no agents yet, just indexing and retrieval. Topics:

Open in Colab View on GitHub 12 code cells · 11 with saved output

Its imports and arguments are checked against shipit-agent 3.0. The outputs were saved when it was written, for an earlier release.

In [5]

python

from pathlib import Path
import sys

ROOT = (
    Path.cwd().resolve().parent
    if Path.cwd().name == "notebooks"
    else Path.cwd().resolve()
)
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from shipit_agent.rag import (
    RAG,
    DocumentChunker,
    HashingEmbedder,
    HybridSearchPipeline,
    IndexFilters,
    InMemoryBM25Store,
    InMemoryVectorStore,
    LLMReranker,
    SearchQuery,
)

1. Build a RAG in one line

RAG.default(embedder=…) gives you an in-memory hybrid index (vector + BM25) with sensible defaults.

In [6]

python

rag = RAG.default(embedder=HashingEmbedder(dimension=512))
rag

Out

<shipit_agent.rag.rag.RAG at 0x10e305b50>

2. Index text

index_text accepts free-form strings. Pass a stable document_id= if you want to reindex it later. Pass source= for a friendly label and metadata= for extra fields used by filters.

In [7]

python

rag.index_text(
    "Shipit supports Python 3.10 and newer. It ships with a CLI and a Python API.",
    source="readme.md",
    metadata={"tags": ["install", "python"]},
)
rag.index_text(
    "Shipit agents stream events in real time. Use agent.stream() to iterate events.",
    source="streaming.md",
    metadata={"tags": ["runtime", "events"]},
)
rag.index_text(
    "Shipit connects to Jira, Linear, Slack, Gmail, Google Drive, Notion, Confluence.",
    source="integrations.md",
    metadata={"tags": ["integrations"]},
)
rag.index_text(
    "Webhooks deliver structured payloads. The webhook_payload tool exposes the body.",
    source="webhooks.md",
    metadata={"tags": ["webhooks"]},
)

print(f"Indexed {rag.count()} chunks across {len(rag.list_sources())} sources")
print("Sources:", rag.list_sources())

Out

Indexed 4 chunks across 4 sources
Sources: ['integrations.md', 'readme.md', 'streaming.md', 'webhooks.md']

3. Index a file

index_file runs the same chunker → embedder → store pipeline on the contents of any file. The built-in TextExtractor supports .txt, .md, .html, .csv, .json with zero dependencies, plus .pdf and .docx via lazy pip install pypdf python-docx.

In [8]

python

import tempfile

tmp = Path(tempfile.mkdtemp())
(tmp / "api.md").write_text(
    "# Shipit API\n\n"
    "Agents can be run with .run(prompt) or streamed with .stream(prompt).\n\n"
    "Tool results are JSON with chunk ids, sources, and scores.\n"
)
indexed = rag.index_file(str(tmp / "api.md"))
print(f"indexed {len(indexed)} chunks from {tmp / 'api.md'}")

Out

indexed 1 chunks from /var/folders/bd/pq4lv0q52pv59m8pthkn60sc0000gn/T/tmpvku86xzt/api.md

4. Hybrid search

rag.search() runs vector and BM25 search in parallel, fuses the results with Reciprocal Rank Fusion (RRF), and returns a RAGContext with timings + per-component scores.

In [9]

python

ctx = rag.search("how do I stream events from an agent?", top_k=3)
print(f"total_found={ctx.total_found} timing_ms={ctx.timing_ms}\n")
for r in ctx.results:
    print(f"score={r.score:.3f}  vector={r.vector_score}  keyword={r.keyword_score}")
    print(f"  chunk_id={r.chunk.id}  source={r.chunk.source}")
    print(f"  text={r.chunk.text}\n")

Out

total_found=5 timing_ms={'embed_ms': 0.058959005400538445, 'vector_ms': 0.205666059628129, 'keyword_ms': 0.048832967877388, 'total_ms': 8.75645806081593}

score=0.017  vector=0.408248290463863  keyword=5.020658621720775
  chunk_id=6bf2b81deb2d4a338cc0b1b424f30ab9::0  source=streaming.md
  text=Shipit agents stream events in real time. Use agent.stream() to iterate events.

score=0.016  vector=0.049507377148833714  keyword=0.6959956911522742
  chunk_id=_var_folders_bd_pq4lv0q52pv59m8pthkn60sc0000gn_T_tmpvku86xzt_api.md::0  source=/var/folders/bd/pq4lv0q52pv59m8pthkn60sc0000gn/T/tmpvku86xzt/api.md
  text=# Shipit API Agents can be run with .run(prompt) or streamed with .stream(prompt). Tool results are JSON with chunk ids, sources, and scores.

score=0.008  vector=0.0625  keyword=None
  chunk_id=ddfd53a4b63341b9a05488981d714a92::0  source=readme.md
  text=Shipit supports Python 3.10 and newer. It ships with a CLI and a Python API.

4.1 Prompt-ready output with citation markers

RAGContext.to_prompt_context(max_chars=…) formats the hits as a block you can drop into any LLM prompt. Every chunk gets a stable [N] marker.

In [10]

python

print(ctx.to_prompt_context(max_chars=600))

Out

[1] source=streaming.md | chunk_id=6bf2b81deb2d4a338cc0b1b424f30ab9::0
    Shipit agents stream events in real time. Use agent.stream() to iterate events.
[2] source=/var/folders/bd/pq4lv0q52pv59m8pthkn60sc0000gn/T/tmpvku86xzt/api.md | chunk_id=_var_folders_bd_pq4lv0q52pv59m8pthkn60sc0000gn_T_tmpvku86xzt_api.md::0
    # Shipit API Agents can be run with .run(prompt) or streamed with .stream(prompt). Tool results are JSON with chunk ids, sources, and scores.
[3] source=readme.md | chunk_id=ddfd53a4b63341b9a05488981d714a92::0
    Shipit supports Python 3.10 and newer. It ships with a CLI an…

4.2 Pure-vector vs pure-keyword vs hybrid

hybrid_alpha weights the two branches: 1.0 = pure vector, 0.0 = pure BM25, 0.5 = balanced (default).

In [11]

python

for alpha in (1.0, 0.5, 0.0):
    ctx = rag.search("webhook payload", top_k=2, hybrid_alpha=alpha)
    label = {1.0: "vector", 0.0: "keyword", 0.5: "hybrid"}[alpha]
    top = ctx.results[0]
    print(f"{label:7s}  → {top.chunk.source:18s} score={top.score:.3f}")

Out

vector   → readme.md          score=0.017
hybrid   → readme.md          score=0.008
keyword  → readme.md          score=0.000

5. Filters

IndexFilters lets you scope a search to specific sources, document ids, metadata fields, or time windows.

In [12]

python

ctx = rag.search("python", top_k=3, filters=IndexFilters(sources=["readme.md"]))
for r in ctx.results:
    print(r.chunk.source, "-", r.chunk.text[:80])

Out

readme.md - Shipit supports Python 3.10 and newer. It ships with a CLI and a Python API.

6. Fetch a specific chunk + context expansion

fetch_chunk(chunk_id, chunks_above=, chunks_below=) returns the chunk by id and (optionally) its neighbours from the same document. Useful when the answer spans multiple paragraphs.

In [13]

python

first_id = next(iter(rag.vector_store.all_chunks())).id
print(f"first chunk id: {first_id}")
chunk = rag.fetch_chunk(first_id)
print(f"text: {chunk.text}\n")

# Search with context expansion baked in:
ctx = rag.search("python version", top_k=1, chunks_above=1, chunks_below=1)
r = ctx.results[0]
print("main:", r.chunk.text)
print("above:", [c.text for c in r.expanded_above])
print("below:", [c.text for c in r.expanded_below])

Out

first chunk id: ddfd53a4b63341b9a05488981d714a92::0
text: Shipit supports Python 3.10 and newer. It ships with a CLI and a Python API.

main: Shipit supports Python 3.10 and newer. It ships with a CLI and a Python API.
above: []
below: []

7. LLM-as-judge reranker

enable_reranking=True runs the top candidates through LLMReranker for higher precision. The default is off because it adds latency. Below we use a fake LLM so the cell stays offline.

In [14]

python

from examples.run_multi_tool_agent import build_llm_from_env

llm = build_llm_from_env("bedrock")


rag_rerank = RAG.default(
    embedder=HashingEmbedder(dimension=256),
    reranker=LLMReranker(llm=llm),
)
for text, src in [
    ("python is great for scripting", "a"),
    ("rust is great for systems", "b"),
    ("go is great for services", "c"),
]:
    rag_rerank.index_text(text, source=src)

ctx = rag_rerank.search("programming language", top_k=3, enable_reranking=True)
for r in ctx.results:
    print(f"rerank={r.rerank_score}  source={r.chunk.source:3s}  text={r.chunk.text}")

Out

rerank=0.7  source=a    text=python is great for scripting
rerank=0.7  source=b    text=rust is great for systems
rerank=0.7  source=c    text=go is great for services

8. Pluggable backends

The RAG.default(...) shortcut uses an InMemoryVectorStore + InMemoryBM25Store. For production swap in any object that implements the VectorStore / KeywordStore protocols. Below we wire one explicitly.

In [15]

python

custom = RAG(
    vector_store=InMemoryVectorStore(),
    keyword_store=InMemoryBM25Store(k1=1.2, b=0.75),
    embedder=HashingEmbedder(dimension=384),
    chunker=DocumentChunker(target_tokens=256, overlap_tokens=32),
)
custom.index_text("hello custom rag", source="custom.md")
print(custom.search("custom", top_k=1).to_prompt_context())

Out

[1] source=custom.md | chunk_id=73ae193d805d48c0b32540a128612b5c::0
    hello custom rag

9. Inspect the search pipeline directly

If you need to fine-tune the pipeline (recency bias, parallel branches, custom rerankers), construct it explicitly:

In [16]

python

pipeline = HybridSearchPipeline(
    vector_store=rag.vector_store,
    embedder=rag.embedder,
    keyword_store=rag.keyword_store,
)
ctx = pipeline.search(
    SearchQuery(
        query="python version",
        top_k=2,
        enable_recency_bias=True,
        recency_half_life_days=14.0,
    )
)
for r in ctx.results:
    print(r.chunk.source, r.chunk.text[:80], "score=", r.score)

Out

readme.md Shipit supports Python 3.10 and newer. It ships with a CLI and a Python API. score= 0.016666384732543362
streaming.md Shipit agents stream events in real time. Use agent.stream() to iterate events. score= 0.008196582662355864

Next: see 23_rag_with_agent.ipynb to plug this RAG into a regular Agent, and 24_rag_with_deep_agents.ipynb for GoalAgent / ReflectiveAgent / Supervisor integrations.

Super RAG Basics

This notebook covers the standalone RAG subsystem at shipit_agent.rag — no agents yet, just indexing and retrieval. Topics:

Its imports and arguments are checked against shipit-agent 3.0. The outputs were saved when it was written, for an earlier release.

In [5]

python

from pathlib import Path
import sys

ROOT = (
    Path.cwd().resolve().parent
    if Path.cwd().name == "notebooks"
    else Path.cwd().resolve()
)
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from shipit_agent.rag import (
    RAG,
    DocumentChunker,
    HashingEmbedder,
    HybridSearchPipeline,
    IndexFilters,
    InMemoryBM25Store,
    InMemoryVectorStore,
    LLMReranker,
    SearchQuery,
)

1. Build a RAG in one line

RAG.default(embedder=…) gives you an in-memory hybrid index (vector + BM25) with sensible defaults.

In [6]

python

rag = RAG.default(embedder=HashingEmbedder(dimension=512))
rag

Out

<shipit_agent.rag.rag.RAG at 0x10e305b50>

2. Index text

index_text accepts free-form strings. Pass a stable document_id= if you want to reindex it later. Pass source= for a friendly label and metadata= for extra fields used by filters.

In [7]

python

rag.index_text(
    "Shipit supports Python 3.10 and newer. It ships with a CLI and a Python API.",
    source="readme.md",
    metadata={"tags": ["install", "python"]},
)
rag.index_text(
    "Shipit agents stream events in real time. Use agent.stream() to iterate events.",
    source="streaming.md",
    metadata={"tags": ["runtime", "events"]},
)
rag.index_text(
    "Shipit connects to Jira, Linear, Slack, Gmail, Google Drive, Notion, Confluence.",
    source="integrations.md",
    metadata={"tags": ["integrations"]},
)
rag.index_text(
    "Webhooks deliver structured payloads. The webhook_payload tool exposes the body.",
    source="webhooks.md",
    metadata={"tags": ["webhooks"]},
)

print(f"Indexed {rag.count()} chunks across {len(rag.list_sources())} sources")
print("Sources:", rag.list_sources())

Out

Indexed 4 chunks across 4 sources
Sources: ['integrations.md', 'readme.md', 'streaming.md', 'webhooks.md']

3. Index a file

In [8]

python

import tempfile

tmp = Path(tempfile.mkdtemp())
(tmp / "api.md").write_text(
    "# Shipit API\n\n"
    "Agents can be run with .run(prompt) or streamed with .stream(prompt).\n\n"
    "Tool results are JSON with chunk ids, sources, and scores.\n"
)
indexed = rag.index_file(str(tmp / "api.md"))
print(f"indexed {len(indexed)} chunks from {tmp / 'api.md'}")

Out

indexed 1 chunks from /var/folders/bd/pq4lv0q52pv59m8pthkn60sc0000gn/T/tmpvku86xzt/api.md

4. Hybrid search

rag.search() runs vector and BM25 search in parallel, fuses the results with Reciprocal Rank Fusion (RRF), and returns a RAGContext with timings + per-component scores.

In [9]

python

ctx = rag.search("how do I stream events from an agent?", top_k=3)
print(f"total_found={ctx.total_found} timing_ms={ctx.timing_ms}\n")
for r in ctx.results:
    print(f"score={r.score:.3f}  vector={r.vector_score}  keyword={r.keyword_score}")
    print(f"  chunk_id={r.chunk.id}  source={r.chunk.source}")
    print(f"  text={r.chunk.text}\n")

Out

total_found=5 timing_ms={'embed_ms': 0.058959005400538445, 'vector_ms': 0.205666059628129, 'keyword_ms': 0.048832967877388, 'total_ms': 8.75645806081593}

score=0.017  vector=0.408248290463863  keyword=5.020658621720775
  chunk_id=6bf2b81deb2d4a338cc0b1b424f30ab9::0  source=streaming.md
  text=Shipit agents stream events in real time. Use agent.stream() to iterate events.

score=0.016  vector=0.049507377148833714  keyword=0.6959956911522742
  chunk_id=_var_folders_bd_pq4lv0q52pv59m8pthkn60sc0000gn_T_tmpvku86xzt_api.md::0  source=/var/folders/bd/pq4lv0q52pv59m8pthkn60sc0000gn/T/tmpvku86xzt/api.md
  text=# Shipit API Agents can be run with .run(prompt) or streamed with .stream(prompt). Tool results are JSON with chunk ids, sources, and scores.

score=0.008  vector=0.0625  keyword=None
  chunk_id=ddfd53a4b63341b9a05488981d714a92::0  source=readme.md
  text=Shipit supports Python 3.10 and newer. It ships with a CLI and a Python API.

4.1 Prompt-ready output with citation markers

RAGContext.to_prompt_context(max_chars=…) formats the hits as a block you can drop into any LLM prompt. Every chunk gets a stable [N] marker.

In [10]

python

print(ctx.to_prompt_context(max_chars=600))

Out

[1] source=streaming.md | chunk_id=6bf2b81deb2d4a338cc0b1b424f30ab9::0
    Shipit agents stream events in real time. Use agent.stream() to iterate events.
[2] source=/var/folders/bd/pq4lv0q52pv59m8pthkn60sc0000gn/T/tmpvku86xzt/api.md | chunk_id=_var_folders_bd_pq4lv0q52pv59m8pthkn60sc0000gn_T_tmpvku86xzt_api.md::0
    # Shipit API Agents can be run with .run(prompt) or streamed with .stream(prompt). Tool results are JSON with chunk ids, sources, and scores.
[3] source=readme.md | chunk_id=ddfd53a4b63341b9a05488981d714a92::0
    Shipit supports Python 3.10 and newer. It ships with a CLI an…

4.2 Pure-vector vs pure-keyword vs hybrid

hybrid_alpha weights the two branches: 1.0 = pure vector, 0.0 = pure BM25, 0.5 = balanced (default).

In [11]

python

for alpha in (1.0, 0.5, 0.0):
    ctx = rag.search("webhook payload", top_k=2, hybrid_alpha=alpha)
    label = {1.0: "vector", 0.0: "keyword", 0.5: "hybrid"}[alpha]
    top = ctx.results[0]
    print(f"{label:7s}  → {top.chunk.source:18s} score={top.score:.3f}")

Out

vector   → readme.md          score=0.017
hybrid   → readme.md          score=0.008
keyword  → readme.md          score=0.000

5. Filters

IndexFilters lets you scope a search to specific sources, document ids, metadata fields, or time windows.

In [12]

python

ctx = rag.search("python", top_k=3, filters=IndexFilters(sources=["readme.md"]))
for r in ctx.results:
    print(r.chunk.source, "-", r.chunk.text[:80])

Out

readme.md - Shipit supports Python 3.10 and newer. It ships with a CLI and a Python API.

6. Fetch a specific chunk + context expansion

fetch_chunk(chunk_id, chunks_above=, chunks_below=) returns the chunk by id and (optionally) its neighbours from the same document. Useful when the answer spans multiple paragraphs.

In [13]

python

first_id = next(iter(rag.vector_store.all_chunks())).id
print(f"first chunk id: {first_id}")
chunk = rag.fetch_chunk(first_id)
print(f"text: {chunk.text}\n")

# Search with context expansion baked in:
ctx = rag.search("python version", top_k=1, chunks_above=1, chunks_below=1)
r = ctx.results[0]
print("main:", r.chunk.text)
print("above:", [c.text for c in r.expanded_above])
print("below:", [c.text for c in r.expanded_below])

Out

first chunk id: ddfd53a4b63341b9a05488981d714a92::0
text: Shipit supports Python 3.10 and newer. It ships with a CLI and a Python API.

main: Shipit supports Python 3.10 and newer. It ships with a CLI and a Python API.
above: []
below: []

7. LLM-as-judge reranker

enable_reranking=True runs the top candidates through LLMReranker for higher precision. The default is off because it adds latency. Below we use a fake LLM so the cell stays offline.

In [14]

python

from examples.run_multi_tool_agent import build_llm_from_env

llm = build_llm_from_env("bedrock")


rag_rerank = RAG.default(
    embedder=HashingEmbedder(dimension=256),
    reranker=LLMReranker(llm=llm),
)
for text, src in [
    ("python is great for scripting", "a"),
    ("rust is great for systems", "b"),
    ("go is great for services", "c"),
]:
    rag_rerank.index_text(text, source=src)

ctx = rag_rerank.search("programming language", top_k=3, enable_reranking=True)
for r in ctx.results:
    print(f"rerank={r.rerank_score}  source={r.chunk.source:3s}  text={r.chunk.text}")

Out

rerank=0.7  source=a    text=python is great for scripting
rerank=0.7  source=b    text=rust is great for systems
rerank=0.7  source=c    text=go is great for services

8. Pluggable backends

In [15]

python

custom = RAG(
    vector_store=InMemoryVectorStore(),
    keyword_store=InMemoryBM25Store(k1=1.2, b=0.75),
    embedder=HashingEmbedder(dimension=384),
    chunker=DocumentChunker(target_tokens=256, overlap_tokens=32),
)
custom.index_text("hello custom rag", source="custom.md")
print(custom.search("custom", top_k=1).to_prompt_context())

Out

[1] source=custom.md | chunk_id=73ae193d805d48c0b32540a128612b5c::0
    hello custom rag

9. Inspect the search pipeline directly

If you need to fine-tune the pipeline (recency bias, parallel branches, custom rerankers), construct it explicitly:

In [16]

python

pipeline = HybridSearchPipeline(
    vector_store=rag.vector_store,
    embedder=rag.embedder,
    keyword_store=rag.keyword_store,
)
ctx = pipeline.search(
    SearchQuery(
        query="python version",
        top_k=2,
        enable_recency_bias=True,
        recency_half_life_days=14.0,
    )
)
for r in ctx.results:
    print(r.chunk.source, r.chunk.text[:80], "score=", r.score)

Out

readme.md Shipit supports Python 3.10 and newer. It ships with a CLI and a Python API. score= 0.016666384732543362
streaming.md Shipit agents stream events in real time. Use agent.stream() to iterate events. score= 0.008196582662355864

Next: see 23_rag_with_agent.ipynb to plug this RAG into a regular Agent, and 24_rag_with_deep_agents.ipynb for GoalAgent / ReflectiveAgent / Supervisor integrations.