创建违禁词过滤服务¶

作为书店老板，您希望在客户提交新的负面评论时，立即在 Slack 频道中收到通知。通过利用 Knative Function，您可以设置一个无服务器函数，其中包含一个简单的违禁词过滤服务，用于判断文本是否包含任何仇恨/侮辱性言论。

我们将学习哪些 Knative 功能？¶

Knative Function 易于使用，可以轻松部署您的服务，并由 Knative Serving 进行管理，这使您能够将服务自动缩放到零，并根据需求进行扩展。

最终交付成果是什么样的？¶

一个正在运行的无服务器 Knative Function，其中包含一个 Python 应用程序，该应用程序以 CloudEvent 的形式接收新的评论，并返回一个结果，告知您的输入文本是否包含任何不当语言。结果以 CloudEvent 的形式发回。

信息

我们正在使用 profanity_check 库来检测文本中的违禁词。这是一个开源库。请参阅此处免责声明。结果可能不是 100% 准确。

该函数的输出将仅为

good
bad

实施¶

这个过程很简单

首先使用 func create 命令生成您的代码模板。
接下来，将您的独特代码整合到此模板中。
最后，执行 func deploy 将您的应用程序无缝部署到 Kubernetes 集群。

此工作流确保了 Knative Functions 生态系统中从开发到部署的顺利过渡。

步骤 1：创建 Knative Function 模板¶

func create -l python bad-word-filter -t cloudevents

验证

文件 tree -a 将如下所示

start/bad-word-filter
.
├── .func
│   └── local.yaml
├── .funcignore
├── .gitignore
├── func.yaml
├── function
│   ├── __init__.py
│   └── func.py
├── pyproject.toml
├── README.md
└── tests
    └── test_func.py

步骤 2：用违禁词过滤逻辑替换生成的代码¶

bad-word-filter/function/func.py 是包含函数代码的文件。您可以用违禁词过滤逻辑替换生成的代码。您可以将以下代码作为起点

bad-word-filter/function/func.py

import logging
from cloudevents.http import CloudEvent
from profanity_check import predict

def new():
    return Function()

class Function:
    async def handle(self, scope, receive, send):
        """ Handle all HTTP requests to this Function. The incoming CloudEvent is in scope["event"]. """
        logging.info("Request Received")

        # 1. Extract the CloudEvent from the scope
        request_event = scope["event"]

        # 2. Extract the data payload from the event, analyze and create CloudEvent
        response_event = self.inappropriate_language_filter(request_event.data)

        # 3. Send the response
        logging.info(f"Sending response: {response_event.data}")
        await send(response_event)

    def create_cloud_event(self, inputText, data):
        attributes = {
            "type": "new-review-comment",
            "source": "book-review-broker",
            "datacontenttype": "application/json",
            "badwordfilter": data,
        }

        data = {"reviewText": inputText, "badWordResult": data}

        return CloudEvent(attributes, data)

    def inappropriate_language_filter(self, text):
        review_text = text.get("reviewText", "")
        profanity_result = predict([review_text])
        result = "good"
        if profanity_result[0] == 1:
            result = "bad"

        return self.create_cloud_event(review_text, result)

步骤 3：配置依赖项¶

bad-word-filter/pyproject.toml 的内容

bad-word-filter/pyproject.toml

[project]
name = "function"
description = ""
version = "0.1.0"
requires-python = ">=3.9"
readme = "README.md"
license = "MIT"
dependencies = [
    "httpx",
    "cloudevents",
    "pytest",
    "pytest-asyncio",
    "alt-profanity-check==1.4.1.post1" # <-- add this dependency
]
authors = [
    { name="Your Name", email="you@example.com"},
]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.pytest.ini_options]
asyncio_mode = "strict"
asyncio_default_fixture_loop_scope = "function"

步骤 4：将函数部署到集群¶

注意

执行以下命令时，请进入 /bad-word-filter。

func deploy -b=s2i -v

验证

预期会看到以下消息

Function deployed in namespace "default" and exposed at URL:
http://bad-word-filter.default.svc.cluster.local

验证¶

func invoke -f=cloudevent --data='{"reviewText":"I love Knative so much"}' -v

验证

预期会收到 CloudEvent 响应

Context Attributes,
specversion: 1.0
type: new-review-comment
source: book-review-broker
id: ebbcd761-3a78-4c44-92e3-de575d1f2d38
time: 2024-05-27T04:44:07.549303Z
datacontenttype: application/json
Extensions,
badwordfilter: good
Data,
{
"reviewText": "I love Knative so much",
"badWordResult": "good"
}

如果您看到响应，则表示函数正在成功运行。

下一步¶

在本教程中，您学习了如何使用 Knative 为一个可以检测文本中不当语言的简单服务创建无服务器函数。

接下来，我们将学习如何使用 Knative Sequence 连接两个机器学习工作流，并确保它们按照您想要的顺序执行。

前往创建 Knative Sequence