前言

以前管 Kubernetes 都是 kubectl 或 Dashboard，出門在外還要 VPN、記指令，超麻煩。這次我直接把 KubeWizard 改造成 LINE Bot Agent API，讓你用手機聊天就能查資源、看 log、重啟服務，還能自動化、擴充工具，維運體驗直接升級。

主要功能：

• 文字互動管 K8s（查資源 / 看 Logs / 重啟）
• REST API 事件入口（Prometheus / GitOps / Webhook）
• 工具可擴充，AI 輔助、自動化腳本隨你加
• 多人獨立記憶 + 智能上下文

模型我選 Google Gemini，因為：

• 有 Free Tier，夠用
• 跟 Python / FastAPI / LangChain 整合很順
• 多語系理解力不錯，維運場景很夠用

Gemini API Rate limits 詳細請看官方：Gemini API Rate limits

接下來直接看：價值 → 架構 → 元件 → 實作 → Demo → 延伸應用。

用 LINE + Agent 管 K8s 有什麼爽點？

✅ 隨時隨地

• 不用 VPN、不用 Terminal，手機就能搞定
• 多人 session 各自記憶，互不干擾

🔗 事件入口整合

• REST API 可接 Prometheus、GitLab、Argo CD、AlertManager Webhook
• 告警來了直接回 LINE 回覆診斷、給操作選項

🧠 智能擴充

• Tools 模式：查 K8s、Pipeline、Jira、Log 分析、Search、HTTP call 其他 API
• AI 幫你生 YAML、摘要錯誤、給修復建議

🎯 對話記憶

• Redis 存上下文 + Token Buffer 自動摘要
• 使用者分流，互不干擾

架構總覽

整體流程：使用者訊息 → LINE → Webhook → Agent → 工具 → 回覆

架構分四層：

LangChain 核心概念（速讀版）

LangChain 就是把「LLM + Tools + 記憶 + Prompt + 執行迴路」組成一顆可擴充的大腦：

流程：問題 → 判斷要不要用工具 → 執行工具 → 整合回覆 → 存記憶（必要時摘要）

prompt = ChatPromptTemplate.from_messages([
  ("system", SYSTEM_PROMPT),
  MessagesPlaceholder("chat_history"),
  ("user", "{input}"),
  MessagesPlaceholder("agent_scratchpad"),
])

tools = [KubeTool(), RequestsGet(), create_search_tool()]
memory = ConversationTokenBufferMemory(
  llm=llm,
  memory_key="chat_history",
  max_token_limit=2000,
  chat_memory=RedisChatMessageHistory(url=REDIS_URL, session_id=user_id)
)

agent = create_openai_tools_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, memory=memory, verbose=True)
reply = executor.invoke({"input": "列出所有 namespace"})

為什麼用 Python Kubernetes SDK 不直接呼叫 kubectl？

自動偵測環境：

def load_k8s_config():
    if os.path.exists("/var/run/secrets/kubernetes.io/serviceaccount/token"):
        config.load_incluster_config()
    else:
        config.load_kube_config()

常見映射：kubectl get pods -n X → v1.list_namespaced_pod(namespace=X)。

Step 1：建立 LINE Bot（前置準備）

1. 前往 LINE Developers：
   https://developers.line.biz/console/

2. 建立 Provider

3. 建立 Messaging API Bot

4. 取得 Channel 設定並寫入 .env：

LINE_CHANNEL_SECRET=你的_channel_secret
LINE_CHANNEL_ACCESS_TOKEN=你的_access_token

5. 設定 Webhook URL，例如：

https://your-domain.com/linebot/callback

Webhook 測試可以用 ngrok 讓本地端公開測試。

Webhook Settings 建議設定

需使用 HTTPS。成功後請點擊 Verify 確認。

Step 2：改造成「LINE Bot Agent API」核心元件

預設 KubeWizard 只能問一句答一句，要變成真正好用的 Agent，必須加上記憶、工具判斷、多入口。

核心設計概念

功能模組化（Tools）

