openedx_ai_extensions.workflows.orchestrators package

openedx_ai_extensions.workflows.orchestrators package#

Submodules#

openedx_ai_extensions.workflows.orchestrators.base_orchestrator module#

Base orchestrator class for AI workflow execution.

class openedx_ai_extensions.workflows.orchestrators.base_orchestrator.BaseOrchestrator(workflow, user, context)#

Bases: object

Base class for workflow orchestrators.

classmethod get_orchestrator(*, workflow, user, context)#

Resolve and instantiate an orchestrator for the given workflow.

This factory method centralizes orchestrator lookup and validation. It ensures that the resolved class exists and is a subclass of BaseOrchestrator, providing a single, consistent entry point for orchestrator creation across the codebase.

Parameters:

  • workflow – AIWorkflowScope instance that defines the workflow configuration.

  • user – User for whom the workflow is being executed.

  • context – Dictionary with runtime context (e.g. course_id, location_id).

Returns:

An instantiated orchestrator for the given workflow.

Return type:

BaseOrchestrator

Raises:

  • AttributeError – If the configured orchestrator class cannot be found.

  • TypeError – If the resolved class is not a subclass of BaseOrchestrator.

run(input_data)#

openedx_ai_extensions.workflows.orchestrators.direct_orchestrator module#

Orchestrators for handling different AI workflow patterns in Open edX.

class openedx_ai_extensions.workflows.orchestrators.direct_orchestrator.DirectLLMResponse(workflow, user, context)#

Bases: BaseOrchestrator

Orchestrator for direct LLM responses. Does a single call to an LLM and gives a response.

run(input_data)#

Executes the content fetching, LLM processing, and handles streaming or structured response return.

class openedx_ai_extensions.workflows.orchestrators.direct_orchestrator.EducatorAssistantOrchestrator(workflow, user, context)#

Bases: SessionBasedOrchestrator

Orchestrator for educator assistant workflows.

Generates quiz questions and optionally stores them in content libraries.

Two modes: - Direct mode (library_id in input_data): generate + commit immediately (legacy). - Iterative mode (no library_id): generate → store in session → review → save separately.

get_current_session_response(_)#

Retrieve the current session state.

  • If a collection was already saved: return the collection URL.

  • If questions were generated but not yet saved: return them for review.

  • Otherwise: return None.

regenerate_question(input_data)#

Refine an existing question at the given slot index.

Uses a dedicated refinement prompt so the LLM improves the provided question rather than generating an unrelated new one.

run(input_data)#

Generate quiz questions.

If library_id is present in input_data, immediately commit to library (legacy path). Otherwise store questions in session metadata for iterative review.

save(input_data)#

Commit selected questions to a content library.

Expects input_data with library_id, questions list, and optional publish flag.

openedx_ai_extensions.workflows.orchestrators.flashcards_orchestrator module#

Orchestrators for handling different AI workflow patterns in Open edX.

class openedx_ai_extensions.workflows.orchestrators.flashcards_orchestrator.FlashCardsOrchestrator(workflow, user, context)#

Bases: ScopedSessionOrchestrator

Orchestrator for flashcards generation using LLM.

Does a single call to an LLM and gives a response.

get_current_session_response(_)#

Retrieve the current session state.

  • If flashcards were generated but not yet saved: return them for review.

  • Otherwise: return None.

run(input_data)#

Executes the content fetching, LLM processing, and handles streaming or structured response return.

save(input_data)#

Saves the generated flashcards to the database or a file. This is a placeholder implementation and should be replaced with actual saving logic.

openedx_ai_extensions.workflows.orchestrators.mock_orchestrator module#

Mock orchestrator for testing and development.

class openedx_ai_extensions.workflows.orchestrators.mock_orchestrator.MockResponse(workflow, user, context)#

Bases: BaseOrchestrator

Complete mock orchestrator. Responds inmediately with a mock answer. Useful for UI testing.

run(input_data)#
class openedx_ai_extensions.workflows.orchestrators.mock_orchestrator.MockStreamResponse(workflow, user, context)#

Bases: BaseOrchestrator

Complete mock orchestrator with streaming. Responds inmediately with a mock answer in a streaming fashion. Useful for UI testing.

run(input_data)#

openedx_ai_extensions.workflows.orchestrators.session_based_orchestrator module#

Session-based orchestrator.

class openedx_ai_extensions.workflows.orchestrators.session_based_orchestrator.ScopedSessionOrchestrator(workflow, user, context)#

Bases: SessionBasedOrchestrator

Orchestrator that follows the scope’s location specificity for sessions.

Intentionally skips SessionBasedOrchestrator.__init__ to avoid creating a location-specific session; instead creates a course-scoped session shared across locations.

run_async(input_data)#

Launch async task for scoped sessions.

Unlike the parent implementation, this does not write location_id to the session row (which has no location_id in its unique-together lookup). Instead the current location is persisted in metadata['location_id'] so the Celery task can recover it without risking an integrity-error collision with any pre-existing location-scoped session.

class openedx_ai_extensions.workflows.orchestrators.session_based_orchestrator.SessionBasedOrchestrator(workflow, user, context)#

Bases: BaseOrchestrator

Orchestrator that provides session-based LLM responses.

clear_session(_)#
get_run_status(input_data)#

Get the status of an async task from session metadata.

Returns:

Status information including task result if completed

Return type:

dict

run(input_data)#
run_async(input_data)#

Launch async task to execute the run method.

Parameters:

input_data – Input data to pass to the run method

openedx_ai_extensions.workflows.orchestrators.threaded_orchestrator module#

Orchestrators Base classes to hold the logic of execution in ai workflows

class openedx_ai_extensions.workflows.orchestrators.threaded_orchestrator.ThreadedLLMResponse(workflow, user, context)#

Bases: SessionBasedOrchestrator

Threaded orchestrator for conversational workflows.

lazy_load_chat_history(input_data)#

Load older messages for infinite scroll. Expects input_data to contain current_messages (count) from frontend. Returns only new messages not already loaded, limited by max_context_messages.

run(input_data)#

Module contents#

Orchestrators package for AI workflow execution.

This package provides the base orchestrator and concrete implementations.

class openedx_ai_extensions.workflows.orchestrators.BaseOrchestrator(workflow, user, context)#

Bases: object

Base class for workflow orchestrators.

classmethod get_orchestrator(*, workflow, user, context)#

Resolve and instantiate an orchestrator for the given workflow.

This factory method centralizes orchestrator lookup and validation. It ensures that the resolved class exists and is a subclass of BaseOrchestrator, providing a single, consistent entry point for orchestrator creation across the codebase.

Parameters:

  • workflow – AIWorkflowScope instance that defines the workflow configuration.

  • user – User for whom the workflow is being executed.

  • context – Dictionary with runtime context (e.g. course_id, location_id).

Returns:

An instantiated orchestrator for the given workflow.

Return type:

BaseOrchestrator

Raises:

  • AttributeError – If the configured orchestrator class cannot be found.

  • TypeError – If the resolved class is not a subclass of BaseOrchestrator.

run(input_data)#
Contents