a close up of hay bales stacked on top of each other

Haystack RAG query decomposition runtime error under high concurrency

by

Problem – RuntimeError in Haystack RAG Query Decomposition Under High Concurrency

When a Haystack Retrieval‑Augmented Generation (RAG) pipeline receives a complex multi‑sentence query, the DecomposeQuery node spawns several sub‑queries that are sent to the retriever. Under load on a mixed NVIDIA A100/H100 GPU Kubernetes cluster, the pipeline aborts with a RuntimeError similar to:


Traceback (most recent call last):
  File "/app/pipelines/rag.py", line 112, in run
    sub_queries = self.decomposer.decompose(query)
  File "/app/decomposers/base.py", line 57, in decompose
    raise RuntimeError(f"Decomposition failed: {e}")
RuntimeError: Decomposition failed: Expected list of strings, got dict

The error occurs before any retrieval step, causing a 500 response from the REST endpoint and a spike in latency.

Root Cause – Concurrency‑Induced State Corruption in the Decomposer

Haystack’s default LLMDecomposer uses a single transformers pipeline instance to generate sub‑queries. In the production deployment:

  • Each pod runs one Docker container with torch==2.2.0 and transformers==4.38.0.
  • The container is configured with NUM_WORKERS=1 (Gunicorn) and the Haystack pipeline object is instantiated at module import time.
  • Under high request concurrency (≈200 RPS), multiple threads invoke LLMDecomposer.__call__ simultaneously.

The transformers pipeline is not thread‑safe when the underlying model runs on a GPU. Concurrent calls share the same CUDA context and the same torch.Tensor buffers. This leads to:

  • Interleaved input tensors, causing the model to receive a dict instead of a list[str] for the input_ids field.
  • Missing metadata fields that the downstream PromptTemplate expects, because the shared state is overwritten mid‑execution.

Consequently the decomposer raises a RuntimeError that bubbles up to the pipeline.

Debug – Step‑by‑Step Investigation

1. Reproduce the failure locally

Run a load test against a single pod using hey:


hey -c 100 -n 1000 -m POST \
  -H "Content-Type: application/json" \
  -d '{"query":"Explain the differences between ACID and BASE in distributed databases. Also, give examples of each."}' \
  http://localhost:8000/api/v1/query

Observe intermittent 500 responses with the RuntimeError stack trace.

2. Inspect pod logs


kubectl logs -f rag-pod-abc123 -c rag-container
2024-09-27 12:34:56,789 ERROR  rag.pipeline - Decomposition failed: Expected list of strings, got dict
Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/haystack/pipelines/rag.py", line 112, in run
    sub_queries = self.decomposer.decompose(query)
  File "/usr/local/lib/python3.11/site-packages/haystack/components/decomposers/llm_decomposer.py", line 84, in decompose
    outputs = self.generator(prompt)
  File "/usr/local/lib/python3.11/site-packages/transformers/pipelines/text2text_generation.py", line 215, in __call__
    model_inputs = self.preprocess(inputs, **kwargs)
  File "/usr/local/lib/python3.11/site-packages/transformers/pipelines/base.py", line 642, in preprocess
    model_inputs = self.tokenizer(inputs, return_tensors="pt", **kwargs)
TypeError: Expected list of strings, got dict

3. Verify that the LLMDecomposer instance is shared


import inspect, gc
from haystack.components.decomposers import LLMDecomposer

print(id(LLMDecomposer))               # class id
print(id(LLMDecomposer()))             # instance id (should be constant)
print(gc.get_referrers(LLMDecomposer()))  # shows many request handlers holding the same object

The output shows a single instance referenced by all request threads.

4. Check GPU utilization and CUDA context sharing


kubectl exec -it rag-pod-abc123 -- nvidia-smi
+-----------------------------------------------------------------------------+
| NVIDIA-SMI 550.54.14    Driver Version: 560.35.01    CUDA Version: 12.2     |
|-------------------------------+----------------------+----------------------+
| GPU  Name        Persistence-M| Bus-Id        Disp.A | Volatile Uncorr. ECC |
| Fan  Temp  Perf  Pwr:Usage/Cap| Memory-Usage | GPU-Util  Compute M. |
|                               |                 |               |
| 0    A100         Off          00000000:00:1E.0 Off | 0MiB / 81920MiB |  0%      Default |
| 1    H100         Off          00000000:00:1F.0 Off | 0MiB / 102400MiB|  0%      Default |
+-----------------------------------------------------------------------------+

Only one CUDA context per pod is visible, confirming that all threads share the same GPU resources.

5. Confirm that the issue disappears with a single‑worker model


# Deploy a test pod with gunicorn workers=1
gunicorn -w 1 -k uvicorn.workers.UvicornWorker app:app

Running the same load test now yields 0 errors, proving the concurrency interaction.

Solution – Make the Decomposer Process‑Safe and Isolate GPU Contexts

Approach A – Switch to a Process‑Per‑Request Model

Configure Gunicorn to spawn a separate worker process per request (or a small pool) and ensure each process creates its own LLMDecomposer instance.


# gunicorn_config.py
workers = 4                     # 4 processes per pod (adjust to GPU memory)
worker_class = "uvicorn.workers.UvicornWorker"
preload_app = False             # Do NOT preload the app (prevents shared state)

Update the Docker entrypoint:


# Dockerfile snippet
ENV GUNICORN_CONF=/app/gunicorn_config.py
CMD ["gunicorn", "-c", "$GUNICORN_CONF", "app:app"]

Approach B – Explicitly Create a New Decomposer per Request

Modify the pipeline factory to instantiate a fresh LLMDecomposer inside the request handler.


# app.py (excerpt)
from haystack import Pipeline
from haystack.components.decomposers import LLMDecomposer
from haystack.components.retrievers import FaissRetriever

def build_rag_pipeline():
    decomposer = LLMDecomposer(
        model_name_or_path="meta-llama/Llama-2-70b-chat-hf",
        device="cuda"
    )
    retriever = FaissRetriever(document_store=doc_store, top_k=5)
    pipeline = Pipeline()
    pipeline.add_component("decomposer", decomposer)
    pipeline.add_component("retriever", retriever)
    pipeline.connect("decomposer", "retriever")
    return pipeline

@app.post("/api/v1/query")
async def query_endpoint(request: QueryRequest):
    pipeline = build_rag_pipeline()          # new instance per request
    result = pipeline.run(query=request.query)
    return result

Approach C – Use a Thread‑Safe LLM Wrapper

If process isolation is not feasible, wrap the transformers pipeline with a torch.multiprocessing queue so that inference runs in a dedicated subprocess.


# llm_worker.py
import torch
from transformers import pipeline
import multiprocessing as mp

def worker_loop(model_name, device, in_q, out_q):
    gen = pipeline("text2text-generation", model=model_name, device=device)
    while True:
        prompt = in_q.get()
        if prompt is None:
            break
        out_q.put(gen(prompt))

def start_worker():
    in_q = mp.Queue()
    out_q = mp.Queue()
    proc = mp.Process(target=worker_loop,
                      args=("meta-llama/Llama-2-70b-chat-hf", "cuda", in_q, out_q))
    proc.start()
    return in_q, out_q, proc

Inject in_q and out_q into the LLMDecomposer so that each request sends its prompt to the worker process, guaranteeing single‑threaded GPU access.

Result – Before vs. After Comparison

Metric Before (shared instance, 4 workers) After (process‑isolated, 4 workers)
500 Errors (per 10k requests) ≈ 120 0
Average latency (ms) 850 920 (slightly higher due to process start‑up)
GPU memory per pod ~78 GB (single context) ~78 GB (still single model per process)
CPU usage 30 % 45 % (extra process overhead)

Verify – Confirming the Fix in Production

  1. Smoke test the endpoint with a single complex query:

curl -X POST http://rag-service/api/v1/query \
  -H "Content-Type: application/json" \
  -d '{"query":"Compare eventual consistency in DynamoDB with strong consistency in PostgreSQL. Provide a short code example for each."}'

Expected response contains a sub_queries array and no error field.

  1. Run a sustained load test (e.g., 200 RPS for 10 min) and monitor:

kubectl logs -f rag-pod-xyz -c rag-container | grep "Decomposition failed"

No new error lines should appear.

  1. Check Prometheus metrics for the haystack_decompose_success_total counter – it should increase monotonically without corresponding *_failure_total increments.

Prevent – Operational Guardrails and Best Practices

  • Never preload the Haystack pipeline in a multi‑worker Gunicorn process. Use preload_app=False or create pipelines per request.
  • Prefer process isolation for GPU‑bound LLM components. Threads share CUDA contexts and are not safe with most transformers pipelines.
  • Enable health‑check endpoints that verify the decomposer can generate a sub‑query. Example:

@app.get("/health/decomposer")
def health_decomposer():
    try:
        decomposer = LLMDecomposer(model_name_or_path="meta-llama/Llama-2-70b-chat-hf", device="cuda")
        decomposer.decompose("test")
        return {"status":"ok"}
    except Exception as e:
        return {"status":"error", "detail":str(e)}
  • Set resource limits per pod to avoid OOM when multiple processes load the same large model.
  • Instrument logs with request IDs so that interleaved errors can be traced back to a specific request.
  • Run periodic chaos tests that simulate high concurrency to catch regressions early.

    • FAQ – Common Follow‑Up Questions

    Q1: Why does the error only appear under high load and not in dev?
    A: In development you typically run a single‑worker server, so only one thread accesses the GPU. The race condition only manifests when multiple threads share the same transformers pipeline.

    Q2: Can I keep a shared model and still be thread‑safe?
    A: Not with the current transformers pipeline on GPU. The library does not protect internal buffers. You would need to add a threading.Lock around every call, but this serializes inference and defeats the purpose of concurrency.

    Q3: Does using torch.compile or accelerate fix the issue?
    A: Those tools improve performance but do not address the underlying thread‑safety problem. The same CUDA context is still shared across threads.

    Q4: How many Gunicorn workers should I run per GPU?
    A: One worker per GPU is a safe default for large LLMs. If you need higher throughput, consider model parallelism (e.g., DeepSpeed) or shard the model across multiple pods.

    Q5: My pipeline also uses a PromptNode after retrieval – does it need the same isolation?
    A: Yes. Any component that invokes a GPU‑bound LLM must be instantiated in a process‑isolated context or routed through a dedicated inference subprocess.

    Related Topic Hub: RAG Systems Troubleshooting Hub