基于Prefect、Weave和RAGAS的RAG應(yīng)用開發(fā)實(shí)戰(zhàn)

作者：朱先忠 2024-08-12 08:28:53

本文介紹了RAG應(yīng)用程序開發(fā)中的一些增強(qiáng)技術(shù)，例如加入線下/線上評估、借助Prefect自動提取數(shù)據(jù)等。

譯者 | 朱先忠

審校 | 重樓

幾個(gè)月前，我發(fā)布了一款電影搜索應(yīng)用程序“Film Search”，這是一個(gè)檢索增強(qiáng)生成（RAG）應(yīng)用程序，旨在根據(jù)用戶查詢實(shí)現(xiàn)電影推薦。例如，用戶可能會發(fā)出下面的英文提問：

“Find me drama movies in English that are less than 2 hours long and feature dogs.”

中文意思：“給我找一部長度不到2小時(shí)、以狗為主角的英文劇情片。”

之后，用戶會收到類似下面這樣的英文推薦：

Title of Film: Hachi: A Dog’s Tale

Runtime: 93 minutes
Release Year: 2009
Streaming: Not available for streaming
This film tells the poignant true story of Hachiko, an Akita dog known for his remarkable loyalty to his owner. The emotional depth and the themes of friendship and loyalty resonate strongly, making it a touching drama that showcases the profound bond between humans and dogs. It’s perfect for anyone looking for a heartfelt story that highlights the importance of companionship.

對應(yīng)的中文回答意思是：

電影名稱：忠犬八公的故事
片長：93分鐘
發(fā)行年份：2009
流媒體：不可用于流媒體
這部電影告訴了秋田犬Hachiko的辛酸真實(shí)故事，這種犬以其對主人的非凡忠誠而聞名。情感深度以及友誼和忠誠的主題引起了強(qiáng)烈的共鳴，使其成為一部感人的電影，展示了人與狗之間的深厚友誼。它非常適合想尋找一個(gè)發(fā)自內(nèi)心的故事來強(qiáng)調(diào)友誼重要性的人。

這個(gè)軟件不僅僅是一個(gè)簡單的RAG應(yīng)用程序。該程序中使用了所謂的自查詢檢索。這意味著，機(jī)器人接受用戶的查詢，并通過添加元數(shù)據(jù)過濾器對其進(jìn)行轉(zhuǎn)換。這樣就確保了拉入聊天模型上下文的任何文檔都遵守用戶查詢設(shè)置的約束。有關(guān)更多信息，我建議查看我文后提供的我早些時(shí)候發(fā)表的文章的鏈接。

但遺憾的是，該應(yīng)用程序尚存在如下一些問題：

除了通過“肉眼測試”外，并沒有進(jìn)行離線評估。這種測試是必要的，但還不夠。
沒有提供可觀察性支持。如果查詢進(jìn)展不順利，你必須手動調(diào)出項(xiàng)目并運(yùn)行一些特別的腳本，以便找出問題所在。
必須手動拉入Pinecone向量數(shù)據(jù)庫。這意味著，如果一部電影從流媒體服務(wù)中撤下，這些文件很快就會過時(shí)。

在本文中，我將簡要介紹對以前開發(fā)的那款電影搜索應(yīng)用程序“Film Search”所做的一些改進(jìn)，具體的改進(jìn)內(nèi)容將包括：

使用RAGAS和Weave進(jìn)行離線評估。
在線評估和可觀察性。
使用Prefect自動提取數(shù)據(jù)。

在我們正式開始之前，還有一個(gè)細(xì)節(jié)需要說明：我發(fā)現(xiàn)“Film Search”這個(gè)名字有點(diǎn)籠統(tǒng)，所以我把這個(gè)應(yīng)用程序重新命名為“Rosebud”，如下圖所示。不用多作解釋，我想任何一位真正的電影迷都會明白這個(gè)意思的（【譯者注】影片Citizen Kane（公民凱恩）的故事是由報(bào)業(yè)巨子凱恩臨死前說的一個(gè)字“玫瑰花蕾”（Rosebud）引出的）。

程序名字“Film Search”更改為“Rosebud”，此圖來自Unsplash網(wǎng)站

