大規模言語モデルによるナレッジグラフ自動化の概念検証[4]

import json
import boto3
from dotenv import load_dotenv
import os
from neo4j import GraphDatabase
import glob

# 環境変数の読み込み
load_dotenv()

# AWS Bedrock初期化
bedrock_runtime = boto3.client(
    service_name='bedrock-runtime',
    region_name=os.getenv("AWS_REGION", "ap-northeast-1"),
    aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
    aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY")
)

# Neo4j初期化
neo4j_driver = GraphDatabase.driver(
    "bolt://localhost:7687", 
    auth=("neo4j", os.getenv('NEO4J_PASSWORD'))
)

def generate_embedding(text, dimensions=None):
    """テキストからエンベディングを生成 (Titan Text Embeddings V2)"""
    # 入力検証
    if not text or not isinstance(text, str):
        raise ValueError(f"Invalid text input: {repr(text)}")
    
    # 空白のみの文字列をチェック
    if not text.strip():
        raise ValueError(f"Text is empty or whitespace only: {repr(text)}")
    
    # リクエストボディ (シンプル版 - normalizeを削除)
    body_dict = {"inputText": text}
    
    # 次元数指定がある場合のみ追加
    if dimensions:
        body_dict["dimensions"] = dimensions
    
    body = json.dumps(body_dict)
    
    try:
        response = bedrock_runtime.invoke_model(
            modelId='amazon.titan-embed-text-v2:0',
            body=body
        )
        
        response_body = json.loads(response['body'].read())
        return response_body['embedding']
    except Exception as e:
        print(f"ERROR generating embedding for text: {text[:100]}...")
        print(f"Dimensions: {dimensions}")
        raise e

def add_embeddings_from_json(json_file_path):
    """JSONからデータを取得して既存グラフにエンベディングを追加"""
    try:
        # グラフデータを読む
        with open(json_file_path, 'r', encoding='utf-8') as f:
            data = json.load(f)
        
        with neo4j_driver.session() as session:
            # 1. BlogPostノードにエンベディング追加(1つのみ)
            blog_post_node = next((node for node in data['nodes'] if 'BlogPost' in node['labels']), None)
            
            if blog_post_node:
                # url(検索キー)とabstract(Embedding対象)を抽出
                url = blog_post_node['properties']['url']
                abstract = blog_post_node['properties'].get('abstract')
                
                if abstract:
                    # Embeddingを依頼 (デフォルト次元数を使用)
                    embedding = generate_embedding(abstract)
                    
                    # urlでノードを検索してabstract_embedding属性を追加
                    session.run("""
                        MATCH (bp:BlogPost {url: $url})
                        SET bp.abstract_embedding = $embedding
                    """, url=url, embedding=embedding)
                    
                    print(f"✓ BlogPost エンベディング追加: {blog_post_node['properties']['title']}")
            
            # 2. 全てのKeywordノードにエンベディング追加(複数のために)
            # 各Keywordノードを処理
            for node in data['nodes']:
                if 'Keyword' in node['labels']:
                    word = node['properties']['word']
                    
                    # Embeddingを依頼 (デフォルト次元数を使用)
                    embedding = generate_embedding(word)
                    
                    # ワードで検索し、word_embedding属性を追加
                    session.run("""
                        MATCH (k:Keyword {word: $word})
                        SET k.word_embedding = $embedding
                    """, word=word, embedding=embedding)
        
        return True
        
    except Exception as e:
        print(f"✗ エラー: {json_file_path} - {str(e)}")
        return False

def process_multiple_files(file_pattern):
    """複数のJSONファイルを処理"""
    # ワイルドカードでファイルを取得
    json_files = glob.glob(file_pattern)
    
    if not json_files:
        print(f"警告: パターン '{file_pattern}' に一致するファイルが見つかりません")
        return
    
    print(f"\n{len(json_files)} 個のファイルを処理します\n")
    
    success_count = 0
    fail_count = 0
    
    for idx, json_file in enumerate(json_files, 1):
        print(f"[{idx}/{len(json_files)}] 処理中: {os.path.basename(json_file)}")
        
        if add_embeddings_from_json(json_file):
            success_count += 1
        else:
            fail_count += 1
        
        print()  # 空行
    
    # サマリー表示
    print("=" * 50)
    print(f"処理完了: 成功 {success_count} / 失敗 {fail_count} / 合計 {len(json_files)}")
    print("=" * 50)

# メイン処理
if __name__ == "__main__":
    # 複数JSONファイルから取得して既存グラフにエンベディング追加
    process_multiple_files('../graph_extract/*.json')
    
    # クリーンアップ
    neo4j_driver.close()
    
    print("\nすべてのファイルの処理が完了しました!")

import json
import boto3
from dotenv import load_dotenv
import os
from neo4j import GraphDatabase

# 環境変数の読み込み
load_dotenv()

# AWS Bedrock初期化
bedrock_runtime = boto3.client(
    service_name='bedrock-runtime',
    region_name=os.getenv("AWS_REGION", "ap-northeast-1"),
    aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
    aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY")
)

# Neo4j初期化
neo4j_driver = GraphDatabase.driver(
    "bolt://localhost:7687", 
    auth=("neo4j", os.getenv('NEO4J_PASSWORD'))
)

def generate_embedding(text):
    """テキストからエンベディングを生成"""
    body = json.dumps({"inputText": text})
    
    response = bedrock_runtime.invoke_model(
        modelId='amazon.titan-embed-text-v2:0',
        body=body
    )
    
    response_body = json.loads(response['body'].read())
    return response_body['embedding']

def vector_search(query_text, top_k=3):
    """Vector検索を実行"""
    # 検索クエリのエンベディングを生成
    query_embedding = generate_embedding(query_text)
    
    with neo4j_driver.session() as session:
        # Vector検索クエリ
        result = session.run("""
            CALL db.index.vector.queryNodes('blog_post_embeddings', $top_k, $query_embedding)
            YIELD node, score
            RETURN node.title AS title, 
                   node.abstract AS abstract, 
                   node.url AS url,
                   score
            ORDER BY score DESC
        """, query_embedding=query_embedding, top_k=top_k)
        
        return list(result)

# メイン処理
if __name__ == "__main__":
    # 検索実行
    query = "Neo4jの新機能について紹介しているブログは"
    print(f"検索クエリ: {query}\n")
    print("=" * 80)
    
    results = vector_search(query, top_k=3)
    
    # 結果表示
    for idx, record in enumerate(results, 1):
        print(f"\n【結果 {idx}】(類似度スコア: {record['score']:.4f})")
        print(f"タイトル: {record['title']}")
        print(f"URL: {record['url']}")
        print(f"\nAbstract:")
        print(record['abstract'])
        print("-" * 80)
    
    # クリーンアップ
    neo4j_driver.close()
    
    print(f"\n検索完了: {len(results)} 件の結果")

import json
import boto3
from dotenv import load_dotenv
import os
from neo4j import GraphDatabase

# 環境変数の読み込み
load_dotenv()

# AWS Bedrock初期化
bedrock_runtime = boto3.client(
    service_name='bedrock-runtime',
    region_name=os.getenv("AWS_REGION", "ap-northeast-1"),
    aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
    aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY")
)

# Neo4j初期化
neo4j_driver = GraphDatabase.driver(
    "bolt://localhost:7687", 
    auth=("neo4j", os.getenv('NEO4J_PASSWORD'))
)

def generate_embedding(text):
    """テキストからエンベディングを生成"""
    body = json.dumps({"inputText": text})
    
    response = bedrock_runtime.invoke_model(
        modelId='amazon.titan-embed-text-v2:0',
        body=body
    )
    
    response_body = json.loads(response['body'].read())
    return response_body['embedding']

def vector_search(query_text, top_k=3):
    """Vector検索を実行"""
    # 検索クエリのエンベディングを生成
    query_embedding = generate_embedding(query_text)
    
    with neo4j_driver.session() as session:
        # Vector検索クエリ
        result = session.run("""
            CALL db.index.vector.queryNodes('blog_post_embeddings', $top_k, $query_embedding)
            YIELD node, score
            MATCH (node)-[:CONTAINS_KEYWORD]->(k:Keyword)
            RETURN node.title AS title, 
                   node.abstract AS abstract, 
                   node.url AS url,
                   collect(k.word) AS keyword,
                   score
            ORDER BY score DESC
        """, query_embedding=query_embedding, top_k=top_k)
        
        return list(result)

# メイン処理
if __name__ == "__main__":
    # 検索実行
    query = "Neo4jの新機能について紹介しているブログは"
    print(f"検索クエリ: {query}\n")
    print("=" * 80)
    
    results = vector_search(query, top_k=3)
    
    # 結果表示
    for idx, record in enumerate(results, 1):
        print(f"\n【結果 {idx}】(類似度スコア: {record['score']:.4f})")
        print(f"タイトル: {record['title']}")
        print(f"URL: {record['url']}")
        print(f"\nAbstract:")
        print(record['abstract'])
        print(f"\nキーワード:")
        print(record['keyword'])
        print("-" * 80)
    
    # クリーンアップ
    neo4j_driver.close()
    
    print(f"\n検索完了: {len(results)} 件の結果")