extensions_call.py

Source: sunholo/vertex/extensions_call.py

Functions

dynamic_extension_call(question, config: sunholo.utils.config_class.ConfigManager, project_id: str = None, model_name: str = 'models/gemini-1.5-pro', **kwargs)

No docstring available.

get_extension_content(question: str, config: sunholo.utils.config_class.ConfigManager, project_id: str = None, **kwargs)

Fetches content from extensions based on the provided question and configuration.

Args: question (str): The question to be processed by the extensions. config (ConfigManager): The configuration manager instance. **kwargs: Additional parameters to be passed to the extension.

Returns: list: A list of responses from the extensions.

Example: Assuming a YAML configuration file as follows:

kind: vacConfig
vac:
    my_vac:
        extensions:
        - extension_id: 8524997435263549440 # or extension_display_name:
          operation_id: post_extension_invoke_one_generic
          vac: our_generic
          operation_params:
            input:
              question: ""
              chat_history: []
              source_filters: []
              source_filters_and_or: false
              search_kwargs:
                k: 0
                filter: ""
                fetch_k: 0
              private_docs: []
              whole_document: false

The function can be called as:

config = ConfigManager()
question = "What is the capital of France?"

responses = get_extension_content(
    question=question,
    config=config,
    input={
        "chat_history": [{"role": "user", "content": "Hello"}],
        "source_filters": ["PPA/"],
        "search_kwargs": {"k": 50, "filter": "source ILIKE '%GermanPolicyforPPA/%'", "fetch_k": 100}
    }
)

In this example, operation_params will be updated to:

{
    "input": {
        "question": "What is the capital of France?",
        "chat_history": [{"role": "user", "content": "Hello"}],
        "source_filters": ["PPA/"],
        "source_filters_and_or": false,
        "search_kwargs": {
            "k": 50,
            "filter": "source ILIKE '%GermanPolicyforPPA/%'",
            "fetch_k": 100
        },
        "private_docs": [],
        "whole_document": false
    }
}

parse_extension_input(question: str, extension_id: str = None, extension_display_name: str = None, config: sunholo.utils.config_class.ConfigManager = None, project_id: str = None, model_name: str = 'models/gemini-1.5-pro', **kwargs)

Takes a question and kwargs and makes an LLM call to extract parameters for an extension call. If no parameters are found, returns None Once parameters are extracted, makes the call to the extension via get_extenstion_content()

Example: Assuming an OpenAPI configuration file as follows:

extract_nested_value(data, key)

Recursively extract a value from nested dictionaries based on the specified key or a dot-separated key path. If the key is not dot-separated, it will find the first occurrence of that key in the nested dictionaries.

Args: data (dict): The dictionary to extract the value from. key (str): The key or dot-separated key path to extract the value.

Returns: Any: The extracted value, or None if the key or key path is not found.

Example:

data = {
    "output": {
        "content": "Some content",
        "metadata": {"key1": "value1"}
    }
}

value = extract_nested_value(data, "content")
# value will be "Some content"

value = extract_nested_value(data, "output.metadata")
# value will be {"key1": "value1"}

inject_question(question, params)

Recursively injects the question into nested dictionaries where the key is 'question' and the value is empty.

Args: question (str): The question to be injected. params (dict): The dictionary where the question should be injected.

Returns: dict: The dictionary with the question injected.

Example:

params = {
    "input": {
        "question": "",
        "chat_history": [],
        "source_filters": [],
        "search_kwargs": {
            "k": 0,
            "filter": "",
            "fetch_k": 0
        },
        "private_docs": [],
        "whole_document": false
    }
}

question = "What is the capital of France?"

updated_params = inject_question(question, params)

# updated_params will be:
# {
#     "input": {
#         "question": "What is the capital of France?",
#         "chat_history": [],
#         "source_filters": [],
#         "search_kwargs": {
#             "k": 0,
#             "filter": "",
#             "fetch_k": 0
#         },
#         "private_docs": [],
#         "whole_document": false
#     }
# }

update_nested_params(original, updates)

Recursively update nested dictionaries with new values.

Args: original (dict): The original dictionary to be updated. updates (dict): The new values to be merged into the original dictionary.

Returns: dict: The updated dictionary.

Example:

original = {
    "param1": "value1",
    "nested_param": {
        "sub_param1": "sub_value1"
    }
}

updates = {
    "param1": "new_value1",
    "nested_param": {
        "sub_param1": "new_sub_value1"
    }
}

updated_params = update_nested_params(original, updates)

# updated_params will be:
# {
#     "param1": "new_value1",
#     "nested_param": {
#         "sub_param1": "new_sub_value1"
#     }
# }