AG-UI

AG-UI

AG-UI는 에이전트 업데이트를 사용자 인터페이스에 스트리밍하기 위한 이벤트 기반 프로토콜이에요. 이 프로토콜은 메시지, 도구 호출, state 이벤트를 표준화해서 서로 다른 에이전트 런타임을 시각적 프론트엔드에 쉽게 연결하게 해 줘요. ragas.integrations.ag_ui 모듈은 그 이벤트 스트림을 Ragas 메시지 객체로 변환하고, 최신 @experiment 데코레이터 패턴을 사용해 실시간 AG-UI 엔드포인트에 대해 실험을 실행할 수 있게 도와줘요.

출처: 문서

본문

이 가이드는 AG-UI 호환 에이전트(예: Google ADK, PydanticAI, CrewAI로 만든 것)가 이미 실행 중이고, Ragas에서 데이터셋을 만드는 것에 익숙하다고 가정해요.

통합 설치

AG-UI 헬퍼는 선택적(optional) extra 뒤에 있어요. 평가자 LLM에 필요한 의존성과 함께 설치하세요. Jupyter나 IPython에서 실행할 때는 notebook의 이벤트 루프를 재사용할 수 있도록 nest_asyncio 를 포함하세요.

pip install "ragas[ag-ui]" python-dotenv nest_asyncio

평가자 LLM 자격증명을 구성하세요. 예를 들어 OpenAI 모델을 사용한다면:

# .env
OPENAI_API_KEY=sk-...

Python 안에서 예시를 실행하기 전에 환경 변수를 로드하세요.

from dotenv import load_dotenv
import nest_asyncio

load_dotenv()

# If you're inside Jupyter/IPython, patch the running event loop once.
nest_asyncio.apply()

실험 데이터셋 만들기

데이터셋은 단일 턴 또는 멀티 턴 샘플을 담을 수 있어요. AG-UI로 두 패턴 모두 테스트할 수 있어요—자유 형식 응답이 있는 단일 질문이나, 도구 호출을 포함한 더 긴 대화요.

단일 턴 샘플

최종 답변 텍스트만 채점하면 될 때 user_input 과 reference 컬럼으로 Dataset.from_pandas() 를 사용하세요.

import pandas as pd
from ragas.dataset import Dataset

scientist_questions = Dataset.from_pandas(
    pd.DataFrame([
        {
            "user_input": "Who originated the theory of relativity?",
            "reference": "Albert Einstein originated the theory of relativity.",
        },
        {
            "user_input": "Who discovered penicillin and when?",
            "reference": "Alexander Fleming discovered penicillin in 1928.",
        },
    ]),
    name="scientist_questions",
    backend="inmemory",
)

도구 기대가 있는 멀티 턴 샘플

중간 에이전트 동작(도구를 올바르게 호출하는지, 사용자 목표를 달성하는지 같은)을 채점하고 싶다면 user_input 으로 대화 목록을 사용하세요. 기대 도구 호출을 JSON으로, 그리고 선택적으로 목표 정확도 평가를 위한 reference 결과를 제공하세요.

import json
import pandas as pd
from ragas.dataset import Dataset
from ragas.messages import HumanMessage

weather_queries = Dataset.from_pandas(
    pd.DataFrame([
        {
            "user_input": [HumanMessage(content="What's the weather in Paris?")],
            "reference_tool_calls": json.dumps([
                {"name": "get_weather", "args": {"location": "Paris"}}
            ]),
            # Expected outcome for AgentGoalAccuracyWithReference
            "reference": "The user received the current weather conditions for Paris.",
        },
        {
            "user_input": [HumanMessage(content="Is it raining in London right now?")],
            "reference_tool_calls": json.dumps([
                {"name": "get_weather", "args": {"location": "London"}}
            ]),
            "reference": "The user received the current weather conditions for London.",
        },
    ]),
    name="weather_queries",
    backend="inmemory",
)

CSV에서 로드

더 큰 데이터셋을 위해 테스트 케이스를 CSV 파일에 저장해 두고 Dataset API로 로드하세요.

from ragas.dataset import Dataset

dataset = Dataset.load(
    name="scientist_biographies",
    backend="local/csv",
    root_dir="./test_data",
)

메트릭과 평가자 모델 선택

이 통합은 모든 Ragas 메트릭과 함께 동작해요. 최신 컬렉션(collection) 포트폴리오를 사용하고(커스텀 체크를 섞으려면) 평가자 프롬프트에 Instructor 호환 LLM을 만들고, 임베딩에는 동기식(synchronous) OpenAI 클라이언트를 사용하세요.

from openai import AsyncOpenAI, OpenAI
from ragas.llms import llm_factory
from ragas.embeddings import embedding_factory
from ragas.metrics import DiscreteMetric
from ragas.metrics.collections import (
    AgentGoalAccuracyWithReference,
    AnswerRelevancy,
    FactualCorrectness,
    ToolCallF1,
)

