Skip to content

Python API Reference

Top-level exports

from agentdelta import AgentTrace, diff_traces, record

agentdelta.trace.AgentTrace(run_id, nodes=list(), edges=list(), metadata=dict()) dataclass

Complete execution trace for a single agent run.

Attributes:

Name Type Description
run_id str

Unique identifier for this run (e.g. "v1.0").
nodes list[TraceNode]

Ordered list of trace steps.
edges list[TraceEdge]

Directed edges connecting steps.
metadata dict[str, Any]

Arbitrary key/value pairs stored in the trace header.

add_node(node)

Append a node to the trace.

Source code in src/agentdelta/trace.py 
129
130
131
def add_node(self, node: TraceNode) -> None:
    """Append a node to the trace."""
    self.nodes.append(node)

add_edge(edge)

Append an edge to the trace.

Source code in src/agentdelta/trace.py 
133
134
135
def add_edge(self, edge: TraceEdge) -> None:
    """Append an edge to the trace."""
    self.edges.append(edge)

save(path)

Write the trace to a JSONL file at path (one record per line).

Source code in src/agentdelta/trace.py 
137
138
139
140
141
142
143
144
145
146
def save(self, path: str | Path) -> None:
    """Write the trace to a JSONL file at *path* (one record per line)."""
    path = Path(path)
    with path.open("w") as f:
        meta = {"type": "trace_meta", "run_id": self.run_id, **self.metadata}
        f.write(json.dumps(meta) + "\n")
        for node in self.nodes:
            f.write(json.dumps({"type": "node", **node.to_dict()}) + "\n")
        for edge in self.edges:
            f.write(json.dumps({"type": "edge", **edge.to_dict()}) + "\n")

load(path) classmethod

Load a trace from a JSONL file previously written by save.

Source code in src/agentdelta/trace.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
@classmethod
def load(cls, path: str | Path) -> AgentTrace:
    """Load a trace from a JSONL file previously written by ``save``."""
    path = Path(path)
    nodes: list[TraceNode] = []
    edges: list[TraceEdge] = []
    run_id = path.stem
    metadata: dict[str, Any] = {}

    with path.open() as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            record = json.loads(line)
            rtype = record.pop("type")
            if rtype == "trace_meta":
                run_id = record.pop("run_id", run_id)
                metadata = record
            elif rtype == "node":
                nodes.append(TraceNode.from_dict(record))
            elif rtype == "edge":
                edges.append(TraceEdge.from_dict(record))

    return cls(run_id=run_id, nodes=nodes, edges=edges, metadata=metadata)

agentdelta.trace.TraceNode(step, node_type, content, metadata=dict(), embedding=None) dataclass

A single step in an agent execution trace.

Attributes:

Name Type Description
step int

1-based sequential position in the trace.
node_type NodeType

Category of this step (LLM reasoning, tool call, etc.).
content str

Human-readable text - reasoning output, tool(args), or tool return value.
metadata dict[str, Any]

Arbitrary key/value pairs for framework-specific data.
embedding list[float] | None

Floating-point sentence embedding, populated by embed_trace().

id property

Content-addressed ID - same content always produces the same ID.

to_dict()

Serialise to a plain dict suitable for JSON encoding.

Source code in src/agentdelta/trace.py 
56
57
58
59
60
61
62
63
64
def to_dict(self) -> dict[str, Any]:
    """Serialise to a plain dict suitable for JSON encoding."""
    return {
        "id": self.id,
        "step": self.step,
        "node_type": self.node_type.value,
        "content": self.content,
        "metadata": self.metadata,
    }

from_dict(d) classmethod

Deserialise from a plain dict (as produced by to_dict).

Source code in src/agentdelta/trace.py 
66
67
68
69
70
71
72
73
74
@classmethod
def from_dict(cls, d: dict[str, Any]) -> TraceNode:
    """Deserialise from a plain dict (as produced by ``to_dict``)."""
    return cls(
        step=d["step"],
        node_type=NodeType(d["node_type"]),
        content=d["content"],
        metadata=d.get("metadata", {}),
    )

