今回の取組み / 取り組みから得た知見

GraphDBは以前から知っていたが、使うタイミングがなく触る機会がなかった
LLMが出てきてGraphDBとLLMとの連動の記事は日本語でなく(投稿時点)、サンプルとして利用可能な状態になれたら有用かと思い、記事を書く
日本語のwikidataを登録するまでに仕事を除く時間で合計2週間詰めていた
今回利用するGraphDBはNeo4jであり、そもそも日本語ドキュメントが弱く、公式ページを見るのがやはり良かった
思っていたよりwikidataをneo4jにインポートする記事はあるが断片的な情報が多く、つなぎ合わせる必要があった
作成した結果のGraphDBは、使い方によってはRDBよりもデータ構造の理解は簡単になりそうで、使いどころが制限されるが有用なDBとなりそう
検索においてはElasticSearchと連携することでLLMでも有用な肌感

参考にしたページ

Wikipediaのダンプからページを取り出す
- 最初に取り組んだ際に出てきた記事
- データがバイト文字で検索できる点はとても使い勝手が良い
- データに精通していない場合、アーキテクチャも含め検討するのが難しいため、Microサービスとしてリリースしていくのが良さそうなデータ
- IndexデータとMultistreamデータで分割されているため、最初はIndexの解析から進めるのが良い
データの分割
- Wikidataの知識をNeo4jで可視化してみる
- 過去(2019年)の情報だが、今も現役で利用可能
- ただし、一部は修正した方が操作性が上昇する
Wikidataについて
- 参考ページ
- 結局公式ドキュメントからイメージを構築した方が早い
- どのようなデータがあるか、は有志のページが参照のこと
wikiextractor
- wikiのデータを加工してくれるツールがGithub上に公開されている
- 結果としてGoogle Colabでも半日-1日かかったし、結果をさらに解析して検索可能な状態にさせる必要がある(Index化)のため、使い方の検討が必要

wiki データの取得方法

データを取得はWikipediaに公式でdumpファイルがあるため、そこからデータを取得する
データが1週間ごとに更新されるため、同時に作業している人がいる場合は同期できるように管理する必要があるため注意
- https://dumps.wikimedia.org/wikidatawiki/20231220/
MultiStream
- wikidatawiki-20231220-pages-articles-multistream-index.txt.bz2(100GB 以上!←時間がかかるので注意)
- wikidatawiki-20231220-pages-articles-multistream.xml.bz2(400 MB)
実際に利用したJsonデータ
以下は、上記の圧縮されたjsonデータを用いて作業を進めます

wiki データの展開方法

ここからは実際のプログラムを記載していきます
プログラム1(latest-all.json.bz2をそのままでneo4jに必要そうなデータを抽出する)

import bz2
import json


def get_claims_mainsnak(relation_p, claim):
    relation_list = []
    if claim.get("mainsnak") is None:
        pass
    else:
        claim_mainsnak = claim["mainsnak"]
        # print(claim_mainsnak)
        claim_mainsnak_datatype = claim_mainsnak.get("datatype", "")
        if claim_mainsnak_datatype == "wikibase-item":
            try:
                # print(claim_mainsnak)
                claim_mainsnak_datavalue = claim_mainsnak.get("datavalue")
                # claim_mainsnak_datavalue_type = (
                #     claim_mainsnak_datavalue.get("type")
                # )
                claim_mainsnak_datavalue_value = claim_mainsnak_datavalue.get("value")
                claim_mainsnak_datavalue_value_id = claim_mainsnak_datavalue_value.get(
                    "id"
                )
                claim_mainsnak_datavalue_value_entity_type = (
                    claim_mainsnak_datavalue_value
                ).get("entity-type")
                if claim_mainsnak_datavalue_value_id is not None and (
                    claim_mainsnak_datavalue_value_entity_type == "item"
                ):
                    # print(claim_mainsnak_datavalue_value_id)
                    # relation の 作成
                    relation_list.append(
                        [
                            relation_p,
                            claim_mainsnak_datavalue_value_id,
                            "direct",
                        ]
                    )
                    # print(relation_list[-1])
                # print(claim_mainsnak_datavalue)
                # print(
                #     # claim_mainsnak_datavalue_type,
                #     claim_mainsnak_datavalue_value
                # )
            except Exception as e:
                # print(e)
                pass
            pass
        pass
    return relation_list