線下評估

能夠判斷對LLM應(yīng)用程序所做的更改是提高還是降低了程序性能，這一點(diǎn)是非常重要的。不幸的是，LLM應(yīng)用程序的評估是一個(gè)困難而新穎的領(lǐng)域。對于什么是好的評估，根本沒有達(dá)成太多的共識。

在新的程序Rosebud中，我決定解決所謂的“RAG Triad（RAG三元組）：https://www.trulens.org/trulens_eval/getting_started/core_concepts/rag_triad/”問題。這種方法由TruLens推出，TruLens是一個(gè)評估和跟蹤LLM應(yīng)用程序的平臺。

RAG Triad（RAG三元組）

概括來看，三元組涵蓋了RAG應(yīng)用程序的三個(gè)方面：

上下文相關(guān)性：當(dāng)用戶進(jìn)行查詢時(shí)，文檔會填充聊天模型的上下文。檢索到的上下文真的有用嗎？如果沒有用，你可能需要調(diào)整文檔嵌入、分塊或元數(shù)據(jù)過濾等操作。
可信度：模型的響應(yīng)是否真的基于檢索到的文檔而生成？你不希望模型編造事實(shí)；RAG的全部目的是通過使用檢索到的文檔來幫助減少幻覺。
答案相關(guān)性：模型的響應(yīng)是否真正回答了用戶的查詢？如果用戶詢問“20世紀(jì)90年代制作的喜劇電影有哪些？”，該模型的答案最好只包含20世紀(jì)90時(shí)代制作的喜劇影片。

目前，已經(jīng)存在幾種方法可以嘗試評估RAG應(yīng)用程序的這三個(gè)功能。一種方法是借助人類專家評估員。不幸的是，這種方法十分昂貴，而且無法擴(kuò)展。在新的程序Rosebud中，我決定使用大型數(shù)據(jù)模型進(jìn)行評估。這意味著，使用聊天模型來查看上述三個(gè)標(biāo)準(zhǔn)中的每一個(gè)，并為每個(gè)標(biāo)準(zhǔn)分配0到1的分?jǐn)?shù)值。這種方法具有成本低、可擴(kuò)展性好的優(yōu)點(diǎn)。為了實(shí)現(xiàn)這一點(diǎn)，我使用了RAGAS（https://github.com/explodinggradients/ragas），這是一個(gè)流行的框架，可以幫助你評估RAG應(yīng)用程序。RAGAS框架包括上述三個(gè)指標(biāo)，可以很容易地使用它們來評估你的應(yīng)用程序。下面是一個(gè)代碼片段，演示了我是如何使用開源的RAGAS框架進(jìn)行離線評估的：

from ragas import evaluate
from ragas.metrics import AnswerRelevancy, ContextRelevancy, Faithfulness
import weave

@weave.op()
def evaluate_with_ragas(query, model_output):
#將數(shù)據(jù)放入一個(gè)數(shù)據(jù)集對象中
data = {
"question": [query],
"contexts": [[model_output['context']]],
"answer": [model_output['answer']]
}
dataset = Dataset.from_dict(data)

# 定義要判斷的指標(biāo)
metrics = [
AnswerRelevancy(),
ContextRelevancy(),
Faithfulness(),
]

judge_model = ChatOpenAI(model=config['JUDGE_MODEL_NAME'])
embeddings_model = OpenAIEmbeddings(model=config['EMBEDDING_MODEL_NAME'])

evaluation = evaluate(dataset=dataset, metrics=metrics, llm=judge_model, embeddings=embeddings_model)

return {
"answer_relevancy": float(evaluation['answer_relevancy']),
"context_relevancy": float(evaluation['context_relevancy']),
"faithfulness": float(evaluation['faithfulness']),
}
def run_evaluation():
#初始化聊天模型
model = rosebud_chat_model()
# 定義評估問題
questions = [
{"query": "Suggest a good movie based on a book."},  # Adaptations
{"query": "Suggest a film for a cozy night in."},  # Mood-Based
{"query": "What are some must-watch horror movies?"},  # Genre-Specific
...
# 共20個(gè)問題
]
#創(chuàng)建Weave評估對象
evaluation = weave.Evaluation(dataset=questions, scorers=[evaluate_with_ragas])
#運(yùn)行評估
asyncio.run(evaluation.evaluate(model))
if __name__ == "__main__":
weave.init('film-search')
run_evaluation()

