Components and Functions

Top-level functions

Prodigy provides the following top level utilities for writing your own scripts and recipes. To use them, import the prodigy module at the top of your file.

prodigy.serve function

Serve a Prodigy recipe and start the web app from Python. Does the same as the prodigy command on the command line. Since you’re already in a Python script, you don’t have to use the -F flag to serve custom recipes (and point that to yet another Python file). Instead, you can just import the recipe function or keep the recipe function in the same file as the call to prodigy.serve. The @prodigy.recipe decorator will take care of making it available.

prodigy.serve("ner.manual ner_news_headlines en_core_web_sm ./news_headlines.jsonl --label PERSON,ORG", port=9000)

prodigy.recipe decorator

Decorator that transforms a recipe function into a Prodigy recipe. The decorated function needs to return a dictionary of recipe components or a Controller. The decorator’s first argument is the recipe name, followed by a variable number of argument annotations, mapping to the arguments of the decorated function. This lets you execute the recipe with Prodigy.

from prodigy.components.stream import get_stream
@prodigy.recipe(
    "example",
    dataset=("Dataset to save annotations to", "positional", None, str),
    source=("Source data to load", "positional", None, str),
    view_id=("Annotation interface to use", "option", "v", str)
)
def example(dataset, source, view_id="text"):
    stream = get_stream(source)
    return {
        "dataset": dataset,
        "view_id": view_id,
        "stream": stream
    }

prodigy.get_recipe function

Get a recipe for a given name.

prodigy.set_recipe function

Register a recipe function with a name. Also adds aliases with - and _ swapped. When you use the @recipe decorator, the recipe will be set automatically.

prodigy.get_config function

Read and combine the user configuration from the available prodigy.json config files. Helpful in recipes to read off database settings, API keys or entirely custom config parameters.

config = prodigy.get_config()
theme = config["theme"]

prodigy.get_stream function

Get an iterable stream of tasks. This function is also used in recipes that allow streaming data from a source standard input. If a loader ID is set, Prodigy will look for a matching loader, and try to load the source. If the source is a file path or Path-like object, Prodigy will try to guess the loader from the file extension (defaults to "jsonl"). If the source is set to "-", Prodigy will read from standard input, letting you pipe data forward on the command line.

stream = prodigy.get_stream("/tmp/data.jsonl")
stream = prodigy.get_stream("/tmp/myfile.tmp", loader="txt")
stream = prodigy.get_stream("/tmp/data.json", input_key="text", skip_invalid=False)

If the source is an iterable stream itself – e.g. a generator or a list – get_stream will simply return the stream. This is useful if you’re calling an existing recipe function from Python – for example, in your custom recipe – and want to use an already loaded stream.

prodigy.get_loader function

Get a loader ID from an ID or guess a loader ID based on a file’s extension.

loader = prodigy.get_loader("jsonl")  # JSONL
loader = prodigy.get_loader(file_path="/tmp/data.json")  # JSON

prodigy.set_hashes function

Set hash IDs on a task based on the task properties. This is usually done by Prodigy automatically as the stream is processed by the controller. However, in some cases, you may want to take care of the hashing yourself, to implement custom filtering.

Input hashes are based on the input data, like the text, image or HTML. Task keys are based on the input hash, plus optional features you’re annotating, like the label or spans. This allows Prodigy to distinguish between two tasks that collect annotations on the same input, but for different features – for example, two different entities in the same text.

from prodigy import set_hashes

stream = (set_hashes(eg) for eg in stream)
stream = (set_hashes(eg, input_keys=("text", "custom_text")) for eg in stream)

prodigy.get_schema function

Get the JSON schema for a given view_id. This is the schema that Prodigy will validate against when you run a recipe. The JSON schemas describe the properties and types needed in order for an interface to render your task. The very first batch of the stream is validated before the server starts. After that, tasks in the stream are validated before they’re sent out to the web app. To disabled validation, set "validate": false in your Prodigy config.

Validation is powered by pydantic, so if you want to implement your own validation of Prodigy tasks and annotations, you can set json=False and receive the pydantic model.

