LlamaIndex • 2024-01-25

构建一个使用 LlamaIndex、Qdrant 和 Render 进行学习的 Slack 机器人

在这篇文章中，我们将引导您完成构建和部署一个 Slack 机器人的过程，该机器人能够监听您的对话、从中学习，并利用这些知识回答关于您 Slack 工作空间中正在发生的事情的问题。我们还将把它部署到 Render 上的生产环境中！

开始前您需要准备的东西

对 LlamaIndex 有初步了解。如果您不了解，我们文档中的入门教程会提供您理解本教程所需的所有知识，并且只需几分钟。
具备 Python 的基础知识，并已安装 Python 3.11 或更高版本
一个您可以安装应用的 Slack 工作空间（您需要是管理员）
在您的本地机器上克隆了我们的 Slack 机器人仓库。本文中我们将引用该仓库中的文件。

步骤 1：创建一个 Slack 应用，并将其安装到您的工作空间

这是最复杂的一步，因为 Slack 对权限非常挑剔。

您的 Slack 机器人的第一个版本大约只有 20 行代码。它所做的就是提供一个 Slack 用来验证您的应用是否可用的“challenge”端点。您可以在仓库中看到这个代码文件 1_flask.py。让我们来逐步讲解。

首先我们引入依赖项。如果您还没有安装，需要使用 pip 或 poetry 进行安装。

from flask import Flask, request, jsonify

现在我们将创建您的 flask 应用并进行设置，以便它可以在开发环境中运行。

flask_app = Flask(__name__)

if __name__ == "__main__":
    flask_app.run(port=3000)

在这些代码行之间，我们将添加基本路由：如果收到包含带有 challenge 键的 JSON 对象的 POST 请求，我们将返回该键的值。否则，我们将不做任何处理。

@flask_app.route("/", methods=["POST"])
def slack_challenge():
    if request.json and "challenge" in request.json:
        print("Received challenge")
        return jsonify({"challenge": request.json["challenge"]})
    else:
        print("Got unknown request incoming")
        print(request.json)
    return

让您的应用可供 Slack 访问

为了配置 Slack 应用，它需要在 Slack 可以访问的地方运行。所以让我们运行我们的 Slack 应用

python 1_flask.py

我们将使用 ngrok 设置它，以便全世界都可以访问。您需要下载并安装 ngrok 来完成这一步。安装完成后，运行以下命令，以便它能找到运行在 3000 端口的我们的应用

ngrok http 3000

ngrok 会提供一个 HTTPS URL，例如 https://1bf6-64-38-189-168.ngrok-free.app。记下它，因为我们需要把它提供给 Slack。同时请记住，如果您停止并再次启动 ngrok，这个 URL 会改变，您需要重新告知 Slack。您只需要在开发期间使用它。

向 Slack 注册您的应用

访问 Slack API 网站并点击“Create New App”（创建新应用）。您会看到一个类似这样的屏幕，您需要选择“from scratch”（从头开始）

选择一个友好易记的名称，并选择您要安装它的工作空间。您会看到一个类似这样的屏幕

接下来您需要设置您的应用所需的权限。点击右下角的“Permissions”（权限）链接

这将带您进入“scopes”（作用域）屏幕，您需要添加图片中显示的所有作用域，具体是

channels:read — 让您的应用查看可用的频道
channels:join — 让您的应用加入频道
channels:history — 让您的应用查看频道中的历史消息
chat:write — 让您的应用发送消息
users:read — 让您的应用查看用户的名称

保存这些作用域后，向上滚动到“Install to workspace”（安装到工作空间）来安装您的应用。

现在您需要告诉 Slack 您的应用在哪里，以便接收来自 Slack 的消息。点击左侧导航栏中的“Event Subscriptions”（事件订阅）链接，并填写，使其看起来像这样，具体来说

将您的 Request URL 设置为 ngrok 之前提供给您的那个 URL
订阅 message.channels 事件

如果您的应用正在运行并且 ngrok 正确地建立了隧道连接，您的 Request URL 应该显示已验证（Verified）。

