Implementation:Vibrantlabsai Ragas AgUI Integration

Knowledge Sources	Vibrantlabsai_Ragas
Domains	LLM Evaluation, Agent Integration, AG-UI Protocol, Streaming
Last Updated	2026-02-12 00:00 GMT

Overview

The AG-UI integration module provides conversion utilities and row enrichment for evaluating AG-UI protocol agents with Ragas, supporting conversion between AG-UI streaming events and Ragas message formats, endpoint calling, and sample building for metric scoring.

Description

AG-UI (Agent-to-UI) is an event-based protocol for agent-to-UI communication that uses typed events for streaming text messages, tool calls, and state synchronization. This integration module bridges the AG-UI protocol with Ragas evaluation by providing a comprehensive set of utilities.

The module is organized into several layers:

Primary API: The run_ag_ui_row function is the top-level entry point designed for use inside @experiment-decorated functions. It takes a data row and an AG-UI endpoint URL, calls the endpoint, collects streaming events, converts them to Ragas messages, and returns the row enriched with response text, messages, tool calls, and contexts.

Event Collection: The AGUIEventCollector class is a stateful collector that reconstructs complete messages from incremental AG-UI streaming events. It handles the Start-Content-End triad pattern for both text messages and tool calls, as well as convenience chunk events (TextMessageChunk, ToolCallChunk). It tracks lifecycle events (run started, step started/finished), accumulates content chunks, reconstructs tool call arguments from streamed JSON fragments, and properly associates tool calls with their parent AI messages.

Conversion Functions:

convert_to_ragas_messages converts a list of AG-UI events into Ragas message format by processing them through an AGUIEventCollector.
convert_messages_snapshot converts an AG-UI MessagesSnapshotEvent (complete conversation history in one event) to Ragas messages.
convert_messages_to_ag_ui converts Ragas messages back to AG-UI message format for sending to endpoints, mapping HumanMessage to UserMessage and AIMessage to AssistantMessage.

Extraction Helpers:

extract_response concatenates all AI message content into a single string.
extract_tool_calls collects all ToolCall objects from AI messages.
extract_contexts gathers content from all ToolMessage instances (tool results).

Sample Building: The build_sample function constructs either a SingleTurnSample or MultiTurnSample suitable for Ragas metric scoring, choosing the type based on whether the input is a conversation list or tool call references are provided.

Endpoint Calling: The call_ag_ui_endpoint function makes async HTTP POST requests to AG-UI FastAPI endpoints, parsing the Server-Sent Events (SSE) stream and deserializing events using Pydantic's TypeAdapter for discriminated union handling. It requires the httpx package and supports configurable timeouts and extra HTTP headers.

The AG-UI dependencies (ag-ui-protocol, httpx) are lazily imported to avoid hard dependencies, with clear error messages when packages are missing.

Usage

Import this module when you need to evaluate AG-UI protocol agents using Ragas metrics. The primary entry point run_ag_ui_row should be used inside @experiment-decorated functions for dataset-driven evaluation. Use the lower-level conversion functions when you need direct control over event processing or message format conversion.

Code Reference

Source Location

Repository: Vibrantlabsai_Ragas
File: src/ragas/integrations/ag_ui.py

Signature

class AGUIEventCollector:
    def __init__(self, metadata: bool = False): ...
    def process_event(self, event: Any) -> None: ...
    def get_messages(self) -> List[Union[HumanMessage, AIMessage, ToolMessage]]: ...
    def clear(self) -> None: ...

def convert_to_ragas_messages(
    events: List[Any],
    metadata: bool = False,
) -> List[Union[HumanMessage, AIMessage, ToolMessage]]: ...

def convert_messages_snapshot(
    snapshot_event: Any,
    metadata: bool = False,
) -> List[Union[HumanMessage, AIMessage, ToolMessage]]: ...

def convert_messages_to_ag_ui(
    messages: List[Union[HumanMessage, AIMessage, ToolMessage]],
) -> List[Any]: ...

async def call_ag_ui_endpoint(
    endpoint_url: str,
    user_input: Union[str, List[Union[HumanMessage, AIMessage, ToolMessage]]],
    thread_id: Optional[str] = None,
    agent_config: Optional[Dict[str, Any]] = None,
    timeout: float = 60.0,
    extra_headers: Optional[Dict[str, str]] = None,
) -> List[Any]: ...

def extract_response(
    messages: List[Union[HumanMessage, AIMessage, ToolMessage]],
) -> str: ...

def extract_tool_calls(
    messages: List[Union[HumanMessage, AIMessage, ToolMessage]],
) -> List[ToolCall]: ...

def extract_contexts(
    messages: List[Union[HumanMessage, AIMessage, ToolMessage]],
) -> List[str]: ...

def build_sample(
    user_input: Union[str, List[Union[HumanMessage, AIMessage, ToolMessage]]],
    messages: List[Union[HumanMessage, AIMessage, ToolMessage]],
    reference: Optional[str] = None,
    reference_tool_calls: Optional[Union[str, List[ToolCall]]] = None,
) -> Union[SingleTurnSample, MultiTurnSample]: ...

