企业官网建设流程全解析

• 文档
• 教程
• 云原生

本文是 Knative Bookstore 端到端教程（欢迎页）的第三个实战环节：书店老板希望在客户提交含侮辱性言论的差评时能及时收到通知，本教程将手把手带你用 Knative Function 编写一个基于 CloudEvent 的不文明用语（bad word）过滤服务。读完本文，你将掌握func create生成模板、编写 Python 过滤逻辑、配置pyproject.toml依赖、用func deploy -b=s2i部署到集群，并用func invoke验证完整的事件流，同时理解 Knative Serving 自动缩容到零背后的原理。

本次实验涉及哪些 Knative 能力

• Knative Function：以极低的上手成本把 Python 业务代码打包成无状态、事件驱动的函数；
• Knative Serving：函数部署后由 Knative Service 托管，具备自动缩容到零（scale-to-zero）与按需扩容能力；
• CloudEvents：函数以 CloudEvent 作为输入与输出，与上游 Broker、下游 Sequence 无缝对接。

最终交付物是什么

最终运行的是一个 Serverless 的 Knative Function：它接收包含新评论的 CloudEvent，解析其中的评论文本，判断是否含有不当语言，再把结果以 CloudEvent 的形式回传。函数的输出只会有两种取值：

• good
• bad

!!! info "关于检测库的说明" 本教程使用开源库profanity_check（Python 包名为alt-profanity-check）检测文本中的脏话。该库的检测结果并非 100% 准确，请阅读其免责声明后合理使用；若检测不准确，可替换为你自己的模型或规则。

实现流程总览

整个过程非常直接：

1. 使用func create命令生成代码模板；
2. 在模板中写入你的过滤逻辑；
3. 执行func deploy将应用无缝部署到 Kubernetes 集群。

这条工作流保证了从开发到部署在 Knative Functions 生态内的平滑衔接。仓库中已有完整可运行的参考实现，位于 code-samples/eventing/bookstore-sample-app/solution/ML-bad-word-filter，下文步骤均可在该目录下对照验证。

Step 1：创建 Knative Function 模板

在终端执行下面的命令，创建一个接收 CloudEvent 的 Python 函数：

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

• -l python：指定语言运行时为 Python；
• bad-word-filter：函数（项目）名，同时也是后续部署的 Knative Service 名称；
• -t cloudevents：使用 CloudEvents 模板，生成带scope["event"]解析能力的函数骨架。

???+ success "验证模板结构" 执行tree -a应该看到如下文件结构：