async_llm_client = AsyncOpenAI()
evaluator_llm = llm_factory("gpt-4o-mini", client=async_llm_client)

# AnswerRelevancy's embeddings still run synchronously, so pair it with a sync client.
embedding_client = OpenAI()
evaluator_embeddings = embedding_factory(
    "openai", model="text-embedding-3-small", client=embedding_client, interface="modern"
)

conciseness_metric = DiscreteMetric(
    name="conciseness",
    allowed_values=["verbose", "concise"],
    prompt=(
        "Is the response concise and efficiently conveys information?\n\n"
        "Response: {response}\n\n"
        "Answer with only 'verbose' or 'concise'."
    ),
)

# Metrics for single-turn Q&A evaluation
qa_metrics = [
    FactualCorrectness(
        llm=evaluator_llm, mode="f1", atomicity="high", coverage="high"
    ),
    AnswerRelevancy(llm=evaluator_llm, embeddings=evaluator_embeddings, strictness=2),
    conciseness_metric,
]

# Metrics for multi-turn agent evaluation
# - ToolCallF1: Rule-based metric for tool call accuracy
# - AgentGoalAccuracyWithReference: LLM-based metric for goal achievement
tool_metrics = [
    ToolCallF1(),
    AgentGoalAccuracyWithReference(llm=evaluator_llm),
]

@experiment로 실험 실행

AG-UI 통합은 run_ag_ui_row() 를 제공해서 엔드포인트를 호출하고 각 행을 에이전트의 응답으로 강화해요. 이를 @experiment 데코레이터와 결합해 평가 파이프라인을 만들어요.

⚠️ 엔드포인트가 AG-UI SSE 스트림을 노출해야 해요. 흔한 경로는 /chat, /agent, /agentic_chat 입니다.

기본 단일 턴 평가

Jupyter나 IPython에서는 asyncio.run 대신 최상위 await(nest_asyncio.apply() 후)를 사용해 "event loop is already running" 오류를 피하세요. 스크립트에서는 asyncio.run 을 유지해도 됩니다.

from ragas import experiment
from ragas.integrations.ag_ui import run_ag_ui_row
from ragas.metrics.collections import FactualCorrectness

@experiment()
async def factual_experiment(row):
    # Call AG-UI endpoint and get enriched row
    enriched = await run_ag_ui_row(row, "http://localhost:8000/chat")

    # Score with metrics
    score = await FactualCorrectness(llm=evaluator_llm).ascore(
        response=enriched["response"],
        reference=row["reference"],
    )

    return {**enriched, "factual_correctness": score.value}

# Run the experiment against the dataset
# In Jupyter/IPython (after calling nest_asyncio.apply())
factual_result = await factual_experiment.arun(
    scientist_questions,
    name="scientist_qa_eval"
)

# In a standalone script, use:
# factual_result = asyncio.run(factual_experiment.arun(scientist_questions, name="scientist_qa_eval"))

factual_result.to_pandas()

결과 데이터프레임은 샘플별 점수, 원시 에이전트 응답, 검색된 컨텍스트(도구 결과)를 포함해요. 결과는 실험 프레임워크에 의해 자동 저장되고, pandas로 CSV 내보내기를 할 수 있어요.

멀티 턴 도구 평가

멀티 턴 데이터셋과 도구 평가를 위해 messages와 reference 도구 호출을 메트릭에 직접 전달하세요.

import json
from ragas import experiment
from ragas.integrations.ag_ui import run_ag_ui_row
from ragas.messages import ToolCall
from ragas.metrics.collections import AgentGoalAccuracyWithReference, ToolCallF1

@experiment()
async def tool_experiment(row):
    # Call AG-UI endpoint and get enriched row
    enriched = await run_ag_ui_row(row, "http://localhost:8000/chat")

    # Parse reference_tool_calls from JSON string (e.g., from CSV)
    ref_tool_calls_raw = row.get("reference_tool_calls")
    if isinstance(ref_tool_calls_raw, str):
        ref_tool_calls = [ToolCall(**tc) for tc in json.loads(ref_tool_calls_raw)]
    else:
        ref_tool_calls = ref_tool_calls_raw or []

    # Score with tool metrics using the modern collections API
    f1_result = await ToolCallF1().ascore(
        user_input=enriched["messages"],
        reference_tool_calls=ref_tool_calls,
    )
    goal_result = await AgentGoalAccuracyWithReference(llm=evaluator_llm).ascore(
        user_input=enriched["messages"],
        reference=row.get("reference", ""),
    )

    return {
        **enriched,
        "tool_call_f1": f1_result.value,
        "agent_goal_accuracy": goal_result.value,
    }

# Run the experiment
# In Jupyter/IPython
tool_result = await tool_experiment.arun(
    weather_queries,
    name="weather_tool_eval"
)

