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

AG-UI

AG-UI 是一种基于事件的协议，用于将 agent 更新流式传输到用户界面。该协议标准化了 message、tool-call 和 state 事件，从而很容易把不同的 agent runtime 接入可视化前端。ragas.integrations.ag_ui 模块帮助你把这些事件流转换成 Ragas message 对象，并使用现代 @experiment decorator 模式对实时 AG-UI endpoints 运行实验。

本指南假设你已经有一个正在运行的 AG-UI 兼容 agent（例如用 Google ADK、PydanticAI 或 CrewAI 构建），并且熟悉在 Ragas 中创建数据集。

安装集成

AG-UI 辅助功能位于可选 extra 中。请与评测用 LLM 所需的依赖一起安装。在 Jupyter 或 IPython 中运行时，请包含 nest_asyncio，以便复用 notebook 的 event loop。

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()

构建实验数据集

Dataset 可以包含单轮或多轮样本。借助 AG-UI，你可以测试任意一种模式——带自由形式回答的单问题，或包含 tool calls 的较长对话。

单轮样本

当你只需要给最终答案文本打分时，使用带 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",
)

带工具期望的多轮样本

当你想给中间 agent 行为打分——例如是否正确调用工具并达成用户目标——请把对话列表作为 user_input。把期望的 tool calls 以 JSON 提供，并可选择为 goal accuracy 评测提供参考结果。

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 指标。要解锁现代 collections 组合（并混入自定义检查），为评测 prompts 构建一个 Instructor 兼容的 LLM，并为 embeddings 使用同步 OpenAI client。

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() 来调用你的 endpoint，并用 agent 的响应丰富每一行。把它与 @experiment decorator 结合，即可构建评测流水线。

⚠️ 该 endpoint 必须暴露 AG-UI SSE 流。常见路径包括 /chat、/agent 或 /agentic_chat。

基本单轮评测

在 Jupyter 或 IPython 中，使用顶层 await（在 nest_asyncio.apply() 之后），而不是 asyncio.run，以避免 "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()

得到的 dataframe 包含每个样本的分数、原始 agent 响应，以及任何检索到的上下文（工具结果）。结果由 experiment 框架自动保存，你也可以通过 pandas 导出为 CSV。

多轮工具评测

对于多轮数据集和工具评测，把 messages 和参考 tool calls 直接传给指标：

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()

如果某个请求失败，实验会记录错误，并为该样本返回占位值，以便实验继续处理剩余样本。

直接处理 AG-UI 事件

有时你可能想单独收集事件日志——例如来自一次录制的运行或预发环境——并离线评测。转换辅助函数暴露了与 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)

转换后的 messages 可用于构建自定义评测工作流，或直接传给指标打分函数。

提取辅助函数

该集成提供从 messages 中提取特定数据的辅助函数：

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"]

生产实验提示

  • 自定义 headers：通过 run_ag_ui_row() 的 extra_headers 参数传递认证 token 或租户 ID。
  • 超时：如果你的 agent 会执行长时间运行的 tool calls，请调整 timeout 参数。
  • 元数据调试：设置 metadata=True，在每条 message 上保留 AG-UI 的 run、thread 和 message ID，便于追溯。
  • 实验命名：为 .arun() 使用描述性的 name 参数，便于识别结果。

完整生产示例见 examples/ragas_examples/ag_ui_agent_experiments/experiments.py，它提供：

  • 用于 endpoint 配置的 CLI 参数
  • 基于 CSV 的测试数据集
  • 妥善的日志和错误处理
  • 带时间戳的结果输出

交互式 walkthrough notebook 也位于 howtos/integrations/ag_ui.ipynb。

API 参考

主要 API

  • run_ag_ui_row(row, endpoint_url, ...) - 针对 AG-UI endpoint 运行单行，并返回带有 response、messages、tool_calls 和 contexts 的丰富数据。

转换函数

  • convert_to_ragas_messages(events, metadata=False) - 将 AG-UI 事件序列转换为 Ragas messages
  • convert_messages_snapshot(snapshot, metadata=False) - 将 AG-UI message snapshots 转换为 Ragas messages
  • convert_messages_to_ag_ui(messages) - 将 Ragas messages 转换为 AG-UI 格式

提取辅助函数

  • extract_response(messages) - 提取拼接后的 AI 响应文本
  • extract_tool_calls(messages) - 从 AI messages 中提取所有 tool calls
  • extract_contexts(messages) - 从 messages 中提取工具结果/contexts

底层

  • call_ag_ui_endpoint(endpoint_url, user_input, ...) - 调用 AG-UI endpoint 并收集流式事件
  • AGUIEventCollector - 从流式事件中收集并重建 messages
← OCI Gen AIAG-UI（旧页） →