转换消息简介

为什么我们需要处理长上下文？这个问题涉及到几个限制和要求：

令牌限制：语言模型有令牌限制，限制了它们可以处理的文本数据量。如果超过这些限制，可能会遇到错误或产生额外的成本。通过预处理聊天历史，我们可以确保保持在可接受的令牌范围内。
上下文相关性：随着对话的进行，保留整个聊天历史可能变得不那么相关甚至适得其反。只保留最近和相关的消息可以帮助语言模型集中于最关键的上下文，从而产生更准确和相关的回复。
效率：处理长上下文可能消耗更多的计算资源，导致响应时间变慢。

转换消息功能

TransformMessages 功能旨在在 LLM 代理处理之前修改传入的消息。这可以包括限制消息数量、截断消息以满足令牌限制等。

要求

安装 pyautogen：

pip install pyautogen

更多信息，请参阅安装指南。

探索和理解转换

让我们从探索可用的转换和理解它们的工作方式开始。我们将首先导入所需的模块。

import copy
import pprint

from autogen.agentchat.contrib.capabilities import transforms

示例 1：限制消息总数

考虑这样一个场景，您希望将上下文历史限制为仅保留最近的消息，以保持效率和相关性。您可以使用 MessageHistoryLimiter 转换来实现这一点：

# 将消息历史限制为最近的 3 条消息
max_msg_transfrom = transforms.MessageHistoryLimiter(max_messages=3)

messages = [
    {"role": "user", "content": "你好"},
    {"role": "assistant", "content": [{"type": "text", "text": "在那里"}]},
    {"role": "user", "content": "你好吗"},
    {"role": "assistant", "content": [{"type": "text", "text": "你好吗？"}]},
    {"role": "user", "content": "非常非常非常非常非常非常长的字符串"},
]

processed_messages = max_msg_transfrom.apply_transform(copy.deepcopy(messages))
pprint.pprint(processed_messages)

[{'content': '你好吗', 'role': 'user'},
{'content': [{'text': '你好吗？', 'type': 'text'}], 'role': 'assistant'},
{'content': '非常非常非常非常非常非常长的字符串', 'role': 'user'}]

通过应用 MessageHistoryLimiter，我们可以看到我们成功将上下文历史限制为最近的 3 条消息。

示例 2：限制令牌数量

为了遵守令牌限制，使用 MessageTokenLimiter 转换。这个转换限制每条消息的令牌数量以及所有消息中的总令牌数量。此外，还可以应用 min_tokens 阈值：

# 将每条消息的令牌限制在3个令牌以内
token_limit_transform = transforms.MessageTokenLimiter(max_tokens_per_message=3, min_tokens=10)

processed_messages = token_limit_transform.apply_transform(copy.deepcopy(messages))

pprint.pprint(processed_messages)

[{'content': '你好', 'role': 'user'},
{'content': [{'text': '在吗', 'type': 'text'}], 'role': 'assistant'},
{'content': '你好', 'role': 'user'},
{'content': [{'text': '你好吗', 'type': 'text'}], 'role': 'assistant'},
{'content': '非常非常非常', 'role': 'user'}]

我们可以看到，我们成功将令牌数量限制在了3个，这相当于这个实例中的3个单词。

在下面的例子中，我们将探讨min_tokens阈值的影响。

short_messages = [
    {"role": "user", "content": "你好，在吗？"},
    {"role": "assistant", "content": [{"type": "text", "text": "你好"}]},
]

processed_short_messages = token_limit_transform.apply_transform(copy.deepcopy(short_messages))

pprint.pprint(processed_short_messages)

[{'content': '你好，在吗？', 'role': 'user'},
 {'content': [{'text': '你好', 'type': 'text'}], 'role': 'assistant'}]

我们可以看到，没有应用任何转换，因为未达到10个令牌的阈值。

使用代理应用转换

到目前为止，我们只测试了MessageHistoryLimiter和MessageTokenLimiter转换，现在让我们测试这些转换与AutoGen的代理一起使用。

设置舞台

import os
import copy

import autogen
from autogen.agentchat.contrib.capabilities import transform_messages, transforms
from typing import Dict, List

config_list = [{"model": "gpt-3.5-turbo", "api_key": os.getenv("OPENAI_API_KEY")}]

# 定义代理；用户代理和助手
assistant = autogen.AssistantAgent(
    "assistant",
    llm_config={"config_list": config_list},
)
user_proxy = autogen.UserProxyAgent(
    "user_proxy",
    human_input_mode="NEVER",
    is_termination_msg=lambda x: "TERMINATE" in x.get("content", ""),
    max_consecutive_auto_reply=10,
)

tip

了解有关为代理配置LLM的更多信息，请点击这里。

我们首先需要编写test函数，通过助手和用户代理之间的消息交换创建一个非常长的聊天记录，并尝试在不清除历史记录的情况下启动新的聊天，可能会触发由于令牌限制而导致的错误。

# 创建一个非常长的聊天历史记录，肯定会导致 gpt 3.5 崩溃
def test(assistant: autogen.ConversableAgent, user_proxy: autogen.UserProxyAgent):
    for _ in range(1000):
        # 定义一个虚假的、非常长的消息
        assitant_msg = {"role": "assistant", "content": "test " * 1000}
        user_msg = {"role": "user", "content": ""}

        assistant.send(assitant_msg, user_proxy, request_reply=False, silent=True)
        user_proxy.send(user_msg, assistant, request_reply=False, silent=True)

    try:
        user_proxy.initiate_chat(assistant, message="绘制并保存一个从 -10 到 10 的 x^2 图形", clear_history=False)
    except Exception as e:
        print(f"遇到了与基础助手相关的错误：\n{e}")

第一次运行将使用默认实现，代理不具备TransformMessages能力。

test(assistant, user_proxy)

运行此测试将由于发送给OpenAI的gpt 3.5的令牌数量过多而导致错误。

user_proxy (to assistant):

绘制并保存从-10到10的x^2图像

--------------------------------------------------------------------------------
遇到基本助手错误
错误代码：429 - {'error': {'message': 'Request too large for gpt-3.5-turbo in organization org-U58JZBsXUVAJPlx2MtPYmdx1 on tokens per min (TPM): Limit 60000, Requested 1252546. The input or output tokens must be reduced in order to run successfully. Visit https://platform.openai.com/account/rate-limits to learn more.', 'type': 'tokens', 'param': None, 'code': 'rate_limit_exceeded'}}

现在让我们给助手添加TransformMessages能力，并运行相同的测试。

context_handling = transform_messages.TransformMessages(
    transforms=[
        transforms.MessageHistoryLimiter(max_messages=10),
        transforms.MessageTokenLimiter(max_tokens=1000, max_tokens_per_message=50, min_tokens=500),
    ]
)
context_handling.add_to_agent(assistant)

test(assistant, user_proxy)

下面的控制台输出显示，代理现在能够处理发送给OpenAI的gpt 3.5的大量令牌。

user_proxy (to assistant):

绘制并保存从-10到10的x^2图像

--------------------------------------------------------------------------------
截断了3804个令牌。令牌从4019个减少到215个
assistant (to user_proxy):

要绘制并保存从-10到10的\( x^2 \)图像，我们可以使用带有matplotlib库的Python。以下是生成图像并将其保存为名为"plot.png"的文件的代码：

