挨拶・はじめに
こんにちは！AI基礎研究チームの二宮です。2024アドベントカレンダー2回目の投稿になります！
このシリーズの記事もあとわずかで終了しますが、弊社のAIへの取り組みについて少しでも関心を持っていただけるきっかけになれたなら幸いです。
さて、RAGに関する記事が何回も出てきて「これ以上RAGについて何が語れるの？」と思っているところかもしれませんが、最後までお付き合いください！この記事では、主流からちょっとだけ外れた実装パターンを紹介するので、試せるRAGのバリエーションが増えると思います。
「RAG」の定義は広い
AI基礎研究チームと連携を取ってRAGの応用を進めているAIサクセスチームがいくつかツールを開発してくれています。ここ最近人事部向けのツールを開発していますが、実装担当の渡部がベクトルDBのドキュメント登録においてさまざまな工夫を試しているのでよかったら合わせてお読みください！
しかし、世の中がRAGで盛り上がっている中、実は知識DBの形が徐々に増えており、ベクトルでないといけないという縛りは特にありません。「LLMにない知識を文脈として渡して答えてもらう」という条件を満たせばそれがRAGです。
そこで、「文章をベクトル化して類似度で検索」という、自然言語処理専門者以外には真相が分かりにくいパターンよりも理解しやすいデータモデルをご紹介していきます。
新しい ≠ 一番強い
一番大事なことなので本題に入る前に言わせてください。
「新しい研究を始めるからには、何かが改善されないといけない」というプレッシャーを受けながら生成AI技術の研究を行っている方が少なくないと思います。
その影響で、例えば本件のGraphRAGでいうと、「普通のRAGにはできないあれこれを、GraphRAGを使うことによって革命的な効果が期待できます」という言い方をしている投稿が多く見えます。
しかし、RAGにできないと言っていることは意外と最新のLLMとベクトルDBでできてしまったりします。例えば「情報の関連性は普通のRAGではつかめません」という断言です。検索の閾値を調整すれば、ベクトルDBは人間から見て質問に関係なさそうな情報まで引っ張ってこれるようになります。その情報をLLMが頭良く関連づけして回答します。
「GraphRAGでできることはRAGにはできない」とは限りないですし、RAG or GraphRAGどちらかに振り切るのではなく、解決したい課題とデータの構成によって使う技術を柔軟に変えればいいと断言させてください。
では、GraphRAGの場合はどのように開発を進めていくか見ていきましょう。
この記事の紹介で使われるデータは全てChatGPTを利用して作成した架空のデータになります。
GraphRAGの紹介
GraphDBの概要
まず、GraphRAGで使うデータベースはGraphDBというものです。GraphDBでは、データを「点」、データ間の関係性を「線」で表します。例えばSNSにおいて構築するなら、「ユーザー」、「投稿」が「点」となり、「ユーザーの投稿」や「ユーザーがいいねした投稿」などの関連付けが「線」になります。
この記事では、GraphDBの一つであるNeo4jを用いて説明を行なっていきます。Neo4jでは、Cypherというクエリ言語を使用しますが、例えば上記のデータを以下のように作ります。
CREATE (:User {name: 'A'}),
(:User {name: 'B'});
MATCH (a:User {name: 'A'})
CREATE (p1:Post {title: 'A投稿1'}),
(p2:Post {title: 'A投稿2'}),
(a)-[:投稿]->(p1),
(a)-[:投稿]->(p2);
MATCH (b:User {name: 'B'})
CREATE (p3:Post {title: 'B投稿1'}),
(p4:Post {title: 'B投稿2'}),
(b)-[:投稿]->(p3),
(b)-[:投稿]->(p4);
MATCH (b:User {name: 'B'}),
(ap1:Post {title: 'A投稿1'})
CREATE (b)-[:いいね]->(ap1)
MySQLなら、「ユーザーがいいねした投稿」を表すには中間テーブルを作ってデータを貯めていくことになりますが、Cypherでは CREATE (user)-[:いいね]->(post) を書くだけで
- 「いいね」という種類の矢印が存在するかどうかを確認
- 存在しない場合は、新しく作成する
- 「ユーザー」と「投稿」を「いいね」で結ぶ
を一気に処理してくれます。
GraphDBにデータを正しく登録するために
上記内容から分かるように、GraphDBを使うということは、ベクトルDBみたいに「ドキュメントから読み込んだ文章をそのまま登録」というわけにはいかなくなります。つまり、ドキュメントデータをどのような構造に変換して保存するかを、開発者が事前に考える必要があります。加えて、登録プロセスをできるだけLLMに任せたいのであれば、LLMが正しいCypherクエリを生成するためのプロンプトエンジニアリングも重要になってきます。
最初に人事部向けのツールを開発しているという話をしましたが、例えばエンジニアのスキルシートを検索できるようにしたいとします。
画像のようなドキュメントを何回読んでも、LLMが点と線の名前を間違えずにGraphDBに登録できるようにしたいですね。その場合、通常の文章のような回答よりも、決まった形式の出力を得てその出力からCypherのクエリーが生成できれば効率的なデータの登録が可能になります。
例えばChatGPTのStructured Outputsを使った実装例を見てみましょう。
model = 'gpt-4o-mini'
client = openai.OpenAI(api_key=OPENAI_API_KEY)
response_format = {
"type": "json_schema",
"json_schema": {
"name": "skillsheet_preformat_response",
"schema": {
"type": "object",
"properties": {
"name": {
"type": "string",
},
"skills": {
"type": "array",
"items": {
"type": "string",
},
},
"certificates": {
"type": "array",
"items": {
"type": "string",
},
},
},
"additionalProperties": False,
"required": ["name", "skills", "certificates"],
},
"strict": True,
},
}
def get_skillsheet_preformat_response(message_content: str) -> dict:
response = client.chat.completions.create(
model=model,
messages=[
{
"role": "system",
"content": "エンジニアのスキルシートを渡されたら、その内容から以下の情報を抽出してください。\n\n1.名前\n2. スキル(環境・言語)\n3. 取得資格\n\nスキルを`skills`、資格を`certificates`として返してください。スキルシートにある内容以外は書かないでください。",
},
{
"role": "user",
"content": message_content,
},
],
response_format=response_format,
)
return response
変換プロンプトの準備ができたら、PDFを読み込んでLLMに投げてみます。
file_list = [
"docs/skillsheets/001 - エンジニアスキルシート.pdf"
]
def load_file(file_path: str) -> str:
pdf_loader = PyPDFLoader(file_path)
documents = pdf_loader.load()
text_content = " ".join([doc.page_content for doc in documents])
return text_content
for file_path in file_list:
text_content = load_file(file_path)
response = get_skillsheet_preformat_response(text_content)
data = json.loads(response.choices[0].message.content)
pprint.pprint(data)
出力：
{'certificates': ['応用情報技術者試験', 'Oracle認定Javaプログラマ'],
'name': '001-山田 太郎',
'skills': ['Java',
'C++',
'Windows Server',
'Linux',
'Spring',
'Hibernate',
'PostgreSQL',
'Qt',
'SQLite']}
これで決まった形式のJSONが手に入るので、あとはCypherクエリーを生成する関数を用意します：
from neo4j import GraphDatabase
class GraphHandler:
def __init__(self, uri, user, password):
self.driver = GraphDatabase.driver(uri, auth=(user, password))
def close(self):
self.driver.close()
def append_skills_and_certificates(self, engineer_name, skills, certificates):
with self.driver.session() as session:
session.execute_write(self._append_data, engineer_name, skills, certificates)
@staticmethod
def _append_data(tx, engineer_name, skills, certificates):
query = """
MERGE (e:Engineer {name: $engineer_name})
WITH e
UNWIND $skills AS skill
MERGE (s:Skill {name: skill})
MERGE (e)-[:HAS_SKILL]->(s)
WITH e
UNWIND $certificates AS certificate
MERGE (c:Certificate {name: certificate})
MERGE (e)-[:HAS_CERTIFICATE]->(c)
"""
tx.run(query, engineer_name=engineer_name, skills=skills, certificates=certificates)
def load_file(file_path: str) -> str:
pdf_loader = PyPDFLoader(file_path)
documents = pdf_loader.load()
text_content = " ".join([doc.page_content for doc in documents])
return text_content
graph_handler = GraphHandler("bolt://localhost:7687", "neo4j", "password")
file_list = [
"docs/skillsheets/001 - エンジニアスキルシート.pdf"
"docs/skillsheets/002 - エンジニアスキルシート.pdf"
"docs/skillsheets/003 - エンジニアスキルシート.pdf"
"docs/skillsheets/004 - エンジニアスキルシート.pdf"
"docs/skillsheets/005 - エンジニアスキルシート.pdf"
"docs/skillsheets/006 - エンジニアスキルシート.pdf"
]
for file_path in file_list:
text_content = load_file(file_path)
response = get_skillsheet_preformat_response(text_content)
data = json.loads(response.choices[0].message.content)
graph_handler.append_skills_and_certificates(data["name"], data["skills"], data["certificates"])
graph_handler.close()
Neo4jに登録されるデータ：
このように、ドキュメント登録時のデータ構造にブレがないようにしてデータを貯めていきます。
GraphDBからデータを正しく検索するために
今度は、ユーザーに聞かれた質問を基に検索クエリーを生成したいので、先ほどと同じくStructured Outputsを使って、例えば案件に必要になりそうな「スキル」の配列を生成します。
load_dotenv()
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
model = 'gpt-4o-mini'
client = openai.OpenAI(api_key=OPENAI_API_KEY)
response_format = {
"type": "json_schema",
"json_schema": {
"name": "job_description_preformat_response",
"schema": {
"type": "object",
"properties": {
"skills": {
"type": "array",
"items": {
"type": "string",
},
},
},
"additionalProperties": False,
"required": ["skills"],
},
"strict": True,
},
}
def get_job_description_preformat_response(message_content: str) -> dict:
response = client.chat.completions.create(
model=model,
messages=[
{
"role": "system",
"content": "エンジニアの求人情報を渡されたら、その内容から以下の情報を抽出してください。\n\n1. 必要なスキル(環境・言語)\n\nスキルを`skills`として返してください。求人情報にある内容以外は書かないでください。",
},
{
"role": "user",
"content": message_content,
},
],
response_format=response_format,
)
return response
なお、少しでもマッチしそうな候補者を集めて、最終的にどちらを提案するかをLLMに考えさせるので、Cypherでの条件を「必要なスキルを一つでも持っていたら」とします。先ほどの GraphHandler に以下を追加します：
def search_skills(self, skills):
with self.driver.session() as session:
return session.execute_read(self._search_skills, skills)
@staticmethod
def _search_skills(tx, skills):
query = """
MATCH (e:Engineer)-[:HAS_SKILL]->(s:Skill)
WHERE s.name IN $skills
WITH e
MATCH (e)-[:HAS_SKILL]->(allSkills:Skill)
RETURN e.name AS engineer, collect(DISTINCT allSkills.name) AS all_skills
"""
result = tx.run(query, skills=skills)
return [{"engineer": record["engineer"], "all_skills": record["all_skills"]} for record in result]
以下の案件概要を投げてみましょう。
job_description = """
主な業務内容
Reactを用いたウェブアプリケーションのUI/UX設計と実装
RESTful APIおよびGraphQLバックエンドとの連携
axiosを用いたAPI通信の実装と最適化
trpcを利用したタイピングセーフなAPIインターフェースの開発
コードのレビューおよび最適化を通じた品質の向上
ユーザーストーリーとテスト仕様に基づいた継続的な機能追加と改善
応募資格
フロントエンド開発の実務経験2年以上
Reactを用いた開発経験およびスキル
axiosを利用したHTTP通信の実装知識
trpcによるAPI開発経験があると尚良い
モジュール化と再利用可能なコンポーネント設計に精通していること
ユーザーインターフェースの設計、アクセシビリティに関する知識
Gitを用いたバージョン管理のスキル
チームでの開発経験およびコミュニケーションスキル
歓迎するスキル
TypeScriptを用いた静的型付けに関する理解
その他のフロントエンド技術スタック（例：Redux、Next.js）に関する知識
UI/UXデザインツール（Figma、Adobe XDなど）を用いた経験
継続的インテグレーションおよびデプロイメント（CI/CD）パイプラインの経験
"""
response = get_job_description_preformat_response(job_description)
data = json.loads(response.choices[0].message.content)
pprint.pprint(data)
graph_handler = GraphHandler("bolt://localhost:7687", "neo4j", "password")
engineer_list = graph_handler.search_skills(data["skills"])
pprint.pprint(engineer_list)
graph_handler.close()
まず、案件概要から出力したキーワードはこちら。
{'skills': ['React',
'RESTful API',
'GraphQL',
'axios',
'trpc',
'TypeScript',
'Git',
'Redux',
'Next.js',
'Figma',
'Adobe XD']}
これで検索をかけて、見つけたエンジニアとそれぞれのスキルリストはこちら。
[{'all_skills': ['JavaScript',
'React',
'Redux',
'TypeScript',
'D3.js',
'Angular',
'MacOS',
'Windows'],
'engineer': '鈴木 健'}]
登録したデータの中に、フロントエンドエンジニアの経験があったのは鈴木さんだけだったのでバッチリですね。他にPythonを使ったバックエンド開発ならどうでしょう？
job_description = """
主な業務内容
Flaskを用いたウェブアプリケーションおよびAPIの設計と実装
セキュリティ、パフォーマンス、スケーラビリティを考慮したシステム開発
ユニットテストおよび統合テストの実装によるコード品質の確保
システムモニタリングとログ管理による運用支援
コードレビューやペアプログラミングを通じたチームの成長促進
応募資格
バックエンド開発の実務経験3年以上
PythonおよびFlaskを用いた開発経験
RESTful APIの設計および開発経験
Docker等を用いたコンテナ化の基本的な理解
チームでの開発経験およびコミュニケーションスキル
歓迎するスキル
クラウドプラットフォーム（AWS, Google Cloud, Azure等）での開発経験
メッセージキュー (RabbitMQ, Kafka) の使用経験
DevOpsの基本的な知識とCI/CDの経験
Webセキュリティに関する知識
"""
response = get_job_description_preformat_response(job_description)
data = json.loads(response.choices[0].message.content)
pprint.pprint(data)
graph_handler = GraphHandler("bolt://localhost:7687", "neo4j", "password")
engineer_list = graph_handler.search_skills(data["skills"])
pprint.pprint(engineer_list)
graph_handler.close()
出力結果：
[{'all_skills': ['Python',
'R',
'Ubuntu',
'CentOS',
'TensorFlow',
'scikit-learn',
'Pandas',
'R Shiny',
'MongoDB',
'MySQL'],
'engineer': '佐藤 花子'},
{'all_skills': ['Python',
'Shell Script',
'Bash',
'Kali Linux',
'C',
'Embedded Linux',
'Nessus',
'Metasploit',
'Wireshark',
'OpenVAS'],
'engineer': '李 小龍'}]
Pythonの経験があったのはこの２人だけで、Neo4jの中身を確認しても「Python」につながっているのはこの２人だけなのでデータの登録も検索も問題ないことが分かりますね。
あとはいつも通りのRAG
ここまでの流れが整理できたら、いつものように作った文脈をLLMに投げて質問に答えてもらいます。
response = get_job_description_preformat_response(job_description)
data = json.loads(response.choices[0].message.content)
graph_handler = GraphHandler("bolt://localhost:7687", "neo4j", "password")
engineer_list = graph_handler.search_skills(data["skills"])
graph_handler.close()
prompt_template = """
案件概要:
{job_description}
エンジニアリスト:
{context}
""".format(context="\n".join([f"{engineer['engineer']} - {', '.join(engineer['all_skills'])}" for engineer in engineer_list]), job_description=job_description)
response = client.chat.completions.create(
model=model,
messages=[
{
"role": "system",
"content": "案件概要と、必要なスキルを一つでも持っているエンジニアを渡しますので、案件に合いそうなエンジニアを提案してください。そのエンジニアの全体的なスキルを見た上で、どれくらい案件に合っているかも表現してください。スキルシートの詳細は人事部が確認するので、できるだけポジティブに紹介してあげてください。",
},
{
"role": "user",
"content": prompt_template,
},
],
)
print(response.choices[0].message.content)
回答：
ご提案させていただくのは、佐藤花子さんです。彼女はPythonのスキルを持っており、バックエンド開発に必要なプログラミング言語の経験があります。また、MySQLの使用経験もあるため、データベースとの連携においても適応可能です。
案件の要件においては、Flaskの具体的な経験が記載されていませんが、Pythonを用いた開発経験がある点が重要です。また、チームでの協業においても、良好なコミュニケーションが期待できます。さらに、さまざまなデータ分析や機械学習の知識を持っているため、システムのパフォーマンスやセキュリティ面でも新しい視点を提供できるかもしれません。
全体的に見て、佐藤さんは案件のスキル要件を一部満たしており、チームにとって有益な存在となり得ると思います。適応力が高い彼女なら、さらに成長しながらプロジェクトに貢献できるでしょう。
まとめ・最後に
GraphDBでのRAGの実装例、如何だったでしょうか？
この記事を通じて、GraphDBを使うハードルが下がるエンジニアが一人でも増えれば嬉しいです。
RAGにおけるGraphDBの活用方法は多岐にわたります。例えば、ベクトルDBでの実装と同様に、ドキュメント内容をチャンクに分割し、チャンクそのものをノードとして扱うアプローチがあります。また、ベクトルDBとGraphDBの両方を使って文脈をさらに広げたり、データの関係性をより深くLLMに理解させたりすることも可能です。
もう一度言いますがRAG or GraphRAGどちらかに振り切る必要はありません。様々な技術を組み合わせることでより効果的なソリューションが見つかることがあるので、柔軟に新しいものを試していくのみです。
DIVXも新しいアプローチを見つけるたびに報告をしていきますのでお楽しみにしていただければと思います。
最後までお読みいただき、ありがとうございました！
お悩みご相談ください