在上述代碼中，有幾點(diǎn)注意事項(xiàng)：

有20個(gè)問題和3個(gè)評判標(biāo)準(zhǔn)，你會看到60次LLM調(diào)用僅需要一次評估！然而，接下來的情況變得更糟了：通過調(diào)用函數(shù)rosebud_chat_model，每個(gè)查詢都需要兩次調(diào)用。其中，一個(gè)用于構(gòu)造元數(shù)據(jù)過濾器，另一個(gè)用于提供答案；所以，實(shí)際上這是對單個(gè)模型計(jì)算的120次調(diào)用！我評估的所有模型都是使用新的gpt-4o-mini，我也強(qiáng)烈推薦使用這種模型。根據(jù)我的經(jīng)驗(yàn)，每次評估的調(diào)用費(fèi)用為0.05美元。
請注意，我們使用了異步的asyncio.run運(yùn)行模型計(jì)算。這種情況下，使用異步調(diào)用是比較合適的，因?yàn)槟悴幌胍粋€(gè)接一個(gè)地以順序方式評估每個(gè)問題。相反，借助于asyncio框架，我們可以在等待之前的I/O操作完成時(shí)開始評估其他的問題。
一次評估共有20個(gè)問題。這些涵蓋了用戶可能會提問的各種典型的電影查詢。這些大多是我自己想出的，但在實(shí)踐中，最好使用生產(chǎn)中用戶實(shí)際提出的查詢。
請注意正在使用的weap.init和@weap.op裝飾器。它們是Weights & Biases(W&B) AI開發(fā)者平臺提供的新的Weave庫的一部分。其中，Weave庫是對傳統(tǒng)W&B庫的補(bǔ)充，專注于LLM應(yīng)用程序。它允許你通過使用簡單的@weap.op裝飾器來捕獲LLM的輸入和輸出。它還允許你使用weave.Evaluation(…)評估結(jié)果。通過集成RAGAS來執(zhí)行評估，并集成Weave框架來捕獲和記錄它們，我們便有了一個(gè)強(qiáng)大的組合，可以幫助GenAI開發(fā)人員以迭代方式不斷改進(jìn)他們的應(yīng)用程序。此外，你還可以記錄下模型延遲、所需成本等其他信息。

集成Weave+RAGAS的示例程序

從理論上講，現(xiàn)在我們可以調(diào)整一個(gè)超參數(shù)（如溫度），重新運(yùn)行評估，看看調(diào)整是否有積極或消極的影響了。但遺憾的是，在實(shí)踐中，我發(fā)現(xiàn)大型語言評判者的評判很挑剔，而且我也不是唯一一個(gè)發(fā)現(xiàn)這一點(diǎn)的人（https://x.com/aparnadhinak/status/1748368364395721128）。

大型語言模型評估似乎不太擅長使用浮點(diǎn)數(shù)來評估這些指標(biāo)。相反，它們似乎在分類方面做得更好些，例如回答“同意/不同意”這樣的問題。當(dāng)前，RAGAS尚不支持使用LLM評判者進(jìn)行分類。直接手寫有關(guān)代碼似乎也并不難，也許在未來的更新中，我可能會自己嘗試一下。

在線評估

離線評估有助于了解調(diào)整超參數(shù)如何影響性能，在我看來，在線評估要有用得多。在新的程序Rosebud中，我現(xiàn)在已經(jīng)使用“同意/不同意”的方案——使用每個(gè)響應(yīng)底部的兩個(gè)相應(yīng)按鈕來提供反饋。

在線反饋示例

當(dāng)用戶點(diǎn)擊上圖中底部任一按鈕時(shí)，就會被告知他們的反饋已被記錄。以下給出在Streamlit應(yīng)用程序界面中如何實(shí)現(xiàn)這一點(diǎn)的代碼片段：

