譯者 | 朱先忠

審校 | 重樓

幾個月前，我發布了一款電影搜索應用程序“Film Search”，這是一個檢索增強生成（RAG）應用程序，旨在根據用戶查詢實現電影推薦。例如，用戶可能會發出下面的英文提問：

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

中文意思：“給我找一部長度不到2小時、以狗為主角的英文劇情片?！?/span>

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

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.

對應的中文回答意思是：

電影名稱：忠犬八公的故事

片長：93分鐘

發行年份：2009

流媒體：不可用于流媒體

這部電影告訴了秋田犬Hachiko的辛酸真實故事，這種犬以其對主人的非凡忠誠而聞名。情感深度以及友誼和忠誠的主題引起了強烈的共鳴，使其成為一部感人的電影，展示了人與狗之間的深厚友誼。它非常適合想尋找一個發自內心的故事來強調友誼重要性的人。

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

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

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

在本文中，我將簡要介紹對以前開發的那款電影搜索應用程序“Film Search”所做的一些改進，具體的改進內容將包括：

• 使用RAGAS和Weave進行離線評估。
• 在線評估和可觀察性。
• 使用Prefect自動提取數據。

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

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

線下評估

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

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

RAG Triad（RAG三元組）

概括來看，三元組涵蓋了RAG應用程序的三個方面：

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

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

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

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

# 定義要判斷的指標
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個問題
]
#創建Weave評估對象
evaluation = weave.Evaluation(dataset=questions, scorers=[evaluate_with_ragas])
#運行評估
asyncio.run(evaluation.evaluate(model))
if __name__ == "__main__":
weave.init('film-search')
run_evaluation()

在上述代碼中，有幾點注意事項：

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

集成Weave+RAGAS的示例程序

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

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

在線評估

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

在線反饋示例

當用戶點擊上圖中底部任一按鈕時，就會被告知他們的反饋已被記錄。以下給出在Streamlit應用程序界面中如何實現這一點的代碼片段：

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發送反饋的過程是在單獨的線程上運行的，而不是在主線程上運行。這是為了防止用戶在等待日志記錄完成時被卡住幾秒鐘。

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

• 情緒（Sentiment）：用戶是否點擊了拇指圖標（同意/不同意）。
• 查詢（Query）：用戶的查詢，例如，查找長度不到2小時的英文戲劇電影和故事狗。
• Query_Constructor：查詢構造函數的結果，它重寫用戶的查詢，并在必要時包含元數據過濾，例如：

{
"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）：基于重建的查詢檢索到的上下文，例如標題“Title: Hachi: A Dog’s Tale. Overview: A drama based on the true story of a college professor’s…”。
• 回應（Response）：模型的回應。

所有這些都可以方便地記錄在與前面顯示的Weave評估相同的項目中?，F在，當查詢“不同意”情況時，只需按下拇指向下的圖標按鈕即可查看到底發生了什么。這將有助于使推薦應用程序Rosebud的迭代和改進加快速度。

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

借助Prefect自動提取數據

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

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

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

@task
def start():
"""
啟動：檢查一切工作或失敗的速度快！
"""
#打印出一些調試信息
print("Starting flow!")
# 確保用戶已經設置了適當的環境變量
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文件中加載數據
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' 設置為'page_content'，其他字段設置為'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):
# 創建空索引
PINECONE_KEY, PINECONE_INDEX_NAME = os.getenv(
'PINECONE_API_KEY'), os.getenv('PINECONE_INDEX_NAME')

pc = Pinecone(api_key=PINECONE_KEY)

# 目標索引和檢查狀態
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函數轉換為Prefect流是非常簡單的事情。你只需要在主函數上使用@task裝飾器和@flow裝飾器來設計一些子函數。還要注意，在將文檔上傳到Pinecone向量數據庫后，我們流程的最后一步是將數據集發布到Weave。這對于再現性很重要。為了學習Prefect的基礎知識，我建議你瀏覽一下他們官網上的教程（https://docs.prefect.io/latest/tutorial/）。

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

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

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

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

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

Prefect中的流正在成功運行的情形

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

總結

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

本文還有未提及的其他一些改進，包括：

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

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

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

譯者介紹

朱先忠，51CTO社區編輯，51CTO專家博客、講師，濰坊一所高校計算機教師，自由編程界老兵一枚。

原文標題：Productionizing a RAG App with Prefect, Weave, and RAGAS，作者：Ed Izaguirre

鏈接：https://towardsdatascience.com/productionizing-a-rag-app-04c857e0966e。