schema = prodigy.get_schema("text")

Example output
{
  "title": "TextTask",
  "type": "object",
  "properties": {
    "meta": { "title": "Meta", "default": {}, "type": "object" },
    "_input_hash": { "title": " Input Hash", "type": "integer" },
    "_task_hash": { "title": " Task Hash", "type": "integer" },
    "_view_id": {
      "title": " View Id",
      "enum": ["text", "classification", "ner", "ner_manual", "pos", "pos_manual", "image", "image_manual", "html", "choice", "diff", "compare", "review", "text_input", "blocks"]
    },
    "_session_id": { "title": " Session Id", "type": "string" },
    "text": { "title": "Text", "type": "string" }
  },
  "required": ["_input_hash", "_task_hash", "text"]
}

prodigy.log function

Add an entry to Prodigy’s log. For more details, see the docs on debugging and logging.

prodigy.log("RECIPE: Starting recipe custom-recipe", locals())

Preprocessors

Preprocessors convert and modify a stream of examples and their properties, or pre-process documents before annotation. They’re available via prodigy.components.preprocess.

split_sentences function

Use spaCy’s sentence boundary detector to split example text into sentences, convert the existing "spans" and their start and end positions accordingly and yield one example per sentence. Setting a min_length will only split texts longer than a certain number of characters. This lets you use your own logic, while still preventing very long examples from throwing off the beam search algorithm and affecting performance. If no min_length is set, Prodigy will check the config for a 'split_sents_threshold' setting.

from prodigy.components.preprocess import split_sentences
import spacy

nlp = spacy.load("en_core_web_sm")
stream = [{"text": "spaCy is a library. It is written in Python."}]
stream = split_sentences(nlp, stream, min_length=30)

split_spans function

Split a stream with multiple spans per example so that there’s one span per task.

add_tokens function

Tokenize the incoming text and add a 'tokens' key to each example in the stream. If the example has spans, each span is updated with a "token_start" and "token_end" key. This pre-processor is mostly used in manual NER annotation to allow entity annotation based on token boundaries.

from prodigy.components.preprocess import add_tokens
import spacy

nlp = spacy.load("en_core_web_sm")
stream = [{"text": "Hello world"}, {"text": "Another text"}]
stream = add_tokens(nlp, stream, skip=True)

split_pages functionNew: 1.17

Split multi-page tasks annotated with the pages UI into separate tasks (one task per page). This is especially useful if you want to train a model on the annotations, and it’s also what the train and data-to-spacy recipes do under the hood. The split pages will preserve a "_page_num" and "_page_title", so you’ll always be able to relate them back to the original paginated annotations.

from prodigy.components.preprocess import split_pages
from prodigy.components.db import connect

db = connect()
examples = db.get_dataset_examples("my_dataset")
split_examples = split_pages(examples)

merge_pages functionNew: 1.17

Merge a stream of split paginated examples (e.g. created using the split_pages preprocessor) into paginated tasks to use with the pages. Examples need to be provided in consecutive order and require a "_page_num" and an optional "_page_title". Whenever a record starts at page 0 again, a new paginated example will be created.

from prodigy.components.preprocess import split_pages, merge_pages
from prodigy.components.db import connect

db = connect()
examples = db.get_dataset_examples("my_dataset")
split_examples = split_pages(examples)
merged_examples = merge_pages(split_examples)

fetch_media functionNew: 1.10

Replace all media paths and URLs in a stream with base64 data URIs. Can be used for converting streams of image or audio files. The skip keyword argument lets you specify whether to skip invalid files that can’t be converted (for example, because the path doesn’t exist, or the URL can’t be fetched). If set to False, Prodigy will raise a ValueError if it encounters invalid files.

from prodigy.components.preprocess import fetch_media

stream = [{"image": "/path/to/image.jpg"}, {"image": "https://example.com/image.jpg"}]
stream = fetch_media(stream, ["image"], skip=True)

make_ner_suggestions functionNew: 1.13.1