def start_log_feedback(feedback):
print("Logging feedback.")
st.session_state.feedback_given = True
st.session_state.sentiment = feedback
thread = threading.Thread(target=log_feedback, args=(st.session_state.sentiment,
st.session_state.query,
st.session_state.query_constructor,
st.session_state.context,
st.session_state.response))
thread.start()
def log_feedback(sentiment, query, query_constructor, context, response):
ct = datetime.datetime.now()
wandb.init(project="film-search",
name=f"query: {ct}")
table = wandb.Table(columns=["sentiment", "query", "query_constructor", "context", "response"])
table.add_data(sentiment,
query,
query_constructor,
context,
response
)
wandb.log({"Query Log": table})
wandb.finish()

請注意，向W&B發(fā)送反饋的過程是在單獨(dú)的線程上運(yùn)行的，而不是在主線程上運(yùn)行。這是為了防止用戶在等待日志記錄完成時(shí)被卡住幾秒鐘。

我們使用了一個(gè)W&B表格用于存儲反饋。表中記錄了五個(gè)數(shù)值：

情緒（Sentiment）：用戶是否點(diǎn)擊了拇指圖標(biāo)（同意/不同意）。
查詢（Query）：用戶的查詢，例如，查找長度不到2小時(shí)的英文戲劇電影和故事狗。
Query_Constructor：查詢構(gòu)造函數(shù)的結(jié)果，它重寫用戶的查詢，并在必要時(shí)包含元數(shù)據(jù)過濾，例如：

{
"query": "drama English dogs", 
"filter": {
"operator": "and", 
"arguments": [
{
"comparator": "eq", "attribute": "Genre", "value": "Drama"
}, 
{
"comparator": "eq", "attribute": "Language", "value": "English"
}, 

{
"comparator": "lt", "attribute": "Runtime (minutes)", "value": 120
}
]
},
}

上下文（Context）：基于重建的查詢檢索到的上下文，例如標(biāo)題“Title: Hachi: A Dog’s Tale. Overview: A drama based on the true story of a college professor’s…”。
回應(yīng)（Response）：模型的回應(yīng)。

所有這些都可以方便地記錄在與前面顯示的Weave評估相同的項(xiàng)目中。現(xiàn)在，當(dāng)查詢“不同意”情況時(shí)，只需按下拇指向下的圖標(biāo)按鈕即可查看到底發(fā)生了什么。這將有助于使推薦應(yīng)用程序Rosebud的迭代和改進(jìn)加快速度。

模型響應(yīng)可觀測性展示（請注意左側(cè)的W&B和Weave之間的無縫過渡）

借助Prefect自動提取數(shù)據(jù)

為了使推薦程序Rosebud保持準(zhǔn)確性，將數(shù)據(jù)提取和上傳到Pinecone向量數(shù)據(jù)庫的過程自動化非常重要。對于這個(gè)任務(wù)，我選擇使用Prefect（https://www.prefect.io/）。Prefect是一個(gè)流行的工作流編排工具。我一直在尋找一些輕量級、易于學(xué)習(xí)和Python風(fēng)格的程序。最后，我在Prefect中找到了這一切。

Prefect提供的用于提取和更新Pinecone向量存儲的自動流程

Prefect支持提供多種方式來規(guī)劃你的工作流程。我決定使用帶有自動基礎(chǔ)設(shè)施配置的推送工作池方式。我發(fā)現(xiàn)這種設(shè)置在簡單性和可配置性之間取得了平衡。它允許用戶委托Prefect自動配置在所選云提供商中運(yùn)行流所需的所有基礎(chǔ)設(shè)施。經(jīng)過幾番權(quán)衡后，我選擇在Azure上部署，但是在GCP或AWS上部署的話只需要更改幾行代碼即可。有關(guān)更多詳細(xì)信息，請參閱pinecone_flow.py文件。下面代碼只是提供了一個(gè)簡化的流程：

