Python 工程師的秘密武器：MongoDB Atlas 向量搜尋的實作

# 設定 MongoDB Atlas 連線字串
try:
    MONGO_CONN_STR = os.environ["MONGODB_CONNECTION_STR"]
except KeyError:
    MONGO_CONN = getpass.getpass("請輸入您的 MongoDB Atlas 連線字串 (按Enter): ")

import pandas as pd
import s3fs

# 從 S3 讀取 JSON Lines 格式的資料
df = pd.read_json("https://ashwin-partner-bucket.s3.eu-west-1.amazonaws.com/movies_sample_dataset.jsonl", 
                  orient="records", lines=True)

# 將資料儲存到本地
df.to_json("./movies_sample_dataset.jsonl", orient="records", lines=True)

# 顯示前三筆資料
df[:3]

import numpy as np
from tqdm import tqdm
import openai

# 合併文字與概要欄位
df['final'] = df['text'] + " Overview: " + df['overview']

# 將資料分批處理以避免超過 OpenAI API 的速率限制
step = int(np.ceil(df['final'].shape[0]/100))
embeddings_t = []
lines = []

# 將資料分成較小的批次
for x, y in list(map(lambda x: (x, x+step), list(range(0, df.shape[0], step)))):
    lines += [df.final.values[x:y].tolist()]

# 為每個批次生成嵌入
for i in tqdm(lines):
    embeddings_t += openai.embeddings.create(
        model='text-embedding-ada-002', input=i).data

# 提取嵌入向量
out = []
for ele in embeddings_t:
    out += [ele.embedding]

# 將嵌入新增到 DataFrame
df['embedding'] = out

from pymongo import MongoClient
import os

# 初始化 MongoDB 客戶端
mongo_client = MongoClient(os.environ["MONGODB_CONNECTION_STR"])

# 上載檔案與向量嵌入到 MongoDB Atlas 集合
output_collection = mongo_client["sample_movies"]["embed_movies"]

# 清空現有集合
if output_collection.count_documents({}) > 0:
    output_collection.delete_many({})

# 插入資料
_ = output_collection.insert_many(df.to_dict("records"))

{
  "fields": [
    {
      "type": "vector",
      "numDimensions": 1536,
      "path": "embedding",
      "similarity": "cosine"
    },
    {
      "type": "filter",
      "path": "year"
    }
  ]
}

def query_vector_search(q, prefilter={}, postfilter={}, path="embedding", topK=2):
    # 將查詢轉換為向量嵌入
    ele = openai.embeddings.create(model='text-embedding-ada-002', input=q).data
    query_embedding = ele[0].embedding
    
    # 建立向量搜尋查詢
    vs_query = {
        "index": "default",
        "path": path,
        "queryVector": query_embedding,
        "numCandidates": 10,
        "limit": topK,
    }
    
    # 新增預過濾條件
    if len(prefilter) > 0:
        vs_query["filter"] = prefilter
    
    # 建立聚合管道
    new_search_query = {"$vectorSearch": vs_query}
    project = {"$project": {"score": {"$meta": "vectorSearchScore"}, "_id": 0, 
                          "title": 1, "release_date": 1, "overview": 1, "year": 1}}
    
    # 執行查詢，新增後過濾條件
    if len(postfilter.keys()) > 0:
        postFilter = {"$match": postfilter}
        res = list(output_collection.aggregate([new_search_query, project, postFilter]))
    else:
        res = list(output_collection.aggregate([new_search_query, project]))
    
    return res

# 使用年份作為預過濾條件的查詢
query_vector_search("我喜歡聖誕節電影，有沒有 1990 年之後的推薦？", 
                   prefilter={"year": {"$gt": 1990}}, topK=5)

# 使用年份預過濾和分數後過濾的查詢
query_vector_search("我喜歡聖誕節電影，有沒有 1990 年之後的推薦？", 
                   prefilter={"year": {"$gt": 1990}}, 
                   postfilter={"score": {"$gt": 0.905}}, topK=5)

from langchain_mongodb.vectorstores import MongoDBAtlasVectorSearch
from langchain_openai import OpenAIEmbeddings
import json

