Orchestration Service V2 API

This page demonstrates how to use the SDK with the Orchestration Service V2, enabling AI-driven workflows with modules such as templating, LLMs, data masking, and content filtering. By leveraging these modules, you can build complex, automated workflows that enhance the capabilities of your AI solutions. For more details on configuring and using these modules, please refer to the Orchestration Service Documentation.

Prerequisite

Before you begin, set up a virtual deployment of the Orchestration Service. See the setup guide.

Authentication

By default, the OrchestrationService initializes a GenAIHubProxyClient, which automatically configures credentials using configuration files or environment variables, as outlined in the Introduction section. If you prefer to set credentials manually, you can provide a custom instance using the proxy_client parameter.

Basic Orchestration Pipeline

Let's walk through a basic orchestration pipeline for a translation task.

Step 1: Define the Template

The Template class is used to define structured message templates for generating dynamic interactions with language models. In this example, the template is designed for a translation assistant, allowing users to specify a language and text for translation.

from gen_ai_hub.orchestration_v2 import Template, SystemMessage, UserMessage

template = Template(
    template=[
        SystemMessage(content="You are a helpful translation assistant."),
        UserMessage(content="Translate the following text to {{?to_lang}}: {{?user_query}}"),
    ],
    defaults={"to_lang": "German"}
)

This template can be used to create translation requests where the language and text to be translated are specified dynamically. The placeholders in the UserMessage will be replaced with the actual values provided at runtime, and the default value for the language is set to German.

Step 2: Define the LLM

The LLM class is used to configure and initialize a language model for generating text based on specific parameters. In this example, we'll use the gpt-5-nano model to perform the translation task.

Note: The Orchestration Service automatically manages the virtual deployment of the language model, so no additional setup is needed on your end.

from gen_ai_hub.orchestration_v2 import LLMModelDetails

llm = LLMModelDetails(name="gpt-5-nano", params={"max_completion_tokens": 2048})

Initializes the language model to use the gpt-5-nano model. It will generate responses up to 2048 tokens in length.

Step 3: Create the Configuration

The OrchestrationConfig class defines a configuration for integrating various modules, such as templates and language models, into a cohesive orchestration setup. It specifies how these components interact and are configured to achieve the desired operational scenario.

from gen_ai_hub.orchestration_v2 import PromptTemplatingModuleConfig, ModuleConfig, OrchestrationConfig

prompt_template = PromptTemplatingModuleConfig(prompt=template, model=llm)
module_config = ModuleConfig(prompt_templating=prompt_template)
config = OrchestrationConfig(modules=module_config)

Step 4: Run the Request

The OrchestrationService class is used to interact with a orchestration service instance by providing configuration details to initiate and manage its operations.

from gen_ai_hub.orchestration_v2 import OrchestrationService

orchestration_service = OrchestrationService(config=config)

Call the run method with the required placeholder values. The service will process the input according to the configuration and return the result.

result = orchestration_service.run(placeholder_values={"user_query": "The Orchestration Service is working!"})
print(result.final_result.choices[0].message.content)

Additional Pipeline Options

Beyond the basic four-step setup, the orchestration pipeline supports several configuration options that you can apply independently or in combination.

Using Fallback Configurations

Use multiple configurations to make your pipeline resilient: if one fails, the service automatically tries the next.

You can provide multiple ModuleConfig objects to OrchestrationConfig instead of a single one. The configurations are tried in order: if a call with one configuration fails, the next one is used as fallback.

This works for both regular and streaming calls.

from gen_ai_hub.orchestration_v2 import (SystemMessage, UserMessage, Template, PromptTemplatingModuleConfig,
                                        LLMModelDetails, ModuleConfig, OrchestrationConfig, OrchestrationService)

working_config = ModuleConfig(
    prompt_templating=PromptTemplatingModuleConfig(
        prompt=Template(template=[UserMessage(content="What is the longest river on planet earth?")]),
        model=LLMModelDetails(name="gpt-4o-mini"),
    )
)

second_working_config = ModuleConfig(
    prompt_templating=PromptTemplatingModuleConfig(
        prompt=Template(template=[UserMessage(content="What is the longest river on planet earth?")]),
        model=LLMModelDetails(name="gpt-4o"),
    )
)

broken_config = ModuleConfig(
    prompt_templating=PromptTemplatingModuleConfig(
        prompt=Template(template=[UserMessage(content="What is the longest river on planet earth?")]),
        model=LLMModelDetails(name="dummy-model"),
    )
)

# broken_config is tried first and will fail, so working_config is used as fallback
config = OrchestrationConfig(modules=[broken_config, working_config, second_working_config])

service = OrchestrationService(config=config)
result = service.run()
print(result.final_result.choices[0].message.content)

Referencing Templates from the Prompt Registry

In Step 3 you can also use a prompt template reference, which allows you to reuse existing templates stored in the Prompt Registry.

from gen_ai_hub.orchestration_v2 import TemplateRefByID, TemplateRefByScenarioNameVersion

template_by_id = TemplateRefByID(id="648871d9-b207-441c-8c13-afee71b0dbec") # this is just an example id
template_by_names = TemplateRefByScenarioNameVersion(scenario="translation", name="translate_text", version="0.1.0")

Response Format Options

The response_format parameter allows the model output to be formatted in several predefined ways, as follows:

1. text: This is the simplest form where the model's output is generated as plain text. It is suitable for applications that require raw text processing.
2. json_object: Under this setting, the model's output is structured as a JSON object. This is useful for applications that handle data in JSON format, enabling easy integration with web applications and APIs.
3. json_schema: This setting allows the model's output to adhere to a defined JSON schema. This is particularly useful for applications that require strict data validation, ensuring the output matches a predefined schema.

Text:

from gen_ai_hub.orchestration_v2 import SystemMessage, UserMessage, Template, ResponseFormatText

template = Template(
    template=[SystemMessage(content="You are a helpful assistant."), UserMessage(content="{{?user_query}}")],
    response_format=ResponseFormatText(),
    defaults={"user_query": "Who was the first person on the moon?"}
)

# Response:
# The first man on the moon was Neil Armstrong.

JSON Object:

from gen_ai_hub.orchestration_v2 import SystemMessage, UserMessage, Template, ResponseFormatJsonObject

template = Template(
    template=[
        SystemMessage(content="You are a helpful assistant. Format the response as json."),
        UserMessage(content="{{?user_query}}")
    ],
    response_format=ResponseFormatJsonObject(),
    defaults={"user_query": "Who was the first person on the moon?"}
)