async def run_ag_ui_row(
    row: Dict[str, Any],
    endpoint_url: str,
    timeout: float = 60.0,
    metadata: bool = False,
    extra_headers: Optional[Dict[str, str]] = None,
) -> Dict[str, Any]: ...

Import

from ragas.integrations.ag_ui import run_ag_ui_row
from ragas.integrations.ag_ui import convert_to_ragas_messages
from ragas.integrations.ag_ui import convert_messages_snapshot
from ragas.integrations.ag_ui import convert_messages_to_ag_ui
from ragas.integrations.ag_ui import call_ag_ui_endpoint
from ragas.integrations.ag_ui import AGUIEventCollector
from ragas.integrations.ag_ui import extract_response, extract_tool_calls, extract_contexts
from ragas.integrations.ag_ui import build_sample

I/O Contract

Inputs (run_ag_ui_row)

Name	Type	Required	Description
row	Dict[str, Any]	Yes	Input row containing at minimum a `"user_input"` field with a string or message list
endpoint_url	str	Yes	URL of the AG-UI FastAPI endpoint (e.g., `"http://localhost:8000/chat"`)
timeout	float	No	Request timeout in seconds (default: 60.0)
metadata	bool	No	Whether to include AG-UI metadata in messages (default: False)
extra_headers	Dict[str, str]	No	Additional HTTP headers for the request

Outputs (run_ag_ui_row)

Name	Type	Description
return	Dict[str, Any]	Original row enriched with `"response"` (str), `"messages"` (List[Message]), `"tool_calls"` (List[ToolCall]), and `"contexts"` (List[str])

Inputs (convert_to_ragas_messages)

Name	Type	Required	Description
events	List[Event]	Yes	List of AG-UI protocol events from an agent run
metadata	bool	No	Whether to include AG-UI metadata in messages (default: False)

Outputs (convert_to_ragas_messages)

Name	Type	Description
return	List[Union[HumanMessage, AIMessage, ToolMessage]]	Ragas messages reconstructed from the event stream, preserving conversation order

Inputs (build_sample)

Name	Type	Required	Description
user_input	str or List[Message]	Yes	Original user input: string for single-turn, message list for multi-turn
messages	List[Message]	Yes	Agent response messages from `convert_to_ragas_messages()`
reference	str	No	Reference/expected answer for evaluation
reference_tool_calls	str or List[ToolCall]	No	Expected tool calls for tool evaluation metrics; accepts JSON string or ToolCall list

Outputs (build_sample)

Name	Type	Description
return	SingleTurnSample or MultiTurnSample	Sample type chosen based on input shape: multi-turn if user_input is a list or reference_tool_calls are provided, single-turn otherwise

Usage Examples

Basic Evaluation with @experiment

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

@experiment()
async def my_experiment(row):
    # Run row against AG-UI endpoint
    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}

# Framework handles dataset iteration
results = await my_experiment.arun(dataset, name="my_eval")

Tool Evaluation with Multi-Turn Samples

from ragas import experiment
from ragas.integrations.ag_ui import run_ag_ui_row, build_sample
from ragas.metrics.collections import ToolCallF1

@experiment()
async def tool_experiment(row):
    enriched = await run_ag_ui_row(row, "http://localhost:8000/chat")

    # Build sample for tool metrics
    sample = build_sample(
        user_input=row["user_input"],
        messages=enriched["messages"],
        reference_tool_calls=row.get("reference_tool_calls"),
    )

    score = await ToolCallF1().multi_turn_ascore(sample)
    return {**enriched, "tool_call_f1": score}

results = await tool_experiment.arun(dataset, name="tool_eval")

Direct Event Conversion

from ragas.integrations.ag_ui import convert_to_ragas_messages, extract_response

# Given a list of AG-UI events from an agent run
ag_ui_events = [...]

# Convert to Ragas messages with metadata
ragas_messages = convert_to_ragas_messages(ag_ui_events, metadata=True)

# Extract the concatenated AI response
response_text = extract_response(ragas_messages)

Using AGUIEventCollector Directly

from ragas.integrations.ag_ui import AGUIEventCollector

collector = AGUIEventCollector(metadata=True)
for event in ag_ui_event_stream:
    collector.process_event(event)

ragas_messages = collector.get_messages()

Constants

Name	Value	Description
MISSING_CONTEXT_PLACEHOLDER	`"[no retrieved contexts provided by agent]"`	Used when no tool results/contexts are available
MISSING_RESPONSE_PLACEHOLDER	`"[no response generated by agent]"`	Used when no AI response content is available

Related Pages

ag-ui-protocol - AG-UI protocol package for event types and message models
httpx - async HTTP client for calling AG-UI endpoints
HumanMessage - Ragas message type from ragas.messages
AIMessage - Ragas message type from ragas.messages
ToolMessage - Ragas message type from ragas.messages
ToolCall - Ragas tool call model from ragas.messages
SingleTurnSample - Single-turn evaluation sample from ragas.dataset_schema
MultiTurnSample - Multi-turn evaluation sample from ragas.dataset_schema
experiment - Ragas experiment decorator for dataset-driven evaluation

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment