作者：來(lái)自 Elastic?Han Xiang Choong

討論并實(shí)現(xiàn) Elastic RAG 的代理流程，其中 LLM 選擇調(diào)用 Elastic KB。

更多閱讀：Elasticsearch：基于 Langchain 的 Elasticsearch Agent 對(duì)文檔的搜索。

簡(jiǎn)介

代理是將 LLM 應(yīng)用于實(shí)際用例的合乎邏輯的下一步。本文旨在介紹代理在 RAG 工作流中的概念和用法?？偠灾?#xff0c;代理代表了一個(gè)極其令人興奮的領(lǐng)域，具有許多雄心勃勃的應(yīng)用程序和用例的可能性。

我希望在未來(lái)的文章中涵蓋更多這些想法?，F(xiàn)在，讓我們看看如何使用 Elasticsearch 作為我們的知識(shí)庫(kù)，使用 LangChain 作為我們的代理框架來(lái)實(shí)現(xiàn) Agentic RAG。

背景

LLMs 的使用始于簡(jiǎn)單地提示 LLM?執(zhí)行諸如回答問(wèn)題和簡(jiǎn)單計(jì)算之類的任務(wù)。

但是，現(xiàn)有模型知識(shí)的不足意味著 LLMs 無(wú)法應(yīng)用于需要專業(yè)技能的領(lǐng)域，例如企業(yè)客戶服務(wù)和商業(yè)智能。

很快，提示過(guò)渡到檢索增強(qiáng)生成 (RAG)，這是 Elasticsearch 的天然優(yōu)勢(shì)。RAG 是一種有效而簡(jiǎn)單的方法，可以在查詢時(shí)快速向 LLM 提供上下文和事實(shí)信息。另一種方法是漫長(zhǎng)而昂貴的再培訓(xùn)過(guò)程，而成功率遠(yuǎn)不能得到保證。

RAG 的主要操作優(yōu)勢(shì)是允許近乎實(shí)時(shí)地向 LLM 應(yīng)用程序提供更新的信息。

實(shí)施涉及采購(gòu)向量數(shù)據(jù)庫(kù)（例如 Elasticsearch）、部署嵌入模型（例如 ELSER）以及調(diào)用 search?API 來(lái)檢索相關(guān)文檔。

檢索到文檔后，可以將其插入 LLM 的提示中，并根據(jù)內(nèi)容生成答案。這提供了背景和事實(shí)，而 LLM 本身可能缺乏這兩者。

但是，標(biāo)準(zhǔn) RAG 部署模型有一個(gè)缺點(diǎn) - 它很死板。LLM 無(wú)法選擇從哪個(gè)知識(shí)庫(kù)中提取信息。它也無(wú)法選擇使用其他工具，例如 Google 或 Bing 等網(wǎng)絡(luò)搜索引擎 API。它無(wú)法查看當(dāng)前天氣，無(wú)法使用計(jì)算器，也無(wú)法考慮除給定知識(shí)庫(kù)之外的任何工具的輸出。

Agentic 模型的不同之處在于選擇。

術(shù)語(yǔ)說(shuō)明

工具使用（Tool usage），即 Langchain 上下文中使用的術(shù)語(yǔ)，也稱為函數(shù)調(diào)用。無(wú)論出于何種目的，這兩個(gè)術(shù)語(yǔ)都是可以互換的 - 兩者都指 LLM 被賦予一組函數(shù)或工具，它可以用來(lái)補(bǔ)充其能力或影響世界。請(qǐng)耐心等待，因?yàn)槲以诒疚牡钠溆嗖糠侄际褂昧?“工具使用 - Tool Usage”。

選擇

賦予 LLM 決策能力，并為其提供一套工具。根據(jù)對(duì)話的狀態(tài)和歷史，LLM 將選擇是否使用每個(gè)工具，并將工具的輸出納入其響應(yīng)中。

