gen_ai_hub.orchestration.service module

Module for orchestration service handling requests and responses.

Provides synchronous and asynchronous methods to run orchestration pipelines.

class gen_ai_hub.orchestration.service.OrchestrationRequest(config: OrchestrationConfig, template_values: List[TemplateValue], history: List[Message])

Bases: JSONSerializable

Represents a request for the orchestration process, including configuration, template values, and message history.

config: OrchestrationConfig

The orchestration configuration for the request.

Returns:

OrchestrationConfig

Return type:

OrchestrationConfig

template_values: List[TemplateValue]

List of template values to be used in the orchestration.

history: List[Message]

List of messages representing the conversation history.

to_dict()

Converts the OrchestrationRequest instance to a dictionary.

Returns:

Dictionary representation of the OrchestrationRequest

Return type:

dict

__init__(config: OrchestrationConfig, template_values: List[TemplateValue], history: List[Message]) → None
gen_ai_hub.orchestration.service.cache_if_not_none(func)

Custom cache decorator that only caches non-None results

Parameters:

func (callable) – The function to be decorated.

Returns:

The decorated function with caching behavior.

Return type:

callable

gen_ai_hub.orchestration.service.discover_orchestration_api_url(base_url: str, auth_url: str, client_id: str, client_secret: str, resource_group: str, config_id: str | None = None, config_name: str | None = None, orchestration_scenario: str = 'orchestration', executable_id: str = 'orchestration') → str | None

Discovers the orchestration API URL based on provided configuration details.

Parameters:

  • base_url (str) – the base URL for the AI Core API.

  • auth_url (str) – the URL for the AI Core authentication service.

  • client_id (str) – the client ID for the AI Core API.

  • client_secret (str) – the client secret for the AI Core API.

  • resource_group (str) – the resource group for the AI Core API.

  • config_id (Optional[str], optional) – the configuration ID, defaults to None

  • config_name (Optional[str], optional) – the configuration name, defaults to None

  • orchestration_scenario (str, optional) – the orchestration scenario ID, defaults to “orchestration”

  • executable_id (str, optional) – the orchestration executable ID, defaults to “orchestration”

Returns:

The orchestration API URL or None if no deployment is found.

Return type:

Optional[str]

gen_ai_hub.orchestration.service.get_orchestration_api_url(proxy_client: GenAIHubProxyClient, deployment_id: str | None = None, config_name: str | None = None, config_id: str | None = None) → str

Retrieves the orchestration API URL based on provided deployment or configuration details.

Parameters:

  • proxy_client (GenAIHubProxyClient) – The GenAIHubProxyClient instance.

  • deployment_id (Optional[str], optional) – the deployment ID, defaults to None

  • config_name (Optional[str], optional) – the configuration name, defaults to None

  • config_id (Optional[str], optional) – the configuration ID, defaults to None

Raises:

ValueError – If no orchestration deployment is found.

Returns:

The orchestration API URL.

Return type:

str

class gen_ai_hub.orchestration.service.OrchestrationService(api_url: str | None = None, config: OrchestrationConfig | None = None, proxy_client: GenAIHubProxyClient | None = None, deployment_id: str | None = None, config_name: str | None = None, config_id: str | None = None, timeout: int | float | Timeout | None = None)

Bases: object

A service for executing orchestration requests, allowing for the generation of LLM-generated content through a pipeline of configured modules. This service supports both synchronous and asynchronous request execution. For streaming responses, special care is taken to not close the underlying HTTP stream prematurely.

https://api.sap.com/api/ORCHESTRATION_API/overview

__init__(api_url: str | None = None, config: OrchestrationConfig | None = None, proxy_client: GenAIHubProxyClient | None = None, deployment_id: str | None = None, config_name: str | None = None, config_id: str | None = None, timeout: int | float | Timeout | None = None)

Initializes the OrchestrationService with the provided parameters.

Parameters:

  • api_url (Optional[str], optional) – The base URL for the orchestration API, defaults to None

  • config (Optional[OrchestrationConfig], optional) – The default orchestration configuration, defaults to None

  • proxy_client (Optional[GenAIHubProxyClient], optional) – The GenAIHubProxyClient instance, defaults to None

  • deployment_id (Optional[str], optional) – the deployment ID, defaults to None

  • config_name (Optional[str], optional) – the configuration name, defaults to None

  • config_id (Optional[str], optional) – the configuration ID, defaults to None

  • timeout (Union[int, float, httpx.Timeout, None], optional) – the timeout for HTTP requests, defaults to None

run(config: OrchestrationConfig | None = None, template_values: List[TemplateValue] | None = None, history: List[Message] | None = None, timeout: int | float | Timeout | None = None) → OrchestrationResponse

Executes an orchestration request synchronously (non-streaming).

Parameters:

  • config (Optional[OrchestrationConfig], optional) – the orchestration configuration, defaults to None

  • template_values (Optional[List[TemplateValue]], optional) – the template values for the request, defaults to None

  • history (Optional[List[Message]], optional) – the message history, defaults to None

  • timeout (Union[int, float, httpx.Timeout, None], optional) – the timeout for the request, defaults to None

Returns:

An OrchestrationResponse object.

Return type:

OrchestrationResponse

stream(config: OrchestrationConfig | None = None, template_values: List[TemplateValue] | None = None, history: List[Message] | None = None, stream_options: dict | None = None, timeout: int | float | Timeout | None = None) → SSEClient

Executes an orchestration request in streaming mode (synchronously).

Parameters:

  • config (Optional[OrchestrationConfig], optional) – the orchestration configuration, defaults to None

  • template_values (Optional[List[TemplateValue]], optional) – the template values for the request, defaults to None

  • history (Optional[List[Message]], optional) – the message history, defaults to None

  • stream_options (Optional[dict], optional) – the additional streaming options, defaults to None

  • timeout (Union[int, float, httpx.Timeout, None], optional) – the timeout for the request, defaults to None

Returns:

An SSEClient instance for iterating over the streaming response.

Return type:

SSEClient

async arun(config: OrchestrationConfig | None = None, template_values: List[TemplateValue] | None = None, history: List[Message] | None = None, timeout: int | float | Timeout | None = None) → OrchestrationResponse

Executes an orchestration request asynchronously (non-streaming).

Parameters:

  • config (Optional[OrchestrationConfig], optional) – the orchestration configuration, defaults to None

  • template_values (Optional[List[TemplateValue]], optional) – the template values for the request, defaults to None

  • history (Optional[List[Message]], optional) – the message history, defaults to None

  • timeout (Union[int, float, httpx.Timeout, None], optional) – the timeout for the request, defaults to None

Returns:

An OrchestrationResponse object.

Return type:

OrchestrationResponse

async astream(config: OrchestrationConfig | None = None, template_values: List[TemplateValue] | None = None, history: List[Message] | None = None, stream_options: dict | None = None, timeout: int | float | Timeout | None = None) → AsyncSSEClient

Executes an orchestration request asynchronously in streaming mode.

Parameters:

  • config (Optional[OrchestrationConfig], optional) – the orchestration configuration, defaults to None

  • template_values (Optional[List[TemplateValue]], optional) – the template values for the request, defaults to None

  • history (Optional[List[Message]], optional) – the message history, defaults to None

  • stream_options (Optional[dict], optional) – the additional streaming options, defaults to None

  • timeout (Union[int, float, httpx.Timeout, None], optional) – the timeout for the request, defaults to None

Returns:

An AsyncSSEClient instance for iterating over the streaming response.

Return type:

AsyncSSEClient

close_http_connection()

Closes the httpx synchronous client.

async aclose_http_connection()

Closes the httpx asynchronous client.

run_with_retries(config: OrchestrationConfig | None = None, template_values: List[TemplateValue] | None = None, history: List[Message] | None = None, timeout: int | float | Timeout | None = None, max_retries: int = 10, base_delay: float = 1.0) → OrchestrationResponseWithRetries | None

Executes an orchestration request with automatic retry on rate limits (429) and server errors.

Parameters:

  • config (Optional[OrchestrationConfig], optional) – the orchestration configuration, defaults to None

  • template_values (Optional[List[TemplateValue]], optional) – the template values for the request, defaults to None

  • history (Optional[List[Message]], optional) – the message history, defaults to None

  • timeout (Union[int, float, httpx.Timeout, None], optional) – the timeout for the request, defaults to None

  • max_retries (int, optional) – the maximum number of retry attempts, defaults to 10

  • base_delay (float, optional) – the initial delay between retries in seconds, defaults to 1.0

Returns:

An OrchestrationResponseWithRetries with retry count information.

Return type:

OrchestrationResponseWithRetries | None

Raises:

  • OrchestrationError – If the request fails after all retries (includes retry count).

  • ValueError – If no configuration is provided.

handle_retry(retry_count: int, base_delay: float, error: OrchestrationError, max_retries: int) → float

Handles retry logic with exponential backoff and jitter. If Retry-After header exists, use it as min_delay to add jitter on top

Parameters:

  • retry_count (int) – the current retry attempt number

  • base_delay (float) – the initial delay between retries in seconds

  • error (OrchestrationError) – the exception that occurred

  • max_retries (int) – the maximum number of retry attempts

Raises:

error – Raises the original error if no more retries should be attempted

Returns:

number of seconds to wait before next retry

Return type:

float

async arun_with_retries(config: OrchestrationConfig | None = None, template_values: List[TemplateValue] | None = None, history: List[Message] | None = None, timeout: int | float | Timeout | None = None, max_retries: int = 10, base_delay: float = 1.0) → OrchestrationResponseWithRetries | None

Executes an orchestration request asynchronously with automatic retry on rate limits (429) and server errors. Uses exponential backoff with jitter to handle rate limiting gracefully.

Parameters:

  • config (Optional[OrchestrationConfig], optional) – the orchestration configuration, defaults to None

  • template_values (Optional[List[TemplateValue]], optional) – the template values for the request, defaults to None

  • history (Optional[List[Message]], optional) – the message history, defaults to None

  • timeout (Union[int, float, httpx.Timeout, None], optional) – the timeout for the request, defaults to None

  • max_retries (int, optional) – the maximum number of retry attempts, defaults to 10

  • base_delay (float, optional) – the initial delay between retries in seconds, defaults to 1.0

Returns:

An OrchestrationResponseWithRetries with retry count information.

Return type:

OrchestrationResponseWithRetries | None

Raises:

  • OrchestrationError – If the request fails after all retries (includes retry count).

  • ValueError – If no configuration is provided.