def get_claims_references(claim_references_list):
    relation_list = []
    for claim_reference in claim_references_list:
        # print(claim_reference)
        # try:
        claim_reference_snaks = claim_reference.get("snaks")
        for relation_p in claim_reference_snaks.keys():
            # print(relation_p)
            # claim_reference_snak = claim_reference_snaks[relation_p]
            # print(claim_reference_snak)
            # print(len(claim_reference_snak))
            for claim_reference_snak in claim_reference_snaks[relation_p]:
                # for claim_reference_snak in claim_reference_snak_list:
                # print(claim_reference_snak)
                ## a
                claim_mainsnak_datatype = claim_reference_snak.get("datatype", "")
                if claim_mainsnak_datatype == "wikibase-item":
                    # print(claim_mainsnak)
                    claim_mainsnak_datavalue = claim_reference_snak.get("datavalue")
                    # claim_mainsnak_datavalue_type = (
                    #     claim_mainsnak_datavalue.get("type")
                    # )
                    claim_mainsnak_datavalue_value = claim_mainsnak_datavalue.get(
                        "value"
                    )
                    claim_mainsnak_datavalue_value_id = (
                        claim_mainsnak_datavalue_value.get("id")
                    )
                    claim_mainsnak_datavalue_value_entity_type = (
                        claim_mainsnak_datavalue_value
                    ).get("entity-type")
                    if claim_mainsnak_datavalue_value_id is not None and (
                        claim_mainsnak_datavalue_value_entity_type == "item"
                    ):
                        # print(claim_mainsnak_datavalue_value_id)
                        # relation の 作成
                        relation_list.append(
                            [
                                relation_p,
                                claim_mainsnak_datavalue_value_id,
                                "reference",
                            ]
                        )
                        # print(relation_list[-1])
                    # print(claim_mainsnak_datavalue)
                    # print(
                    #     # claim_mainsnak_datavalue_type,
                    #     claim_mainsnak_datavalue_value
                    # )
                    pass
                pass
            pass
        pass
        # except Exception as e:
        #     print(e)
        #     pass
        # pass
        ## a
    return relation_list


def get_claims_qualifiers(claim_qualifiers_dict):
    relation_list = []
    for relation_p in claim_qualifiers_dict.keys():
        for claim_qualifier_snak in claim_qualifiers_dict[relation_p]:
            # for claim_reference_snak in claim_reference_snak_list:
            # print(claim_reference_snak)
            ## a
            claim_mainsnak_datatype = claim_qualifier_snak.get("datatype", "")
            if claim_mainsnak_datatype == "wikibase-item":
                try:
                    # print(claim_mainsnak)
                    claim_mainsnak_datavalue = claim_qualifier_snak.get("datavalue")
                    # claim_mainsnak_datavalue_type = (
                    #     claim_mainsnak_datavalue.get("type")
                    # )
                    claim_mainsnak_datavalue_value = claim_mainsnak_datavalue.get(
                        "value"
                    )
                    claim_mainsnak_datavalue_value_id = (
                        claim_mainsnak_datavalue_value.get("id")
                    )
                    claim_mainsnak_datavalue_value_entity_type = (
                        claim_mainsnak_datavalue_value
                    ).get("entity-type")
                    if claim_mainsnak_datavalue_value_id is not None and (
                        claim_mainsnak_datavalue_value_entity_type == "item"
                    ):
                        # print(claim_mainsnak_datavalue_value_id)
                        # relation の 作成
                        relation_list.append(
                            [
                                relation_p,
                                claim_mainsnak_datavalue_value_id,
                                "qualifiers",
                            ]
                        )
                        # print(relation_list[-1])
                    # print(claim_mainsnak_datavalue)
                    # print(
                    #     # claim_mainsnak_datavalue_type,
                    #     claim_mainsnak_datavalue_value
                    # )
                except Exception as e:
                    # print(e)
                    pass
                pass
            pass
        pass
    return relation_list