呼！这真不少。您的 Slack 应用现在已经注册了，Slack 会向它发送消息。但是要收到这些消息，您必须告诉它加入一个频道。

步骤 2：加入一个频道，并回复消息

为了做到这一点，我们需要扩展我们的应用。您可以在 2_join_and_reply.py 中看到这一步的最终结果。让我们逐步讲解我们添加的内容

import dotenv, os
dotenv.load_dotenv()

我们需要一些环境变量，所以您需要添加这些行并安装 python-dotenv。您还需要在项目根目录下创建一个 .env 文件，包含三个值

OPENAI_API_KEY：您的 OpenAI API 密钥。现在还不需要用到，但您不妨现在就获取。
SLACK_BOT_TOKEN：您可以在 Slack 应用的“OAuth and Permissions”（OAuth 和权限）部分找到它。
SLACK_SIGNING_SECRET：您可以在 Slack 应用的“Basic Information”（基本信息）部分找到它。

我们将使用 Slack 方便的 Python SDK 构建我们的应用，所以请 pip 安装 slack-bolt，然后更新所有的导入语句

from slack_bolt import App
from flask import Flask, request, jsonify
from slack_bolt.adapter.flask import SlackRequestHandler

现在使用我们刚刚设置的那些秘密信息初始化一个 Slack Bolt 应用

app = App(
    token=os.environ.get("SLACK_BOT_TOKEN"),
    signing_secret=os.environ.get("SLACK_SIGNING_SECRET")
)
handler = SlackRequestHandler(app)

要监听消息，机器人必须加入一个频道。您可以让它加入任何和所有公共频道，但为了测试目的，我创建了一个名为 #bot-testing 的频道，它在这里加入的就是这个频道

channel_list = app.client.conversations_list().data
channel = next((channel for channel in channel_list.get('channels') if channel.get("name") == "bot-testing"), None)
channel_id = channel.get('id')
app.client.conversations_join(channel=channel_id)

app.client 是 Bolt 框架的 Slack WebClient，因此您可以直接在框架内完成 WebClient 可以做的任何事情。此处最后添加的是一个非常简单的消息监听器

@app.message()
def reply(message, say):
    print(message)
    say("Yes?")

在 Bolt 框架中，@app.message 装饰器告诉框架在收到消息事件时触发此方法。say 参数是一个函数，它会将消息发送回消息来源的频道。因此，这段代码将在每次收到消息时向频道发送一条回复“Yes?”的消息。

让我们来试试！停止运行 1_flask.py，改为运行 python 2_join_and_reply.py。您无需重新启动 ngrok，它会像以前一样继续将消息发送到 3000 端口。这是我尝试的结果

成功了！我们现在有一个非常烦人的机器人，它会回复任何人说的任何一句话。我们可以做得更好！

步骤 3：只回复提及机器人的消息

表面上看这是一个相当简单的改动，但 Slack 的入站消息格式有点复杂，所以我们不得不添加相当多的代码。您可以在 3_reply_to_mentions.py 中看到最终结果。

首先，为了判断何时提到了我们的机器人，我们需要机器人的用户 ID。在底层，Slack 不使用用户名甚至 @-handles，而是使用一个跨所有 Slack 安装的全局唯一 ID。我们必须获取它

auth_response = app.client.auth_test()
bot_user_id = auth_response["user_id"]

现在我们添加了一段令人烦恼的复杂代码，它解析 Slack 的消息对象，查看入站消息中提到了哪个用户。如果是提到了机器人，机器人就回复，否则就忽略该消息。随着我们深入，我们将把发给机器人的消息视为“查询”，把其他任何消息视为要存储的“事实”，但我们暂时还不会进行存储。

@app.message()
def reply(message, say):
    if message.get('blocks'):
        for block in message.get('blocks'):
            if block.get('type') == 'rich_text':
                for rich_text_section in block.get('elements'):
                    for element in rich_text_section.get('elements'):
                        if element.get('type') == 'user' and element.get('user_id') == bot_user_id:
                            for element in rich_text_section.get('elements'):
                                if element.get('type') == 'text':
                                    query = element.get('text')
                                    print(f"Somebody asked the bot: {query}")
                                    say("Yes?")
                                    return
    # otherwise do something else with it
    print("Saw a fact: ", message.get('text'))

