使用 Ragas 评估基于 openGauss 的 RAG 管道 ​

class OpenGaussClient:
    """A simple openGauss client using psycopg2."""

    def __init__(self, host=None, port=None, user=None, password=None, database=None):
        self.host = host or os.getenv("OPENGAUSS_HOST", "localhost")
        self.port = port or int(os.getenv("OPENGAUSS_PORT", 5432))
        self.user = user or os.getenv("OPENGAUSS_USER", "gaussdb")
        self.password = password or os.getenv("OPENGAUSS_PASSWORD", "openGauss@123")
        self.database = database or os.getenv("OPENGAUSS_DB", "postgres")
        self.conn = psycopg2.connect(
            host=self.host,
            port=self.port,
            user=self.user,
            password=self.password,
            dbname=self.database
        )
        self.cursor = self.conn.cursor()

    def drop_table(self, table_name: str):
        """Drop table."""
        self.cursor.execute(
            sql.SQL(
                """
                DROP TABLE IF EXISTS public.{table_name};
                """
            ).format(table_name=sql.Identifier(table_name))
        )
        return self.cursor

    def create_table(self, table_name: str, dim: int = 1024):
        """Create table."""
        self.cursor.execute(
            sql.SQL(
                """
                CREATE TABLE IF NOT EXISTS public.{table_name}
                (
                    id TEXT PRIMARY KEY,
                    content TEXT NOT NULL,
                    embedding vector({dim}) NOT NULL,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                    );
                """
            ).format(table_name=sql.Identifier(table_name), dim=sql.Literal(dim)),
        )
        self.conn.commit()
        return self.cursor

    def insert_into_table(self, table_name: str, rows: List[List[Any]]) -> None:
        """Insert into the table."""
        self.cursor.executemany(
            sql.SQL(
                """
                INSERT INTO public.{table_name} (id, content, embedding)
                VALUES (%s, %s, %s)
                """
            )
            .format(table_name=sql.Identifier(table_name))
            .as_string(self.cursor),
            rows,
        )
        self.conn.commit()
        return self.cursor

    def query(self, table_name: str, question_vec: List[float], top_k: int = 3) -> List[Any]:
        """Execute a query and return the results."""
        self.cursor.execute(
            sql.SQL(
                """
                SELECT content
                FROM public.{table_name}
                ORDER BY embedding <-> %s::vector
                LIMIT %s:: int;
                """
            )
            .format(table_name=sql.Identifier(table_name)),
            (question_vec, top_k),
        )

        return [row[0] for row in self.cursor.fetchall()]

    def close(self):
        """Close the connection."""
        self.cursor.close()
        self.conn.close()


def chunk_source(urls: List[str]) -> List[str]:
    """Chunk text into chunks."""
    all_chunks: List[str] = []

    for url in tqdm(urls, desc="Chunking text"):
        random_name = "".join(random.choices(string.ascii_letters + string.digits, k=8))
        file_path = f"./{random_name}.md"
        try:
            urllib.request.urlretrieve(url, file_path)
            with open(file_path, "r") as file:
                file_text = file.read()
            text_lines = file_text.split("# ")
            text_lines = [line for line in text_lines if line.strip()]
            all_chunks.extend(text_lines)
        finally:
            if os.path.exists(file_path):
                os.remove(file_path)

    return all_chunks


class RAG:
    """RAG (Retrieval-Augmented Generation) class built upon OpenAI and openGauss."""

    def __init__(self, openai_client: OpenAI = None, opengauss_client: OpenGaussClient = None):
        self._prepare_openai(openai_client)
        self._prepare_opengauss(opengauss_client)

    def _embedder(self, text: str) -> List[float]:
        """embed text into embedding vector."""
        response = self.openai_client.embeddings.create(
            input=text,
            model=self.embedding_model
        )
        return response.data[0].embedding

    def _prepare_openai(self, openai_client: OpenAI, llm_model: str = "qwen3.5-plus", embedding_model: str = "text-embedding-v4") -> None:
        """Prepare OpenAI."""
        self.openai_client = openai_client
        self.llm_model = llm_model
        self.embedding_model = embedding_model
        self.SYSTEM_PROMPT = """你是一个智能问答助手。
你的任务是根据提供的参考资料（Context）回答用户问题。
请遵循以下规则：
1. 仅根据参考资料回答，不要编造信息。
2. 如果资料中没有答案，请直接说“根据提供的资料，无法回答该问题”。
3. 回答要简洁、准确。"""
        self.USER_PROMPT = """请根据以下参考资料回答问题。

<参考资料>
{contexts_text}
</参考资料>

<问题>
{question}
</问题>

请开始回答："""

    def _prepare_opengauss(self, opengauss_client: OpenGaussClient, table_name: str = "ragas_test"):
        """Prepare OpenGauss."""
        self.opengauss_client = opengauss_client
        self.table_name = table_name

        self.opengauss_client.drop_table(self.table_name)
        dim = len(self._embedder("get_dim"))
        self.opengauss_client.create_table(self.table_name, dim)

    def load(self, urls: List[str]):
        """Load the urls."""
        rows: list[Any] = []
        texts = chunk_source(urls)
        for text in tqdm(texts, desc="Loading data"):
            rows.append(
                (
                    str(uuid.uuid4()),
                    text,
                    self._embedder(text)
                )
            )

        self.opengauss_client.insert_into_table(table_name=self.table_name, rows=rows)

    def retrieve(self, question: str, top_k: int=3):
        """Retrieve the top k results."""
        question_vec = self._embedder(question)
        retrieve_texts = self.opengauss_client.query(self.table_name, question_vec, top_k)
        return retrieve_texts

    def answer(self, question: str, top_k: int = 3, return_retrieved_text: bool = False):
        """Answer the question."""
        retrieve_texts = self.retrieve(question, top_k)

        retrieve_texts_str : str = ""
        for i, c in enumerate(retrieve_texts, start=1):
            retrieve_texts_str += f"[{i}] {c}\n\n"

        completion = self.openai_client.chat.completions.create(
            model=self.llm_model,
            messages = [
                {"role": "system", "content": self.SYSTEM_PROMPT},
                {
                    "role": "user",
                    "content": self.USER_PROMPT.format(contexts_text=retrieve_texts_str, question=question),
                }
            ]
        )
        if return_retrieved_text:
            return completion.choices[0].message.content, retrieve_texts
        else:
            return completion.choices[0].message.content