def get_claims(claims):
    relation_list = []
    for relation_p in claims.keys():
        # print(relation_p)
        for claim in claims[relation_p]:
            relation_list.extend(get_claims_mainsnak(relation_p, claim))
            relation_list.extend(get_claims_references(claim.get("references", [])))
            relation_list.extend(get_claims_qualifiers(claim.get("qualifiers", {})))
            # qualifiers
            pass
        pass
    return relation_list


file_path = "./latest-all.json.bz2"
wf_file_path = "./tmp.csv"
wf = open(wf_file_path, "w", encoding="utf-8")
wf.write("q_id_start,relation,q_id_end,relation_type_rank" + "\n")
q_id_wf = open("./q_id/q_id.txt", "w", encoding="utf-8")

q_id_wf.write("base_q_id,label_en,label_ja,descriptions_en,descriptions_ja" + "\n")
# rdf_wf = open("./rdf/rdf.txt", "w", encoding="utf-8")
# relation_list = []
with bz2.BZ2File(file_path, "r") as rf:
    next(rf)  # 1行目を飛ばす
    for i, line in enumerate(rf, 1):
        try:
            line = json.loads(line[:-2])
        except json.decoder.JSONDecodeError as e:
            print(e)
            # print(i)
            # triples = []
            # qs = []
            continue
        # print(line)
        # print(i, "*" * 100)
        # if i >= 5 * 1e2:
        #     break

        # ############################################################
        # ############################################################
        # ############################################################

        # relation_list = get_claims(claims)
        # relation_list.extend(get_claims(claims))
        try:
            key = "id"
            base_q_id = line[key]
            label_en = line["labels"]["en"]["value"]
            label_ja = line["labels"]["ja"]["value"]
            descriptions_en = line["descriptions"]["en"]["value"]
            descriptions_ja = line["descriptions"]["ja"]["value"]
            claims = line["claims"]
            # claims_relation_list = []
            q_id_wf.write(
                '"'
                + '","'.join(
                    [base_q_id, label_en, label_ja, descriptions_en, descriptions_ja]
                )
                + '"'
                + "\n"
            )

            claims_relation_list = get_claims(claims)
            # save_claims_relation_list = []
            for row in claims_relation_list:
                _save_claims_relation_list = []
                _save_claims_relation_list.append(base_q_id)
                _save_claims_relation_list.extend(row)
                # save_claims_relation_list.append(_save_claims_relation_list)
                wf.write(",".join(_save_claims_relation_list) + "\n")
                del _save_claims_relation_list

            # relation_list.extend(save_claims_relation_list)
            del claims_relation_list
        except Exception as e:
            pass
        # print(relation_list)
        # print(len(relation_list))

        # if i >= 2:
        #     break
q_id_wf.close()
# rdf_wf.close()
wf.close()

# 923 / 60 hours
# 15.xxx hours

プログラム2(作成したデータ内で重複がある場合、インポート時にエラーとなり、また、データ構造として重複は不要なため今回は削除)
- sqlite3を利用しライトに重複を削除した後、再度csv化する

sqlite3に作成したcsvをインポートする

sqlite3のコマンドでデータを作成する

以下はslite3のコマンド

// 作成したデータをインポートする
.mode csv  
.import tmp.csv rdf

sqlite3に作成したDBからcsvとしてエクスポートする

// 上記でインポートしたデータをrdf_n.csvに出力する
// distinctで重複を削除する
.headers on  
.mode csv  
.output rdf_n.csv  
SELECT distinct "q_id_start", relation, "q_id_end", "relation_type_rank"  
FROM rdf;  
.quit

プログラム3(最終的なデータがcsv化されたので、このデータを基準にGraphDBに登録できる形(Node, Relation)に再作成する)

### q_id を neo4j にインポートするためのデータに変換する
cnt = 0
q_wf = open(f"./q_id/q_id_neo4j_{str(cnt).zfill(2)}.csv", mode="w", encoding="utf8")
q_wf.write(
    "entityId:ID,label_en,descriptions_en,label_ja,descriptions_ja,:LABEL" + "\n"
)

