openedx_ai_extensions.workflows package

Contents

openedx_ai_extensions.workflows package#

Subpackages#

Submodules#

openedx_ai_extensions.workflows.models module#

AI Workflow models for managing flexible AI workflow execution

class openedx_ai_extensions.workflows.models.AIWorkflowProfile(*args, **kwargs)#

Bases: Model

Workflow profile combining a disk-based template with database overrides.

Templates are read-only JSON files on disk (versioned, immutable). Profiles point to a template and store JSON patch overrides in the DB. Effective config = merge(base_template, content_patch)

exception DoesNotExist#

Bases: ObjectDoesNotExist

exception MultipleObjectsReturned#

Bases: MultipleObjectsReturned

aiworkflowscope_set#

Accessor to the related objects manager on the reverse side of a many-to-one relation.

In the example:

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Parent.children is a ReverseManyToOneDescriptor instance.

Most of the implementation is delegated to a dynamically defined manager class built by create_forward_many_to_many_manager() defined below.

aiworkflowsession_set#

Accessor to the related objects manager on the reverse side of a many-to-one relation.

In the example:

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Parent.children is a ReverseManyToOneDescriptor instance.

Most of the implementation is delegated to a dynamically defined manager class built by create_forward_many_to_many_manager() defined below.

base_filepath#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

clean()#

Validate the effective configuration before saving.

config#

Get the effective configuration by merging base template with overrides.

Cached per instance to avoid repeated disk reads and merging.

Returns:

Merged configuration dict

content_patch#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

property content_patch_dict: dict#

Parse content_patch as JSON5 and return as dict.

Returns:

Parsed dict from JSON5 string, or empty dict if empty/invalid

description#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

get_config() → dict#

Get the effective configuration (backward compatibility).

Use .config property instead for better performance.

get_ui_components() → dict#

Extract UIComponents from the effective configuration.

id#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

objects = <django.db.models.manager.Manager object>#
property orchestrator_class: str | None#

Get orchestrator class name from effective config.

property processor_config: dict#

Get processor config from effective config.

save(*args, **kwargs)#

Override save to validate and clear cached config.

slug#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

validate() → tuple[bool, list[str]]#

Validate the effective configuration.

Returns:

Tuple of (is_valid, error_messages)

class openedx_ai_extensions.workflows.models.AIWorkflowScope(*args, **kwargs)#

Bases: Model

exception DoesNotExist#

Bases: ObjectDoesNotExist

exception MultipleObjectsReturned#

Bases: MultipleObjectsReturned

SERVICE_VARIANTS = [('lms', 'LMS'), ('cms', 'CMS - Studio')]#
property action#

Get the runtime action if set.

aiworkflowsession_set#

Accessor to the related objects manager on the reverse side of a many-to-one relation.

In the example:

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Parent.children is a ReverseManyToOneDescriptor instance.

Most of the implementation is delegated to a dynamically defined manager class built by create_forward_many_to_many_manager() defined below.

clean()#

Validate the scope before saving.

course_id#

DO NOT REUSE THIS CLASS. Provided for backwards compatibility only!

A placeholder class that provides a way to set the attribute on the model.

enabled#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

execute(user_input, action, user, running_context) → dict[str, str | dict[str, str]] | Any#

Execute this workflow using its configured orchestrator This is where the actual AI processing happens

Returns: Dictionary with execution results

classmethod get_profile(course_id=None, location_id=None, ui_slot_selector_id=None)#

Resolve the best-matching AIWorkflowScope for the given context.

ui_slot_selector_id is required for a scope to be found. Each frontend widget sends its own identifier and only receives the scope explicitly configured for that identifier. If no ui_slot_selector_id is provided, or no scope exists for the given value, None is returned and the widget does not render.

Resolution strategy:

Phase 1 — DB filter: query all enabled scopes that match ui_slot_selector_id exactly. course_id and location_regex still act as wildcards (NULL = match any). Results are ordered by specificity_index descending so the most specific scope wins.

Phase 2 — Python regex loop: iterates over ordered candidates and returns the first scope whose location_regex matches location_id (or is NULL). The first match wins — no tie-breaking needed.

Results are cached using functools.lru_cache (max 128 entries). Cache is cleared automatically when AIWorkflowScope or AIWorkflowProfile objects are saved or deleted.

get_service_variant_display(*, field=<django.db.models.fields.CharField: service_variant>)#
id#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

classmethod list_profiles_for_context(course_id=None, location_id=None, ui_slot_selector_id=None, service_variant=None)#

Return all distinct AIWorkflowProfile objects reachable for the given context.

Unlike get_profile, which returns the single best-matching scope, this method collects every enabled scope whose course_id and location_regex match the context and returns the unique set of associated AIWorkflowProfile objects (deduplicated by profile pk).

When ui_slot_selector_id is provided, only scopes matching that exact value or the empty-string wildcard are included. When it is omitted, all profiles for the course are returned regardless of slot assignment — useful for the Studio settings panel where all available workflows should be visible.

When service_variant is provided, results are filtered to that variant only. When omitted, scopes from all service variants are returned.

Parameters:

  • course_id (str | None) – Opaque course key string.

  • location_id (str | None) – Opaque usage key string for location filtering.

  • ui_slot_selector_id (str | None) – Optional UI slot filter.

  • service_variant (str | None) – Optional service variant filter ("lms" or "cms"). When None, all variants are included.

Returns:

Deduplicated list of matching profiles, ordered by descending specificity_index of the first matching scope per profile. Each profile has a matched_scopes attribute containing all AIWorkflowScope instances that linked to it in this context.

Return type:

list[AIWorkflowProfile]

property location_id#

Get the runtime location_id if set.

location_regex#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

objects = <django.db.models.manager.Manager object>#
profile#

Accessor to the related object on the forward side of a many-to-one or one-to-one (via ForwardOneToOneDescriptor subclass) relation.

In the example:

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Child.parent is a ForwardManyToOneDescriptor instance.

profile_id#
save(*args, **kwargs)#

Override save to compute specificity_index and clear cache on changes.

service_variant#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

specificity_index#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

ui_slot_selector_id#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

class openedx_ai_extensions.workflows.models.AIWorkflowSession(*args, **kwargs)#

Bases: Model

Sessions for tracking user interactions within AI workflows

exception DoesNotExist#

Bases: ObjectDoesNotExist

exception MultipleObjectsReturned#

Bases: MultipleObjectsReturned

course_id#

DO NOT REUSE THIS CLASS. Provided for backwards compatibility only!

A placeholder class that provides a way to set the attribute on the model.

created_at#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

get_combined_thread()#

Build a unified chronological thread combining local and remote data.

The remote thread is the backbone (it has system messages, reasoning, tool calls). Local thread enriches with submission_id and timestamp. Messages are deduplicated across responses since each remote response’s input replays the full history.

Returns:

Flat list of message dicts with all available metadata.

Return type:

list or None

get_local_thread()#

Fetch the full local conversation thread from submissions.

Returns:

Messages in chronological order, or None if no submission exists.

Return type:

list or None

get_remote_thread()#

Fetch the full remote conversation thread from the LLM provider via LiteLLM.

Instantiates an LLMProcessor with the profile’s processor config so that provider credentials (api_key, api_base, etc.) are resolved and passed through.

Returns:

Chronologically ordered response dicts, or None if no remote ID exists.

Return type:

list or None

id#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

local_submission_id#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

location_id#

DO NOT REUSE THIS CLASS. Provided for backwards compatibility only!

A placeholder class that provides a way to set the attribute on the model.

metadata#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

objects = <django.db.models.manager.Manager object>#
profile#

Accessor to the related object on the forward side of a many-to-one or one-to-one (via ForwardOneToOneDescriptor subclass) relation.

In the example:

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Child.parent is a ForwardManyToOneDescriptor instance.

profile_id#
remote_response_id#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

scope#

Accessor to the related object on the forward side of a many-to-one or one-to-one (via ForwardOneToOneDescriptor subclass) relation.

In the example:

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Child.parent is a ForwardManyToOneDescriptor instance.

scope_id#
updated_at#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

user#

Accessor to the related object on the forward side of a many-to-one or one-to-one (via ForwardOneToOneDescriptor subclass) relation.

In the example:

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Child.parent is a ForwardManyToOneDescriptor instance.

user_id#

openedx_ai_extensions.workflows.template_utils module#

Utilities for discovering, loading, and validating workflow templates.

Templates are read-only JSON5 files stored on disk (allowing comments). Security: Only load from configured directories to prevent path traversal.

openedx_ai_extensions.workflows.template_utils.discover_templates() → list[tuple[str, str]]#

Discover all available workflow templates.

Returns:

List of (relative_path, display_name) tuples for Django choices

openedx_ai_extensions.workflows.template_utils.get_effective_config(base_filepath: str, content_patch: dict) → dict | None#

Get the effective configuration by merging base template with patch.

This is the main function used by AIWorkflowProfile.

Parameters:

  • base_filepath – Relative path to base template

  • content_patch – JSON patch to apply

Returns:

Effective configuration, or None if base template cannot be loaded

openedx_ai_extensions.workflows.template_utils.get_template_directories() → list[Path]#

Get list of allowed template directories from settings.

Returns:

List of Path objects pointing to template directories

openedx_ai_extensions.workflows.template_utils.is_safe_template_path(template_path: str) → bool#

Verify that a template path is safe (no path traversal attacks).

Parameters:

template_path – Relative path to template file

Returns:

True if path is safe, False otherwise

openedx_ai_extensions.workflows.template_utils.load_template(template_path: str) → dict | None#

Load a workflow template from disk.

Supports JSON5 format (allows comments, trailing commas, etc).

Parameters:

template_path – Relative path to template file

Returns:

Template data as dict, or None if not found/invalid

openedx_ai_extensions.workflows.template_utils.merge_template_with_patch(base_template: dict, patch: dict) → dict#

Merge a base template with a JSON patch.

Uses RFC 7386 JSON Merge Patch via jsonmerge library.

Parameters:

  • base_template – Base template configuration

  • patch – JSON patch to apply

Returns:

Merged configuration

openedx_ai_extensions.workflows.template_utils.parse_json5_string(json5_string: str) → dict#

Parse a JSON5 string into a dict.

Allows comments, trailing commas, etc.

Parameters:

json5_string – JSON5-formatted string

Returns:

Parsed dict

Raises:

json5.JSON5DecodeError – If string is invalid JSON5

openedx_ai_extensions.workflows.template_utils.validate_workflow_config(config: dict) → tuple[bool, list[str]]#

Validate a workflow configuration against the JSON schema.

Enforces schema version 1.0 requirements.

Parameters:

config – Configuration to validate

Returns:

Tuple of (is_valid, error_messages)

Module contents#

Contents