agentdelta.trace.TraceEdge(source_step, target_step, edge_type, label='') dataclass

A directed connection between two steps in a trace.

Attributes:

Name Type Description
source_step int

Step number of the originating node.
target_step int

Step number of the destination node.
edge_type EdgeType

Semantic category of this transition.
label str

Optional human-readable label (tool name, phase, etc.).

to_dict()

Serialise to a plain dict suitable for JSON encoding.

Source code in src/agentdelta/trace.py 
 93
 94
 95
 96
 97
 98
 99
100
def to_dict(self) -> dict[str, Any]:
    """Serialise to a plain dict suitable for JSON encoding."""
    return {
        "source_step": self.source_step,
        "target_step": self.target_step,
        "edge_type": self.edge_type.value,
        "label": self.label,
    }

from_dict(d) classmethod

Deserialise from a plain dict (as produced by to_dict).

Source code in src/agentdelta/trace.py 
102
103
104
105
106
107
108
109
110
@classmethod
def from_dict(cls, d: dict[str, Any]) -> TraceEdge:
    """Deserialise from a plain dict (as produced by ``to_dict``)."""
    return cls(
        source_step=d["source_step"],
        target_step=d["target_step"],
        edge_type=EdgeType(d["edge_type"]),
        label=d.get("label", ""),
    )

agentdelta.trace.NodeType

Bases: str, Enum

Classification of a single step in an agent trace.

agentdelta.trace.EdgeType

Bases: str, Enum

Classification of a directed edge between two trace nodes.

Diff

agentdelta.diff.diff_traces(trace_a, trace_b, fork_threshold=0.7, match_threshold=0.85)

Compute a semantic diff between two agent traces.

Parameters:

Name Type Description Default
trace_a AgentTrace

Baseline trace.

 required
trace_b AgentTrace

Comparison trace.

 required
fork_threshold float

Similarity below this triggers a fork point.

 0.7
match_threshold float

Similarity above this is considered a match.

 0.85

Returns:

Type Description
DiffResult

DiffResult with aligned steps and the first fork point if found.
Source code in src/agentdelta/diff.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
def diff_traces(
    trace_a: AgentTrace,
    trace_b: AgentTrace,
    fork_threshold: float = 0.70,
    match_threshold: float = 0.85,
) -> DiffResult:
    """
    Compute a semantic diff between two agent traces.

    Args:
        trace_a: Baseline trace.
        trace_b: Comparison trace.
        fork_threshold: Similarity below this triggers a fork point.
        match_threshold: Similarity above this is considered a match.

    Returns:
        DiffResult with aligned steps and the first fork point if found.
    """
    # Ensure both traces are embedded
    embed_trace(trace_a)
    embed_trace(trace_b)

    alignment = align_traces(trace_a, trace_b, threshold=fork_threshold)

    steps: list[StepDiff] = []
    fork_point: ForkPoint | None = None

    for na, nb, score in alignment:
        if na is None:
            if nb is None:
                continue
            summary = f"+ [{nb.node_type.value}] {nb.content[:80]}"
            steps.append(StepDiff(None, nb, 0.0, "added", summary))
        elif nb is None:
            summary = f"- [{na.node_type.value}] {na.content[:80]}"
            steps.append(StepDiff(na, None, 0.0, "removed", summary))
        elif score >= match_threshold:
            steps.append(StepDiff(na, nb, score, "match"))
        else:
            desc = _describe_fork(na, nb, score)
            step = StepDiff(na, nb, score, "changed", desc)
            steps.append(step)
            # Record the first fork point
            if fork_point is None:
                fork_point = ForkPoint(
                    step_a=na.step,
                    step_b=nb.step,
                    node_a=na,
                    node_b=nb,
                    similarity=score,
                    description=desc,
                )

    total = len(alignment)
    matched = sum(1 for s in steps if s.status == "match")
    changed = sum(1 for s in steps if s.status == "changed")

    result = DiffResult(
        run_id_a=trace_a.run_id,
        run_id_b=trace_b.run_id,
        steps=steps,
        fork_point=fork_point,
        summary={
            "total_steps": total,
            "matched": matched,
            "changed": changed,
            "added": len([s for s in steps if s.status == "added"]),
            "removed": len([s for s in steps if s.status == "removed"]),
            "similarity_pct": round(matched / total * 100, 1) if total else 100.0,
            "has_regression": fork_point is not None or (total > 0 and matched == 0),
            "fork_step": fork_point.step_a if fork_point else None,
        },
    )
    return result

