Repository: emcie-co/parlant
File: server/tests/engines/alpha/test_guideline_filter.py (Python)

Code before:
        g_name: create_guideline_by_name(context, g_name) for g_name in conversation_guideline_names
    }

    irrelevant_guidelines = [
        conversation_guidelines[g_name]
        for g_name in conversation_guidelines
        if g_name in irrelevant_guideline_names
    ]

    retrieved_guidelines = retrieve_guidelines(context, session_id, conversation_context)

    for guideline in retrieved_guidelines:
        assert guideline not in irrelevant_guidelines


def retrieve_guidelines(
    context: TestContext,
    session_id: SessionId,
    conversation_context: list[tuple[str, str]],
) -> Iterable[Guideline]:
    guideline_store = context.container[GuidelineStore]
    guide_filter = GuidelineFilter()

    interaction_history = [
        create_event_message(context, session_id, cast(roles, r), m)
        for r, m in conversation_context
    ]

    all_possible_guidelines = context.sync_await(
        guideline_store.list_guidelines(guideline_set=context.agent_id)
    )

    retrieved_guidelines = context.sync_await(
        guide_filter.find_relevant_guidelines(
            guidelines=all_possible_guidelines,
            context_variables=[],
            interaction_history=interaction_history,
        )
    )

    return retrieved_guidelines


Review comment (suggestion):
`r, m` are too short in this context. We're getting a list of tuples of two strings each as a parameter. How are we supposed to look at this function and figure out what they mean at a quick glance?

Also, regarding the `roles` variable:
1. As a type alias, it should at least be capitalized, `Roles`
2. The name "roles" is a new term in our terminology—no need to introduce a new term.
3. There already exists a type for this. Look at `EventSource` under `core/sessions.py`

Code after:
from dataclasses import dataclass
from typing import Iterable, cast
from pytest import fixture, mark
from emcie.server.core.guidelines import Guideline, GuidelineId
from emcie.server.core.sessions import Event, EventId, EventSource
from emcie.server.engines.alpha.guideline_filter import GuidelineFilter
from tests.test_utilities import SyncAwaiter
from datetime import datetime, timezone

from emcie.server.core import common


@dataclass
class _TestContext:
    sync_await: SyncAwaiter
    guidelines: list[Guideline]


@fixture
def context(
    sync_await: SyncAwaiter,
) -> _TestContext:
    return _TestContext(sync_await, guidelines=list())


def retrieve_guidelines(
    context: _TestContext,
    conversation_context: list[tuple[str, str]],
) -> Iterable[Guideline]:
    guideline_filter = GuidelineFilter()

    interaction_history = [
        create_event_message(
            offset=i,
            source=cast(EventSource, source),
            message=message,
        )
        for i, (source, message) in enumerate(conversation_context)
    ]

    retrieved_guidelines = context.sync_await(
        guideline_filter.find_relevant_guidelines(
            guidelines=context.guidelines,
            context_variables=[],
            interaction_history=interaction_history,
        )
    )

    return retrieved_guidelines


def create_event_message(
    offset: int,
    source: EventSource,
    message: str,
) -> Event:

    event = Event(
        id=EventId(common.generate_id()),
        source=source,
        type=Event.MESSAGE_TYPE,
        offset=offset,
        data={"message": message},
        creation_utc=datetime.now(timezone.utc),
    )

    return event


def create_guideline_by_name(
    context: _TestContext,
    guideline_name: str,
) -> Guideline:
    guidelines = {
        "check_drinks_in_stock": {
            "predicate": "a client asks for a drink",
            "content": "check if the drink is available in the following stock: "
            "['Sprite', 'Coke', 'Fanta']",
        },
        "check_toppings_in_stock": {
            "predicate": "a client asks for toppings",
            "content": "check if the toppings are available in the following stock: "
            "['Pepperoni', 'Tomatoes', 'Olives']",
        },
        "payment_process": {
            "predicate": "a client is in the payment process",
            "content": "Follow the payment instructions, "
            "which are: 1. Pay in cash only, 2. Pay only at the location.",
        },
        "address_location": {
            "predicate": "the client needs to know our address",
            "content": "Inform the client that our address is at Sapir 2, Herzliya.",
        },
        "mood_support": {
            "predicate": "the client expresses stress or dissatisfaction",
            "content": "Provide comforting responses and suggest alternatives "
            "or support to alleviate the client's mood.",
        },
        "class_booking": {
            "predicate": "the client asks about booking a class or an appointment",
            "content": "Provide available times and facilitate the booking process, "
            "ensuring to clarify any necessary details such as class type, date, and requirements.",
        },
    }

    guideline = Guideline(
        GuidelineId(common.generate_id()),
        predicate=guidelines[guideline_name]["predicate"],
        content=guidelines[guideline_name]["content"],
        creation_utc=datetime.now(timezone.utc),
    )
    context.guidelines.append(guideline)
    return guideline


