watsonx.governance日本語HAP&PII評価＆監視の備忘録

```python:templates.py
# %%
template_rag = """# 指示：
与えられた文脈にもとづいて質問に回答してください。

# 文脈：
{context}

# 質問：
{question}

# 回答：
"""
# %%

```

```python:create_detached_prompte_template.py
# %%
#!pip install ibm_aigov_facts_client setuptools
#!pip install tabulate
#!pip install python-dotenv

# %%
from dotenv import load_dotenv

load_dotenv(override=True)

# %%
import os

CPD_URL = os.environ.get("CPD_URL")
CPD_USERNAME = os.environ.get("CPD_USERNAME")
CPD_API_KEY = os.environ.get("CPD_API_KEY")
PROJECT_ID = os.environ.get("PROJECT_ID")

AZURE_OPENAI_DEPLOYMENT_NAME = os.environ.get("AZURE_OPENAI_DEPLOYMENT_NAME")
AZURE_OPENAI_ENDPOINT = os.environ.get("AZURE_OPENAI_ENDPOINT")

# %%
from ibm_aigov_facts_client import AIGovFactsClient
from ibm_aigov_facts_client import CloudPakforDataConfig

creds=CloudPakforDataConfig(
    service_url=CPD_URL,
    username=CPD_USERNAME,
    api_key=CPD_API_KEY
)

facts_client = AIGovFactsClient(
    cloud_pak_for_data_configs=creds,
    container_id=PROJECT_ID,
    container_type="project",
    disable_tracing=True
)

# %%
from ibm_aigov_facts_client import DetachedPromptTemplate, PromptTemplate

detached_information = DetachedPromptTemplate(
    prompt_id="rag-gpt-3.5-turbo",
    model_id=AZURE_OPENAI_DEPLOYMENT_NAME,
    model_provider="openai",
    model_name=AZURE_OPENAI_DEPLOYMENT_NAME,
    model_url=AZURE_OPENAI_ENDPOINT,
    prompt_url="prompt_url",
    prompt_additional_info={"model_owner": "Microsoft"}
)

# %%
from templates import template_rag

prompt_template = PromptTemplate(
    input=template_rag,
    prompt_variables={"context": "", "question": ""},
    input_prefix="",
    output_prefix="",
)

# %%
response = facts_client.assets.create_detached_prompt(
    name="rag-gpt-3.5-turbo",
    model_id=AZURE_OPENAI_DEPLOYMENT_NAME,
    task_id="retrieval_augmented_generation",
    detached_information=detached_information,
    description="detached prompt template for RAG based on Azure OpenAI GPT-3.5 turbo",
    prompt_details=prompt_template)

# %%
response

# %%

```

```python:execute_detached_prompte_template_production.py
# %%
from dotenv import load_dotenv

load_dotenv(override=True)

# %%
import os

CPD_URL = os.environ.get("CPD_URL")
CPD_USERNAME = os.environ.get("CPD_USERNAME")
CPD_API_KEY = os.environ.get("CPD_API_KEY")

PROJECT_ID = os.environ.get("PROJECT_ID")

SPACE_ID = os.environ.get("SPACE_ID")
PROMPTE_TEMPLATE_ASSET_ID = os.environ.get("PROMPT_TEMPLATE_ASSET_ID_PRODUCTION")
DEPLOYMENT_ID = os.environ.get("DEPLOYMENT_ID")

# %%
from ibm_cloud_sdk_core.authenticators import CloudPakForDataAuthenticator
from ibm_watson_openscale import *
from ibm_watson_openscale.supporting_classes.enums import *
from ibm_watson_openscale.supporting_classes import *

authenticator = CloudPakForDataAuthenticator(
    url=CPD_URL,
    username=CPD_USERNAME,
    apikey=CPD_API_KEY,
    disable_ssl_verification=True
)

wos_client = APIClient(
    service_url=CPD_URL,
    authenticator=authenticator,
)
data_mart_id = wos_client.service_instance_id
data_mart_id

# %%
wos_client.version

# %%
from ibm_watson_openscale.base_classes import ApiRequestFailure

try:
  wos_client.wos.add_instance_mapping(                
    service_instance_id=data_mart_id,
    space_id=SPACE_ID     # <- ここが大切！
  )
except ApiRequestFailure as arf:
    if arf.response.status_code == 409:
      # Instance mapping already exists
      pass
    else:
      raise arf

# %%
#!pip install ibm_metrics_plugin

# %%
language_code = "ja"

supporting_monitors = {
    "generative_ai_quality": {
        "parameters": {
            "min_sample_size": 10,
            "metrics_configuration": {
                "hap_score": {},
                "pii": {
                   "language_code" : language_code
                }
            }
        }
    }
}

# %%
response = wos_client.wos.execute_prompt_setup(
    prompt_template_asset_id=PROMPTE_TEMPLATE_ASSET_ID,
    space_id=SPACE_ID,
    deployment_id=DEPLOYMENT_ID,
	label_column="reference_text",
	operational_space_id="production",
	problem_type="retrieval_augmented_generation",
	input_data_type="unstructured_text",
    data_input_locale=[language_code],
    generated_output_locale=[language_code],
    context_fields=["context"],     
    question_field="question",
	supporting_monitors=supporting_monitors,
	background_mode=True)
response.result.to_dict()

# %%
response = wos_client.monitor_instances.mrm.get_prompt_setup(
    prompt_template_asset_id=PROMPTE_TEMPLATE_ASSET_ID,
    deployment_id=DEPLOYMENT_ID,
    space_id=SPACE_ID)
response.result.to_dict()

# %%
```

