RagPipeline

Plug-and-play boilerplate that wires real-time events to any embedding model and any vector store.

RagPipeline (sync) and AsyncRagPipeline (async) connect the Autoplay event stream to your vector store with minimal code. You provide two functions — embed and upsert — and the pipeline handles the rest.

---

Sync pipeline

import openai
from autoplay_sdk import ConnectorClient
from autoplay_sdk.rag import RagPipeline

openai_client = openai.OpenAI()

pipeline = RagPipeline(
    embed=lambda text: openai_client.embeddings.create(
        input=text, model="text-embedding-3-small"
    ).data[0].embedding,
    upsert=lambda id, vector, meta: pinecone_index.upsert([(id, vector, meta)]),
)

ConnectorClient(url=URL, token=TOKEN) \
    .on_actions(pipeline.on_actions) \
    .on_summary(pipeline.on_summary) \
    .run()

---

Async pipeline

import openai
from autoplay_sdk import AsyncConnectorClient
from autoplay_sdk.rag import AsyncRagPipeline

openai_client = openai.AsyncOpenAI()

async def embed(text: str) -> list[float]:
    r = await openai_client.embeddings.create(input=text, model="text-embedding-3-small")
    return r.data[0].embedding

async def upsert(id: str, vector: list[float], meta: dict) -> None:
    await pinecone_index.upsert([(id, vector, meta)])

pipeline = AsyncRagPipeline(embed=embed, upsert=upsert)

async with AsyncConnectorClient(url=URL, token=TOKEN) as client:
    client.on_actions(pipeline.on_actions)
    client.on_summary(pipeline.on_summary)
    await client.run()

---

With a SessionSummarizer

Attach a SessionSummarizer to automatically condense actions before embedding. This keeps your vector store entries compact and your context window small.

from autoplay_sdk.rag import RagPipeline
from autoplay_sdk.summarizer import SessionSummarizer

summarizer = SessionSummarizer(llm=my_llm_fn, threshold=10)

pipeline = RagPipeline(
    embed=embed_fn,
    upsert=upsert_fn,
    summarizer=summarizer,  # actions go through summarizer first
)

client.on_actions(pipeline.on_actions).on_summary(pipeline.on_summary).run()

When the summarizer fires (every 10 actions), the LLM-generated summary is embedded and upserted — not the raw action batch.

---

Compatible vector stores

The upsert callable works with any vector store. Examples:

Pinecone

from pinecone import Pinecone

pc = Pinecone(api_key="...")
index = pc.Index("my-index")

upsert = lambda id, vector, meta: index.upsert(vectors=[{"id": id, "values": vector, "metadata": meta}])

import chromadb

client = chromadb.Client()
collection = client.get_or_create_collection("sessions")

upsert = lambda id, vector, meta: collection.upsert(ids=[id], embeddings=[vector], metadatas=[meta])

import weaviate

wv = weaviate.connect_to_local()
collection = wv.collections.get("Session")

upsert = lambda id, vector, meta: collection.data.insert(properties=meta, vector=vector, uuid=id)

import psycopg2

conn = psycopg2.connect("...")

def upsert(id, vector, meta):
    with conn.cursor() as cur:
        cur.execute(
            "INSERT INTO sessions (id, embedding) VALUES (%s, %s) ON CONFLICT (id) DO UPDATE SET embedding = EXCLUDED.embedding",
            (id, vector)
        )
    conn.commit()

---

What gets upserted

Event type	ID used	Text embedded
ActionsPayload	session_id	payload.to_text() — numbered action list
SummaryPayload	session_id	payload.to_text() — prose summary
Client summary (via SessionSummarizer)	session_id	LLM-generated summary text

Each event type upserts with the same session_id as the key — so your vector store always has one up-to-date entry per session.

---

Constructor

RagPipeline(embed, upsert, summarizer=None)

embed · Callable[[str], list[float]] · required

Any embedding function. Receives payload.to_text() and must return a vector.

upsert · Callable[[str, list[float], dict], None] · required

Writes the embedding to your vector store. Called with (session_id, vector, metadata).

summarizer · SessionSummarizer | None · default: "None"

Optional SessionSummarizer. When set, actions are summarised before embedding.

AsyncRagPipeline(embed, upsert, summarizer=None)

Same parameters but embed and upsert are async callables.

---

Callbacks

Method	Description
.on_actions(payload)	Wire to client.on_actions(...)
.on_summary(payload)	Wire to client.on_summary(...)