Pipelines¶

Pipelines ingest content, chunk it, generate embeddings, and store vectors in Redis for RAG.

CLI¶

The CLI is the easiest local path:

rak pipelines run ./docs --pattern "*.md" --chunk-size 800
rak pipelines status
rak pipelines clear

Synchronous Python API¶

from pathlib import Path

import redis.asyncio as redis
from redis_agent_kit.pipelines import FileScraper, Pipeline, ProcessorConfig

client = redis.from_url("redis://localhost:6379", decode_responses=True)

pipeline = Pipeline(
    client,
    processor_config=ProcessorConfig(chunk_size=800, chunk_overlap=100),
    embedding_model="text-embedding-3-small",
)

scraper = FileScraper(Path("./docs"), patterns=["*.md"], recursive=True)
result = await pipeline.run(scraper)

print(result.documents_processed, result.chunks_created)

Two-Stage Pipeline¶

Use the orchestrator when you want to prepare files, inspect artifacts, then ingest them later.

from pathlib import Path

import redis.asyncio as redis
from redis_agent_kit import Vectorizer
from redis_agent_kit.pipelines import (
    PipelineConfig,
    PipelineOrchestrator,
    SourceConfig,
    VectorStore,
)

client = redis.from_url("redis://localhost:6379", decode_responses=True)

config = PipelineConfig(
    sources=[SourceConfig(name="docs", path_pattern="**/*.md")],
)

orchestrator = PipelineOrchestrator(
    config=config,
    base_path=Path("./docs"),
    vectorizer=Vectorizer(model="text-embedding-3-small"),
    vector_store=VectorStore(client),
)

batch_id = orchestrator.prepare()
manifest = orchestrator.ingest(batch_id)

print(batch_id, manifest.chunks_embedded)

GitHub Repository Source & Monitoring¶

Ingest a GitHub repo's files and keep the vector store fresh as the repo changes. This is the repo-shaped counterpart to the local-file pipeline: instead of globbing a directory, it pulls files over the GitHub REST API and re-syncs only what changed.

Change detection is cheap by design:

The repo's HEAD commit SHA is a one-call change signal — when it matches the last sync, the pass returns immediately without fetching the tree or any files.
Each file's blob SHA (from the git tree) is a free content hash, so only files whose content actually changed are downloaded and re-embedded.

Chunks are keyed by a stable per-file id (github://owner/repo@ref/path), so a modified file's chunks are replaced and a removed file's chunks are deleted — no stale chunks linger in search results.

CLI¶

# One sync pass (re-run anytime; unchanged HEAD is a no-op)
rak pipelines github redis/redis-vl-python --ref main -p "**/*.md"

# Poll on an interval and sync changes until stopped (Ctrl-C)
rak pipelines github-watch redis/redis-vl-python --ref main --interval 60

Auth: pass --token or set GITHUB_TOKEN (required for private repos; also lifts the rate limit from 60 to 5000 requests/hour). For GitHub Enterprise, set --api-base https://your-ghe-host/api/v3. Steady-state polling costs ~1 API call per tick thanks to the HEAD short-circuit.

Python API¶

import redis.asyncio as redis
from redis_agent_kit import Vectorizer
from redis_agent_kit.pipelines import (
    GitHubScraper,
    GitHubRepoMonitor,
    GitHubSyncEngine,
    RedisRepoStateStore,
    VectorStore,
)

client = redis.from_url("redis://localhost:6379", decode_responses=True)

scraper = GitHubScraper("redis", "redis-vl-python", ref="main", patterns=["**/*.md"])
monitor = GitHubRepoMonitor(scraper, RedisRepoStateStore(client, prefix="rak"))
engine = GitHubSyncEngine(
    vectorizer=Vectorizer(model="text-embedding-3-small"),
    vector_store=VectorStore(client, prefix="rak"),
)

result = await engine.sync(monitor)
print(result.added, result.modified, result.removed, result.chunks_embedded)

# Or run the built-in polling daemon:
# await engine.watch(monitor, interval=60)

Background submission¶

github_sync is exported in PIPELINE_WORKER_TASKS, so a worker can run repo syncs in the background:

from redis_agent_kit.pipelines.tasks import github_sync, PIPELINE_WORKER_TASKS

execution = await docket.add(github_sync)(
    owner="redis", repo="redis-vl-python", ref="main",
    patterns=["**/*.md"], redis_url=docket.url, prefix="rak",
)

Patterns match the full repo-relative path with fnmatch semantics (* spans /), so *.md and **/*.md both match Markdown at any depth. Binary / undecodable files are skipped. Very large repos may return a truncated tree (logged as a warning).

Vector Search¶

VectorStore.search() expects an embedding vector:

from redis_agent_kit import Vectorizer
from redis_agent_kit.pipelines import VectorStore

vectorizer = Vectorizer(model="text-embedding-3-small")
vector_store = VectorStore(client, prefix="rak")

query_embedding = await vectorizer.embed("How do I configure Redis?")
results = await vector_store.search(query_embedding, limit=5)

for result in results:
    print(result.score, result.content[:100])

Two-Stage Background Submission¶

PipelineOrchestrator.submit_prepare(), submit_ingest(), submit_full(), and submit_document() enqueue Docket tasks for the two-stage pipeline. The workers that execute these are exported as PIPELINE_WORKER_TASKS and must be registered on the worker side:

# worker.py
from redis_agent_kit.pipelines.tasks import PIPELINE_WORKER_TASKS

# A worker that handles both AgentKit tasks and pipeline jobs:
tasks = [kit.worker_task, *PIPELINE_WORKER_TASKS]

Then point rak worker --tasks at the module exposing the list, e.g.:

rak worker --name pipeline_worker --tasks worker:tasks

The submit methods return a Docket execution key; poll it via TaskManager (per-task channel) just like any other background task.

REST API¶

The API includes pipeline endpoints under /pipelines. The staged endpoints submit background tasks via the orchestrator above — workers must have PIPELINE_WORKER_TASKS registered.

curl -X POST http://localhost:8000/pipelines/run \
  -H "Content-Type: application/json" \
  -d '{
    "documents": [
      {"title": "Doc 1", "content": "Redis is fast.", "source": "api"}
    ],
    "chunk_size": 500,
    "chunk_overlap": 100
  }'

The staged endpoints submit background tasks and return task IDs:

curl -X POST http://localhost:8000/pipelines/prepare \
  -H "Content-Type: application/json" \
  -d '{"source_path": "./docs"}'

curl -X POST http://localhost:8000/pipelines/ingest \
  -H "Content-Type: application/json" \
  -d '{"batch_id": "01J..."}'

curl -X POST http://localhost:8000/pipelines/full \
  -H "Content-Type: application/json" \
  -d '{"source_path": "./docs"}'

curl -X POST http://localhost:8000/pipelines/documents \
  -H "Content-Type: application/json" \
  -d '{"filename": "note.md", "content": "# Note\nRedis content"}'

Status and clear:

curl http://localhost:8000/pipelines/status
curl -X DELETE http://localhost:8000/pipelines