這些工具可能是知識(shí)庫(kù)、計(jì)算器、網(wǎng)絡(luò)搜索引擎和爬蟲 - 種類繁多，沒有限制或結(jié)束。LLM 能夠執(zhí)行復(fù)雜的操作和任務(wù)，而不僅僅是生成文本。

讓我們實(shí)現(xiàn)一個(gè)代理的簡(jiǎn)單示例。Elastic 的核心優(yōu)勢(shì)在于我們的知識(shí)庫(kù)。因此，此示例將重點(diǎn)介紹如何使用相對(duì)較大且復(fù)雜的知識(shí)庫(kù)，方法是制作比簡(jiǎn)單的向量搜索更復(fù)雜的查詢。

設(shè)置

首先，在項(xiàng)目目錄中定義一個(gè) .env 文件，并填寫這些字段。我使用帶有 GPT-4o 的 Azure OpenAI 部署來(lái)學(xué)習(xí)我的 LLM，并使用 Elastic Cloud 部署來(lái)學(xué)習(xí)我的知識(shí)庫(kù)。我的 Python 版本是 Python 3.12.4，我使用 Macbook 進(jìn)行操作。

ELASTIC_ENDPOINT="YOUR ELASTIC ENDPOINT"
ELASTIC_API_KEY="YOUR ELASTIC API KEY"OPENAI_API_TYPE="azure"
AZURE_OPENAI_ENDPOINT="YOUR AZURE ENDPOINT"
AZURE_OPENAI_API_VERSION="2024-06-01"
AZURE_OPENAI_API_KEY="YOUR AZURE API KEY"
AZURE_OPENAI_GPT4O_MODEL_NAME="gpt-4o"
AZURE_OPENAI_GPT4O_DEPLOYMENT_NAME="YOUR AZURE OPENAI GPT-4o DEPLOYMENT NAME"

你可能必須在終端中安裝以下依賴項(xiàng)。

pip install langchain elasticsearch 

在你的項(xiàng)目目錄中創(chuàng)建一個(gè)名為 chat.py 的 python 文件，并粘貼此代碼片段以初始化你的 LLM 和與 Elastic Cloud 的連接：

import os
from dotenv import load_dotenv
load_dotenv()from langchain.chat_models import AzureChatOpenAI
from langchain.agents import initialize_agent, AgentType, Tool
from langchain.tools import StructuredTool  # Import StructuredTool
from langchain.memory import ConversationBufferMemory
from typing import Optional
from pydantic import BaseModel, Field# LLM setup
llm = AzureChatOpenAI(openai_api_version=os.getenv("AZURE_OPENAI_API_VERSION"),azure_deployment=os.getenv("AZURE_OPENAI_GPT4O_DEPLOYMENT_NAME"),temperature=0.5,max_tokens=4096
)from elasticsearch import Elasticsearch
# Elasticsearch Setup
try:# Elasticsearch setupes_endpoint = os.environ.get("ELASTIC_ENDPOINT")es_client = Elasticsearch(es_endpoint,api_key=os.environ.get("ELASTIC_API_KEY"))
except Exception as e:es_client = None

Hello World！我們的第一個(gè)工具

初始化并定義我們的 LLM 和 Elastic 客戶端后，讓我們來(lái)做一個(gè) Elastic 版的 Hello World。我們將定義一個(gè)函數(shù)來(lái)檢查與 Elastic Cloud 的連接狀態(tài)，并定義一個(gè)簡(jiǎn)單的代理對(duì)話鏈來(lái)調(diào)用它。

將以下函數(shù)定義為 langchain Tool。名稱和描述是提示（prompt）工程的關(guān)鍵部分。LLM 依靠它們來(lái)確定是否在對(duì)話期間使用該工具。

# Define a function to check ES status
def es_ping(*args, **kwargs):if es_client is None:return "ES client is not initialized."else:try:if es_client.ping():return "ES ping returning True, ES is connected."else:return "ES is not connected."except Exception as e:return f"Error pinging ES: {e}"es_status_tool = Tool(name="ES Status",func=es_ping,description="Checks if Elasticsearch is connected.",
)tools = [es_status_tool]

現(xiàn)在，讓我們初始化一個(gè)對(duì)話記憶組件來(lái)跟蹤對(duì)話以及我們的 agent 本身。

# Initialize memory to keep track of the conversation
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)# Initialize agent
agent_chain = initialize_agent(tools,llm,agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION,memory=memory,verbose=True,
)

最后，讓我們用這個(gè)代碼片段運(yùn)行對(duì)話循環(huán)：

# Interactive conversation with the agent
def main():print("Welcome to the chat agent. Type 'exit' to quit.")while True:user_input = input("You: ")if user_input.lower() in ['exit', 'quit']:print("Goodbye!")breakresponse = agent_chain.run(input=user_input)print("Assistant:", response)if __name__ == "__main__":main()

在終端中，運(yùn)行 python chat.py 來(lái)初始化對(duì)話。

python chat.py 

以下是我的操作：

You: Hello
Assistant: Hello! How can I assist you today?
You: Is Elastic search connected?> Entering new AgentExecutor chain...
Thought: Do I need to use a tool? Yes
Action: ES Status
Action Input: Observation: ES ping returning True, ES is connected.
Thought:Do I need to use a tool? No
AI: Yes, Elasticsearch is connected. How can I assist you further?

當(dāng)我詢問(wèn) Elasticsearch 是否已連接時(shí)，LLM 使用 ES Status 工具，ping 了我的 Elastic Cloud 部署，返回 True，然后確認(rèn) Elastic Cloud 確實(shí)已連接。

恭喜！這是一個(gè)成功的 Hello World :)

請(qǐng)注意，觀察結(jié)果是 es_ping 函數(shù)的輸出。此觀察結(jié)果的格式和內(nèi)容是我們快速工程的關(guān)鍵部分，因?yàn)檫@是 LLM 用來(lái)決定下一步的內(nèi)容。

讓我們看看如何針對(duì) RAG 修改此工具。

Agentic RAG

我最近在我的 Elastic Cloud 部署中使用 POLITICS 數(shù)據(jù)集構(gòu)建了一個(gè)大型而復(fù)雜的知識(shí)庫(kù)（knowledge base）。該數(shù)據(jù)集包含從美國(guó)新聞來(lái)源抓取的大約 246 萬(wàn)篇政治文章。我將其導(dǎo)入 Elastic Cloud 并將其嵌入 elser_v2 推理端點(diǎn)，遵循上一篇博客中定義的流程。

要部署 elser_v2 推理端點(diǎn)，請(qǐng)確保啟用了 ML 節(jié)點(diǎn)自動(dòng)擴(kuò)展，然后在 Elastic Cloud 控制臺(tái)中運(yùn)行以下命令。

PUT _inference/sparse_embedding/elser_v2
{"service": "elser","service_settings": {"num_allocations": 4,"num_threads": 8}
}

現(xiàn)在，讓我們定義一個(gè)新工具，對(duì)我們的政治知識(shí)庫(kù)索引進(jìn)行簡(jiǎn)單的語(yǔ)義搜索。我將其稱為 bignews_embedded。此函數(shù)接受搜索查詢，將其添加到標(biāo)準(zhǔn)語(yǔ)義搜索查詢模板，然后使用 Elasticsearch 運(yùn)行查詢。一旦獲得搜索結(jié)果，它就會(huì)將文章內(nèi)容連接成一個(gè)文本塊，并將其作為 LLM 觀察（observation）返回。

我們將搜索結(jié)果的數(shù)量限制為 3。這種 Agentic RAG 風(fēng)格的一個(gè)優(yōu)點(diǎn)是我們可以通過(guò)多個(gè)對(duì)話步驟來(lái)制定答案。換句話說(shuō)，可以使用引導(dǎo)性問(wèn)題來(lái)設(shè)置階段和上下文來(lái)回答更復(fù)雜的查詢。問(wèn)答變成了基于事實(shí)的對(duì)話，而不是一次性的答案生成。

日期

為了突出使用代理的重要優(yōu)勢(shì)，RAG 搜索函數(shù)除了查詢之外還包含 dates 參數(shù)。在搜索新聞文章時(shí)，我們可能希望將搜索結(jié)果限制在特定的時(shí)間范圍內(nèi)，例如“In 2020”或 “Between 2008 and 2012”。通過(guò)添加日期以及解析器，我們?cè)试S LLM 指定搜索的日期范圍。

簡(jiǎn)而言之，如果我指定 “California wildfires in 2020” 之類的內(nèi)容，我不希望看到 2017 年或任何其他年份的新聞。

此 rag_search 函數(shù)是一個(gè)日期解析器（從輸入中提取日期并將其添加到查詢中）和一個(gè) Elastic semantic_search 查詢。

# Define the RAG search function
def rag_search(query: str, dates: str):if es_client is None:return "ES client is not initialized."else:try:# Build the Elasticsearch querymust_clauses = []# If dates are provided, parse and include in queryif dates:# Dates must be in format 'YYYY-MM-DD' or 'YYYY-MM-DD to YYYY-MM-DD'date_parts = dates.strip().split(' to ')if len(date_parts) == 1:# Single datestart_date = date_parts[0]end_date = date_parts[0]elif len(date_parts) == 2:start_date = date_parts[0]end_date = date_parts[1]else:return "Invalid date format. Please use YYYY-MM-DD or YYYY-MM-DD to YYYY-MM-DD."date_range = {"range": {"date": {"gte": start_date,"lte": end_date}}}must_clauses.append(date_range)# Add the main query clausemain_query = {"nested": {"path": "text.inference.chunks","query": {"sparse_vector": {"inference_id": "elser_v2","field": "text.inference.chunks.embeddings","query": query}},"inner_hits": {"size": 2,"name": "bignews_embedded.text","_source": False}}}must_clauses.append(main_query)es_query = {"_source": ["text.text", "title", "date"],"query": {"bool": {"must": must_clauses}},"size": 3}response = es_client.search(index="bignews_embedded", body=es_query)hits = response["hits"]["hits"]if not hits:return "No articles found for your query."result_docs = []for hit in hits:source = hit["_source"]title = source.get("title", "No Title")text_content = source.get("text", {}).get("text", "")date = source.get("date", "No Date")doc = f"Title: {title}\nDate: {date}\n{text_content}\n"result_docs.append(doc)return "\n".join(result_docs)except Exception as e:return f"Error during RAG search: {e}"

運(yùn)行完整的搜索查詢后，結(jié)果將連接成一個(gè)文本塊并作為 “observation” 返回，供 LLM 使用。

為了考慮多個(gè)可能的參數(shù)，請(qǐng)使用 pydantic 的 BaseModel 定義有效的輸入格式：

class RagSearchInput(BaseModel):query: str = Field(..., description="The search query for the knowledge base.")dates: str = Field(...,description="Date or date range for filtering results. Specify in format YYYY-MM-DD or YYYY-MM-DD to YYYY-MM-DD.")

我們還需要利用 StructuredTool 定義一個(gè)多輸入函數(shù)，使用上面定義的輸入格式：

# Define the RAG search tool using StructuredTool
rag_search_tool = StructuredTool(name="RAG_Search",func=rag_search,description=("Use this tool to search for information about American politics from the knowledge base. ""**Input must include a search query and a date or date range.** ""Dates must be specified in this format YYYY-MM-DD or YYYY-MM-DD to YYYY-MM-DD."),args_schema=RagSearchInput
)

描述是工具定義的關(guān)鍵要素，也是你快速工程的一部分。它應(yīng)該全面而詳細(xì)，并為 LLM 提供足夠的背景信息，以便它知道何時(shí)使用該工具以及出于什么目的。

描述還應(yīng)包括 LLM 必須提供的輸入類型，以便正確使用該工具。指定格式和期望在這里會(huì)產(chǎn)生巨大影響。

不具信息性的描述可能會(huì)嚴(yán)重影響 LLM 使用該工具的能力！

請(qǐng)記住將新工具添加到代理要使用的工具列表中：

tools = [es_status_tool, rag_search_tool]

我們還需要使用系統(tǒng)提示進(jìn)一步修改代理，以提供對(duì)代理行為的額外控制。系統(tǒng)提示對(duì)于確保不會(huì)發(fā)生與格式錯(cuò)誤的輸出和函數(shù)輸入相關(guān)的錯(cuò)誤至關(guān)重要。我們需要明確說(shuō)明每個(gè)函數(shù)的期望以及模型應(yīng)輸出的內(nèi)容，因?yàn)槿绻?langchain 看到格式不正確的 LLM 響應(yīng)，它將拋出錯(cuò)誤。

我們還需要設(shè)置 agent=AgentType.OPENAI_FUNCTIONS 以使用 OpenAI 的函數(shù)調(diào)用功能。這允許 LLM 根據(jù)我們指定的結(jié)構(gòu)化模板與函數(shù)交互。

請(qǐng)注意，系統(tǒng)提示包括 LLM 應(yīng)生成的輸入的確切格式的規(guī)定，以及具體示例。

LLM 不僅應(yīng)該檢測(cè)應(yīng)該使用哪種工具，還應(yīng)該檢測(cè)工具期望的輸入！ Langchain 只負(fù)責(zé)函數(shù)調(diào)用/工具使用，但是否正確使用取決于 LLM。

agent_chain = initialize_agent(tools,llm,agent=AgentType.OPENAI_FUNCTIONS,memory=memory,verbose=True,handle_parsing_errors=True,system_message="""You are an AI assistant that helps with questions about American politics using a knowledge base. Be concise, sharp, to the point, and respond in one paragraph.You have access to the following tools:- **ES_Status**: Checks if Elasticsearch is connected.- **RAG_Search**: Use this to search for information in the knowledge base. **Input must include a search query and a date or date range.** Dates must be specified in this format YYYY-MM-DD or YYYY-MM-DD to YYYY-MM-DD.**Important Instructions:**- **Extract dates or date ranges from the user's question.**- **If the user does not provide a date or date range, politely ask them to provide one before proceeding.**When you decide to use a tool, use the following format *exactly*:Thought: [Your thought process about what you need to do next]Action: [The action to take, should be one of [ES_Status, RAG_Search]]Action Input: {"query": "the search query", "dates": "the date or date range"}If you receive an observation after an action, you should consider it and then decide your next step. If you have enough information to answer the user's question, respond with:Thought: [Your thought process]Assistant: [Your final answer to the user]**Examples:**- **User's Question:** "Tell me about the 2020 California wildfires."Thought: I need to search for information about the 2020 California wildfires.Action: RAG_SearchAction Input: {"query" : "California wildfires", "dates" : "2020-01-01 to 2020-12-31"}- **User's Question:** "What happened during the presidential election?"Thought: The user didn't specify a date. I should ask for a date range.Assistant: Could you please specify the date or date range for the presidential election you're interested in?Always ensure that your output strictly follows one of the above formats, and do not include any additional text or formatting.Remember:- **Do not** include any text before or after the specified format.- **Do not** add extra explanations.- **Do not** include markdown, bullet points, or numbered lists unless it is part of the Assistant's final answer.Your goal is to assist the user by effectively using the tools when necessary and providing clear and concise answers."""
)

現(xiàn)在，在你的終端中運(yùn)行 python chat.py 并讓我們測(cè)試一下！

測(cè)試 Agentic RAG

讓我們通過(guò)詢問(wèn)以下查詢來(lái)測(cè)試它：

