0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

Azure 環境の RAG システム構築 ローカル開発手順~デプロイ①

Last updated at Posted at 2024-12-11

Azure へのデプロイ(本記事の続き)はこちら

Azure 環境の RAG システム構築 ローカル開発手順~デプロイ②
https://qiita.com/KUROMAGOORO/items/9f4a942e4bee651bf850

概要

  • Azure 環境でのアプリ開発~デプロイの流れをまとめました(①ではローカル環境開発まで)
  • 各サービスと接続する事を優先したので「①情報取得」でデータをいろいろな所から集めたり、「③生成結果保存」が複数発生します
  • 実施には偉い権限(所有者以上)が必要です
  • ローカル環境での接続例
    image.png

各サービスのデプロイ

  • 以下サービスをデプロイしていきます
  • ローカル環境開発から実施するので「AppService」は後の手順でデプロイします

リソースグループ

image.png

仮想ネットワーク

image.png

  • サービスエンドポイントに「Microsoft.CognitiveServices」を追加します(AzureOpenAI用)

image.png

ストレージアカウント

  • 作成した仮想ネットワークを設定します
    image.png

  • 自身の IP アドレスを許可します
    image.png

  • 自身に「ストレージBLOBデータ共同作成者」を付与します
    image.png

Azure Database for MySQL Flexible Server

image.png

  • 自身の IP アドレスを許可します
    image.png

Azure OpenAI

  • 作成した仮想ネットワークを設定します
    image.png

  • 自身の IP アドレスを許可します

  • 「信頼されたサービスリストにある Azure サービスに対し、この検索サービスにアクセスすることを許可します。」にチェックします
    image.png

  • 自身に「Cognitive Services OpenAI User」を付与します
    image.png

  • モデルをデプロイします
    image.png

Azure AI Document Intelligence

  • AISearch のインデクサー用。レイアウトスキルを使わない場合は不要
    image.png

  • 作成した仮想ネットワークに追加します

    • リージョン違いだから(?)作成時に選択出来ない
  • 自身の IP アドレスを許可します
    image.png

AISearch

image.png

データベース構築・登録

AISearch

image.png

Blob ストレージ

  • input コンテナを作成して適当なデータを配備します
    image.png

MySQL

  • DB を作って input テーブルを作って適当なデータを登録します
    image.png

アプリ構築

image.png

.env
AISEARCH_ENDPOINT = "https://XXX.search.windows.net"
AISEARCH_KEY = "KEYKEYKEY"
AISEARCH_INDEX = "vector-1733938719844"

BLOB_URL = "https://XXX.blob.core.windows.net"
CONTAINER_NAME_INPUT = "input"
CONTAINER_NAME_OUTPUT = "output"

MYSQL_HOST = "XXX.mysql.database.azure.com"
MYSQL_USER = "rrr"
MYSQL_PASSWORD = "KEYKEYKEY"
MYSQL_DATABASE = "rag"

AZURE_OPENAI_ENDPOINT = "https://XXX.openai.azure.com/"
AZURE_OPENAI_API_VERSION = "2024-07-01-preview"

requirements.txt
gradio==5.8.0
azure-search-documents==11.5.2
PyMySQL==1.1.1
azure-storage-blob==12.24.0
azure-identity==1.19.0
openai==1.57.2
python-dotenv==1.0.1
app.py
from action.rag import rag
import gradio as gr
from dotenv import load_dotenv
load_dotenv()

demo = gr.Interface(
    fn=rag,
    inputs=["text"],
    outputs=["text"],
    allow_flagging="never"
)

demo.launch()
action.rag.py
import task.retrieve as ret
import task.generate as gen
import task.register as reg

def rag(input: str):
    ret_result = ret._run(input)
    gen_result = gen._run(input, ret_result)
    reg._run(gen_result)
    return gen_result
task.retrieve.py
import os

# aisearch から取得
def aisearch(text: str):
    from azure.search.documents import SearchClient
    from azure.search.documents.models import QueryType, VectorizableTextQuery
    from azure.core.credentials import AzureKeyCredential
    credential = AzureKeyCredential(os.getenv("AISEARCH_KEY"))
    search_client = SearchClient(endpoint=os.getenv("AISEARCH_ENDPOINT"),
                    index_name=os.getenv("AISEARCH_INDEX"),
                    credential=credential)
    vector_query = VectorizableTextQuery(text=text, k_nearest_neighbors=50, fields="chunk_vector", exhaustive=True)
    results = search_client.search(
        search_text = text,
        query_type = QueryType.SEMANTIC,
        top = 10,
        semantic_configuration_name = "semantic-configuration",
        vector_queries = [vector_query],
        select = "title, chunk"
    )
    # 1件だけなので適当に取得する
    for result in results:
        print(f"aisearch contents: {result['chunk']}")
        return result["chunk"]

