An Implementation of Totally Traced and Evaluated Native LLM Pipeline Utilizing Opik for Clear, Measurable, and Reproducible AI Workflows

!pip set up -q opik transformers speed up torch


import torch
from transformers import pipeline
import textwrap


import opik
from opik import Opik, Immediate, monitor
from opik.analysis import consider
from opik.analysis.metrics import Equals, LevenshteinRatio


machine = 0 if torch.cuda.is_available() else -1
print("Utilizing machine:", "cuda" if machine == 0 else "cpu")


opik.configure()
PROJECT_NAME = "opik-hf-tutorial"

llm = pipeline(
   "text-generation",
   mannequin="distilgpt2",
   machine=machine,
)


def hf_generate(immediate: str, max_new_tokens: int = 80) -> str:
   end result = llm(
       immediate,
       max_new_tokens=max_new_tokens,
       do_sample=True,
       temperature=0.3,
       pad_token_id=llm.tokenizer.eos_token_id,
   )[0]["generated_text"]
   return end result[len(prompt):].strip()

plan_prompt = Immediate(
   title="hf_plan_prompt",
   immediate=textwrap.dedent("""
       You might be an assistant that creates a plan to reply a query
       utilizing ONLY the given context.


       Context:
       {{context}}


       Query:
       {{query}}


       Return precisely 3 bullet factors as a plan.
   """).strip(),
)


answer_prompt = Immediate(
   title="hf_answer_prompt",
   immediate=textwrap.dedent("""
       You reply based mostly solely on the given context.


       Context:
       {{context}}


       Query:
       {{query}}


       Plan:
       {{plan}}


       Reply the query in 2–4 concise sentences.
   """).strip(),
)

DOCS = {
   "overview": """
       Opik is an open-source platform for debugging, evaluating,
       and monitoring LLM and RAG purposes. It gives tracing,
       datasets, experiments, and analysis metrics.
   """,
   "tracing": """
       Tracing in Opik logs nested spans, LLM calls, token utilization,
       suggestions scores, and metadata to examine complicated LLM pipelines.
   """,
   "analysis": """
       Opik evaluations are outlined by datasets, analysis duties,
       scoring metrics, and experiments that combination scores,
       serving to detect regressions or points.
   """,
}


@monitor(project_name=PROJECT_NAME, sort="software", title="retrieve_context")
def retrieve_context(query: str) -> str:
   q = query.decrease()
   if "hint" in q or "span" in q:
       return DOCS["tracing"]
   if "metric" in q or "dataset" in q or "consider" in q:
       return DOCS["evaluation"]
   return DOCS["overview"]

@monitor(project_name=PROJECT_NAME, sort="llm", title="plan_answer")
def plan_answer(context: str, query: str) -> str:
   rendered = plan_prompt.format(context=context, query=query)
   return hf_generate(rendered, max_new_tokens=80)


@monitor(project_name=PROJECT_NAME, sort="llm", title="answer_from_plan")
def answer_from_plan(context: str, query: str, plan: str) -> str:
   rendered = answer_prompt.format(
       context=context,
       query=query,
       plan=plan,
   )
   return hf_generate(rendered, max_new_tokens=120)


@monitor(project_name=PROJECT_NAME, sort="common", title="qa_pipeline")
def qa_pipeline(query: str) -> str:
   context = retrieve_context(query)
   plan = plan_answer(context, query)
   reply = answer_from_plan(context, query, plan)
   return reply


print("Pattern reply:n", qa_pipeline("What does Opik assist builders do?"))

consumer = Opik()


dataset = consumer.get_or_create_dataset(
   title="HF_Opik_QA_Dataset",
   description="Small QA dataset for HF + Opik tutorial",
)


dataset.insert([
   {
       "question": "What kind of platform is Opik?",
       "context": DOCS["overview"],
       "reference": "Opik is an open-source platform for debugging, evaluating and monitoring LLM and RAG purposes.",
   },
   {
       "query": "What does tracing in Opik log?",
       "context": DOCS["tracing"],
       "reference": "Tracing logs nested spans, LLM calls, token utilization, suggestions scores, and metadata.",
   },
   {
       "query": "What are the elements of an Opik analysis?",
       "context": DOCS["evaluation"],
       "reference": "An Opik analysis makes use of datasets, analysis duties, scoring metrics and experiments that combination scores.",
   },
])

equals_metric = Equals()
lev_metric = LevenshteinRatio()


def evaluation_task(merchandise: dict) -> dict:
   output = qa_pipeline(merchandise["question"])
   return {
       "output": output,
       "reference": merchandise["reference"],
   }

evaluation_result = consider(
   dataset=dataset,
   process=evaluation_task,
   scoring_metrics=[equals_metric, lev_metric],
   experiment_name="HF_Opik_QA_Experiment",
   project_name=PROJECT_NAME,
   task_threads=1,
)


print("nExperiment URL:", evaluation_result.experiment_url)

agg = evaluation_result.aggregate_evaluation_scores()


print("nAggregated scores:")
for metric_name, stats in agg.aggregated_scores.gadgets():
   print(metric_name, "=>", stats)