Trace-level Evaluation: Beyond Input/Output Checks

import json


def _as_list(value):
    """Span attributes can arrive as a JSON string or an already-parsed list."""
    if isinstance(value, str):
        try:
            return json.loads(value)
        except json.JSONDecodeError:
            return []
    return value if isinstance(value, list) else []


def user_query(input_messages):
    """The first user message in an LLM span's input."""
    for message in _as_list(input_messages):
        if message.get("message.role") == "user":
            contents = message.get("message.contents") or []
            return message.get("message.content") or "".join(
                c.get("message_content.text", "") for c in contents
            )
    return None


def assistant_text(output_messages):
    """The assistant's text reply in an LLM span (empty on tool-calling turns)."""
    texts = []
    for message in _as_list(output_messages):
        if message.get("message.role") == "assistant":
            if message.get("message.content"):
                texts.append(message["message.content"])
            for c in message.get("message.contents") or []:
                if c.get("message_content.type") == "text":
                    texts.append(c.get("message_content.text", ""))
    return " ".join(t for t in texts if t).strip() or None


# agent_trace_ids = the trace ids whose root span has span_kind == "AGENT"
# (filtering out the evaluators' own LLM traces). See the notebook.
llm_spans = primary_df[
    (primary_df["span_kind"] == "LLM") & (primary_df["context.trace_id"].isin(agent_trace_ids))
].sort_values("start_time")

# Endpoint input  = the user's question (first user message).
# Endpoint output = the agent's FINAL answer only (its last text turn).
trace_df = pd.DataFrame(
    {
        "input": llm_spans.groupby("context.trace_id")["attributes.llm.input_messages"].apply(
            lambda s: next((q for q in s.map(user_query) if q), None)
        ),
        "output": llm_spans.groupby("context.trace_id")["attributes.llm.output_messages"].apply(
            lambda s: next((a for a in reversed(list(s.map(assistant_text))) if a), None)
        ),
    }
).dropna(subset=["input", "output"])

tool_spans = primary_df[primary_df["span_kind"] == "TOOL"].sort_values("start_time")


# tool_path: the ordered tool calls WITH their arguments (selection, order, args).
def format_decision_path(group):
    return " -> ".join(
        f"{row['name']}({row['attributes.input.value']})" for _, row in group.iterrows()
    )


trace_df["tool_path"] = (
    tool_spans.groupby("context.trace_id")[["name", "attributes.input.value"]]
    .apply(format_decision_path)
    .reindex(trace_df.index)
    .fillna("No tools called")
)


# tool_io: each tool call's input AND output.
def format_tool_calls(group):
    lines = []
    for i, (_, row) in enumerate(group.iterrows(), start=1):
        lines.append(
            f"{i}. {row['name']} | input: {row['attributes.input.value']} "
            f"| output: {row['attributes.output.value']}"
        )
    return "\n".join(lines)


trace_df["tool_io"] = (
    tool_spans.groupby("context.trace_id")[["name", "attributes.input.value", "attributes.output.value"]]
    .apply(format_tool_calls)
    .reindex(trace_df.index)
    .fillna("No tools called")
)

RECOMMENDATION_RELEVANCE = """
You are evaluating the relevance of movie recommendations provided by an LLM application.

You will be given:
1. The user input that initiated the trace
2. The list of movie recommendations output by the system

##
User Input:
{input}

Recommendations:
{output}
##

Respond with exactly one word: `correct` or `incorrect`.
1. `correct` →
- All recommended movies match the requested genre or criteria in the user input.
- The recommendations are relevant to the user's request and are not repetitive.
2. `incorrect` →
- One or more recommendations do not match the requested genre or criteria, or the
  recommendations are repetitive.
"""

DECISION_PATH = """
You are evaluating an agent's DECISION PATH: the ordered tool calls it made to
answer a request — which tools, in what order, and with what arguments. You are
NOT judging the final answer text, and you are NOT judging whether each tool's
output was correct — only whether the agent's choices were sensible.

The agent has three tools available:
- movie_selector_llm(genre): returns candidate movies. Must come FIRST, because the
  other tools operate on the selected movies.
- reviewer_llm(movies): reviews and sorts movies. Only meaningful AFTER selection,
  and should be called with the movies that were actually selected.
- preview_summarizer_llm(movie): summarizes a movie. Only meaningful AFTER selection.

You will be given:
1. The user input that initiated the trace
2. The ordered tool calls the agent executed, with their arguments

##
User Input:
{input}

Decision Path (ordered tool calls with arguments):
{tool_path}
##

Respond with exactly one word: `correct` or `incorrect`.
1. `correct` →
- movie_selector_llm is called before reviewer_llm or preview_summarizer_llm, AND
- each tool is called with sensible arguments (e.g. reviewer_llm receives the
  movies that were selected, not an empty or unrelated list).
2. `incorrect` →
- a tool that operates on movies (reviewer_llm / preview_summarizer_llm) runs
  before any movies have been selected, the selection step is missing, OR a tool is
  called with nonsensical arguments (empty/placeholder inputs, or movies that were
  never selected).
"""

REASONING_SUPPORT = """
You are checking whether an agent's FINAL ANSWER is SUPPORTED by the actual
results its tools returned.

You are NOT judging tool order (that's the decision-path check) or genre match
(that's the relevance check). Judge ONLY whether every concrete claim in the final
answer — titles, ratings, scores, review quotes, plot facts — is grounded in the
tool outputs below. An answer that asserts a fact no tool produced (for example a
specific rating) is unsupported, even if it sounds plausible.

##
User Input:
{input}

Tool calls and their results (in order):
{tool_io}

Final Answer:
{output}
##

Respond with exactly one word: `correct` or `incorrect`.
1. `correct` → every concrete claim in the final answer is supported by the tool
   results above.
2. `incorrect` → the final answer asserts at least one concrete fact (a rating,
   score, review, or title) that does not appear in the tool results.
"""

from phoenix.trace import suppress_tracing
from phoenix.evals import LLM, ClassificationEvaluator, async_evaluate_dataframe

llm = LLM(provider="openai", model="gpt-4o-mini")

relevance_evaluator = ClassificationEvaluator(
    name="relevance", llm=llm, prompt_template=RECOMMENDATION_RELEVANCE,
    choices={"correct": 1.0, "incorrect": 0.0},
)
path_evaluator = ClassificationEvaluator(
    name="decision path", llm=llm, prompt_template=DECISION_PATH,
    choices={"correct": 1.0, "incorrect": 0.0},
)
reasoning_evaluator = ClassificationEvaluator(
    name="reasoning", llm=llm, prompt_template=REASONING_SUPPORT,
    choices={"correct": 1.0, "incorrect": 0.0},
)

with suppress_tracing():
    results_df = await async_evaluate_dataframe(
        dataframe=trace_df,
        evaluators=[relevance_evaluator, path_evaluator, reasoning_evaluator],
    )