```python
# 文件名：plot_quadratic.py
import matplotlib.pyplot as plt
import numpy as np

# 创建一个从-10到10的x值数组
x = np.linspace(-10, 10, 100)
y = x**2

# 绘制图像
plt.plot(x, y)
plt.xlabel('x')
plt.ylabel('x^2')
plt.title('x^2的图像')
plt.grid(True)

# 将图像保存为图像文件
plt.savefig('plot.png')

# 显示图像
plt.show()

您可以在Python环境中运行此脚本。它将生成一个从-10到10的( x^2 )图像，并将其保存为在执行脚本的相同目录中名为"plot.png"的文件。

执行Python脚本以创建并保存图像。执行代码后，您应该在当前目录中看到一个名为"plot.png"的文件，其中包含从-10到10的( x^2 )图像。您可以查看此文件以查看绘制的图像。

还有其他事情您想做或需要帮助吗？如果没有，您可以输入"TERMINATE"来结束我们的对话。

创建自定义转换以处理敏感内容

您可以通过实现MessageTransform协议来创建自定义转换，这提供了处理各种用例的灵活性。一个实际的应用是创建一个自定义转换，从聊天记录或日志中删除敏感信息，例如API密钥、密码或个人数据。这样可以确保机密数据不会被意外泄露，增强您的对话AI系统的安全性和隐私性。

我们将通过实现一个名为MessageRedact的自定义转换来演示这一点，该转换可以检测并删除对话历史中的OpenAI API密钥。当您希望防止API密钥的意外泄露，从而危及系统安全时，这个转换特别有用。

import os
import pprint
import copy
import re

import autogen
from autogen.agentchat.contrib.capabilities import transform_messages, transforms
from typing import Dict, List, Tuple

# 转换必须遵循 transform_messages.MessageTransform 协议。
class MessageRedact:
    def __init__(self):
        self._openai_key_pattern = r"sk-([a-zA-Z0-9]{48})"
        self._replacement_string = "REDACTED"

    def apply_transform(self, messages: List[Dict]) -> List[Dict]:
        temp_messages = copy.deepcopy(messages)

        for message in temp_messages:
            if isinstance(message["content"], str):
                message["content"] = re.sub(self._openai_key_pattern, self._replacement_string, message["content"])
            elif isinstance(message["content"], list):
                for item in message["content"]:
                    if item["type"] == "text":
                        item["text"] = re.sub(self._openai_key_pattern, self._replacement_string, item["text"])
        return temp_messages

    def get_logs(self, pre_transform_messages: List[Dict], post_transform_messages: List[Dict]) -> Tuple[str, bool]:
        keys_redacted = self._count_redacted(post_transform_messages) - self._count_redacted(pre_transform_messages)
        if keys_redacted > 0:
            return f"已屏蔽 {keys_redacted} 个 OpenAI API 密钥。", True
        return "", False

    def _count_redacted(self, messages: List[Dict]) -> int:
        # 统计消息内容中 "REDACTED" 的出现次数
        count = 0
        for message in messages:
            if isinstance(message["content"], str):
                if "REDACTED" in message["content"]:
                    count += 1
            elif isinstance(message["content"], list):
                for item in message["content"]:
                    if isinstance(item, dict) and "text" in item:
                        if "REDACTED" in item["text"]:
                            count += 1
        return count


assistant_with_redact = autogen.AssistantAgent(
    "assistant",
    llm_config=llm_config,
    max_consecutive_auto_reply=1,
)
redact_handling = transform_messages.TransformMessages(transforms=[MessageRedact()])

redact_handling.add_to_agent(assistant_with_redact)

user_proxy = autogen.UserProxyAgent(
    "user_proxy",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=1,
)

messages = [
    {"content": "api key 1 = sk-7nwt00xv6fuegfu3gnwmhrgxvuc1cyrhxcq1quur9zvf05fy"},  # 别担心，这个密钥是随机生成的
    {"content": [{"type": "text", "text": "API key 2 = sk-9wi0gf1j2rz6utaqd3ww3o6c1h1n28wviypk7bd81wlj95an"}]},
]

for message in messages:
    user_proxy.send(message, assistant_with_redact, request_reply=False, silent=True)

result = user_proxy.initiate_chat(
    assistant_with_redact, message="我刚刚提供的两个 API 密钥是什么", clear_history=False

user_proxy (向助手):

我刚刚提供的两个 API 密钥是什么？

--------------------------------------------------------------------------------
已隐去的 2 个 OpenAI API 密钥。
assistant (向 user_proxy):

作为一个 AI，我必须告诉您，公开共享 API 密钥是不安全的，因为它们可以用于访问您的私人数据或产生费用的服务。鉴于您输入的是“已隐去”而不是实际的密钥，看起来您意识到了隐私问题，并且可能正在测试我的回答或模拟交流而不暴露真实凭据，这是出于隐私和安全考虑的良好做法。

直接回答您的问题：您提供的两个 API 密钥都是占位符，用“已隐去”的文本表示，并非真实的 API 密钥。如果这些是真实的密钥，我会再次强调保持它们的安全性，并不会在这里显示它们。

请记住，为了防止未经授权的使用，请保密您的实际 API 密钥。如果您意外地公开了真实的 API 密钥，您应该尽快通过相应服务的 API 管理控制台撤销或重新生成它们。

--------------------------------------------------------------------------------
user_proxy (向助手):

--------------------------------------------------------------------------------
已隐去的 2 个 OpenAI API 密钥。

转换消息简介

转换消息功能​

探索和理解转换​

示例 1：限制消息总数​

示例 2：限制令牌数量​

使用代理应用转换​

设置舞台​

创建自定义转换以处理敏感内容​