Processors

The two processors are the main entry points for the SDK. Configure once, then call ingest() or search() as many times as needed.

IngestProcessor

chatbot_plugin_sdk.processors.ingest.IngestProcessor

IngestProcessor()

文章向量化寫入處理器。

Pipeline: normalize → chunk → embed (dense / sparse) → upsert via backend

Usage::

# ThreadPoolExecutor (sync psycopg2):
backend = SyncPgBackend(DatabaseConfig(...))

# FastAPI / native async (asyncpg):
backend = AsyncPgBackend(DatabaseConfig(...))

processor = IngestProcessor()
processor.configure(
    backend=backend,
    dense=EndpointProvider(url="http://embed:8080", dimension=768),
)
await processor.ingest(
    full_text="...",
    articles_column_values={
        "url": "https://example.com/article",  # required — used as idempotent key
        "title": "My Article",
    },
)

Thread-safety notes

• The processor itself holds no per-call mutable state after configure().
• _ready may be set concurrently by multiple threads during startup; the worst case is backend.setup() being called twice, which is idempotent.
• Use :class:SyncPgBackend for ThreadPoolExecutor + asyncio.run() patterns. :class:AsyncPgBackend must live inside a single event loop.

configure

configure(
    backend: DatabaseBackend,
    dense: DenseEmbeddingProvider | None = None,
    sparse: SparseEmbeddingProvider | None = None,
    embed_batch_size: int = 16,
    chunk_size: int = DEFAULT_CHUNK_SIZE,
    chunk_overlap: int = DEFAULT_CHUNK_OVERLAP,
) -> None

Bind backend + providers. Pure sync, no I/O.

Parameters:

Name	Type	Description	Default
embed_batch_size	int	Max chunks sent to each provider's embed() per call. Smaller values reduce peak memory when using local ONNX models (e.g. SPLADE). Default: 16.	16
chunk_size	int	Maximum characters per chunk. Default: 500.	DEFAULT_CHUNK_SIZE
chunk_overlap	int	Overlap characters between consecutive chunks. Default: 50.	DEFAULT_CHUNK_OVERLAP

ingest async

ingest(
    full_text: str,
    articles_column_values: dict[str, Any] | None = None,
    metadata: dict[str, Any] | None = None,
) -> None

Full ingest pipeline: normalize → chunk → embed → upsert.

Parameters:

Name	Type	Description	Default
full_text	str	Raw article text (HTML-stripped or plain).	required
articles_column_values	dict[str, Any] | None	SQL column values for the articles table. Must include url — it is used to derive a deterministic article_id via uuid.uuid5(NAMESPACE_URL, url) for idempotent upserts. Any other keys become INSERT columns; column existence is the caller's responsibility.	None
metadata	dict[str, Any] | None	Opaque JSONB metadata — the SDK never interprets its keys.	None

RetrieveProcessor

chatbot_plugin_sdk.processors.retrieve.RetrieveProcessor

RetrieveProcessor()

向量語意搜尋處理器（read-only）。

Supports three retrieval modes depending on configured providers:

• Dense-only: embed query → backend.search_dense() → cosine similarity score
• Sparse-only: embed query → backend.search_sparse() → inner-product score
• Hybrid (dense + sparse): both searches → RRF merge → optional reranker

When a reranker is configured, top_k * 3 candidates are fetched and re-scored before returning the final top_k results.

SDK 在回傳 SearchResponse（chunks）後即完成職責；LLM 生成由 caller 負責。

Usage::

retriever = RetrieveProcessor()
retriever.configure(
    backend=AsyncPgBackend(config),
    dense=EndpointProvider(url="http://embed:8080", dimension=768),
    sparse=LocalProvider(fn=splade_fn, dimension=30522),
    reranker=FastEmbedReranker(),
)
result = await retriever.retrieve("What is RAG?")
# result.chunks: list[ChunkResult] — pass to your LLM

configure

configure(
    backend: DatabaseBackend,
    dense: DenseEmbeddingProvider | None = None,
    sparse: SparseEmbeddingProvider | None = None,
    reranker: Reranker | None = None,
) -> None