@task
def start():
"""
啟動：檢查一切工作或失敗的速度快！
"""
#打印出一些調(diào)試信息
print("Starting flow!")
# 確保用戶已經(jīng)設(shè)置了適當(dāng)?shù)沫h(huán)境變量
assert os.environ['LANGCHAIN_API_KEY']
assert os.environ['OPENAI_API_KEY']
...
@task(retries=3, retry_delay_seconds=[1, 10, 100])
def pull_data_to_csv(config):
TMBD_API_KEY = os.getenv('TMBD_API_KEY')
YEARS = range(config["years"][0], config["years"][-1] + 1)
CSV_HEADER = ['Title', 'Runtime (minutes)', 'Language', 'Overview', ...]

for year in YEARS:
# 獲取所有在{Year}中制作的電影的id列表
movie_list = list(set(get_id_list(TMBD_API_KEY, year)))

FILE_NAME = f'./data/{year}_movie_collection_data.csv'

#生成文件
with open(FILE_NAME, 'w') as f:
writer = csv.writer(f)
writer.writerow(CSV_HEADER)
...
print("Successfully pulled data from TMDB and created csv files in data/")
@task
def convert_csv_to_docs():
#從所有csv文件中加載數(shù)據(jù)
loader = DirectoryLoader(
...
show_progress=True)
docs = loader.load()
metadata_field_info = [
AttributeInfo(name="Title",
description="The title of the movie", type="string"),
AttributeInfo(name="Runtime (minutes)",
description="The runtime of the movie in minutes", type="integer"),
...
]

def convert_to_list(doc, field):
if field in doc.metadata and doc.metadata[field] is not None:
doc.metadata[field] = [item.strip()
for item in doc.metadata[field].split(',')]
...
fields_to_convert_list = ['Genre', 'Actors', 'Directors',
'Production Companies', 'Stream', 'Buy', 'Rent']
...
# 將'overview' 和'keywords' 設(shè)置為'page_content'，其他字段設(shè)置為'metadata'
for doc in docs:
#將page_counte字符串解析為字典
page_content_dict = dict(line.split(": ", 1)
for line in doc.page_content.split("\n") if ": " in line)

doc.page_content = (
'Title: ' + page_content_dict.get('Title') +
'. Overview: ' + page_content_dict.get('Overview') +
...
)
...
print("Successfully took csv files and created docs")

return docs
@task
def upload_docs_to_pinecone(docs, config):
# 創(chuàng)建空索引
PINECONE_KEY, PINECONE_INDEX_NAME = os.getenv(
'PINECONE_API_KEY'), os.getenv('PINECONE_INDEX_NAME')

pc = Pinecone(api_key=PINECONE_KEY)

# 目標(biāo)索引和檢查狀態(tài)
pc_index = pc.Index(PINECONE_INDEX_NAME)
print(pc_index.describe_index_stats())
embeddings = OpenAIEmbeddings(model=config['EMBEDDING_MODEL_NAME'])
namespace = "film_search_prod"
PineconeVectorStore.from_documents(
docs,
...
)
print("Successfully uploaded docs to Pinecone vector store")
@task
def publish_dataset_to_weave(docs):
#初始化Weave
weave.init('film-search')
rows = []
for doc in docs:
row = {
'Title': doc.metadata.get('Title'),
'Runtime (minutes)': doc.metadata.get('Runtime (minutes)'),
...
}
rows.append(row)

dataset = Dataset(name='Movie Collection', rows=rows)
weave.publish(dataset)
print("Successfully published dataset to Weave")

@flow(log_prints=True)
def pinecone_flow():
with open('./config.json') as f:
config = json.load(f)
start()
pull_data_to_csv(config)
docs = convert_csv_to_docs()
upload_docs_to_pinecone(docs, config)
publish_dataset_to_weave(docs)
if __name__ == "__main__":
pinecone_flow.deploy(
name="pinecone-flow-deployment",
work_pool_name="my-aci-pool",
cron="0 0 * * 0",
image=DeploymentImage(
name="prefect-flows:latest",
platform="linux/amd64",
)
)

請注意，將Python函數(shù)轉(zhuǎn)換為Prefect流是非常簡單的事情。你只需要在主函數(shù)上使用@task裝飾器和@flow裝飾器來設(shè)計(jì)一些子函數(shù)。還要注意，在將文檔上傳到Pinecone向量數(shù)據(jù)庫后，我們流程的最后一步是將數(shù)據(jù)集發(fā)布到Weave。這對于再現(xiàn)性很重要。為了學(xué)習(xí)Prefect的基礎(chǔ)知識，我建議你瀏覽一下他們官網(wǎng)上的教程（https://docs.prefect.io/latest/tutorial/）。

在上面腳本的最后，我們看到部署是如何在Prefect中完成的。

我們需要為部署提供一個(gè)名稱，這個(gè)名稱是自由決定的。
我們還需要指定一個(gè)work_pool_name。Prefect中的推送工作池會自動將任務(wù)發(fā)送到無服務(wù)器計(jì)算機(jī)，而不需要中介。此名稱需要與用于創(chuàng)建池的名稱相匹配，我們將在下面看到。
你還需要指定一個(gè)cron，它是計(jì)時(shí)器的縮寫。這允許你指定重復(fù)工作流的頻率。值“0 0**0”表示每周重復(fù)此工作流。
最后，你需要指定一個(gè)DeploymentImage。在這里，你可以指定名稱和平臺。名稱是任意的，但平臺不是。由于我想部署到Azure計(jì)算實(shí)例，而這些實(shí)例運(yùn)行Linux操作系統(tǒng)，所以我在DeploymentImage中指定這一點(diǎn)很重要。

要使用命令行方式在Azure上部署此流，請運(yùn)行以下命令：

prefect work-pool create --type azure-container-instance:push --provision-infra my-aci-pool
prefect deployment run 'get_repo_info/my-deployment'

這些命令將自動在Azure上提供所有必要的基礎(chǔ)設(shè)施。這包括一個(gè)Azure容器注冊表（ACR），它將保存一個(gè)Docker映像，其中包含目錄中的所有文件以及requirements.txt中列出的任何必要的依賴庫。它還將包括一個(gè)Azure容器實(shí)例（ACI）標(biāo)識，該標(biāo)識將具有部署具有上述Docker映像的容器所需的權(quán)限。最后，使用deployment run命令安排每周運(yùn)行的代碼。你可以通過Prefect控制面板來查看你的流是否運(yùn)行：

Prefect中的流正在成功運(yùn)行的情形

通過每周更新我的Pinecone向量庫，我可以確保來自程序Rosebud的推薦結(jié)果準(zhǔn)確。

總結(jié)

在本文中，我介紹了我的改進(jìn)后的Rosebud應(yīng)用程序的一些改進(jìn)方案。這包括整合離線和在線評估的過程，以及自動更新我的Pinecone向量庫等。

本文還有未提及的其他一些改進(jìn)，包括：

在電影數(shù)據(jù)中包括電影數(shù)據(jù)庫的評級。現(xiàn)在，你可以使用“好評電影（highly rated films）”，這種聊天模式能夠過濾掉7/10以上的電影。
升級了聊天模式。現(xiàn)在查詢和摘要模型使用的是模型gpt-4o-mini。請回想一下，LLM判斷模型也是使用了模型gpt-4o-mini。
嵌入模型從text-Embedding-ada-002升級為text-embeading-3-small。
現(xiàn)在的年份跨越1950年至2023年，而不是從1920年開始。1920年至1950年的電影數(shù)據(jù)質(zhì)量不高，只有糟糕的推薦。
用戶界面更加清晰，所有關(guān)于項(xiàng)目的細(xì)節(jié)都放在側(cè)邊欄中。
程序在GitHub上的文檔得到了進(jìn)一步的改進(jìn)。
修復(fù)了一些錯(cuò)誤。

正如文章一開始提到的，該應(yīng)用程序現(xiàn)在可以100%免費(fèi)使用！在可預(yù)見的未來，我將為查詢買單（因此選擇gpt-4o-mini而不是更昂貴的gpt-4o）。我真的很想獲得在生產(chǎn)環(huán)境中運(yùn)行應(yīng)用程序的經(jīng)驗(yàn)，并讓我的讀者測試Rosebud，這是一個(gè)很好的方法。萬一應(yīng)用程序真的“火爆”了，我將不得不想出其他的融資模式。但這會是一個(gè)很大的問題。

下面，請盡情享受使用Rosebud程序搜索精彩電影的樂趣吧！