Simulating Customer Support Conversations

from netra import Netra
from netra.simulation.task import BaseTask
from netra.simulation.models import TaskResult
from openai import OpenAI
import uuid

Netra.init(app_name="support-simulation")
client = OpenAI()

# Store conversation history per session
conversations: dict[str, list] = {}

class SupportAgentTask(BaseTask):
    """Wraps a customer support agent for simulation."""

    def run(self, message: str, session_id: str | None = None) -> TaskResult:
        session = session_id or str(uuid.uuid4())

        if session not in conversations:
            conversations[session] = [
                {
                    "role": "system",
                    "content": (
                        "You are a customer support agent for an e-commerce store. "
                        "Help customers with refunds, order status, and return policies. "
                        "Be professional and empathetic. Always confirm the action taken "
                        "before ending the conversation."
                    ),
                }
            ]

        conversations[session].append({"role": "user", "content": message})

        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=conversations[session],
        )

        content = response.choices[0].message.content
        conversations[session].append({"role": "assistant", "content": content})

        return TaskResult(message=content, session_id=session)

result = Netra.simulation.run_simulation(
    name="Customer Support — All Personas",
    dataset_id="your-dataset-id",
    task=SupportAgentTask(),
    context={"model": "gpt-4o-mini", "agent_version": "v1"},
    max_concurrency=3,
)

print(f"Total scenarios: {result['total_items']}")
print(f"Completed: {len(result['completed'])}")
print(f"Failed: {len(result['failed'])}")

for failure in result["failed"]:
    print(f"  Failed {failure['run_item_id']}: {failure['error']}")

Netra.shutdown()