RAG + Deep Agents

Name: SHIPIT Agent
Author: SHIPIT

All five deep-agent patterns in shipit_agent.deep accept the same rag= parameter as Agent. They forward it to every inner Agent they build, so every step of a multi-step agentic workflow has access to the knowledge base.

Open in Colab View on GitHub 10 code cells · 9 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 [1]

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 import Agent
from shipit_agent.deep import (
    AdaptiveAgent,
    Goal,
    GoalAgent,
    PersistentAgent,
    ReflectiveAgent,
    Supervisor,
    Worker,
)
from shipit_agent.llms.base import LLMResponse
from shipit_agent.rag import RAG, HashingEmbedder
from shipit_agent.tools.base import ToolContext

Setup — shared knowledge base + mock LLM

In [2]

python

rag = RAG.default(embedder=HashingEmbedder(dimension=512))
rag.index_text(
    "Shipit supports Python 3.10 and newer. CLI and Python API.",
    document_id="readme",
    source="readme.md",
)
rag.index_text(
    "Shipit agents stream events in real time via agent.stream().",
    document_id="streaming",
    source="streaming.md",
)
rag.index_text(
    "Shipit deep agents include GoalAgent, ReflectiveAgent, Supervisor, AdaptiveAgent, PersistentAgent.",
    document_id="deep",
    source="deep_agents.md",
)
rag.index_text(
    "Shipit integrates with Jira, Linear, Slack, Gmail, Google Drive, Notion, Confluence.",
    document_id="integrations",
    source="integrations.md",
)
print(f"{rag.count()} chunks indexed")


class ChatLLM:
    def __init__(self):
        self.calls = 0

    def complete(self, messages, **kwargs):
        self.calls += 1
        return LLMResponse(
            content="Based on the docs, Shipit supports Python 3.10+ and streams events. [1][2]",
            usage={"prompt_tokens": 5, "completion_tokens": 5, "total_tokens": 10},
        )


llm = ChatLLM()

Out

4 chunks indexed

1. GoalAgent + RAG

GoalAgent decomposes a goal into sub-tasks and runs each through an inner Agent. Pass rag= and every sub-task can search the knowledge base.

In [3]

python

goal_agent = GoalAgent(
    llm=llm,
    goal=Goal(
        objective="Write a 2-sentence summary of Shipit runtime features",
        success_criteria=["mentions python version", "mentions streaming"],
    ),
    rag=rag,
)

inner = goal_agent._build_agent()
print("inner agent tool names:", sorted(t.name for t in inner.tools))
print("inner agent prompt has RAG section:", "rag_search" in inner.prompt)

Out

inner agent tool names: ['rag_fetch_chunk', 'rag_list_sources', 'rag_search']
inner agent prompt has RAG section: True

Driving the inner Agent for a citation demo

Each inner Agent run captures sources into the same RAG instance. We drive the rag_search tool manually so the notebook works without a real LLM.

In [4]

python

rag.begin_run()
search = next(t for t in inner.tools if t.name == "rag_search")
search.run(ToolContext(prompt=""), query="python version", top_k=1)
search.run(ToolContext(prompt=""), query="streaming", top_k=1)
sources = rag.end_run()
for s in sources:
    print(f"[{s.index}] {s.source} → {s.text[:80]}")

Out

[1] readme.md → Shipit supports Python 3.10 and newer. CLI and Python API.
[2] streaming.md → Shipit agents stream events in real time via agent.stream().

2. ReflectiveAgent + RAG

ReflectiveAgent generates an answer, critiques it, and revises if needed. The same rag= works.

In [5]

python

reflective = ReflectiveAgent(llm=llm, rag=rag)
inner = reflective._build_agent()
print("reflective inner tools:", sorted(t.name for t in inner.tools))

Out

reflective inner tools: ['rag_fetch_chunk', 'rag_list_sources', 'rag_search']

3. AdaptiveAgent + RAG

AdaptiveAgent can write new tools at runtime and still use RAG for grounding.

In [6]

python

