_summarize_chat can return a non-string object

[MarianoMolina]

The _summarize_chat method is defined as so:

    def _summarize_chat(
        self,
        summary_method,
        summary_args,
        recipient: Optional[Agent] = None,
        cache: Optional[AbstractCache] = None,
    ) -> str:
...
        if isinstance(summary_method, Callable):
            summary = summary_method(self, recipient, summary_args)
        else:
            raise ValueError(
                "If not None, the summary_method must be a string from [`reflection_with_llm`, `last_msg`] or a callable."
            )
        return summary

Seems that it is expected for registered replies to return a non-string result (this is a snippet from the register_reply docstrings):

def reply_func(
    recipient: ConversableAgent,
    messages: Optional[List[Dict]] = None,
    sender: Optional[Agent] = None,
    config: Optional[Any] = None,
) -> Tuple[bool, Union[str, Dict, None]]:

Moreover, ChatResult seems to expect summary to be a string:

@dataclass
class ChatResult:
    """(Experimental) The result of a chat. Almost certain to be changed."""

    chat_id: int = None
    """chat id"""
    chat_history: List[Dict[str, any]] = None
    """The chat history."""
    summary: str = None
    """A summary obtained from the chat."""
    cost: Dict[str, dict] = None  # keys: "usage_including_cached_inference", "usage_excluding_cached_inference"
    """The cost of the chat.
       The value for each usage type is a dictionary containing cost information for that specific type.
           - "usage_including_cached_inference": Cost information on the total usage, including the tokens in cached inference.
           - "usage_excluding_cached_inference": Cost information on the usage of tokens, excluding the tokens in cache. No larger than "usage_including_cached_inference".
    """
    human_input: List[str] = None
    """A list of human input solicited during the chat."""

To clarify, right now this means _summarize_chat returns a tuple when using a registered reply callable, which is clearly not the expected behavior.

Its a minor thing, but I think it would be a good idea to:

• Check if the summary_method is returning a tuple, and return the second argument if so
• Update the summary return typings to be Union[str, any], or at least Union[str, Dict, None]

[MarianoMolina]

I think part of the issue is that register_reply expects a structure like generate_oai_reply:

    def generate_oai_reply(
        self,
        messages: Optional[List[Dict]] = None,
        sender: Optional[Agent] = None,
        config: Optional[OpenAIWrapper] = None,
    ) -> Tuple[bool, Union[str, Dict, None]]:
        """Generate a reply using autogen.oai."""
        client = self.client if config is None else config
        if client is None:
            return False, None
        if messages is None:
            messages = self._oai_messages[sender]
        extracted_response = self._generate_oai_reply_from_client(
            client, self._oai_system_message + messages, self.client_cache
        )
        return (False, None) if extracted_response is None else (True, extracted_response)

But the summarization logic jumps to use _generate_oai_reply_from_client directly, which looks like this:

def _generate_oai_reply_from_client(self, llm_client, messages, cache) -> Union[str, Dict, None]:

With this in mind, seems like the following changes could make sense:

• _summarize_chat flow should use methods that, like the register_reply structure, return a tuple (bool, Union[str, Dict, None])
• _reflection_with_llm should then use generate_oai_reply instead of _generate_oai_reply_from_client
• _last_msg_as_summary should also work the same way, returning a tuple instead of a string

Then, _summarize_chat returns the second value of the tuple if the first value is true, else returns empty.

[ekzhu]

I think this makes sense. Do you want to create a PR to address the issue?

[MarianoMolina]

Sounds good. What solution do you think makes the most sense? My inclination is to allow the summary parameter to be Union[str, Dict, None].

A little unrelated, but I'm building the logic to have chatmanagers be in charge of creating a json/dict summary of the "task" (the assumption here is that each chat resolution is considered a task, with a set of inputs and a string output), and the idea is that the summary would be a dict that would track the "outcome" of the task/conversation. I'm using guidance to force the grammar and retrieve the values instead of doing some regex retrieval, hoping this will make it consistent. Looks something like this:

def generate_json_summary(recipient, messages: List[Union[str, dict]], sender, config)-> Tuple[bool, Union[str, Dict, None]]:
    # config = recipient.llm_config if config is None else config
    # if config is None:
    #     return False, None
    # TODO: Consolidate the guidance model building with Autogen's model config
    model = guidance_llama_8b_chat()
    if messages is None:
        messages = recipient._oai_messages[sender]
    with system():
        lm = model + recipient.system_message
    for message in messages:
        if isinstance(message, str):
            with user():
                lm += message
        else:
            if message.get("role") == "user":
                with user():
                    lm += message.get("content")
            else:
                with assistant():
                    lm += message.get("content")
    status_options = ["complete", "pending", "failed"]
    code_options = [0,1,2]
    with assistant():
        lm += f"""
        TASK SUMMARY:
        - **status**: {select(status_options, 'status')}
        - **result_code**:  {select(code_options, 'result_code')}
        - **result_description**: {gen('result_description', stop='- **')}
        - **result_content**: 
        ```
        {gen('result_content', stop='- **')}
        ```
        - **result_diagnostic**: {gen('result_diagnostic', max_tokens=200, stop="TERMINATE")}
        TERMINATE
        ---
        """
    response = lm._variables
    if response is None:
        return False, None
    else:
        return True, response

Basically, the ChatManager agent has this callable registered as a reply, with an accompanying system_message to hopefully generate consistent results. The unfortunate part is that (this implementation at least) requires a local model so that guidance can restrict the grammar, but I have a feeling that this can be solved with a slightly different logic.

This is just to say, I think it makes sense for the ChatResult.summary to potentially be a dict.

[ekzhu]

This is just to say, I think it makes sense for the ChatResult.summary to potentially be a dict.