with open("./q_id/q_id.txt", "r", encoding="utf8") as f:
    for i, line in enumerate(f):
        if i == 0:
            continue
        line = line.strip()
        s_line = line.split(",")
        q_id = s_line[0].replace('"', "")
        write_txt = q_id + "," + q_id + ",Entity"
        entityId = s_line[0]
        label_en = s_line[1]
        label_ja = s_line[2]
        descriptions_en = s_line[3]
        descriptions_ja = s_line[4]
        s_line2 = [label_en, descriptions_en, label_ja, descriptions_ja, "Entity"]
        write_txt = (
            q_id
            + ',"'
            + '","'.join([x.replace('"', "").replace('"', "") for x in s_line2])
            + '"'
        )
        q_wf.write(write_txt + "\n")
q_wf.close()

### q_id で作成した情報で、ノード一覧を取得し、relation作成後に存在しないノードを確認できるようにする
cnt = 0

node_list = set()
with open(f"./q_id/q_id_neo4j_{str(cnt).zfill(2)}.csv", "r", encoding="utf8") as f:
    for i, line in enumerate(f):
        line = line.strip()
        # print(line)
        if i == 0:
            continue
        # if i > 10:
        #     break
        node_str = line.split(",")[0]
        if node_str not in node_list:
            node_list.add(node_str)


### relation を neo4j にインポートするためのデータに変換する
### relation に紐づくノードが存在しない場合は、ノードを保存する
none_node_list = set()
rdf_wf = open(f"./rdf_n2.csv", mode="w", encoding="utf8")
rdf_wf.write(":start_id,rank,:end_id,:type" + "\n")

# print(":start_id,role,:end_id,:type")
with open("./rdf_n.csv", mode="r", encoding="utf8") as f:
    for i, line in enumerate(f):
        line = line.strip()
        # print(line)
        if i == 0:
            continue
        s_line = line.split(",")
        from_node = s_line[0]
        relation = s_line[1]
        to_node = s_line[2]
        rank = s_line[3]
        # rdf_wf.write(from_node + "," + relation + "," + to_node + ",n," + rank + "\n")

        rdf_wf.write(from_node + "," + rank + "," + to_node + "," + relation + "\n")
        # if i > 10:
        #     break
        if from_node not in node_list:
            none_node_list.add(from_node)
        if to_node not in node_list:
            none_node_list.add(to_node)
rdf_wf.close()
print(i)
print(len(none_node_list))

# 足りないノードを保存する
cnt = 0
q_wf = open(
    f"./q_id/q_id_neo4j_{str(cnt).zfill(2)}_none.csv", mode="w", encoding="utf8"
)
# q_wf = open("./q_id/q_id_neo4j.csv", mode="w", encoding="sjis")
"""entityId:ID,name,:LABEL
Q31,Q31,Entity"""
q_wf.write("entityId:ID,name,:LABEL" + "\n")


for x in none_node_list:
    # print(x)
    q_wf.write(",".join([x, x, "Entity"]) + "\n")
    # break

q_wf.close()

bin\neo4j-admin database import full --skip-bad-entries-logging --skip-duplicate-nodes --nodes=import\q_id_neo4j_00.csv --nodes=import\q_id_neo4j_00_none.csv  --relationships=import\rdf_n2.csv --skip-bad-relationships --overwrite-destination wiki-all --verbose

とりあえず取り込みがしたかったので、構文の最適化は別途調整の必要有

実際に作業してみて

Neo4jが思っていた以上に曲者で、かつ、ChatGPTを利用しても良い回答が得られなかった.
結果として必要な作業内容は、デバッグをしながらエラーを取り除く、または、wikidataの構造理解、などで、AIが出てきても結局はプログラマーがやってきた原始的な作業は変わらないことを感じた
- 上記の原始的な作業の詳細を書くと、AGIによって人の作業が無くなったSF的な未来が訪れたら無職になると思っているので、それまでの間のことを表しています。
2024年に入りChatGPTが出てから約9ヵ月ほど汎用AIの民主化が起こっているが、GraphDBが使われた記事がなかなか出てきていないことから、企業内で知見としてはあるが出せないか、そもそも有用でないか、または、取り組みが進んでいないか、などが考えられます。
- 良いサイトがあればぜひ教えてください！
終わりに
- 去年から今年にかけて加速するAI時代、良い感じに触れながらEnjoyしていきましょう!