> ## Documentation Index
> Fetch the complete documentation index at: https://docs.seekr.com/llms.txt
> Use this file to discover all available pages before exploring further.
# Simple RAG agent
> Learn the SeekrFlow building blocks: upload a file, index it in a vector store, and give an agent a FileSearch tool to answer from it.
export const DownloadNotebook = ({href, filename}) =>
;
};
This recipe walks through the fundamental parts of the SeekrFlow SDK by building
a retrieval-augmented generation (RAG) agent end to end: upload a document,
index it in a vector database, wrap that database in a FileSearch tool, attach
the tool to an agent, promote the agent, and ask it a question.
The example writes its own sample document, so you can run the whole thing
without preparing any files first. Once it works, swap in your own documents.
## What you'll build
An agent that:
1. Indexes a document into a vector database.
2. Searches that database through a FileSearch tool.
3. Answers questions grounded in what it retrieves.
## Prerequisites
* A SeekrFlow API key, set as the `SEEKR_API_KEY` environment variable
* Python 3.8 or later
* The SeekrFlow SDK: `pip install seekrai`
This recipe creates billable resources. When you are done, remove them with the
[cleanup step](#clean-up).
## Build it
Create `rag_agent.py` with the imports, configuration, and client. `RUN_ID` adds
a short unique suffix to the resource names so repeat runs don't collide.
```python rag_agent.py theme={null}
import os
import time
import tempfile
import uuid
from pathlib import Path
from seekrai import SeekrFlow
from seekrai.types import FilePurpose
from seekrai.types.tools import CreateFileSearch, FileSearchConfig
from seekrai.types.agents.agent import CreateAgentRequest
API_KEY = os.environ["SEEKR_API_KEY"]
API_URL = "https://flow.seekr.com/v1/"
MODEL_ID = "meta-llama/Llama-3.3-70B-Instruct"
EMBED_MODEL = "intfloat/e5-mistral-7b-instruct"
RUN_ID = uuid.uuid4().hex[:8]
client = SeekrFlow(api_key=API_KEY, base_url=API_URL)
```
Write a short Markdown document to a temporary file and upload it with
`purpose=FilePurpose.Alignment`, which marks it for data-engine ingestion.
```python rag_agent.py theme={null}
document = """\
# Seekr Technologies — Quick Facts
Seekr Technologies is an AI company focused on developing trustworthy AI systems.
Key facts:
- Headquarters: Reston, Virginia, USA
- Product: SeekrFlow, a platform for building and deploying AI agents
- Mission: responsible AI that people can trust
- SeekrFlow supports multi-agent orchestration, RAG, and custom Python tools
"""
with tempfile.NamedTemporaryFile(mode="w", suffix=".md", delete=False, prefix="seekr_facts_") as f:
f.write(document)
tmp_path = Path(f.name)
file_obj = client.files.upload(tmp_path, purpose=FilePurpose.Alignment)
print(f"Uploaded: {file_obj.id} ({file_obj.filename})")
```
To index several documents, upload them with `client.files.bulk_upload`, collect
the returned IDs, and include them all in a single `create_ingestion_job` call.
```python theme={null}
paths = ["guidebook.pdf", "policies.md", "faq.docx"]
uploaded = client.files.bulk_upload(paths, purpose=FilePurpose.Alignment)
file_ids = [f.id for f in uploaded]
# Then ingest them together:
# client.vector_database.create_ingestion_job(database_id=db.id, files=file_ids)
```
Create a vector database. Documents you ingest are embedded with the model you
name here and stored for retrieval.
```python rag_agent.py theme={null}
db = client.vector_database.create(
name=f"RAG Demo DB {RUN_ID}",
model=EMBED_MODEL,
description="Demo vector database for the RAG agent walkthrough.",
)
print(f"Created vector database: {db.id}")
```
Start an ingestion job to chunk, embed, and store the file, then poll until it
completes.
```python rag_agent.py theme={null}
job = client.vector_database.create_ingestion_job(database_id=db.id, files=[file_obj.id])
print(f"Ingestion job: {job.id} ({job.status})")
deadline = time.time() + 300
while time.time() < deadline:
job = client.vector_database.retrieve_ingestion_job(db.id, job.id)
print(f" status: {job.status}")
if job.status == "completed":
break
if job.status in ("failed", "error"):
raise RuntimeError(f"Ingestion failed: {job.error_message}")
time.sleep(5)
else:
raise TimeoutError("Ingestion did not complete within 300s.")
print("Ingestion complete.")
```
Wrap the vector database in a FileSearch tool. The `file_search_index` points the
tool at your database, and the `description` tells the agent when to use it.
```python rag_agent.py theme={null}
tool = client.tools.create(
CreateFileSearch(
name=f"seekr_facts_search_{RUN_ID}",
description="Search the Seekr Technologies knowledge base for facts about the company.",
config=FileSearchConfig(file_search_index=db.id),
)
)
print(f"Created tool: {tool.id}")
```
Create an agent, attach the tool by ID, and instruct it to answer from what the
tool retrieves.
```python rag_agent.py theme={null}
agent = client.agents.create(
CreateAgentRequest(
name=f"RAG Demo Agent {RUN_ID}",
instructions=(
"You are a helpful assistant with access to a knowledge base about Seekr Technologies. "
"When the user asks a question, search the knowledge base using your tool and answer "
"based on what you find."
),
model_id=MODEL_ID,
tool_ids=[tool.id],
)
)
print(f"Created agent: {agent.id}")
```
Promote the agent to deploy it, then wait for it to become `Active`.
```python rag_agent.py theme={null}
client.agents.promote(agent.id)
deadline = time.time() + 120
while time.time() < deadline:
agent = client.agents.retrieve(agent.id)
print(f" status: {agent.status.value}")
if agent.status.value == "Active":
break
if agent.status.value == "Failed":
raise RuntimeError("Agent failed to deploy.")
time.sleep(3)
else:
raise TimeoutError("Agent did not become Active within 120s.")
print("Agent is Active.")
```
Create a thread, send a question, wait for the run to finish, and print the
agent's reply.
```python rag_agent.py theme={null}
thread = client.agents.threads.create()
client.agents.threads.create_message(
thread_id=thread.id, role="user",
content="Where is Seekr Technologies headquartered and what do they make?",
)
run_response = client.agents.runs.run(agent_id=agent.id, thread_id=thread.id, stream=False)
deadline = time.time() + 300
while time.time() < deadline:
run = client.agents.runs.retrieve(run_response.run_id, thread.id)
if run.status.value == "completed":
break
if run.status.value in ("failed", "canceled"):
raise RuntimeError(f"Run ended with status: {run.status.value}")
time.sleep(3)
else:
raise TimeoutError("Run did not complete within 300s.")
messages = client.agents.threads.list_messages(thread.id, limit=10, order="desc")
answer = next((m.content for m in messages if m.role == "assistant"), "(no assistant message found)")
print("\nAgent answer:\n")
print(answer if isinstance(answer, str) else str(answer))
```
Run the finished script:
```bash theme={null}
python rag_agent.py
```
The agent searches the indexed document and answers that Seekr Technologies is
headquartered in Reston, Virginia, and builds SeekrFlow.
Clean up resources (optional)
This shuts down the agent and removes the tool, vector database, and file the
recipe created.
```python rag_agent.py theme={null}
client.agents.demote(agent.id)
client.agents.delete(agent.id)
client.tools.delete(tool.id)
client.vector_database.delete(db.id)
client.files.delete(file_obj.id)
print("Deleted the agent, tool, vector database, and file.")
```
## Next steps
* **Use your own documents.** Replace the inline document with real files (PDF,
DOCX, or Markdown) to build a knowledge base for your own domain.
* **Add citations and confidence.** See [New hire onboarding agent with citations](/flow/recipes/new-hire-onboarding-agent-with-citations) for a version that rates its confidence and cites sources.
* **Tune retrieval.** Set `top_k` and `score_threshold` on `FileSearchConfig` to
control how much context the tool returns.