[#46] やっと出てきた　Claudeの「MCP」

2024年12月3日 13:38

始めに

先日Claudeが新機能「MCP（Model Context Protocol）」を発表しましたね！
簡単に説明すると、画面から自然言語でデータベースにアクセスして結果を取り出したりする処理です。つまりMCPを使うことで、AIシステムがGoogle DriveやSlack、GitHub、Postgresなどの複数のデータやツールと共通の標準プロトコルで連携可能になるということです。

事例

実は、私たちは1年半前(23/5月)から　Python とOpenAIを使って同様の仕組みを作ってRAG対応を行っていました、当時はまだほとんど注目されていなかったと記憶しています、時々間違ったSQLを生成したりはしていましたが。
環境　Ubuntu20.04(Mac) + OpenAI +Python + DB(SQLite)
OpenAIのAPIを使って自然言語でのデータ検索を行なっています、今まではテーブルに合わせたSQLを記述しなけれはせなりませんでしたが、これからは必要の無い世界がやってくるかもです。
--------------------------------------------------------------------

メタデータが読み込まれました
>staffから全件表示して

SQL文: SELECT * FROM staff;　　 ---> AIが生成したSQL

説明: staffテーブルから全てのデータを取得し、表示するSQL文です。SELECT文で全ての列を取得するため、*を使用しています。

| id | name | job_id |

|------|--------|----------|

| 1 | Yamada | 3 |
| 2 | Kitano | 3 |
| 3 | 藤井 | 3 |

この単純なPythonプログラムの中には、GPTによる文脈変換、日本語->SQLというドメイン変換、推論、という概念がすべて詰まっています

終りに

Claudeはプログラムを組まなくてもノーコードで同様の仕組みを実現しました。今まで、AIアシスタント運用についてもデータを事前にクレンジングしてナレッジ更新をしていましたが、これからは直接データにアクセスできるので事前準備無しでAIアシスタントが運用できるようになるでしょう
・予約データへのアクセス
・注文、実績、顧客対応データへのアクセス
・関連Webサイトへのアクセスから必要データ抽出、要約
など、利用の可能性は無限大、今後のバージョンアップに期待したいと思います。
やっと、我々にClaude が追いついたという事に・・・・

事例の、Pythonサンプルコード、最新のOpenAIでは動かないと思いますので、コードなどは修正をしてください。
import openai
import os
import sqlite3
import sys
from sqlalchemy import create_engine, MetaData
from sqlalchemy.orm import sessionmaker
from sqlalchemy.sql import text
from sqlalchemy.exc import SQLAlchemyError
from tabulate import tabulate
import json
from json.decoder import JSONDecodeError
openai.api_key = os.environ['OPENAI_API_KEY'] = "sk-XXX"

engine = create_engine('sqlite:////var/tmp/kenshin.db', echo=False)

def init_meta_data():

# メタデータを作成、テーブル情報を反映させるためにエンジンからテーブル情報を読み込む
metadata = MetaData()
metadata.reflect(bind=engine)

# メタデータのテーブル情報からテーブル名と列情報を編集する
meta_text = ''
for table in metadata.tables.values():
    meta_text += 'テーブル名:'+ table.name + '\n'
    for column in table.columns:
        meta_text += f'列名:{column.name}, 型: {column.type}\n'
    meta_text += '\n'
return meta_text

def exec_sql(sql):
Session = sessionmaker(bind=engine)

error_text = "None"
rows = []
tabled = ""
dict_array = []

try:
    with Session() as session:
        t = text (sql)
        result = session.execute(t)

        header = [k for k in result.keys()]

        rows = result.fetchall()
        tabled = tabulate(rows,header, tablefmt="plain")

        dict_array = []
        for row in rows:
            dict_row = {}
            for i in range(len(header)):
                dict_row[header[i]] = row[i]
            dict_array.append(dict_row)
        #print(tabled)
        #print()

except SQLAlchemyError as e:
    print(f'Exception Excute SQL: {e}\n')
    error_text = str(e)

return {
    "read_count": len(rows),
    "tabled": tabled,
    "dict_array": dict_array,
    "error": error_text
}

def exec_api(user_text="",user_info="",prompt="",meta_text="",summary="",temperature=0):
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
temperature = temperature,
messages=[
{
"role": "user",
"content": prompt.format(
user_text=user_text,
user_info=user_info,
prompt=prompt,
meta_text=meta_text,
summary=summary)
}
],
)
return response.choices[0].message.content.strip()

def find_user(user_text,meta_text):
prompt="""
{{
"prompt": "
## DB情報
{meta_text}

    ## 処理
    ユーザテーブルを参照し、次の要求文に対してユーザを一意に特定したい。
    その回答(SQL文を含まない)と、SQL文(ID,名前,電話番号,メール)を
    厳密なJSON形式で以下のように提供してください。
    なお、日本語の名前は最初がlast_nameです。
    説明文は不要ですので非表示にしてください。：
      {{
          "response": {{
            "message": "ユーザをDBに問い合わせます)",
            "sql_query": "SQL文"
          }}
      }}

      要求: [{user_text}]
    }}
}}

"""
response = exec_api(user_text=user_text,prompt=prompt,meta_text=meta_text)
#print (response)
json_answer = json.loads(response)
sql = json_answer['response']['sql_query']
if sql:
result = exec_sql(sql)
#print (result['read_count'])
#print (result['tabled'])
#print (result['dict_array'])
if result['read_count'] == 1:
return result['tabled']
else:
return ""

def get_user_info(user,meta_text):
prompt="""
{{
"prompt": "
## DB情報
{meta_text}

  ## ユーザ情報
    {user_info}

  ## 処理
  DB情報を参照し、
  ユーザ情報をもとにそのユーザに関するすべてがわかるためのタスク
  (購買履歴、対応履歴、購入済商品情報など）を挙げ、
  そのタスクの説明(SQL文を含まない)と、タイトル、SQL文を
  厳密なJSON形式で以下のように提供してください。
  説明文は不要ですので非表示にしてください。：
    {{
        "message":"(回答メッセージ)",
        "queries": [{{
          "title";(タスクのタイトル)
          "description": "(タスクの説明)",
          "sql_query": "(SQL文)"
        }}]
    }}

}}
"""
user_text = ""
response = exec_api(user_info=user,prompt=prompt,meta_text=meta_text)
#print (response)
json_answer = json.loads(response)
user_info = "\nユーザ情報\n"+user+"\n"
for query in json_answer['queries']:
description = query['description']
user_info += "\n"+query['title']+"\n"
#print (title)
sql = query['sql_query']
if sql:
result = exec_sql(sql)
user_info += result['tabled']+"\n"
return user_info

def user_support(user_text,user_info,summary):
summary = ""
prompt="""
{{
"prompt": "
## 会話の要約
{summary}
## ユーザ情報
{user_info}

    ## 処理
    あなたは接客のエキスパートです。お客様の要求文にたいして的確に答えてください。
    接客のさい、ユーザ情報を参照し親切丁寧にお客様サポートをしてください。
    ただし、DB情報にある事実だけを答えてください。
    お客様からの要求文が「ありがとう。ではまた。さよなら。」などの終了を暗示する言葉が入力された場合サポート終了とみなします。
    回答を含めて以下のように厳密なJSON形式で出力してください。:
      {{
          "message":"(回答メッセージ)",
          "status": "(サポートの状態。開始、サポート中、サポート終了)",
          "summary":"要求文と回答の内容の要約"
      }}

    要求文[{user_text}]
    "
}}

"""
response = exec_api(user_text=user_text,user_info=user_info,prompt=prompt,summary=summary,temperature=0.5)
json_answer = json.loads(response)
return json_answer

if name == "main":

#----- 初期設定 ------
meta_text = init_meta_data()
if not meta_text:
print("メタデータが読み込めませんでした。")
sys.exit(0)

#------ ユーザの特定 ------
print("お客様の情報を確認します。お名前、電話番号、ユーザID、メールなどお客様を特定できるデータを入力してください。")
while True:
user_text = input("\n>")
user = find_user(user_text,meta_text)
print(user)
if user:
print("お客様の情報を確認できました。ありがとうございます。")
break
else:
print("お客様の情報を確認できませんでした。再度入力してください。")

#------ ユーザ情報の取得　-------
print("\nお客様の情報をDBに問い合わせます。少々お待ちください。")
user_info = get_user_info(user,meta_text)
#print (user_info)
print("\nお客様の情報をDBから取得できました。")

#------ ユーザサポート　-------
print("\nお客様、何かご質問やお困り事はございますか？\n")
summary = ""
while True:
user_text = input(">")
if user_text == "quit":
break

result = user_support(user_text,user_info,summary)
print(result['message']+"\n")
if result['status'] == 'サポート終了':
  break
summary += f"要求: {user_text}\n回答: {result['summary']}\n"

print("\nありがとうございました。それでは失礼いたします。")

[#46] やっと出てきた Claudeの「MCP」

始めに

事例

終りに

いいなと思ったら応援しよう！

[#46] やっと出てきた　Claudeの「MCP」