def main()

in assets/large_language_models/rag/components/src/validate_deployments.py [0:0]

• 32 lines of code
• 31 McCabe index (conditional complexity)

def main(parser_args, activity_logger: Logger):
    """Extract main method."""
    # Determine if embeddings model is AOAI or OpenAI
    embeddings_model = parser_args.embeddings_model
    check_aoai_embeddings = (parser_args.check_embeddings == "true" or
                             parser_args.check_embeddings == "True") and embeddings_model.startswith("azure_open_ai")
    check_openai_embeddings = (parser_args.check_embeddings == "true" or
                               parser_args.check_embeddings == "True") and embeddings_model.startswith("open_ai")

    # Determine if completion model is AOAI or OpenAI
    llm_config = json.loads(parser_args.llm_config)
    model_type = llm_config.get("type")
    check_aoai_completion = (parser_args.check_completion == "true" or
                             parser_args.check_completion == "True") and model_type == "azure_open_ai"
    check_openai_completion = (parser_args.check_completion == "true" or
                               parser_args.check_completion == "True") and model_type == "open_ai"

    # Check if ACS needs to be validated
    check_acs = (parser_args.check_acs == "true" or parser_args.check_acs == "True")

    # Validate aoai models, if any
    if check_aoai_embeddings or check_aoai_completion:
        completion_to_check = "Completion model" if check_aoai_completion else ""
        embeddings_to_check = "Embeddings model" if check_aoai_embeddings else ""
        use_and = " and " if (check_aoai_embeddings and check_aoai_completion) else ""
        activity_logger.info(
            f"[Validate Deployments]: Validating {completion_to_check}{use_and}{embeddings_to_check} using AOAI")
        validate_aoai_deployments(parser_args, check_aoai_completion, check_aoai_embeddings, activity_logger)

    # validate openai models, if any
    if check_openai_completion or check_openai_embeddings:
        completion_to_check = "Completion model" if check_openai_completion else ""
        embeddings_to_check = "Embeddings model" if check_openai_embeddings else ""
        use_and = " and " if (check_openai_embeddings and check_openai_completion) else ""
        activity_logger.info(
            f"[Validate Deployments]: Validating {completion_to_check}{use_and}{embeddings_to_check} using OpenAI")
        validate_openai_deployments(parser_args, check_openai_completion, check_openai_embeddings, activity_logger)

    # validate acs connection is valid, if needed
    if check_acs:
        acs_config = json.loads(parser_args.acs_config)
        activity_logger.info(
            "[Validate Deployments]: Validating ACS config and connection for ACS index creation...")
        validate_acs(acs_config, activity_logger)