AG-UI 통합

AG-UI 통합

Ragas는 AG-UI 프로토콜 을 통해 이벤트를 스트리밍하는 에이전트에서 실험(experiment)을 실행할 수 있어요. 이 노트북은 최신 @experiment 데코레이터 패턴을 사용해 실험 데이터셋을 만들고, 메트릭을 구성하고, AG-UI 엔드포인트에 점수를 매기는 방법을 보여줘요.

출처: 문서

본문

사전 요구사항

  • 의존성 설치: pip install "ragas[ag-ui]" python-dotenv nest_asyncio
  • AG-UI 호환 에이전트를 로컬에서 실행(Google ADK, PydanticAI, CrewAI 등)
  • 평가자 LLM 자격증명이 든 .env 파일 생성(예: OPENAI_API_KEY, GOOGLE_API_KEY 등)
  • 이 노트북을 실행한다면 nest_asyncio.apply() 를 호출해(아래 참고) 코루틴을 제자리에서 await 할 수 있게 하세요.
# !pip install "ragas[ag-ui]" python-dotenv nest_asyncio

Import와 환경 설정

환경 변수를 로드하고 워크스루에서 쓰는 클래스들을 import 해요.

import json

import nest_asyncio
import pandas as pd
from dotenv import load_dotenv
from IPython.display import display

from ragas.dataset import Dataset
from ragas.messages import HumanMessage

load_dotenv()
# Patch the existing notebook loop so we can await coroutines safely
nest_asyncio.apply()

단일 턴 실험 데이터 만들기

최종 답변 텍스트만 채점하면 될 때 Dataset.from_pandas() 로 user_input 과 reference 를 담은 데이터셋 엔트리를 만들 수 있어요.

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",
)

scientist_questions

멀티 턴 대화 만들기

도구 사용·목표 정확도 메트릭을 위해서는 다음을 제공하세요.

  • reference_tool_calls : ToolCallF1 에 필요한 기대 도구 호출(JSON)
  • reference : AgentGoalAccuracyWithReference 에 필요한 기대 결과 설명
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 - phrased to match what LLM extracts as end_state
                "reference": "The AI provided 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 AI provided the current weather conditions for London.",
            },
        ]
    ),
    name="weather_queries",
    backend="inmemory",
)

weather_queries

메트릭과 평가자 LLM 구성

단일 턴 Q&A 실험에서는 다음을 사용해요.

  • FactualCorrectness : 응답 사실을 reference와 비교
  • AnswerRelevancy : 응답이 질문에 얼마나 관련 있는지 측정
  • DiscreteMetric : 간결성(conciseness)을 위한 커스텀 메트릭

멀티 턴 에이전트 실험에서는 다음을 사용해요.

  • ToolCallF1 : 실제 vs 기대 도구 호출을 비교하는 규칙 기반 메트릭
  • AgentGoalAccuracyWithReference : 에이전트가 사용자 목표를 달성했는지 평가하는 LLM 기반 메트릭
from openai import AsyncOpenAI

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

# Async client for evaluator prompts
async_llm_client = AsyncOpenAI()
evaluator_llm = llm_factory("gpt-4o-mini", client=async_llm_client)

embedding_client = AsyncOpenAI()
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 experiments
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 experiments
# - ToolCallF1: Rule-based metric for tool call accuracy
# - AgentGoalAccuracyWithReference: LLM-based metric for goal achievement
tool_metrics = [
    ToolCallF1(),
    AgentGoalAccuracyWithReference(llm=evaluator_llm),
]

실시간 AG-UI 엔드포인트에 대해 실험 실행

에이전트가 노출한 엔드포인트 URL을 설정하세요. run_ag_ui_row() 함수는 엔드포인트를 호출하고 강화된(enriched) 행 데이터를 반환해요. 이를 평가 파이프라인을 위한 @experiment 데코레이터와 결합하세요.

실험을 실행할 준비가 되면 플래그를 토글하세요. Jupyter/IPython에서 nest_asyncio.apply() 를 호출했다면 실험을 직접 await 할 수 있어요.

AG_UI_ENDPOINT = "http://localhost:8000"  # Update to match your agent

RUN_FACTUAL_EXPERIMENT = True
RUN_TOOL_EXPERIMENT = True

from ragas import experiment
from ragas.integrations.ag_ui import run_ag_ui_row


