Skip to content

feat: Resolve Interactions API data for agent trace display in eval results. #6970

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
copybara-service wants to merge 1 commit into
base: main
Choose a base branch
from
+609 −0
Open

feat: Resolve Interactions API data for agent trace display in eval results. #6970

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
325 changes: 325 additions & 0 deletions agentplatform/_genai/_evals_common.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -575,6 +575,326 @@ def _is_gemini_agent_resource(agent: str) -> bool:
)


def _interaction_dict_to_agent_data(
interaction: dict[str, Any],
) -> dict[str, Any]:
"""Converts an Interaction API JSON response to an AgentData-compatible dict.

Maps the flat list of Interaction steps (user_input, model_output,
function_call, function_result, etc.) into a single ConversationTurn
with AgentEvents, matching the AgentData structure expected by the
evaluation pipeline.

NOTE: This function is shared with cl/944601748. When that CL lands
first, this copy should be removed during the rebase.

Args:
interaction: A dict from the Interactions API GET response.

Returns:
A dict matching the AgentData schema with a single turn containing
all events from the interaction.
"""
events = []
for step in interaction.get("steps", []):
step_type = step.get("type")

if step_type == "user_input":
parts = []
for content_item in step.get("content", []):
if content_item.get("type") == "text":
parts.append({"text": content_item.get("text", "")})
if parts:
events.append({
"author": "user",
"content": {"role": "user", "parts": parts},
})

elif step_type == "model_output":
parts = []
for content_item in step.get("content", []):
if content_item.get("type") == "text":
parts.append({"text": content_item.get("text", "")})
if parts:
events.append({
"author": "agent",
"content": {"role": "model", "parts": parts},
})

elif step_type == "function_call":
events.append({
"author": "agent",
"content": {
"role": "model",
"parts": [{
"function_call": {
"name": step.get("name", ""),
"args": step.get("arguments", {}),
"id": step.get("id", ""),
},
}],
},
})

elif step_type == "function_result":
result = step.get("result")
if isinstance(result, dict):
result_str = json.dumps(result)
elif isinstance(result, str):
result_str = result
else:
result_str = str(result) if result is not None else ""
events.append({
"author": "user",
"content": {
"role": "user",
"parts": [{
"function_response": {
"name": step.get("name", ""),
"response": {"result": result_str},
"id": step.get("callId", step.get("call_id", "")),
},
}],
},
})

agent_data: dict[str, Any] = {
"turns": [{
"turn_index": 0,
"events": events,
}],
}
return agent_data


def _merge_text_parts_in_agent_data(
agent_data: dict[str, Any],
) -> None:
"""Merges consecutive text events and parts for cleaner trace display.

The Interaction API may return multiple consecutive ``model_output``
steps (one per paragraph) and/or multiple text content items within a
single step. ``_interaction_dict_to_agent_data`` maps each step to a
separate event, and each content item to a separate ``part``, causing
the trace renderer to display them as separate visual blocks.

This function performs two merges:

1. **Event merge** -- consecutive events from the same author that
contain only text parts are collapsed into a single event.
2. **Part merge** -- within each (possibly merged) event, consecutive
text-only parts are collapsed into a single part.

Mutates ``agent_data`` in place.

Args:
agent_data: A dict matching the AgentData schema.
"""
for turn in agent_data.get("turns", []):
events = turn.get("events")
if not events:
continue

# --- Pass 1: merge consecutive text-only events from the same author ---
merged_events: list[dict[str, Any]] = []
for event in events:
parts = (event.get("content") or {}).get("parts", [])
is_text_only = parts and all(
"text" in p and len(p) == 1 for p in parts
)
if (
merged_events
and is_text_only
and event.get("author") == merged_events[-1].get("author")
):
prev_parts = (
merged_events[-1].get("content") or {}
).get("parts", [])
prev_is_text_only = prev_parts and all(
"text" in p and len(p) == 1 for p in prev_parts
)
if prev_is_text_only:
prev_parts.extend(parts)
continue
merged_events.append(event)
turn["events"] = merged_events

# --- Pass 2: merge consecutive text parts within each event ---
for event in turn["events"]:
content = event.get("content")
if not content:
continue
parts = content.get("parts")
if not parts or len(parts) <= 1:
continue
merged_parts: list[dict[str, Any]] = []
text_buffer: list[str] = []
for part in parts:
if "text" in part and len(part) == 1:
text_buffer.append(part["text"])
else:
if text_buffer:
merged_parts.append({"text": "\n".join(text_buffer)})
text_buffer = []
merged_parts.append(part)
if text_buffer:
merged_parts.append({"text": "\n".join(text_buffer)})
content["parts"] = merged_parts


