Flexible IO proposal #1161
Flexible IO proposal #1161
Conversation
|
proposal.md
Outdated
| class PipelineIO(ABC): | ||
|
|
||
| def before_stt_node(self, source: AsyncIterator[rtc.AudioFrame]) -> AsyncIterator[rtc.AudioFrame]: | ||
| return source | ||
|
|
||
| def after_stt_node(self, source: AsyncIterator[SpeechEvent]) -> AsyncIterator[SpeechEvent]: | ||
| return source | ||
|
|
||
| def before_llm_node(self, chat_ctx: ChatContext) -> AsyncIterator[ChatChunk] | None: | ||
| return None | ||
|
|
||
| def after_llm_node(self, source: AsyncIterator[ChatChunk]) -> AsyncIterator[ChatChunk]: | ||
| return source | ||
|
|
||
| def before_tts_node(self, source: AsyncIterator[str]) -> AsyncIterator[rtc.AudioFrame] | None: | ||
| return source | ||
|
|
||
| def after_tts_node(self, source: AsyncIterator[rtc.AudioFrame]) -> AsyncIterator[rtc.AudioFrame]: | ||
| return source |
There was a problem hiding this comment.
Not my wheelhouse, but is this in any way more convenient or idiomatic than if each pipeline stage managed its own pre/post transform callbacks? E.G.:
def passthrough_audio(source: AsyncIterator[rtc.AudioFrame]) -> AsyncIterator[rtc.AudioFrame]:
return source
def filter_swearwords(source: AsyncIterator[SpeechEvent]) -> AsyncIterator[SpeechEvent]:
return source
agent = PipelineAgent(
stt=STT(pre=passthrough_audio, post=filter_swearwords)
...
)There was a problem hiding this comment.
The scope is different, ideally, we can add new parameters like speech_id inside each step like before_tts_node, ...
| class TextOutput(Protocol): | ||
| async def write(self, text: str) -> None: ... | ||
|
|
||
| def flush(self) -> None: ... |
There was a problem hiding this comment.
how should frontend applications reason about "turns" with these two output types? is that what "flush" means? UI will likely want to render each complete "message" in a chat bubble, for instance. maybe having a unique id somewhere could help?
| STT -> LLM -> TTS | ||
|
|
||
| ```python | ||
| AudioInput = AsyncIterator[rtc.AudioFrame | rtc.AudioFrameEvent] |
There was a problem hiding this comment.
not sure the scope of what you're working on but where would text input/image/file "chat" input fit in?
| def clear_queue(self) -> None: ... | ||
|
|
||
|
|
||
| class TextOutput(Protocol): |
There was a problem hiding this comment.
In addition to text and audio output that are essentially "talking" or "chat", many applications "output" other structured things either by returning images or through function calls (i.e. JSON output). do we have any thoughts about whether it would make sense to provide an affordance for that in pipelineagent?
| return source | ||
|
|
||
| def before_tts_node(self, source: AsyncIterator[str] | str) -> AsyncIterator[rtc.AudioFrame] | None: | ||
| return source |
There was a problem hiding this comment.
shouldn't this method return AsyncIterator[str] | str?
|
|
||
| class PipelineIO(ABC): | ||
|
|
||
| def before_stt_node(self, source: AsyncIterator[rtc.AudioFrame]) -> AsyncIterator[rtc.AudioFrame]: |
There was a problem hiding this comment.
shouldn't these methods all be async?
| return source | ||
|
|
||
| def before_llm_node(self, chat_ctx: ChatContext) -> AsyncIterator[ChatChunk] | None: | ||
| return None |
There was a problem hiding this comment.
why is this one different than the others? it feels odd that it doesn't have the same return type as input type, and the default implementation returns None which implies its actually very semantically different than the other methods which are all open-ended hooks to transform data in the pipeline or add logging or other side effects.
| def flush(self) -> None: ... | ||
|
|
||
|
|
||
| class PipelineIO(ABC): |
There was a problem hiding this comment.
this name feels a little odd, given that in addition to PipelineIO we also have PipelineOutput (and maybe some forthcoming Input protocol too), but the "IO" in PipelineIO is not related to the pipeline's input nor output itself... might be better as PipelineHooks
No description provided.