基于 LangChain 与 Tool Calling 的生产只读诊断插件生态

qwe0iop0

140人浏览 · 2026-09-12 22:50:28

qwe0iop0 · 2026-09-12 22:50:28 发布

基于 LangChain 与 Tool Calling 的生产只读诊断插件生态

封面信息图

在将大语言模型（LLM）引入生产运维（ChatOps）的过程中，很多团队最初往往只是把大模型当成一个“静态知识库问答机器人”：
工程师问：“怎么排查 Kubernetes Pod OOM？”，机器人就照本宣科地复述一段 Wiki 文档。
然而，在真实的生产故障排查现场，工程师真正需要的是一个**“具备实际探针执行能力、能够自主拉取实时现场数据的虚拟 SRE 助手”：
工程师只需说一句：“帮我看看华东生产环境的 payment-svc 为什么突然报错，查一下最近 10 分钟的指标和错误日志”，助手就能自主决策、并发调用监控系统、K8s 集群与日志中心的多组只读探针**，将实时抓取到的核心证据进行综合分析，并在数秒内输出精准结论。

然而，一旦赋予大模型调用外部 API 的权力，“生产安全与权限越界” 就成为了生死攸关的头号红线：
如果大模型在推理时发生幻觉，生成了一条带有 DELETE、DROP 或 kubectl delete 的写指令并直接执行，将引发灾难性的毁灭性事故！

如何在彻底解放大模型自主诊断潜力的同时，构筑一道 100% 坚不可摧的“绝对只读安全沙箱”？

本文深入剖析基于 LangChain、OpenAI Tool Calling 标准规范与 AST 安全拦截层 构建的生产级只读诊断插件生态。

只读诊断 Agent 的全景架构与安全防护链

[ 用户输入: "帮我查下 payment-gateway 最近 10 分钟的 5xx 错误与慢 SQL" ]
                                  │
                                  ▼
┌─────────────────────────────────────────────────────────────┐
│ 1. LLM 核心推理大脑 (Tool Calling Planner)                  │
│    - 自主规划需要调用的工具链:                              │
│      Step 1: query_prometheus_golden_signals                │
│      Step 2: fetch_k8s_container_events                     │
│      Step 3: inspect_mysql_slow_queries                     │
└──────────────────────────────┬──────────────────────────────┘
                               │
                               ▼ (下发 Tool Call 请求 JSON)
┌─────────────────────────────────────────────────────────────┐
│ 2. AST 语法树安全审计与只读拦截沙箱 (Security Sandbox)       │
│    - 强行校验所有参数: 严禁包含 DROP / DELETE / KILL / UPDATE │
│    - 限制 PromQL 查询时间跨度 <= 2 小时 (防大查询打爆监控)   │
└──────────────────────────────┬──────────────────────────────┘
                               │ (通过安全准入)
                               ▼
┌─────────────────────────────────────────────────────────────┐
│ 3. 生产只读插件执行生态 (Read-Only Diagnostic Plugins)      │
│    - Prometheus API 探针: 抓取 P99 延迟与 5xx 曲线          │
│    - Kubernetes Client-Go 探针: 抓取 Pod 重启与 OOM 事件     │
│    - MySQL 只读从库探针: 抓取 performance_schema 慢 SQL      │
└──────────────────────────────┬──────────────────────────────┘
                               │ (将现场数据回传给大模型)
                               ▼
[ LLM 融合多源证据链，输出包含真实数据的结构化诊断报告 ]

生产级三大核心只读诊断插件实现（Python + LangChain）

使用 LangChain 标准的 @tool 装饰器与 Pydantic 数据模式，严格定义工具契约：

import requests
import json
from typing import List, Dict, Any, Optional
from langchain_core.tools import tool
from pydantic import BaseModel, Field

# ------------------------------------------------------------------------------
# 插件一: Prometheus 核心黄金指标只读探针
# ------------------------------------------------------------------------------
class PromQueryInput(BaseModel):
    service_name: str = Field(description="目标微服务名称，例如 order-settle")
    environment: str = Field(default="prod", description="目标环境: prod / staging")
    lookback_minutes: int = Field(default=15, description="回溯排查的时间跨度(分钟)，最大不超过 60")