Adds the named entity annotations suggestions from an nlp model to the stream. Can be used with any spaCy nlp pipeline that has a component that can provide NER suggestions. This includes spacy-llm pipelines.

In general you’ll want to make sure that you apply add_tokens to the stream after adding the spans. This ensures that the token information is attached to the spans as well.

import spacy
from prodigy.components.stream import get_stream
from prodigy.preprocess import make_ner_suggestions, add_tokens

nlp = spacy.load("en_core_web_md")
stream = get_stream("/path/to/examples.jsonl")

# Apply predictions
stream.apply(
    make_ner_suggestions,
    stream=stream,
    nlp=nlp,
    component="ner",
    labels=["PERSON"],
)

# Make sure tokens are added everywhere, including in span annotations
stream.apply(add_tokens)

make_spancat_suggestions functionNew: 1.13.1

Adds the spancat annotations suggestions from an nlp model to the stream. Can be used with any spaCy nlp pipeline that has a component that can provide spancat suggestions. This includes spacy-llm pipelines.

In general you’ll want to make sure that you apply add_tokens to the stream after adding the spans. This ensures that the token information is attached to the spans as well.

Example
import spacy
from prodigy.components.stream import get_stream
from prodigy.preprocess import make_spancat_suggestions, add_tokens

nlp = spacy.load("model_with_spans")
stream = get_stream("/path/to/examples.jsonl")

# Apply predictions
stream.apply(
    make_spancat_suggestions,
    stream=stream,
    nlp=nlp,
    component="spancat",
    labels=["LABEL_1", "LABEL_2"],
)

# Make sure tokens are added everywhere, including in span annotations
stream.apply(add_tokens)

make_textcat_suggestions functionNew: 1.13.1

Adds the textcat predictions from an nlp model to the stream. Can be used with any spaCy nlp pipeline that has a component that can provide NER suggestions. This includes spacy-llm pipelines.

Example
import spacy
from prodigy.components.stream import get_stream
from prodigy.preprocess import make_textcat_suggestions

nlp = spacy.load("model_with_textcat")
stream = get_stream("/path/to/examples.jsonl")

# Apply predictions
stream.apply(
    make_textcat_suggestions,
    stream=stream,
    nlp=nlp,
    component="textcat",
    threshold=0.65
    labels=["LABEL_1", "LABEL_2"],
)

# Make sure tokens are added everywhere, including in span annotations
stream.apply(add_tokens)

Sorters

Sorters are helper functions that wrap a stream of (score, example) tuples, (usually returned by a model), resort it and yield examples in the new order. They’re available via prodigy.components.sorters.

All sorters follow the same API and take two arguments:

The following sorters are available and can be imported from prodigy.components.sorters:

The prefer_uncertain function also supports an additional keyword argument algorithm, that lets you specify either "probability" or "ema" (exponential moving average).

from prodigy.components.sorters import prefer_uncertain

def score_stream(stream):
    for example in stream:
        score = model.predict(example["text"])
        yield (score, example)

stream = prefer_uncertain(score_stream(stream))

Filters

Filters are helper functions used across recipes that wrap a stream and filter it based on certain conditions. They’re available via prodigy.components.filters.

filter_empty function

Remove examples with a missing, empty or otherwise falsy value from a stream. This filter can also be enabled by specifying an input_key argument on the get_stream helper function.

from prodigy.components.filters import filter_empty
stream = [{"text": "test"}, {"image": "test.jpg"}, {"text": ""}]
stream = filter_empty(stream, key="text")
# [{'text': 'test'}]

filter_duplicates function

Filter duplicate examples from a stream. You can choose to filter by task, which includes the input data as well as the added spans, labels etc., or by input data only. This filter can also enabled by setting dedup=True on the get_stream helper function.

from prodigy.components.filters import filter_duplicates
from prodigy import set_hashes