# Response:
# {
#     "First_man_on_the_moon": "Neil Armstrong"
# }

Important: When using response_format as json_object, ensure that messages contain the word 'json' in some form.

JSON Schema:

from gen_ai_hub.orchestration_v2 import SystemMessage, UserMessage, Template, ResponseFormatJsonSchema, JSONResponseSchema

json_schema = {
    "title": "Person", "type": "object",
    "properties": {
        "firstName": {"type": "string", "description": "The person's first name."},
        "lastName": {"type": "string", "description": "The person's last name."}
    }
}
template = Template(
    template=[SystemMessage(content="You are a helpful assistant."), UserMessage(content="{{?user_query}}")],
    response_format=ResponseFormatJsonSchema(
        json_schema=JSONResponseSchema(name="person", description="person mapping", schema=json_schema)
    ),
    defaults={"user_query": "Who was the first person on the moon?"}
)

# Response:
# {
#     "firstName": "Neil",
#     "lastName": "Armstrong"
# }

Understanding Deployment Resolution

The OrchestrationService class provides multiple ways to specify and target orchestration deployments when sending requests. Below are the available options:

Default Behavior

If no parameters are provided, the OrchestrationService automatically searches for a RUNNING deployment. If multiple running deployments exist, the service selects the most recently created one.

Direct Deployment Specification

You can explicitly define the target deployment using the following options:

1. API URL (api_url):
   • Specify the exact URL assigned to the deployment during its creation.
   • Refer to the Prerequisites section for more details on obtaining the deployment URL.
2. Deployment ID (deployment_id):
   • Use the unique identifier assigned to the deployment instead of the URL.

Config-Based Specification

If you want to target deployments based on their configuration source, use one of the following options:

1. Configuration ID (config_id):
   • The OrchestrationService searches for a RUNNING deployment created using the provided configuration ID.
2. Configuration Name (config_name):
   • The service looks for a RUNNING deployment that matches the specified configuration name.

If multiple deployments match the given configuration criteria, the most recently created one will be selected automatically.

Optional Modules

Data Masking

The Data Masking module anonymizes or pseudonymizes personally identifiable information (PII) before it is processed by the LLM module. Currently, SAPDataPrivacyIntegration is the only available masking provider.

Masking Types

• Anonymization: All identifying information is replaced with placeholders (e.g., MASKED_ENTITY), and the original data cannot be recovered, ensuring that no trace of the original information is retained.
• Pseudonymization: Data is substituted with unique placeholders (e.g., MASKED_ENTITY_ID), allowing the original information to be restored if needed.

In both cases, the masking module identifies sensitive data and replaces it with appropriate placeholders before further processing.

Configuration Options

• entities: Specify which types of entities to mask (e.g., EMAIL, PHONE, PERSON).
• allowlist: Provide specific terms or patterns that should be excluded from masking, even if they match entity types.
• mask_grounding_input: When enabled, ensures that masking is also applied to the context provided to the grounding module.

from gen_ai_hub.orchestration_v2.utils import load_text_file
from gen_ai_hub.orchestration_v2 import (
    SystemMessage, UserMessage, Template, PromptTemplatingModuleConfig,
    LLMModelDetails, ModuleConfig, OrchestrationConfig, OrchestrationService,
    MaskingModuleConfig, MaskingProviderConfig, MaskingMethod, DPIStandardEntity, ProfileEntity
)

data_masking_config = MaskingModuleConfig(
    providers=[MaskingProviderConfig(
        method=MaskingMethod.ANONYMIZATION,
        entities=[
            DPIStandardEntity(type=ProfileEntity.ADDRESS),
            DPIStandardEntity(type=ProfileEntity.EMAIL),
            DPIStandardEntity(type=ProfileEntity.PHONE),
            DPIStandardEntity(type=ProfileEntity.PERSON),
        ],
        allowlist=["M&K Group"],
    )],
)

template = Template(template=[
    SystemMessage(content="You are a helpful AI assistant."),
    UserMessage(content="Summarize the following CV in 10 sentences: {{?text}}"),
])
llm = LLMModelDetails(name="gpt-4o")
module_config = ModuleConfig(
    prompt_templating=PromptTemplatingModuleConfig(prompt=template, model=llm),
    masking=data_masking_config
)
config = OrchestrationConfig(modules=module_config)

cv_as_string = load_text_file("data/cv.txt")
result = OrchestrationService().run(config=config, placeholder_values={"text": cv_as_string})
print(result.final_result.choices[0].message.content)

Content Filtering

The Content Filtering module can be configured to filter both the input to the LLM module (input filter) and the output generated by the LLM (output filter). The module uses predefined classification services to detect inappropriate or unwanted content. Azure Content Filter sensitivity is controlled by customizable thresholds, assuring the content aligns with the desired standards before processing or generating as output. Llama Guard 3 Filter, equipped with 14 categories, runs on a binary mechanism, accepting only true or false. Setting a category to true enables filtering for it. It's possible to execute both filters in a single request, optimizing efficiency.

from gen_ai_hub.orchestration_v2 import (
    AzureContentSafetyInput, AzureContentSafetyOutput, AzureThreshold,
    LlamaGuard38bFilter, FilteringModuleConfig, InputFiltering, OutputFiltering,
    AzureContentSafetyInputFilterConfig, AzureContentSafetyOutputFilterConfig, LlamaGuard38bFilterConfig,
    Template, SystemMessage, UserMessage, LLMModelDetails, PromptTemplatingModuleConfig,
    ModuleConfig, OrchestrationConfig, OrchestrationService, OrchestrationError
)

template = Template(template=[
    SystemMessage(content="You are a helpful AI assistant."),
    UserMessage(content="{{?text}}"),
])
prompt_template = PromptTemplatingModuleConfig(prompt=template, model=LLMModelDetails(name="gpt-4o"))

content_filter_config = FilteringModuleConfig(
    input=InputFiltering(filters=[
        AzureContentSafetyInputFilterConfig(config=AzureContentSafetyInput(
            hate=AzureThreshold.ALLOW_SAFE, violence=AzureThreshold.ALLOW_SAFE,
            self_harm=AzureThreshold.ALLOW_SAFE, sexual=AzureThreshold.ALLOW_SAFE)),
        LlamaGuard38bFilterConfig(config=LlamaGuard38bFilter(hate=True))
    ]),
    output=OutputFiltering(filters=[
        AzureContentSafetyOutputFilterConfig(config=AzureContentSafetyOutput(
            hate=AzureThreshold.ALLOW_SAFE, violence=AzureThreshold.ALLOW_SAFE,
            self_harm=AzureThreshold.ALLOW_SAFE, sexual=AzureThreshold.ALLOW_SAFE)),
        LlamaGuard38bFilterConfig(config=LlamaGuard38bFilter(hate=True))
    ])
)

module_config = ModuleConfig(prompt_templating=prompt_template, filtering=content_filter_config)

config = OrchestrationConfig(modules=module_config)

client = OrchestrationService(config=config)

try:
    result = client.run(placeholder_values={"text": "I hate you"})
    print(result.final_result.choices[0].message.content)
except OrchestrationError as er:
    print(er.message)

Streaming

When you initiate an orchestration request, the full response is typically processed and delivered in one go. For longer responses, this can lead to delays in receiving the complete output. To mitigate this, you have the option to stream the results as they are being generated. This helps in rapidly processing or displaying initial portions of the results without waiting for the entire computation to finish.

To activate streaming, use the stream method of the OrchestrationService with the stream option in OrchestrationConfig. This method returns an object that streams chunks of the response as they become available. You can then extract relevant information from the delta field.

Here's how you can set up a simple configuration to stream orchestration results:

from gen_ai_hub.orchestration_v2 import (
    Template, SystemMessage, UserMessage, LLMModelDetails, PromptTemplatingModuleConfig,
    ModuleConfig, OrchestrationConfig, OrchestrationService, GlobalStreamOptions
)

template = Template(template=[
    SystemMessage(content="You are a helpful AI assistant."),
    UserMessage(content="{{?text}}"),
])
llm = LLMModelDetails(name="gpt-4o")
prompt_template = PromptTemplatingModuleConfig(prompt=template, model=llm)
module_config = ModuleConfig(prompt_templating=prompt_template)

config = OrchestrationConfig(modules=module_config, stream=GlobalStreamOptions(enabled=True))
client = OrchestrationService(config=config)

result = client.stream(placeholder_values={"text": "Which color is the sky? Answer in one sentence."})
for part in result:
    print(part.final_result.choices[0].delta.content)
    print("*" * 20)

Note: As shown above, streaming responses contain a delta field instead of a message field.

You can customize the global stream behavior by setting options like chunk_size which controls the amount of data processed in each chunk:

config = OrchestrationConfig(modules=module_config, stream=GlobalStreamOptions(enabled=True, chunk_size=25))
client = OrchestrationService(config=config)

result = client.stream(placeholder_values={"text": "Which color is the sky? Answer in one sentence."})
for part in result:
    print(part.final_result.choices[0].delta.content)
    print("*" * 20)

Modules that influence or process streaming results, such as OutputFiltering, might need specific stream options. The overlap option allows you to include extra context during the filtering process.

from gen_ai_hub.orchestration_v2 import (
    Template, SystemMessage, UserMessage, LLMModelDetails, PromptTemplatingModuleConfig,
    ModuleConfig, OrchestrationConfig, OrchestrationService, GlobalStreamOptions,
    AzureContentSafetyOutput, AzureContentSafetyOutputFilterConfig,
    FilteringModuleConfig, OutputFiltering, FilteringStreamOptions
)

content_filter_config = FilteringModuleConfig(
    output=OutputFiltering(
        filters=[AzureContentSafetyOutputFilterConfig(config=AzureContentSafetyOutput(hate=0))],
        stream_options=FilteringStreamOptions(overlap=20))
)

template = Template(
    template=[
        SystemMessage(content="You are a helpful AI assistant."),
        UserMessage(content="{{?text}}"),
    ]
    )

llm=LLMModelDetails(name="gpt-4o")

prompt_template = PromptTemplatingModuleConfig(prompt=template,
                                               model=llm)

module_config = ModuleConfig(prompt_templating=prompt_template, filtering=content_filter_config)

config = OrchestrationConfig(modules=module_config,
                             stream=GlobalStreamOptions(enabled=True))

client = OrchestrationService(config=config)

response = client.stream(placeholder_values={"text": "Which color is the sky? Answer in one sentence."})

for chunk in response:
    print(chunk.final_result.choices[0].delta.content, end='')

Reasoning

The Orchestration Service accepts a harmonized parameter reasoning_effort to control the level of reasoning a model performs. This parameter can take the values minimal, low, medium, high, and none, which are mapped to the closest equivalent in the model's native reasoning configuration. See the Orchestration Service reasoning documentation for details.

When reasoning is enabled, models can expose their intermediate reasoning content in addition to the final answer. This reasoning content is returned as a list of reasoning blocks, which contain the human-readable reasoning text in their content attribute.

config = OrchestrationConfig(
    modules=ModuleConfig(
        prompt_templating=PromptTemplatingModuleConfig(
            prompt=Template(
                template=[
                    UserMessage(
                        content="What is the longest river on planet earth?"
                    )
                ]
            ),
            # pass the reasoning_effort parameter in params
            model=LLMModelDetails(name="gemini-3.5-flash", params={ "reasoning_effort": "high" }),
        )
    )
)
service = OrchestrationService(config=config)
result = service.run()
reasoning_content = result.final_result.choices[0].message.reasoning_content
if reasoning_content:
    # print the human-readable reasoning content
    print("".join([block.content for block in reasoning_content]))

In streaming responses, the reasoning content is spread across different chunks which are again lists of reasoning blocks available in the reasoning_content attribute of each delta.

config = OrchestrationConfig(
    modules=ModuleConfig(
        prompt_templating=PromptTemplatingModuleConfig(
            prompt=Template(
                template=[
                    UserMessage(
                        content="What is the longest river on planet earth?"
                    )
                ]
            ),
            model=LLMModelDetails(name="gemini-3.5-flash", params={ "reasoning_effort": "high" }),
        )
    ),
    stream=GlobalStreamOptions(enabled=True),
)

service = OrchestrationService(config=config)
stream = service.stream()
for chunk in stream:
    reasoning_content_chunk = chunk.final_result.choices[0].delta.reasoning_content
    if reasoning_content_chunk:
        # print the human-readable reasoning content chunk
        print("".join([block.content for block in reasoning_content_chunk]))

note

In multi-turn conversations, the full and unmodified list of reasoning_content blocks needs to be passed in the message history. In a streaming setting, the chunks have to be accumulated before using them in a follow-up request.

Tool Calling (Function Calling)

The Orchestration Service supports tool calling, which allows large language models (LLMs) to request the execution of external operations such as Python functions, API calls, or other tools as part of their workflow. This feature enables you to build advanced AI solutions that can perform calculations, access data, or interact with external systems in response to user input.

Defining Tools

You can define tools in several ways, depending on your requirements and the level of control you need.

Using the Python Decorator

The simplest way to define a tool is to decorate a Python function with @function_tool(). The function's signature and docstring are used to describe the tool to the LLM.

from gen_ai_hub.orchestration_v2 import function_tool

@function_tool()
def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

@function_tool()
def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

tools = [multiply, add]

Using the FunctionTool Class

For more control, you can use the FunctionTool class directly. This is useful if you want to customize the schema, enable strict argument checking, or wrap an existing function.

from gen_ai_hub.orchestration_v2 import FunctionTool, FunctionObject

def get_weather(location: str) -> str:
    """Get current temperature for a given location."""
    return "22°C"

weather_tool_func = FunctionObject(
    name="get_weather",
    description="Get current temperature for a given location.",
    parameters={
        "type": "object",
        "properties": {"location": {"type": "string", "description": "City and country e.g. Bogotá, Colombia"}},
        "required": ["location"],
        "additionalProperties": False
    },
    strict=True,
    function=get_weather
)
weather_tool = FunctionTool(function=weather_tool_func)

You can also create a FunctionTool from a function using the from_function static method:

weather_tool = FunctionTool.from_function(get_weather, strict=True)

Using a JSON Schema Dictionary

You can define a tool directly as a JSON schema dictionary. This is useful if you want to specify the tool interface without implementing the function in Python, or if you want to integrate with external systems.

tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "Get current temperature for a given location.",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "City and country e.g. Bogotá, Colombia"
                }
            },
            "required": [
                "location"
            ],
            "additionalProperties": False
        },
        "strict": True
    }
}]

You can then attach any of these tool definitions to your template:

from gen_ai_hub.orchestration_v2 import Template, SystemMessage, UserMessage

template = Template(
    template=[
        SystemMessage(content="You are a weather assistant."),
        UserMessage(content="What is the temperature in {{?location}}?"),
    ],
    tools=tools,
)

Synchronous Tool Call Workflow

When the LLM decides to call a tool, the orchestration response will include a tool_calls field. You are responsible for executing the tool(s), adding the results to the conversation history, and running the orchestration again to get the final answer.

from typing import List
from gen_ai_hub.orchestration_v2 import (
    LLMModelDetails, PromptTemplatingModuleConfig, ModuleConfig, OrchestrationConfig, OrchestrationService,
    ChatMessage, ToolChatMessage
)

llm = LLMModelDetails(name="gpt-4o-mini", params={"max_completion_tokens": 200, "temperature": 0.0})
prompt_template = PromptTemplatingModuleConfig(prompt=template, model=llm)
module_config = ModuleConfig(prompt_templating=prompt_template)
config = OrchestrationConfig(modules=module_config)

service = OrchestrationService()
template_values = {"location": "Bogotá, Colombia"}

response = service.run(config=config, placeholder_values=template_values)
tool_calls = response.final_result.choices[0].message.tool_calls

history: List[ChatMessage] = []
history.extend(response.intermediate_results.templating)
history.append(response.final_result.choices[0].message)

for tool_call in tool_calls:
    result = weather_tool.execute(**tool_call.function.parse_arguments())
    history.append(ToolChatMessage(content=str(result), tool_call_id=tool_call.id))

response2 = service.run(config=config, placeholder_values=template_values, history=history)
print(response2.final_result.choices[0].message.content)

Streaming Tool Calls

When using streaming, tool calls may be split across multiple chunks. The delta.tool_calls field in each chunk contains partial or complete tool call information. You may need to buffer and concatenate arguments if they arrive in pieces.

from gen_ai_hub.orchestration_v2 import OrchestrationConfig, GlobalStreamOptions, OrchestrationService

config = OrchestrationConfig(modules=module_config, stream=GlobalStreamOptions(enabled=True))
service = OrchestrationService()
stream = service.stream(config=config, placeholder_values=template_values)

final_tool_calls = {}

for chunk in stream:
    for tool_call in chunk.final_result.choices[0].delta.tool_calls or []:
        index = tool_call.index
        if index not in final_tool_calls:
            final_tool_calls[index] = tool_call
        else:
            # Concatenate arguments if split across chunks
            final_tool_calls[index].function.arguments += tool_call.function.arguments

# Now final_tool_calls contains all tool calls with complete arguments

⚠️ Note on Agentic Loop Support:

The current SDK does not provide built-in abstractions or convenience methods for managing the agentic loop (the process of automatically handling tool call detection, execution, and iterative orchestration until a final answer is produced).

As a user, you are responsible for:

• Detecting tool calls in the LLM response
• Executing the corresponding Python functions
• Appending tool results to the conversation history (as ToolMessage)
• Re-invoking the orchestration service as needed

This approach gives you maximum flexibility, but you must implement the orchestration loop logic yourself.

Prompt Caching

Cache control improves performance by caching requests at the model provider level. Add a cache_control breakpoint to a content block to mark a cache breakpoint in the prompt. Availability of different time-to-live (TTL) options may depend on the model.

This feature covers explicit cache control for AWS Bedrock-hosted models (Anthropic Claude and Amazon Nova). Other model providers, such as OpenAI and Gemini, use implicit context caching that is enabled by default and requires no additional configuration.

Note: Prompt caching requires a minimum number of tokens in the cached content. If the prompt is too short, the cache breakpoint may not take effect.

Attaching cache_control to Content Blocks

Set cache_control on individual content parts inside a message. The example below caches a long system prompt and marks the last user turn as a second cache breakpoint, following the few-shot pattern recommended for classification tasks.

from gen_ai_hub.orchestration_v2 import (
    CacheControl, TextPart,
    SystemMessage, UserMessage, AssistantMessage, Template,
    LLMModelDetails, PromptTemplatingModuleConfig, ModuleConfig,
    OrchestrationConfig, OrchestrationService,
)

# A long system prompt whose tokens will be cached after the first request.
system_text = "You are an expert news article classifier. Classify each article into exactly one category."
system_text += " Classify each article into exactly one category." * 512  # repeat to exceed the minimum token threshold

template = Template(
    template=[
        SystemMessage(
            content=[
                TextPart(
                    text=system_text,
                    cache_control=CacheControl(type="ephemeral"),
                )
            ]
        ),
        # Few-shot examples (plain strings, not cached)
        UserMessage(content="input: Comcast launches prepaid plans"),
        AssistantMessage(content="Business"),
        # Dynamic user turn: second cache breakpoint
        UserMessage(
            content=[
                TextPart(
                    text="input: {{?article}}",
                    cache_control=CacheControl(type="ephemeral"),
                )
            ]
        ),
    ]
)

llm = LLMModelDetails(name="anthropic--claude-4.5-sonnet")

prompt_template = PromptTemplatingModuleConfig(prompt=template, model=llm)
module_config = ModuleConfig(prompt_templating=prompt_template)
config = OrchestrationConfig(modules=module_config)

orchestration_service = OrchestrationService(config=config)

result = orchestration_service.run(
    placeholder_values={
        "article": "Scaling up neural models has yielded significant advancements in language generation"
    }
)
print(result.final_result.choices[0].message.content)

Inspecting Cache Token Usage

For supported models the prompt_tokens_details field on the usage object reports how many tokens were written to the cache (cache_creation_tokens) and how many were read from it (cached_tokens). On a warm cache at least one of the two values will be greater than zero.

usage = result.final_result.usage
details = usage.prompt_tokens_details
print(f"cache_creation_tokens : {details.cache_creation_tokens}")
print(f"cached_tokens         : {details.cached_tokens}")

Attaching cache_control to a Tool Definition

Some models also support caching tool definitions. Pass cache_control directly to the FunctionTool constructor.

from gen_ai_hub.orchestration_v2 import FunctionTool, FunctionObject, CacheControl

def get_weather(location: str) -> str:
    """Get current temperature for a given location."""
    # Replace with your actual implementation
    return "22°C"

def get_climate_report(location: str, start_date: str, end_date: str, **kwargs) -> str:
    """Get a verbose climate report for a given location and time interval"""
    # Replace with your actual implementation
    return "{}"

weather_tool = FunctionTool(
    function=FunctionObject(
        name="get_weather",
        description="Get the current temperature for a given location.",
        parameters={
            "type": "object",
            "properties": {
                "location": {"type": "string", "description": "City and country e.g. Bogotá, Colombia"},
            },
            "required": ["location"],
            "additionalProperties": False,
        },
        strict=True,
        function=get_weather,
    ),
)

# Verbose tool definition, pushing total schema past Anthropic's 1024-token cache minimum.
# cache_control goes on the last tool in the array, caching the entire tools prefix.
climate_tool = FunctionTool(
    function=FunctionObject(
        name="get_climate_report",
        description=(
            "Retrieves a comprehensive climate and environmental report for a specified location "
            "and date range. The report covers historical weather patterns, seasonal temperature "
            "trends, air quality indices (AQI), UV exposure levels, precipitation forecasts, "
            "relative humidity statistics, wind speed and direction analysis, atmospheric pressure "
            "readings, pollen counts by category (tree, grass, weed), and general climate "
            "advisories issued by local meteorological authorities. Intended for environmental "
            "researchers, travel planners, outdoor event organizers, agricultural analysts, and "
            "health-conscious users requiring detailed atmospheric data beyond simple temperature."
        ) * 6,
        parameters={
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "Target location as city and country e.g. Berlin, Germany.",
                },
                "start_date": {
                    "type": "string",
                    "description": "Start of the reporting period in ISO 8601 format (YYYY-MM-DD).",
                },
                "end_date": {
                    "type": "string",
                    "description": "End of the reporting period in ISO 8601 format (YYYY-MM-DD).",
                },
                "metrics": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": (
                        "Subset of metrics to include. Accepted values: temperature, humidity, "
                        "precipitation, wind_speed, wind_direction, aqi, uv_index, pressure, "
                        "pollen_tree, pollen_grass, pollen_weed. Omit to retrieve all metrics."
                    ),
                },
                "units": {
                    "type": "string",
                    "enum": ["metric", "imperial"],
                    "description": "Unit system for numerical values. Defaults to metric.",
                },
                "interval": {
                    "type": "string",
                    "enum": ["hourly", "daily", "weekly"],
                    "description": "Temporal resolution of the returned data series. Defaults to daily.",
                },
                "include_advisories": {
                    "type": "boolean",
                    "description": (
                        "When true, appends any active climate or health advisories issued for "
                        "the location within the requested date range."
                    ),
                },
                "language": {
                    "type": "string",
                    "description": "BCP 47 language tag for advisory text e.g. en, de, es. Defaults to en.",
                },
            },
            "required": ["location", "start_date", "end_date"],
            "additionalProperties": False,
        },
        strict=False,
        function=get_climate_report,
    ),
    cache_control=CacheControl(type="ephemeral"),
)

Use the cached tools in a template and verify that the second run hits the cache:

from gen_ai_hub.orchestration_v2 import Template, SystemMessage, UserMessage

template = Template(
    template=[
        SystemMessage(content="You are a weather assistant."),
        UserMessage(content="What is the temperature in {{?location}}?"),
    ],
    tools=[weather_tool, climate_tool],
)

config = OrchestrationConfig(
    modules=ModuleConfig(
        prompt_templating=PromptTemplatingModuleConfig(
            prompt=template,
            model=LLMModelDetails(name="anthropic--claude-4.6-sonnet", params={"max_tokens": 50}),
        )
    )
)

service = OrchestrationService()
placeholder_values = {"location": "Bogotá, Colombia"}

# Run 1: writes the cache (cache_creation_tokens > 0, cached_tokens = 0)
response1 = service.run(config=config, placeholder_values=placeholder_values)
u1 = response1.final_result.usage.prompt_tokens_details
print(f"Run 1 — cache_creation_tokens: {u1.cache_creation_tokens}, cached_tokens: {u1.cached_tokens}")

# Run 2: hits the cache (cache_creation_tokens = 0, cached_tokens > 0)
response2 = service.run(config=config, placeholder_values=placeholder_values)
u2 = response2.final_result.usage.prompt_tokens_details
print(f"Run 2 — cache_creation_tokens: {u2.cache_creation_tokens}, cached_tokens: {u2.cached_tokens}")

Using Images as Input

The Orchestration Service supports multimodal prompts, enabling you to include images alongside text in your messages. This powerful feature unlocks a variety of applications, such as visual question answering (VQA), image captioning, object recognition, and generating text creatively based on visual input.

This guide details how to prepare image inputs, integrate them into your prompts, and execute the orchestration to get insightful responses.

1. Preparing Image Inputs

To use an image, you first need to represent it as an ImageItem object. The ImageItem class provides two convenient ways to do this.

a) From a URL or Data URL

This method is ideal for images hosted online or when you have the image data encoded as a Data URL (base64 encoded).

• Standard URL: Provide a direct web link to the image file.
• Data URL: Provide the image data directly embedded in the URL string.

Note: For web URLs, ensure the image is publicly accessible, as the service will need to fetch it.

from gen_ai_hub.orchestration_v2 import ImageItem

# From URL
image_from_web = ImageItem(url="https://picsum.photos/id/1/200/300")

# From Data URL
image_from_data_url = ImageItem(
    url="data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAoAAAAKCAIAAAACUFjqAAAAE0lEQVR4nGP8z4APMOGVZRip0gBBLAETee26JgAAAABJRU5ErkJggg=="
)

b) From a Local File

If your image resides on your local filesystem, you can load it directly using the ImageItem.from_file() class method. The from_file method handles opening, reading, and base64 encoding the image data for you, packaging it into an ImageItem.

# From local file
try:
    image_from_local_file = ImageItem.from_file("path/to/your/local/image.jpeg")
except FileNotFoundError:
    print("Error: The specified image file was not found.")
except Exception as e:
    print(f"An error occurred while loading the image: {e}")

2. Adding Images to a Prompt

Once you have your ImageItem object(s), you can combine them with text to create a multimodal prompt. This is done by passing a list containing ImageItem instances and text strings to the content parameter of a UserMessage.

from gen_ai_hub.orchestration_v2 import (
    UserMessage, LLMModelDetails, Template, PromptTemplatingModuleConfig,
    ModuleConfig, OrchestrationConfig, OrchestrationService
)

llm=LLMModelDetails(name="gpt-4o")

# Simple visual question answering
content_vqa = [image_from_web, "What objects are prominent in this image?"]

# Create a UserMessage with the mixed content
user_message = UserMessage(content=content_vqa)
image_template = Template(template=[user_message])

# Create a Template containing the UserMessage
prompt_template = PromptTemplatingModuleConfig(prompt=image_template, model=llm)

module_config = ModuleConfig(prompt_templating=prompt_template)

config = OrchestrationConfig(modules=module_config)
service = OrchestrationService(config=config)
response = service.run()
print(response.final_result.choices[0].message.content)

Translation

The Translation module can translate text from one language to another. It can translate input text before it is processed by the LLM module, or translate output generated by the LLM module.

from gen_ai_hub.orchestration_v2 import (
    Template, SystemMessage, UserMessage, LLMModelDetails, PromptTemplatingModuleConfig,
    ModuleConfig, OrchestrationConfig, OrchestrationService,
    TranslationModuleConfig, SAPDocumentTranslationInput, SAPDocumentTranslationOutput,
    InputTranslationConfig, OutputTranslationConfig
)

template = Template(template=[
    SystemMessage(content="You are a helpful assistant."),
    UserMessage(content="{{?text}}"),
])
prompt_template = PromptTemplatingModuleConfig(prompt=template, model=LLMModelDetails(name="gpt-4o"))

translation_config = TranslationModuleConfig(
    input=SAPDocumentTranslationInput(config=InputTranslationConfig(source_language="en-US", target_language="de-DE")),
    output=SAPDocumentTranslationOutput(config=OutputTranslationConfig(source_language="de-DE", target_language="fr-FR"))
)

template = Template(template=[
    SystemMessage(content="You are a helpful assistant."),
    UserMessage(content="{{?text}}"),
])
llm = LLMModelDetails(name="gpt-4o")
prompt_template = PromptTemplatingModuleConfig(prompt=template, model=llm)

module_config = ModuleConfig(prompt_templating=prompt_template, translation=translation_config)
config = OrchestrationConfig(modules=module_config)

result = OrchestrationService().run(config=config, placeholder_values={"text": "What is the capital of Germany?"})
print(result.final_result.choices[0].message.content)

Embeddings

The Orchestration Service provides an embeddings endpoint for generating vector representations of text. Embeddings capture the semantic meaning of text, enabling powerful applications like semantic search, document clustering, and retrieval-augmented generation (RAG).

Key Use Cases:

• Semantic Search: Find documents based on meaning, not just keywords
• RAG (Retrieval-Augmented Generation): Retrieve relevant context for LLM prompts
• Document Clustering: Group similar documents together
• Similarity Comparison: Measure how semantically similar two texts are

Basic Usage

Generate an embedding for a single text string with minimal configuration.

from gen_ai_hub.orchestration_v2 import (
    OrchestrationService, EmbeddingsOrchestrationConfig, EmbeddingsModuleConfigs,
    EmbeddingsModelConfig, EmbeddingsModelDetails, EmbeddingsInput
)

service = OrchestrationService()
embeddings_config = EmbeddingsOrchestrationConfig(
    modules=EmbeddingsModuleConfigs(
        embeddings=EmbeddingsModelConfig(model=EmbeddingsModelDetails(name="text-embedding-3-large"))
    )
)
response = service.embed(config=embeddings_config, input=EmbeddingsInput(text="Hello World!"))
embedding = response.final_result.data[0].embedding
print(f"Embedding dimensions: {len(embedding)}")
print(f"First 5 values: {embedding[:5]}")

Customizing Embedding Parameters

You can customize the embedding output with parameters like dimensions, encoding_format, and normalize.

Parameter	Description	Values
dimensions	Number of dimensions in the output	e.g. 256, 512, 1536, 3072
encoding_format	Output format	FLOAT, BASE64, BINARY
normalize	Normalize the vector	True, False

from gen_ai_hub.orchestration_v2 import (
    EmbeddingsOrchestrationConfig, EmbeddingsModuleConfigs, EmbeddingsModelConfig,
    EmbeddingsModelDetails, EmbeddingsInput, EmbeddingsModelParams, EmbeddingsEncodingFormat
)

embeddings_config_custom = EmbeddingsOrchestrationConfig(
    modules=EmbeddingsModuleConfigs(
        embeddings=EmbeddingsModelConfig(
            model=EmbeddingsModelDetails(
                name="text-embedding-3-large",
                params=EmbeddingsModelParams(dimensions=256, encoding_format=EmbeddingsEncodingFormat.FLOAT, normalize=True)
            )
        )
    )
)
response = service.embed(config=embeddings_config_custom, input=EmbeddingsInput(text="Hello World!"))
print(f"Embedding dimensions: {len(response.final_result.data[0].embedding)}")

Batch Embeddings

Generate embeddings for multiple texts in a single request for better efficiency.

documents = [
    "Artificial intelligence is transforming industries worldwide.",
    "Machine learning models require large amounts of training data.",
    "Neural networks are inspired by the human brain structure.",
    "Deep learning has achieved breakthroughs in image recognition."
]
response = service.embed(config=embeddings_config, input=EmbeddingsInput(text=documents))
print(f"Generated {len(response.final_result.data)} embeddings")
for result in response.final_result.data:
    print(f"  Index {result.index}: {len(result.embedding)} dimensions")

Input Type Hints (Asymmetric Search)

Some embedding models support asymmetric search, where queries and documents are embedded differently for better retrieval. Use the type parameter to hint the purpose of your text.

Type	Use Case
TEXT	General purpose (default)
DOCUMENT	Content to be indexed and searched
QUERY	Search queries to find relevant documents

from gen_ai_hub.orchestration_v2 import EmbeddingsInput, EmbeddingsInputType

# Embed a document for storage in a vector database
doc_response = service.embed(
    config=embeddings_config,
    input=EmbeddingsInput(
        text="SAP is a German multinational software company that develops enterprise software.",
        type=EmbeddingsInputType.DOCUMENT
    )
)

# Embed a query for searching
query_response = service.embed(
    config=embeddings_config,
    input=EmbeddingsInput(
        text="What is SAP?",
        type=EmbeddingsInputType.QUERY
    )
)

Embeddings with Data Masking

When embedding sensitive data, use the data masking module to anonymize PII before generating embeddings. This ensures sensitive information is not exposed to the embedding model.

from gen_ai_hub.orchestration_v2 import (
    OrchestrationService, EmbeddingsOrchestrationConfig, EmbeddingsModuleConfigs,
    EmbeddingsModelConfig, EmbeddingsModelDetails, EmbeddingsInput,
    MaskingModuleConfig, MaskingMethod, MaskingProviderConfig, DPIStandardEntity, ProfileEntity
)

service = OrchestrationService()
embeddings_config_with_masking = EmbeddingsOrchestrationConfig(
    modules=EmbeddingsModuleConfigs(
        embeddings=EmbeddingsModelConfig(
            model=EmbeddingsModelDetails(name="text-embedding-3-large")
        ),
        masking=MaskingModuleConfig(
            providers=[
                MaskingProviderConfig(
                    method=MaskingMethod.ANONYMIZATION,
                    entities=[
                        DPIStandardEntity(type=ProfileEntity.PERSON),
                        DPIStandardEntity(type=ProfileEntity.EMAIL),
                        DPIStandardEntity(type=ProfileEntity.PHONE),
                    ]
                )
            ]
        )
    )
)

response = service.embed(
    config=embeddings_config_with_masking,
    input=EmbeddingsInput(
        text="Contact John Smith at john.smith@example.com or call 555-123-4567."
    )
)

print(f"Embedding generated with PII masked")
print(f"Intermediate results: {response.intermediate_results}")
print(f"Dimensions: {len(response.final_result.data[0].embedding)}")

Masking with Custom Entities and Allowlist

Use regular expressions to mask custom patterns and allowlists to exclude specific terms from masking.

from gen_ai_hub.orchestration_v2 import (
    OrchestrationService, EmbeddingsOrchestrationConfig, EmbeddingsModuleConfigs,
    EmbeddingsModelConfig, EmbeddingsModelDetails, EmbeddingsInput,
    MaskingModuleConfig, MaskingMethod, MaskingProviderConfig, DPIStandardEntity, ProfileEntity,
    DPICustomEntity, DPIMethodConstant
)

service = OrchestrationService()
embeddings_config_advanced_masking = EmbeddingsOrchestrationConfig(
    modules=EmbeddingsModuleConfigs(
        embeddings=EmbeddingsModelConfig(
            model=EmbeddingsModelDetails(name="text-embedding-3-large")
        ),
        masking=MaskingModuleConfig(
            providers=[
                MaskingProviderConfig(
                    method=MaskingMethod.ANONYMIZATION,
                    entities=[
                        DPIStandardEntity(type=ProfileEntity.PERSON),
                        DPIStandardEntity(type=ProfileEntity.ORG),
                        # Custom pattern for internal IDs like "89-SAP-550"
                        DPICustomEntity(
                            regex=r"\b[0-9]{2}-SAP-[0-9]{3}\b",
                            replacement_strategy=DPIMethodConstant(
                                method="constant",
                                value="REDACTED_ID"
                            )
                        ),
                    ],
                    # These terms will NOT be masked
                    allowlist=["SAP", "Microsoft"]
                )
            ]
        )
    )
)

response = service.embed(
    config=embeddings_config_advanced_masking,
    input=EmbeddingsInput(
        text="Employee John Doe (ID: 89-SAP-550) works at SAP with Microsoft partners."
    )
)

Async Embeddings

For non-blocking operations, use the async aembed method.

from gen_ai_hub.orchestration_v2 import OrchestrationService, EmbeddingsInput

async def embed_async():
    async_service = OrchestrationService()
    response = await async_service.aembed(config=embeddings_config, input=EmbeddingsInput(text="Hello async world!"))
    print(f"Async embedding dimensions: {len(response.final_result.data[0].embedding)}")
    await async_service.aclose_http_connection()

await embed_async()

Advanced Examples

Translation Service

This example extends the initial walkthrough of a basic orchestration pipeline by abstracting the translation task into its own reusable TranslationService class. Once the configuration is established, it can be easily adapted and reused for different translation scenarios.

from gen_ai_hub.orchestration_v2 import (
    OrchestrationConfig, ModuleConfig, LLMModelDetails, SystemMessage, UserMessage,
    Template, PromptTemplatingModuleConfig, OrchestrationService
)


