テキスト埋め込みを超えて: 構造化データクエリの RAG アプリケーションのギャップに対処する

誰もがテキスト埋め込みモデルを好みますが、それには理由があります。テキスト埋め込みモデルは非構造化テキストのエンコードに優れており、意味的に類似したコンテンツを見つけやすくなります。特に現在は、ドキュメントやその他のテキスト リソースから関連情報をエンコードして取得することに重点が置かれているため、ほとんどの RAG アプリケーションのバックボーンとしてテキスト埋め込みモデルが採用されているのは当然のことです。ただし、RAG アプリケーションに対するテキスト埋め込みアプローチが不十分で、誤った情報を提供するという疑問が明確に生じます。

前述のように、テキスト埋め込みは非構造化テキストのエンコードに最適です。一方、構造化情報やフィルタリング、並べ替え、集計などの操作の処理にはそれほど適していません。次のような簡単な質問を想像してみてください。

2024年に公開される映画の中で最も評価が高い映画は何ですか?

この質問に答えるには、まずリリース年でフィルタリングし、次に評価で並べ替える必要があります。テキスト埋め込みを使用した単純なアプローチがどのように機能するかを調べ、次にこのような質問に対処する方法を説明します。このブログ投稿では、フィルタリング、並べ替え、集約などの構造化データ操作を処理する場合は、ナレッジ グラフなどの構造を提供する他のツールを使用する必要があることを示しています。

コードはGitHubで入手できます。

環境設定

このブログ投稿では、 Neo4j Sandbox の推奨プロジェクトを使用します。推奨プロジェクトは、映画、俳優、評価などの情報が含まれるMovieLens データセットを使用します。

次のコードは、LangChain ラッパーをインスタンス化して Neo4j データベースに接続します。

 os.environ["NEO4J_URI"] = "bolt://44.204.178.84:7687" os.environ["NEO4J_USERNAME"] = "neo4j" os.environ["NEO4J_PASSWORD"] = "minimums-triangle-saving" graph = Neo4jGraph(refresh_schema=False)

さらに、次のコードで渡す OpenAI API キーが必要になります。

 os.environ["OPENAI_API_KEY"] = getpass.getpass("OpenAI API Key:")

データベースには 10,000 本の映画が含まれていますが、テキスト埋め込みはまだ保存されていません。すべての映画の埋め込みを計算しないように、評価の高い 1,000 本の映画にTargetという 2 番目のラベルを付けます。

 graph.query(""" MATCH (m:Movie) WHERE m.imdbRating IS NOT NULL WITH m ORDER BY m.imdbRating DESC LIMIT 1000 SET m:Target """)

テキスト埋め込みの計算と保存

何を埋め込むかを決めることは重要な考慮事項です。年によるフィルタリングと評価による並べ替えのデモンストレーションを行うため、埋め込みテキストからこれらの詳細を除外するのは公平ではありません。そのため、各映画の公開年、評価、タイトル、説明をキャプチャすることにしました。

以下は、映画「ウルフ・オブ・ウォールストリート」に埋め込むテキストの例です。

 plot: Based on the true story of Jordan Belfort, from his rise to a wealthy stock-broker living the high life to his fall involving crime, corruption and the federal government. title: Wolf of Wall Street, The year: 2013 imdbRating: 8.2

これは構造化データを埋め込むための良い方法ではないと言う人もいるかもしれませんが、私は最善の方法を知らないので、異論は唱えません。おそらく、キー値項目の代わりに、テキストなどに変換する必要があります。もっと良い方法について何かアイデアがあれば教えてください。

LangChain の Neo4j Vector オブジェクトには、エンコードするテキスト プロパティを選択できる便利なメソッド from_existing_graph があります。

 embedding = OpenAIEmbeddings(model="text-embedding-3-small") neo4j_vector = Neo4jVector.from_existing_graph( embedding=embedding, index_name="movies", node_label="Target", text_node_properties=["plot", "title", "year", "imdbRating"], embedding_node_property="embedding", )

この例では、埋め込み生成に OpenAI の text-embedding-3-small モデルを使用します。from_existing_graph メソッドを使用して Neo4jVector オブジェクトを初期化します。node_label パラメータは、エンコードするノード、具体的にはTargetというラベルの付いたノードをフィルタリングします。text_node_properties パラメータは、 plot 、 title 、 year 、 imdbRatingなど、埋め込まれるノードのプロパティを定義します。最後に、embedding_node_property は、生成された埋め込みが保存されるプロパティを定義し、 embeddingとして指定します。

素朴なアプローチ

まず、映画のあらすじや説明に基づいて映画を探してみましょう。

 pretty_print( neo4j_vector.similarity_search( "What is a movie where a little boy meets his hero?" ) )

結果：

 plot: A young boy befriends a giant robot from outer space that a paranoid government agent wants to destroy. title: Iron Giant, The year: 1999 imdbRating: 8.0 plot: After the death of a friend, a writer recounts a boyhood journey to find the body of a missing boy. title: Stand by Me year: 1986 imdbRating: 8.1 plot: A young, naive boy sets out alone on the road to find his wayward mother. Soon he finds an unlikely protector in a crotchety man and the two have a series of unexpected adventures along the way. title: Kikujiro (Kikujirô no natsu) year: 1999 imdbRating: 7.9 plot: While home sick in bed, a young boy's grandfather reads him a story called The Princess Bride. title: Princess Bride, The year: 1987 imdbRating: 8.1

結果は全体的にかなりしっかりしているようです。小さな男の子が常に登場しますが、彼がいつもヒーローに会うかどうかはわかりません。とはいえ、データセットには 1,000 本の映画しか含まれていないので、選択肢はやや限られています。

次に、基本的なフィルタリングを必要とするクエリを試してみましょう。

 pretty_print( neo4j_vector.similarity_search( "Which movies are from year 2016?" ) )

結果：

 plot: Six short stories that explore the extremities of human behavior involving people in distress. title: Wild Tales year: 2014 imdbRating: 8.1 plot: A young man who survives a disaster at sea is hurtled into an epic journey of adventure and discovery. While cast away, he forms an unexpected connection with another survivor: a fearsome Bengal tiger. title: Life of Pi year: 2012 imdbRating: 8.0 plot: Based on the true story of Jordan Belfort, from his rise to a wealthy stock-broker living the high life to his fall involving crime, corruption and the federal government. title: Wolf of Wall Street, The year: 2013 imdbRating: 8.2 plot: After young Riley is uprooted from her Midwest life and moved to San Francisco, her emotions - Joy, Fear, Anger, Disgust and Sadness - conflict on how best to navigate a new city, house, and school. title: Inside Out year: 2015 imdbRating: 8.3

面白いことに、2016 年の映画は 1 本も選ばれませんでした。エンコード用にテキストを準備することで、より良い結果が得られるかもしれません。ただし、ここでは、メタデータ プロパティに基づいてドキュメント (この例では映画) をフィルターする必要がある単純な構造化データ操作を扱っているため、テキスト埋め込みは適用できません。メタデータ フィルタリングは、RAG システムの精度を高めるためによく使用される確立された手法です。

次に試すクエリでは、少し並べ替えが必要です。

 pretty_print( neo4j_vector.similarity_search("Which movie has the highest imdb score?") )

結果：

 plot: A silent film production company and cast make a difficult transition to sound. title: Singin' in the Rain year: 1952 imdbRating: 8.3 plot: A film about the greatest pre-Woodstock rock music festival. title: Monterey Pop year: 1968 imdbRating: 8.1 plot: This movie documents the Apollo missions perhaps the most definitively of any movie under two hours. Al Reinert watched all the footage shot during the missions--over 6,000,000 feet of it, ... title: For All Mankind year: 1989 imdbRating: 8.2 plot: An unscrupulous movie producer uses an actress, a director and a writer to achieve success. title: Bad and the Beautiful, The year: 1952 imdbRating: 7.9

IMDb の評価に詳しい方なら、8.3 を超えるスコアを獲得した映画がたくさんあることをご存知でしょう。私たちのデータベースで最も評価の高いタイトルは、実はシリーズ作品「バンド・オブ・ブラザース」で、9.6 という素晴らしい評価を得ています。繰り返しになりますが、テキスト埋め込みは、結果の並べ替えに関してはパフォーマンスが劣っています。

何らかの集計を必要とする質問も評価してみましょう。

 pretty_print(neo4j_vector.similarity_search("How many movies are there?"))

結果：

 plot: Ten television drama films, each one based on one of the Ten Commandments. title: Decalogue, The (Dekalog) year: 1989 imdbRating: 9.2 plot: A documentary which challenges former Indonesian death-squad leaders to reenact their mass-killings in whichever cinematic genres they wish, including classic Hollywood crime scenarios and lavish musical numbers. title: Act of Killing, The year: 2012 imdbRating: 8.2 plot: A meek Hobbit and eight companions set out on a journey to destroy the One Ring and the Dark Lord Sauron. title: Lord of the Rings: The Fellowship of the Ring, The year: 2001 imdbRating: 8.8 plot: While Frodo and Sam edge closer to Mordor with the help of the shifty Gollum, the divided fellowship makes a stand against Sauron's new ally, Saruman, and his hordes of Isengard. title: Lord of the Rings: The Two Towers, The year: 2002 imdbRating: 8.7

ここでは、ランダムに 4 本の映画が返されるため、結果はまったく役に立ちません。これらのランダムに 4 本の映画から、この例のためにタグ付けして埋め込んだ映画が合計 1,000 本あるという結論を導き出すことは、事実上不可能です。

では、解決策は何でしょうか? それは簡単です。フィルタリング、並べ替え、集計などの構造化された操作を伴う質問には、構造化されたデータで動作するように設計されたツールが必要です。

構造化データのためのツール

現時点では、ほとんどの人が text2query アプローチについて考えているようです。このアプローチでは、LLM がデータベース クエリを生成し、提供された質問とスキーマに基づいてデータベースと対話します。Neo4j の場合、これは text2cypher ですが、SQL データベース用の text2sql もあります。ただし、実際には信頼性が低く、実稼働での使用には十分な堅牢性がないことが判明しています。

Cypher ステートメント生成の評価。Cypher 評価に関する私のブログ投稿から抜粋。

思考の連鎖、少数の例、微調整などのテクニックを使うこともできますが、この段階では高い精度を達成することはほぼ不可能です。text2query アプローチは、単純なデータベース スキーマに関する単純な質問にはうまく機能しますが、実稼働環境ではそうではありません。これに対処するために、データベース クエリ生成の複雑さを LLM から切り離し、関数入力に基づいて決定論的にデータベース クエリを生成するコードの問題として扱います。利点は堅牢性が大幅に向上することですが、柔軟性が低下するという代償があります。すべての質問に答えようとして不正確な答えを出すよりも、RAG アプリケーションの範囲を狭めてそれらの質問に正確に答える方がよいでしょう。

関数入力に基づいてデータベース クエリ (この場合は Cypher ステートメント) を生成しているため、LLM のツール機能を活用できます。このプロセスでは、LLM がユーザー入力に基づいて関連パラメータを設定し、関数が必要な情報の取得を処理します。このデモでは、まず映画の数をカウントするツールと映画を一覧表示するツールの 2 つのツールを実装し、次に LangGraph を使用して LLM エージェントを作成します。

映画を数えるツール

まず、定義済みのフィルターに基づいて映画をカウントするツールを実装します。まず、それらのフィルターが何であるかを定義し、いつどのように使用するかを LLM に説明する必要があります。

 class MovieCountInput(BaseModel): min_year: Optional[int] = Field( description="Minimum release year of the movies" ) max_year: Optional[int] = Field( description="Maximum release year of the movies" ) min_rating: Optional[float] = Field(description="Minimum imdb rating") grouping_key: Optional[str] = Field( description="The key to group by the aggregation", enum=["year"] )

LangChain では関数の入力を定義する方法がいくつか提供されていますが、私は Pydantic アプローチを好みます。この例では、映画の結果を絞り込むために、min_year、max_year、min_rating の 3 つのフィルターが用意されています。これらのフィルターは構造化データに基づいており、オプションです。ユーザーはフィルターのいずれか、すべて、またはいずれも含めないことを選択できます。さらに、関数に特定のプロパティでカウントをグループ化するかどうかを指示する grouping_key 入力を導入しました。この場合、enumsection で定義されているように、サポートされているグループ化は年によるグループ化のみです。

それでは実際の関数を定義しましょう。

 @tool("movie-count", args_schema=MovieCountInput) def movie_count( min_year: Optional[int], max_year: Optional[int], min_rating: Optional[float], grouping_key: Optional[str], ) -> List[Dict]: """Calculate the count of movies based on particular filters""" filters = [ ("t.year >= $min_year", min_year), ("t.year <= $max_year", max_year), ("t.imdbRating >= $min_rating", min_rating), ] # Create the parameters dynamically from function inputs params = { extract_param_name(condition): value for condition, value in filters if value is not None } where_clause = " AND ".join( [condition for condition, value in filters if value is not None] ) cypher_statement = "MATCH (t:Target) " if where_clause: cypher_statement += f"WHERE {where_clause} " return_clause = ( f"t.`{grouping_key}`, count(t) AS movie_count" if grouping_key else "count(t) AS movie_count" ) cypher_statement += f"RETURN {return_clause}" print(cypher_statement) # Debugging output return graph.query(cypher_statement, params=params)

movie_count 関数は、オプションのフィルターとグループ化キーに基づいて映画をカウントする Cypher クエリを生成します。まず、対応する値を引数として指定したフィルターのリストを定義します。フィルターは、値が None ではない条件のみを含む、Cypher ステートメントで指定されたフィルタリング条件を適用する WHERE 句を動的に構築するために使用されます。

次に、Cypher クエリの RETURN 句が構築され、指定された grouping_key でグループ化するか、または単に映画の合計数をカウントします。最後に、関数はクエリを実行し、結果を返します。

関数は、必要に応じて、より多くの引数とより複雑なロジックを使用して拡張できますが、LLM が正しく正確に呼び出せるように、関数が明確なままであることを確認することが重要です。

映画をリストするツール

ここでも、関数の引数を定義することから始める必要があります。

 class MovieListInput(BaseModel): sort_by: str = Field( description="How to sort movies, can be one of either latest, rating", enum=["latest", "rating"], ) k: Optional[int] = Field(description="Number of movies to return") description: Optional[str] = Field(description="Description of the movies") min_year: Optional[int] = Field( description="Minimum release year of the movies" ) max_year: Optional[int] = Field( description="Maximum release year of the movies" ) min_rating: Optional[float] = Field(description="Minimum imdb rating")

映画カウント関数と同じ 3 つのフィルターを保持しますが、説明引数を追加します。この引数により、ベクトル類似性検索を使用して、プロットに基づいて映画を検索してリストできます。構造化されたツールとフィルターを使用しているからといって、テキスト埋め込みとベクトル検索メソッドを組み込むことができないわけではありません。ほとんどの場合、すべての映画を返す必要はないため、デフォルト値を持つオプションの k 入力を含めます。さらに、リスト表示では、最も関連性の高い映画のみを返すように映画を並べ替えます。この場合、評価または公開年で並べ替えることができます。

次の関数を実装してみましょう。

 @tool("movie-list", args_schema=MovieListInput) def movie_list( sort_by: str = "rating", k : int = 4, description: Optional[str] = None, min_year: Optional[int] = None, max_year: Optional[int] = None, min_rating: Optional[float] = None, ) -> List[Dict]: """List movies based on particular filters""" # Handle vector-only search when no prefiltering is applied if description and not min_year and not max_year and not min_rating: return neo4j_vector.similarity_search(description, k=k) filters = [ ("t.year >= $min_year", min_year), ("t.year <= $max_year", max_year), ("t.imdbRating >= $min_rating", min_rating), ] # Create parameters dynamically from function arguments params = { key.split("$")[1]: value for key, value in filters if value is not None } where_clause = " AND ".join( [condition for condition, value in filters if value is not None] ) cypher_statement = "MATCH (t:Target) " if where_clause: cypher_statement += f"WHERE {where_clause} " # Add the return clause with sorting cypher_statement += " RETURN t.title AS title, t.year AS year, t.imdbRating AS rating ORDER BY " # Handle sorting logic based on description or other criteria if description: cypher_statement += ( "vector.similarity.cosine(t.embedding, $embedding) DESC " ) params["embedding"] = embedding.embed_query(description) elif sort_by == "rating": cypher_statement += "t.imdbRating DESC " else: # sort by latest year cypher_statement += "t.year DESC " cypher_statement += " LIMIT toInteger($limit)" params["limit"] = k or 4 print(cypher_statement) # Debugging output data = graph.query(cypher_statement, params=params) return data

この関数は、説明、年の範囲、最低評価、並べ替え設定など、複数のオプション フィルターに基づいて映画のリストを取得します。他のフィルターなしで説明のみが指定されている場合は、ベクター インデックス類似性検索を実行して関連する映画を検索します。追加のフィルターが適用されると、関数は、リリース年や IMDb 評価などの指定された基準に基づいて映画を一致させる Cypher クエリを作成し、オプションの説明ベースの類似性と組み合わせます。結果は、類似性スコア、IMDb 評価、または年のいずれかで並べ替えられ、 k 本の映画に制限されます。

すべてをLangGraphエージェントとしてまとめる

LangGraph を使用して、簡単なReActエージェントを実装します。

エージェントは LLM とツールのステップで構成されています。エージェントと対話する際、まず LLM を呼び出してツールを使用するかどうかを決定します。次にループを実行します。

1. エージェントがアクションを実行するように指示した場合 (つまり、ツールを呼び出す場合)、ツールを実行し、結果をエージェントに返します。
2. エージェントがツールの実行を要求しなかった場合は、終了します (ユーザーに応答します)。

コードの実装は非常に簡単です。まず、ツールを LLM にバインドし、アシスタント ステップを定義します。

 llm = ChatOpenAI(model='gpt-4-turbo') tools = [movie_count, movie_list] llm_with_tools = llm.bind_tools(tools) # System message sys_msg = SystemMessage(content="You are a helpful assistant tasked with finding and explaining relevant information about movies.") # Node def assistant(state: MessagesState): return {"messages": [llm_with_tools.invoke([sys_msg] + state["messages"])]}

次に、LangGraph フローを定義します。

 # Graph builder = StateGraph(MessagesState) # Define nodes: these do the work builder.add_node("assistant", assistant) builder.add_node("tools", ToolNode(tools)) # Define edges: these determine how the control flow moves builder.add_edge(START, "assistant") builder.add_conditional_edges( "assistant", # If the latest message (result) from assistant is a tool call -> tools_condition routes to tools # If the latest message (result) from assistant is a not a tool call -> tools_condition routes to END tools_condition, ) builder.add_edge("tools", "assistant") react_graph = builder.compile()

LangGraph に 2 つのノードを定義し、条件付きエッジでそれらをリンクします。ツールが呼び出されると、フローはそのツールに向けられ、それ以外の場合は結果がユーザーに返されます。

それではエージェントをテストしてみましょう。

 messages = [ HumanMessage( content="What are the some movies about a girl meeting her hero?" ) ] messages = react_graph.invoke({"messages": messages}) for m in messages["messages"]: m.pretty_print()

結果：

最初のステップでは、エージェントは適切な説明パラメータを使用して映画リスト ツールを使用することを選択します。k 値として 5 を選択する理由は不明ですが、その数値が好まれるようです。ツールはプロットに基づいて最も関連性の高い上位 5 つの映画を返し、LLM は最後にそれらをユーザー向けに要約します。

ChatGPT に k 値が 5 である理由を尋ねると、次の応答が返されます。

次に、メタデータのフィルタリングを必要とする、もう少し複雑な質問をしてみましょう。

 messages = [ HumanMessage( content="What are the movies from the 90s about a girl meeting her hero?" ) ] messages = react_graph.invoke({"messages": messages}) for m in messages["messages"]: m.pretty_print()

結果：

今回は、追加の引数を使用して、1990 年代の映画のみをフィルタリングしました。この例は、事前フィルタリング アプローチを使用したメタデータ フィルタリングの典型的な例です。生成された Cypher ステートメントは、最初に公開年でフィルタリングして映画を絞り込みます。次の部分では、Cypher ステートメントはテキスト埋め込みとベクトル類似性検索を使用して、小さな女の子がヒーローに出会う映画を検索します。

さまざまな条件に基づいて映画を数えてみましょう。

 messages = [ HumanMessage( content="How many movies are from the 90s have the rating higher than 9.1?" ) ] messages = react_graph.invoke({"messages": messages}) for m in messages["messages"]: m.pretty_print()

結果：

カウント専用のツールを使用すると、複雑さは LLM からツールに移行し、LLM は関連する関数パラメータを入力することだけを担当するようになります。このタスクの分離により、システムの効率と堅牢性が向上し、LLM 入力の複雑さが軽減されます。

エージェントは複数のツールを順番にまたは並行して呼び出すことができるため、さらに複雑なものでテストしてみましょう。

 messages = [ HumanMessage( content="How many were movies released per year made after the highest rated movie?" ) ] messages = react_graph.invoke({"messages": messages}) for m in messages["messages"]: m.pretty_print()

結果

前述のように、エージェントは複数のツールを呼び出して、質問に答えるために必要なすべての情報を収集できます。この例では、まず評価の高い映画をリストして、評価の高い映画がいつ公開されたかを特定します。そのデータを取得すると、映画カウント ツールを呼び出して、質問で定義されたグループ化キーを使用して、指定された年以降に公開された映画の数を収集します。

まとめ

テキスト埋め込みは非構造化データの検索には優れていますが、フィルタリング、並べ替え、集計などの構造化操作には不十分です。これらのタスクには、操作の処理に必要な精度と柔軟性を備えた、構造化データ用に設計されたツールが必要です。重要な点は、システム内のツール セットを拡張することで、より幅広いユーザー クエリに対応できるようになり、アプリケーションの堅牢性と汎用性が向上することです。構造化データ アプローチと非構造化テキスト検索手法を組み合わせることで、より正確で関連性の高い応答が得られ、最終的に RAG アプリケーションでのユーザー エクスペリエンスが向上します。

いつものように、コードはGitHubで入手できます。

このトピックについて詳しく知るには、11 月 7 日に開催されるインテリジェント アプリ、ナレッジ グラフ、AI に関する無料の仮想開発者会議 NODES 2024 にご参加ください。今すぐ登録してください。