Class BaseInteraction

internbootcamp/src/base_interaction.py:20–94

import json

class BaseInteraction:
    def __init__(self, config: dict[str, Any], *args, **kwargs):
        self.config = config
        self.name: str = config.get("name", "interaction_agent")  # More general agent default role name
        self._instance_dict: dict[str, Any] = {}

    async def start_interaction(self, instance_id: Optional[str] = None, identity: dict[str, Any] = None, **kwargs) -> str:
        """Create a tool instance.

        Args:
            instance_id: The instance id of the tool.
            identity: The identity of the interaction of a specific task.

        Returns:
            The instance id of the tool.
        """
        # identity类型限制
        if not isinstance(identity, dict):
            try:
                identity = json.loads(identity)
            except:
                print(f"[DEBUG BaseInteraction] Error in start_interaction: identity is not a dict")
                return None
        if instance_id is None:
            instance_id = str(uuid4())
            self._instance_dict[instance_id] = {
                "identity": identity,
            }
            return instance_id
        else:
            if instance_id not in self._instance_dict:
                self._instance_dict[instance_id] = {
                "identity": identity,
            }
            return instance_id

    async def generate_response(
        self, instance_id: str, messages: list[dict[str, Any]], **kwargs
    ) -> tuple[bool, str, float, dict[str, Any]]:  # More clear response generation method
        """
        Generates a response for the current turn of interaction.
        Returns a tuple containing:
        - should_terminate_sequence (bool): True if the interaction sequence should end.
        - response_content (str): The textual content of the response.
        - current_turn_score (float): The score for this specific turn/response.
        - additional_data (dict): Any extra information or metadata.
        """
        should_terminate_sequence: bool = False  # if True, end rollout
        response_content: str = "Your current result seems acceptable."
        current_turn_score: float = 0.8
        additional_data: dict[str, Any] = {}
        return should_terminate_sequence, response_content, current_turn_score, additional_data

    async def calculate_score(self, instance_id: str, **kwargs) -> float:  # More clear score calculation method
        """
        Calculates a score for the interaction,
        potentially considering aspects like partial exposure & in-context task switching.
        should be invoke at turn-level