Ragas Ragas
stable · 中文译文
中文译文 · 原文：https://docs.ragas.io/en/stable/howtos/integrations/_ag_ui/ · 许可证 Apache-2.0

AG-UI 集成

Ragas 可以对通过 AG-UI 协议 流式发送事件的 agents 运行实验。本 notebook 展示如何构建实验数据集、配置指标，并使用现代 @experiment decorator 模式对 AG-UI endpoints 打分。

前置条件

  • 安装依赖：pip install "ragas[ag-ui]" python-dotenv nest_asyncio
  • 在本地启动一个 AG-UI 兼容的 agent（Google ADK、PydanticAI、CrewAI 等）
  • 创建包含评测用 LLM 凭证的 .env 文件（例如 OPENAI_API_KEY、GOOGLE_API_KEY 等）
  • 如果运行本 notebook，请调用 nest_asyncio.apply()（如下所示），以便就地 await 协程。
# !pip install "ragas[ag-ui]" python-dotenv nest_asyncio

导入与环境设置

加载环境变量，并导入本 walkthrough 全程使用的类。

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

构建多轮对话

对于工具使用和 goal accuracy 指标，请提供：

  • reference_tool_calls：作为 JSON 的期望 tool calls，供 ToolCallF1 使用
  • 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

对于单轮问答实验，我们使用：

  • FactualCorrectness：将响应中的事实与 reference 比较
  • AnswerRelevancy：衡量响应对问题的相关程度
  • DiscreteMetric：用于简洁性的自定义指标

对于多轮 agent 实验，我们使用：

  • ToolCallF1：比较实际与期望 tool calls 的基于规则的指标
  • AgentGoalAccuracyWithReference：评估 agent 是否达成用户目标的基于 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 endpoint 运行实验

设置你的 agent 暴露的 endpoint URL。run_ag_ui_row() 函数调用该 endpoint 并返回丰富后的行数据。把它与 @experiment decorator 结合，用于评测流水线。

准备好运行实验时再切换这些 flags。在 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() 函数。

这种方式让你可以：

  • 自定义事件处理
  • 添加按行的 endpoint 配置
  • 实现自定义 message 处理
  • 添加额外日志或调试
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 decorator 提供 .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 层级 函数 何时使用
高层 run_ag_ui_row() 标准实验——处理 endpoint 调用、转换和提取
底层 call_ag_ui_endpoint() + convert_to_ragas_messages() 自定义事件处理、按行 endpoint 配置、高级调试

两种方式都可与 @experiment decorator 配合——根据你需要多少控制来选择。

← AG-UIGriptape →