Summarize the federal response to california wildfires in 2020

有幫助的是，langchain 將輸出中間步驟，包括 RAG_Search 函數(shù)的輸入、搜索結(jié)果和最終輸出。

> Entering new AgentExecutor chain...Invoking: `RAG_Search` with `{'query': 'federal response to California wildfires', 'dates': '2020-01-01 to 2020-12-31'}`Title: California federal wildfire aid: Newsom says Trump changed mind
Date: 2020-10-16T00:00:00.000Z
President Donald Trump abruptly changed course...Title: California governor declares state of emergency in multiple counties as wildfires rage
Date: 2020-09-07T00:00:00.000Z
In a statement, Newsom noted that tens...Title: Trump approves disaster declaration for California, reversing course
Date: 2020-10-16T00:00:00.000Z
California secured a Major Disaster Declaration ...> Finished chain.Assistant: The federal response to the California wildfires in 2020 included several key actions and moments:1. **Initial Denial and Reversal of Federal Aid**: President Trump initially denied California's request for federal aid for the wildfires. However, after Governor Gavin Newsom vowed to appeal the decision, Trump reversed course and approved the Major Disaster Declaration. This declaration allowed California to receive hundreds of millions of dollars in federal aid.2. **Statewide Emergency Declarations**: Governor Newsom declared a statewide emergency due to the widespread fires and extreme weather conditions. The state worked closely with federal authorities to secure assistance, including Fire Management Assistance Grants from FEMA.3. **Presidential Major Disaster Declarations**: Newsom secured a Presidential Major Disaster Declaration to bolster the state's emergency response to the Northern California wildfires. This declaration made people in impacted counties eligible for various forms of federal support, including crisis counseling, housing, unemployment assistance, and legal services.4. **Federal and State Collaboration**: Despite ongoing tensions and disagreements between the state and the Trump administration, particularly over forest management and climate change, federal agencies such as the National Park Service, U.S. Forest Service, and Bureau of Land Management were involved in managing and supporting firefighting efforts in California.5. **Impact and Scale of Wildfires**: The 2020 wildfire season in California was historically devastating, with more than 8,500 blazes scorching 6,400 square miles, destroying thousands of structures, and claiming lives. The federal aid and disaster declarations were crucial in supporting the state's response and recovery efforts.Overall, the federal response involved a combination of initial resistance followed by critical support and collaboration to address the unprecedented wildfire crisis in California.

最值得注意的是，LLM 創(chuàng)建了一個(gè)搜索查詢，然后添加了從 2020 年初到年末的日期范圍。通過(guò)將搜索結(jié)果限制在指定年份，我們確保只有相關(guān)文檔才會(huì)傳遞給 LLM。

我們可以用它做更多的事情，例如根據(jù)類別、某些實(shí)體的外觀或與其他事件的關(guān)系進(jìn)行約束。

可能性無(wú)窮無(wú)盡，我認(rèn)為這很酷！

關(guān)于錯(cuò)誤處理的注意事項(xiàng)

在某些情況下，LLM 可能無(wú)法在需要時(shí)使用正確的工具/功能。例如，它可能選擇使用自己的知識(shí)而不是使用可用的知識(shí)庫(kù)來(lái)回答有關(guān)當(dāng)前事件的問(wèn)題。

必須仔細(xì)測(cè)試和調(diào)整系統(tǒng)提示和工具/功能描述。

另一種選擇可能是增加可用工具的種類，以增加基于知識(shí)庫(kù)內(nèi)容而不是 LLM 的固有知識(shí)生成答案的可能性。

請(qǐng)注意，LLMs 偶爾會(huì)失敗，這是其概率性質(zhì)的自然結(jié)果。有用的錯(cuò)誤消息或免責(zé)聲明也可能是用戶體驗(yàn)的重要組成部分。

結(jié)論和未來(lái)前景

