代码之家  ›  专栏  ›  技术社区  ›  Neil C. Obremski

如何在Microsoft AutoGen中指定对话历史记录?

  •  0
  • Neil C. Obremski  · 技术社区  · 2 年前

    背景:我正在整合 AutoGen 作为在我的应用程序中推进对话的一种方法(而不是简单地调用OpenAI chat completion 例如直接)。

    如何将我自己的消息注入到user_proxy聊天历史记录中?

    到目前为止,我有以下内容:

    assistant = AssistantAgent(
        "assistant",
        human_input_mode="NEVER",
        llm_config={"config_list": config_list},
        system_message=str(system_message))
    
    user_proxy = UserProxyAgent(
        "user_proxy",
        human_input_mode="NEVER",
        max_consecutive_auto_reply=0,
    )
    
    # TODO: populate chat messages from existing conversation here
    
    user_proxy.initiate_chat(
        assistant,
        clear_history=True,
        message=prompt,
    )
    
    last_message = user_proxy.last_message()
    logger.info("AUTOGEN: last_message: %s", last_message)
    return last_message['content']
    

    注意:我必须设置 max_consecutive_auto_reply=0 否则,AutoGen将永远循环向助手发送空白提示。

    1 回复  |  直到 2 年前
        1
  •  1
  •   Brian Chan    2 年前

    我认为一个肮脏但有效的解决方案是改变 _oai_messages assistant和user_proxy。

    如果你看一下源代码,你会发现它只是传递给自己_oai_system_message+self_oai_消息发送到Openai API。

    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]
    
            # TODO: #1143 handle token limit exceeded error
            response = client.create(
                context=messages[-1].pop("context", None), messages=self._oai_system_message + messages
            )
    
            # TODO: line 301, line 271 is converting messages to dict. Can be removed after ChatCompletionMessage_to_dict is merged.
            extracted_response = client.extract_text_or_completion_object(response)[0]
            if not isinstance(extracted_response, str):
                extracted_response = extracted_response.model_dump(mode="dict")
            return True, extracted_response