def _resolve_interactions_for_display(
api_client: BaseApiClient,
dataset_list: list[types.EvaluationDataset],
) -> list[types.EvaluationDataset]:
"""Resolves interactions_data_source on EvalCases for display.

For each EvalCase that has ``interactions_data_source`` set but no
``agent_data``, fetches the Interaction and Agent config from the
API, converts to AgentData, and populates it on the EvalCase so
that ``show()`` can render the System Topology and Conversation
Trace sections.

Args:
api_client: The API client used to fetch interactions and agents.
dataset_list: The original evaluation datasets.

Returns:
A list of EvaluationDatasets with agent_data populated from
resolved interactions.
"""
resolved_datasets = []
for dataset in dataset_list:
if not dataset.eval_cases:
resolved_datasets.append(dataset)
continue

resolved_cases = []
any_resolved = False
for case in dataset.eval_cases:
ids = getattr(case, "interactions_data_source", None)
existing_agent_data = getattr(case, "agent_data", None)
if ids and not existing_agent_data:
interaction_name = getattr(ids, "interaction", None)
gemini_cfg = getattr(ids, "gemini_agent_config", None)
agent_name = (
getattr(gemini_cfg, "gemini_agent", None)
if gemini_cfg else None
)

if interaction_name:
try:
# Extract the interaction ID from the resource name.
parts = interaction_name.split("/")
if len(parts) >= 6 and parts[4] == "interactions":
interaction_id = parts[5]
else:
interaction_id = interaction_name

logger.info(
"Fetching interaction %s for display.",
interaction_name,
)
path = f"interactions/{interaction_id}"
response = api_client.request("get", path, {}, None)
if not response.body:
logger.warning(
"Empty response fetching interaction %s.",
interaction_name,
)
resolved_cases.append(case)
continue
interaction_dict = json.loads(response.body)

agent_data = _interaction_dict_to_agent_data(
interaction_dict
)

# Derive the agent short name from the resource name.
agent_id = "agent"
if agent_name:
agent_parts = agent_name.split("/")
if len(agent_parts) >= 2:
agent_id = agent_parts[-1]

# Best-effort: fetch agent config (instruction,
# tools, description) from the Agent API.
agent_config: dict[str, Any] = {"agent_id": agent_id}
if agent_name:
try:
agent_short_id = agent_name.split("/")[-1]
agent_resp = api_client.request(
"get", f"agents/{agent_short_id}", {}, None
)
if agent_resp.body:
agent_dict = json.loads(agent_resp.body)
if agent_dict.get("system_instruction"):
agent_config["instruction"] = (
agent_dict["system_instruction"]
)
if agent_dict.get("description"):
agent_config["description"] = (
agent_dict["description"]
)
if agent_dict.get("base_agent"):
agent_config["agent_type"] = (
agent_dict["base_agent"]
)
if agent_dict.get("tools"):
cleaned_tools = []
for tool in agent_dict["tools"]:
if isinstance(tool, dict):
tool = {
k: v
for k, v in tool.items()
if k != "type"
}
if tool:
cleaned_tools.append(tool)
if cleaned_tools:
agent_config["tools"] = (
cleaned_tools
)
except Exception as e: # pylint: disable=broad-exception-caught
logger.warning(
"Failed to fetch agent config for '%s'"
" (continuing without it): %s",
agent_name,
e,
)
agent_data["agents"] = {agent_id: agent_config}

_merge_text_parts_in_agent_data(agent_data)

case_dict = case.model_dump(
mode="python", exclude_none=True
)
case_dict["agent_data"] = agent_data
resolved_cases.append(types.EvalCase(**case_dict))
any_resolved = True
continue
except Exception as e: # pylint: disable=broad-exception-caught
logger.warning(
"Failed to resolve interaction %s for display: %s",
interaction_name,
e,
)

resolved_cases.append(case)

if any_resolved:
ds_dict = dataset.model_dump(mode="python", exclude_none=True)
ds_dict.pop("eval_cases", None)
resolved_datasets.append(
types.EvaluationDataset(
eval_cases=resolved_cases, **ds_dict
)
)
else:
resolved_datasets.append(dataset)

return resolved_datasets


def _add_evaluation_run_labels(
labels: Optional[dict[str, str]] = None,
agent: Optional[str] = None,
Expand Down Expand Up @@ -1897,6 +2217,11 @@ def _execute_evaluation( # type: ignore[no-untyped-def]
t2 = time.perf_counter()
logger.info("Evaluation took: %f seconds", t2 - t1)

# Resolve interactions_data_source to agent_data for display.
# This fetches Interaction trace data client-side so that show() can
# render the System Topology and Conversation Trace sections.
dataset_list = _resolve_interactions_for_display(api_client, dataset_list)

evaluation_result.evaluation_dataset = dataset_list
evaluation_result.agent_info = validated_agent_info

Expand Down
Loading
Loading