對(duì)我來(lái)說(shuō)，主要的收獲是創(chuàng)建更高級(jí)的搜索應(yīng)用程序的可能性。LLM 可能能夠在自然語(yǔ)言對(duì)話的背景下即時(shí)制作非常復(fù)雜的搜索查詢。這為大幅提高搜索應(yīng)用程序的準(zhǔn)確性和相關(guān)性開辟了道路，也是我興奮地探索的領(lǐng)域。

通過(guò) LLM 媒介，知識(shí)庫(kù)與其他工具（例如 Web 搜索引擎和監(jiān)控工具 API）的交互也可以實(shí)現(xiàn)一些令人興奮且復(fù)雜的用例。來(lái)自 KB 的搜索結(jié)果可能會(huì)補(bǔ)充實(shí)時(shí)信息，從而使 LLM 能夠執(zhí)行有效且時(shí)間敏感的即時(shí)推理。

還有多代理工作流的可能性。在 Elastic 上下文中，這可能是多個(gè)代理探索不同的知識(shí)庫(kù)集，以協(xié)作構(gòu)建復(fù)雜問(wèn)題的解決方案。也許是一個(gè)聯(lián)合搜索模型，其中多個(gè)組織構(gòu)建協(xié)作、共享的應(yīng)用程序，類似于聯(lián)合學(xué)習(xí)（federated learning）的想法？

我想探索 Elasticsearch 的一些用例，希望你也能這樣做。

下次見！

附錄：chat.py 的完整代碼