adaptive = AdaptiveAgent(llm=llm, rag=rag)
print("adaptive agent_kwargs has rag:", "rag" in adaptive.agent_kwargs)
print("adaptive can_create_tools:", adaptive.can_create_tools)

Out

adaptive agent_kwargs has rag: True
adaptive can_create_tools: True

4. Supervisor + RAG (one shared knowledge base for the whole team)

Supervisor.with_builtins(..., rag=rag) wires the same RAG into every worker. A researcher and a writer can both retrieve from — and cite — the same chunks.

In [7]

python

supervisor = Supervisor.with_builtins(
    llm=llm,
    worker_configs=[
        {"name": "researcher", "prompt": "You research the codebase."},
        {"name": "writer", "prompt": "You write user-facing answers."},
    ],
    rag=rag,
)
for name, worker in supervisor.workers.items():
    tools = sorted(t.name for t in worker.agent.tools)
    rag_tools = [t for t in tools if t.startswith("rag_")]
    print(f"worker={name}: rag tools = {rag_tools}")

Out

worker=researcher: rag tools = ['rag_fetch_chunk', 'rag_list_sources', 'rag_search']
worker=writer: rag tools = ['rag_fetch_chunk', 'rag_list_sources', 'rag_search']

4.1 Building a supervisor manually with RAG-equipped Agents

If you want to construct workers by hand instead of using with_builtins, just create each worker's Agent with rag=rag directly:

In [8]

python

researcher_agent = Agent(llm=llm, prompt="Research mode.", rag=rag)
writer_agent = Agent(llm=llm, prompt="Writer mode.", rag=rag)

manual_supervisor = Supervisor(
    llm=llm,
    workers=[
        Worker(name="researcher", agent=researcher_agent),
        Worker(name="writer", agent=writer_agent),
    ],
    rag=rag,
)
print("supervisor.rag is rag:", manual_supervisor.rag is rag)
for name, w in manual_supervisor.workers.items():
    has_rag = any(t.name == "rag_search" for t in w.agent.tools)
    print(f"  worker={name}: rag_search wired = {has_rag}")

Out

supervisor.rag is rag: True
  worker=researcher: rag_search wired = True
  worker=writer: rag_search wired = True

5. PersistentAgent + RAG

PersistentAgent checkpoints long runs and forwards rag= to every Agent it builds, so RAG state survives across resumes (the chunks are in the store, not the checkpoint).

In [9]

python

persistent = PersistentAgent(llm=llm, rag=rag, checkpoint_interval=1, max_steps=2)
print("persistent agent_kwargs has rag:", "rag" in persistent.agent_kwargs)

Out

persistent agent_kwargs has rag: True

6. Source citations across the whole team

Whatever deep-agent pattern you use, every chunk retrieved during the run flows back into the same rag_sources panel. This is the DRK_CACHE-style grounded UX, free of charge.

In [10]

python

rag.begin_run()
for tool_name, query in [
    ("rag_search", "python version"),
    ("rag_search", "streaming events"),
    ("rag_search", "integrations"),
]:
    tool = next(t for t in goal_agent._build_agent().tools if t.name == tool_name)
    tool.run(ToolContext(prompt=""), query=query, top_k=1)

consolidated = rag.end_run()
print(f"consolidated {len(consolidated)} unique sources for the run:")
for s in consolidated:
    print(f"  [{s.index}] {s.source} (chunk_id={s.chunk_id})")

Out

consolidated 3 unique sources for the run:
  [1] readme.md (chunk_id=readme::0)
  [2] streaming.md (chunk_id=streaming::0)
  [3] integrations.md (chunk_id=integrations::0)

Summary

Deep agent	Accepts `rag=`	How it forwards
`GoalAgent`	✅ explicit param	inner Agent built per task
`ReflectiveAgent`	✅ via `**agent_kwargs`	inner Agent built once
`AdaptiveAgent`	✅ via `**agent_kwargs`	inner Agent + dynamic tools
`Supervisor`	✅ explicit param	every worker Agent in `with_builtins`
`PersistentAgent`	✅ explicit param	inner Agent + checkpoints

The DX rule of thumb: if you can pass rag= to Agent, you can pass it to any deep agent the same way.

RAG + Deep Agents

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

In [1]

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 import Agent
from shipit_agent.deep import (
    AdaptiveAgent,
    Goal,
    GoalAgent,
    PersistentAgent,
    ReflectiveAgent,
    Supervisor,
    Worker,
)
from shipit_agent.llms.base import LLMResponse
from shipit_agent.rag import RAG, HashingEmbedder
from shipit_agent.tools.base import ToolContext

Setup — shared knowledge base + mock LLM

In [2]

python

rag = RAG.default(embedder=HashingEmbedder(dimension=512))
rag.index_text(
    "Shipit supports Python 3.10 and newer. CLI and Python API.",
    document_id="readme",
    source="readme.md",
)
rag.index_text(
    "Shipit agents stream events in real time via agent.stream().",
    document_id="streaming",
    source="streaming.md",
)
rag.index_text(
    "Shipit deep agents include GoalAgent, ReflectiveAgent, Supervisor, AdaptiveAgent, PersistentAgent.",
    document_id="deep",
    source="deep_agents.md",
)
rag.index_text(
    "Shipit integrates with Jira, Linear, Slack, Gmail, Google Drive, Notion, Confluence.",
    document_id="integrations",
    source="integrations.md",
)
print(f"{rag.count()} chunks indexed")


class ChatLLM:
    def __init__(self):
        self.calls = 0

    def complete(self, messages, **kwargs):
        self.calls += 1
        return LLMResponse(
            content="Based on the docs, Shipit supports Python 3.10+ and streams events. [1][2]",
            usage={"prompt_tokens": 5, "completion_tokens": 5, "total_tokens": 10},
        )


llm = ChatLLM()

Out

4 chunks indexed

1. GoalAgent + RAG

GoalAgent decomposes a goal into sub-tasks and runs each through an inner Agent. Pass rag= and every sub-task can search the knowledge base.

In [3]

python

goal_agent = GoalAgent(
    llm=llm,
    goal=Goal(
        objective="Write a 2-sentence summary of Shipit runtime features",
        success_criteria=["mentions python version", "mentions streaming"],
    ),
    rag=rag,
)

inner = goal_agent._build_agent()
print("inner agent tool names:", sorted(t.name for t in inner.tools))
print("inner agent prompt has RAG section:", "rag_search" in inner.prompt)

Out

inner agent tool names: ['rag_fetch_chunk', 'rag_list_sources', 'rag_search']
inner agent prompt has RAG section: True

Driving the inner Agent for a citation demo

Each inner Agent run captures sources into the same RAG instance. We drive the rag_search tool manually so the notebook works without a real LLM.

In [4]

python

rag.begin_run()
search = next(t for t in inner.tools if t.name == "rag_search")
search.run(ToolContext(prompt=""), query="python version", top_k=1)
search.run(ToolContext(prompt=""), query="streaming", top_k=1)
sources = rag.end_run()
for s in sources:
    print(f"[{s.index}] {s.source} → {s.text[:80]}")

Out

[1] readme.md → Shipit supports Python 3.10 and newer. CLI and Python API.
[2] streaming.md → Shipit agents stream events in real time via agent.stream().

2. ReflectiveAgent + RAG

ReflectiveAgent generates an answer, critiques it, and revises if needed. The same rag= works.

In [5]

python

reflective = ReflectiveAgent(llm=llm, rag=rag)
inner = reflective._build_agent()
print("reflective inner tools:", sorted(t.name for t in inner.tools))

Out

reflective inner tools: ['rag_fetch_chunk', 'rag_list_sources', 'rag_search']

3. AdaptiveAgent + RAG

AdaptiveAgent can write new tools at runtime and still use RAG for grounding.

In [6]

python

adaptive = AdaptiveAgent(llm=llm, rag=rag)
print("adaptive agent_kwargs has rag:", "rag" in adaptive.agent_kwargs)
print("adaptive can_create_tools:", adaptive.can_create_tools)