class TranslationService:
    def __init__(self, orchestration_service: OrchestrationService):
        self.service = orchestration_service
        self.template = Template(
            template=[
                SystemMessage(content="You are a helpful AI assistant."),
                UserMessage(content="Translate the following text to {{?to_lang}}: {{?text}}"),
            ],
            defaults={"to_lang": "en-US"}
        )
        self.llm = LLMModelDetails(name="gpt-4o")
        self.config = OrchestrationConfig(modules=ModuleConfig(
            prompt_templating=PromptTemplatingModuleConfig(prompt=self.template, model=self.llm)
        ))

    def translate(self, text, to_lang):
        response = self.service.run(config=self.config, placeholder_values={"to_lang": to_lang, "text": text})
        return response.final_result.choices[0].message.content


service = OrchestrationService()
translator = TranslationService(orchestration_service=service)
print(translator.translate(text="Hello, world!", to_lang="French"))
print(translator.translate(text="Hello, world!", to_lang="Spanish"))
print(translator.translate(text="Hello, world!", to_lang="German"))

Chatbot with Memory

This example demonstrates how to integrate the OrchestrationService with a chatbot to handle conversational flow.

When making requests to the orchestration service, you can specify a list of messages as history that will be prepended to the templated content and processed by the templating module. These messages are plain, non-templated messages, as they typically represent past conversation outputs — such as in this chatbot scenario.

It's important to note that managing conversation history / state is handled locally in the ChatBot class, not by the orchestration service itself.

from typing import List
from gen_ai_hub.orchestration_v2 import (
    OrchestrationConfig, ModuleConfig, LLMModelDetails, ChatMessage, SystemMessage,
    UserMessage, Template, PromptTemplatingModuleConfig, OrchestrationService
)


class ChatBot:
    def __init__(self, orchestration_service: OrchestrationService):
        self.service = orchestration_service
        self.template = Template(template=[
            SystemMessage(content="You are a helpful chatbot assistant."),
            UserMessage(content="{{?user_query}}")
        ])
        self.llm = LLMModelDetails(name="gpt-4o")
        self.config = OrchestrationConfig(modules=ModuleConfig(
            prompt_templating=PromptTemplatingModuleConfig(prompt=self.template, model=self.llm)
        ))
        self.history: List[ChatMessage] = []

    def chat(self, user_input):
        response = self.service.run(
            config=self.config,
            placeholder_values={"user_query": user_input},
            history=self.history,
        )
        message = response.final_result.choices[0].message
        self.history = response.intermediate_results.templating
        self.history.append(message)
        return message.content

    def reset(self):
        self.history = []


bot = ChatBot(orchestration_service=OrchestrationService())
print(bot.chat("Hello, how are you?"))
print(bot.chat("What's the weather like today?"))
print(bot.chat("Can you remember what I first asked you?"))
bot.reset()

Sentiment Analysis with Few-Shot Learning

This example demonstrates the different message roles in the templating module through a few-shot learning use case with the FewShotLearner class.

• Message Types: Different message types (SystemMessage, UserMessage, AssistantMessage) structure the interaction and guide the model's behavior.
• Templating: The template includes these examples, ending with a placeholder ({{?user_input}}) for dynamic user input.
• Few-Shot Examples: Pairs of UserMessage and AssistantMessage show how the model should respond to similar queries.

The FewShotLearner class manages the dynamic creation of the template and ensures the correct message roles are used for each user input.

from gen_ai_hub.orchestration_v2 import (
    OrchestrationConfig, ModuleConfig, LLMModelDetails, SystemMessage, UserMessage,
    AssistantMessage, Template, PromptTemplatingModuleConfig, OrchestrationService
)
class FewShotLearner:
    def __init__(self, orchestration_service, system_message, examples):
        self.service = orchestration_service
        self.llm = LLMModelDetails(name="gpt-4o-mini")
        self.config = OrchestrationConfig(modules=ModuleConfig(
            prompt_templating=PromptTemplatingModuleConfig(
                prompt=self._create_few_shot_template(system_message, examples),
                model=self.llm
            )
        ))

    @staticmethod
    def _create_few_shot_template(system_message, examples):
        messages = [system_message]
        for example in examples:
            messages.append(example[0])
            messages.append(example[1])
        messages.append(UserMessage(content="{{?user_input}}"))
        return Template(template=messages)

    def predict(self, user_input):
        response = self.service.run(config=self.config, placeholder_values={"user_input": user_input})
        return response.final_result.choices[0].message.content


sentiment_examples = [
    (UserMessage(content="I love this product!"), AssistantMessage(content="Positive")),
    (UserMessage(content="This is terrible service."), AssistantMessage(content="Negative")),
    (UserMessage(content="The weather is okay today."), AssistantMessage(content="Neutral")),
]
sentiment_analyzer = FewShotLearner(
    orchestration_service=OrchestrationService(),
    system_message=SystemMessage(
        content="You are a sentiment analysis assistant. Classify the sentiment as Positive, Negative, or Neutral."
    ),
    examples=sentiment_examples,
)
print(sentiment_analyzer.predict("The movie was a complete waste of time!"))
print(sentiment_analyzer.predict("The traffic was fortunately unusually light today."))
print(sentiment_analyzer.predict("I'm not sure how I feel about the recent events."))

Async Support

The OrchestrationService also supports asynchronous calls. Use:

• arun from the async version of run
• astream from the async version of stream

from gen_ai_hub.orchestration_v2 import (
    SystemMessage, UserMessage, Template, PromptTemplatingModuleConfig,
    LLMModelDetails, OrchestrationConfig, ModuleConfig, OrchestrationService
)

template = Template(template=[
    SystemMessage(content="This is a system message."),
    UserMessage(content="Write a markdown cheatsheet!"),
])
llm = LLMModelDetails(name="gemini-3.5-flash")
module_config = ModuleConfig(
    prompt_templating=PromptTemplatingModuleConfig(prompt=template, model=llm)
)
config = OrchestrationConfig(modules=module_config)
orchestration_service = OrchestrationService(config=config)

async def test_async():
    result = await orchestration_service.arun()
    print(result.final_result.choices[0].message.content)

await test_async()

Async streaming:

from gen_ai_hub.orchestration_v2 import GlobalStreamOptions

config_stream = OrchestrationConfig(modules=module_config, stream=GlobalStreamOptions(enabled=True))

async def test_streaming_async():
    streamed_content = ""
    async for chunk in await orchestration_service.astream(config=config_stream):
        streamed_content += chunk.final_result.choices[0].delta.content
    print(streamed_content)

await test_streaming_async()