呼。花了一段时间才弄好！但现在我们的机器人只在被提及时代回复了

步骤 4：使用 LlamaIndex 存储事实并回答问题

我们已经到了第 4 步，但还没有用 LlamaIndex 做任何事情！但现在是时候了。在 4_incremental_rag.py 中，您将看到一个简单的命令行 Python 脚本演示，该脚本使用 LlamaIndex 存储事实并回答问题。我不会逐行讲解（脚本中有有用的注释），但让我们看看重要的部分。记住要 pip install llama-index！

首先我们创建一个新的 VectorStoreIndex，这是一个内存中的向量存储，我们将在这里存储我们的事实。开始时它是空的。

index = VectorStoreIndex([])

接下来我们创建 3 个 Document 对象，并将它们分别插入到我们的索引中。真实的文档可以是巨大的文本块、整个 PDF，甚至是图像，但这只是简单的、Slack 消息大小的事实。

doc1 = Document(text="Molly is a cat")
doc2 = Document(text="Doug is a dog")
doc3 = Document(text="Carl is a rat")

index.insert(doc1)
index.insert(doc2)
index.insert(doc3)

最后，我们从索引中创建一个查询引擎并向它提问

# run a query
query_engine = index.as_query_engine()
response = query_engine.query("Who is Molly?")
print(response)

结果是“Molly is a cat”加上大量的调试信息，因为我们在 4_incremental_rag.py 中开启了详细的调试模式。您可以看到我们发送给 LLM 的提示、它从索引中检索到的上下文，以及它生成并发送给我们的回复。

步骤 5：使用 LlamaIndex 在 Slack 中存储事实并回答问题

在 5_rag_in_slack.py 中，我们将之前做过的两件事结合起来：脚本 3（回复查询）和脚本 4（存储事实并回答问题）。我们再次不会逐行讲解，但这里是一些重要的改动

首先，如果您还没有安装，请 pip install llama-index，并引入依赖项。同时初始化您的索引

from llama_index import VectorStoreIndex, Document

index = VectorStoreIndex([])

之前我们只是简单地回复“Yes?”（第 73 行），现在我们将向查询引擎发送一个查询并回复其结果

query = element.get('text')
query_engine = index.as_query_engine()
response = query_engine.query(query)
say(str(response))

之前我们只是记录我们看到一个事实（第 82 行），现在我们将其存储到索引中

index.insert(Document(text=message.get('text')))

结果是，一个能够回答关于它被告知的事情的 Slack 机器人诞生了

太棒了！您可以轻松想象一个监听所有人对话并能回答关于几周或几个月前人们说过的事情的机器人，这能节省每个人搜索旧消息的时间和精力。

步骤 6：持久化我们的记忆

然而，我们的机器人有一个致命的缺陷：索引只存储在内存中。如果我们重启机器人，它就会忘记一切

在 6_qdrant.py 中，我们引入了 Qdrant，这是一个开源的本地向量数据库，它将这些事实存储在磁盘上。这样，如果我们重启机器人，它就能记住之前说过的话。pip install qdrant-client 并引入一些新的依赖项

import qdrant_client
from llama_index.vector_stores.qdrant import QdrantVectorStore

现在我们将初始化 Qdrant 客户端，将其附加到存储上下文，并在初始化索引时将该存储上下文提供给它

client = qdrant_client.QdrantClient(
    path="./qdrant_data"
)
vector_store = QdrantVectorStore(client=client, collection_name="slack_messages")
storage_context = StorageContext.from_defaults(vector_store=vector_store)

index = VectorStoreIndex([],storage_context=storage_context)

这一步就完成了！您的机器人现在可以在重启后继续工作，并且还记得我把“Doug”打成了“Dough”，而且懒得为截图修改过来

步骤 7：让最近的消息更重要

我们现在有了一个相当有能力的机器人！但它有一个微妙的问题：人们可能会说冲突的话，而且它无法判断谁是“对”的，比如当我改变主意决定狗狗应该叫什么名字时

在真实的 Slack 对话中，随着情况的发展，人们可能会从说一个项目“正在计划中”转变为“正在进行中”，再到“已启动”。所以我们需要一种方法来告诉机器人，最近的消息比旧的消息更重要。

为了实现这一点，我们不得不进行相当多的重构，最终结果可以在 7_recency.py 中看到。首先我们需要一堆新的依赖项

import datetime, uuid
from llama_index.schema import TextNode
from llama_index.prompts import PromptTemplate
from llama_index.postprocessor import FixedRecencyPostprocessor
from llama_index import set_global_handler

为了让最近的消息更重要，我们必须知道消息发送的时间。为此，我们将停止向索引中插入 Documents，而是插入 Nodes，并将时间戳作为元数据附加到节点上（在底层，我们的 Documents 无论如何都会被转换为 Nodes，所以这并不会改变太多）

dt_object = datetime.datetime.fromtimestamp(float(message.get('ts')))
formatted_time = dt_object.strftime('%Y-%m-%d %H:%M:%S')

# get the message text
text = message.get('text')
# create a node with metadata
node = TextNode(
    text=text,
    id_=str(uuid.uuid4()),
    metadata={
        "when": formatted_time
    }
)
index.insert_nodes([node])

我还将回复逻辑从消息处理中提取出来，放到它自己的函数 answer_question 中，只是为了让代码更易读。我们要改变的第一件事是我们提供给 LLM 的提示：我们必须告诉它最近的消息很重要。为此我们创建一个提示模板

template = (
    "Your context is a series of chat messages. Each one is tagged with 'who:' \n"
    "indicating who was speaking and 'when:' indicating when they said it, \n"
    "followed by a line break and then what they said. There can be up to 20 chat messages.\n"
    "The messages are sorted by recency, so the most recent one is first in the list.\n"
    "The most recent messages should take precedence over older ones.\n"
    "---------------------\n"
    "{context_str}"
    "\n---------------------\n"
    "You are a helpful AI assistant who has been listening to everything everyone has been saying. \n"
    "Given the most relevant chat messages above, please answer this question: {query_str}\n"
)
qa_template = PromptTemplate(template)

使用 LLM 的有趣之处在于，您经常只需要用英语描述您正在做的事情，然后将描述发送给 LLM 即可。提示模板将自动从查询引擎获取 context_str 和 query_str。但是我们必须像这样在我们的查询引擎上设置这个模板

query_engine.update_prompts(
    {"response_synthesizer:text_qa_template": qa_template}
)

现在我们要改变另外两件事。我们将获取从向量存储中得到的结果，并按新近度对其进行排序，LlamaIndex 有一个内置类可以实现这一点。它叫做 FixedRecencyPostprocessor。我们告诉它存储时间戳的键（我们之前在节点上定义的，见上文）以及它应该返回多少结果

postprocessor = FixedRecencyPostprocessor(
    top_k=20, 
    date_key="when", # the key in the metadata to find the date
    service_context=ServiceContext.from_defaults()
)

然后我们需要创建一个带有附加后处理器的查询引擎

query_engine = index.as_query_engine(similarity_top_k=20, node_postprocessors=[postprocessor])

与此同时，我们做了最后一件事，那就是传递 similarity_top_k=20，这意味着向量存储会给我们 20 条 Slack 消息作为上下文（默认是 2，因为通常 Node 中的文本块要大得多）。

瞧！现在机器人知道将最近的陈述视为事实了。

步骤 8：完善剩下的功能

现在这个机器人运行得相当不错，但我在构建它时玩得很开心，一时兴起又添加了两个功能

我添加了关于谁在说话的元数据，而不仅仅是时间，这样机器人就可以回答诸如“Logan 对这个项目说了什么？”的问题了
我的同事们与机器人交互时，试图像我们彼此交流一样在某个话题（thread）中提出后续问题。所以我添加了一种方式让机器人理解它正处于一个话题中，并将话题中的回复视为后续问题，即使用户没有直接提及机器人