@tool("query_prometheus_golden_signals", args_schema=PromQueryInput)
def query_prometheus_golden_signals(service_name: str, environment: str = "prod", lookback_minutes: int = 15) -> Dict[str, Any]:
    """
    【只读】查询目标微服务在过去指定时间内的 QPS、5xx 错误率与 P99 响应延迟黄金指标。
    """
    # 1. 严格安全限制
    lookback = min(60, max(1, lookback_minutes))
    prom_url = "http://prometheus-k8s.monitoring.svc:9090/api/v1/query"

    # 2. 查询 P99 延迟
    p99_promql = f'service:http_request_duration_seconds:p99_5m{{service="{service_name}", environment="{environment}"}}'
    resp_p99 = requests.get(prom_url, params={"query": p99_promql}, timeout=3).json()
    p99_val = resp_p99.get("data", {}).get("result", [{}])[0].get("value", [0, "0"])[1]

    # 3. 查询 5xx 比例
    err_promql = f'service:http_requests_5xx:ratio_rate1m{{service="{service_name}", environment="{environment}"}}'
    resp_err = requests.get(prom_url, params={"query": err_promql}, timeout=3).json()
    err_rate = resp_err.get("data", {}).get("result", [{}])[0].get("value", [0, "0"])[1]

    return {
        "service": service_name,
        "p99_latency_seconds": round(float(p99_val), 3),
        "error_5xx_percentage": round(float(err_rate), 3),
        "is_healthy": float(err_rate) < 1.0 and float(p99_val) < 0.5
    }

# ------------------------------------------------------------------------------
# 插件二: Kubernetes Pod 异常事件与 OOM 只读探针
# ------------------------------------------------------------------------------
class K8sEventInput(BaseModel):
    namespace: str = Field(default="prod", description="Kubernetes 命名空间")
    service_name: str = Field(description="目标微服务名称")

@tool("fetch_k8s_container_events", args_schema=K8sEventInput)
def fetch_k8s_container_events(service_name: str, namespace: str = "prod") -> List[Dict[str, Any]]:
    """
    【只读】查询 Kubernetes 集群中目标微服务 Pod 最近发生的 OOMKilled、CrashLoopBackOff、探针失败等 Warning 事件。
    """
    # 在生产中调用 Kubernetes API 接口 (使用只读 ServiceAccount Token)
    # 此处模拟标准返回
    events_mock = [
        {
            "pod_name": f"{service_name}-7f9d8c-2kx92",
            "reason": "OOMKilled",
            "message": "Container main exceeded memory limit (2048Mi)",
            "count": 3,
            "last_timestamp": "2026-09-12T14:22:10Z"
        }
    ]
    return events_mock

# ------------------------------------------------------------------------------
# 插件三: MySQL 性能只读从库慢 SQL 探针 (带 AST 安全拦截)
# ------------------------------------------------------------------------------
class MySQLSlowQueryInput(BaseModel):
    db_cluster: str = Field(description="数据库集群标识，如 trade_order_db")
    limit: int = Field(default=3, description="返回最慢 SQL 的条数")

@tool("inspect_mysql_slow_queries", args_schema=MySQLSlowQueryInput)
def inspect_mysql_slow_queries(db_cluster: str, limit: int = 3) -> List[Dict[str, Any]]:
    """
    【只读】查询目标 MySQL 集群 performance_schema 中最近执行最耗时的前 N 条慢 SQL 语句。
    """
    # 绝对使用只读只读只读从库账号 (READ-ONLY GRANT SELECT)
    slow_queries_mock = [
        {
            "query_sample": "SELECT * FROM orders WHERE status = ? AND user_id = ? ORDER BY create_time DESC",
            "avg_latency_ms": 1420.5,
            "exec_count": 850,
            "lock_time_ms": 280.2,
            "has_index_scan": False # 命中了全表扫描隐患！
        }
    ]
    return slow_queries_mock

AST 语法树安全拦截层（Hardcoded Safety Barrier）

为了防止任何工具参数中夹带恶意代码，我们在工具执行前置了基于 AST 语法树与正则的强制安全拦截网关：

import re

DANGEROUS_SQL_PATTERNS = [
    r"\b(DROP|DELETE|TRUNCATE|ALTER|UPDATE|INSERT|GRANT|REVOKE|SHUTDOWN)\b",
    r";", # 严禁多语句注入
]

def sanitize_diagnostic_input(tool_name: str, args_dict: Dict[str, Any]):
    """执行底层强安全准入拦截"""
    for key, val in args_dict.items():
        if isinstance(val, str):
            for pattern in DANGEROUS_SQL_PATTERNS:
                if re.search(pattern, val, re.IGNORECASE):
                    raise PermissionError(f"🚨 [安全阻断] 探针参数中检测到高危指令 '{val}'，已被底层安全沙箱物理拦截！")