上个月，我们团队的 Slack 频道简直成了个笑话。每次凌晨两点服务挂掉，值班工程师就得磕磕绊绊地拼凑那些半生不熟的 kubectl 命令，拼命回想扩容部署或抓取特定 pod 日志的准确语法。我就眼睁睁看着一位高级工程师花了 20 分钟才搞清楚容器为什么被 OOM 杀掉，期间一遍又一遍地敲错 kubectl logs 的参数。
那一刻我突然恍然大悟：既然我们跟这些基础设施的交互只是偶尔为之，干嘛还要死记硬背它的 CLI 语法呢？我想给我们的 DevOps 流水线做个对话式接口——只要说一句“把 API 扩容到 5 个副本”或者“给我看看支付服务的报错”，它就能安全地执行对应命令。评估了几个框架后，我最终选择了 Rasa。
下面就来聊聊我是怎么用 Rasa 搭建这个 DevOps 对话助手的，以及一路走来踩过的坑。
为什么 DevOps 要选 Rasa？
大多数人一提到 Rasa，想到的都是客服机器人。但 Rasa 真正的强项在于处理具有严格业务逻辑的多轮、任务导向型对话——这正是修改基础设施时的刚需。你肯定不希望大语言模型（LLM）随心所欲地给你拼出个 kubectl delete namespace 命令来。你需要的是让它收集好必填参数，跟用户确认后，再去执行一个定义明确的动作。
Rasa 的 CALM 架构（Conversational AI with Language Models，结合语言模型的对话式 AI）正是为此量身定制的：它用 LLM 来理解自然语言，但把实际的执行逻辑保持在确定且受控的状态。当你的助手能直接影响生产系统时，这一点是没得商量的。
准备工作：项目配置
首先，你需要申请一个 Rasa Developer Edition 许可证（免费的）。然后，建一个干净的虚拟环境：
python3 -m venv rasa-devops
source rasa-devops/bin/activate
pip install rasa
初始化一个新项目：
rasa init --no-prompt devops-agent
cd devops-agent
这会搭起 Rasa 的基础项目结构。你会看到 domain.yml、data/nlu.yml、data/stories.yml 和 actions/actions.py 这些文件，它们就是接下来你要重点修改的核心文件了。
定义助手能做什么
domain.yml 文件是用来声明意图（intents）、实体（entities）、槽位（slots）和响应（responses）的地方。对于 DevOps 助手，我一开始先搞定了三个核心操作：扩容服务、查看日志和列出部署。
# domain.yml
intents:
- scale_service
- check_logs
- list_deployments
- confirm
- deny
entities:
- service_name
- replica_count
- namespace
slots:
service_name:
type: text
mappings:
- type: from_entity
entity: service_name
replica_count:
type: float
mappings:
- type: from_entity
entity: replica_count
namespace:
type: text
mappings:
- type: from_entity
entity: namespace
actions:
- action_scale_service
- action_check_logs
- action_list_deployments
我早期踩过一个坑：一开始我把 replica_count 槽位的类型设成了 type: any。千万别这么干。用 type: float 能确保 Rasa 在把参数传给你的动作之前先做校验。你肯定不想把 "five" 这种字符串直接传给 kubectl scale --replicas= 命令吧。
训练 NLU
接下来，在 data/nlu.yml 里定义你的训练样本：
# data/nlu.yml
nlu:
- intent: scale_service
examples: |
- scale [api-service](service_name) to [3](replica_count) replicas
- I need [payment](service_name) scaled up to [5](replica_count)
- increase [auth](service_name) to [2](replica_count)
- can you scale [frontend](service_name) to [10](replica_count)?
- [worker](service_name) needs [4](replica_count) replicas
- intent: check_logs
examples: |
- show me the logs for [api-service](service_name)
- tail [payment](service_name) logs
- what's happening in [auth](service_name)?
- check [frontend](service_name) error logs
- intent: list_deployments
examples: |
- what's running in [production](namespace)?
- list deployments in [staging](namespace)
- show me all services in [default](namespace)
Rasa 只需要这么点样本就能达到不错的准确率，这让我挺惊讶的。我一开始每个意图大概只写了 8-10 个样本，它就已经能正确识别“把 API 提升到 5 个”这种变体说法了。不过，你绝对应该多加一些边缘情况——尤其是那些有歧义的表达。
编写自定义动作
这就是见证 DevOps 魔法的时刻了。Rasa 里的自定义动作就是运行你业务逻辑的 Python 类，它们放在 actions/actions.py 里。
下面是扩容服务的动作代码：
# actions/actions.py
from typing import Any, Text, Dict, List
from rasa_sdk import Action, Tracker
from rasa_sdk.executor import CollectingDispatcher
from rasa_sdk.events import SlotSet
import subprocess
class ActionScaleService(Action):
def name(self) -> Text:
return "action_scale_service"
def run(self, dispatcher: CollectingDispatcher,
tracker: Tracker,
domain: Dict[Text, Any]) -> List[Dict[Text, Any]]:
service_name = tracker.get_slot("service_name")
replica_count = tracker.get_slot("replica_count")
namespace = tracker.get_slot("namespace") or "default"
if not service_name or not replica_count:
dispatcher.utter_message(
text="I need both a service name and replica count to scale."
)
return []
# 执行 kubectl 命令
try:
result = subprocess.run(
[
"kubectl", "scale", "deployment", service_name,
f"--replicas={int(replica_count)}",
"-n", namespace
],
capture_output=True,
text=True,
timeout=30
)
if result.returncode == 0:
msg = f"Done. {service_name} scaled to {int(replica_count)} replicas in {namespace}."
dispatcher.utter_message(text=msg)
else:
dispatcher.utter_message(
text=f"Something went wrong: {result.stderr.strip()}"
)
except subprocess.TimeoutExpired:
dispatcher.utter_message(
text="The command timed out. Check your cluster connection."
)
except Exception as e:
dispatcher.utter_message(
text=f"Unexpected error: {str(e)}"
)
return [SlotSet("service_name", None),
SlotSet("replica_count", None),
SlotSet("namespace", None)]
血泪教训： 一定要在动作结束时清空你的槽位。我一开始忘了这茬，结果助手一直反复尝试扩容同一个服务，就因为槽位在多轮对话中一直保留着。把 SlotSet 事件重置为 None 就解决了这个问题。
另外注意，我在用 subprocess.run 时加了超时时间。当你向集群跑 kubectl 命令时，卡死是常有的事。加上超时可以防止动作无限期阻塞。
增加确认流程
对于 DevOps 助手来说，在执行破坏性操作前绝对必须进行确认。Rasa 通过故事（stories）和表单（forms）来处理这个。下面是我在 data/stories.yml 里设置确认流程的方法：
# data/stories.yml
stories:
- story: scale with confirmation
steps:
- intent: scale_service
- action: utter_confirm_scale
- intent: confirm
- action: action_scale_service
- story: scale denied
steps:
- intent: scale_service
- action: utter_confirm_scale
- intent: deny
- action: utter_scale_cancelled
然后在 domain 里加上对应的响应：
responses:
utter_confirm_scale:
- text: "Please confirm: you want to scale {service_name} to {replica_count} replicas in {namespace}?"
utter_scale_cancelled:
- text: "Scale operation cancelled."
这就是 Rasa 结构化方法真正发挥威力的地方。不受约束的 LLM 可能会跳过确认步骤，或者措辞含糊。而用 Rasa，确认是强制且明确的。
运行助手
先启动动作服务器：
cd actions
rasa run actions
然后在另一个终端里，启动带交互式 shell 的 Rasa 服务器：
rasa shell
下面是一段真实的对话效果：
Your input -> scale the payment service to 5 replicas in production
Bot: Please confirm: you want to scale payment to 5.0 replicas in production?
Your input -> yes
Bot: Done. payment scaled to 5 replicas in production.
当出问题时：
Your input -> show me logs for the api-service
Bot: Running kubectl logs for api-service in default namespace...
Bot: [Last 10 lines of logs appear here]
对接真实基础设施
在生产环境中，你肯定不会用 subprocess 去调 kubectl。相反，你应该用 Kubernetes 的 Python 客户端库，或者直接调你们平台的 API——就像 Scalingo API 项目（github.com/SC-Samir/chatbotsamirrasa，一个很好的参考）通过 REST 调用集成 PaaS 供应商那样。
比如，用官方客户端替换掉 subprocess 调用：
from kubernetes import client, config
class ActionScaleService(Action):
def run(self, dispatcher, tracker, domain):
config.load_incluster_config()
apps_v1 = client.AppsV1Api()
service_name = tracker.get_slot("service_name")
replica_count = int(tracker.get_slot("replica_count"))
namespace = tracker.get_slot("namespace") or "default"
body = {'spec': {'replicas': replica_count}}
apps_v1.patch_namespaced_deployment_scale(
name=service_name,
namespace=namespace,
body=body
)
# ... 处理响应
这样更干净、更可靠，而且你可以用 service account token 做认证，不用依赖本地的 kubeconfig 文件。
实用建议与坦诚的局限性
建议：
- 部署前用
rasa interactive多测试。 它能让你实时纠正助手的理解，并把纠正结果保存为训练数据。 - 用 Rasa X 做对话驱动开发。 它提供了一个 UI 界面，方便你复盘对话并标注助手理解错误的地方。
- 给槽位加上命名空间限制。 我遇到过这样一个 bug：当有人在完全不相关的语境下提到 "in production" 时，
namespace槽位也被填上了。一定要加上严格的实体提取条件。 - 给动作加上限流。 对话循环一旦出 bug，可能会触发疯狂的扩容指令。我加了个简单的内存级限流器。
局限性：
- Rasa 的学习曲线确实陡。 基于 YAML 的配置很快就会变得臃肿。搞懂 stories、rules 和 forms 之间的交互，我足足花了一周时间反复试错。
- 生产环境的 LLM 集成需要许可证。 Developer Edition 是免费的，但要用生产级的 CALM 功能就得买 Rasa Pro。如果你想省钱，可以自己托管他们文档里提到的微调模型，但这又增加了运维成本。
- 多集群管理很复杂。 我目前的设定是单集群上下文。要支持多集群，就得在对话流程里加集群选择环节，复杂度会大增。
- 实时日志流不太契合请求-响应模型。 对于
check_logs，我只能妥协，返回最后 N 行日志，而不是流式输出。真正的流式方案需要 WebSocket 集成，Rasa 支持这个，但需要自定义 channel 开发。
对我而言，最大的体会是：Rasa 的强项不在于它是个多聪明的对话者——而在于它是最靠谱的。当你的助手能直接扩容生产环境基础设施时，可靠性永远比小聪明重要。它那结构化的确认流程、槽位校验和确定性的动作执行，为你提供了纯 LLM 提示词根本给不了的安全护栏。