Evaluate Support Swarm Example

This page embeds the full source for examples/evaluate_support_swarm.py.

"""Example: generate heuristic evaluation inputs and score recorded swarm execution artifacts."""

import json

from swarmforge.authoring import build_swarm_definition
from swarmforge.evaluation import (
    build_graph_snapshot,
    build_heuristic_swarm_intents,
    build_intent_based_swarm_scenario_seeds,
    classify_scenario_feasibility,
    evaluate_scenario_artifacts,
)
from swarmforge.swarm.models import SessionCheckpoint


swarm = build_swarm_definition(
    {
        "nodes": [
            {
                "node_key": "triage",
                "name": "Triage",
                "persona": "",
                "is_entry_node": True,
                "sub_agents": [
                    {
                        "sub_agent": "billing",
                        "handoff_description": "Transfer after confirming the request is billing-related.",
                        "required_variables": ["account_id"],
                    }
                ],
            },
            {
                "node_key": "billing",
                "name": "Billing",
                "persona": "",
                "is_entry_node": False,
            },
        ],
    },
    swarm_id="support",
    name="Support Swarm",
)

graph_snapshot = build_graph_snapshot(swarm)
intents = build_heuristic_swarm_intents(graph_snapshot, 2)
scenario_seed = build_intent_based_swarm_scenario_seeds(
    graph_snapshot,
    [intent["title"] for intent in intents],
    1,
    2,
)[0]

event_log = [
    {"event": "open", "data": {"currentAgent": "Triage"}},
    {
        "event": "handoff",
        "data": {
            "from": "Triage",
            "to": "Billing",
            "fromNodeKey": "triage",
            "toNodeKey": "billing",
            "handoffType": "agent_transfer",
            "requiredVariables": ["account_id"],
            "resolvedVariables": {"account_id": "ACME-991"},
            "state": {"account_id": "ACME-991"},
            "globalVariables": {"account_id": "ACME-991"},
        },
    },
    {"event": "text_chunk", "data": {"text_chunk": "I can help with that billing issue.", "agentName": "Billing"}},
    {"event": "done", "data": {"fullText": "I can help with that billing issue."}},
]

checkpoints = [
    SessionCheckpoint(
        session_id="session-1",
        acting_node_id="triage",
        step_number=1,
        action_type="handoff",
        state_snapshot={"state": {"account_id": "ACME-991"}},
    ),
    SessionCheckpoint(
        session_id="session-1",
        acting_node_id="billing",
        step_number=2,
        action_type="agent_response",
        state_snapshot={"state": {"account_id": "ACME-991"}},
    ),
]

print("Scenario seed:")
print(json.dumps(scenario_seed, indent=2))
print()
print("Feasibility:")
print(json.dumps(classify_scenario_feasibility(graph_snapshot, scenario_seed), indent=2))
print()
print("Score:")
print(json.dumps(evaluate_scenario_artifacts(graph_snapshot, scenario_seed, event_log=event_log, checkpoints=checkpoints), indent=2))

Run:

python examples/evaluate_support_swarm.py