Out

adaptive agent_kwargs has rag: True
adaptive can_create_tools: True

4. Supervisor + RAG (one shared knowledge base for the whole team)

Supervisor.with_builtins(..., rag=rag) wires the same RAG into every worker. A researcher and a writer can both retrieve from — and cite — the same chunks.

In [7]

python

supervisor = Supervisor.with_builtins(
    llm=llm,
    worker_configs=[
        {"name": "researcher", "prompt": "You research the codebase."},
        {"name": "writer", "prompt": "You write user-facing answers."},
    ],
    rag=rag,
)
for name, worker in supervisor.workers.items():
    tools = sorted(t.name for t in worker.agent.tools)
    rag_tools = [t for t in tools if t.startswith("rag_")]
    print(f"worker={name}: rag tools = {rag_tools}")

Out

worker=researcher: rag tools = ['rag_fetch_chunk', 'rag_list_sources', 'rag_search']
worker=writer: rag tools = ['rag_fetch_chunk', 'rag_list_sources', 'rag_search']

4.1 Building a supervisor manually with RAG-equipped Agents

If you want to construct workers by hand instead of using with_builtins, just create each worker's Agent with rag=rag directly:

In [8]

python

researcher_agent = Agent(llm=llm, prompt="Research mode.", rag=rag)
writer_agent = Agent(llm=llm, prompt="Writer mode.", rag=rag)

manual_supervisor = Supervisor(
    llm=llm,
    workers=[
        Worker(name="researcher", agent=researcher_agent),
        Worker(name="writer", agent=writer_agent),
    ],
    rag=rag,
)
print("supervisor.rag is rag:", manual_supervisor.rag is rag)
for name, w in manual_supervisor.workers.items():
    has_rag = any(t.name == "rag_search" for t in w.agent.tools)
    print(f"  worker={name}: rag_search wired = {has_rag}")

Out

supervisor.rag is rag: True
  worker=researcher: rag_search wired = True
  worker=writer: rag_search wired = True

5. PersistentAgent + RAG

PersistentAgent checkpoints long runs and forwards rag= to every Agent it builds, so RAG state survives across resumes (the chunks are in the store, not the checkpoint).

In [9]

python

persistent = PersistentAgent(llm=llm, rag=rag, checkpoint_interval=1, max_steps=2)
print("persistent agent_kwargs has rag:", "rag" in persistent.agent_kwargs)

Out

persistent agent_kwargs has rag: True

6. Source citations across the whole team

Whatever deep-agent pattern you use, every chunk retrieved during the run flows back into the same rag_sources panel. This is the DRK_CACHE-style grounded UX, free of charge.

In [10]

python

rag.begin_run()
for tool_name, query in [
    ("rag_search", "python version"),
    ("rag_search", "streaming events"),
    ("rag_search", "integrations"),
]:
    tool = next(t for t in goal_agent._build_agent().tools if t.name == tool_name)
    tool.run(ToolContext(prompt=""), query=query, top_k=1)

consolidated = rag.end_run()
print(f"consolidated {len(consolidated)} unique sources for the run:")
for s in consolidated:
    print(f"  [{s.index}] {s.source} (chunk_id={s.chunk_id})")

Out

consolidated 3 unique sources for the run:
  [1] readme.md (chunk_id=readme::0)
  [2] streaming.md (chunk_id=streaming::0)
  [3] integrations.md (chunk_id=integrations::0)

Summary

Deep agent	Accepts `rag=`	How it forwards
`GoalAgent`	✅ explicit param	inner Agent built per task
`ReflectiveAgent`	✅ via `**agent_kwargs`	inner Agent built once
`AdaptiveAgent`	✅ via `**agent_kwargs`	inner Agent + dynamic tools
`Supervisor`	✅ explicit param	every worker Agent in `with_builtins`
`PersistentAgent`	✅ explicit param	inner Agent + checkpoints

The DX rule of thumb: if you can pass rag= to Agent, you can pass it to any deep agent the same way.