Ragas × AG-UI 集成实战用 experiment 装饰器评估流式 Agent 端点【免费下载链接】ragasSupercharge Your LLM Application Evaluations 项目地址: https://gitcode.com/gh_mirrors/ra/ragasRagas 可以通过 AG-UI 协议 为骨架讲解如何基于现代experiment装饰器模式构建实验数据集、配置指标并对运行中的 AG-UI 端点进行打分。读完本文你将掌握单轮问答与多轮工具调用两类 Agent 评估实验的完整搭建方法并理解底层事件收集与消息转换原理从而能够对 Google ADK、PydanticAI、CrewAI 等 AG-UI 兼容 Agent 建立自动化评测流水线。前置条件在开始之前需要准备以下环境安装依赖pip install ragas[ag-ui] python-dotenv nest_asyncioragas[ag-ui]额外安装ag-ui-protocol与httpx等依赖源码中通过延迟导入lazy import方式按需加载 AG-UI 类型若缺失会提示AG-UI integration requires the ag-ui-protocol package见 src/ragas/integrations/ag_ui.py。在本地启动一个 AG-UI 兼容 Agent如 Google ADK、PydanticAI、CrewAI 等并暴露 HTTP 端点。创建.env文件写入评估器 LLM 的凭据如OPENAI_API_KEY、GOOGLE_API_KEY等。如果在 Notebook 中运行需要调用nest_asyncio.apply()见下文以便在现有事件循环中安全地await协程。# !pip install ragas[ag-ui] python-dotenv nest_asyncio导入与环境初始化加载环境变量并导入本文后续使用的类。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()ragas.messages模块定义了HumanMessage、AIMessage、ToolCall、ToolMessage等统一消息类型它们是 AG-UI 事件转换后的目标格式也是后续多轮指标如ToolCallF1评分时的输入单位。构建单轮实验数据当你只需要对最终回答文本进行评分时可以使用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., }, ] ), namescientist_questions, backendinmemory, ) scientist_questions仓库示例中提供了一个更完整的单轮数据集 examples/ragas_examples/ag_ui_agent_experiments/test_data/datasets/scientist_biographies.csv包含 5 条关于科学家生平的问答样本如Who originated the theory of relativity and where were they born?并通过Dataset.load(..., backendlocal/csv, root_dir...)加载详见 experiments.py。从源码结构可以推断backendinmemory表示数据驻留在内存中而local/csv则从 CSV 文件按需读取。构建多轮对话数据对于需要评测工具调用与目标达成度的场景需要额外提供reference_tool_calls以 JSON 字符串形式给出的期望工具调用供ToolCallF1使用reference期望的结果描述供AgentGoalAccuracyWithReference使用。weather_queries Dataset.from_pandas( pd.DataFrame( [ { user_input: [HumanMessage(contentWhats 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(contentIs 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., }, ] ), nameweather_queries, backendinmemory, ) weather_queries注意此处user_input是HumanMessage列表多轮对话起始片段而不是普通字符串。仓库示例中的 weather_tool_calls.csv 用 CSV 存储了相同的结构user_input、reference_tool_callsJSON 字符串内部使用双引号转义与reference三列。配置指标与评估器 LLM单轮问答实验使用的指标FactualCorrectness将回答中的事实与参考文本比对AnswerRelevancy衡量回答与问题的相关性DiscreteMetric自定义离散指标用于评估回答的简洁性。多轮 Agent 实验使用的指标ToolCallF1基于规则的指标比较实际工具调用与期望工具调用AgentGoalAccuracyWithReference基于 LLM 的指标评估 Agent 是否达成了用户目标。这些指标统一导出自 src/ragas/metrics/collections/init.py。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, clientasync_llm_client) embedding_client AsyncOpenAI() evaluator_embeddings embedding_factory( openai, modeltext-embedding-3-small, clientembedding_client, interfacemodern, ) conciseness_metric DiscreteMetric( nameconciseness, 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 QA experiments qa_metrics [ FactualCorrectness( llmevaluator_llm, modef1, atomicityhigh, coveragehigh, ), AnswerRelevancy( llmevaluator_llm, embeddingsevaluator_embeddings, strictness2, ), 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(llmevaluator_llm), ]参数说明结合源码与仓库示例FactualCorrectness(modef1, atomicityhigh, coveragehigh)以 F1 方式计算事实一致性atomicity与coverage控制拆解与覆盖粒度AnswerRelevancy(strictness2)strictness控制相关性判定严格程度DiscreteMetric通过allowed_values约束输出取值prompt中的{response}是占位符评分时会替换为实际回答。对运行中的 AG-UI 端点运行实验设置 Agent 暴露的端点 URL。run_ag_ui_row()函数负责调用端点并返回富化后的行数据将其与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 QA experiment with factual correctness scoring. # Call AG-UI endpoint and get enriched row enriched await run_ag_ui_row(row, AG_UI_ENDPOINT, metadataTrue) # Score with factual correctness metric fc_result await qa_metrics[0].ascore( responseenriched[response], referencerow[reference], ) # Score with answer relevancy metric ar_result await qa_metrics[1].ascore( user_inputrow[user_input], responseenriched[response], ) # Score with conciseness metric concise_result await conciseness_metric.ascore( responseenriched[response], llmevaluator_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, namescientist_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_inputenriched[messages], reference_tool_callsref_tool_calls, ) goal_result await tool_metrics[1].ascore( user_inputenriched[messages], referencerow.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, nameweather_tool_experiment ) display(tool_result.to_pandas())experiment 装饰器与 run_ag_ui_row 的底层逻辑experiment()装饰器定义于 src/ragas/experiment.py它为函数提供.arun(dataset, name..., backend...)方法框架负责遍历数据集、以Experiment形式收集结果Experiment继承自DataTable支持to_pandas()导出并通过可选backend参数决定结果的存储后端未指定时沿用数据集的 backend。run_ag_ui_row()是推荐的一站式入口见 src/ragas/integrations/ag_ui.py其内部完成四步调用call_ag_ui_endpoint()向端点发起 POST 请求并收集 SSE 事件流通过convert_to_ragas_messages()将事件转换为 Ragas 消息用extract_response()、extract_tool_calls()、extract_contexts()分别提取回答文本、工具调用与上下文返回富化后的行{response, messages, tool_calls, contexts}被并入原行字段。需要注意的边界行为当行中缺少user_input或端点调用失败时函数不会抛错而是返回占位符response为[no response generated by agent]contexts为[no retrieved contexts provided by agent]便于流水线继续运行call_ag_ui_endpoint()使用httpx.AsyncClient以Accept: text/event-stream头请求 SSE 流逐行解析data: {...}前缀的 JSON 事件并用TypeAdapter(Event)按type判别字段反序列化为 AG-UI 事件对象默认超时 60 秒可通过timeout调整仓库示例脚本中设为 300 秒见 experiments.py。进阶底层精细控制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_urlAG_UI_ENDPOINT, user_inputrow[user_input], timeout60.0, ) # Convert AG-UI events to Ragas messages messages convert_to_ragas_messages(events, metadataTrue) # Extract response using helper (or custom logic) response extract_response(messages) # Score with a custom metric score_result await conciseness_metric.ascore( responseresponse, llmevaluator_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, namecustom_ag_ui_experiment ) display(custom_result.to_pandas())底层 API 一览API 层级函数使用时机高层run_ag_ui_row()标准实验——自动完成端点调用、消息转换与结果提取低层call_ag_ui_endpoint()convert_to_ragas_messages()自定义事件处理、按行配置端点、深度调试两种方式都可以与experiment装饰器配合使用按需选择控制粒度即可。深入理解AG-UI 事件流如何转换为 Ragas 消息要真正用好底层 API需要理解消息转换机制。convert_to_ragas_messages()见 src/ragas/integrations/ag_ui.py内部实例化AGUIEventCollector并逐事件调用process_event()。AGUIEventCollector维护三类内部状态_active_text_messages正在流式累积的文本消息Start → Content → End 三元组_active_tool_calls与_completed_tool_calls正在累积与已完成解析的工具调用_current_run_id、_current_thread_id、_current_step来自生命周期事件RUN_STARTED、STEP_STARTED等的上下文信息。其处理规则包括文本消息TEXT_MESSAGE_START初始化TEXT_MESSAGE_CONTENT累积delta分片TEXT_MESSAGE_END时拼接内容并依据role生成AIMessage或HumanMessage工具调用TOOL_CALL_START/TOOL_CALL_ARGS/TOOL_CALL_END重建ToolCallargs通过json.loads解析解析失败时回退为{raw_args: ...}TOOL_CALL_RESULT生成ToolMessage并确保其前驱AIMessage携带tool_calls必要时进行回溯附加以通过MultiTurnSample校验便捷分块事件TEXT_MESSAGE_CHUNK与TOOL_CALL_CHUNK单事件即包含完整消息绕过流式三元组重建快照事件MESSAGES_SNAPSHOT提供完整对话历史可通过convert_messages_snapshot()直接转换效率更高无关事件过滤生命周期、状态管理等非消息事件被静默忽略。metadataTrue时run_id、thread_id、step_name、message_id、timestamp等会保留在转换后消息的metadata字段中便于追踪与调试。上述行为均有对应的单元测试验证参见 tests/unit/integrations/test_ag_ui.py例如test_basic_text_message_conversion分片拼接、test_tool_call_conversion工具调用重建、test_step_context_in_metadata步骤上下文、test_call_ag_ui_endpointSSE 解析等。若未安装ag-ui-protocol相关测试会被pytestmark自动跳过。从 Notebook 到可执行脚本文档以 Notebook 形式给出实验流程仓库同时提供了对应的可执行脚本 examples/ragas_examples/ag_ui_agent_experiments/experiments.py将两个实验封装为命令行工具适合 CI 或定时评测python experiments.py --endpoint-url http://localhost:8000/chat python experiments.py --endpoint-url http://localhost:8000/chat --skip-tool-experiment python experiments.py --endpoint-url http://localhost:8000 --skip-factual脚本支持的参数--endpoint-urlAG-UI 端点地址默认http://localhost:8000--evaluator-model评估 LLM 模型名默认gpt-4o-mini--skip-factual跳过事实正确性实验--skip-tool-experiment跳过工具调用实验。脚本在实验前会先对 Embedding 端点做一次健全性检查并在日志中输出平均factual_correctness、tool_call_f1、agent_goal_accuracy以及完美得分样本占比等汇总统计方便快速判断 Agent 的当前质量水位。小结借助 Ragas 的 AG-UI 集成你可以在不侵入 Agent 实现的前提下通过统一的experiment装饰器模式对任何 AG-UI 兼容端点完成自动化评估单轮场景使用FactualCorrectness、AnswerRelevancy与自定义DiscreteMetric评估回答质量多轮场景使用ToolCallF1与AgentGoalAccuracyWithReference评估工具调用与目标达成度。高层 APIrun_ag_ui_row()封装了端点调用、事件转换与结果提取的完整链路而底层 API 则提供了完全可控的精细化操作空间——两者共享同一套消息转换与收集机制可在实际项目中按需选用。进一步阅读完整的 Notebook 版指南见 docs/howtos/integrations/_ag_ui.md集成实现见 src/ragas/integrations/ag_ui.pyexperiment装饰器源码见 src/ragas/experiment.py可运行的完整示例见 examples/ragas_examples/ag_ui_agent_experiments/experiments.py 及其 测试数据集。【免费下载链接】ragasSupercharge Your LLM Application Evaluations 项目地址: https://gitcode.com/gh_mirrors/ra/ragas创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Ragas × AG-UI 集成实战：用 @experiment 装饰器评估流式 Agent 端点
Ragas × AG-UI 集成实战用 experiment 装饰器评估流式 Agent 端点【免费下载链接】ragasSupercharge Your LLM Application Evaluations 项目地址: https://gitcode.com/gh_mirrors/ra/ragasRagas 可以通过 AG-UI 协议 为骨架讲解如何基于现代experiment装饰器模式构建实验数据集、配置指标并对运行中的 AG-UI 端点进行打分。读完本文你将掌握单轮问答与多轮工具调用两类 Agent 评估实验的完整搭建方法并理解底层事件收集与消息转换原理从而能够对 Google ADK、PydanticAI、CrewAI 等 AG-UI 兼容 Agent 建立自动化评测流水线。前置条件在开始之前需要准备以下环境安装依赖pip install ragas[ag-ui] python-dotenv nest_asyncioragas[ag-ui]额外安装ag-ui-protocol与httpx等依赖源码中通过延迟导入lazy import方式按需加载 AG-UI 类型若缺失会提示AG-UI integration requires the ag-ui-protocol package见 src/ragas/integrations/ag_ui.py。在本地启动一个 AG-UI 兼容 Agent如 Google ADK、PydanticAI、CrewAI 等并暴露 HTTP 端点。创建.env文件写入评估器 LLM 的凭据如OPENAI_API_KEY、GOOGLE_API_KEY等。如果在 Notebook 中运行需要调用nest_asyncio.apply()见下文以便在现有事件循环中安全地await协程。# !pip install ragas[ag-ui] python-dotenv nest_asyncio导入与环境初始化加载环境变量并导入本文后续使用的类。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()ragas.messages模块定义了HumanMessage、AIMessage、ToolCall、ToolMessage等统一消息类型它们是 AG-UI 事件转换后的目标格式也是后续多轮指标如ToolCallF1评分时的输入单位。构建单轮实验数据当你只需要对最终回答文本进行评分时可以使用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., }, ] ), namescientist_questions, backendinmemory, ) scientist_questions仓库示例中提供了一个更完整的单轮数据集 examples/ragas_examples/ag_ui_agent_experiments/test_data/datasets/scientist_biographies.csv包含 5 条关于科学家生平的问答样本如Who originated the theory of relativity and where were they born?并通过Dataset.load(..., backendlocal/csv, root_dir...)加载详见 experiments.py。从源码结构可以推断backendinmemory表示数据驻留在内存中而local/csv则从 CSV 文件按需读取。构建多轮对话数据对于需要评测工具调用与目标达成度的场景需要额外提供reference_tool_calls以 JSON 字符串形式给出的期望工具调用供ToolCallF1使用reference期望的结果描述供AgentGoalAccuracyWithReference使用。weather_queries Dataset.from_pandas( pd.DataFrame( [ { user_input: [HumanMessage(contentWhats 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(contentIs 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., }, ] ), nameweather_queries, backendinmemory, ) weather_queries注意此处user_input是HumanMessage列表多轮对话起始片段而不是普通字符串。仓库示例中的 weather_tool_calls.csv 用 CSV 存储了相同的结构user_input、reference_tool_callsJSON 字符串内部使用双引号转义与reference三列。配置指标与评估器 LLM单轮问答实验使用的指标FactualCorrectness将回答中的事实与参考文本比对AnswerRelevancy衡量回答与问题的相关性DiscreteMetric自定义离散指标用于评估回答的简洁性。多轮 Agent 实验使用的指标ToolCallF1基于规则的指标比较实际工具调用与期望工具调用AgentGoalAccuracyWithReference基于 LLM 的指标评估 Agent 是否达成了用户目标。这些指标统一导出自 src/ragas/metrics/collections/init.py。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, clientasync_llm_client) embedding_client AsyncOpenAI() evaluator_embeddings embedding_factory( openai, modeltext-embedding-3-small, clientembedding_client, interfacemodern, ) conciseness_metric DiscreteMetric( nameconciseness, 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 QA experiments qa_metrics [ FactualCorrectness( llmevaluator_llm, modef1, atomicityhigh, coveragehigh, ), AnswerRelevancy( llmevaluator_llm, embeddingsevaluator_embeddings, strictness2, ), 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(llmevaluator_llm), ]参数说明结合源码与仓库示例FactualCorrectness(modef1, atomicityhigh, coveragehigh)以 F1 方式计算事实一致性atomicity与coverage控制拆解与覆盖粒度AnswerRelevancy(strictness2)strictness控制相关性判定严格程度DiscreteMetric通过allowed_values约束输出取值prompt中的{response}是占位符评分时会替换为实际回答。对运行中的 AG-UI 端点运行实验设置 Agent 暴露的端点 URL。run_ag_ui_row()函数负责调用端点并返回富化后的行数据将其与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 QA experiment with factual correctness scoring. # Call AG-UI endpoint and get enriched row enriched await run_ag_ui_row(row, AG_UI_ENDPOINT, metadataTrue) # Score with factual correctness metric fc_result await qa_metrics[0].ascore( responseenriched[response], referencerow[reference], ) # Score with answer relevancy metric ar_result await qa_metrics[1].ascore( user_inputrow[user_input], responseenriched[response], ) # Score with conciseness metric concise_result await conciseness_metric.ascore( responseenriched[response], llmevaluator_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, namescientist_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_inputenriched[messages], reference_tool_callsref_tool_calls, ) goal_result await tool_metrics[1].ascore( user_inputenriched[messages], referencerow.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, nameweather_tool_experiment ) display(tool_result.to_pandas())experiment 装饰器与 run_ag_ui_row 的底层逻辑experiment()装饰器定义于 src/ragas/experiment.py它为函数提供.arun(dataset, name..., backend...)方法框架负责遍历数据集、以Experiment形式收集结果Experiment继承自DataTable支持to_pandas()导出并通过可选backend参数决定结果的存储后端未指定时沿用数据集的 backend。run_ag_ui_row()是推荐的一站式入口见 src/ragas/integrations/ag_ui.py其内部完成四步调用call_ag_ui_endpoint()向端点发起 POST 请求并收集 SSE 事件流通过convert_to_ragas_messages()将事件转换为 Ragas 消息用extract_response()、extract_tool_calls()、extract_contexts()分别提取回答文本、工具调用与上下文返回富化后的行{response, messages, tool_calls, contexts}被并入原行字段。需要注意的边界行为当行中缺少user_input或端点调用失败时函数不会抛错而是返回占位符response为[no response generated by agent]contexts为[no retrieved contexts provided by agent]便于流水线继续运行call_ag_ui_endpoint()使用httpx.AsyncClient以Accept: text/event-stream头请求 SSE 流逐行解析data: {...}前缀的 JSON 事件并用TypeAdapter(Event)按type判别字段反序列化为 AG-UI 事件对象默认超时 60 秒可通过timeout调整仓库示例脚本中设为 300 秒见 experiments.py。进阶底层精细控制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_urlAG_UI_ENDPOINT, user_inputrow[user_input], timeout60.0, ) # Convert AG-UI events to Ragas messages messages convert_to_ragas_messages(events, metadataTrue) # Extract response using helper (or custom logic) response extract_response(messages) # Score with a custom metric score_result await conciseness_metric.ascore( responseresponse, llmevaluator_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, namecustom_ag_ui_experiment ) display(custom_result.to_pandas())底层 API 一览API 层级函数使用时机高层run_ag_ui_row()标准实验——自动完成端点调用、消息转换与结果提取低层call_ag_ui_endpoint()convert_to_ragas_messages()自定义事件处理、按行配置端点、深度调试两种方式都可以与experiment装饰器配合使用按需选择控制粒度即可。深入理解AG-UI 事件流如何转换为 Ragas 消息要真正用好底层 API需要理解消息转换机制。convert_to_ragas_messages()见 src/ragas/integrations/ag_ui.py内部实例化AGUIEventCollector并逐事件调用process_event()。AGUIEventCollector维护三类内部状态_active_text_messages正在流式累积的文本消息Start → Content → End 三元组_active_tool_calls与_completed_tool_calls正在累积与已完成解析的工具调用_current_run_id、_current_thread_id、_current_step来自生命周期事件RUN_STARTED、STEP_STARTED等的上下文信息。其处理规则包括文本消息TEXT_MESSAGE_START初始化TEXT_MESSAGE_CONTENT累积delta分片TEXT_MESSAGE_END时拼接内容并依据role生成AIMessage或HumanMessage工具调用TOOL_CALL_START/TOOL_CALL_ARGS/TOOL_CALL_END重建ToolCallargs通过json.loads解析解析失败时回退为{raw_args: ...}TOOL_CALL_RESULT生成ToolMessage并确保其前驱AIMessage携带tool_calls必要时进行回溯附加以通过MultiTurnSample校验便捷分块事件TEXT_MESSAGE_CHUNK与TOOL_CALL_CHUNK单事件即包含完整消息绕过流式三元组重建快照事件MESSAGES_SNAPSHOT提供完整对话历史可通过convert_messages_snapshot()直接转换效率更高无关事件过滤生命周期、状态管理等非消息事件被静默忽略。metadataTrue时run_id、thread_id、step_name、message_id、timestamp等会保留在转换后消息的metadata字段中便于追踪与调试。上述行为均有对应的单元测试验证参见 tests/unit/integrations/test_ag_ui.py例如test_basic_text_message_conversion分片拼接、test_tool_call_conversion工具调用重建、test_step_context_in_metadata步骤上下文、test_call_ag_ui_endpointSSE 解析等。若未安装ag-ui-protocol相关测试会被pytestmark自动跳过。从 Notebook 到可执行脚本文档以 Notebook 形式给出实验流程仓库同时提供了对应的可执行脚本 examples/ragas_examples/ag_ui_agent_experiments/experiments.py将两个实验封装为命令行工具适合 CI 或定时评测python experiments.py --endpoint-url http://localhost:8000/chat python experiments.py --endpoint-url http://localhost:8000/chat --skip-tool-experiment python experiments.py --endpoint-url http://localhost:8000 --skip-factual脚本支持的参数--endpoint-urlAG-UI 端点地址默认http://localhost:8000--evaluator-model评估 LLM 模型名默认gpt-4o-mini--skip-factual跳过事实正确性实验--skip-tool-experiment跳过工具调用实验。脚本在实验前会先对 Embedding 端点做一次健全性检查并在日志中输出平均factual_correctness、tool_call_f1、agent_goal_accuracy以及完美得分样本占比等汇总统计方便快速判断 Agent 的当前质量水位。小结借助 Ragas 的 AG-UI 集成你可以在不侵入 Agent 实现的前提下通过统一的experiment装饰器模式对任何 AG-UI 兼容端点完成自动化评估单轮场景使用FactualCorrectness、AnswerRelevancy与自定义DiscreteMetric评估回答质量多轮场景使用ToolCallF1与AgentGoalAccuracyWithReference评估工具调用与目标达成度。高层 APIrun_ag_ui_row()封装了端点调用、事件转换与结果提取的完整链路而底层 API 则提供了完全可控的精细化操作空间——两者共享同一套消息转换与收集机制可在实际项目中按需选用。进一步阅读完整的 Notebook 版指南见 docs/howtos/integrations/_ag_ui.md集成实现见 src/ragas/integrations/ag_ui.pyexperiment装饰器源码见 src/ragas/experiment.py可运行的完整示例见 examples/ragas_examples/ag_ui_agent_experiments/experiments.py 及其 测试数据集。【免费下载链接】ragasSupercharge Your LLM Application Evaluations 项目地址: https://gitcode.com/gh_mirrors/ra/ragas创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考