文本嵌入和向量搜索技術可以幫助我們根據文檔的含義及其相似性來檢索文檔。但當需要根據日期或類別等特定標準來篩選信息時，這些技術就顯得力不從心。為了解決這個問題，我們可以引入元數據過濾或過濾向量搜索，這允許我們根據用戶的特定需求來縮小搜索范圍。
圖片
例如，用戶可能想要了解 2021 年實施的新政策。通過使用元數據過濾器，系統可以先篩選出 2021 年的文檔，然后在這些文檔中執行向量相似性搜索，以找到與用戶興趣最相關的文檔。這種先進行元數據過濾再執行向量搜索的兩步策略，能夠顯著提高搜索的相關性和準確性。
近期，Neo4j 引入了基于節點屬性的 LangChAIn 元數據過濾支持。由于圖形數據庫能夠存儲復雜的結構化和非結構化數據，我們可以利用這些數據來執行更精細的元數據過濾。
圖片
以一個包含文章和組織信息的數據集為例，文章節點包含了文本和嵌入值，而與文章相關聯的組織節點則包含了日期、情感、作者等更多信息。通過這些信息，我們可以構建復雜的查詢，以回答如
- Rod Johnson 所在的公司是否實施了新的在家工作政策？
- Neo4j 投資的公司是否有負面新聞？
- 與為現代汽車供應的公司相關的供應鏈問題是否有任何值得注意的新聞？
等問題。
在本篇博客中，Tomaz Bratanic 將向我們展示如何結合 LangChain 和 OpenAI 函數調用代理來實現基于圖的元數據過濾。相關代碼已在 https://Github.com/tomasonjo/blogs/blob/master/llm/graph_based_prefiltering.ipynb 上提供。
概覽
我們將使用 Neo4j 托管的公共演示服務器上的 companies 圖數據集。您可以通過以下憑據訪問該數據集：
URI: https://demo.neo4jlabs.com:7473/browser/
用戶名: companies
密碼: companies
數據庫: companies
圖片
數據集的完整模式包括以 Organization 節點為中心的豐富信息，涵蓋供應商、競爭對手、位置、董事會成員等。此外，還有提及特定組織的文章及其相應的文本塊。
我們將實現一個 OpenAI 代理，它可以根據用戶輸入動態生成 Cypher 語句，并從圖形數據庫檢索相關文本塊。這個工具將提供四個可選輸入參數：
- 主題：用戶感興趣的特定信息或主題。
- 組織：用戶希望查詢信息的組織。
- 國家：用戶感興趣的組織的國家。
- 情感：文章的情感傾向。
我們將根據這些輸入參數動態構建相應的 Cypher 語句，從圖形數據庫檢索相關信息，并利用大型語言模型（LLM）生成最終答案。
要跟隨代碼實踐，您將需要一個 OpenAI API 密鑰。
功能實現
我們從設置 Neo4j 的連接憑證和相關連接開始。
import os
os.environ["OPENAI_API_KEY"] = "sk-"
os.environ["NEO4J_URI"] = "neo4j+s://demo.neo4jlabs.com"
os.environ["NEO4J_USERNAME"] = "companies"
os.environ["NEO4J_PASSword"] = "companies"
os.environ["NEO4J_DATABASE"] = "companies"
embeddings = OpenAIEmbeddings()
graph = Neo4jGraph()
vector_index = Neo4jVector.from_existing_index(
embeddings,
index_name="news"
)
我們使用 OpenAI 的文本嵌入技術，您需要一個 API 密鑰來使用它。接下來，我們定義了與 Neo4j 的連接，這使我們能夠執行任意的 Cypher 語句。最后，我們創建了一個 Neo4jVector 連接，它可以通過查詢現有的向量索引來檢索信息。目前，我們不能將向量索引與預過濾方法結合使用，只能與后過濾方法結合使用。但本文將專注于預過濾方法與全面向量相似性搜索的結合使用。
本文的核心是一個名為 get_organization_news 的函數，它能夠根據用戶的需求動態生成 Cypher 查詢語句并檢索相關信息。為了清晰起見，我將代碼分成了多個部分。
- 首先，我們定義了一組輸入參數，這些參數都是可選的文字輸入。特別地，topic 參數用來在文檔中搜索特定的信息。在實際應用中，我們會將 topic 參數的值用于向量相似性搜索。另外三個參數則用于展示預過濾的方法。如果所有預過濾參數都沒有提供，我們可以直接利用現有的向量索引來檢索相關文檔。如果提供了預過濾參數，我們會開始構建一個基礎的 Cypher 查詢語句，這個語句將用于后續的預過濾元數據方法。我們使用 CYPHER runtime = parallel parallelRuntimeSupport=all 指令來告訴 Neo4j 數據庫，在可能的情況下使用 并行運行時。然后，我們準備一個匹配語句來選擇 Chunk 節點和它們關聯的 Article 節點。
def get_organization_news(
topic: Optional[str] = None,
organization: Optional[str] = None,
country: Optional[str] = None,
sentiment: Optional[str] = None,
) -> str:
# 如果沒有預過濾條件，我們可以直接使用向量索引進行搜索
if topic and not organization and not country and not sentiment:
return vector_index.similarity_search(topic)
# 使用并行運行時（如果可用）
base_query = (
"CYPHER runtime = parallel parallelRuntimeSupport=all "
"MATCH (c:Chunk)<-[:HAS_CHUNK]-(a:Article) WHERE "
)
where_queries = []
params = {"k": 5} # 設置要檢索的文本塊數量
- 接下來，我們動態地向 Cypher 語句添加元數據過濾器。我們從 Organization 過濾器開始。
if organization:
# 將組織名稱映射到數據庫中的候選項
candidates = get_candidates(organization)
if len(candidates) > 1: # 如果候選選項太多，則需要用戶進一步明確
return f"請明確指出用戶指的是以下哪個組織：{candidates}"
# 添加一個過濾條件，篩選出提及特定組織的 articles
where_queries.Append(f"EXISTS {{(a)-[:MENTIONS]->(:Organization {{name: $organization}})}}")
# 將組織名稱作為參數傳入
params["organization"] = candidates[0]
如果系統識別出用戶感興趣的特定組織，我們會使用 get_candidates 函數將該組織的名稱映射到數據庫中的候選項。如果找到多個匹配項，我們會要求用戶進一步明確。如果沒有找到多個匹配項，我們會添加一個過濾條件，篩選出提及特定組織的 articles。為了安全起見，我們使用參數化查詢而不是直接拼接查詢字符串。
- 隨后，我們處理用戶可能基于提及的組織的國家進行預過濾的情況。
if country:
# 由于國家名稱標準化，不需要額外的映射
where_queries.append(f"EXISTS {{(a)-[:MENTIONS]->(:Organization)-[:IN_CITY]->()-[:IN_COUNTRY]->(:Country {{name: $country}})}}")
params["country"] = country
由于國家名稱通常是標準化的，我們不需要將國家名稱映射到數據庫中的值，因為大型語言模型（LLM）已經熟悉大多數國家的名稱。
- 隨后，我們處理情感元數據的過濾。
if sentiment:
if sentiment == "positive":
where_queries.append("a.sentiment > $sentiment")
params["sentiment"] = 0.5
else:
where_queries.append("a.sentiment < $sentiment")
params["sentiment"] = -0.5
我們要求 LLM 僅接受正面或負面兩種情感輸入值，并將這些值映射到適當的過濾器上。
- 對于 topic 參數，我們采取了略有不同的處理方式，因為它不用于預過濾，而是用于向量相似性搜索。
if topic: # 執行向量比較
vector_snippet = (
"WITH c, a, vector.similarity.cosine(c.embedding,$embedding) AS score "
"ORDER BY score DESC LIMIT toInteger($k)"
)
params["embedding"] = embeddings.embed_query(topic)
else: # 只返回最新的數據
vector_snippet = "WITH c, a ORDER BY a.date DESC LIMIT toInteger($k)"
如果系統識別出用戶對新聞中的特定主題感興趣，我們使用主題輸入的文本嵌入來找到最相關的文檔。如果沒有識別出特定主題，我們簡單地返回最新的幾篇文章，并避免向量相似性搜索。
- 最后，我們將 Cypher 語句組合起來，并用它來從數據庫中檢索信息。
return_snippet = "RETURN '#title ' + a.title + 'n#date ' + toString(a.date) + 'n#text ' + c.text AS output"
complete_query = (
base_query + " AND ".join(where_queries) + vector_snippet + return_snippet
)
# 從數據庫檢索信息
data = graph.query(complete_query, params)
print(f"Cypher: {complete_query}n")
# 在打印前安全地移除嵌入
params.pop('embedding', None)
print(f"參數: {params}")
return "###文章: ".join([el["output"] for el in data])
我們通過組合所有查詢片段來構建最終的 complete_query。然后，我們使用動態生成的 Cypher 語句從數據庫檢索信息并返回結果。讓我們通過一個示例輸入來看看生成的 Cypher 語句。
get_organization_news(
organizatinotallow='neo4j',
sentiment='positive',
topic='遠程工作'
)
# Cypher: CYPHER runtime = parallel parallelRuntimeSupport=all
# MATCH (c:Chunk)<-[:HAS_CHUNK]-(a:Article) WHERE
# EXISTS {(a)-[:MENTIONS]->(:Organization {name: $organization})} AND
# a.sentiment > $sentiment
# WITH c, a, vector.similarity.cosine(c.embedding,$embedding) AS score
# ORDER BY score DESC LIMIT toInteger($k)
# RETURN '#title ' + a.title + 'n#date ' + toString(a.date) + 'n#text ' + c.text AS output
# 參數: {'k': 5, 'organization': 'Neo4j', 'sentiment': 0.5}
動態查詢生成按預期工作，能夠從數據庫中檢索到相關的信息。
構建新聞信息代理工具
接下來，我們將創建一個代理工具，用于處理新聞信息查詢。首先，我們需要為輸入參數編寫一些說明。
fewshot_examples = """{輸入：google員工的健康福利在新聞中有哪些？查詢：健康福利}
{輸入：關于Google的最新正面新聞是什么？查詢：無}
{輸入：有關VertexAI和Google的新聞有哪些？查詢：VertexAI}
{輸入：關于Google的新產品有哪些新聞？查詢：新產品}
"""
class NewsInput(BaseModel):
topic: Optional[str] = Field(
descriptinotallow="除了組織、國家和情感傾向之外，如果您對其他特定信息或話題感興趣，請告訴我們。以下是一些示例："
+ fewshot_examples
)
organization: Optional[str] = Field(
descriptinotallow="您希望了解信息的組織名稱"
)
country: Optional[str] = Field(
descriptinotallow="您感興趣的組織的所在國家。請使用正式的國家名稱，例如‘美利堅合眾國’或‘法國’。"
)
sentiment: Optional[str] = Field(
descriptinotallow="您想要查詢的文章情感傾向", enum=["正面", "負面"]
)
在定義預過濾參數時，我遇到了一些困難，特別是如何讓 topic 參數按預期工作。為了解決這個問題，我提供了一些示例，幫助語言模型更好地理解用戶的需求。同時，我們還向模型提供了關于國家名稱格式的指導，并對情感傾向選項進行了枚舉。
現在，我們可以定義一個自定義工具，為其指定一個名稱和一段包含使用說明的描述。
class NewsTool(BaseTool):
name = "新聞信息工具"
description = (
"當你需要在新聞中查找相關信息時，這個工具會非常有用。"
)
args_schema:Type[BaseModel] = NewsInput
def _run(
self,
topic: Optional[str] = None,
organization: Optional[str] = None,
country: Optional[str] = None,
sentiment: Optional[str] = None,
run_manager: Optional[CallbackManagerForToolRun] = None,
) -> str:
""“使用這個工具來獲取新聞信息。”""
return get_organization_news(topic, organization, country, sentiment)
最后，我們需要定義一個代理執行器。這里，我使用了之前實現的 OpenAI 代理的 LCEL 實現。
llm = ChatOpenAI(temperature=0, model="GPT-4-turbo", streaming=True)
tools = [NewsTool()]
llm_with_tools = llm.bind(functinotallow=[format_tool_to_openai_function(t) for t in tools])
prompt = ChatPromptTemplate.from_messages(
[
(
"system",
“你是一個樂于助人的助手，可以找到關于電影的信息并進行推薦。如果工具需要進一步的問題，請確保向用戶詢問以獲得澄清。確保在后續問題中包含任何需要澄清的可用選項。只做用戶明確請求的事情。”
),
MessagesPlaceholder(variable_name="chat_history"),
("user", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
]
)
agent = (
{
"input": lambda x: x["input"],
"chat_history": lambda x: _format_chat_history(x["chat_history"])
if x.get("chat_history")
else [],
"agent_scratchpad": lambda x: format_to_openai_function_messages(
x["intermediate_steps"]
),
}
| prompt
| llm_with_tools
| OpenAIFunctionsAgentOutputParser()
)
agent_executor = AgentExecutor(agent=agent, tools=tools)
這個代理工具可以用于檢索新聞信息。我們還添加了 聊天記錄 消息占位符，這樣代理就可以進行對話，并允許提出后續問題和回復。
實施測試
讓我們嘗試幾個查詢，看看生成的 Cypher 語句和參數是什么樣的。
agent_executor.invoke(
{"輸入": "關于 neo4j 的一些正面新聞是什么？"}
)
# Cypher: CYPHER runtime = parallel parallelRuntimeSupport=all
# MATCH (c:Chunk)<-[:HAS_CHUNK]-(a:Article) WHERE
# EXISTS {(a)-[:MENTIONS]->(:Organization {name: $organization})} AND
# a.sentiment > $sentiment WITH c, a
# ORDER BY a.date DESC LIMIT toInteger($k)
# RETURN '#標題 ' + a.title + '日期 ' + toString(a.date) + '文本 ' + c.text AS output
# 參數: {'k': 5, 'organization': 'Neo4j', 'sentiment': 0.5}
生成的 Cypher 語句是有效的。由于沒有指定具體的主題，它返回了提到 Neo4j 的最后五篇正面文章的文本塊。讓我們嘗試一個更復雜的例子：
agent_executor.invoke(
{"輸入": "關于法國公司的員工幸福感，有哪些最新的負面新聞？"}
)
# Cypher: CYPHER runtime = parallel parallelRuntimeSupport=all
# MATCH (c:Chunk)<-[:HAS_CHUNK]-(a:Article) WHERE
# EXISTS {(a)-[:MENTIONS]->(:Organization)-[:IN_CITY]->()-[:IN_COUNTRY]->(:Country {name: $country})} AND
# a.sentiment < $sentiment
# WITH c, a, vector.similarity.cosine(c.embedding,$embedding) AS score
# ORDER BY score DESC LIMIT toInteger($k)
# RETURN '#標題 ' + a.title + '日期 ' + toString(a.date) + '文本 ' + c.text AS output
# 參數: {'k': 5, 'country': 'France', 'sentiment': -0.5, 'topic': '員工幸福感'}
語言模型代理正確地生成了預過濾參數，并且還識別出了一個特定的“員工幸福感”主題。這個主題被用作向量相似性搜索的輸入，使我們能夠進一步優化檢索過程。
總結
在這篇博客文章中，我們實現了基于圖的元數據過濾器的示例，以提高向量搜索的準確性。數據集擁有廣泛且相互關聯的選項，這允許進行更精細的預過濾查詢。結合圖數據表示和語言模型的函數調用功能，可以動態生成 Cypher 語句，從而為結構化過濾器提供了幾乎無限的可能性。
此外，你的代理可以擁有檢索非結構化文本的工具，如本文所示，以及能夠檢索結構化信息的其他工具，這使得知識圖譜成為許多 RAG應用的理想解決方案。