agentdelta.diff.DiffResult(run_id_a, run_id_b, steps=list(), fork_point=None, summary=dict()) dataclass

Full diff result between two agent traces.

Attributes:

Name Type Description
run_id_a str

Run identifier of the baseline trace.
run_id_b str

Run identifier of the candidate trace.
steps list[StepDiff]

All aligned step pairs, in order.
fork_point ForkPoint | None

The first divergent step, or None if the traces are equivalent.
summary dict[str, Any]

Pre-computed aggregate statistics (total, matched, changed, etc.).

has_regression property

True if the traces diverged (a fork point was detected).

changed_steps property

Steps where both traces have a node but they diverged semantically.

added_steps property

Steps present only in trace B (inserted relative to baseline).

removed_steps property

Steps present only in trace A (removed relative to baseline).

agentdelta.diff.ForkPoint(step_a, step_b, node_a, node_b, similarity, description) dataclass

The first step where two traces take meaningfully different paths.

Attributes:

Name Type Description
step_a int

Step number in trace A where the fork occurred.
step_b int

Step number in trace B where the fork occurred.
node_a TraceNode

The divergent node from trace A.
node_b TraceNode

The divergent node from trace B.
similarity float

Cosine similarity between the two nodes at the fork (< fork_threshold).
description str

Human-readable explanation of why this step diverged.

is_tool_change()

Return True if the fork is a tool-selection or tool-return change.

Source code in src/agentdelta/diff.py 
32
33
34
35
36
37
def is_tool_change(self) -> bool:
    """Return True if the fork is a tool-selection or tool-return change."""
    return self.node_a.node_type in (
        NodeType.TOOL_CALL,
        NodeType.TOOL_RETURN,
    ) and self.node_b.node_type in (NodeType.TOOL_CALL, NodeType.TOOL_RETURN)

is_reasoning_change()

Return True if the fork is an LLM reasoning divergence.

Source code in src/agentdelta/diff.py 
39
40
41
def is_reasoning_change(self) -> bool:
    """Return True if the fork is an LLM reasoning divergence."""
    return self.node_a.node_type == NodeType.LLM and self.node_b.node_type == NodeType.LLM

agentdelta.diff.StepDiff(step_a, step_b, similarity, status, summary='') dataclass

A single aligned step pair with its comparison result.

Attributes:

Name Type Description
step_a TraceNode | None

Node from trace A, or None if this step was added in B.
step_b TraceNode | None

Node from trace B, or None if this step was removed in A.
similarity float

Cosine similarity between the two nodes (0.0 for added/removed).
status str

One of "match", "changed", "added", or "removed".
summary str

Human-readable one-line description of this diff entry.

Embeddings

agentdelta.embed.embed_trace(trace, batch_size=64)

Compute embeddings for all nodes in a trace (in-place) and return the trace.

Source code in src/agentdelta/embed.py 
33
34
35
36
37
38
39
40
41
42
def embed_trace(trace: AgentTrace, batch_size: int = 64) -> AgentTrace:
    """Compute embeddings for all nodes in a trace (in-place) and return the trace."""
    model = _get_model()
    contents = [node.content for node in trace.nodes]
    if not contents:
        return trace
    embeddings = model.encode(contents, batch_size=batch_size, show_progress_bar=False)
    for node, emb in zip(trace.nodes, embeddings, strict=False):
        node.embedding = emb.tolist()
    return trace

agentdelta.embed.align_traces(trace_a, trace_b, window=5, threshold=0.75)

Align nodes from two traces by semantic similarity within a sliding window.

Uses greedy 1:1 matching: each node in trace_a is paired with the closest unmatched node in trace_b within ±window positions.

Returns:

Type Description
list[tuple[TraceNode | None, TraceNode | None, float]]

List of (node_a, node_b, similarity) triples.
list[tuple[TraceNode | None, TraceNode | None, float]]

Unmatched nodes appear as (node, None, 0.0) or (None, node, 0.0).
Source code in src/agentdelta/embed.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def align_traces(
    trace_a: AgentTrace,
    trace_b: AgentTrace,
    window: int = 5,
    threshold: float = 0.75,
) -> list[tuple[TraceNode | None, TraceNode | None, float]]:
    """Align nodes from two traces by semantic similarity within a sliding window.

    Uses greedy 1:1 matching: each node in *trace_a* is paired with the
    closest unmatched node in *trace_b* within ±*window* positions.

    Returns:
        List of ``(node_a, node_b, similarity)`` triples.
        Unmatched nodes appear as ``(node, None, 0.0)`` or ``(None, node, 0.0)``.
    """
    nodes_a = trace_a.nodes
    nodes_b = trace_b.nodes

    alignment: list[tuple[TraceNode | None, TraceNode | None, float]] = []
    used_b: set[int] = set()
    node_to_idx: dict[int, int] = {id(nb): j for j, nb in enumerate(nodes_b)}

    for i, na in enumerate(nodes_a):
        start = max(0, i - window)
        end = min(len(nodes_b), i + window + 1)
        candidates = [nodes_b[j] for j in range(start, end) if j not in used_b]

        match, score = find_best_match(na, candidates, threshold)
        if match is not None:
            used_b.add(node_to_idx[id(match)])
            alignment.append((na, match, score))
        else:
            alignment.append((na, None, 0.0))

    for j, nb in enumerate(nodes_b):
        if j not in used_b:
            alignment.append((None, nb, 0.0))

    return alignment

agentdelta.embed.cosine_similarity(a, b)

Return cosine similarity between two embedding vectors. Returns 0.0 for zero vectors.

Source code in src/agentdelta/embed.py 
45
46
47
48
49
50
51
def cosine_similarity(a: list[float], b: list[float]) -> float:
    """Return cosine similarity between two embedding vectors. Returns 0.0 for zero vectors."""
    va, vb = np.array(a), np.array(b)
    denom = np.linalg.norm(va) * np.linalg.norm(vb)
    if denom == 0:
        return 0.0
    return float(np.dot(va, vb) / denom)

agentdelta.embed.find_best_match(node, candidates, threshold=0.75)

Find the candidate most semantically similar to node.

Returns (best_node, score). If the best score is below threshold, returns (None, best_score) rather than a low-confidence match.

Source code in src/agentdelta/embed.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def find_best_match(
    node: TraceNode,
    candidates: list[TraceNode],
    threshold: float = 0.75,
) -> tuple[TraceNode | None, float]:
    """Find the candidate most semantically similar to *node*.

    Returns ``(best_node, score)``. If the best score is below *threshold*,
    returns ``(None, best_score)`` rather than a low-confidence match.
    """
    if node.embedding is None or not candidates:
        return None, 0.0

    best_node, best_score = None, -1.0
    for candidate in candidates:
        if candidate.embedding is None:
            continue
        score = cosine_similarity(node.embedding, candidate.embedding)
        if score > best_score:
            best_score = score
            best_node = candidate

    if best_score < threshold:
        return None, best_score
    return best_node, best_score

Instrumentation

agentdelta.instrument.record(output_path, run_id=None)

Context manager that records an agent run and saves the trace on exit.

Usage

with agentdelta.record("run_a.jsonl") as cb: agent.invoke({"input": "..."}, config={"callbacks": [cb]})

trace_a.jsonl is now saved

Source code in src/agentdelta/instrument.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
@contextlib.contextmanager
def record(output_path: str | Path, run_id: str | None = None) -> Iterator[AgentdeltaCallback]:
    """
    Context manager that records an agent run and saves the trace on exit.

    Usage:
        with agentdelta.record("run_a.jsonl") as cb:
            agent.invoke({"input": "..."}, config={"callbacks": [cb]})
        # trace_a.jsonl is now saved
    """
    callback = AgentdeltaCallback(run_id=run_id)
    try:
        yield callback
    finally:
        callback.trace.save(output_path)