# 初始化嵌入模型
embedding_model = OpenAIEmbeddings(model="text-embedding-ada-002")

# 建立向量搜尋例項
vector_search = MongoDBAtlasVectorSearch(output_collection, embedding_model, text_key='final')

# 設定過濾條件
fquery = {"year": {"$gt": 1990}}
search_kwargs = {
    "k": 5,
    'filter': fquery,
}

# 建立檢索器
retriever = vector_search.as_retriever(search_kwargs=search_kwargs)

# 執行查詢
docs = retriever.invoke("我喜歡聖誕節電影，有沒有 1990 年之後的推薦？")

# 顯示結果
for doc in docs:
    foo = {}
    foo['title'] = doc.metadata['title']
    foo['year'] = doc.metadata['year']
    foo['final'] = doc.metadata['text']
    print(json.dumps(foo, indent=1))

from openai import OpenAI
client = OpenAI()

def invoke_llm(prompt, model_name='gpt-3.5-turbo-0125'):
    """
    使用聊天完成API向OpenAI傳送提示並取得模型回應
    """
    response = client.chat.completions.create(
        model=model_name,
        messages=[
            {
                "role": "user",
                "content": prompt
            }
        ],
        temperature=0.2,
        max_tokens=256,
        top_p=1,
        frequency_penalty=0,
        presence_penalty=0
    )
    chatbot_response = response.choices[0].message.content.strip()
    return chatbot_response

invoke_llm("This is a test")

'Great! What do you need help with?'

def get_prompt(question, context):
    prompt = f"""Question: {question}
System: Let's think step by step.
Context: {context}
"""
    return prompt

def get_recommendation_prompt(query, context):
    prompt = f"""
From the given movie listing data, choose a few great movie
recommendations.
User query: {query}
Context: {context}
Movie Recommendations:
1. Movie_name: Movie_overview
"""
    return prompt

print(invoke_llm("In which movie does a footballer go completely blind?"))

"The Game of Their Lives" (2005), where the character Davey Herold, a
footballer, goes completely blind after being hit in the head during a game

idea = "In which movie does a footballer go completely blind?"
search_response = query_vector_search(idea, prefilter={"year":{"$gt": 1990}}, postfilter={"score": {"$gt":0.8}},topK=10)
premise = "\n".join(list(map(lambda x:x['final'], search_response)))
print(invoke_llm(get_prompt(idea, premise)))

The movie in which a footballer goes completely blind is "23 Blast."

question = "I like Christmas movies, any recommendations for movies release after 1990?"
search_response = query_vector_search(question,topK=10)
context = "\n".join(list(map(lambda x:x['final'], search_response)))
print(invoke_llm(get_recommendation_prompt("I like Christmas movies, any recommendations for movies release after 1990?", context)))

import pandas as pd
import s3fs
import os
import boto3

s3_uri= "https://ashwin-partner-bucket.s3.eu-west-1.amazonaws.com/fashion_dataset.jsonl"
df = pd.read_json(s3_uri, orient="records", lines=True)
print(df[:3])

from pymongo import MongoClient
mongo_client = MongoClient(os.environ["MONGODB_CONNECTION_STR"])
# 上載檔案及其向量嵌入到MongoDB Atlas集合
col = mongo_client["search"]["catalog_final_myn"]
col.insert_many(df.to_dict(orient="records"))

{
  "fields": [
    {
      "type": "vector",
      "numDimensions": 1536,
      "path": "openAIVec",
      "similarity": "cosine"
    }
  ]
}

from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain_community.embeddings import OpenAIEmbeddings
from langchain_mongodb.vectorstores import MongoDBAtlasVectorSearch
from pymongo import MongoClient
from typing import List
from itertools import chain
import certifi
import os
from dotenv import load_dotenv
load_dotenv()
from functools import lru_cache