# blob から取得
def download_blob():
    from azure.identity import DefaultAzureCredential
    from azure.storage.blob import BlobServiceClient
    account_url = os.getenv("BLOB_URL")
    credential = DefaultAzureCredential()
    blob_service_client = BlobServiceClient(account_url, credential=credential)
    blob_client = blob_service_client.get_blob_client(container=os.getenv("CONTAINER_NAME_INPUT"), blob="blob_input.txt")

    downloader = blob_client.download_blob(max_concurrency=1, encoding='UTF-8')
    blob_text = downloader.readall()
    print(f"Blob contents: {blob_text}")
    return blob_text

# mysql から取得
def select_mysql():
    import pymysql.cursors
    # データベースに接続
    connection = pymysql.connect(host=os.getenv("MYSQL_HOST"),
                                user=os.getenv("MYSQL_USER"),
                                password=os.getenv("MYSQL_PASSWORD"),
                                database=os.getenv("MYSQL_DATABASE"),
                                cursorclass=pymysql.cursors.DictCursor,
                                ssl_ca='ssl/DigiCertGlobalRootCA.crt.pem')
    
    with connection:
        with connection.cursor() as cursor:
            # データ読み込み
            sql = "SELECT `input_text` FROM `input`"
            cursor.execute(sql)
            result = cursor.fetchone()
            print(f"mysql contents: {result['input_text']}")
            return result['input_text']

def _run(text: str):
    aisearch_text = aisearch(text)
    blob_text = download_blob()
    mysql_text = select_mysql()
    return f"{aisearch_text}.{blob_text}.{mysql_text}"
task.generate.py
def llm(input: str, retrieve_text: str):
    import os
    from azure.identity import DefaultAzureCredential, get_bearer_token_provider
    from openai import AzureOpenAI

    token_provider = get_bearer_token_provider(
        DefaultAzureCredential(), "https://cognitiveservices.azure.com/.default"
    )

    client = AzureOpenAI(
        api_version=os.getenv("AZURE_OPENAI_API_VERSION"),
        azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"),
        azure_ad_token_provider=token_provider,
    )
    message = f"""
# 質問
{input}
# ソース
{retrieve_text}
"""
    chat_completion = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "あなたは「ソース」の情報を使ってユーザの質問に答える親切なアシスタントです。"},
            {"role": "user", "content": message}
        ]
    )
    print(f"llm generate contents: {chat_completion.choices[0].message.content}")
    return chat_completion.choices[0].message.content

def _run(input: str, retrieve_text: str):
    generate_text = llm(input, retrieve_text)
    return generate_text
task.register.py
import os

# blob から取得
def upload_blob(text):
    from azure.identity import DefaultAzureCredential
    from azure.storage.blob import BlobServiceClient
    account_url = os.getenv("BLOB_URL")
    credential = DefaultAzureCredential()
    blob_service_client = BlobServiceClient(account_url, credential=credential)
    blob_client = blob_service_client.get_blob_client(container=os.getenv("CONTAINER_NAME_OUTPUT"), blob="blob_output.txt")

    blob_client.upload_blob(text.encode(), blob_type="BlockBlob", overwrite=True)

# mysql から取得
def insert_mysql(text: str):
    import pymysql.cursors
    # データベースに接続
    connection = pymysql.connect(host=os.getenv("MYSQL_HOST"),
                                user=os.getenv("MYSQL_USER"),
                                password=os.getenv("MYSQL_PASSWORD"),
                                database=os.getenv("MYSQL_DATABASE"),
                                cursorclass=pymysql.cursors.DictCursor,
                                ssl_ca='ssl/DigiCertGlobalRootCA.crt.pem')
    
    with connection:
        with connection.cursor() as cursor:
            # レコードを挿入
            sql = "INSERT INTO `output` (`output_text`) VALUES (%s)"
            cursor.execute(sql, (text))
        # コミットしてトランザクション実行
        connection.commit()

def _run(text: str):
    upload_blob(text)
    insert_mysql(text)

アプリの動作確認

  • 実行してメッセージを送るとデータ取得後に文言生成されて DB、Blob にデータが登録される事を確認できます
    image.png
0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?