templates.py
# %%
template_rag = """# 指示：
与えられた文脈にもとづいて質問に回答してください。
# 文脈：
{context}
# 質問：
{question}
# 回答：
"""
# %%
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
# %%
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()
# %%
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)
# %%
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")
# %%
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": "保険証券の紛失時には、再発行のお手続きが必要です。"
}
]
監視@watsonx.governance