@lru_cache
def get_openai_emb_transformers():
    """
    回傳OpenAI轉換器模型的OpenAIEmbeddings例項。
    此函式建立並回傳OpenAIEmbeddings類別的例項，
    提供對OpenAI轉換器模型的存取，用於自然語言處理。
    該例項使用lru_cache裝飾器進行快取，以便高效重用。
    
    Returns:
        embeddings (OpenAIEmbeddings): OpenAIEmbeddings類別的例項。
    """
    embeddings = OpenAIEmbeddings()
    return embeddings

@lru_cache
def get_vector_store():
    """
    取得MongoDB Atlas的向量儲存。
    
    Returns:
        MongoDBAtlasVectorSearch: 向量儲存物件。
    """
    vs = MongoDBAtlasVectorSearch(
        collection=col, 
        embedding=get_openai_emb_transformers(), 
        index_name="vector_index_openAi_cosine", 
        embedding_key="openAIVec", 
        text_key="title"
    )
    return vs

@lru_cache(10)
def get_conversation_chain_conv():
    """
    取得用於聊天對話的對話鏈模型。
    
    Returns:
        ChatOpenAI: 用於聊天對話的對話鏈模型。
    """
    llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.2, max_tokens=2048)
    return llm

from pymongo import MongoClient
from dotenv import load_dotenv
import os
from functools import lru_cache
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores.mongodb_atlas import MongoDBAtlasVectorSearch

# 載入環境變數
load_dotenv()

# 連線MongoDB Atlas
@lru_cache
def get_openai_emb_transformers():
    """取得OpenAI嵌入模型例項"""
    return OpenAIEmbeddings()

# 連線向量儲存
@lru_cache
def get_vector_store():
    """取得MongoDB Atlas向量儲存例項"""
    client = MongoClient(os.environ.get("MONGODB_ATLAS_URI"))
    collection = client[os.environ.get("MONGODB_DATABASE")][os.environ.get("MONGODB_COLLECTION")]
    
    return MongoDBAtlasVectorSearch.from_connection_string(
        connection_string=os.environ.get("MONGODB_ATLAS_URI"),
        namespace=f"{os.environ.get('MONGODB_DATABASE')}.{os.environ.get('MONGODB_COLLECTION')}",
        embedding=get_openai_emb_transformers(),
        index_name=os.environ.get("MONGODB_INDEX_NAME")
    )

from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferMemory
from langchain.chat_models import ChatOpenAI

def get_conversation_chain():
    """建立基本的對話檢索鏈"""
    vectorstore = get_vector_store()
    llm = ChatOpenAI(temperature=0)
    
    memory = ConversationBufferMemory(
        memory_key="chat_history",
        return_messages=True
    )
    
    chain = ConversationalRetrievalChain.from_llm(
        llm=llm,
        retriever=vectorstore.as_retriever(),
        memory=memory
    )
    
    return chain

from typing import List
from langchain.output_parsers import JsonOutputParser
from langchain.prompts import PromptTemplate
from pydantic import BaseModel, Field
from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferWindowMemory

@lru_cache
def get_conversation_chain_conv():
    """取得對話鏈"""
    llm = ChatOpenAI(temperature=0)
    chain = ConversationChain(
        llm=llm,
        memory=ConversationBufferWindowMemory(k=5)
    )
    return llm

# 定義資料結構
class ProductRecoStatus(BaseModel):
    """
    表示產品推薦狀態
    """
    relevancy_status: bool = Field(description="根據查詢連貫的背景與環境判斷是否與購買時尚服飾/配飾相關")
    recommendations: List[str] = Field(description="根據查詢連貫的背景與環境的推薦產品標題列表")

class Product(BaseModel):
    """
    表示一個產品
    """
    title: str = Field(description="產品標題")
    baseColour: List[str] = Field(description="產品基本顏色列表")
    gender: List[str] = Field(description="產品目標性別列表")
    articleType: str = Field(description="產品類別")
    mfg_brand_name: str = Field(description="製造商或品牌名稱")

