Find out how to Construct a Self-Evaluating Agentic AI System with LlamaIndex and OpenAI Utilizing Retrieval, Software Use, and Automated High quality Checks

!pip -q set up -U llama-index llama-index-llms-openai llama-index-embeddings-openai nest_asyncio


import os
import asyncio
import nest_asyncio
nest_asyncio.apply()


from getpass import getpass


if not os.environ.get("OPENAI_API_KEY"):
   os.environ["OPENAI_API_KEY"] = getpass("Enter OPENAI_API_KEY: ")

from llama_index.core import Doc, VectorStoreIndex, Settings
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding


Settings.llm = OpenAI(mannequin="gpt-4o-mini", temperature=0.2)
Settings.embed_model = OpenAIEmbedding(mannequin="text-embedding-3-small")


texts = [
   "Reliable RAG systems separate retrieval, synthesis, and verification. Common failures include hallucination and shallow retrieval.",
   "RAG evaluation focuses on faithfulness, answer relevancy, and retrieval quality.",
   "Tool-using agents require constrained tools, validation, and self-review loops.",
   "A robust workflow follows retrieve, answer, evaluate, and revise steps."
]


docs = [Document(text=t) for t in texts]
index = VectorStoreIndex.from_documents(docs)
query_engine = index.as_query_engine(similarity_top_k=4)

from llama_index.core.analysis import FaithfulnessEvaluator, RelevancyEvaluator


faith_eval = FaithfulnessEvaluator(llm=Settings.llm)
rel_eval = RelevancyEvaluator(llm=Settings.llm)


def retrieve_evidence(q: str) -> str:
   r = query_engine.question(q)
   out = []
   for i, n in enumerate(r.source_nodes or []):
       out.append(f"[{i+1}] {n.node.get_content()[:300]}")
   return "n".be part of(out)


def score_answer(q: str, a: str) -> str:
   r = query_engine.question(q)
   ctx = [n.node.get_content() for n in r.source_nodes or []]
   f = faith_eval.consider(question=q, response=a, contexts=ctx)
   r = rel_eval.consider(question=q, response=a, contexts=ctx)
   return f"Faithfulness: {f.rating}nRelevancy: {r.rating}"

from llama_index.core.agent.workflow import ReActAgent
from llama_index.core.workflow import Context


agent = ReActAgent(
   instruments=[retrieve_evidence, score_answer],
   llm=Settings.llm,
   system_prompt="""
At all times retrieve proof first.
Produce a structured reply.
Consider the reply and revise as soon as if scores are low.
""",
   verbose=True
)


ctx = Context(agent)

async def run_brief(subject: str):
   q = f"Design a dependable RAG + tool-using agent workflow and methods to consider it. Subject: {subject}"
   handler = agent.run(q, ctx=ctx)
   async for ev in handler.stream_events():
       print(getattr(ev, "delta", ""), finish="")
   res = await handler
   return str(res)


subject = "RAG agent reliability and analysis"
loop = asyncio.get_event_loop()
outcome = loop.run_until_complete(run_brief(subject))


print("nnFINAL OUTPUTn")
print(outcome)