stream = [{"text": "foo", "label": "bar"}, {"text": "foo", "label": "bar"}, {"text": "foo"}]
stream = [set_hashes(eg) for eg in stream]
stream = filter_duplicates(stream, by_input=False, by_task=True)
# [{'text': 'foo', 'label': 'bar', '_input_hash': ..., '_task_hash': ...}, {'text': 'foo', '_input_hash': ..., '_task_hash': ...}]
stream = filter_duplicates(stream, by_input=True, by_task=True)
# [{'text': 'foo', 'label': 'bar', '_input_hash': ..., '_task_hash': ...}]

filter_inputs function

Filter out tasks based on a list of input hashes, referring to the input data. Useful for filtering out already annotated tasks. To get the task hashes of one or more datasets, you can use db.get_input_hashes(*dataset_names).

from prodigy.components.filters import filter_inputs
stream = [{"_input_hash": 5, "text": "foo"}, {"_input_hash": 9, "text": "bar"}]
stream = filter_inputs(stream, [1, 2, 3, 4, 5])
# [{'_input_hash': 9, 'text': 'bar'}]

filter_tasks function

Filter out tasks based on a list of task hashes, referring to the input data plus the added spans, label etc. Useful for filtering out already annotated tasks and used by Prodigy’s built-in --exclude logic. To get the task hashes of one or more datasets, you can use db.get_task_hashes(*dataset_names).

from prodigy.components.filters import filter_tasks
stream = [{"_task_hash": 5, "text": "foo"}, {"_task_hash": 9, "text": "bar"}]
stream = filter_tasks(stream, [1, 2, 3, 4, 5])
# [{'_task_hash': 9, 'text': 'bar'}]

filter_seen_before functionNew: 1.13

Filter out tasks that you may have seen before.

from prodigy.components.filters import filter_tasks
stream_a = [{"_task_hash": 5, "text": "foo"}, {"_task_hash": 9, "text": "bar"}]
stream_b = [{"_task_hash": 5, "text": "foo"}, {"_task_hash": 10, "text": "buz"}]
stream = filter_seen_before(stream_a, stream_b, exclude_by="task")
# [{'_task_hash': 9, 'text': 'bar'}]

Task routers

Task routers are Python functions that determine which annotator will annotate each example in a stream. You can read the full guide to learn more about how they work. Built-in routers (no_overlap, full_overlap and route_average_per_task) are available via config settings or programatically via routers module. Importantly, the task_router component also accepts custom routers. Check the task routing guide for example implementations.

no_overlap functionNew: 1.12

Route each example to a single annotator either by setting "feed_overlap": false (default) in the Prodigy config file or passing the router function as the task_router component like so:

recipe.py (except)
import prodigy
from prodigy.components.routers import no_overlap

@prodigy.recipe(
    "my-custom-recipe",
    dataset=("Dataset to save answers to", "positional", None, str),
    view_id=("Annotation interface", "option", "v", str)
)
def my_custom_recipe(dataset, view_id="text"):
    # Load your own streams from anywhere you want
    stream = load_my_custom_stream()

    return {
        "dataset": dataset,
        "view_id": view_id,
        "stream": stream,
        "task_router": no_overlap
    }

full_overlap functionNew: 1.12

Route each example to each annotator either by setting "feed_overlap": true in the Prodigy config file or passing the router function as the task_router component like so:

recipe.py (except)
import prodigy
from prodigy.components.routers import full_overlap

@prodigy.recipe(
    "my-custom-recipe",
    dataset=("Dataset to save answers to", "positional", None, str),
    view_id=("Annotation interface", "option", "v", str)
)
def my_custom_recipe(dataset, view_id="text"):
    # Load your own streams from anywhere you want
    stream = load_my_custom_stream()

    return {
        "dataset": dataset,
        "view_id": view_id,
        "stream": stream,
        "task_router": full_overlap
    }

route_average_per_task functionNew: 1.12

Assign partial annotator overlap by specifying how many annotators need to see each example on average. In line with the other built-in routers, this can be set in the config file via annotations_per_task setting or programmatically:

recipe.py (except)
import prodigy
from prodigy.components.routers import route_average_per_task

@prodigy.recipe(
    "my-custom-recipe",
    dataset=("Dataset to save answers to", "positional", None, str),
    view_id=("Annotation interface", "option", "v", str)
)
def my_custom_recipe(dataset, view_id="text"):
    # Load your own streams from anywhere you want
    stream = load_my_custom_stream()

    return {
        "dataset": dataset,
        "view_id": view_id,
        "stream": stream,
        "task_router": route_average_per_task(2.5)
    }

PatternMatcher

The PatternMatcher wraps spaCy’s Matcher and PhraseMatcher and will match both token-based and exact string match pattern on a stream of incoming examples. It’s typically used in recipes like ner.teach or textcat.teach to use match pattern files to add suggestions. The PatternMatcher can be updated with annotations and will score the individual patterns. Patterns that are accepted more often will be scored higher than patterns that are mostly rejected. Combined with a sorter, this lets you focus on the most uncertain or the highest/lowest scoring patterns. The pattern matcher is available via prodigy.models.matcher.

PatternMatcher.__init__ method

Initialize a pattern matcher.

from prodigy.models.matcher import PatternMatcher
import spacy

nlp = spacy.blank("en")
matcher = PatternMatcher(nlp)

score = (n_accept + prior_correct) / ((n_reject + prior_incorrect) / (n_accept + prior_correct))

PatternMatcher.__call__ method

Process a stream of examples and add pattern matches. Will add an entry to the example dict’s "spans" for each match. Each span includes a "pattern" key, mapped to the ID of the pattern that was used to produce this match. This is also later used in PatternMatcher.update.

PatternMatcher.from_disk method

Load patterns from a patterns file.

matcher = PatternMatcher(nlp).from_disk("./patterns.jsonl")

PatternMatcher.add_patterns method

Add patterns to the pattern matcher.

patterns = [
    {"label": "FRUIT", "pattern": [{"lower": "goji"}, {"lower": "berry"}]}
    {"label": "VEGETABLE", "pattern": "Lamb's lettuce"}
]
matcher.add_patterns(patterns)

PatternMatcher.update method

Update the pattern matcher from annotation and update its scores. Typically called as part of a recipe’s update callback and with answers received from the web app. Expects the examples to have an "answer" key ("accept", "reject" or "ignore") and will use all "spans" that have a "pattern" key, which is the ID of the pattern assigned by PatterMatcher.__call__.

answers = [
    {
      "text": "Hello Apple",
      "spans": [{"start": 0, "end": 11, "label": "ORG", "pattern": 0}],
      "answer": "reject"
    }
]
matcher.update(answers)

Utility functions

combine_models function

Combine two models and return a predict and update function. Predictions of both models are combined using the toolz.interleave function. Requires both model objects to have a __call__ and an update() method. This helper function is mostly used to combine annotation models with a PatternMatcher to mix pattern matches and model suggestions. For an example, see the docs on custom text classification models.

from prodigy.util import combine_models
from prodigy.models.matcher import PatternMatcher

class CustomModel:
    def __call__(self, stream):
        yield from predict_something(stream)

    def update(self, answers):
        update_something(answers)

predict, update = combine_models(CustomModel, PatternMatcher)

b64_to_bytes function

Convert a base64-encoded data URI to bytes. Can be used to convert the inlined base64 "image", "audio" or "video" values annotation tasks to byte strings that can be consumed by models. See the docs on integrating image models for examples.

from prodigy.util import b64_to_bytes
image = "data:image/jpeg;base64,/9j/4AAQSkZJRgABAQEASABIA..."  # and so on
image_bytes = b64_to_bytes(image)

file_to_b64 function

Convert a file to a base64-encoded data URI. Used under the hood by the media loaders.

from prodigy.util import file_to_b64

b64_str = file_to_b64("/path/to/image.jpg")

bytes_to_b64 function

Convert a bytestring to a base64-encoded data URI. Used under the hood by the media loaders.

from prodigy.util import bytes_to_b64

image_bytes = open("/path/to/image.jpg", "r").read()
b64_str = bytes_to_b64(image_bytes, "image/jpeg")