# 建立解析器和提示範本
reco_status_parser = JsonOutputParser(pydantic_object=ProductRecoStatus)
reco_status_prompt = PromptTemplate(
    template="你是一個AI助手，負責識別查詢中是否存在產品購買意圖並提供合適的時尚推薦。\n{format_instructions}\n{query}\n\n#聊天歷史摘要: {chat_history}\n\n根據查詢連貫的背景與環境，請提供相關性狀態和推薦產品列表。",
    input_variables=["query", "chat_history"],
    partial_variables={"format_instructions": reco_status_parser.get_format_instructions()},
)

# 定義產品推薦狀態取得函式
def get_product_reco_status(query: str, chat_history: List[str] = []):
    """
    根據給定查詢和聊天歷史取得產品推薦狀態
    """
    llm = get_conversation_chain_conv()
    chain = reco_status_prompt | llm | reco_status_parser
    resp = chain.invoke({"query": query, "chat_history": chat_history})
    return resp

query = "能推薦一些適合與男友約會的休閒洋裝嗎？"
status = get_product_reco_status(query)
print(status)
print(get_product_recommendations(query, reco_queries=status["recommendations"], chat_history=[]))

{'relevancy_status': True,
'recommendations': ['花卉印花裹身洋裝',
'一字領荷葉邊洋裝',
'蕾絲合身傘擺洋裝',
'中長絲綢洋裝',
'牛仔襯衫洋裝']}

query = "我該帶男友去哪裡約會？"
status = get_product_reco_status(query)
print(status)
print(get_conversation_chain_conv().invoke(query).content)

{'relevancy_status': False, 'recommendations': []}

from prettytable import PrettyTable

input_relevance_guardrail_data = [
    {
        "input": "七月份在紐約應該做什麼？",
        "output": True,
        "expected": True
    },
    {
        "input": "你能幫我解數學作業嗎？",
        "output": False,
        "expected": False
    },
    {
        "input": "法國的首都是哪裡？",
        "output": False,
        "expected": True
    },
]

def evaluate_correctness(output, expected):
    return 1 if output == expected else 0

def calculate_average(scores):
    return sum(scores) / len(scores)

def create_table(data):
    table = PrettyTable()
    table.field_names = ["Input", "Output", "Expected", "Score"]
    scores = [evaluate_correctness(case["output"], case["expected"]) for case in data]
    
    for case, score in zip(data, scores):
        table.add_row([case["input"], case["output"], case["expected"], score])
    
    # 增加視覺分隔
    table.add_row(["", "", "", ""])
    
    # 新增平均分數至表格底部
    average_score = calculate_average(scores)
    table.add_row(["Average", "", "", f"{average_score:.4f}"])
    
    return table

+--------------------------------------------+--------+----------+--------+
| Input                                      | Output | Expected | Score  |
+--------------------------------------------+--------+----------+--------+
| What should I do in New York City in July? | True   | True     | 1      |
| Can you help me with my math homework?     | False  | False    | 1      |
| What's the capital of France?              | False  | True     | 0      |
|                                            |        |          |        |
| Average                                    |        |          | 0.6667 |
+--------------------------------------------+--------+----------+--------+

from prettytable import PrettyTable
import sacrebleu
from rouge_score import rouge_scorer

# 計算BLEU分數的函式
def calculate_bleu(output, reference):
    bleu = sacrebleu.corpus_bleu([output], [[reference]])
    return bleu.score

# 計算ROUGE分數的函式
def calculate_rouge(output, contexts):
    if not contexts:
        return 0.0  # 如果沒有連貫的背景與環境可用，回傳0
    
    scorer = rouge_scorer.RougeScorer(['rouge1'], use_stemmer=True)
    
    # 計算輸出與每個連貫的背景與環境的ROUGE分數
    scores = [scorer.score(output, context)['rouge1'].fmeasure for context in contexts]
    
    # 回傳最高分數，表示輸出與最相關連貫的背景與環境的相似度
    return max(scores) if scores else 0.0

# 評估並格式化結果
def evaluate_responses(data):
    table = PrettyTable()
    table.field_names = ["Input", "BLEU Score", "ROUGE Score"]
    
    bleu_scores = []
    rouge_scores = []
    
    for case in data:
        bleu = calculate_bleu(case["output"], case["golden_answer"])
        rouge = calculate_rouge(case["output"], case["contexts"])
        
        table.add_row([case["input"], f"{bleu:.4f}", f"{rouge:.4f}"])
        
        bleu_scores.append(bleu)
        rouge_scores.append(rouge)
    
    # 增加視覺分隔
    table.add_row(["", "", ""])
    
    # 新增平均分數
    avg_bleu = sum(bleu_scores) / len(bleu_scores)
    avg_rouge = sum(rouge_scores) / len(rouge_scores)
    table.add_row(["Average", f"{avg_bleu:.4f}", f"{avg_rouge:.4f}"])
    
    return table

# 統計評估工具
def evaluate_bleu(output, golden_answer):
    bleu = sacrebleu.corpus_bleu([output], [[golden_answer]])
    return bleu.score / 100  # 將BLEU分數標準化至0到1之間

def evaluate_rouge(output, contexts):
    context_text = ("\n").join(contexts)
    scorer = rouge_scorer.RougeScorer(['rouge1', 'rouge2', 'rougeL'], use_stemmer=True)
    scores = scorer.score(context_text, output)
    return scores['rougeL'].fmeasure

def calculate_average(scores):
    return sum(scores) / len(scores)

# 截斷字串以便在表格中更容易顯示
def truncate_string(s, max_length=10):
    return (s[:max_length] + '...') if len(s) > max_length else s

def create_table(data):
    table = PrettyTable()
    table.field_names = ["Input", "Output", "Golden Answer", "# Contexts", "BLEU", "ROUGE"]
    
    bleu_scores = [evaluate_bleu(case["output"], case["golden_answer"]) for case in data]
    rouge_scores = [evaluate_rouge(case["output"], case["contexts"]) for case in data]
    
    for case, bleu, rouge in zip(data, bleu_scores, rouge_scores):
        table.add_row([
            truncate_string(case["input"]),
            truncate_string(case["output"]),
            truncate_string(case["golden_answer"]),
            len(case["contexts"]),
            f"{bleu:.4f}",
            f"{rouge:.4f}"])
    
    # 加入空白行以便視覺分隔
    table.add_row(["", "", "", "", "", ""])
    
    # 在表格底部加入平均分數
    average_bleu = calculate_average(bleu_scores)
    average_rouge = calculate_average(rouge_scores)
    table.add_row(["Average", "", "", "", f"{average_bleu:.4f}", f"{average_rouge:.4f}"])
    
    return table

+---------------+---------------+---------------+------------+--------+--------+
| Input         | Output        | Golden Answer | # Contexts | BLEU   | ROUGE  |
+---------------+---------------+---------------+------------+--------+--------+
| What shoul... | Check out ... | Explore Ce... | 3          | 0.0951 | 0.2857 |
| Can you he... | I'm design... | I am a tra... | 0          | 0.0270 | 0.0000 |
| What's the... | The capita... | Paris is t... | 2          | 0.2907 | 0.2857 |
|               |               |               |            |        |        |
| Average       |               |               |            | 0.1376 | 0.1905 |
+---------------+---------------+---------------+------------+--------+--------+

import json
from prettytable import PrettyTable
import openai
import os

# Add your OpenAI API key to call the model
openai.api_key = os.getenv("OPENAI_API_KEY")

# Data to evaluate
evaluation_data = [
    {
        "input": "What should I do in New York City in July?",
        "output": "Check out Times Square, go to an outdoor concert, and visit the Statue of Liberty.",
    },
    {
        "input": "Can you help me with my math homework?",
        "output": "I'm designed to assist with travel queries. For math help, try using online resources like Khan Academy or Mathway.",
    },
    {
        "input": "What's the capital of France?",
        "output": "The capital of France is Paris.",
    }
]

def evaluate_includes_recommendation(input, output):
    # Few-shot examples to help the model produce better answers.
    few_shot_examples = [
        {
            "input": "What are some good restaurants in Paris?",
            "output": "Try Le Jules Verne for an upscale dining experience, or visit Le Relais de l'Entrecôte for a classic steak frites.",
            "recommendation": True
        },
        # 其他範例...
    ]
    
    # Constructing the prompt
    prompt = """Determine whether the following output includes a recommendation based on the input.
Format response as a JSON object with the shape { "recommendation": boolean }.
Examples:
"""

# Append few-shot examples to the prompt.
for example in few_shot_examples:
    prompt += f"""Input: {example['input']}
Output: {example['output']}
Recommendation: {{ "recommendation": {str(example['recommendation']).lower()} }}
"""

prompt += f"""Input: {input}
Output: {output}
Recommendation:"""

# Call the OpenAI API
response = openai.chat.completions.create(
    # Use strong evaluator LLM
    model="gpt-4o",
    ## Format response as JSON, so it is easier to parse
    response_format={ "type": "json_object" },
    messages=[{ "role": "user", "content": prompt }],
    # Make sure temperature=0 for consistent outputs
    temperature=0
)

recommendation = json.loads(response.choices[0].message.content)["recommendation"]
return 1 if recommendation == True else 0

def calculate_average(scores):
    return sum(scores) / len(scores)

def truncate_string(s, max_length=30):
    return (s[:max_length] + '...') if len(s) > max_length else s

def create_table(data):
    table = PrettyTable()
    table.field_names = ["Input", "Output", "Score"]
    scores = [evaluate_includes_recommendation(case["input"], case["output"]) for case in data]
    # 表格新增行的程式碼

def evaluate_helpfulness(input, output):
    # 評估回應的有用性
    # 實作類別似於evaluate_includes_recommendation的邏輯

def evaluate_accuracy(input, output):
    # 評估回應的準確性
    # 可能需要外部知識來源來驗證

def evaluate_professionalism(input, output):
    # 評估回應的專業性和語氣

# 新增空白列以增加視覺分隔
table.add_row(["", "", ""])

# 計算並新增平均分數到表格底部
average = calculate_average(scores)
table.add_row(["Average", "", f"{average:.4f}"])
return table

# 建立並列印表格
result_table = create_table(evaluation_data)
print(result_table)

+-----------------------------------+-----------------------------------+--------+
| Input                             | Output                            | Score  |
+-----------------------------------+-----------------------------------+--------+
| What should I do in New York C... | Check out Times Square, go to ... | 1      |
| Can you help me with my math h... | I'm designed to assist with tr... | 1      |
| What's the capital of France?     | The capital of France is Paris... | 0      |
|                                   |                                   |        |
| Average                           |                                   | 0.6667 |
+-----------------------------------+-----------------------------------+--------+

忠實度 = 可從連貫的背景與環境推斷的宣告數量 / 回答中的總宣告數量

pip3 install ragas==0.1.13 langchain-openai==0.1.20 datasets==2.20.0

from ragas.metrics import faithfulness
from ragas import evaluate
from datasets import Dataset
from langchain_openai.chat_models import ChatOpenAI
import os

openai_api_key = os.getenv("OPENAI_API_KEY")
evaluation_data = [
    {
        "input": "What should I do in New York City in July?",
        "output": "Check out Times Square, go to an outdoor concert, and visit the Statue of Liberty.",
        "contexts": [
            "Times Square is known for its Broadway theaters, bright lights, and bustling atmosphere.",
            "Outdoor concerts in Central Park are popular summer events attracting many visitors.",
            "The Statue of Liberty is a symbol of freedom and a must-see landmark in NYC."
        ]
    },
    {
        "input": "Can you help me with my math homework?",
        "output": "I'm designed to assist with travel queries. For math help, try using online resources like Khan Academy or Mathway.",
        "contexts": []
    },
    {
        "input": "What's the capital of France?",
        "output": "The capital of France is Paris.",
        "contexts": [
            "Paris, known as the City of Light, is the most populous city of France.",
            "European capitals: Paris, France; Berlin, Germany; Madrid, Spain",
        ]
    }
]

def prepare_data_for_ragas(data_list):
    data_table = {
        'question': [],
        'answer': [],
        'contexts': []
    }
    for data_item in data_list:
        data_table["question"].append(data_item["input"])
        data_table["answer"].append(data_item["output"])
        data_table["contexts"].append(data_item["contexts"])
    return data_table

def create_report(data):
    ragas_dict = prepare_data_for_ragas(data)
    dataset = Dataset.from_dict(prepare_data_for_ragas(data))
    langchain_llm = ChatOpenAI(
        model_name="gpt-4o-mini",
        api_key=openai_api_key)
    score = evaluate(dataset, metrics=[faithfulness], llm=langchain_llm)
    return score

# 建立並列印評估表
results = create_report(evaluation_data)
print(results.to_pandas())
print(results)

Evaluating: 100%
3/3 [00:05<00:00, 1.72s/it]
                               question                                        \
0  What should I do in New York City in July?
1       Can you help me with my math homework?
2               What's the capital of France?

                                             answer                                        \
0  Check out Times Square, go to an outdoor conce...
1  I'm designed to assist with travel queries. Fo...
2                    The capital of France is Paris.

                                           contexts faithfulness
0  [Times Square is known for its Broadway theate...          1.0
1                                                []          0.0
2  [Paris, known as the City of Light, is the mos...          1.0
{'faithfulness': 0.6667}

pip3 install ragas==0.1.13 langchain-openai==0.1.20 datasets==2.20.0

from ragas.metrics import answer_relevancy
from ragas import evaluate
from datasets import Dataset
from langchain_openai.chat_models import ChatOpenAI
from langchain_openai.embeddings import OpenAIEmbeddings
import os

openai_api_key = os.getenv("OPENAI_API_KEY")

evaluation_data = [
    {
        "input": "What should I do in New York City in July?",
        "output": "Check out Times Square, go to an outdoor concert, and visit the Statue of Liberty.",
        "contexts": [
            "Times Square is known for its Broadway theaters, bright lights, and bustling atmosphere.",
            "Outdoor concerts in Central Park are popular summer events attracting many visitors.",
            "The Statue of Liberty is a symbol of freedom and a must-see landmark in NYC."
        ]
    },
    {
        "input": "Can you help me with my math homework?",
        "output": "I'm designed to assist with travel queries. For math help, try using online resources like Khan Academy or Mathway.",
        "contexts": []
    },
    {
        "input": "What's the capital of France?",
        "output": "The capital of France is Paris.",
        "contexts": [
            "Paris, known as the City of Light, is the most populous city of France.",
            "European capitals: Paris, France; Berlin, Germany; Madrid, Spain",
        ]
    }
]

def prepare_data_for_ragas(data_list):
    data_table = {
        'question': [],
        'answer': [],
        'contexts': []
    }
    for data_item in data_list:
        data_table["question"].append(data_item["input"])
        data_table["answer"].append(data_item["output"])
        data_table["contexts"].append(data_item["contexts"])
    return data_table

def create_report(data):
    dataset = Dataset.from_dict(prepare_data_for_ragas(data))
    
    # 設定LLM和嵌入模型
    langchain_llm = ChatOpenAI(
        model_name="gpt-4o-mini",
        api_key=openai_api_key)
    
    embeddings = OpenAIEmbeddings(
        api_key=openai_api_key)
    
    # 進行評估
    score = evaluate(
        dataset, 
        metrics=[answer_relevancy], 
        llm=langchain_llm,
        embeddings=embeddings)
    
    return score

# 執行評估並輸出結果
results = create_report(evaluation_data)
print(results.to_pandas())
print(results)

from ragas.metrics import faithfulness, answer_relevancy, context_precision, context_recall

def comprehensive_evaluation(data):
    dataset = Dataset.from_dict(prepare_data_for_ragas(data))
    
    langchain_llm = ChatOpenAI(
        model_name="gpt-4o-mini",
        api_key=openai_api_key)
    
    embeddings = OpenAIEmbeddings(
        api_key=openai_api_key)
    
    # 使用多種評估指標
    metrics = [
        faithfulness,
        answer_relevancy,
        context_precision,
        context_recall
    ]
    
    score = evaluate(
        dataset, 
        metrics=metrics, 
        llm=langchain_llm,
        embeddings=embeddings)
    
    return score