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)

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