DeepEval

pip install deepeval surrealdb openai   # swap OpenAI for any embedder you like
# optional – start a local SurrealDB node
docker run -p 8000:8000 surrealdb/surrealdb:latest \
       start --user root --pass secret file:/data/db

-- SurrealQL – run in the DB console once.
DEFINE TABLE rag_docs SCHEMALESS;
DEFINE FIELD id        ON rag_docs TYPE string;  -- primary key
DEFINE FIELD text      ON rag_docs TYPE string;
DEFINE FIELD source    ON rag_docs TYPE string;
DEFINE FIELD embedding ON rag_docs TYPE array;   -- float[]

-- Fast approximate NN (cosine, 1536-D, OpenAI embeddings here)
DEFINE INDEX IF NOT EXISTS rag_docs_vec
  ON rag_docs FIELDS embedding
  HNSW DIMENSION 1536 DIST COSINE;

surreal_rag.py

from surrealdb import AsyncSurreal
import hashlib, json, os
from typing import List, Dict, Any
import openai                         # or any local embedding model

_EMBED_DIM = 1536

def embed(text: str) -> List[float]:
    resp = openai.Embedding.create(
        model="text-embedding-3-small",
        input=[text],
        dimensions=_EMBED_DIM,
        api_key=os.getenv("OPENAI_API_KEY"),
    )
    return resp["data"][0]["embedding"]


class SurrealRAG:
    def __init__(
        self,
        url: str = "ws://localhost:8000/rpc",
        namespace: str = "rag",
        database: str = "demo",
        user: str = "root",
        password: str = "secret",
    ):
        self.url = url
        self.namespace = namespace
        self.database = database
        self.user = user
        self.password = password

    async def _ensure_table(self):
        async with AsyncSurreal(self.url) as db:
            await db.signin({"username": self.user, "password": self.password})
            await db.use(self.namespace, self.database)
            await db.query(
                """
                DEFINE TABLE rag_docs SCHEMALESS;
                DEFINE FIELD id        ON rag_docs TYPE string;
                DEFINE FIELD text      ON rag_docs TYPE string;
                DEFINE FIELD source    ON rag_docs TYPE string;
                DEFINE FIELD embedding ON rag_docs TYPE array;
                DEFINE INDEX IF NOT EXISTS rag_docs_vec
                    ON rag_docs FIELDS embedding
                    HNSW DIMENSION $dim DIST COSINE;
                """,
                {"dim": _EMBED_DIM}
            )

    # --- ingest --------------------------------------------------------
    async def add(self, docs: List[Dict[str, str]]):
        await self._ensure_table()
        async with AsyncSurreal(self.url) as db:
            await db.signin({"username": self.user, "password": self.password})
            await db.use(self.namespace, self.database)
            for d in docs:
                rec = {
                    "id": hashlib.sha1(d["text"].encode()).hexdigest(),
                    "text": d["text"],
                    "source": d["source"],
                    "embedding": embed(d["text"]),
                }
                await db.create("rag_docs", rec)

    # --- retrieve ------------------------------------------------------
    async def query(self, text: str, k: int = 4) -> List[Dict[str, Any]]:
        await self._ensure_table()
        vec = embed(text)
        async with AsyncSurreal(self.url) as db:
            await db.signin({"username": self.user, "password": self.password})
            await db.use(self.namespace, self.database)
            result = await db.query(
                """
                SELECT text, source,
                       vector::distance::cosine(embedding, $vec) AS score
                FROM rag_docs
                WHERE embedding <|$k|> $vec
                ORDER BY score ASC
                """,
                {
                    "vec": vec,
                    "k": k
                }
            )
            rows = result[0]["result"]
            return [{"context": r["text"], "source": r["source"], "score": r["score"]} for r in rows]

import asyncio
from surreal_rag import SurrealRAG
from deepeval.test_case import LLMTestCase
from deepeval.metrics import (
    AnswerRelevancyMetric,
    FaithfulnessMetric,
    ContextualPrecisionMetric,
)
from deepeval import evaluate
import openai

async def main():
    rag = SurrealRAG()

    # 1. Populate SurrealDB (first run only)
    await rag.add([
        {
            "text": "The tomato is botanically classified as a berry because it develops \
                     from a single ovary and contains seeds.",
            "source": "https://en.wikipedia.org/wiki/Tomato",
        },
        {
            "text": "A cucumber is a pepo, a type of berry with a hard rind.",
            "source": "https://en.wikipedia.org/wiki/Cucumber",
        },
        {
            "text": "Strawberries are accessory fruits; their 'seeds' are achenes.",
            "source": "https://en.wikipedia.org/wiki/Strawberry",
        },
    ])

    # 2. Retrieve context
    query = "Which fruit is a berry but people think it's a vegetable?"
    context = await rag.query(query, k=3)

    # 3. Build prompt & generate answer
    prompt = f"""Answer the question.\n\nContext:\n{context}\n\nQ: {query}\nA:"""
    resp = openai.ChatCompletion.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
    )
    answer = resp.choices[0].message.content.strip()

    # 4. DeepEval test case
    test_case = LLMTestCase(
        input=query,
        actual_output=answer,
        expected_output="tomato",
        retrieval_context=context,
    )

    # 5. Evaluate
    evaluate(
        test_cases=[test_case],
        metrics=[
            AnswerRelevancyMetric(threshold=0.7),
            FaithfulnessMetric(threshold=0.7),
            ContextualPrecisionMetric(top_k=3),
        ],
    )

if __name__ == "__main__":
    asyncio.run(main())