openai_client = OpenAI(
    api_key=os.getenv(
        "DASHSCOPE_API_KEY",
        "sk-5e589e8d7788451fae664ec63be51685"
    ),
    base_url="https://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/compatible-mode/v1",
)

opengauss_client = OpenGaussClient()

rag = RAG(openai_client=openai_client, opengauss_client=opengauss_client)

urls = ["https://raw.gitcode.com/opengauss/docs/raw/master/docs/zh/datavec/datavec_overview.md"]

rag.load(urls)

question = "Datavec从openGauss的哪个版本开始引入？"
response = rag.answer(question=question, top_k=3, return_retrieved_text=True)
print(response)

user_input_list = [
    "Datavec从openGauss的哪个版本开始引入？",
    "DataVec目前支持的向量功能有哪些？",
    "openGauss DataVec支持的向量数据类型有哪些？"
]

reference_list = [
    "从openGauss 6.0.3版本开始引入。",
    "精确和近似的最近邻搜索、L2距离&余弦距离&内积、向量索引、向量操作函数和操作符。",
    "vector、bitvec、sparsevec、halfvec四种向量数据类型。"
]

retrieved_contexts_list: List[List[str]] = []
response_list: List[str] = []

for user_input in tqdm(user_input_list, desc="Answering questions"):
    response, retrieved_context = rag.answer(question=user_input, top_k=3, return_retrieved_text=True)
    retrieved_contexts_list.append(retrieved_context)
    response_list.append(response)

df = pd.DataFrame(
    {
        "user_input": user_input_list,
        "retrieved_contexts": retrieved_contexts_list,
        "response": response_list,
        "reference": reference_list,
    }
)

pd.set_option("display.max_columns", None)
pd.set_option("display.max_colwidth", 200)
pd.set_option("display.width", 2000)
print(df)

import asyncio
from openai import AsyncOpenAI
from pydantic import BaseModel

from ragas import experiment
from ragas.dataset import Dataset
from ragas.metrics.collections import AnswerRelevancy, Faithfulness, ContextRecall, ContextPrecision

from ragas.llms import llm_factory
from ragas.embeddings.base import embedding_factory


class EvalResult(BaseModel):
    user_input: str
    answer_relevancy: float
    faithfulness: float
    context_recall: float
    context_precision: float


evaluator_client = AsyncOpenAI(
    api_key=os.getenv(
        "DASHSCOPE_API_KEY",
        "sk-5e589e8d7788451fae664ec63be51685"
    ),
    base_url="https://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/compatible-mode/v1",
)

evaluator_llm = llm_factory(
    model="qwen3.6-plus",
    client=evaluator_client,
)

evaluator_embeddings = embedding_factory(
    provider="openai",
    model="text-embedding-v4",
    client=evaluator_client,
)

answer_relevancy = AnswerRelevancy(llm=evaluator_llm, embeddings=evaluator_embeddings)
faithfulness = Faithfulness(llm=evaluator_llm)
context_recall = ContextRecall(llm=evaluator_llm)
context_precision = ContextPrecision(llm=evaluator_llm)


@experiment(EvalResult)
async def run_evaluation(row):
    ar, f, cr, cp = await asyncio.gather(
        answer_relevancy.ascore(
            user_input=row["user_input"],
            response=row["response"]
        ),
        faithfulness.ascore(
            user_input=row["user_input"],
            response=row["response"],
            retrieved_contexts=row["retrieved_contexts"]
        ),
        context_recall.ascore(
            user_input=row["user_input"],
            retrieved_contexts=row["retrieved_contexts"],
            reference=row["reference"]
        ),
        context_precision.ascore(
            user_input=row["user_input"],
            reference=row["reference"],
            retrieved_contexts=row["retrieved_contexts"]
        ),
    )

    return EvalResult(
        user_input=row["user_input"],
        answer_relevancy=ar.value,
        faithfulness=f.value,
        context_recall=cr.value,
        context_precision=cp.value,
    )


dataset = Dataset.from_pandas(df, name="rag_eval", backend="inmemory")
results = asyncio.run(run_evaluation.arun(dataset=dataset))
df_results = results.to_pandas()

print(df_results[["answer_relevancy", "faithfulness", "context_recall", "context_precision"]].mean())