Skip to content

Push *TaskFrame downstream #893

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
markbackman merged 1 commit into from
Jun 11, 2026
Merged

Push *TaskFrame downstream #893

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 5 additions & 6 deletions pipecat/learn/pipeline-termination.mdx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,7 @@ from pipecat.frames.frames import EndFrame, TTSSpeakFrame
await task.queue_frame(EndFrame())
```

Push an `EndTaskFrame` upstream from inside your pipeline:
Push an `EndTaskFrame` downstream from inside your pipeline:

```python
# From inside a function call
Expand All @@ -65,7 +65,7 @@ async def end_conversation(params: FunctionCallParams):
await params.llm.push_frame(TTSSpeakFrame("Have a nice day!"))

# Signal that the task should end after processing this frame
await params.llm.push_frame(EndTaskFrame(), FrameDirection.UPSTREAM)
await params.llm.push_frame(EndTaskFrame(), FrameDirection.DOWNSTREAM)
```

**How graceful termination works:**
Expand Down Expand Up @@ -186,7 +186,7 @@ async def check_conversation_complete(params: FunctionCallParams):

if conversation_complete:
await params.llm.push_frame(TTSSpeakFrame("Thank you for using our service!"))
await params.llm.push_frame(EndTaskFrame(), FrameDirection.UPSTREAM)
await params.llm.push_frame(EndTaskFrame(), FrameDirection.DOWNSTREAM)

await params.result_callback({"status": "complete" if conversation_complete else "continuing"})
```
Expand Down Expand Up @@ -232,9 +232,8 @@ async def process_frame(self, frame: Frame, direction: FrameDirection):
**Solution:** Use the appropriate frame type and direction:

```python
# From inside the pipeline, push upstream to reach the source
await self.push_frame(EndTaskFrame(), FrameDirection.UPSTREAM)
await self.push_frame(CancelTaskFrame(), FrameDirection.UPSTREAM)
await self.push_frame(EndTaskFrame(), FrameDirection.DOWNSTREAM)
await self.push_frame(CancelTaskFrame(), FrameDirection.DOWNSTREAM)

# The pipeline source will then convert these to proper termination frames
# and push them downstream through the entire pipeline
Expand Down
Loading