@mark.parametrize(
    "conversation_context, conversation_guideline_names, relevant_guideline_names",
    [
        (
            [
                ("client", "I'd like to order a pizza, please."),
                ("server", "No problem. What would you like to have?"),
                ("client", "I'd like a large pizza. What toppings do you have?"),
                ("server", "Today, we have pepperoni, tomatoes, and olives available."),
                ("client", "I'll take pepperoni, thanks."),
                (
                    "server",
                    "Awesome. I've added a large pepperoni pizza. "
                    "Would you like a drink on the side?",
                ),
                ("client", "Sure. What types of drinks do you have?"),
                ("server", "We have Sprite, Coke, and Fanta."),
                ("client", "I'll take two Sprites, please."),
                ("server", "Anything else?"),
                ("client", "No, that's all. I want to pay."),
                ("server", "No problem! We accept only cash."),
                ("client", "Sure, I'll pay the delivery guy."),
                ("server", "Unfortunately, we accept payments only at our location."),
                ("client", "So what should I do now?"),
            ],
            [
                "check_toppings_in_stock",
                "check_drinks_in_stock",
                "payment_process",
                "address_location",
            ],
            [
                "payment_process",
            ],
        ),
        (
            [
                (
                    "client",
                    "I'm feeling a bit stressed about coming in. Can I cancel my class for today?",
                ),
                (
                    "server",
                    "I'm sorry to hear that. While cancellation is not possible now, "
                    "how about a lighter session? Maybe it helps to relax.",
                ),
                ("client", "I suppose that could work. What do you suggest?"),
                (
                    "server",
                    "How about our guided meditation session? "
                    "It’s very calming and might be just what you need right now.",
                ),
                ("client", "Alright, please book me into that. Thank you for understanding."),
                (
                    "server",
                    "You're welcome! I've switched your booking to the meditation session. "
                    "Remember, it's okay to feel stressed. We're here to support you.",
                ),
                ("client", "Thanks, I really appreciate it."),
                ("server", "Anytime! Is there anything else I can assist you with today?"),
                ("client", "No, that's all for now."),
                (
                    "server",
                    "Take care and see you soon at the meditation class. "
                    "Our gym is at Sapir 2, Herzliya, in case you need directions.",
                ),
            ],
            [
                "class_booking",
                "mood_support",
                "address_location",
            ],
            [
                "mood_support",
            ],
        ),
    ],
)
def test_that_relevant_guidelines_are_retrieved(
    context: _TestContext,
    conversation_context: list[tuple[str, str]],
    conversation_guideline_names: list[str],
    relevant_guideline_names: list[str],
) -> None:
    conversation_guidelines = {
        name: create_guideline_by_name(context, name) for name in conversation_guideline_names
    }
    relevant_guidelines = [
        conversation_guidelines[name]
        for name in conversation_guidelines
        if name in relevant_guideline_names
    ]

    retrieved_guidelines = retrieve_guidelines(context, conversation_context)

    for guideline in relevant_guidelines:
        assert guideline in retrieved_guidelines


@mark.parametrize(
    "conversation_context, conversation_guideline_names, irrelevant_guideline_names",
    [
        (
            [
                ("client", "I'd like to order a pizza, please."),
                ("server", "No problem. What would you like to have?"),
                ("client", "I'd like a large pizza. What toppings do you have?"),
                ("server", "Today we have pepperoni, tomatoes, and olives available."),
                ("client", "I'll take pepperoni, thanks."),
                (
                    "server",
                    "Awesome. I've added a large pepperoni pizza. "
                    "Would you like a drink on the side?",
                ),
                ("client", "Sure. What types of drinks do you have?"),
                ("server", "We have Sprite, Coke, and Fanta."),
                ("client", "I'll take two Sprites, please."),
                ("server", "Anything else?"),
                ("client", "No, that's all."),
                ("server", "How would you like to pay?"),
                ("client", "I'll pick it up and pay in cash, thanks."),
            ],
            ["check_toppings_in_stock", "check_drinks_in_stock"],
            ["check_toppings_in_stock", "check_drinks_in_stock"],
        ),
        (
            [
                ("client", "Could you add some pretzels to my order?"),
                ("server", "Pretzels have been added to your order. Anything else?"),
                ("client", "Do you have Coke? I'd like one, please."),
                ("server", "Coke has been added to your order."),
                ("client", "Great, where are you located at?"),
            ],
            ["check_drinks_in_stock"],
            ["check_drinks_in_stock"],
        ),
    ],
)
def test_that_irrelevant_guidelines_are_not_retrieved(
    context: _TestContext,
    conversation_context: list[tuple[str, str]],
    conversation_guideline_names: list[str],
    irrelevant_guideline_names: list[str],
) -> None:
    conversation_guidelines = {
        name: create_guideline_by_name(context, name) for name in conversation_guideline_names
    }

    irrelevant_guidelines = [
        conversation_guidelines[name]
        for name in conversation_guidelines
        if name in irrelevant_guideline_names
    ]

    retrieved_guidelines = retrieve_guidelines(context, conversation_context)

    for guideline in retrieved_guidelines:
        assert guideline not in irrelevant_guidelines