import os
from dotenv import load_dotenv
load_dotenv()from langchain.chat_models import AzureChatOpenAI
from langchain.agents import initialize_agent, AgentType, Tool
from langchain.tools import StructuredTool  # Import StructuredTool
from langchain.memory import ConversationBufferMemory
from typing import Optional
from pydantic import BaseModel, Fieldllm = AzureChatOpenAI(openai_api_version=os.getenv("AZURE_OPENAI_API_VERSION"),azure_deployment=os.getenv("AZURE_OPENAI_GPT4O_DEPLOYMENT_NAME"),temperature=0.5,max_tokens=4096
)from elasticsearch import Elasticsearchtry:# Elasticsearch setupes_endpoint = os.environ.get("ELASTIC_ENDPOINT")es_client = Elasticsearch(es_endpoint,api_key=os.environ.get("ELASTIC_API_KEY"))
except Exception as e:es_client = None# Define a function to check ES status
def es_ping(_input):if es_client is None:return "ES client is not initialized."else:try:if es_client.ping():return "ES is connected."else:return "ES is not connected."except Exception as e:return f"Error pinging ES: {e}"# Define the ES status tool
es_status_tool = Tool(name="ES_Status",func=es_ping,description="Checks if Elasticsearch is connected.",
)# Define the RAG search function
def rag_search(query: str, dates: str):if es_client is None:return "ES client is not initialized."else:try:# Build the Elasticsearch querymust_clauses = []# If dates are provided, parse and include in queryif dates:# Dates must be in format 'YYYY-MM-DD' or 'YYYY-MM-DD to YYYY-MM-DD'date_parts = dates.strip().split(' to ')if len(date_parts) == 1:# Single datestart_date = date_parts[0]end_date = date_parts[0]elif len(date_parts) == 2:start_date = date_parts[0]end_date = date_parts[1]else:return "Invalid date format. Please use YYYY-MM-DD or YYYY-MM-DD to YYYY-MM-DD."date_range = {"range": {"date": {"gte": start_date,"lte": end_date}}}must_clauses.append(date_range)# Add the main query clausemain_query = {"nested": {"path": "text.inference.chunks","query": {"sparse_vector": {"inference_id": "elser_v2","field": "text.inference.chunks.embeddings","query": query}},"inner_hits": {"size": 2,"name": "bignews_embedded.text","_source": False}}}must_clauses.append(main_query)es_query = {"_source": ["text.text", "title", "date"],"query": {"bool": {"must": must_clauses}},"size": 3}response = es_client.search(index="bignews_embedded", body=es_query)hits = response["hits"]["hits"]if not hits:return "No articles found for your query."result_docs = []for hit in hits:source = hit["_source"]title = source.get("title", "No Title")text_content = source.get("text", {}).get("text", "")date = source.get("date", "No Date")doc = f"Title: {title}\nDate: {date}\n{text_content}\n"result_docs.append(doc)return "\n".join(result_docs)except Exception as e:return f"Error during RAG search: {e}"class RagSearchInput(BaseModel):query: str = Field(..., description="The search query for the knowledge base.")dates: str = Field(...,description="Date or date range for filtering results. Specify in format YYYY-MM-DD or YYYY-MM-DD to YYYY-MM-DD.")# Define the RAG search tool using StructuredTool
rag_search_tool = StructuredTool(name="RAG_Search",func=rag_search,description=("Use this tool to search for information about American politics from the knowledge base. ""**Input must include a search query and a date or date range.** ""Dates must be specified in this format YYYY-MM-DD or YYYY-MM-DD to YYYY-MM-DD."),args_schema=RagSearchInput
)# List of tools
tools = [es_status_tool, rag_search_tool]# Initialize memory to keep track of the conversation
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)agent_chain = initialize_agent(tools,llm,agent=AgentType.OPENAI_FUNCTIONS,memory=memory,verbose=True,handle_parsing_errors=True,system_message="""You are an AI assistant that helps with questions about American politics using a knowledge base. Be concise, sharp, to the point, and respond in one paragraph.You have access to the following tools:- **ES_Status**: Checks if Elasticsearch is connected.- **RAG_Search**: Use this to search for information in the knowledge base. **Input must include a search query and a date or date range.** Dates must be specified in this format YYYY-MM-DD or YYYY-MM-DD to YYYY-MM-DD.**Important Instructions:**- **Extract dates or date ranges from the user's question.**- **If the user does not provide a date or date range, politely ask them to provide one before proceeding.**When you decide to use a tool, use the following format *exactly*:Thought: [Your thought process about what you need to do next]Action: [The action to take, should be one of [ES_Status, RAG_Search]]Action Input: {"query": "the search query", "dates": "the date or date range"}If you receive an observation after an action, you should consider it and then decide your next step. If you have enough information to answer the user's question, respond with:Thought: [Your thought process]Assistant: [Your final answer to the user]**Examples:**- **User's Question:** "Tell me about the 2020 California wildfires."Thought: I need to search for information about the 2020 California wildfires.Action: RAG_SearchAction Input: {"query": "California wildfires", "dates": "2020-01-01 to 2020-12-31"}- **User's Question:** "What happened during the presidential election?"Thought: The user didn't specify a date. I should ask for a date range.Assistant: Could you please specify the date or date range for the presidential election you're interested in?Always ensure that your output strictly follows one of the above formats, and do not include any additional text or formatting.Remember:- **Do not** include any text before or after the specified format.- **Do not** add extra explanations.- **Do not** include markdown, bullet points, or numbered lists unless it is part of the Assistant's final answer.Your goal is to assist the user by effectively using the tools when necessary and providing clear and concise answers."""
)# Interactive conversation with the agent
def main():print("Welcome to the chat agent. Type 'exit' to quit.")while True:user_input = input("You: ")if user_input.lower() in ['exit', 'quit']:print("Goodbye!")break# Update method call to address deprecation warningresponse = agent_chain.invoke(input=user_input)print("Assistant:", response['output'])if __name__ == "__main__":main()

Elasticsearch 包含許多新功能，可幫助你針對(duì)自己的用例構(gòu)建最佳搜索解決方案。深入了解我們的示例筆記本以了解更多信息，開始免費(fèi)云試用，或立即在本地機(jī)器上試用 Elastic。

原文：https://www.elastic.co/search-labs/blog/rag-agent-tool-elasticsearch-langchain