split_string function

Split a comma-separated string and strip whitespace. A very simple utility that’s mostly used as a converter function in the recipe argument annotations to convert labels passed in from the command line.

from prodigy.util import split_string
assert split_string("PERSON,ORG,PRODUCT") == ["PERSON", "ORG", "PRODUCT"]

get_labels function

Utility function used in recipe argument annotations to handle command-line arguments that can either take a comma-separated list of labels or a file with one label per line. If the string is a valid file path, the file contents are read in line by line. Otherwise, the string is split on commas.

from prodigy.util import get_labels
assert get_labels("PERSON,ORG,PRODUCT") == ["PERSON", "ORG", "PRODUCT"]
assert get_labels("./labels.txt") == ["SOME", "LABELS", "FROM", "FILE"]

Controller

The controller takes care of putting the individual recipe components together and exposes methods that allow the application to interact with the REST API. This is usually done when you use the @recipe decorator on a function that returns a dictionary of components. However, you can also choose to initialize a Controller yourself and make your recipe return it.

Controller.from_components classmethodNew: 1.11.8

Initialize a controller from the dictionary of components returned by a recipe function.

from prodigy.core import Controller

components = your_recipe("dataset", some_argument="some value")
controller = Controller.from_components("your_recipe", components)

Controller.__init__ method

Initialize the controller.

from prodigy.core import Controller
controller = Controller(dataset, view_id, stream, update, db,
                        progress, on_load, on_exit, before_db,
                        validate_answer, get_session_id, exclude,
                        config, None)

All arguments of the controller are also accessible as attributes, for example controller.dataset. In addition, the controller exposes the following attributes:

Controller.get_questions method

Get a batch of annotation tasks from the queue.

next_batch = controller.get_questions()

Controller.receive_answers method

Receive a batch of annotated tasks. If available, stores the tasks in the database and calls the update function.

tasks = [
    {"_input_hash": 0, "_task_hash": 0, "text": "x", "answer": "accept"},
    {"_input_hash": 1, "_task_hash": 1, "text": "y", "answer": "reject"}
]
controller.receive_answers(tasks)

Controller.save method

Saves the current project and progress when the user exits the application. If available, calls the store’s save method and the on_exit function.

Controller.get_total_by_session methodNew: 1.10

Get the total number of examples available in the dataset for a given named session.

Controller.progress property

Get the current progress. If available, calls the progress function. Otherwise, it checks whether the stream has a length and returns the quotient of the session annotations and stream length. Otherwise, it returns None. A progress of None is visualized with an infinity symbol in the web application.

Controller.get_progress New: 1.12

Get the current progress for a specific session_id. If available, calls the progress function provided to the Controller. If not, the Controller has a few default Progress estimators it will choose from based on other Controller settings.

1. If an update callback is provided, a progress estimator will be used that has been in Prodigy since the beginning for active learning workflows. It assumes the return value of update represents a loss value and will use gradient descent to estimate when model in the loop has enough annotations (visualised as PROGRESS in the web app).

1. If no update callback is provided but the total_examples_target config variable is set, the progress is the quotient of the number of saved examples for this session and the value of total_examples_target (visualised as TARGET PROGRESS in the web app).

2. If no update callback is provided and the total_examples_target config variable is not set, progress is estimated (and in some cases calculated) as the quotient of the Source position and its size (visualised as SOURCE PROGRESSin the web app).

For the best annotator experience, we recommend setting a total_examples_target value. The progress bar will be the most accurate when this is set.

You can also always provide a custom progress function via the "progress" component to the dictionary of Recipe components returned by your custom recipe.

def progress(ctrl, update_return_value):
    return ctrl.session_annotated / 2500

@prodigy.recipe(
    "example",
    dataset=("Dataset to save annotations to", "positional", None, str),
    source=("Source data to load", "positional", None, str),
    view_id=("Annotation interface to use", "option", "v", str)
)
def example(dataset, source, view_id="text"):
    stream = get_stream(source)
    return {
        "dataset": dataset,
        "view_id": view_id,
        "stream": stream,
        "progress": progress
    }

If the Source has a known length, (e.g. a list of tasks or loading from a Parquet file or a JSONL file) progress can be calculated exactly. When using a file based source (e.g. loading a CSV file) progress is estimated by calculating the size in bytes, reading the file in chunks and using the quotient of byte offset and total bytes.

If none of these conditions are met, and the stream provided to the Controller is a simple generator, this method returns None. A progress of None is visualized with an infinity symbol in the web application.

Controller.get_session_name New: 1.12

Get the appropriate session name for an annotator name. This method can be very useful when writing your own custom task routers when you want to make sure that you’re able to map a name to the appropriate session id.

Controller.confirm_session New: 1.12

Ensure that a session ID exists. This is useful in custom task routers when you want to add a task to a queue that might not have started annotating yet. More details on how this mechanic works can be found on the task routing documentation page. There’s a also a detailed example that shows a use-case for such a function here.

Controller.add_session New: 1.12

Add session to the controller’s internal list of available sessions. This is useful when custom session_factory is used.

Controller.reset_stream New: 1.14.11

Reset the stream such that all sessions will get different examples to annotate. Meant to be used in custom event callbacks.

Stream New: 1.12

Loads and transform items from a raw data source, and enqueues them to one or more queues. Queues can be added during iteration and can be populated by any number of items from the history. As of v1.12 the Stream type can be used instead of the Generator type for the stream of annotations tasks. The Generator based stream will be deprecated in Prodigy v2.0. Note that this function is different from the old prodigy.get_stream function, which would return a Generator.

stream.get_stream functionNew: 1.12

Get an iterable stream of tasks of Stream type. Under the hood it resolves the input type so that in can be handled with the appropriate loader, unless a custom loader is provided. If the source is set to "-", Prodigy will read from standard input, letting you pipe data forward on the command line.

from prodigy.components.stream import get_stream

stream = get_stream("/tmp/data.jsonl")
stream = get_stream("/tmp/myfile.tmp", loader="txt")
stream = get_stream("/tmp/data.json", input_key="text", skip_invalid=False)

stream.apply methodNew: 1.12

Transforms the stream of tasks by applying a function that returns a generator of tasks. Can be used for any preprocessing of the annotation tasks such as adding tokens, splitting sentences or defining tasks data structures from scratch.

If data in the pages format is provided, Stream.apply will automatically apply the function to each page instead of the top-level task.

def my_wrapper(nlp, stream):
    for eg in stream:
        eg["new_key"] = new_value
        yield eg

stream.apply(my_wrapper, nlp, stream)

You can also import and use the special value STREAM_PLACEHOLDER to indicate which argument is the stream:

from prodigy.stream import STREAM_PLACEHOLDER

def my_wrapper(arg1, arg2):
    for eg in arg2:
        eg["new_key"] = new_value
        yield eg

stream.apply(my_wrapper, nlp, STREAM_PLACEHOLDER)

The same works for keyword arguments:

def my_wrapper(arg1, arg2, *, arbitrary_name):
    for eg in arbitrary_name:
        eg["new_key"] = new_value
        yield eg

stream.apply(my_wrapper, arg1, arg2, arbitrary_name=stream)

If the stream instance is not in the args or kwargs, it is assumed to be the first argument to the wrapper.

Session New: 1.12

The session represents the Prodigy annotator session with a queue of tasks to pull from and the temporary buffer of pulled but unanswered tasks. It also exposes a number of attributes relevant to history tracking and progress estimation. A new session is initialized every time a user accesses the Prodigy server with a URL. As of v1.12 this initialization can be customized by providing session_factory callback to the Controller.

Session.__init__ methodNew: 1.12

Initialize the session object.

from prodigy.components import Session
session = Session(session_id, stream, batch_size, answered_input_hashes, answered_task_hashes,
                  total_annotated, target_annotated)

All arguments of the session are also accessible as attributes, for example session.total_annotated. In addition, the session exposes the following attributes: