This PR focuses on restructuring the pipeline.py and subgraph execution such that async and non-async functions are separated. For each of these, a Generator or AsyncGenerator can now also be returned, if the pipeline has set streaming in its inference_state. With this restructuring, we can now enable streaming.
Important Changes
Subgraph execution is now moved to a new helper class called SubGraphExecutor. This will execute a list of subgraphs depends on if the user is using the normal sync pathway, async pathway, sync pathway with streaming or async pathway with streaming. This should make execution cleaner and hopefully easier to follow.
To identify if the user is returning an output that should be yielded, there is a new StreamingOutput pydantic model which should be used by all pipelines that have streaming enabled. This model includes a field for the output that should be yielded to the user, and the output that should be returned to the operator to execute the next step in the pipeline. This was adapted from @dbogunowicz's previous PR
To simplify streaming vs non-streaming routes, pipelines can provide a generator_router which will be used instead when in streaming mode. This is simply to make it easier to determine the next step between the two pathways and not complicate the can_operate() functions, especially as the normal, non-streaming pathway is what we expect for most pipelines majority of the time. This is available in the TextGenerationPipeline
Other changes
Streaming tests are now enabled
When an AsyncGenerator is returned by the Pipeline, a fastapi StreamingResponse is now being returned. This is also what is done by OpenAI. Doesn't seem like the server had a good way to handle streaming/generator outputs before
Testing:
The two previously disabled streaming tests are now enabled
Summary
This PR focuses on restructuring the
pipeline.py
and subgraph execution such thatasync
and non-async functions are separated. For each of these, aGenerator
orAsyncGenerator
can now also be returned, if the pipeline has setstreaming
in itsinference_state
. With this restructuring, we can now enable streaming.Important Changes
SubGraphExecutor
. This will execute a list of subgraphs depends on if the user is using the normal sync pathway, async pathway, sync pathway with streaming or async pathway with streaming. This should make execution cleaner and hopefully easier to follow.StreamingOutput
pydantic model which should be used by all pipelines that have streaming enabled. This model includes a field for the output that should be yielded to the user, and the output that should be returned to the operator to execute the next step in the pipeline. This was adapted from @dbogunowicz's previous PRgenerator_router
which will be used instead when in streaming mode. This is simply to make it easier to determine the next step between the two pathways and not complicate thecan_operate()
functions, especially as the normal, non-streaming pathway is what we expect for most pipelines majority of the time. This is available in theTextGenerationPipeline
Other changes
AsyncGenerator
is returned by the Pipeline, a fastapiStreamingResponse
is now being returned. This is also what is done by OpenAI. Doesn't seem like the server had a good way to handle streaming/generator outputs beforeTesting:
Server Testing:
sample_config.yaml
:We can get streaming responses through the following request:
Outside of the server: