Zenn
株式会社マインディア テックブログPublicationへの投稿
❄️

RAGをSnowflake CortexとStreamlitで構築する

ShintaroAmaike
2024/07/08に公開
Streamlit
Snowflake
cortex
LangChain
RAG
tech

生成AIの精度と信頼性を高める技術として注目されているRAG（Retrieval Augmented Generation）
本記事では、Snowflakeを活用してRAGシステムを構築する方法を、PDFドキュメントの取り込みからチャットUIの作成まで、実践的に解説します。
CortexとStreamlitを組み合わせることで、比較的シンプルにRAGを実現できる方法をご紹介します。

3行まとめ

  • Snowflakeを使ってRAG（Retrieval Augmented Generation）システムを構築する方法を紹介
  • CortexとStreamlitを組み合わせることで、比較的シンプルにRAGを実現可能
  • 日本語対応モデルの選択、コスト計算、PDFの取り込みからチャットUIの構築までのプロセスを解説

RAGとは

検索拡張生成（Retrieval Augmented Generation、RAG）があり、これはLLMを文書検索を使用して拡張するもので、ときにはベクトルデータベース（英語版）を使うこともある。クエリが与えられると、文書検索ツールが呼び出され、もっとも関連性が高い文書が取得される（通常、初めにクエリと文書をベクトルで符号化し、次にクエリベクトルにユークリッドノルムで最も近いベクトルを持つ文書を検索する）。その後、LLMは、クエリと取得した文書の両方に基づいて出力を生成する。[1]

RAGによるハルシネーション対策

生成AIの大きな課題の一つに、「ハルシネーション」と呼ばれる現象があります。
これは、AIが事実に基づかない情報を自信を持って提示してしまうリスクを指します。
この問題は、特に重要な意思決定や正確性が求められる場面で深刻な影響を及ぼす可能性があります。
RAG（Retrieval Augmented Generation）は、このハルシネーションリスクを軽減する有力な手法として注目されています。

RAGの仕組みは以下の通りです：

  • 信頼性の高い文書データベースを事前に用意する
  • ユーザーの質問に関連する情報をこのデータベースから検索する
  • 検索結果を参照しながら、AIが回答を生成する

この方法により、AIの回答は信頼できる情報源に基づいたものとなり、根拠のない情報の生成リスクを大幅に低減できます。
結果として、より信頼性の高い、事実に基づいたAI対話システムの実現が期待されています。

[2]

Snowflakeの利点

Cortex ML が特定リージョンで使えるようになった。[3]

RAGシステムの構築には様々なアプローチがありますが、Snowflakeを活用することで特有の利点が得られます
Snowflakeを活用したRAG構築は、統合環境、スケーラビリティ、セキュリティ、簡易な実装、コスト効率、豊富なAIモデル選択という多面的な利点を提供します。
CortexとStreamlitの組み合わせにより、複雑なインフラ構築なしに比較的容易にRAGシステムを実装でき、使用量ベースの課金モデルでコスト最適化も図れます。
さらに、多様なAIモデルへのアクセスで、用途に最適なモデル選択が可能です。

※2024/7/17現在awsのAsia Pacificで試しています。

対応モデル

モデル名 Asia Pacific(Tokyo)での利用可否 日本語対応
snowflake-arctic NG -
mistral-large NG -
reka-flash OK OK
reka-core NG -
mixtral-8x7b OK OK
llama2-70b-chat NG -
llama3-8b OK NG
llama3-70b OK NG
mistral-7b OK OK

※2024/07/17現在、aws Asia Pacific(Tokyo)

利用できたのはmixtral-8x7b、llama3-8b、llama3-70b、mistral-7b、
reka-flashでそれ以外は400エラーが返却されました。
日本語が利用できたのはmixtral-8x7b、mistral-7b、reka-flashでした。

100351 (P0000): Request failed for external function COMPLETE$V2 with remote service error: 400 '"unknown model \"snowflake-arctic\""

日本語対応

mistral-7bの例

SELECT SNOWFLAKE.CORTEX.COMPLETE('mistral-7b', '日本語に対応しているモデルですか？');

はい、日本語に対応しています。

(Yes, I can handle Japanese.)

この回答は英語ですが、日本語でも同じ意味の答えを与えることができます。

(This answer is in English, but I can give the same meaning answer in Japanese.)

例: 「What is your favorite color?」「私のお気に入りの色は、紺色です。」(Watashi no okiniiri no iro wa, aoi-iro desu.)

