Arxiv 논문 검색 LLM 에이전트 성능 평가하기
Arxiv 논문 검색 LLM 에이전트 성능 평가하기 (Phoenix Observability)
사용자 질의와 관련된 Arxiv 연구 논문을 찾고 요약하는 LLM 에이전트를 만들고, LlamaIndex에 내장된 Arize Phoenix 관찰 레이어로 그 성능을 평가하는 튜토리얼이에요. 에이전트의 RAG 기술과 함수 호출 정확도를 판사(LLM-as-a-Judge) 방식으로 측정합니다.
출처: 문서
본문
이 튜토리얼에서는 MistralAI 언어 모델 기반 LLM 에이전트를 만들고, 그 용도를 Arxiv에서 사용자 질의와 관련된 연구 논문을 찾아 요약하는 것으로 삼습니다. 에이전트 구축에는 LlamaIndex 프레임워크를 사용합니다.
에이전트가 사용하는 도구는 다음과 같아요.
- RAG Query Engine: 최근 Arxiv 논문을 저장·검색해 지식 베이스로 활용합니다.
- Paper Fetch Tool: RAG 쿼리 엔진에 없는 주제를 사용자가 지정하면 해당 주제의 최근 논문을 Arxiv에서 직접 가져옵니다.
- PDF Download Tool: Arxiv가 제공하는 링크로 연구 논문의 PDF를 로컬에 다운로드하게 해 줍니다.
이 노트북은 Andrei Chernov가 만들었습니다. (Github, Linkedin)
설치와 설정
Mistral과 Phoenix API 키가 필요합니다.
!pip install arxiv==2.1.3 llama_index==0.12.3 llama-index-llms-mistralai==0.3.0 llama-index-embeddings-mistralai==0.3.0
!pip install arize-phoenix==7.2.0 arize-phoenix-evals==0.18.0 openinference-instrumentation-llama-index==3.0.2
from getpass import getpass
import requests
import sys
import arxiv
from llama_index.llms.mistralai import MistralAI
from llama_index.embeddings.mistralai import MistralAIEmbedding
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Document, StorageContext, load_index_from_storage, PromptTemplate, Settings
from llama_index.core.tools import FunctionTool, QueryEngineTool
from llama_index.core.agent import ReActAgent
api_key= getpass("Type your API Key")
llm = MistralAI(api_key=api_key, model='mistral-large-latest')
이 튜토리얼에서는 MistralAI 임베딩 모델을 사용합니다.
model_name = "mistral-embed"
embed_model = MistralAIEmbedding(model_name=model_name, api_key=api_key)
Arxiv에서 언어 모델 관련 논문 내려받기
이 튜토리얼을 무료 Mistral API 버전에서 접근 가능하게 유지하기 위해 최근 논문 10개만 내려받습니다. 더 많이 내려받으면 나중에 RAG 쿼리 엔진을 만들 때 한도를 초과할 수 있어요.
def fetch_arxiv_papers(title :str, papers_count: int):
search_query = f'all:"{title}"'
search = arxiv.Search(
query=search_query,
max_results=papers_count,
sort_by=arxiv.SortCriterion.SubmittedDate,
sort_order=arxiv.SortOrder.Descending
)
papers = []
# Use the Client for searching
client = arxiv.Client()
# Execute the search
search = client.results(search)
for result in search:
paper_info = {
'title': result.title,
'authors': [author.name for author in result.authors],
'summary': result.summary,
'published': result.published,
'journal_ref': result.journal_ref,
'doi': result.doi,
'primary_category': result.primary_category,
'categories': result.categories,
'pdf_url': result.pdf_url,
'arxiv_url': result.entry_id
}
papers.append(paper_info)
return papers
papers = fetch_arxiv_papers("Language Models", 10)
[[p['title']] for p in papers]
이 과정은 임베딩 모델로 문서의 각 청크에 대한 벡터 표현을 만듭니다. Arxiv 포맷을 LlamaIndex가 이해하는 문서로 변환합니다.
def create_documents_from_papers(papers):
documents = []
for paper in papers:
content = f"Title: {paper['title']}\n" \
f"Authors: {', '.join(paper['authors'])}\n" \
f"Summary: {paper['summary']}\n" \
f"Published: {paper['published']}\n" \
f"Journal Reference: {paper['journal_ref']}\n" \
f"DOI: {paper['doi']}\n" \
f"Primary Category: {paper['primary_category']}\n" \
f"Categories: {', '.join(paper['categories'])}\n" \
f"PDF URL: {paper['pdf_url']}\n" \
f"arXiv URL: {paper['arxiv_url']}\n"
documents.append(Document(text=content))
return documents
#Create documents for LlamaIndex
documents = create_documents_from_papers(papers)
Settings.chunk_size = 1024
Settings.chunk_overlap = 50
index = VectorStoreIndex.from_documents(documents, embed_model=embed_model)
인덱스 저장하기
많은 텍스트를 인덱싱하는 것은 임베딩 모델에 API 호출을 하므로 시간과 비용이 들 수 있어요. 실제 애플리케이션에서는 재인덱싱을 피하려고 벡터 데이터베이스에 인덱스를 저장하는 게 좋지만, 이 튜토리얼에서는 단순하게 인덱스를 로컬 디렉터리에 저장합니다.
index.storage_context.persist('index/')
# rebuild storage context
storage_context = StorageContext.from_defaults(persist_dir='index/')
#load index
index = load_index_from_storage(storage_context, embed_model=embed_model)
각 도구에 의미 있는 이름과 명확한 설명을 제공하는 것이 좋은 습관이에요. 에이전트가 필요할 때 가장 적절한 도구를 고르는 데 도움이 됩니다.
query_engine = index.as_query_engine(llm=llm, similarity_top_k=5)
rag_tool = QueryEngineTool.from_defaults(
query_engine,
name="research_paper_query_engine_tool",
description="A RAG engine with recent research papers.",
)
RAG 도구가 문맥을 기반으로 질의에 답할 때 쓰는 프롬프트를 살펴봅니다. 기본적으로 LlamaIndex는 답을 반환하기 전에 refine 프롬프트를 사용합니다.
from llama_index.core import PromptTemplate
from IPython.display import Markdown, display
# define prompt viewing function
def display_prompt_dict(prompts_dict):
for k, p in prompts_dict.items():
text_md = f"**Prompt Key**: {k}" f"**Text:** "
display(Markdown(text_md))
print(p.get_template())
display(Markdown(""))
prompts_dict = query_engine.get_prompts()
display_prompt_dict(prompts_dict)
나머지 두 도구는 단순히 Python 함수이므로 만들기 쉽습니다.
def download_pdf(pdf_url, output_file):
"""
Downloads a PDF file from the given URL and saves it to the specified file.
Args:
pdf_url (str): The URL of the PDF file to download.
output_file (str): The path and name of the file to save the PDF to.
Returns:
str: A message indicating success or the nature of an error.
"""
try:
# Send a GET request to the PDF URL
response = requests.get(pdf_url)
response.raise_for_status() # Raise an error for HTTP issues
# Write the content of the PDF to the output file
with open(output_file, "wb") as file:
file.write(response.content)
return f"PDF downloaded successfully and saved as '{output_file}'."
except requests.exceptions.RequestException as e:
return f"An error occurred: {e}"
download_pdf_tool = FunctionTool.from_defaults(
download_pdf,
name='download_pdf_file_tool',
description='python function, which downloads a pdf file by link'
)
fetch_arxiv_tool = FunctionTool.from_defaults(
fetch_arxiv_papers,
name='fetch_from_arxiv',
description='download the {max_results} recent papers regarding the topic {title} from arxiv'
)
에이전트와 대화하기
세 도구로 ReAct 에이전트를 만듭니다.
# building an ReAct Agent with the three tools.
agent = ReActAgent.from_tools([download_pdf_tool, rag_tool, fetch_arxiv_tool], llm=llm, verbose=True)
ReAct 에이전트는 두 가지 주요 단계로 동작해요.
- 추론(Reasoning): 쿼리를 받으면 에이전트는 직접 답할 충분한 정보가 있는지, 아니면 도구를 써야 하는지 평가합니다.
- 행동(Acting): 도구를 쓰기로 결정하면 도구를 실행하고, 다시 추론 단계로 돌아가 이제 쿼리에 답할 수 있는지 또는 추가 도구 사용이 필요한지 판단합니다.
# create a prompt template to chat with an agent
q_template = (
"I am interested in {topic}. \n"
"Find papers in your knowledge database related to this topic; use the following template to query research_paper_query_engine_tool tool: 'Provide title, summary, authors and link to download for papers related to {topic}'. If there are not, could you fetch the recent one from arXiv? \n"
)
answer = agent.chat(q_template.format(topic="Audio Models"))
Markdown(answer.response)
answer = agent.chat("Download the papers, which you mentioned above")
answer = agent.chat(q_template.format(topic="Gaussian process"))
에이전트 실행의 더 자세한 내용은 Phoenix 대시보드에서 확인할 수 있어요. LlamaIndex에는 Arize Phoenix가 제공하는 내장 관찰 레이어가 있습니다. 이를 사용해 에이전트의 실행을 추적하고 성능을 평가할 수 있죠.
Phoenix로 에이전트 추적하기
Phoenix API 키가 없다면 여기에서 얻을 수 있어요.
from phoenix.otel import register
from openinference.instrumentation.llama_index import LlamaIndexInstrumentor
import os
PHOENIX_API_KEY = getpass("Type your Phoenix API Key")
os.environ["PHOENIX_CLIENT_HEADERS"] = f"api_key={PHOENIX_API_KEY}"
os.environ["PHOENIX_COLLECTOR_ENDPOINT"] = "https://app.phoenix.arize.com"
tracer = register(project_name="arxiv-agentic-rag")
LlamaIndexInstrumentor().instrument(tracer_provider=tracer)
이제 LlamaIndex에 대한 모든 호출이 추적되어 Phoenix 인스턴스에 기록됩니다. 방금 추적을 켰으므로 트레이스 데이터를 보려면 에이전트를 다시 실행해야 해요. 보통은 에이전트의 전체 실행을 캡처하려면 노트북 앞부분에서 추적을 켜는 게 좋습니다.
answer = agent.chat(q_template.format(topic="Audio Models"))
answer = agent.chat("Download the papers, which you mentioned above")
answer = agent.chat(q_template.format(topic="Gaussian process"))
이제 Phoenix 인스턴스로 가면 에이전트 실행의 트레이스 데이터를 볼 수 있어요. 에이전트 실행의 처음 몇 번 반복을 수동으로 확인하는 건 쉽지만, 모든 반복을 그렇게 하는 건 비현실적입니다. 에이전트 성능을 평가하는 더 확장 가능한 방법을 추가해 봅니다.
에이전트 성능을 평가하는 방법은 무한히 많습니다. 두 가지 일반적인 방법을 살펴볼게요.
- 에이전트의 RAG 기술 평가
- 에이전트의 함수 호출 정확도 평가
두 평가 모두 LLM-as-a-Judge를 사용하며, Mistral이 판사입니다.
from phoenix.session.evaluation import get_retrieved_documents, get_qa_with_reference
from phoenix.trace import SpanEvaluations, DocumentEvaluations
import phoenix as px
from phoenix.evals import (
MistralAIModel,
RelevanceEvaluator,
HallucinationEvaluator,
QAEvaluator,
run_evals,
)
import nest_asyncio
nest_asyncio.apply()
eval_model = MistralAIModel(api_key=api_key)
에이전트의 RAG 기술 평가하기
# First retrieve documents from Phoenix
retrieved_documents_df = get_retrieved_documents(px.Client(), project_name="arxiv-agentic-rag")
retrieved_documents_df.head()
# Use Phoenix's RelevanceEvaluator to evaluate the relevance of the retrieved documents
relevance_evaluator = RelevanceEvaluator(eval_model)
retrieved_documents_relevance_df = run_evals(
evaluators=[relevance_evaluator],
dataframe=retrieved_documents_df,
provide_explanation=True,
concurrency=5,
)[0]
# Retrieve Question and Answer pairs with reference answers
qa_with_reference_df = get_qa_with_reference(px.Client(), project_name="arxiv-agentic-rag")
# Evaluate the correctness of the Q&A pairs
qa_evaluator = QAEvaluator(eval_model)
# Evaluate the hallucination of the Q&A pairs
hallucination_evaluator = HallucinationEvaluator(eval_model)
# Run evaluations for Q&A correctness and hallucination
qa_correctness_eval_df, hallucination_eval_df = run_evals(
evaluators=[qa_evaluator, hallucination_evaluator],
dataframe=qa_with_reference_df,
provide_explanation=True,
concurrency=5,
)
RAG 기술에 대해 계산된 이 세 메트릭을 Phoenix에 기록해 트레이스 데이터와 함께 볼 수 있습니다.
px.Client().log_evaluations(
SpanEvaluations(dataframe=qa_correctness_eval_df, eval_name="Q&A Correctness"),
SpanEvaluations(dataframe=hallucination_eval_df, eval_name="Hallucination"),
DocumentEvaluations(dataframe=retrieved_documents_relevance_df, eval_name="relevance"),
)
에이전트의 함수 호출 정확도 평가하기
이제 에이전트의 함수 호출 정확도, 즉 질의에 답할 때 에이전트가 올바른 도구를 얼마나 자주 사용하는지 평가합니다.
from phoenix.trace.dsl import SpanQuery
from phoenix.evals import (
llm_classify,
TOOL_CALLING_PROMPT_RAILS_MAP,
TOOL_CALLING_PROMPT_TEMPLATE,
)
이전 섹션과 마찬가지로 관련 트레이스 데이터를 검색하는 것부터 시작합니다. 이전에는 Phoenix SDK의 헬퍼 메서드를 사용했지만, 여기서는 더 일반적이며 설정한 필터에 따라 트레이스 데이터를 검색하는 SpanQuery DSL을 사용할게요.
query = (
SpanQuery()
.where(
# Filter for the `LLM` span kind.
# The filter condition is a string of valid Python boolean expression.
"span_kind == 'LLM'",
)
.select(
# Extract and rename the following span attributes
question="llm.input_messages",
tool_call="llm.function_call",
)
)
trace_df = px.Client().query_spans(query, project_name="arxiv-agentic-rag")
trace_df["tool_call"] = trace_df["tool_call"].fillna("No tool used")
trace_df["question"] = trace_df["question"].fillna("No question")
평가자가 에이전트가 사용할 수 있는 가능한 도구를 알 수 있도록 도구 정의도 전달해야 합니다.
tool_definitions = ""
for current_tool in [download_pdf_tool, rag_tool, fetch_arxiv_tool]:
tool_definitions += f"""
{current_tool.metadata.name}: {current_tool.metadata.description}
"""
tool_definitions = tool_definitions.replace("{", "").replace("}", "")
trace_df["tool_definitions"] = tool_definitions
print(tool_definitions)
이제 평가를 실행할 준비가 됐습니다. llm_classify 메서드로 도구 호출을 올바른지/올바르지 않은지 분류합니다.
rails = list(TOOL_CALLING_PROMPT_RAILS_MAP.values())
template = TOOL_CALLING_PROMPT_TEMPLATE.explanation_template[0].template.replace("{tool_definitions}", tool_definitions)
function_calling_evals = llm_classify(
dataframe=trace_df,
template=TOOL_CALLING_PROMPT_TEMPLATE,
model=eval_model,
rails=rails,
concurrency=5,
provide_explanation=True,
)
function_calling_evals["score"] = function_calling_evals.apply(
lambda x: 1 if x["label"] == "correct" else 0, axis=1
)
마지막으로 평가를 Phoenix에 기록해 트레이스 데이터와 함께 볼 수 있습니다.
px.Client().log_evaluations(
SpanEvaluations(dataframe=function_calling_evals, eval_name="Function Calling Accuracy"),
)
축하합니다! 이제 LlamaIndex로 LLM 에이전트를 만들었고, Phoenix를 사용해 그 성능을 평가했습니다.
더 알아보기 (Learn more)
- Arize Phoenix 문서 — LLM 트레이스 관찰·평가 플랫폼
LlamaIndexInstrumentor— LlamaIndex 계측기RelevanceEvaluator/QAEvaluator/HallucinationEvaluator— Phoenix LLM Evals 평가기SpanQuery/llm_classify— 트레이스 검색·도구 호출 분류 DSL- 모델:
mistral-large-latest(LLM),mistral-embed(임베딩),MistralAIModel(판사)