Implementation:CrewAIInc CrewAI Crew Replay Method

Overview

Concrete method for replaying crew execution from a specific task using stored prior outputs provided by the CrewAI framework.

Source

src/crewai/crew.py:L1577-1615

Signature

def replay(
    self,
    task_id: str,
    inputs: dict[str, Any] | None = None,
) -> CrewOutput

Parameters

Parameter	Type	Required	Description
task_id	str	Yes	The unique identifier of the task from which to begin replay
inputs	None	No	Optional dictionary of input variables for task interpolation

I/O

• Input: task_id (unique identifier of the task to replay from) and optional inputs (variable dictionary for task templates)
• Output: CrewOutput containing the results of the replayed execution from the specified task through the end of the pipeline

Internal Behavior

The replay method performs the following steps:

1. Loads stored outputs from TaskOutputStorageHandler — Retrieves persisted TaskOutput objects from previous crew executions. These outputs were automatically saved during the original kickoff() execution.
2. Finds task index by task_id — Searches through self.tasks to locate the task matching the provided task_id. If the task ID is not found, an error is raised.
3. Restores TaskOutput for all preceding tasks — For each task before the replay start point, the stored output is loaded and assigned to the task's output attribute. This reconstructs the execution context as it existed when the specified task originally ran.
4. Executes from specified task through end — The crew execution engine resumes from the specified task, running it and all subsequent tasks in sequence. Each task receives the restored upstream context as if the entire pipeline had run from the beginning.

Import

from crewai import Crew

Example

from crewai import Crew, Agent, Task, Process

# Define agents and tasks
researcher = Agent(
    role="Senior Research Analyst",
    goal="Uncover cutting-edge developments in AI",
    backstory="You are an expert research analyst.",
)

writer = Agent(
    role="Content Writer",
    goal="Craft compelling content about AI",
    backstory="You are a skilled writer.",
)

research_task = Task(
    description="Conduct thorough research about AI.",
    expected_output="A detailed research report.",
    agent=researcher,
)

writing_task = Task(
    description="Write a comprehensive article about AI.",
    expected_output="A well-structured article.",
    agent=writer,
)

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential,
)

# First, run the full crew to generate stored outputs
result = crew.kickoff()

# Later, replay from the writing task using its task_id
# This skips re-running the research task and uses its stored output
replay_result = crew.replay(task_id="abc-123")

# The replay_result contains the output from the writing task onward,
# using the stored research output as context
print(replay_result)

Obtaining the Task ID

To obtain the task_id for replay, developers can:

• Inspect task objects — Each Task instance has an id attribute (a UUID string) that is assigned at creation time
• Check execution logs — When verbose=True, the crew logs task IDs during execution
• Use the CLI — CrewAI provides CLI commands to list tasks from previous executions

# Accessing task IDs programmatically
for task in crew.tasks:
    print(f"Task: {task.description}, ID: {task.id}")

Key Implementation Details

• TaskOutputStorageHandler persists task outputs automatically during kickoff() execution. The replay method depends on these stored outputs being available from a prior run.
• The method raises an exception if the provided task_id does not match any task in the crew's task list.
• Replay executes from the specified task through the end of the pipeline. It is not possible to replay only a single task in isolation; all downstream tasks will also execute.
• If inputs are provided, they are used for task template interpolation during the replayed execution, potentially overriding the original input values.

Principle

Principle:CrewAIInc_CrewAI_Task_Replay

References

• crewAI GitHub Repository