JuliusBrussee/caveman · error · TypeError

Synchronous document compression requires MiddlewareRuntime

Error message

Synchronous document compression requires MiddlewareRuntime

What it means

_compress is the synchronous document-compressor path and requires self.runtime to be a MiddlewareRuntime. An async runtime (AsyncMiddlewareRuntime) or None cannot run the sync optimize() call, so a TypeError is raised before computing options.

Solutions

Construct the compressor with a MiddlewareRuntime.
Use the async compression path (acompress_documents) when only an async runtime is available.
If compression is intentionally disabled, pass documents through without the compressor.

Example fix

// before
compressor = CavemanDocumentCompressor(async_runtime, scope)
compressor.compress_documents(docs, query)  # TypeError
// after
compressor = CavemanDocumentCompressor(MiddlewareRuntime(...), scope)
compressor.compress_documents(docs, query)

Defensive patterns

Strategy: type-guard

Validate before calling

if not isinstance(compressor.runtime, MiddlewareRuntime):
    raise RuntimeError('sync compression requires MiddlewareRuntime; use acompress_documents otherwise')

Type guard

def supports_sync_compression(c): return isinstance(getattr(c, 'runtime', None), MiddlewareRuntime)

Try / catch

try:
    docs = compressor.compress_documents(documents, query)
except TypeError as e:
    if 'Synchronous document compression' in str(e):
        docs = await compressor.acompress_documents(documents, query)
    else:
        raise

Prevention

Match compressor runtime type to the retriever's sync/async flavor.
In async RAG chains, always call acompress_documents.
Validate the runtime type at compressor construction, not at first use.

When it happens

Trigger: Calling compress_documents (sync) on a compressor constructed with an AsyncMiddlewareRuntime or with no runtime.

Common situations: Using the async retriever/compressor with a sync VectorStore.get_relevant_documents pipeline; RAG chain refactor swapped the runtime type but kept the sync compressor.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

AI-assisted analysis of JuliusBrussee/caveman@3ee70a1026 (2026-09-20). Data as JSON: /api/errors/030db30f35027ca1. Report an issue: GitHub.

Appendix: source

Thrown at packages/middleware/python/caveman_middleware/langchain.py:397

            self.runtime.report(reason="off" if self.runtime.mode == "off" else "unsupported_version", adapter="langchain-rag")
            return None
        if any(type(document) is not Document for document in documents):
            self.runtime.report(reason="unsupported_shape", adapter="langchain-rag")
            return None
        context = manifest([{"id": d.id, "metadata": d.metadata, "page_content": d.page_content} for d in documents])
        if context is None:
            self.runtime.report(reason="unsupported_shape", adapter="langchain-rag")
            return None
        scope = _scope(self.scope)
        reader = self.source_expansion
        binding = reader if (isinstance(reader, RecoveryBinding) and callable(reader.execute)
                             and self.runtime.owns_binding(reader, scope)) else None
        return dict(scope=scope, adapter=Adapter("langchain-rag", "0.1.0", "1.4.0", "langchain-document-v1"), manifest=context,
                    candidates=[Candidate(f"document-{i}", d.page_content, d.id or f"document-{i}", kind="artifact") for i, d in enumerate(documents)], binding=binding)

    def _compress(self, documents, query):
        if not isinstance(self.runtime, MiddlewareRuntime):
            raise TypeError("Synchronous document compression requires MiddlewareRuntime")
        documents = list(documents)
        options = self._options(documents, query)
        return documents if options is None else self._apply_documents(documents, self.runtime.optimize(**options))

    def _apply_documents(self, documents, outcome):
        replacements = {r["segment_id"]: r["text"] for r in outcome.replacements}
        if not replacements.keys() <= {f"document-{i}" for i in range(len(documents))}:
            self.runtime.report(reason="invalid_replacement_plan", adapter="langchain-rag")
            return documents
        result = [document.model_copy(update={"page_content": replacements[f"document-{i}"]}) if f"document-{i}" in replacements else document for i, document in enumerate(documents)]
        self.runtime.report(outcome, adapter="langchain-rag")
        return result

View on GitHub (pinned to 3ee70a1026)