Snowflake CortexAIの Cortex COMPLETE Multimodalを使ってみた

CREATE OR REPLACE STAGE input_stage
    DIRECTORY = ( ENABLE = true )
    ENCRYPTION = ( TYPE = 'SNOWFLAKE_SSE' );

CREATE OR REPLACE STAGE input_stage
    URL='s3://<s3-path>/'
    CREDENTIALS=(AWS_KEY_ID=<aws_key_id>
    AWS_SECRET_KEY=<aws_secret_key>)
    ENCRYPTION=( TYPE = 'AWS_SSE_S3' );

SELECT SNOWFLAKE.CORTEX.COMPLETE('claude-3-5-sonnet',
    'Summarize the insights from this pie chart in 100 words',
    TO_FILE('@myimages', 'science-employment-slide.jpeg'));

SELECT SNOWFLAKE.CORTEX.COMPLETE('claude-3-5-sonnet',
    PROMPT('Compare this image {0} to this image {1} and describe the ideal audience for each in two concise bullets no longer than 10 words',
    TO_FILE('@myimages', 'adcreative_1.png'),
    TO_FILE('@myimages', 'adcreative_2.png')
));

def get_snowpark_session():
    connection_parameters = {
        "account": os.environ["SNOWFLAKE_ACCOUNT"],
        "user": os.environ["SNOWFLAKE_USER"],
        "password": os.environ["SNOWFLAKE_PASSWORD"]
    }
    return Session.builder.configs(connection_parameters).create()

def init_snowflake_object(session):
    db = os.environ["SNOWFLAKE_DATABASE"]
    schema = os.environ["SNOWFLAKE_SCHEMA"]
    stage = os.environ["SNOWFLAKE_STAGE"]
    wh = os.environ["SNOWFLAKE_WAREHOUSE"]
    
    sql_statements = [
        f"CREATE OR REPLACE DATABASE {db};",
        f"CREATE OR REPLACE SCHEMA {db}.{schema};",
        f"CREATE OR REPLACE WAREHOUSE {wh};",
        f"USE DATABASE {db};",
        f"USE SCHEMA {schema};",
        f"USE WAREHOUSE {wh};",
        f"""
            CREATE OR REPLACE STAGE {stage}
            DIRECTORY = ( ENABLE = true )
            ENCRYPTION = ( TYPE = 'SNOWFLAKE_SSE' );
        """
    ]
    
    for sql in sql_statements:
        session.sql(sql).collect()

put_sql = f"""
    PUT 'file://{tmp_file_path}'
    @{os.environ["SNOWFLAKE_DATABASE"]}.{os.environ["SNOWFLAKE_SCHEMA"]}.{os.environ["SNOWFLAKE_STAGE"]}
    AUTO_COMPRESS=FALSE
"""
result = session.sql(put_sql).collect()

import streamlit as st
from dotenv import load_dotenv

from snowflake.session_manager import get_snowpark_session
from snowflake.object_initializer import init_snowflake_object
from mode.single import single_image
from mode.multiple import multiple_images

def init_page():
    st.set_page_config(
        page_title="Cortex COMPLETE Multimodal Demo",
        page_icon="❄️"
    )
    st.header("Cortex COMPLETE Multimodal Demo ❄️")
    st.markdown("""
        SnowflakeのCortex COMPLETE Multimodal機能を使用して、画像処理タスクを実行します。
        			<div class="blog-card">
				<div class="blog-card-body-outer">
					<div class="blog-card-body">
																		<div class="blog-card-site-title">
							<a href="http://docs.snowflake.com">
																	<img decoding="async" loading="lazy" src="/images/favicon/favicon.ico" width="16" height="16" alt="" >
																http://docs.snowflake.com							</a>
						</div>
					</div>
				</div>
							</div>

			
    """)
    
def main():
 
    init_page()
    
    if "session" not in st.session_state:
        session = get_snowpark_session()
        init_snowflake_object(session)
        st.session_state["session"] = session
        tabs = st.tabs(["SINGLE", "MULTIPLE"])
        
        with tabs[0]:
            single_image()
        with tabs[1]:
            multiple_images()
            
if __name__ == "__main__":
    main()

user_prompt = st.text_area("プロンプトを入力してください", "100文字程度でキャプションを生成してください。")
    if st.button("実行", key="single_image"):
    try:
        with st.spinner("処理中…"):
            result = session.sql(f"""
                SELECT SNOWFLAKE.CORTEX.COMPLETE('claude-3-5-sonnet',
                '{user_prompt}',
                TO_FILE('@{os.environ["SNOWFLAKE_STAGE"]}', '{os.path.basename(tmp_file_path)}'));
            """).collect()
            st.subheader("結果")
            st.write(result[0][0])
    except Exception as e:
        st.error(f"処理中にエラーが発生しました: {e}")

user_prompt = st.text_area("プロンプトを入力してください", "2つの画像の違いを説明してください。")
    if st.button("実行", key="multiple_images"):
    try:
        with st.spinner("処理中..."):
            result = session.sql(f"""
                SELECT SNOWFLAKE.CORTEX.COMPLETE('claude-3-5-sonnet',
                PROMPT('2つの画像 {{0}} と {{1}} に対し、ユーザーの指示に従って処理してください。指示は次の通りです: {user_prompt}',                
                TO_FILE('@{os.environ["SNOWFLAKE_STAGE"]}', '{os.path.basename(file_path[0])}'),
                TO_FILE('@{os.environ["SNOWFLAKE_STAGE"]}', '{os.path.basename(file_path[1])}')));
            """).collect()
            st.subheader("結果")
            st.write(result[0][0])
    except Exception as e:
        st.error(f"処理中にエラーが発生しました: {e}")