Skip to content

Context manager

ContextManager #

ContextManager(openapi_service: OpenAPIService)

Shortens a datapoint's context for the specific question an annotator answers.

A long, general context (e.g. a full scene description) is often far more detail than a single question needs. This manager tunes a context down to what is relevant for the question, which keeps it within the length the backend accepts and focuses the annotator. Results are cached server-side.

Source code in src/rapidata/rapidata_client/context/context_manager.py 
36
37
38
def __init__(self, openapi_service: OpenAPIService):
    self._openapi_service = openapi_service
    logger.debug("ContextManager initialized")

shorten_context #

shorten_context(context: str, question: str) -> str

Shorten a single context for the given question.

Parameters:

Name Type Description Default
context str

The (potentially long) context to shorten.

 required
question str

The question the context will be shown alongside. The context is tuned to what this question needs.

 required

Returns:

Type Description
str

The shortened context.
Source code in src/rapidata/rapidata_client/context/context_manager.py 
40
41
42
43
44
45
46
47
48
49
50
51
def shorten_context(self, context: str, question: str) -> str:
    """Shorten a single context for the given question.

    Args:
        context: The (potentially long) context to shorten.
        question: The question the context will be shown alongside. The
            context is tuned to what this question needs.

    Returns:
        The shortened context.
    """
    return self.shorten_contexts([(context, question)])[0]

shorten_contexts #

shorten_contexts(
    pairs: Sequence[tuple[str, str]],
) -> list[str]

Shorten a batch of (context, question) pairs.

The pairs are sent in concurrent batched requests, with a progress bar while they run (suppressed by rapidata_config.logging.silent_mode).

Parameters:

Name Type Description Default
pairs Sequence[tuple[str, str]]

The (context, question) pairs to shorten.

 required

Returns:

Type Description
list[str]

The shortened contexts, in the same order as pairs.
Source code in src/rapidata/rapidata_client/context/context_manager.py 
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def shorten_contexts(self, pairs: Sequence[tuple[str, str]]) -> list[str]:
    """Shorten a batch of ``(context, question)`` pairs.

    The pairs are sent in concurrent batched requests, with a progress bar
    while they run (suppressed by ``rapidata_config.logging.silent_mode``).

    Args:
        pairs: The ``(context, question)`` pairs to shorten.

    Returns:
        The shortened contexts, in the same order as ``pairs``.
    """
    if not pairs:
        return []

    with tracer.start_as_current_span("ContextManager.shorten_contexts"):
        if len(pairs) <= SHORTEN_BATCH_SIZE:
            return self._shorten_batch(pairs)

        batches = [
            pairs[start : start + SHORTEN_BATCH_SIZE]
            for start in range(0, len(pairs), SHORTEN_BATCH_SIZE)
        ]
        results: list[list[str]] = [[] for _ in batches]
        current_context = otel_context.get_current()

        def shorten_batch(index: int) -> None:
            token = otel_context.attach(current_context)
            try:
                results[index] = self._shorten_batch(batches[index])
            finally:
                otel_context.detach(token)

        with ThreadPoolExecutor(
            max_workers=rapidata_config.upload.maxWorkers
        ) as executor:
            futures = {
                executor.submit(shorten_batch, index): index
                for index in range(len(batches))
            }
            with tqdm(
                total=len(pairs),
                desc="Shortening contexts",
                disable=rapidata_config.logging.silent_mode,
            ) as progress:
                for future in as_completed(futures):
                    future.result()
                    progress.update(len(batches[futures[future]]))

        return [context for batch in results for context in batch]