每個功能都獨立成 Tool，擴充超方便。以下是 KubeTool 實作：

from langchain_core.tools import BaseTool
from kubernetes import client, config
from pydantic import BaseModel, Field

class KubeInput(BaseModel):
    """Kubernetes 工具的參數模型"""
    commands: str = Field(
        ...,
        example="kubectl get pods",
        description="要執行的 kubectl 相關命令"
    )

class KubeTool(BaseTool):
    """Kubernetes 工具 - 使用 Python SDK 執行 K8s 操作"""
    
    name: str = "KubeTool"
    description: str = """在 Kubernetes 集群上執行 k8s 相關命令的工具。
    支援 get/describe/logs/list 等操作。
    特別功能：
    - 使用 'kubectl list all' 快速查看所有 namespace 和 pods 概覽
    - 使用 'kubectl list namespaces' 查看所有 namespace
    - 使用 'kubectl list pods' 按 namespace 分組查看所有 pods
    """
    args_schema: Type[BaseModel] = KubeInput
    
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        # 自動判斷環境並載入配置
        try:
            config.load_incluster_config()  # Pod 內環境
            logger.info("使用集群內配置 (Pod 環境)")
        except:
            config.load_kube_config()  # 本地環境
            logger.info("使用本地 kubeconfig 配置")
        
        self.v1 = client.CoreV1Api()
        self.apps_v1 = client.AppsV1Api()
    
    def _run(self, commands: str) -> str:
        """執行 kubectl 命令並返回結果"""
        # 解析命令並轉換為 SDK API 調用
        # 例如：kubectl get pods -n default
        # 轉換為：self.v1.list_namespaced_pod(namespace="default")
        ...

Agent 判斷邏輯

# Agent 初始化
from langchain.agents import create_openai_tools_agent, AgentExecutor
from langchain_google_genai import ChatGoogleGenerativeAI

# 定義可用工具
tools = [
    KubeTool(),
    SearchTool(),
    RequestsGet(),
    human_console_input()
]

# 創建 Agent
agent = create_openai_tools_agent(
    llm=ChatGoogleGenerativeAI(model="gemini-2.0-flash"),
    tools=tools,
    prompt=system_prompt
)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    verbose=True
)

# 執行流程：使用者問題 → Agent 分析 → 選擇工具 → 執行 → 回傳結果
result = agent_executor.invoke({"input": "列出所有 pods"})

記憶功能（Redis + Token 管理）

使用 Redis 儲存對話上下文，並使用 Token Buffer 管理對話長度：

from langchain_community.chat_message_histories import RedisChatMessageHistory
from langchain.memory import ConversationTokenBufferMemory

# 1. Redis 聊天歷史（每個用戶獨立 session）
chat_history = RedisChatMessageHistory(
    url="redis://localhost:6379/0",
    session_id=user_id,  # 每個用戶有獨立的 session
    key_prefix="chat_history:",
    ttl=3600  # 1 小時過期
)

# 2. Token Buffer Memory（自動管理對話長度）
memory = ConversationTokenBufferMemory(
    llm=llm,
    memory_key="chat_history",
    output_key="output",
    return_messages=True,
    max_token_limit=2000,  # 限制 Token 數量
    chat_memory=chat_history
)

# 3. 自動總結機制（當對話過長時）
if len(chat_history.messages) > 10:
    # 使用 AI 總結對話
    summary = llm.invoke(f"總結以下對話：{chat_history.messages}")
    chat_history.clear()
    chat_history.add_message(summary)

FastAPI 路由設計

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI(title="KubeWizard LINE Bot API")

class ChatRequest(BaseModel):
    message: str
    user_id: str

@app.post("/api/chat")
async def chat(request: ChatRequest):
    """聊天端點 - 支援 REST API 調用"""
    agent = KubeAgent(user_id=request.user_id)
    result = agent.invoke(request.message)
    return {"reply": result["output"], "user_id": request.user_id}

@app.post("/linebot/callback")
async def linebot_callback(request: Request):
    """LINE Bot Webhook 端點"""
    # 驗證簽名
    signature = request.headers.get("X-Line-Signature")
    body = await request.body()
    
    try:
        handler.handle(body.decode(), signature)
    except InvalidSignatureError:
        raise HTTPException(status_code=400, detail="Invalid signature")
    
    return {"status": "ok"}

@linebot_handler.add(MessageEvent, message=TextMessage)
def handle_message(event):
    """處理 LINE 訊息"""
    user_id = event.source.user_id
    user_message = event.message.text
    
    # 使用 KubeAgent 處理訊息
    agent = KubeAgent(user_id=user_id)
    result = agent.invoke(user_message)
    
    # 回覆 LINE 訊息
    line_bot_api.reply_message(
        event.reply_token,
        TextSendMessage(text=result["output"])
    )

我的「Vibe Coding」迭代心得

「Vibe Coding」就是我這次專案的開發心法，感覺驅動、輕量迭代，不先寫滿設計文件，讓程式、模型、工具行為自己告訴我下一步該修什麼。流程大致如下：

1. 最小可行：只建立一個 KubeAgent + 基本 Prompt + 單一工具 (KubeTool)。
2. 觀察行為：在 CLI / FastAPI 中丟「列出 pods」看看模型是不是會錯誤幻想→如果會，就加關鍵字補充。
3. 加工具：逐步加入 search, RequestsGet, human_console_input 工具，每加一個就測試 2~3 條指令。
4. 人設調整：Prompt 裡加「先分析問題」避免模型直接亂猜原因。
5. 加安全欄：分成 KubeTool（安全）與 KubeToolWithApprove（危險操作如刪除 / 重啟）。
6. 加記憶：用 Redis 保存上下文，超過 10 則自動摘要 → 觀察 Token 使用是否回落。
7. 自動上下文：加入「輸入含 ‘列出’ + 資源關鍵字」才附加整體集群 snapshot，避免每題都浪費資源。
8. Helm 化：把所有環境變數、Secrets、RBAC 拆進 values.yaml → 部署一次可複用。

「Vibe Coding」的核心心法：

• 只解決現在真的出現的問題，不預先過度設計。
• 每次迭代都確認：輸入 → Agent 行為 → 工具呼叫 → 回覆格式。
• 用日誌觀察「模型是否該用工具卻沒用」→ 代表 Prompt 或工具描述需要調整。
• 把難 Debug 的東西（像 kubectl 字串）換成可程式化（Python SDK）。

小技巧：

• Prompt 明列口頭禪，回覆語氣一致
• 工具描述用「使用者語言」取代技術術語
• 危險操作拆分工具 + 人工確認
• DEBUG 看推理、INFO 看工具使用，方便定位問題

技術亮點、挑戰與解法

簡化的 kubectl → SDK 映射範例：

def map_command(cmd: str):
  parts = cmd.split()
  if parts[:3] == ["kubectl", "get", "pods"]:
      ns = "default"
      if "-n" in parts:
          ns = parts[parts.index("-n") + 1]
      return v1.list_namespaced_pod(namespace=ns)
  raise ValueError("暫不支援的指令")

註：此表列出已完成與預留強化的項目，後續可加入：自動重試、工具使用統計、Prometheus 指標。

（技術選型已前移，原段落合併至前面）

RBAC 權限設定（建議）

以下為建議的最小必要權限，支援 Namespace Scoped 和 Cluster Wide 兩種模式。

Namespace Scoped（推薦，最小權限原則）

適用場景：

• 只需要管理特定 namespace 的資源
• 不需要 cluster-admin 權限即可部署
• 更安全、更符合企業安全規範

apiVersion: v1
kind: ServiceAccount
metadata:
  name: kubewizard-bot
  namespace: default
---
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: kubewizard-bot-role
  namespace: default
rules:
  # Core 資源
  - apiGroups: [""]
    resources: ["pods", "services", "endpoints", "events", "configmaps", "secrets", "persistentvolumeclaims"]
    verbs: ["get", "list", "watch", "create", "update", "patch", "delete"]
  
  # Pod logs（只需讀取權限）
  - apiGroups: [""]
    resources: ["pods/log"]
    verbs: ["get", "list"]
  
  # Apps 資源
  - apiGroups: ["apps"]
    resources: ["deployments", "replicasets", "statefulsets", "daemonsets"]
    verbs: ["get", "list", "watch", "create", "update", "patch", "delete"]
  
  # Batch 資源
  - apiGroups: ["batch"]
    resources: ["jobs", "cronjobs"]
    verbs: ["get", "list", "watch", "create", "update", "patch", "delete"]
  
  # Networking 資源
  - apiGroups: ["networking.k8s.io"]
    resources: ["ingresses", "networkpolicies"]
    verbs: ["get", "list", "watch", "create", "update", "patch", "delete"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: kubewizard-bot-binding
  namespace: default
subjects:
  - kind: ServiceAccount
    name: kubewizard-bot
    namespace: default
roleRef:
  kind: Role
  name: kubewizard-bot-role
  apiGroup: rbac.authorization.k8s.io

Cluster Wide（進階用）

適用場景：

• 需要跨 namespace 管理資源
• 需要查看 nodes、namespaces 等集群級別資源
• 需要 cluster-admin 權限來部署

apiVersion: v1
kind: ServiceAccount
metadata:
  name: kubewizard-bot
  namespace: default
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: kubewizard-bot-cluster-role
rules:
  # Namespace 級別資源（所有 namespace）
  - apiGroups: [""]
    resources: ["pods", "services", "endpoints", "events", "configmaps", "secrets", "persistentvolumeclaims"]
    verbs: ["get", "list", "watch", "create", "update", "patch", "delete"]
  
  - apiGroups: [""]
    resources: ["pods/log"]
    verbs: ["get", "list"]
  
  - apiGroups: ["apps"]
    resources: ["deployments", "replicasets", "statefulsets", "daemonsets"]
    verbs: ["get", "list", "watch", "create", "update", "patch", "delete"]
  
  - apiGroups: ["batch"]
    resources: ["jobs", "cronjobs"]
    verbs: ["get", "list", "watch", "create", "update", "patch", "delete"]
  
  - apiGroups: ["networking.k8s.io"]
    resources: ["ingresses", "networkpolicies"]
    verbs: ["get", "list", "watch"]
  
  # Cluster 級別資源（只讀）
  - apiGroups: [""]
    resources: ["nodes", "namespaces", "persistentvolumes"]
    verbs: ["get", "list", "watch"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: kubewizard-bot-cluster-binding
subjects:
  - kind: ServiceAccount
    name: kubewizard-bot
    namespace: default
roleRef:
  kind: ClusterRole
  name: kubewizard-bot-cluster-role
  apiGroup: rbac.authorization.k8s.io

Helm Chart 配置（推薦）

專案已包含完整的 Helm Chart，可透過 values.yaml 輕鬆配置：

# values.yaml

# RBAC 配置
rbac:
  # 是否創建 RBAC 資源
  create: true
  
  # 使用 Role (false) 或 ClusterRole (true)
  # 設為 false 可避免需要 cluster-admin 權限
  clusterWide: false
  
  # 自定義 RBAC 規則
  rules:
    - apiGroups: [""]
      resources: ["pods", "services"]
      verbs: ["get", "list", "watch"]

部署方式：

# 1. Namespace Scoped 部署（推薦）
helm install kubewizard ./helm \
  --set rbac.clusterWide=false \
  --set secrets.AI_GOOGLE_API_KEY="your-api-key" \
  --namespace default

# 2. Cluster Wide 部署
helm install kubewizard ./helm \
  --set rbac.clusterWide=true \
  --set secrets.AI_GOOGLE_API_KEY="your-api-key" \
  --namespace default

# 3. 驗證 RBAC 權限
kubectl auth can-i list pods \
  --as=system:serviceaccount:default:kubewizard-bot \
  -n default

權限驗證與測試

# 測試 ServiceAccount 權限
kubectl create serviceaccount kubewizard-bot -n default
kubectl apply -f rbac.yaml

# 驗證權限
kubectl auth can-i get pods \
  --as=system:serviceaccount:default:kubewizard-bot \
  -n default
# 應該返回：yes

# 測試在 Pod 中執行
kubectl run test-pod \
  --image=python:3.11-slim \
  --serviceaccount=kubewizard-bot \
  -n default \
  -- sleep 3600

# 進入 Pod 測試
kubectl exec -it test-pod -n default -- python3
>>> from kubernetes import client, config
>>> config.load_incluster_config()
>>> v1 = client.CoreV1Api()
>>> pods = v1.list_namespaced_pod(namespace="default")
>>> print(len(pods.items))

安全最佳實踐

1. 最小權限原則：優先使用 Namespace Scoped Role
2. 避免通配符：不要使用 resources: ["*"]
3. 限制危險操作：刪除操作建議使用人工審批
4. 定期審查：定期檢查 RBAC 權限是否過大
5. 使用 Pod Security Policies：限制 Pod 的安全上下文

資訊流示意（白話版）

使用者輸入：

「幫我看一下 kube-system 裡的 Pods」

流程：

1. LINE 收到訊息
2. Webhook 傳至 FastAPI
3. FastAPI 將內容交給 Agent
4. Agent 從 Redis 取得上下文
5. Agent 判定需使用 K8sTool
6. 執行 Python SDK 查詢叢集
7. 回傳內容至 LINE

Demo 示範（文字版）

範例 1：查詢 Pods 狀態

你： 幫我看 default namespace 的 Pods
Bot：

📦 Namespace: default
================================
NAME                         READY   STATUS              RESTARTS   AGE
nginx-deployment-7d8f...     1/1     Running            0          2d
redis-master-0               1/1     Running            0          5d
payment-service-abc123       0/1     CrashLoopBackOff   5          10m

總計: 3 pods (2 Running, 0 Pending, 1 CrashLoopBackOff)
⚠️ 發現問題: payment-service-abc123 處於 CrashLoopBackOff 狀態
需要查看 Logs 嗎？

你： 看一下那個 Crash 的 Log
Bot：

🔍 payment-service-abc123 的最近日誌：

2025-11-22 10:30:15 [ERROR] Failed to connect to database
2025-11-22 10:30:15 [ERROR] Connection refused: mysql:3306
2025-11-22 10:30:15 [FATAL] Application startup failed

💡 可能原因：
1. MySQL Service 不存在或無法訪問
2. 環境變數配置錯誤（DB_HOST）
3. 網絡策略阻擋連線

建議檢查：
- kubectl get svc -n default | grep mysql
- 確認 Pod 的環境變數設定

你： 幫我重啟看看
Bot：

🔄 正在重啟 Deployment: payment-service

✅ 已執行：kubectl rollout restart deployment/payment-service -n default
⏳ 等待新 Pod 啟動中...

📊 更新狀態：
- 舊版 Pods: 正在終止
- 新版 Pods: 正在創建
- 預計完成時間: ~30 秒

我會持續監控狀態，稍後回報結果。

範例 2：診斷網路問題

你： 為什麼我的 Ingress 無法訪問？
Bot： 讓我幫你診斷 Ingress 問題。

🔍 診斷步驟：

1️⃣ 檢查 Ingress 資源
✅ Ingress 'my-app' 已存在
✅ Ingress Controller 正常運行

2️⃣ 檢查 Service
✅ Service 'my-app-svc' 已存在 (ClusterIP: 10.96.1.100)

3️⃣ 檢查 Pod
⚠️ 問題發現：Service 後端沒有可用的 Pod！

EndpointSlice 資訊：
- my-app-svc: 0/3 endpoints ready

📋 建議解決方案：
1. 檢查 Deployment 的 selector 是否與 Service 匹配
2. 檢查 Pod 的 labels 是否正確
3. 執行：kubectl describe service my-app-svc -n default

需要我幫你執行這些檢查嗎？

範例 3：智能上下文記憶

你： 列出所有 namespace
Bot：

📋 所有 Namespaces：
1. default
2. kube-system
3. kube-public
4. production
5. staging
6. monitoring

總計: 6 個 namespaces

你： production 裡面有什麼？
Bot： (智能記憶前一句的 “production” namespace)

📦 Namespace: production
================================

Deployments:
- api-gateway (3/3 Ready)
- user-service (2/2 Ready)
- order-service (5/5 Ready)

Services:
- api-gateway-svc (LoadBalancer)
- user-service-svc (ClusterIP)
- order-service-svc (ClusterIP)

總計: 3 Deployments, 3 Services, 10 Pods (全部正常)

你： 重啟 api-gateway
Bot： (記得我們在談論 production namespace 的 api-gateway)

確認要重啟 production namespace 的 api-gateway 嗎？
這會導致短暫的服務中斷。

[已確認]

🔄 正在執行：kubectl rollout restart deployment/api-gateway -n production
✅ 重啟成功！新 Pods 已啟動並就緒。

更多應用場景

（延伸應用提前放，方便大家快速想像擴充場景：告警 / GitOps / 定時任務）

1. 告警自動處理

# Prometheus AlertManager Webhook
@app.post("/webhook/prometheus")
async def prometheus_webhook(alert: PrometheusAlert):
  agent = KubeAgent()
  result = agent.invoke(f"診斷問題: {alert.description}")
  send_line_message(result)

2. GitOps 整合

@app.post("/webhook/argocd")
async def argocd_webhook(event: ArgoCDEvent):
  if event.type == "sync-failed":
    agent = KubeAgent()
    result = agent.invoke(f"分析同步失敗原因: {event.app_name}")
    send_line_message(result)

3. 定時健康檢查

@app.get("/cron/health-check")
async def health_check():
  agent = KubeAgent()
  result = agent.invoke("檢查所有 namespace 的 pod 狀態")
  if "CrashLoopBackOff" in result:
    send_alert("發現異常 Pod！")

結語

這篇筆記示範怎麼把 KubeWizard 整合到 LINE Bot，打造出能聊天管 K8s 的助理，重點：

✅ 智能 Agent：LangChain + Gemini，能自動判斷工具、支援多步驟操作 ✅ 對話記憶：Redis 持久化、Token Buffer 管理長度、多用戶 session ✅ Python SDK 操作叢集：不用 kubectl 二進制，Docker 映像更輕量，能自動判斷 Pod 內外 ✅ RBAC 權限控管：支援 Namespace Scoped/Cluster Wide，最小權限原則，Helm Chart 一次到位 ✅ 生產就緒：Helm Chart、Docker Compose、健康檢查、錯誤處理、HPA 水平擴展

🚀 立即開始

GitHub 專案： KubeWizard

# 1. Clone 專案
git clone https://github.com/markku636/kubewizard.git
cd kubewizard

# 2. 配置環境變數
cp .env.example .env
# 編輯 .env，填入你的 API Key

# 3. 使用 Docker Compose 啟動
docker-compose up -d

# 4. 測試 API
curl -X POST http://localhost:8000/api/chat \
  -H "Content-Type: application/json" \
  -d '{"message":"列出所有 pods","user_id":"test"}'

# 5. 或使用 Helm 部署到 K8s
helm install kubewizard ./helm \
  --set secrets.AI_GOOGLE_API_KEY="your-api-key" \
  --set secrets.LINE_CHANNEL_SECRET="your-secret" \
  --set secrets.LINE_CHANNEL_ACCESS_TOKEN="your-token"

📚 參考資料

• 快速開始指南
• LINE Bot 整合文檔
• Kubernetes 配置說明
• Helm Chart 部署指南

特別感謝以下開源專案：

• LangChain - 強大的 LLM 應用框架
• Google Gemini - 優秀的 AI 模型
• Kubernetes Python Client - 官方 Python SDK
• FastAPI - 現代化的 Web 框架