diff --git a/usersimcrs/simulator/agenda_based/agenda_based_simulator.py b/usersimcrs/simulator/agenda_based/agenda_based_simulator.py index e4659b58..baa93c3b 100644 --- a/usersimcrs/simulator/agenda_based/agenda_based_simulator.py +++ b/usersimcrs/simulator/agenda_based/agenda_based_simulator.py @@ -41,8 +41,12 @@ def __init__( nlg: NLG module generating textual responses. ratings: Historical ratings. """ - super().__init__(id=id, domain=domain, item_collection=item_collection) - self._preference_model = preference_model + super().__init__( + id=id, + domain=domain, + item_collection=item_collection, + preference_model=preference_model, + ) self._interaction_model = interaction_model self._interaction_model.initialize_agenda(self.information_need) self._nlu = nlu diff --git a/usersimcrs/simulator/llm/llm_dual_prompt_user_simulator.py b/usersimcrs/simulator/llm/llm_dual_prompt_user_simulator.py index 4036a57d..911ffbbd 100644 --- a/usersimcrs/simulator/llm/llm_dual_prompt_user_simulator.py +++ b/usersimcrs/simulator/llm/llm_dual_prompt_user_simulator.py @@ -5,6 +5,8 @@ the user response. """ +from typing import Optional + from dialoguekit.core.utterance import Utterance from dialoguekit.participant import DialogueParticipant from usersimcrs.core.simulation_domain import SimulationDomain @@ -19,6 +21,7 @@ UtteranceGenerationPrompt, ) from usersimcrs.simulator.user_simulator import UserSimulator +from usersimcrs.user_modeling.preference_model import PreferenceModel from usersimcrs.user_modeling.persona import Persona @@ -32,7 +35,8 @@ def __init__( item_type: str, task_definition: str = DEFAULT_TASK_DEFINITION, stop_definition: str = DEFAULT_STOP_DEFINITION, - persona: Persona = None, + persona: Optional[Persona] = None, + preference_model: Optional[PreferenceModel] = None, ) -> None: """Initializes the user simulator. @@ -45,14 +49,23 @@ def __init__( stop_definition: Definition of the stop task. Defaults to DEFAULT_STOP_DEFINITION. persona: Persona of the user. Defaults to None. + preference_model: Preference model. Defaults to None. """ - super().__init__(id, domain, item_collection) + super().__init__(id, domain, item_collection, preference_model) self.llm_interface = llm_interface self.generation_prompt = UtteranceGenerationPrompt( - self.information_need, item_type, task_definition, persona + self.information_need, + item_type, + task_definition, + persona, + preference_model, ) self.stop_prompt = StopPrompt( - self.information_need, item_type, stop_definition, persona + self.information_need, + item_type, + stop_definition, + persona, + preference_model, ) def _generate_response(self, agent_utterance: Utterance) -> Utterance: diff --git a/usersimcrs/simulator/llm/llm_single_prompt_user_simulator.py b/usersimcrs/simulator/llm/llm_single_prompt_user_simulator.py index 3d413242..2cf35fab 100644 --- a/usersimcrs/simulator/llm/llm_single_prompt_user_simulator.py +++ b/usersimcrs/simulator/llm/llm_single_prompt_user_simulator.py @@ -4,6 +4,8 @@ model. """ +from typing import Optional + from dialoguekit.core.utterance import Utterance from dialoguekit.participant import DialogueParticipant from usersimcrs.core.simulation_domain import SimulationDomain @@ -14,6 +16,7 @@ UtteranceGenerationPrompt, ) from usersimcrs.simulator.user_simulator import UserSimulator +from usersimcrs.user_modeling.preference_model import PreferenceModel from usersimcrs.user_modeling.persona import Persona @@ -26,7 +29,8 @@ def __init__( llm_interface: LLMInterface, item_type: str, task_definition: str = DEFAULT_TASK_DEFINITION, - persona: Persona = None, + persona: Optional[Persona] = None, + preference_model: Optional[PreferenceModel] = None, ) -> None: """Initializes the user simulator. @@ -37,11 +41,16 @@ def __init__( task_definition: Definition of the task to be performed. Defaults to DEFAULT_TASK_DEFINITION. persona: Persona of the user. Defaults to None. + preference_model: Preference model. Defaults to None. """ - super().__init__(id, domain, item_collection) + super().__init__(id, domain, item_collection, preference_model) self.llm_interface = llm_interface self.prompt = UtteranceGenerationPrompt( - self.information_need, item_type, task_definition, persona + self.information_need, + item_type, + task_definition, + persona, + preference_model, ) def _generate_response(self, agent_utterance: Utterance) -> Utterance: diff --git a/usersimcrs/simulator/llm/prompt/prompt.py b/usersimcrs/simulator/llm/prompt/prompt.py index 619130ab..d0e74899 100644 --- a/usersimcrs/simulator/llm/prompt/prompt.py +++ b/usersimcrs/simulator/llm/prompt/prompt.py @@ -1,10 +1,12 @@ """Interface for prompt.""" from abc import ABC, abstractmethod +from typing import Optional from dialoguekit.core.utterance import Utterance from dialoguekit.participant.participant import DialogueParticipant from usersimcrs.core.information_need import InformationNeed +from usersimcrs.user_modeling.preference_model import PreferenceModel from usersimcrs.user_modeling.persona import Persona @@ -14,7 +16,8 @@ def __init__( information_need: InformationNeed, item_type: str, prompt_definition: str, - persona: Persona = None, + persona: Optional[Persona] = None, + preference_model: Optional[PreferenceModel] = None, ) -> None: """Initializes the prompt. @@ -23,18 +26,66 @@ def __init__( item_type: The type of the item to be recommended. prompt_definition: The definition of the task to be performed. persona: The persona of the user. Defaults to None. + preference_model: Preference model. Defaults to None. """ self.information_need = information_need self.item_type = item_type self.prompt_definition = prompt_definition self.persona = persona + self.preference_model = preference_model self._initial_prompt = self.build_new_prompt() self._prompt_context = "" @property def prompt_text(self) -> str: """Prompt for the user simulator.""" - return self._initial_prompt + "\n" + self._prompt_context + return ( + self._initial_prompt + + "\n" + + self._preference_context + + "\n" + + self._prompt_context + ) + + @property + def _preference_context(self) -> str: + """Returns current preference context for the prompt.""" + if not self.preference_model: + return "" + + preference_summary = self.preference_model.get_preference_summary() + if not preference_summary: + return "" + + return ( + "PREFERENCES: " + f"{preference_summary}\n" + f"{self._preference_context_guidance}\n" + ) + + @property + def _preference_context_guidance(self) -> str: + """Returns the guidance appended after the preference summary.""" + return ( + "Treat these preferences as soft background tastes, not as new " + "requirements." + ) + + @property + def _persona_text(self) -> str: + """Returns a compact persona description.""" + if not self.persona: + return "" + + if self.persona.persona_description: + return self.persona.persona_description + + return ", ".join( + [ + f"{key}={value}" + for key, value in self.persona.characteristics.items() + ] + ) @abstractmethod def build_new_prompt(self, **kwargs) -> str: diff --git a/usersimcrs/simulator/llm/prompt/stop_prompt.py b/usersimcrs/simulator/llm/prompt/stop_prompt.py index c776e27b..5fc14d22 100644 --- a/usersimcrs/simulator/llm/prompt/stop_prompt.py +++ b/usersimcrs/simulator/llm/prompt/stop_prompt.py @@ -1,7 +1,10 @@ """Define the prompt for stopping the conversation.""" +from typing import Optional + from usersimcrs.core.information_need import InformationNeed from usersimcrs.simulator.llm.prompt.prompt import Prompt +from usersimcrs.user_modeling.preference_model import PreferenceModel from usersimcrs.user_modeling.persona import Persona DEFAULT_STOP_DEFINITION = ( @@ -21,7 +24,8 @@ def __init__( information_need: InformationNeed, item_type: str, prompt_definition: str = DEFAULT_STOP_DEFINITION, - persona: Persona = None, + persona: Optional[Persona] = None, + preference_model: Optional[PreferenceModel] = None, ) -> None: """Initializes the prompt. @@ -31,20 +35,37 @@ def __init__( prompt_definition: The definition of the task to be performed. Defaults to DEFAULT_STOP_DEFINITION. persona: The persona of the user. Defaults to None. + preference_model: Preference model. Defaults to None. """ super().__init__( - information_need, item_type, prompt_definition, persona + information_need, + item_type, + prompt_definition, + persona, + preference_model, ) @property def prompt_text(self) -> str: """Prompt for the user simulator.""" + prompt_text = self._initial_prompt + + if self._preference_context: + prompt_text += self._preference_context.rstrip("\n") + + prompt_text += "\n" + + if self._prompt_context: + prompt_text += self._prompt_context.rstrip("\n") + + return prompt_text + "\nCONTINUE: " + + @property + def _preference_context_guidance(self) -> str: + """Returns the guidance appended after the preference summary.""" return ( - self._initial_prompt - + "\n" - + self._prompt_context - + "\n" - + "CONTINUE: " + "Use these preferences only as soft context when deciding " + "whether the conversation is still productive." ) def build_new_prompt(self) -> str: @@ -60,13 +81,7 @@ def build_new_prompt(self) -> str: " Take into account your PERSONA when deciding to stop the " "conversation.\n" ) - stringified_characteristics = ", ".join( - [ - f"{key}={value}" - for key, value in self.persona.characteristics.items() - ] - ) - initial_prompt += f"PERSONA: {stringified_characteristics}\n" + initial_prompt += f"PERSONA: {self._persona_text}\n" initial_prompt += "\nHISTORY:\n" return initial_prompt diff --git a/usersimcrs/simulator/llm/prompt/utterance_generation_prompt.py b/usersimcrs/simulator/llm/prompt/utterance_generation_prompt.py index 6f65a4e6..0c76e31e 100644 --- a/usersimcrs/simulator/llm/prompt/utterance_generation_prompt.py +++ b/usersimcrs/simulator/llm/prompt/utterance_generation_prompt.py @@ -9,10 +9,13 @@ for Task-Oriented Dialog Systems", arXiv 2306.00774. """ +from typing import Optional + from dialoguekit.core.utterance import Utterance from dialoguekit.participant.participant import DialogueParticipant from usersimcrs.core.information_need import InformationNeed from usersimcrs.simulator.llm.prompt.prompt import Prompt +from usersimcrs.user_modeling.preference_model import PreferenceModel from usersimcrs.user_modeling.persona import Persona DEFAULT_TASK_DEFINITION = ( @@ -34,7 +37,8 @@ def __init__( information_need: InformationNeed, item_type: str, prompt_definition: str = DEFAULT_TASK_DEFINITION, - persona: Persona = None, + persona: Optional[Persona] = None, + preference_model: Optional[PreferenceModel] = None, ) -> None: """Initializes the prompt. @@ -44,9 +48,14 @@ def __init__( prompt_definition: The definition of the task to be performed. Defaults to DEFAULT_TASK_DEFINITION. persona: The persona of the user. Defaults to None. + preference_model: Preference model. Defaults to None. """ super().__init__( - information_need, item_type, prompt_definition, persona + information_need, + item_type, + prompt_definition, + persona, + preference_model, ) def build_new_prompt(self) -> str: @@ -61,13 +70,7 @@ def build_new_prompt(self) -> str: initial_prompt += ( " Adapt your responses considering your PERSONA.\n" ) - stringified_characteristics = ", ".join( - [ - f"{key}={value}" - for key, value in self.persona.characteristics.items() - ] - ) - initial_prompt += f"PERSONA: {stringified_characteristics}\n" + initial_prompt += f"PERSONA: {self._persona_text}\n" else: initial_prompt += ( "Be precise with the REQUIREMENTS, clear and concise.\n" @@ -90,6 +93,14 @@ def build_new_prompt(self) -> str: ) return initial_prompt + @property + def _preference_context_guidance(self) -> str: + """Returns the guidance appended after the preference summary.""" + return ( + "Use these preferences to shape what the user accepts, rejects, " + "asks to avoid, or follows up on." + ) + def update_prompt_context( self, utterance: Utterance, participant: DialogueParticipant ) -> None: diff --git a/usersimcrs/simulator/user_simulator.py b/usersimcrs/simulator/user_simulator.py index af238c4e..a8750f1f 100644 --- a/usersimcrs/simulator/user_simulator.py +++ b/usersimcrs/simulator/user_simulator.py @@ -1,6 +1,7 @@ """User simulator abstract class.""" from abc import ABC, abstractmethod +from typing import Optional from dialoguekit.core.annotated_utterance import AnnotatedUtterance from dialoguekit.core.utterance import Utterance @@ -8,6 +9,7 @@ from usersimcrs.core.information_need import generate_random_information_need from usersimcrs.core.simulation_domain import SimulationDomain from usersimcrs.items.item_collection import ItemCollection +from usersimcrs.user_modeling.preference_model import PreferenceModel class UserSimulator(User, ABC): @@ -16,11 +18,13 @@ def __init__( id: str, domain: SimulationDomain, item_collection: ItemCollection, + preference_model: Optional[PreferenceModel] = None, ) -> None: """Initializes the user simulator.""" super().__init__(id, UserType.SIMULATOR) self._domain = domain self._item_collection = item_collection + self.preference_model = preference_model self.get_new_information_need() def get_new_information_need(self) -> None: @@ -53,4 +57,6 @@ def receive_utterance(self, utterance: Utterance) -> None: response = self._generate_response(utterance) if not isinstance(response, AnnotatedUtterance): response = AnnotatedUtterance.from_utterance(response) + if self.preference_model: + self.preference_model.update_from_utterance(response) self._dialogue_connector.register_user_utterance(response) diff --git a/usersimcrs/user_modeling/__init__.py b/usersimcrs/user_modeling/__init__.py index 233b4c33..7de59057 100644 --- a/usersimcrs/user_modeling/__init__.py +++ b/usersimcrs/user_modeling/__init__.py @@ -6,6 +6,9 @@ from usersimcrs.user_modeling.simple_preference_model import ( SimplePreferenceModel, ) +from usersimcrs.user_modeling.structured_preference_model import ( + StructuredPreferenceModel, +) from usersimcrs.user_modeling.persona import Persona __all__ = [ @@ -13,5 +16,6 @@ "PreferenceModel", "PKGPreferenceModel", "SimplePreferenceModel", + "StructuredPreferenceModel", "Persona", ] diff --git a/usersimcrs/user_modeling/persona.py b/usersimcrs/user_modeling/persona.py index 33fc78fd..d60446b5 100644 --- a/usersimcrs/user_modeling/persona.py +++ b/usersimcrs/user_modeling/persona.py @@ -2,12 +2,32 @@ (e.g., age, gender, education), personality types, and behavioral tendencies (e.g., patience, conscientiousness, or curiosity).""" +from __future__ import annotations + from dataclasses import dataclass -from typing import Any, Dict +from typing import Any, Dict, Optional @dataclass class Persona: - """Represents personal user characteristics as key-value pairs.""" characteristics: Dict[str, Any] + persona_description: Optional[str] = None + + @classmethod + def from_config(cls, persona_config: Dict[str, Any]) -> Persona: + """Creates a persona from a configuration dictionary. + + The configuration must provide characteristics under + ``characteristics`` and may include an optional ``persona_description``. + + Args: + persona_config: Persona configuration. + + Returns: + Persona instance. + """ + return cls( + persona_config.get("characteristics", {}), + persona_config.get("persona_description"), + ) diff --git a/usersimcrs/user_modeling/pkg_preference_model.py b/usersimcrs/user_modeling/pkg_preference_model.py index b424aa94..2ec3cec9 100644 --- a/usersimcrs/user_modeling/pkg_preference_model.py +++ b/usersimcrs/user_modeling/pkg_preference_model.py @@ -9,6 +9,8 @@ See: https://github.com/iai-group/UserSimCRS/issues/110 """ +from dialoguekit.core.utterance import Utterance + from usersimcrs.core.simulation_domain import SimulationDomain from usersimcrs.items.item_collection import ItemCollection from usersimcrs.items.ratings import Ratings @@ -45,11 +47,11 @@ def get_item_preference(self, item_id: str) -> float: Args: item_id: Item ID. - Returns: - Item preference, which is in [-1,1]. - Raises: ValueError: If the item does not exist in the collection. + + Returns: + Item preference, which is in [-1,1]. """ self._assert_item_exists(item_id) # TODO: Query PKG to retrieve item preference. @@ -70,3 +72,23 @@ def get_slot_value_preference(self, slot: str, value: str) -> float: # TODO: Query PKG to retrieve slot-value preference. preference = None return preference + + def get_preference_summary(self, max_preferences: int = 10) -> str: + """Returns an empty summary until PKG prompt support is added. + + Args: + max_preferences: Maximum number of preferences to show in the + summary. + + Returns: + Empty summary. + """ + return "" + + def update_from_utterance(self, utterance: Utterance) -> None: + """PKG-backed utterance updates are not implemented yet. + + Args: + utterance: User utterance used to update preferences. + """ + return None diff --git a/usersimcrs/user_modeling/preference_model.py b/usersimcrs/user_modeling/preference_model.py index a64cfe20..042cd804 100644 --- a/usersimcrs/user_modeling/preference_model.py +++ b/usersimcrs/user_modeling/preference_model.py @@ -11,9 +11,10 @@ import random import string from abc import ABC, abstractmethod -from typing import Tuple +from typing import Optional, Tuple import joblib +from dialoguekit.core.utterance import Utterance from usersimcrs.core.simulation_domain import SimulationDomain from usersimcrs.items.item_collection import ItemCollection @@ -37,7 +38,7 @@ def __init__( domain: SimulationDomain, item_collection: ItemCollection, historical_ratings: Ratings, - historical_user_id: str = None, + historical_user_id: Optional[str] = None, ) -> None: """Initializes the preference model of a simulated user. @@ -109,11 +110,11 @@ def get_item_preference(self, item_id: str) -> float: Args: item_id: Item ID. + Raises: + NotImplementedError: If the method is not implemented in a subclass. + Returns: Item preference, which is generally in [-1,1]. - - Raises: - NotImplementedError: If not implemented in derived class. """ raise NotImplementedError @@ -127,11 +128,11 @@ def get_slot_value_preference(self, slot: str, value: str) -> float: slot: Slot name (needs to exist in the domain). value: Slot value. + Raises: + NotImplementedError: If the method is not implemented in a subclass. + Returns: Slot-value preference. - - Raises: - NotImplementedError: If not implemented in derived class. """ raise NotImplementedError @@ -174,6 +175,34 @@ def get_slot_preference(self, slot: str) -> Tuple[str, float]: return value, preference + @abstractmethod + def get_preference_summary(self, max_preferences: int = 10) -> str: + """Returns a compact preference summary for prompts. + + Args: + max_preferences: Maximum number of preferences to show in the + summary. + + Raises: + NotImplementedError: If the method is not implemented in a subclass. + + Returns: + Compact preference summary. + """ + raise NotImplementedError + + @abstractmethod + def update_from_utterance(self, utterance: Utterance) -> None: + """Updates preferences from an utterance if supported by the model. + + Args: + utterance: User utterance used to update preferences. + + Raises: + NotImplementedError: If the method is not implemented in a subclass. + """ + raise NotImplementedError + @classmethod def load_preference_model(cls, path: str) -> PreferenceModel: """Loads preference model from a file. diff --git a/usersimcrs/user_modeling/simple_preference_model.py b/usersimcrs/user_modeling/simple_preference_model.py index acdc2469..6c5bc348 100644 --- a/usersimcrs/user_modeling/simple_preference_model.py +++ b/usersimcrs/user_modeling/simple_preference_model.py @@ -23,6 +23,7 @@ import random +from dialoguekit.core.utterance import Utterance from dialoguekit.participant.user_preferences import UserPreferences from usersimcrs.core.simulation_domain import SimulationDomain @@ -67,11 +68,11 @@ def get_item_preference(self, item_id: str) -> float: Args: item_id: Item ID. - Returns: - Randomly chosen preference, which is either -1 or +1. - Raises: ValueError: If the item does not exist in the collection. + + Returns: + Randomly chosen preference, which is either -1 or +1. """ self._assert_item_exists(item_id) preference = self._item_preferences.get_preference(KEY_ITEM_ID, item_id) @@ -98,3 +99,23 @@ def get_slot_value_preference(self, slot: str, value: str) -> float: preference = random.choice([-1, 1]) self._slot_value_preferences.set_preference(slot, value, preference) return preference + + def get_preference_summary(self, max_preferences: int = 10) -> str: + """Returns an empty summary for non-structured preferences. + + Args: + max_preferences: Maximum number of preferences to show in the + summary. + + Returns: + Empty summary. + """ + return "" + + def update_from_utterance(self, utterance: Utterance) -> None: + """Simple preferences do not update from utterances. + + Args: + utterance: User utterance used to update preferences. + """ + return None diff --git a/usersimcrs/user_modeling/structured_preference/__init__.py b/usersimcrs/user_modeling/structured_preference/__init__.py new file mode 100644 index 00000000..31135ca8 --- /dev/null +++ b/usersimcrs/user_modeling/structured_preference/__init__.py @@ -0,0 +1,15 @@ +"""Helpers for structured preference modeling.""" + +from .heuristic_signal_extractor import HeuristicPreferenceSignalsExtractor +from .preference_memory import PreferenceMemory +from .signal_extractor import ( + PreferenceSignal, + PreferenceSignalsExtractor, +) + +__all__ = [ + "HeuristicPreferenceSignalsExtractor", + "PreferenceMemory", + "PreferenceSignal", + "PreferenceSignalsExtractor", +] diff --git a/usersimcrs/user_modeling/structured_preference/heuristic_signal_extractor.py b/usersimcrs/user_modeling/structured_preference/heuristic_signal_extractor.py new file mode 100644 index 00000000..e38a9331 --- /dev/null +++ b/usersimcrs/user_modeling/structured_preference/heuristic_signal_extractor.py @@ -0,0 +1,204 @@ +"""Heuristic signal extractor for structured preference modeling.""" + +from __future__ import annotations + +import re +from typing import Dict, List, Tuple + +from dialoguekit.core.utterance import Utterance + +from usersimcrs.core.simulation_domain import SimulationDomain +from usersimcrs.items.item_collection import ItemCollection +from usersimcrs.user_modeling.structured_preference.signal_extractor import ( + PreferenceSignal, + PreferenceSignalsExtractor, +) + + +class HeuristicPreferenceSignalsExtractor(PreferenceSignalsExtractor): + """Extracts preference signals with simple text heuristics.""" + + PREFERENCE_SCORE = 0.7 + NEGATIVE_PATTERNS = ( + r"\b(?:avoid|nothing|not|no|without|not into|not too|too much)\b.*" + r"\b{value}\b", + r"\bnot interested in\b.*\b{value}\b", + r"\b(?:anything but|other than|except|rather than)\b.*\b{value}\b", + ) + POSITIVE_PATTERNS = ( + r"\b(?:like|love|prefer|enjoy)\b.*\b{value}\b", + r"\blooking for\b.*\b{value}\b", + r"\binterested in\b.*\b{value}\b", + r"\b(?:with|about|featuring|set in)\b.*\b{value}\b", + r"\bmore\b.*\b{value}\b", + ) + + def __init__( + self, + domain: SimulationDomain, + item_collection: ItemCollection, + ) -> None: + """Initializes the heuristic signal extractor. + + Args: + domain: Simulation domain. + item_collection: Item collection. + """ + self._domain = domain + self._item_collection = item_collection + self._catalog_slot_values = self._build_catalog_slot_values() + self._item_names = self._build_item_names() + + def _build_catalog_slot_values(self) -> Dict[str, List[Tuple[str, str]]]: + """Builds catalog slot values and normalized forms. + + Returns: + Mapping from slot to original and normalized values. + """ + catalog_slot_values: Dict[str, List[Tuple[str, str]]] = {} + for slot in self._domain.get_slot_names(): + if slot.upper() in {"TITLE", "NAME"}: + continue + values = {} + for value in self._item_collection.get_possible_property_values( + slot + ): + if value is None: + continue + normalized_value = self.normalize_preference_value(value) + if len(normalized_value) >= 3: + values[normalized_value] = str(value) + catalog_slot_values[slot] = [ + (value, normalized_value) + for normalized_value, value in sorted( + values.items(), key=lambda item: len(item[0]), reverse=True + ) + ] + return catalog_slot_values + + def _build_item_names(self) -> Dict[str, str]: + """Builds normalized item titles and names. + + Returns: + Mapping from item identifier to normalized item name. + """ + item_names: Dict[str, str] = {} + available_slots = [ + slot + for slot in ("TITLE", "NAME") + if slot in self._domain.get_slot_names() + ] + if not available_slots: + return item_names + + query = ( + f"SELECT id, {', '.join(available_slots)} " + f"FROM {self._item_collection._table_name}" + ) + self._item_collection._cursor.execute(query) + for row in self._item_collection._cursor.fetchall(): + for slot in available_slots: + value = row[slot] + if value is None: + continue + normalized = self.normalize_preference_value(value) + if len(normalized) < 3: + continue + item_names[row["id"]] = normalized + break + return item_names + + def _item_metadata_preference_candidates( + self, item_id: str + ) -> List[Tuple[str, str]]: + """Returns item metadata as slot-value preference candidates.""" + item = self._item_collection.get_item(item_id) + if item is None: + return [] + + slot_values = [] + for slot in self._domain.get_informable_slots(): + value = item.get_property(slot) + if value is None: + continue + values = value if isinstance(value, list) else [value] + for entry in values: + slot_values.append((slot, str(entry))) + return slot_values + + def _score_value_in_text(self, text: str, value: str) -> float: + """Scores a matched value in text. + + Args: + text: Normalized text. + value: Normalized matched value. + + Returns: + Preference score. + """ + contexts = [ + clause.strip() + for clause in re.split(r"[.!?;]", text) + if value in clause + ] + for context in contexts: + if any( + re.search(pattern.format(value=re.escape(value)), context) + for pattern in self.NEGATIVE_PATTERNS + ): + return -self.PREFERENCE_SCORE + if any( + re.search(pattern.format(value=re.escape(value)), context) + for pattern in self.POSITIVE_PATTERNS + ): + return self.PREFERENCE_SCORE + return 0 + + def extract(self, utterance: Utterance) -> List[PreferenceSignal]: + """Extracts preference signals from user text. + + Args: + utterance: User utterance. + + Returns: + Extracted preference signals. + """ + text = self.normalize_preference_value(utterance.text) + if not text: + return [] + signals: List[PreferenceSignal] = [] + + for slot, values in self._catalog_slot_values.items(): + for value, normalized_value in values: + if normalized_value not in text: + continue + score = self._score_value_in_text(text, normalized_value) + if score: + signals.append( + PreferenceSignal( + slot=slot, + value=value, + score=score, + source="attribute", + ) + ) + + for item_id, item_name in self._item_names.items(): + if item_name not in text: + continue + score = self._score_value_in_text(text, item_name) + if score: + for slot, value in self._item_metadata_preference_candidates( + item_id + ): + signals.append( + PreferenceSignal( + slot=slot, + value=value, + score=score, + item_id=item_id, + source="item", + ) + ) + + return signals diff --git a/usersimcrs/user_modeling/structured_preference/preference_memory.py b/usersimcrs/user_modeling/structured_preference/preference_memory.py new file mode 100644 index 00000000..04bef002 --- /dev/null +++ b/usersimcrs/user_modeling/structured_preference/preference_memory.py @@ -0,0 +1,147 @@ +"""Preference memory helpers for structured preference modeling.""" + +from __future__ import annotations + +from typing import Dict, List, Tuple + +from dialoguekit.participant.user_preferences import UserPreferences + + +class PreferenceMemory: + """Stores slot-value preferences and their evidence counts.""" + + def __init__(self, user_id: str) -> None: + """Initializes preference memory. + + Args: + user_id: User identifier. + """ + self._user_id = user_id + self._preferences = UserPreferences(user_id) + self._counts: Dict[Tuple[str, str], float] = {} + + def get(self, slot: str, value: str) -> float | None: + """Returns a stored preference. + + Args: + slot: Slot name. + value: Slot value. + + Returns: + Stored preference or None. + """ + return self._preferences.get_preference(slot, value) + + def set( + self, slot: str, value: str, score: float, count: float | None = None + ) -> None: + """Stores a preference. + + Args: + slot: Slot name. + value: Slot value. + score: Preference score. + count: Optional evidence count. + """ + self._preferences.set_preference(slot, value, score) + if count is not None: + self._counts[(slot, value)] = count + + def increment_towards( + self, slot: str, value: str, score: float, step: float + ) -> None: + """Moves a stored preference toward a new signal. + + Args: + slot: Slot name. + value: Slot value. + score: New signal score. + step: Update step size. + """ + current = self.get(slot, value) + if current is None: + self.set(slot, value, score, 1) + return + + count = self.get_count(slot, value) + weighted_step = step / max(1, count) + updated = current + weighted_step * (score - current) + updated = max(-1.0, min(1.0, updated)) + self.set(slot, value, updated, count + 1) + + def get_count(self, slot: str, value: str) -> float: + """Returns evidence count for a preference. + + Args: + slot: Slot name. + value: Slot value. + + Returns: + Evidence count. + """ + return self._counts.get((slot, value), 0) + + def clear(self) -> None: + """Clears all stored preferences.""" + self._preferences = UserPreferences(self._user_id) + self._counts.clear() + + def items(self) -> List[Tuple[str, str, float, float]]: + """Returns stored preferences. + + Returns: + List of slot, value, score, and evidence count. + """ + preferences = self._preferences._preferences.items() + return [ + (slot, value, score, self.get_count(slot, value)) + for slot, value_preferences in preferences + for value, score in value_preferences.items() + ] + + def ranked_preferences(self) -> List[Tuple[str, str, float]]: + """Returns preferences ranked by evidence count and score.""" + return sorted( + [(slot, value, score) for slot, value, score, _ in self.items()], + key=lambda preference: ( + -self.get_count(preference[0], preference[1]), + -abs(preference[2]), + preference[0], + preference[1], + ), + ) + + def set_weighted_scores( + self, + weighted_sources: List[ + Tuple[Dict[Tuple[str, str], List[float]], float] + ], + preference_threshold: float, + ) -> None: + """Stores weighted scores from historical sources. + + Args: + weighted_sources: Score dictionaries paired with source weights. + preference_threshold: Minimum absolute score to store. + """ + keys: set[Tuple[str, str]] = set() + for scores, _ in weighted_sources: + keys.update(scores) + + for slot, value in keys: + weighted_count = sum( + weight * len(scores.get((slot, value), [])) + for scores, weight in weighted_sources + ) + if weighted_count <= 0: + continue + + score = ( + sum( + weight * sum(scores.get((slot, value), [])) + for scores, weight in weighted_sources + ) + / weighted_count + ) + if abs(score) >= preference_threshold: + self.set(slot, value, score, weighted_count) diff --git a/usersimcrs/user_modeling/structured_preference/signal_extractor.py b/usersimcrs/user_modeling/structured_preference/signal_extractor.py new file mode 100644 index 00000000..25511b50 --- /dev/null +++ b/usersimcrs/user_modeling/structured_preference/signal_extractor.py @@ -0,0 +1,54 @@ +"""Signal extraction interfaces for structured preference modeling.""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from dataclasses import dataclass +from typing import List, Optional + +from dialoguekit.core.utterance import Utterance + + +@dataclass +class PreferenceSignal: + """Represents one extracted preference signal. + + Args: + score: Signal score in [-1, 1]. + slot: Slot name for an explicit attribute signal. + value: Slot value for an explicit attribute signal. + item_id: Item identifier for an item-based signal. + source: Signal source. + """ + + score: float + slot: Optional[str] = None + value: Optional[str] = None + item_id: Optional[str] = None + source: str = "attribute" + + +class PreferenceSignalsExtractor(ABC): + @staticmethod + def normalize_preference_value(value: str) -> str: + """Normalizes a preference value for matching and memory keys. + + Args: + value: Raw preference value. + + Returns: + Normalized preference value. + """ + return " ".join(str(value).lower().replace("-", " ").split()) + + @abstractmethod + def extract(self, utterance: Utterance) -> List[PreferenceSignal]: + """Extracts preference signals from user input. + + Args: + utterance: User utterance. + + Returns: + Extracted preference signals. + """ + raise NotImplementedError diff --git a/usersimcrs/user_modeling/structured_preference_model.py b/usersimcrs/user_modeling/structured_preference_model.py new file mode 100644 index 00000000..481898e5 --- /dev/null +++ b/usersimcrs/user_modeling/structured_preference_model.py @@ -0,0 +1,263 @@ +"""Structured preference model backed by historical ratings and metadata.""" + +from __future__ import annotations + +from collections import defaultdict +from typing import Dict, List, Optional, Tuple + +from dialoguekit.core.dialogue import Dialogue +from dialoguekit.core.utterance import Utterance +from dialoguekit.participant.participant import DialogueParticipant +from dialoguekit.participant.user_preferences import UserPreferences + +from usersimcrs.core.simulation_domain import SimulationDomain +from usersimcrs.items.item import Item +from usersimcrs.items.item_collection import ItemCollection +from usersimcrs.items.ratings import Ratings +from usersimcrs.user_modeling.preference_model import ( + KEY_ITEM_ID, + PreferenceModel, +) +from usersimcrs.user_modeling.structured_preference import ( + PreferenceMemory, + PreferenceSignalsExtractor, +) + + +class StructuredPreferenceModel(PreferenceModel): + + UPDATE_STEP = 0.25 + RATING_HISTORY_WEIGHT = 1.0 + DIALOGUE_HISTORY_WEIGHT = 1.0 + + def __init__( + self, + domain: SimulationDomain, + item_collection: ItemCollection, + signals_extractor: PreferenceSignalsExtractor, + historical_ratings: Optional[Ratings] = None, + historical_user_id: Optional[str] = None, + preference_threshold: float = PreferenceModel.PREFERENCE_THRESHOLD, + dialogue_history: Optional[List[Dialogue]] = None, + rating_history_weight: float = RATING_HISTORY_WEIGHT, + dialogue_history_weight: float = DIALOGUE_HISTORY_WEIGHT, + ) -> None: + """Initializes structured preferences. + + Args: + domain: Domain. + item_collection: Item collection. + signals_extractor: Preference signal extractor. + historical_ratings: Optional historical ratings. Defaults to None. + historical_user_id: Historical user ID. Defaults to None. + preference_threshold: Minimum absolute score to store as a + meaningful preference. Defaults to + `PreferenceModel.PREFERENCE_THRESHOLD`. + dialogue_history: Optional previous dialogues to initialize session + preferences from. Defaults to None. + rating_history_weight: Weight for evidence extracted from + historical ratings. Defaults to 1.0. + dialogue_history_weight: Weight for evidence extracted from + historical dialogues. Defaults to 1.0. + """ + historical_ratings = historical_ratings or Ratings(item_collection) + super().__init__( + domain, item_collection, historical_ratings, historical_user_id + ) + self._preference_threshold = preference_threshold + self._dialogue_history = list(dialogue_history or []) + self._rating_history_weight = rating_history_weight + self._dialogue_history_weight = dialogue_history_weight + self._item_preferences = UserPreferences(self._user_id) + self._long_term_preferences = PreferenceMemory(self._user_id) + self._session_preferences = PreferenceMemory(self._user_id) + self._signals_extractor = signals_extractor + self._initialize_preferences() + + def _initialize_preferences(self) -> None: + """Initializes preferences from historical ratings and dialogues.""" + self._long_term_preferences.set_weighted_scores( + [ + ( + self._rating_preference_scores(), + self._rating_history_weight, + ), + ( + self._dialogue_preference_scores(), + self._dialogue_history_weight, + ), + ], + self._preference_threshold, + ) + + def _rating_preference_scores(self) -> Dict[Tuple[str, str], List[float]]: + """Returns preference scores extracted from historical ratings. + + Returns: + Rating scores keyed by slot-value pair. + """ + rating_scores: Dict[Tuple[str, str], List[float]] = defaultdict(list) + for item_id, rating in self._historical_ratings.get_user_ratings( + self._historical_user_id + ).items(): + item = self._item_collection.get_item(item_id) + if item is None: + continue + if abs(rating) >= self._preference_threshold: + self._item_preferences.set_preference( + KEY_ITEM_ID, item_id, rating + ) + for slot, value in self._item_slot_values(item): + rating_scores[(slot, value)].append(rating) + return rating_scores + + def _dialogue_preference_scores(self) -> Dict[Tuple[str, str], List[float]]: + """Returns preference scores extracted from historical dialogues. + + Returns: + Dialogue scores keyed by slot-value pair. + """ + dialogue_scores: Dict[Tuple[str, str], List[float]] = defaultdict(list) + for dialogue in self._dialogue_history: + for utterance in self._user_utterances(dialogue): + for signal in self._signals_extractor.extract(utterance): + if not signal.slot or signal.value is None: + continue + score = max(-1.0, min(1.0, signal.score)) + dialogue_scores[(signal.slot, signal.value)].append(score) + return dialogue_scores + + def _user_utterances(self, dialogue: Dialogue) -> List[Utterance]: + """Returns user utterances from a dialogue. + + Args: + dialogue: Dialogue to read. + + Returns: + User utterances. + """ + return [ + utterance + for utterance in dialogue.utterances + if utterance.participant is DialogueParticipant.USER + ] + + def _item_slot_values(self, item: Item) -> List[Tuple[str, str]]: + """Returns preference slot values for an item. + + Args: + item: Item to read. + + Returns: + Slot-value pairs from informable item properties. + """ + slot_values = [] + for slot in self._domain.get_informable_slots(): + value = item.get_property(slot) + if value is None: + continue + values = value if isinstance(value, list) else [value] + for entry in values: + slot_values.append((slot, str(entry))) + return slot_values + + def get_item_preference(self, item_id: str) -> float: + """Returns preference score for an item. + + Args: + item_id: Item identifier. + + Returns: + Long-term item preference if available, otherwise 0. + """ + self._assert_item_exists(item_id) + preference = self._item_preferences.get_preference(KEY_ITEM_ID, item_id) + return preference if preference is not None else 0 + + def get_slot_value_preference(self, slot: str, value: str) -> float: + """Returns preference score for a slot-value pair. + + Session preferences are checked first because they capture the current + dialogue context; long-term preferences are used as a fallback. + + Args: + slot: Slot name. + value: Slot value. + + Returns: + Preference score for the slot-value pair, or 0 if unavailable. + """ + self._assert_slot_exists(slot) + preference = self._session_preferences.get(slot, value) + if preference is None: + preference = self._long_term_preferences.get(slot, value) + return preference if preference is not None else 0 + + def update_slot_value_preference( + self, slot: str, value: str, score: float + ) -> None: + """Updates one slot-value preference. + + Args: + slot: Slot name. + value: Slot value. + score: Preference score to store. + """ + self._assert_slot_exists(slot) + self._session_preferences.increment_towards( + slot, + value, + max(-1.0, min(1.0, score)), + self.UPDATE_STEP, + ) + + def get_preference_summary(self, max_preferences: int = 20) -> str: + """Returns a preference summary for the LLM. + + Args: + max_preferences: Total number of preferences. Defaults to 20. + + Returns: + Summary for the LLM of session and long-term preferences. + """ + session_preferences = self._session_preferences.ranked_preferences()[ + : max(1, max_preferences // 2) + ] + long_term_preferences = self._long_term_preferences.ranked_preferences() + positive_preferences = [ + preference + for preference in long_term_preferences + if preference[2] >= self._preference_threshold + ][: max(1, max_preferences // 2)] + negative_preferences = [ + preference + for preference in long_term_preferences + if preference[2] <= -self._preference_threshold + ][: max(1, max_preferences // 2)] + + sections = [ + ("Session preferences", session_preferences), + ("Positive preferences", positive_preferences), + ("Negative preferences", negative_preferences), + ] + return " | ".join( + f"{title}: " + + "; ".join( + f"{slot}={value}:{score:.2f}" + for slot, value, score in preferences + ) + for title, preferences in sections + if preferences + ) + + def update_from_utterance(self, utterance: Utterance) -> None: + """Updates preferences from a user utterance. + + Args: + utterance: User utterance used to update preferences. + """ + for signal in self._signals_extractor.extract(utterance): + if not signal.slot or signal.value is None: + continue + score = max(-1.0, min(1.0, signal.score)) + self.update_slot_value_preference(signal.slot, signal.value, score) diff --git a/usersimcrs/utils/simulation_utils.py b/usersimcrs/utils/simulation_utils.py index 003ffcae..10f3db38 100644 --- a/usersimcrs/utils/simulation_utils.py +++ b/usersimcrs/utils/simulation_utils.py @@ -23,6 +23,12 @@ from dialoguekit.utils.dialogue_reader import json_to_dialogues from usersimcrs.core.simulation_domain import SimulationDomain +from usersimcrs.information_need_management.information_need import ( + generate_random_information_need, +) +from usersimcrs.information_need_management.information_need_tracker import ( + InformationNeedTracker, +) from usersimcrs.items.item_collection import ItemCollection from usersimcrs.items.ratings import Ratings from usersimcrs.nlu.llm.llm_dialogue_act_extractor import ( @@ -30,10 +36,32 @@ ) from usersimcrs.simulator.agenda_based.interaction_model import InteractionModel from usersimcrs.llm_interfaces.llm_interface import LLMInterface +from usersimcrs.user_modeling.preference_model import PreferenceModel from usersimcrs.user_modeling.persona import Persona -from usersimcrs.user_modeling.simple_preference_model import ( - SimplePreferenceModel, -) + + +def _get_information_need_tracker( + config: confuse.Configuration, + domain: SimulationDomain, + item_collection: ItemCollection, +) -> InformationNeedTracker: + """Creates an information need tracker from configuration. + + Args: + config: Configuration of the run. + domain: Domain. + item_collection: Item collection. + + Returns: + Information need tracker. + """ + tracker_config = config["information_need_tracker"].get() + tracker_class = map_path_to_class(tracker_config["class_path"]) + tracker_args = dict(tracker_config.get("args", {})) + tracker_args["information_need"] = generate_random_information_need( + domain, item_collection + ) + return tracker_class(**tracker_args) def map_path_to_class(cls_path: str) -> Type: @@ -149,11 +177,8 @@ def _get_agenda_based_simulator_config( config["historical_ratings_ratio"].get(confuse.Number(default=0.8)) ) - preference_model = SimplePreferenceModel( - domain, - item_collection, - historical_ratings, - historical_user_id="13", + preference_model = _get_preference_model( + config, domain, item_collection, historical_ratings ) # Loads dialogue sample @@ -180,6 +205,9 @@ def _get_agenda_based_simulator_config( return { "preference_model": preference_model, "interaction_model": interaction_model, + "information_need_tracker": _get_information_need_tracker( + config, domain, item_collection + ), "nlu": nlu, "nlg": nlg, "domain": domain, @@ -309,21 +337,35 @@ def _get_llm_single_prompt_user_simulator_config( delimiter=config["csv_delimiter"].get(confuse.String(default=",")), ) + ratings = Ratings(item_collection) + if config["ratings"].get() is not None: + ratings.load_ratings_csv(file_path=config["ratings"].get()) + historical_ratings, _ = ratings.create_split( + config["historical_ratings_ratio"].get(confuse.Number(default=0.8)) + ) + preference_model = _get_preference_model( + config, domain, item_collection, historical_ratings + ) + llm_interface = get_llm_interface(config["llm_interface"].get()) task_definition = config["task_definition"].get() persona = None if "persona" in config: - persona = Persona(config["persona"].get()) + persona = Persona.from_config(config["persona"].get()) return { "domain": domain, "item_collection": item_collection, "llm_interface": llm_interface, "item_type": item_type, + "information_need_tracker": _get_information_need_tracker( + config, domain, item_collection + ), "task_definition": task_definition, "persona": persona, + "preference_model": preference_model, } @@ -344,6 +386,51 @@ def get_llm_interface(config: Dict[str, Any]) -> LLMInterface: return llm_interface +def _get_preference_model( + config: confuse.Configuration, + domain: SimulationDomain, + item_collection: ItemCollection, + historical_ratings: Ratings, +) -> PreferenceModel: + """Initializes the configured preference model.""" + preference_model_config = {} + if "preference_model" in config: + preference_model_config = config["preference_model"].get() + + preference_model_class_path = preference_model_config.get( + "class_path", + "usersimcrs.user_modeling.simple_preference_model." + "SimplePreferenceModel", + ) + preference_model_args = dict(preference_model_config.get("args", {})) + + signals_extractor_config = preference_model_args.get("signals_extractor") + if signals_extractor_config: + signals_extractor_class = map_path_to_class( + signals_extractor_config["class_path"] + ) + signals_extractor_args = dict(signals_extractor_config.get("args", {})) + preference_model_args["signals_extractor"] = signals_extractor_class( + domain=domain, + item_collection=item_collection, + **signals_extractor_args, + ) + + preference_model_class = map_path_to_class(preference_model_class_path) + if not issubclass(preference_model_class, PreferenceModel): + raise TypeError( + f"Preference model class {preference_model_class} must inherit " + "from PreferenceModel." + ) + + return preference_model_class( + domain=domain, + item_collection=item_collection, + historical_ratings=historical_ratings, + **preference_model_args, + ) + + def _get_llm_dual_prompt_user_simulator_config( config: confuse.Configuration, ) -> Dict[str, Any]: