Simple RAG agent

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:

Indexes a document into a vector database.
Searches that database through a FileSearch tool.
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.

Build it

Set up the client

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.

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)

Create and upload a document

Write a short Markdown document to a temporary file and upload it with purpose=FilePurpose.Alignment, which marks it for data-engine ingestion.

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})")

Upload several files at once

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.

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

Create a vector database. Documents you ingest are embedded with the model you name here and stored for retrieval.

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}")

Ingest the document

Start an ingestion job to chunk, embed, and store the file, then poll until it completes.

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.")

Create a FileSearch tool

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.

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 the agent

Create an agent, attach the tool by ID, and instruct it to answer from what the tool retrieves.

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

Promote the agent to deploy it, then wait for it to become Active.

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.")

Ask a question

Create a thread, send a question, wait for the run to finish, and print the agent’s reply.

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 script

Run the finished script:

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.

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 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.

​What you’ll build

​Prerequisites

​Build it

​Clean up resources (optional)

​Next steps

What you’ll build

Prerequisites

Build it

Clean up resources (optional)

Next steps