```python:store_records.py
# %%
from dotenv import load_dotenv

load_dotenv(override=True)

# %%
import os

CPD_URL = os.environ.get("CPD_URL")
CPD_USERNAME = os.environ.get("CPD_USERNAME")
CPD_API_KEY = os.environ.get("CPD_API_KEY")

PROJECT_ID = os.environ.get("PROJECT_ID")

SPACE_ID = os.environ.get("SPACE_ID")
SUBSCRIPTION_ID = os.environ.get("SUBSCRIPTION_ID_PRODUCTION")

# %%
from ibm_cloud_sdk_core.authenticators import CloudPakForDataAuthenticator
from ibm_watson_openscale import *
from ibm_watson_openscale.supporting_classes.enums import *
from ibm_watson_openscale.supporting_classes import *

authenticator = CloudPakForDataAuthenticator(
    url=CPD_URL,
    username=CPD_USERNAME,
    apikey=CPD_API_KEY,
    disable_ssl_verification=True
)

wos_client = APIClient(
    service_url=CPD_URL,
    authenticator=authenticator,
)
data_mart_id = wos_client.service_instance_id
data_mart_id

# %%
wos_client.version

# %%
from ibm_watson_openscale.supporting_classes.enums import *

data_set_id = None
response = wos_client.data_sets.list(
    type=DataSetTypes.PAYLOAD_LOGGING,
    target_target_id=SUBSCRIPTION_ID,
    target_target_type=TargetTypes.SUBSCRIPTION)
response.result.to_dict()

# %%
data_set_id = response.result.to_dict()['data_sets'][0]['metadata']['id']
data_set_id

# %%
#import json
#import pandas as pd

#with open(file="test.json", mode="r", encoding="utf-8") as f:
#    data = json.load(f)
#df = pd.json_normalize(data=data)
#df

# %%
#i = 0

# %%
#request_body = []
#question = df.iloc[i]["question"]
#context = df.iloc[i]["context"]
#generated_text = df.iloc[i]["generated_text"]
#tmp = {
#    "request": {
#        "parameters": {
#            "template_variables": {
#                "context": context,
#                "question": question
#            }
#        }
#    },
#    "response": {
#        "results": [
#            {
#                "generated_text": generated_text
#            }
#        ]
#    }
#}
#request_body.append(tmp)

# %%
#print(json.dumps(request_body, indent=2, ensure_ascii=False))

# %%
#response = wos_client.data_sets.store_records(
#data_set_id=data_set_id,
#request_body=request_body,
#background_mode=True)
#response.result.to_dict()

# %%
#wos_client.data_sets.get_records_count(data_set_id=data_set_id)

# %%
```

```python:app.py
# %%
#!pip install gradio

# %%
import gradio as gr
import json
import pandas as pd
from store_records import wos_client, data_set_id

# %%
def json2questions(filepath):
    try:
        with open(filepath, "r", encoding="utf-8") as f:
            data = json.load(f)
        df = pd.json_normalize(data)
        df["i_question"] = df.index.astype(str) + ":" + df["question"]
        questions = df["i_question"].tolist()
        return gr.update(choices=questions, value=questions[0])
    except Exception as e:
        print(f"{e}")
        return gr.update(choices=[f"{e}"], value=f"{e}")

def rag(filepath, question):
    try:
        with open(filepath, mode="r", encoding="utf-8") as f:
            data = json.load(f)
        df = pd.json_normalize(data)

        df["i_question"] = df.index.astype(str) + ":" + df["question"]
        i = df.index[df["i_question"] == question].tolist()

        context = "|".join(df["context"].iloc[i].astype(str))
        generated_text = "|".join(df["generated_text"].iloc[i].astype(str))
        reference_text = "|".join(df["reference_text"].iloc[i].astype(str))

        request_body = []
        for index, row in df.iloc[i].iterrows():
            question = row["question"]
            context = row["context"]
            generated_text = row["generated_text"]
            tmp = {
                "request": {
                    "parameters": {
                        "template_variables": {
                            "context": context,
                            "question": question
                        }
                    }
                },
                "response": {
                    "results": [
                        {
                            "generated_text": generated_text
                        }
                    ]
                }
            }
            request_body.append(tmp)

        response = wos_client.data_sets.store_records(
            data_set_id=data_set_id,
            request_body=request_body,
            background_mode=True
        )
        response.result.to_dict()

        wos_client.data_sets.get_records_count(data_set_id=data_set_id)

        return context, generated_text, reference_text
    except Exception as e:
        context = f"{e}"
        generated_text = f"{e}"
        reference_text = f"{e}"
        return context, generated_text, reference_text

with gr.Blocks() as app:
    filepath = gr.File(label="json", file_types=[".json"])
    question = gr.Dropdown(label="question", choices=[])
    button = gr.Button()
    context = gr.Textbox(label="context")
    generated_text = gr.Textbox(label="generated_text")
    reference_text = gr.Textbox(label="reference_text")

    filepath.change(fn=json2questions, inputs=[filepath], outputs=[question])
    button.click(fn=rag, inputs=[filepath, question], outputs=[context, generated_text, reference_text])

app.launch(server_name="0.0.0.0")

# %%
```

```json:test.json
[
  {
    "question": "保険の申し込み方法を教えてください。",
    "context": "保険に加入するには、申込書への記入と本人確認書類の提出が必要です。",
    "generated_text": "保険の申し込みには、申込書の記入と本人確認書類の提出が必要です。",
    "reference_text": "保険の申し込みは、申込書の記入と必要書類の提出によって完了します。"
  },
  {
    "question": "保険金の請求には何が必要ですか？",
    "context": "保険金請求には、請求書と診断書などの証明書類が必要です。",
    "generated_text": "保険金の請求には、必要な書類の提出が求められます。",
    "reference_text": "請求書や診断書などの証明書類を提出することで、保険金請求が可能です。"
  },
  {
    "question": "保険を解約するにはどうすればよいですか？",
    "context": "解約は、契約者本人による書面での手続きが必要です。",
    "generated_text": "保険の解約は、書面での申請が必要です。",
    "reference_text": "契約者本人が所定の手続きを行うことで、保険を解約できます。"
  },
  {
    "question": "保険料の支払い方法は変更できますか？",
    "context": "支払い方法は、口座振替やクレジットカードなどから選択可能です。",
    "generated_text": "はい、支払い方法は変更可能です。手続きが必要です。",
    "reference_text": "お手続きにより、保険料の支払い方法を変更できます。"
  },
  {
    "question": "担当者に連絡するにはどうすればいい？",
    "context": "連絡先は契約時に発行された書類に記載されています。",
    "generated_text": "担当者の連絡先は、契約時の資料をご確認ください。",
    "reference_text": "ご契約時にお渡しした書類に、担当者の連絡先が記載されています。"
  },
  {
    "question": "補償内容の確認方法を教えてください。",
    "context": "補償内容は契約書やマイページで確認できます。",
    "generated_text": "契約内容は、契約書またはオンラインで確認できます。",
    "reference_text": "ご契約の補償内容は、契約書やマイページ上でご確認いただけます。"
  },
  {
    "question": "申込に必要な書類は何ですか？",
    "context": "本人確認書類と申込書の提出が必要です。",
    "generated_text": "本人確認書類と申込書が必要になります。",
    "reference_text": "申込の際には、本人確認書類と申込書をご提出いただきます。"
  },
  {
    "question": "契約内容を変更したいときはどうすれば？",
    "context": "契約内容の変更は、カスタマーサポートを通じて可能です。",
    "generated_text": "契約の変更は、カスタマーサポートにご連絡ください。",
    "reference_text": "変更をご希望の場合は、カスタマーサポートまでご連絡ください。"
  },
  {
    "question": "保険の対象にペットは含まれますか？",
    "context": "一部のプランではペット保険も提供されています。",
    "generated_text": "はい、プランによってはペットも対象となる場合があります。",
    "reference_text": "ご加入のプランによっては、ペットも補償対象になる場合があります。"
  },
  {
    "question": "保険証券を紛失した場合はどうすればいい？",
    "context": "紛失時は再発行の手続きを行う必要があります。",
    "generated_text": "保険証券を紛失した場合は、再発行の手続きをしてください。",
    "reference_text": "保険証券の紛失時には、再発行のお手続きが必要です。"
  }
]
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3756172/1ec9d8fc-50a0-4fcb-a2a5-92c40197ba3f.png)

#### 監視@watsonx.governance

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3756172/8c6aa4aa-6d22-44b8-b7e2-1dee45c5c069.png)