Yes. I think this makes sense. For this feature, it is possible to use summary_method=generate_json_summary to customize initiate_chat. You are welcome to submit this as a notebook PR (see: https://microsoft.github.io/autogen/docs/contributor-guide/documentation#how-to-get-a-notebook-rendered-on-the-website) and add a link page in our Ecosystem (https://microsoft.github.io/autogen/docs/ecosystem) for Guidance.

[ekzhu]

My inclination is to allow the summary parameter to be Union[str, Dict, None]

Do you mean the second value in the tuple returned by the summary method? If so then yes I think it should be Union[str, Dict, None] to match with reply_func.

[MarianoMolina]

Yes, exactly. Cool then, I'll give it a crack.

[MarianoMolina]

I was thinking about the _summarize_chat() logic and was wondering what you thought:

    def _summarize_chat(
        self,
        summary_method,
        summary_args,
        recipient: Optional[Agent] = None,
        cache: Optional[AbstractCache] = None,
    ) -> str:
        """Get a chat summary from an agent participating in a chat.

        Args:
            summary_method (str or callable): the summary_method to get the summary.
                The callable summary_method should take the recipient and sender agent in a chat as input and return a string of summary. E.g,
                ```python
                def my_summary_method(
                    sender: ConversableAgent,
                    recipient: ConversableAgent,
                    summary_args: dict,
                ):
                    return recipient.last_message(sender)["content"]
                ```
            summary_args (dict): a dictionary of arguments to be passed to the summary_method.
            recipient: the recipient agent in a chat.
            prompt (str): the prompt used to get a summary when summary_method is "reflection_with_llm".

        Returns:
            str: a chat summary from the agent.
        """
        summary = ""
        if summary_method is None:
            return summary
        if "cache" not in summary_args:
            summary_args["cache"] = cache
        if summary_method == "reflection_with_llm":
            summary_method = self._reflection_with_llm_as_summary
        elif summary_method == "last_msg":
            summary_method = self._last_msg_as_summary

        if isinstance(summary_method, Callable):
            summary = summary_method(self, recipient, summary_args)
        else:
            raise ValueError(
                "If not None, the summary_method must be a string from [`reflection_with_llm`, `last_msg`] or a callable."
            )
        return summary

    def initiate_chat(
...
        summary = self._summarize_chat(
            summary_method,
            summary_args,
            recipient,
            cache=cache,
        )

Right now, the idea is that:
1- Every chat has a summary
2- There are 2 "default" options: last message or generate oai reply, where an additional prompt is injected to the request but not added to the agent's history.
3- The alternative is that you pass a callable as the summary_method and do something else with the sender, recipient and summary_args

I'm thinking it would be useful for _reflection_with_llm_as_summary and _last_msg_as_summary to have the structure of a reply_func:

def reply_func(
    recipient: ConversableAgent,
    messages: Optional[List[Dict]] = None,
    sender: Optional[Agent] = None,
    config: Optional[Any] = None,
) -> Tuple[bool, Union[str, Dict, None]]:

This means that the "summary_method" callable can have a compatible structure to reply_func. This could look something like this:

    def _summarize_chat(
        self,
        summary_method,
        summary_args,
        recipient: Optional[Agent] = None,
        cache: Optional[AbstractCache] = None,
    ) -> Union[str, Dict, None]:
        """Get a chat summary from an agent participating in a chat.

        Args:
            summary_method (str or callable): the summary_method to get the summary.
                The callable summary_method should take the recipient and sender agent in a chat as input and return a string of summary. E.g,
                ```python
                def my_summary_method(
                    sender: ConversableAgent,
                    recipient: ConversableAgent,
                    summary_args: dict,
                ):
                    return recipient.last_message(sender)["content"]
                ```
            summary_args (dict): a dictionary of arguments to be passed to the summary_method.
            recipient: the recipient agent in a chat.
            prompt (str): the prompt used to get a summary when summary_method is "reflection_with_llm".

        Returns:
            str: a chat summary from the agent.
        """
        summary = ""
        messages = recipient.chat_messages_for_summary(self)
        if summary_method is None:
            return summary
        if "cache" not in summary_args:
            summary_args["cache"] = cache
        if summary_method == "reflection_with_llm":
            summary_method = self._reflection_with_llm_as_summary
            messages = self.add_summary_prompt_message(messages, summary_args)
        elif summary_method == "last_msg":
            summary_method = self._last_msg_as_summary

        if isinstance(summary_method, Callable):
            bool, summary = summary_method(self, messages, recipient)
        else:
            raise ValueError(
                "If not None, the summary_method must be a string from [`reflection_with_llm`, `last_msg`] or a callable."
            )
        return summary
    
    @staticmethod
    def add_summary_prompt_message(messages, summary_args):
        prompt = summary_args.get("summary_prompt")
        prompt = ConversableAgent.DEFAULT_SUMMARY_PROMPT if prompt is None else prompt
        if not isinstance(prompt, str):
            raise ValueError("The summary_prompt must be a string.")
        role = summary_args.get("summary_role", "system")
        if not isinstance(role, str):
            raise ValueError("The summary_role in summary_arg must be a string.")

        system_msg = [
            {
                "role": role,
                "content": prompt,
            }
        ]

        messages = messages + system_msg
        return messages
    @staticmethod
    def _last_msg_as_summary(sender, messages, recipient, config) -> Tuple[bool, Union[str, Dict, None]]:
        """Get a chat summary from the last message of the recipient."""
        summary = ""
        try:
            content = messages[-1]["content"]
            if isinstance(content, str):
                summary = content.replace("TERMINATE", "")
            elif isinstance(content, list):
                # Remove the `TERMINATE` word in the content list.
                summary = "\n".join(
                    x["text"].replace("TERMINATE", "") for x in content if isinstance(x, dict) and "text" in x
                )
        except (IndexError, AttributeError) as e:
            warnings.warn(f"Cannot extract summary using last_msg: {e}. Using an empty str as summary.", UserWarning)
        return (False, None) if summary is None else (True, summary)

    @staticmethod
    def _reflection_with_llm_as_summary(sender, messages, recipient, config) -> Tuple[bool, Union[str, Dict, None]]:
        try:
            if recipient and recipient.client is not None:
                llm_client = recipient.client
            elif sender.client is not None:
                llm_client = sender.client
            else:
                raise ValueError("No OpenAIWrapper client is found.")
            return recipient.generate_oai_reply(llm_client=llm_client, messages=messages, cache=cache)
        except BadRequestError as e:
            warnings.warn(
                f"Cannot extract summary using reflection_with_llm: {e}. Using an empty str as summary.", UserWarning
            )
            summary = ""
        return (False, None) if summary is None else (True, summary)

This would allow a user that builds any reply_func to deploy it with either register_reply or summary_method without making any adjustments. It provides a consistent interface for creating agent response functions.
Issues:

• This, however, does mean that anyone that built a summary function that relied on the summary_args being passed to it would not work anymore. I.e. the summary_method's inputs would go from "sender, recipient, summary_args" to "sender, messages, recipient, config".
• Not sure if an issue or not, but this would mean summary_args is exclusively used for reflection_with_llm, no other flow cares about it
• Would change the logic and structure of several conversableagent methods that people may be actively using.

[ekzhu]

Thanks for the suggestion.

This would allow a user that builds any reply_func to deploy it with either register_reply or summary_method without making any adjustments. It provides a consistent interface for creating agent response functions.

It would be nice to have some usage cases to show why unifying summary method and reply functions would be useful. For now I think we can start with making the return type consistent.

[MarianoMolina]

Yeah, definitely. The key notion I'm working with is that the summary is a flow that happens every single time at the end of a conversation.

But first, lets imagine a reply_func that looks at a conversation, retrieves the last code block in a conversation and saves it to a file in the "functions" dir. For the sake of argument, something like this maybe:

def save_code_to_dir(
    recipient: ConversableAgent,
    messages: Optional[List[Dict]] = None,
    sender: Optional[Agent] = None,
    config: Optional[Any] = None
) -> Tuple[bool, Union[str, Dict, None]]:
    """
    Scans through a list of messages, extracts the last Python code block, finds the name of the first function,
    and saves the block to a file with that function name.

    Args:
    recipient: The agent intended to receive the result of this function.
    messages: A list of message dictionaries.
    sender: The agent who sent the messages.
    config: Optional configuration for processing messages.

    Returns:
    A tuple (success flag, result/message/file path or None).
    """
    if not messages:
        return False, "No messages provided."

    # Search for the last message with a Python code block
    code_block = None
    for message in reversed(messages):
        if 'content' in message and '```python' in message['content']:
            code_block = message['content'].split('```python')[-1].split('```')[0].strip()
            break

    if not code_block:
        return False, "No Python code block found."

    # Extract the first function name using regex
    match = re.search(r'def (\w+)\s*\(', code_block)
    if not match:
        return False, "No function definition found in the code block."

    function_name = match.group(1)
    file_path = os.path.join(config.get('directory', ''), f"{function_name}.py") if config and 'directory' in config else f"{function_name}.py"

    # Write the code to a file named after the function
    try:
        with open(file_path, 'w') as file:
            file.write(code_block)
    except IOError as e:
        return False, f"Failed to write to file: {str(e)}"

    return True, file_path

We can think of a workflow (Code + Run) with the following steps:

• Coder creates a piece of code that meets the prompt details
• Executioner runs the code and if error, Coder iterates until code runs
• Tester creates a few unit tests for the code
• Executioner runs unit tests, if execution fails, tester iterates. If assertions fail, coder iterates.
• If assertions pass, a "Saver" agent is tasked with retrieving the last version of the code and saving it to a file in a specific dir, which adds it to available functions
• With the function added, an agent is tasked with executing a run of the new function with the expectation of being able to do something with its output

In this context, we can think of the "save" function as a function that could be useful to give to an agent in the middle of this conversation flow, but it can also make sense to execute at the end of a workflow where the goal was to create the function itself.

So, we have potentially multiple flows that are useful in different scenarios, that could use the same function at either the end of the chat, or the middle of it.

In essence, the argument is that the summary is just another "reply" we are asking from an agent (GroupChatManager), which can range from a normal LLM response, simply using the last message, or even executing a function where the string generated is the less important part.

I have been thinking about summaries a lot lately, so maybe I'm overthinking it, but in general it seems weird to me that the "summary_method" structure is distinct to the "reply_func" which basically performs the same assignment (reflection_with_llm uses basically the same logic used to create a normal agent response).