1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
| from openai import OpenAI
import pandas as pd
import json
import time
import requests
from rapidfuzz.process import extract
# OpenAI APIのクライアントを作成
OPENAI_CLIENT = OpenAI()
# 質問回答用のスレッドを作成
THREAD = OPENAI_CLIENT.beta.threads.create()
# モデルの設定
MODEL = "gpt-4-1106-preview" # 使用するGPTモデル名
# カード情報の読み込み
CARD_INFO_DF = pd.read_csv("cardinfo.csv", encoding="utf-8")
CARD_NAME_LIST = CARD_INFO_DF["name_jp"].tolist()
# カード名の類似度を計算してカードidを返す関数
def get_card_id_by_name(arguments):
card_name = arguments.get("card_name")
print(f"[DEBUG]card_name: {card_name}")
result = extract(card_name, CARD_NAME_LIST)
print(f"[DEBUG]result: {result}")
# result[:][2]にCARD_INFO_DF.iloc[result[:][2]]["card_id"]を代入
result = [(result[i][0], result[i][1], CARD_INFO_DF.iloc[result[i][2]]["card_id"]) for i in range(len(result))]
if result[0][1] > 800: # 類似度のトップが80%以上の場合はそのカードのidを返す
card_id = CARD_INFO_DF.iloc[result[0][2]]["card_id"]
return (True, card_id)
else: # 類似度のトップが90%未満の場合は上位5件のカード名と類似度を返す
return (False, result)
# YU-GI-OH APIにカード名を渡してカード情報を取得する関数
def get_card_info_by_id(arguments):
card_id = arguments.get("card_id")
url = f"https://db.ygoprodeck.com/api/v7/cardinfo.php?id={card_id}"
response = requests.get(url)
if response.status_code == 200:
# APIから取得した情報をJSON→辞書→文字列に変換
data = response.json()
# 'data'キーの最初の要素から指定されたキーの情報を取得
card_info = data["data"][0]
keys_to_extract = ["id", "name", "type", "frameType", "desc", "atk", "def", "level", "race", "attribute", "archetype"]
extracted_data = {key: card_info.get(key, None) for key in keys_to_extract} # keyが存在しない項目はNoneを返す(魔法罠のatkなど)
# 辞書を文字列に変換
result_string = ", ".join([f"'{key}': {value!r}" for key, value in extracted_data.items()])
return result_string
else:
return f"card_id[{card_id}]と一致するカードが見つかりませんでした。"
# Functions Callingで使用する関数の辞書
FUNCTIONS_DICT = {
"get_card_info_by_id": get_card_info_by_id,
"get_card_id_by_name": get_card_id_by_name,
}
# ツールの設定
TOOLS = [
{"type": "code_interpreter"},
{"type": "retrieval"},
{
"type": "function",
"function": {
"name": "get_card_id_by_name",
"description": "カード辞書との類似度計算によって遊戯王カードの日本語名からカードIDを取得、類似度の上位5件の[カード名、類似度、カードID]を返す",
"parameters": {
"type": "object",
"properties": {
"card_name": {
"type": "string",
"description": "日本語カード名のみを文字列で入力"
}
},
"required": ["card_name"]
}
}
},
{
"type": "function",
"function": {
"name": "get_card_info_by_id",
"description": "YU-Gi-OH APIによってカードIDからカード情報を取得",
"parameters": {
"type": "object",
"properties": {
"card_id": {
"type": "string",
"description": "カードIDのみを文字列で入力"
}
},
"required": ["card_id"]
}
}
},
]
# アシスタントの設定
def set_assistant(name, instructions, tools, model):
assistant = OPENAI_CLIENT.beta.assistants.create(
name=name,
instructions=instructions,
tools=tools,
model=model
)
return assistant
# ユーザーメッセージをスレッドに追加
def add_message(role, content):
message = OPENAI_CLIENT.beta.threads.messages.create(
thread_id=THREAD.id,
role=role,
content=content
)
return message
# アシスタント実行時の指示
def set_run_instruction(assistant, instructions):
run = OPENAI_CLIENT.beta.threads.runs.create(
thread_id=THREAD.id,
assistant_id=assistant.id,
instructions=instructions
)
return run
# アシスタントの実行 (Code Interpreter, Retrieval, and Function calling)
def run_assistant(run):
while(1): # ループしないと中身がユーザーメッセージだけのまま更新されず終わる(多分並列処理してる)
run = OPENAI_CLIENT.beta.threads.runs.retrieve(
thread_id=THREAD.id,
run_id=run.id
)
if(run.status) == "in_progress": # 実行中("in_progress")の場合は2秒待機
print("実行中...")
time.sleep(2)
elif(run.status) == "requires_action": # 実行中(requires_action")の場合は2秒待機
print("関数待ち...")
if(run.required_action.type) == "submit_tool_outputs":
tool_calls = run.required_action.submit_tool_outputs.tool_calls
tool_outputs=[]
for tool_call in tool_calls:
arguments_str = tool_call.function.arguments
arguments = json.loads(arguments_str)
tool_call_id = tool_call.id
function_name = tool_call.function.name
function_to_call = FUNCTIONS_DICT[function_name]
function_result = function_to_call(arguments)
tool_output = {
"tool_call_id": tool_call_id,
"output": str(function_result),
}
tool_outputs.append(tool_output)
# tool実行結果を送信
run = OPENAI_CLIENT.beta.threads.runs.submit_tool_outputs(
thread_id=THREAD.id,
run_id=run.id,
tool_outputs=tool_outputs
)
time.sleep(2)
elif(run.status) == "completed": break # 実行完了("completed")の場合はループを抜ける
else:
print(run.status)
time.sleep(2)
# 実行結果の取得
def print_threads():
messages = OPENAI_CLIENT.beta.threads.messages.list(
thread_id=THREAD.id
)
data = messages.data
# 実行結果の出力
for d in data[::-1]: # 0から昇順に過去のメッセージが入っているので逆順ループ
role = d.role
content = d.content[0].text.value
print(f"[{role}]:\n{content}\n")
# tool実行結果を取得
def print_tool_result(run):
run_steps = OPENAI_CLIENT.beta.threads.runs.steps.list(
thread_id=THREAD.id,
run_id=run.id
)
# tool実行結果の出力
for run_step in run_steps.data:
if run_step.type == "tool_calls":
tool = run_step.step_details.tool_calls[0]
tooltype = tool.type
tool_result = getattr(tool, tooltype)
input = tool_result.input
outputs = tool_result.outputs
print(f"[{tooltype}]:\n{input}\n\n{outputs}\n")
# アシスタントの設定
name="師匠"
instructions="遊戯王カードに関する情報に答える"
assistant = set_assistant(name, instructions, TOOLS, MODEL)
run_instruction = "APIを利用して正確な情報を返す"
while(1):
# ユーザーメッセージをスレッドに追加
user_message = input(f"テキストを入力してください: ") # メッセージ入力
if user_message == "exit": break
message = add_message("user", user_message)
# アシスタント実行時の指示
run = set_run_instruction(assistant, run_instruction )
# アシスタントの実行
run_assistant(run)
# 実行結果の取得
print_threads()
|