agentdelta.instrument.AgentdeltaCallback(run_id=None)

LangChain BaseCallbackHandler-compatible callback that records agent runs as AgentTrace objects.

Usage

callback = AgentdeltaCallback() agent.invoke({"input": "..."}, config={"callbacks": [callback]}) trace = callback.trace trace.save("run.jsonl")

Source code in src/agentdelta/instrument.py 
26
27
28
29
def __init__(self, run_id: str | None = None) -> None:
    self.run_id = run_id or str(uuid.uuid4())[:8]
    self.trace = AgentTrace(run_id=self.run_id)
    self._step = 0

on_llm_start(serialized, prompts, **kwargs)

No-op - input prompts are not captured; only LLM output is recorded.

Source code in src/agentdelta/instrument.py 
37
38
def on_llm_start(self, serialized: dict[str, Any], prompts: list[str], **kwargs: Any) -> None:
    """No-op - input prompts are not captured; only LLM output is recorded."""

on_llm_end(response, **kwargs)

Record an LLM generation as a NodeType.LLM trace node.

Source code in src/agentdelta/instrument.py 
40
41
42
43
44
45
46
47
48
49
50
def on_llm_end(self, response: Any, **kwargs: Any) -> None:
    """Record an LLM generation as a ``NodeType.LLM`` trace node."""
    try:
        text = response.generations[0][0].text
    except (AttributeError, IndexError):
        text = str(response)
    step = self._next_step()
    node = TraceNode(step=step, node_type=NodeType.LLM, content=text[:2000])
    self.trace.add_node(node)
    if step > 1:
        self.trace.add_edge(TraceEdge(step - 1, step, EdgeType.LLM_DECISION, "llm_output"))

on_tool_start(serialized, input_str, **kwargs)

Record a tool invocation as a NodeType.TOOL_CALL trace node.

Source code in src/agentdelta/instrument.py 
52
53
54
55
56
57
58
59
60
def on_tool_start(self, serialized: dict[str, Any], input_str: str, **kwargs: Any) -> None:
    """Record a tool invocation as a ``NodeType.TOOL_CALL`` trace node."""
    tool_name = serialized.get("name", "unknown_tool")
    step = self._next_step()
    content = f"{tool_name}({input_str[:500]})"
    node = TraceNode(step=step, node_type=NodeType.TOOL_CALL, content=content)
    self.trace.add_node(node)
    if step > 1:
        self.trace.add_edge(TraceEdge(step - 1, step, EdgeType.TOOL_CALL, tool_name))

on_tool_end(output, **kwargs)

Record a tool result as a NodeType.TOOL_RETURN trace node.

Source code in src/agentdelta/instrument.py 
62
63
64
65
66
67
68
def on_tool_end(self, output: str, **kwargs: Any) -> None:
    """Record a tool result as a ``NodeType.TOOL_RETURN`` trace node."""
    step = self._next_step()
    node = TraceNode(step=step, node_type=NodeType.TOOL_RETURN, content=str(output)[:500])
    self.trace.add_node(node)
    if step > 1:
        self.trace.add_edge(TraceEdge(step - 1, step, EdgeType.TOOL_RETURN, "tool_output"))

on_chain_start(serialized, inputs, **kwargs)

Record the initial chain input as a NodeType.START node (first call only).

Source code in src/agentdelta/instrument.py 
70
71
72
73
74
75
76
77
78
79
80
81
def on_chain_start(
    self, serialized: dict[str, Any], inputs: dict[str, Any], **kwargs: Any
) -> None:
    """Record the initial chain input as a ``NodeType.START`` node (first call only)."""
    if self._step == 0:
        step = self._next_step()
        node = TraceNode(
            step=step,
            node_type=NodeType.START,
            content=str(inputs.get("input", inputs))[:500],
        )
        self.trace.add_node(node)

on_chain_end(outputs, **kwargs)

Record the final chain output as a NodeType.END node.

Source code in src/agentdelta/instrument.py 
83
84
85
86
87
88
89
90
91
92
93
def on_chain_end(self, outputs: dict[str, Any], **kwargs: Any) -> None:
    """Record the final chain output as a ``NodeType.END`` node."""
    step = self._next_step()
    node = TraceNode(
        step=step,
        node_type=NodeType.END,
        content=str(outputs.get("output", outputs))[:500],
    )
    self.trace.add_node(node)
    if step > 1:
        self.trace.add_edge(TraceEdge(step - 1, step, EdgeType.SEQUENCE, "chain_end"))

on_agent_action(action, **kwargs)

No-op - agent actions are captured via on_tool_start.

Source code in src/agentdelta/instrument.py 
95
96
def on_agent_action(self, action: Any, **kwargs: Any) -> None:
    """No-op - agent actions are captured via ``on_tool_start``."""

on_agent_finish(finish, **kwargs)

No-op - agent finish is captured via on_chain_end.

Source code in src/agentdelta/instrument.py 
98
99
def on_agent_finish(self, finish: Any, **kwargs: Any) -> None:
    """No-op - agent finish is captured via ``on_chain_end``."""

on_llm_error(error, **kwargs)

No-op error shim required for LangGraph compatibility.

Source code in src/agentdelta/instrument.py 
102
103
def on_llm_error(self, error: Exception, **kwargs: Any) -> None:
    """No-op error shim required for LangGraph compatibility."""

on_tool_error(error, **kwargs)

No-op error shim required for LangGraph compatibility.

Source code in src/agentdelta/instrument.py 
105
106
def on_tool_error(self, error: Exception, **kwargs: Any) -> None:
    """No-op error shim required for LangGraph compatibility."""

on_chain_error(error, **kwargs)

No-op error shim required for LangGraph compatibility.

Source code in src/agentdelta/instrument.py 
108
109
def on_chain_error(self, error: Exception, **kwargs: Any) -> None:
    """No-op error shim required for LangGraph compatibility."""

Report

agentdelta.report.print_diff(result, show_matches=False, console=None)

Print a Rich-formatted diff to the terminal.

Source code in src/agentdelta/report.py 
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def print_diff(
    result: DiffResult,
    show_matches: bool = False,
    console: Console | None = None,
) -> None:
    """Print a Rich-formatted diff to the terminal."""
    con = console or _console

    # Header
    title = f"[bold]agentdelta[/bold]  [dim]{result.run_id_a}[/dim] vs [dim]{result.run_id_b}[/dim]"
    con.print(Panel(title, expand=False))

    # Summary line
    s = result.summary
    status_color = "red" if s["has_regression"] else "green"
    status_word = "REGRESSION DETECTED" if s["has_regression"] else "NO REGRESSION"
    con.print(
        f"  [{status_color}]{status_word}[/{status_color}]  "
        f"[dim]{s['matched']}/{s['total_steps']} steps matched "
        f"({s['similarity_pct']}%)[/dim]  "
        f"[yellow]{s['changed']} changed[/yellow]  "
        f"[green]+{s['added']} added[/green]  "
        f"[red]-{s['removed']} removed[/red]"
    )

    # Fork point callout
    if result.fork_point:
        fp = result.fork_point
        con.print()
        con.print(
            Panel(
                f"[bold yellow]⚡ First fork at step {fp.step_a}[/bold yellow]\n"
                f"[white]{fp.description}[/white]\n\n"
                f"  [dim]Before:[/dim] {_truncate(fp.node_a.content)}\n"
                f"  [dim]After: [/dim] {_truncate(fp.node_b.content)}",
                title="[bold yellow]Fork Point[/bold yellow]",
                border_style="yellow",
                expand=False,
            )
        )

    # Step table
    if result.changed_steps or result.added_steps or result.removed_steps or show_matches:
        con.print()
        table = Table(show_header=True, header_style="bold", box=None, padding=(0, 1))
        table.add_column("Step", style="dim", width=5)
        table.add_column("Status", width=8)
        table.add_column("Type", width=12)
        table.add_column("Detail", no_wrap=False)

        for step in result.steps:
            if step.status == "match" and not show_matches:
                continue

            style = _STATUS_STYLE[step.status]
            node = step.step_a or step.step_b
            if node is None:
                continue

            icon = _NODE_ICONS.get(node.node_type, "?")
            step_num = str(node.step)
            node_type = f"{icon} {node.node_type.value}"

            if step.status == "match":
                detail = Text(_truncate(node.content), style="dim")
            else:
                detail = Text(step.summary or _truncate(node.content), style=style)

            table.add_row(step_num, step.status.upper(), node_type, detail)

        con.print(table)

    con.print()

agentdelta.report.to_json(result)

Serialize a DiffResult to JSON for CI/CD consumption.

Source code in src/agentdelta/report.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def to_json(result: DiffResult) -> str:
    """Serialize a DiffResult to JSON for CI/CD consumption."""
    steps_data = []
    for step in result.steps:
        node = step.step_a or step.step_b
        steps_data.append(
            {
                "status": step.status,
                "similarity": round(step.similarity, 4),
                "step_a": step.step_a.step if step.step_a else None,
                "step_b": step.step_b.step if step.step_b else None,
                "node_type": node.node_type.value if node else None,
                "summary": step.summary,
            }
        )

    data: dict[str, Any] = {
        "run_id_a": result.run_id_a,
        "run_id_b": result.run_id_b,
        "summary": result.summary,
        "fork_point": (
            {
                "step_a": result.fork_point.step_a,
                "step_b": result.fork_point.step_b,
                "similarity": round(result.fork_point.similarity, 4),
                "description": result.fork_point.description,
                "node_a_content": result.fork_point.node_a.content[:200],
                "node_b_content": result.fork_point.node_b.content[:200],
            }
            if result.fork_point
            else None
        ),
        "steps": steps_data,
    }
    return json.dumps(data, indent=2)

agentdelta.report.to_markdown(result)

Generate a GitHub PR comment in Markdown.

Source code in src/agentdelta/report.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
def to_markdown(result: DiffResult) -> str:
    """Generate a GitHub PR comment in Markdown."""
    s = result.summary
    status_emoji = "🔴" if s["has_regression"] else "🟢"
    status_word = "Regression detected" if s["has_regression"] else "No regression"

    lines = [
        "## agentdelta behavior diff",
        "",
        f"{status_emoji} **{status_word}** &nbsp;·&nbsp; "
        f"{s['matched']}/{s['total_steps']} steps matched ({s['similarity_pct']}%) &nbsp;·&nbsp; "
        f"**{s['changed']}** changed &nbsp; "
        f"**+{s['added']}** added &nbsp; **-{s['removed']}** removed",
        "",
    ]

    if result.fork_point:
        fp = result.fork_point
        lines += [
            "### ⚡ First fork point",
            "",
            f"> **Step {fp.step_a}** - {fp.description}",
            "",
            "```diff",
            f"- {fp.node_a.content[:200]}",
            f"+ {fp.node_b.content[:200]}",
            "```",
            "",
        ]

    changed = [s for s in result.steps if s.status != "match"]
    if changed:
        lines += [
            "### Changed steps",
            "",
            "| Step | Status | Type | Detail |",
            "|------|--------|------|--------|",
        ]
        for step in changed[:20]:  # cap at 20 rows
            node = step.step_a or step.step_b
            if not node:
                continue
            icon = {"added": "+", "removed": "-", "changed": "~"}.get(step.status, "")
            detail = (step.summary or node.content[:60]).replace("|", "\\|")
            lines.append(
                f"| {node.step} | {icon} {step.status} | {node.node_type.value} | {detail} |"
            )
        if len(changed) > 20:
            lines.append(f"| … | | | *{len(changed) - 20} more rows* |")

    lines += [
        "",
        "<details><summary>Full JSON report</summary>",
        "",
        "```json",
        to_json(result),
        "```",
        "",
        "</details>",
        "",
        "*Generated by [agentdelta](https://github.com/sandeep-alluru/agentdelta)*",
    ]

    return "\n".join(lines)