(What is your favorite color? "My favorite color is blue.")

llama3-8bに「日本語に対応していますか？」と尋ねてみた例

🙋‍♂️

Yes, I am a multilingual model, and I have been trained on a significant amount of Japanese text data. I can understand and respond to Japanese input to some extent. However, please note that my proficiency in Japanese is not as high as it is in English, and I may make mistakes or not fully understand certain nuances of the language.

...以下略

Yesとは回答されましたが、英文で回答されました。

reka-flashの例

はい、私は日本語に対応しているモデルです。お気軽に日本語で質問や会話をしていただけます。どうぞお待ちしております！

どのモデルよりもシンプルに日本語のみで回答されました。

費用

https://www.snowflake.com/legal-files/CreditConsumptionTable.pdf#page=9

Snowflake Cortex Functions

COMPLETEは利用するモデルに応じて費用が異なり、TRANSLATE、SUMMARIZE、EXTRACT_ANSWER、SENTIMENT関数はそれぞれ費用が異なります。
また、埋め込みベクトルを取得する関数はEMBED_TEXT_768、EMBED_TEXT_1024でそれぞれモデルに応じて費用が異なります。

モデルによっては非常に高額になる可能性もあるため、一旦トークン数を取得し、コスト感を計算してから実行すると良いでしょう。

COUNT_TOKENS

SELECT SNOWFLAKE.CORTEX.COUNT_TOKENS( 'snowflake-arctic', 'what is a large language model?' );

6

英語は単語ごとに1トークンとして計算しますが、

日本語は１文字ごとにトークンカウントされるようです。
モデルの受け入れ上限より少なめにすると良さそうです。

SELECT SNOWFLAKE.CORTEX.COUNT_TOKENS( 'snowflake-arctic', '日本語のトークン数のカウントテスト' );

19

費用計算の例

SELECT SNOWFLAKE.CORTEX.COUNT_TOKENS( 'mistral-7b', text ) * 0.12 / 1000000;

0.00000228 credit.

費用計算をするUDFを作成すると良さそうです。

RAGの構築

続いてSnowflakeでRAGを構築していきます。

権限の付与

Cortex の必要な権限

各Cortexを利用するためには権限が必要ですので、先に付与します。

USE ROLE ACCOUNTADMIN;

CREATE ROLE cortex_user_role;
GRANT DATABASE ROLE SNOWFLAKE.CORTEX_USER TO ROLE cortex_user_role;

GRANT ROLE cortex_user_role TO USER some_user;

既に権限がある場合は特定のRoleに付与することもできます。

GRANT DATABASE ROLE SNOWFLAKE.CORTEX_USER TO ROLE analyst;

ファイルのダウンロード

テキスト生成AI利活用におけるリスクへの対策ガイドブック
こちらのページにアクセスし、資料のダウンロードから、pdfをダウンロードします。

ファイル名：20240530_resources_generalitve-ai-guidebook_01.pdf

作業用データベースの作成

まずは一時作業用のデータベースとスキーマを作成しておきます。

CREATE DATABASE DIGITAL_AGENCY_CORTEX_DOCS;
CREATE SCHEMA DATA;

PDFファイルをLangchainで読み取りチャンクテーブルにする

create or replace function pdf_text_chunker(file_url string)
returns table (chunk varchar)
language python
runtime_version = '3.9'
handler = 'pdf_text_chunker'
packages = ('snowflake-snowpark-python','PyPDF2', 'langchain')
as
$$
from snowflake.snowpark.types import StringType, StructField, StructType
from langchain.text_splitter import RecursiveCharacterTextSplitter
from snowflake.snowpark.files import SnowflakeFile
import PyPDF2, io
import logging
import pandas as pd

class pdf_text_chunker:

    def read_pdf(self, file_url: str) -> str:
    
        logger = logging.getLogger("udf_logger")
        logger.info(f"Opening file {file_url}")
    
        with SnowflakeFile.open(file_url, 'rb') as f:
            buffer = io.BytesIO(f.readall())
            
        reader = PyPDF2.PdfReader(buffer)   
        text = ""
        for page in reader.pages:
            try:
                text += page.extract_text().replace('\n', ' ').replace('\0', ' ')
            except:
                text = "Unable to Extract"
                logger.warn(f"Unable to extract from file {file_url}, page {page}")
        
        return text

    def process(self,file_url: str):

        text = self.read_pdf(file_url)
        
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size = 4000, #Adjust this as you see fit
            chunk_overlap  = 400, #This let's text have some form of overlap. Useful for keeping chunks contextual
            length_function = len
        )
    
        chunks = text_splitter.split_text(text)
        df = pd.DataFrame(chunks, columns=['chunks'])
        
        yield from df.itertuples(index=False, name=None)
$$;

stageを作成する

PDFをアップロードするstageを作成します。

create or replace stage docs ENCRYPTION = (TYPE = 'SNOWFLAKE_SSE') DIRECTORY = ( ENABLE = true );

ダウンロードしたpdfをアップロードします。

ls @docs;

stageの内容を確認します。

アップロードされました。

ベクトルストア(Vector Store)の作成

[4]

次に、pdfドキュメントをテキストエンべディングし、VECTOR型のデータタイプに変換します。
これにより、ドキュメントのテキストデータが数値ベクトルとしてテーブルに保存され、
ベクトル検索を行うことによって意味的に類似している情報を探せるようになります。

create or replace TABLE DOCS_CHUNKS_TABLE ( 
    RELATIVE_PATH VARCHAR(16777216), -- PDFファイルへの相対パス
    SIZE NUMBER(38,0), -- PDFファイルのサイズ
    FILE_URL VARCHAR(16777216), -- PDFのURL
    SCOPED_FILE_URL VARCHAR(16777216), -- スコープ付きURL
    CHUNK VARCHAR(16777216), -- テキストの一部
    CHUNK_VEC VECTOR(FLOAT, 768) );  -- VECTORデータタイプの埋め込み

PDFファイルをチャンクテーブルにインサート

insert into docs_chunks_table (relative_path, size, file_url,
                            scoped_file_url, chunk, chunk_vec)
    select relative_path, 
            size,
            file_url, 
            build_scoped_file_url(@docs, relative_path) as scoped_file_url,
            func.chunk as chunk,
            SNOWFLAKE.CORTEX.EMBED_TEXT_768('e5-base-v2',chunk) as chunk_vec
    from 
        directory(@docs),
        TABLE(pdf_text_chunker(build_scoped_file_url(@docs, relative_path))) as func;

先に作成した関数pdf_text_chunkerを使用してチャンクを抽出し、埋め込みを作成します。
先のテーブルdocs_chunks_tableに情報をインサートします。

SNOWFLAKE.CORTEX.EMBED_TEXT_768('e5-base-v2',chunk) as chunk_vec

ここのコードでは、ドキュメントから読み込まれたテキストの一部であるチャンクをCortexに渡し
テキストエンべディングを作成しています。

select relative_path, size, chunk, chunk_vec from docs_chunks_table limit 5;

作成されたテーブルを確認します。

ファイルに対するチャンクの数を確認します。

select relative_path, count(*) as num_chunks 
    from docs_chunks_table
    group by relative_path;

チャットUIとチャットのロジックの構築

ベクトル埋め込み関数

SELECT SNOWFLAKE.CORTEX.EMBED_TEXT_768(model, text)

モデル名とチャンクにしたテキストを渡します。

SnowflakeではStreamlitが使えるため
Streamlitで簡単なUIを作成します。

import streamlit as st # Import python packages
from snowflake.snowpark.context import get_active_session
session = get_active_session() # Get the current credentials

import pandas as pd

pd.set_option("max_colwidth",None)
num_chunks = 3 # Num-chunks provided as context. Play with this to check how it affects your accuracy

def create_prompt (myquestion, rag):

    if rag == 1:    

        cmd = """
         with results as
         (SELECT RELATIVE_PATH,
           VECTOR_COSINE_SIMILARITY(docs_chunks_table.chunk_vec,
                    SNOWFLAKE.CORTEX.EMBED_TEXT_768('e5-base-v2', ?)) as similarity,
           chunk
         from docs_chunks_table
         order by similarity desc
         limit ?)
         select chunk, relative_path from results 
         """
    
        df_context = session.sql(cmd, params=[myquestion, num_chunks]).to_pandas()      
        
        context_lenght = len(df_context) -1

        prompt_context = ""
        for i in range (0, context_lenght):
            prompt_context += df_context._get_value(i, 'CHUNK')

        prompt_context = prompt_context.replace("'", "")
        relative_path =  df_context._get_value(0,'RELATIVE_PATH')
    
        prompt = f"""
          'You are an expert assistance extracting information from context provided. 
           Answer the question based on the context. Be concise and do not hallucinate. 
           If you don´t have the information just say so.
          Context: {prompt_context}
          Question:  
           {myquestion} 
           Answer: '
           """
        cmd2 = f"select GET_PRESIGNED_URL(@docs, '{relative_path}', 360) as URL_LINK from directory(@docs)"
        df_url_link = session.sql(cmd2).to_pandas()
        url_link = df_url_link._get_value(0,'URL_LINK')

    else:
        prompt = f"""
         'Question:  
           {myquestion} 
           Answer: '
           """
        url_link = "None"
        relative_path = "None"
        
    return prompt, url_link, relative_path

def complete(myquestion, model_name, rag = 1):

    prompt, url_link, relative_path =create_prompt (myquestion, rag)
    cmd = f"""
             select SNOWFLAKE.CORTEX.COMPLETE(?,?) as response
           """
    
    df_response = session.sql(cmd, params=[model_name, prompt]).collect()
    return df_response, url_link, relative_path

def display_response (question, model, rag=0):
    response, url_link, relative_path = complete(question, model, rag)
    res_text = response[0].RESPONSE
    st.markdown(res_text)
    if rag == 1:
        display_url = f"Link to [{relative_path}]({url_link}) that may be useful"
        st.markdown(display_url)

#Main code

st.title("Snowflake Cortex を使用して自分のドキュメントに質問する:")
st.write("あなたは質問をし、文脈のためにあなたの文書を使用するか、モデルに独自の応答を作成させるかを決定することができます。")
st.write("これは、あなたがすでに持っている書類のリストです。:")
docs_available = session.sql("ls @docs").collect()
list_docs = []
for doc in docs_available:
    list_docs.append(doc["name"])
st.dataframe(list_docs)

#Here you can choose what LLM to use. Please note that they will have different cost & performance
model = st.sidebar.selectbox('Select your model:',(
                                    'mixtral-8x7b',
                                     'mistral-7b'))

question = st.text_input("Enter question", placeholder="生成AIを利用する上でのリスクは？", label_visibility="collapsed")

rag = st.sidebar.checkbox('自分の文書をコンテキストとして使う?')

print (rag)

if rag:
    use_rag = 1
else:
    use_rag = 0

if question:
    display_response (question, model, use_rag)

本要件では日本語での回答を望んでいるため、
モデルは2つから選ぶようにしています。
※利用しているリージョンによって適宜追加してください。

プロンプトの結果例

Q:ライセンスに関するリスクは？

A:

PDFドキュメントの42ページをもとに回答を生成してくれた。

Q:生成AIを利用する上でのリスクは？

A:

ドキュメントの広範囲をソースとして回答を生成してくれた。

Q:テキスト生成AIの利活用が不適切であるケースは？

A:

ドキュメント2-1をソースとして回答を生成してくれた。

※日本語で回答してくれない場合は、プロンプトの最後に「日本語で回答してください」と付け加えると日本語で回答してくれます。

さいごに

本記事では、Snowflakeを活用してRAG（Retrieval Augmented Generation）システムを構築する方法を解説しました。
Snowflakeの強みであるCortexとStreamlitを組み合わせることで、比較的シンプルにRAGを構築することができました。

RAGシステムは、生成AIの精度と信頼性を向上させる重要な技術です。
Snowflakeを使用することで、データの管理からAIモデルの運用まで一貫して行えるため、企業のAI導入をより効率的に進められる可能性があります。
一方で、モデルの選択やコスト管理、データの準備、セキュリティ対策など、考慮すべき点も多々あります。
本記事で紹介した方法を参考に、各組織のニーズに合わせてカスタマイズし、効果的なRAGシステムを構築していただければ幸いです。
今後もSnowflakeの機能は進化し続けると予想されます。
常に最新の情報をキャッチアップし、より効率的で高度なAIシステムの構築に挑戦していくことが重要です。

引用

テキスト生成AI利活用におけるリスクへの対策ガイドブック（α版）
snowflake quickstarts

脚注

  1. 大規模言語モデル(Wikipedia) ↩︎

  2. quickstarts p1 RAG Overview より ↩︎

  3. 大規模言語モデル（LLM）関数（Snowflake Cortex）一部リージョンで利用できるようになった。(7/17現在) ↩︎

  4. quickstarts p3 Build the Vector Store より ↩︎

株式会社マインディア テックブログ
株式会社マインディア テックブログPublication

株式会社マインディアのテックブログになります。 サービスやシステム、カルチャーの話題はリンクの「採用」のスライドをご覧ください！

Discussion