@experiment()
async def factual_experiment(row):
    """Single-turn Q&A experiment with factual correctness scoring."""
    # Call AG-UI endpoint and get enriched row
    enriched = await run_ag_ui_row(row, AG_UI_ENDPOINT, metadata=True)

    # Score with factual correctness metric
    fc_result = await qa_metrics[0].ascore(
        response=enriched["response"],
        reference=row["reference"],
    )

    # Score with answer relevancy metric
    ar_result = await qa_metrics[1].ascore(
        user_input=row["user_input"],
        response=enriched["response"],
    )

    # Score with conciseness metric
    concise_result = await conciseness_metric.ascore(
        response=enriched["response"],
        llm=evaluator_llm,
    )

    return {
        **enriched,
        "factual_correctness": fc_result.value,
        "answer_relevancy": ar_result.value,
        "conciseness": concise_result.value,
    }


if RUN_FACTUAL_EXPERIMENT:
    # Run the experiment against the dataset
    factual_result = await factual_experiment.arun(
        scientist_questions, name="scientist_qa_experiment"
    )
    display(factual_result.to_pandas())

from ragas.messages import ToolCall


@experiment()
async def tool_experiment(row):
    """Multi-turn experiment with tool call and goal accuracy scoring."""
    # Call AG-UI endpoint and get enriched row
    enriched = await run_ag_ui_row(row, AG_UI_ENDPOINT)

    # 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 tool_metrics[0].ascore(
        user_input=enriched["messages"],
        reference_tool_calls=ref_tool_calls,
    )
    goal_result = await tool_metrics[1].ascore(
        user_input=enriched["messages"],
        reference=row.get("reference", ""),
    )

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


if RUN_TOOL_EXPERIMENT:
    # Run the experiment against the dataset
    tool_result = await tool_experiment.arun(
        weather_queries, name="weather_tool_experiment"
    )
    display(tool_result.to_pandas())

고급: 더 낮은 레벨의 제어

run_ag_ui_row() 가 권장 API지만 더 많은 제어가 필요할 때가 있어요. 하위 레벨의 call_ag_ui_endpoint() 함수를 직접 사용할 수 있어요.

이 접근법을 사용하면 다음을 할 수 있어요.

  • 이벤트 처리를 커스터마이즈
  • 행별 엔드포인트 구성을 추가
  • 커스텀 메시지 처리를 구현
  • 추가 로깅·디버깅을 더함
from ragas.integrations.ag_ui import (
    call_ag_ui_endpoint,
    convert_to_ragas_messages,
    extract_response,
)


@experiment()
async def custom_ag_ui_experiment(row):
    """
    Custom experiment function with full control over endpoint calls.
    """
    # Call the AG-UI endpoint directly (lower-level than run_ag_ui_row)
    events = await call_ag_ui_endpoint(
        endpoint_url=AG_UI_ENDPOINT,
        user_input=row["user_input"],
        timeout=60.0,
    )

    # Convert AG-UI events to Ragas messages
    messages = convert_to_ragas_messages(events, metadata=True)

    # Extract response using helper (or custom logic)
    response = extract_response(messages)

    # Score with a custom metric
    score_result = await conciseness_metric.ascore(
        response=response,
        llm=evaluator_llm,
    )

    # Return result with custom fields
    return {
        **row,
        "response": response or "[No response]",
        "message_count": len(messages),
        "conciseness": score_result.value,
    }

커스텀 실험을 데이터셋에 대해 실행하세요. @experiment 데코레이터는 병렬 실행과 자동 결과 수집을 위한 .arun() 을 제공해요.

RUN_CUSTOM_EXPERIMENT = True

if RUN_CUSTOM_EXPERIMENT:
    # Run the custom experiment
    custom_result = await custom_ag_ui_experiment.arun(
        scientist_questions, name="custom_ag_ui_experiment"
    )
    display(custom_result.to_pandas())

API 비교

API 레벨 함수 사용 시점
High-level run_ag_ui_row() 표준 실험 - 엔드포인트 호출·변환·추출을 처리
Low-level call_ag_ui_endpoint() + convert_to_ragas_messages() 커스텀 이벤트 처리, 행별 엔드포인트 구성, 고급 디버깅

두 접근법 모두 @experiment 데코레이터와 함께 동작해요. 필요한 제어 수준에 따라 선택하면 돼요.

더 알아보기 (Learn more)