```txt start/bad-word-filter . ├── .func │ └── local.yaml ├── .funcignore ├── .gitignore ├── func.yaml ├── function │ ├── __init__.py │ └── func.py ├── pyproject.toml ├── README.md └── tests └── test_func.py ``` 其中 `func.yaml` 是函数的元数据描述文件。参考实现的 [func.yaml](https://link.gitcode.com/i/2f9e9efb164aa5176fb96c6ecb6d9f37) 显示，模板会生成 `specVersion`、`name`、`runtime`、`namespace`、`created` 时间戳、`invoke: cloudevent` 以及默认的 `build.builder: s2i` 配置；`created` 字段与时间戳是函数被识别为合法 Knative Function 的关键，请勿删除。

Step 2：用过滤逻辑替换生成的代码

bad-word-filter/function/func.py是存放函数代码的文件。将生成的内容替换为下面的不良词汇过滤逻辑：

???+ abstract "bad-word-filter/function/func.py"

```python 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) ```

这段代码的关键点：

• new()入口：Knative Python Function 要求实现new()方法返回函数实例，参考实现 ML-bad-word-filter/function/func.py 与仓库模板说明一致；函数类还可选实现start()、stop()、alive()/ready()等生命周期与健康检查钩子；
• handle(scope, receive, send)：所有 HTTP 请求都会进入这里，入站 CloudEvent 存放在scope["event"]中；
• 数据提取：request_event.data是评论 JSON，通过text.get("reviewText", "")取评论文本；
• 检测逻辑：predict([review_text])返回列表，1表示含有不当语言，因此把结果置为bad，否则为good；
• 响应构造：create_cloud_event保留了type: new-review-comment、source: book-review-broker，并新增badwordfilter扩展属性；datacontenttype设为application/json，负载包含reviewText与badWordResult两个字段。保留原type与source是有意为之——下游 Sequence 和 Trigger 会按ce-type: new-review-comment继续匹配事件。

仓库还提供了配套的单元测试 test_func.py：它构造一个测试 CloudEvent，调用new()实例化函数并执行handle，断言send回调被触发且收到的是合法的 CloudEvent 对象，可作为本地回归验证的参考。

Step 3：配置依赖

bad-word-filter/pyproject.toml是函数的项目配置，Knative Function 在构建时会自动安装这里声明的依赖。在dependencies数组中追加alt-profanity-check：

???+ abstract "bad-word-filter/pyproject.toml"

```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" ```

说明：

• httpx与cloudevents：函数作为 ASGI 应用（仓库测试注释指出由 hypercorn 提供服务）与 CloudEvents 协议支持所必需；
• alt-profanity-check==1.4.1.post1：固定版本以保证构建可复现（参考实现 ML-bad-word-filter/pyproject.toml 使用了完全一致的版本号）；
• [tool.pytest.ini_options]声明了 pytest-asyncio 的严格异步模式，与上面测试文件中的@pytest.mark.asyncio配合使用。

Step 4：把函数部署到集群

!!! note 执行下面的命令前，请先进入函数目录：cd /bad-word-filter。

func deploy -b=s2i -v

• -b=s2i：指定使用 source-to-image（s2i）构建器，另一种可选方案是pack构建器；
• -v：输出详细的构建与部署日志。

func deploy的本质是：为函数构建 OCI 容器镜像并推送到你的镜像仓库，然后以Knative Service的形式部署到集群；重复部署会更新镜像与集群上对应的 Service（参见仓库 部署指南 与 部署过程说明）。部署前请先通过export FUNC_REGISTRY=<你的容器仓库>配置镜像仓库地址；若使用私有仓库，还可通过环境变量FUNC_USERNAME/FUNC_PASSWORD注入认证信息，这比命令行--password参数更安全（避免被ps进程列表泄露）。

???+ success "验证部署结果" 预期看到如下输出：

```sh Function deployed in namespace "default" and exposed at URL: http://bad-word-filter.default.svc.cluster.local ``` 也可以用 `kubectl get kservice` 查看 Knative Service 的状态与访问地址： ```sh NAME URL LATESTCREATED LATESTREADY READY REASON bad-word-filter http://bad-word-filter.default.svc.cluster.local bad-word-filter-00001 bad-word-filter-00001 True ```

验证：调用函数并检查 CloudEvent 响应

部署完成后，使用func invoke向集群上的函数直接发送一个 CloudEvent 测试请求：

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

• -f=cloudevent：指定数据格式为 CloudEvent（另一种取值是HTTP）；
• --data：输入的事件数据负载；
• -v：显示详细输出。

???+ success "验证响应" 预期收到一个 CloudEvent 格式的响应：

```sh 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" } ```

如果看到上述响应，说明函数已成功运行："I love Knative so much"不含不当语言，因此badwordfilter扩展属性与badWordResult字段均为good。若输入包含脏话，则这两个位置都会变成bad。

延伸：自动缩容到零与本地调试

• 自动缩容到零：函数以 Knative Service 方式部署后，kubectl get pods会先看到运行中的 Pod；如果一段时间没有请求，Knative Serving 的自动扩缩器会把副本缩到0 个 Pod，再收到请求时又能迅速拉起到处理规模。这正是本实验部署方式（而非普通 Deployment）带来的核心收益。更多细节可参考 Serving 自动扩缩 相关文档。
• 本地运行与调试（可选）：先用func build -b=s2i -v构建镜像，再执行func run -b=s2i -v在本地 8080 端口运行函数，随后同样用func invoke -f=cloudevent --data='...'模拟 CloudEvent 做本地联调；func run会自动按需构建，也可用--build强制重建。如果在非函数目录执行命令，会收到 “does not contain an initialized function” 错误，此时用pwd确认当前目录、并检查func.yaml中created字段与时间戳是否存在即可（详见 运行函数说明）。

下一步：接入 Knative Sequence

至此，你已经掌握了用 Knative Function 编写并部署一个能够检测不当语言的 Serverless 服务。在本教程中，这个服务将与 情感分析服务 一起组成两个 ML 工作流。

下一步将学习如何用Knative Sequence把这两个 ML 工作流串联起来，并按你期望的顺序依次执行——在 Sequence 中bad-word-filter作为第一步、sentiment-analysis-app作为第二步（参考实现见 100-create-sequence.yaml），最终把结果作为 reply 回传给 Broker，从而在书店的全链路事件流中实现“先过滤、再分析情感”的编排。

继续学习：创建 Knative Sequence 来编排 ML 工作流

• 文档
• 教程
• 云原生