Repository: cpacker/MemGPT
File: letta/agent.py (Python)

Code before:
            )
            for msg in added_messages
        ]
        self._append_to_messages(added_messages_objs)

    def _get_ai_reply(
        self,
        message_sequence: List[Message],
        function_call: str = "auto",
        first_message: bool = False,
        stream: bool = False,  # TODO move to config?
        empty_response_retry_limit: int = 3,
        backoff_factor: float = 0.5,  # delay multiplier for exponential backoff
        max_delay: float = 10.0,  # max delay between retries
    ) -> ChatCompletionResponse:
        """Get response from LLM API with robust retry mechanism."""

        allowed_tool_names = self.tool_rules_solver.get_allowed_tool_names()
        allowed_functions = (
            self.functions if not allowed_tool_names else [func for func in self.functions if func["name"] in allowed_tool_names]
        )

        # For the first message, force the initial tool if one is specified
        force_tool_call = None
        if first_message and self.tool_rules_solver.init_tool_rules:
            assert len(self.tool_rules_solver.init_tool_rules) == 1, "Multiple initial tools not supported"
            force_tool_call = self.tool_rules_solver.init_tool_rules[0].tool_name

        for attempt in range(1, empty_response_retry_limit + 1):
            try:
                response = create(
                    llm_config=self.agent_state.llm_config,
                    messages=message_sequence,
                    user_id=self.agent_state.created_by_id,
                    functions=allowed_functions,
                    functions_python=self.functions_python,
                    function_call=function_call,
                    first_message=first_message,
                    force_tool_call=force_tool_call,
                    stream=stream,
                    stream_interface=self.interface,
                )

                # These bottom two are retryable
                if len(response.choices) == 0 or response.choices[0] is None:
                    raise ValueError(f"API call returned an empty message: {response}")

                if response.choices[0].finish_reason not in ["stop", "function_call", "tool_calls"]:
                    if response.choices[0].finish_reason == "length":
                        # This is not retryable, hence RuntimeError v.s. ValueError
                        raise RuntimeError("Finish reason was length (maximum context length)")

Review comment (refactor):
maybe move this into the create agent function? and we should allow this (not throw an error) if the LLM config specifies OpenAI + one of these structured output models https://platform.openai.com/docs/guides/structured-outputs#supported-models

Code after:
        self.messages_total += len(added_messages)

    def append_to_messages(self, added_messages: List[dict]):
        """An external-facing message append, where dict-like messages are first converted to Message objects"""
        added_messages_objs = [
            Message.dict_to_message(
                agent_id=self.agent_state.id,
                user_id=self.agent_state.created_by_id,
                model=self.model,
                openai_message_dict=msg,
            )
            for msg in added_messages
        ]
        self._append_to_messages(added_messages_objs)

    def _get_ai_reply(
        self,
        message_sequence: List[Message],
        function_call: str = "auto",
        first_message: bool = False,
        stream: bool = False,  # TODO move to config?
        empty_response_retry_limit: int = 3,
        backoff_factor: float = 0.5,  # delay multiplier for exponential backoff
        max_delay: float = 10.0,  # max delay between retries
        step_count: Optional[int] = None,
    ) -> ChatCompletionResponse:
        """Get response from LLM API with robust retry mechanism."""

        allowed_tool_names = self.tool_rules_solver.get_allowed_tool_names()
        allowed_functions = (
            self.functions if not allowed_tool_names else [func for func in self.functions if func["name"] in allowed_tool_names]
        )

        # For the first message, force the initial tool if one is specified
        force_tool_call = None
        if (
            step_count is not None
            and step_count == 0
            and not self.supports_structured_output
            and len(self.tool_rules_solver.init_tool_rules) > 0
        ):
            force_tool_call = self.tool_rules_solver.init_tool_rules[0].tool_name

        for attempt in range(1, empty_response_retry_limit + 1):
            try:
                response = create(
                    llm_config=self.agent_state.llm_config,
                    messages=message_sequence,
                    user_id=self.agent_state.created_by_id,
                    functions=allowed_functions,
                    functions_python=self.functions_python,