zylon-ai/private-gpt · error · AsyncIteratorError

Item transformation failed: {e!s}

Error message

Item transformation failed: {e!s}

What it means

Wrapped as AsyncIteratorError by to_async_iterator when the user-supplied transform_fn raises while processing an item. transform_fn is executed in the thread pool through loop.run_in_executor; any exception it throws (TypeError, KeyError, unexpected input shape) is caught, re-raised as AsyncIteratorError('Item transformation failed: ...'), and the async generator terminates - no later items are produced.

Source

Thrown at private_gpt/utils/async_utils.py:87

                        f"Iterator next() operation failed: {e!s}"
                    ) from e

            if not chunk:
                break

            # Process the chunk
            for item in chunk:
                try:
                    if transform_fn:
                        # Run transform in executor if it's CPU-intensive
                        result = await loop.run_in_executor(
                            internal_executor, transform_fn, item
                        )
                        yield result
                    else:
                        yield item
                except Exception as e:
                    raise AsyncIteratorError(
                        f"Item transformation failed: {e!s}"
                    ) from e

    except asyncio.CancelledError:
        raise
    except Exception as e:
        raise AsyncIteratorError(f"Async iterator conversion failed: {e!s}") from e
    finally:
        if not executor:
            internal_executor.shutdown(wait=False)

View on GitHub (pinned to 4a030776a3)

Solutions

1. Make transform_fn total: wrap its body in try/except and return a fallback (e.g. None or an error-record) for bad items, then filter downstream
2. Validate/normalize items before passing them to to_async_iterator
3. Add logging inside transform_fn including the offending item so failures are diagnosable
4. If failures are transient (network), add retry with backoff inside the transform instead of letting it raise

Example fix

# before
ait = to_async_iterator(iter(docs), transform_fn=lambda d: d['text'])

# after
def get_text(d):
    try:
        return d['text']
    except KeyError:
        logger.warning(f"item missing text: {d!r}")
        return None
ait = to_async_iterator(iter(docs), transform_fn=get_text)

def safe_transform(fn, fallback=None):
    def wrapper(item):
        try:
            return fn(item)
        except Exception as e:
            logger.warning("transform failed for %r: %s", item, e)
            return fallback
    return wrapper

# pass wrapper instead of fn to to_async_iterator

null

try:
    async for out in to_async_iterator(it, transform_fn=fn):
        ...
except AsyncIteratorError as e:
    if "Item transformation" in str(e):
        logger.error("bad item: %s", e.__cause__)
        # decide: skip item (needs defensive fn) or abort stream

When it happens

Trigger: Passing transform_fn=lambda d: d['text'] when some items lack the 'text' key; a transform that deserializes JSON and hits malformed payload; type mismatches between what the iterator yields and what transform_fn expects; a transform with side effects (HTTP call) that fails transiently.

Common situations: Document-processing pipelines (chunkers, embedders, parsers) where one malformed document kills the whole stream; upgrading a dependency that changes item schema so the transform no longer matches; LLM response parsing where the model occasionally returns an unexpected structure.

Related errors

• Iterator next() operation failed: {e!s}
• Async iterator conversion failed: {e!s}
• INVALID_REQUEST_ERROR
• No items returned from astream_structured_predict
• API base URL and poll interval must be provided in async mod

AI-assisted analysis of zylon-ai/private-gpt@4a030776a3 (2026-08-15). Data as JSON: /api/errors/8604daa5f9f85fb0. Report an issue: GitHub.