# Or in a script
# tool_result = asyncio.run(tool_experiment.arun(weather_queries, name="weather_tool_eval"))

tool_result.to_pandas()

요청이 실패하면 실험은 오류를 로깅하고 해당 샘플에 placeholder 값을 반환해서 나머지 샘플로 실험을 계속해요.

AG-UI 이벤트로 직접 작업

때로는 이벤트 로그를 별도로 수집하고 싶을 수 있어요—기록된 run이나 스테이징 환경에서—오프라인으로 평가하고 싶을 때요. 변환 헬퍼는 run_ag_ui_row() 가 사용하는 것과 동일한 파싱 로직을 노출해요.

from ragas.integrations.ag_ui import convert_to_ragas_messages
from ag_ui.core import TextMessageChunkEvent

events = [
    TextMessageChunkEvent(
        message_id="assistant-1",
        role="assistant",
        delta="Hello from AG-UI!",
        timestamp="2024-12-01T00:00:00Z",
    )
]

ragas_messages = convert_to_ragas_messages(events, metadata=True)

이미 MessagesSnapshotEvent 가 있다면 스트리밍 재구성을 건너뛰고 convert_messages_snapshot 을 호출할 수 있어요.

from ragas.integrations.ag_ui import convert_messages_snapshot
from ag_ui.core import MessagesSnapshotEvent, UserMessage, AssistantMessage

snapshot = MessagesSnapshotEvent(
    messages=[
        UserMessage(id="msg-1", content="Hello?"),
        AssistantMessage(id="msg-2", content="Hi! How can I help you today?"),
    ]
)

ragas_messages = convert_messages_snapshot(snapshot)

변환된 메시지는 커스텀 평가 워크플로우를 만드는 데 사용하거나 메트릭 점수 함수에 직접 전달할 수 있어요.

추출 헬퍼

통합은 메시지에서 특정 데이터를 추출하는 헬퍼 함수를 제공해요.

from ragas.integrations.ag_ui import (
    extract_response,    # Get concatenated AI response text
    extract_tool_calls,  # Get all tool calls from AI messages
    extract_contexts,    # Get tool results/contexts
)

messages = convert_to_ragas_messages(events)

response = extract_response(messages)      # "Hello! The weather is sunny."
tool_calls = extract_tool_calls(messages)  # [ToolCall(name="get_weather", args={"location": "SF"})]
contexts = extract_contexts(messages)      # ["Sunny, 72F in San Francisco"]

프로덕션 실험을 위한 팁

  • 커스텀 헤더 : run_ag_ui_row() 의 extra_headers 파라미터로 인증 토큰이나 테넌트 ID를 전달하세요.
  • 타임아웃 : 에이전트가 오래 실행되는 도구 호출을 수행한다면 timeout 파라미터를 조정하세요.
  • 메타데이터 디버깅 : metadata=True 로 설정해 모든 메시지에 AG-UI run, thread, message ID를 유지하면 추적이 쉬워요.
  • 실험 네이밍 : 결과 식별이 쉽도록 .arun() 에 설명적인 name 인자를 사용하세요.

완전한 프로덕션 예시는 examples/ragas_examples/ag_ui_agent_experiments/experiments.py 를 보세요. 여기에는 다음이 제공돼요.

  • 엔드포인트 구성을 위한 CLI 인자
  • CSV 기반 테스트 데이터셋
  • 적절한 로깅과 오류 처리
  • 타임스탬프가 있는 결과 출력

인터랙티브 워크스루 노트북도 howtos/integrations/ag_ui.ipynb 에서 볼 수 있어요.

API 참조

기본 API

  • run_ag_ui_row(row, endpoint_url, ...) - 단일 행을 AG-UI 엔드포인트에 대해 실행하고 response, messages, tool_calls, contexts가 담긴 강화된 데이터를 반환해요.

변환 함수

  • convert_to_ragas_messages(events, metadata=False) - AG-UI 이벤트 시퀀스를 Ragas 메시지로 변환
  • convert_messages_snapshot(snapshot, metadata=False) - AG-UI 메시지 스냅샷을 Ragas 메시지로 변환
  • convert_messages_to_ag_ui(messages) - Ragas 메시지를 AG-UI 형식으로 변환

추출 헬퍼

  • extract_response(messages) - 연결된 AI 응답 텍스트 추출
  • extract_tool_calls(messages) - AI 메시지에서 모든 도구 호출 추출
  • extract_contexts(messages) - 메시지에서 도구 결과/컨텍스트 추출

저수준(Low-Level)

  • call_ag_ui_endpoint(endpoint_url, user_input, ...) - AG-UI 엔드포인트를 호출하고 스트리밍 이벤트를 수집
  • AGUIEventCollector - 스트리밍 이벤트에서 메시지를 수집·재구성

더 알아보기 (Learn more)