AI 智能体如何参与光伏电站运维?

引言

本文按 5 个生产环境真实在跑的用例，拆解光伏电站 AI Agent 的技术实现。每个用例给出技术架构、tool calling、Prompt 设计和评估指标。

用例一：自动日报生成

架构

┌────────────┐    ┌──────────┐    ┌────────────┐
│ Cron 18:00 │ → │ Agent    │ ← │ tools/     │
└────────────┘    │ (LLM)    │    │ - tsdb_q   │
                  │          │    │ - alert_q  │
                  └──────────┘    │ - device_q │
                       ↓          └────────────┘
                  ┌──────────┐
                  │ Markdown │
                  │ → IM/Mail│
                  └──────────┘

Tool 设计

from pydantic import BaseModel
from typing import Optional

class DailyReportTools:
    @tool
    def query_generation(self, station_id: str, date: str) -> dict:
        """查询电站当日发电数据（kWh / 满发小时 / 辐照）"""
        pass

    @tool
    def query_alerts(self, station_id: str, date: str, severity: Optional[str] = None) -> list[dict]:
        """查询当日告警，可按严重度过滤"""
        pass

    @tool
    def query_device_status(self, station_id: str) -> dict:
        """查询设备可用率 + 在线状态"""
        pass

    @tool
    def query_history_compare(self, station_id: str, metric: str, days: int) -> dict:
        """对比历史同期数据"""
        pass

Prompt 模板

System:
你是众壹能源 ZenovaAI 光伏运营助手。任务是生成站长视角的简洁日报。

要求：
1. 中文。专业但不生硬。
2. 关键数据点必须有链接（用 [描述](id) 格式，前端会替换）。
3. 异常分级标记：P0（停机 / 大幅损失），P1（部分容量损失），P2（可观察）。
4. 不写废话「今天没有什么特别情况」之类。无内容直接简报「今日运行正常」。
5. 末尾不要加促销或 AI 标识。

输入：
站点 ID: {station_id}
日期: {date}

请通过 tool 查询数据后生成日报。

评估

class DailyReportEval:
    def __init__(self, station_id: str):
        self.station_id = station_id

    def precision_critical(self, report: str, ground_truth: list[Alert]) -> float:
        """日报中提到的 P0/P1 告警与实际比例"""
        mentioned = extract_alerts_from_report(report)
        actual_critical = [a for a in ground_truth if a.severity in ('P0', 'P1')]
        return len(set(mentioned) & set(actual_critical)) / len(actual_critical)

    def hallucination_rate(self, report: str, all_data: dict) -> float:
        """日报中提到的数字不在 ground truth 里的比例"""
        numbers = extract_numbers_from_report(report)
        return sum(1 for n in numbers if not is_in_data(n, all_data)) / len(numbers)

目标指标：

• precision_critical ≥ 95%
• hallucination_rate ≤ 1%

用例二：告警归因分析

架构

alert_stream → [event_aggregator] → [topology_correlator] → [LLM root_cause] → [priority_ranker] → alert_clusters

关键算法

class AlertClusterer:
    def __init__(self, time_window_min: int = 5):
        self.time_window = time_window_min * 60

    def cluster(self, alerts: list[Alert]) -> list[AlertCluster]:
        # 按时间窗 + 拓扑关联聚类
        sorted_alerts = sorted(alerts, key=lambda a: a.timestamp)
        clusters = []
        current_cluster = []
        for alert in sorted_alerts:
            if not current_cluster:
                current_cluster.append(alert)
                continue
            last = current_cluster[-1]
            if (alert.timestamp - last.timestamp).total_seconds() <= self.time_window \
               and self._topo_related(alert, last):
                current_cluster.append(alert)
            else:
                clusters.append(AlertCluster(current_cluster))
                current_cluster = [alert]
        if current_cluster:
            clusters.append(AlertCluster(current_cluster))
        return clusters

    def _topo_related(self, a: Alert, b: Alert) -> bool:
        # 同一逆变器下的设备视为拓扑关联
        return a.inverter_id == b.inverter_id or \
               self._is_descendant(a.device_id, b.device_id)

Root Cause Prompt

System:
你是光伏电站告警根因分析助手。给定一组关联告警，输出最可能的根因 + 置信度 + 建议操作。

输入：告警列表（时间、设备、严重度、描述）+ 设备拓扑树

输出 JSON 严格格式：
{
  "root_cause": "...",
  "confidence": 0.0-1.0,
  "reasoning": "...",
  "recommended_action": "...",
  "estimated_impact_kw": ...
}

评估

建立 100-200 条已知根因的告警集群作为评估集，定期跑：

def evaluate_root_cause(model, eval_set):
    correct = 0
    for case in eval_set:
        prediction = model.predict(case.cluster)
        if prediction.root_cause == case.actual_root_cause:
            correct += 1
    return correct / len(eval_set)

目标：≥ 85% 精度。

用例三：巡检任务调度

架构

[历史告警] ──┐
[IV 趋势] ────┼─→ [Agent] ─→ [任务清单 + 路径]
[天气预报] ──┤      ↑
[巡检记录] ──┘   tools

Tool

class InspectionTools:
    @tool
    def query_alert_hotspots(self, station_id: str, days: int) -> list[dict]:
        """查询过去 N 天告警热点设备"""

    @tool
    def query_iv_anomalies(self, station_id: str, days: int) -> list[dict]:
        """查询过去 N 天 IV 异常组件"""

    @tool
    def query_weather_forecast(self, station_id: str, days: int) -> list[dict]:
        """查询未来 N 天天气"""

    @tool
    def query_last_inspection(self, station_id: str) -> dict:
        """查询最近一次巡检记录"""

    @tool
    def compute_optimal_path(self, devices: list[str]) -> list[str]:
        """按位置坐标优化巡检路径（TSP 近似）"""

用例四：业主对账问答 RAG

架构

业主 question → embedding → vector search → context retrieval → LLM 回答
                                  ↓
                       [time series db]
                       [电价 / 合同 / 工单库]

检索策略

class OwnerQARetriever:
    def __init__(self, station_id: str):
        self.station_id = station_id
        self.vector_db = ChromaClient()
        self.tsdb = TSDBClient()

    def retrieve(self, question: str) -> list[Context]:
        # 1. 关键词 + 时间识别（用 LLM 做意图分类）
        intent = self._classify_intent(question)
        # 2. 按 intent 类型分发
        if intent.type == 'generation_query':
            return self._fetch_generation_context(intent.time_range)
        elif intent.type == 'cost_savings':
            return self._fetch_cost_context(intent.time_range)
        elif intent.type == 'comparison':
            return self._fetch_comparison_context(intent.periods)
        ...

用例五：设备健康度评估

健康指数模型

def compute_health_index(device: Device) -> float:
    weights = {
        'runtime_ratio': 0.20,
        'alert_frequency': 0.30,
        'performance_decay': 0.30,
        'maintenance_history': 0.10,
        'environmental_stress': 0.10,
    }
    scores = {
        'runtime_ratio': _runtime_score(device),
        'alert_frequency': _alert_score(device),
        'performance_decay': _decay_score(device),
        'maintenance_history': _maint_score(device),
        'environmental_stress': _env_score(device),
    }
    return sum(scores[k] * weights[k] for k in weights)

def _runtime_score(device: Device) -> float:
    expected_lifetime_h = device.spec.expected_lifetime_hours
    return max(0, 100 * (1 - device.total_runtime_h / expected_lifetime_h))

剩余寿命预测

def predict_remaining_years(device: Device, current_health: float, decay_rate: float) -> float:
    # 按线性退化 + 当前健康度推算
    threshold = 50  # 健康指数 < 50 视为需要更换
    remaining_points = current_health - threshold
    return max(0, remaining_points / decay_rate)

部署架构

[Web Admin] ──API──→ [Agent Server (Python)]
                          │
                          ├─ LLM Provider (Anthropic / OpenAI / Qwen)
                          ├─ Vector DB (Chroma / PGVector)
                          ├─ TSDB (InfluxDB / TDengine)
                          ├─ PG (告警 / 工单 / 设备)
                          └─ Cache (Redis)

[Cron Worker] ──→ daily report, weekly inspection plan, monthly health eval
[Stream Worker] ──→ real-time alert clustering
[Webhook] ──→ owner Q&A bot

评估集设计

至少包含：

• 100 条已标注根因的告警集群
• 50 条业主问答典型 query
• 30 个设备健康度历史 case
• 20 份高质量人工日报作为对比

按周或按月跑一次评估，看：

• 准确率
• 幻觉率
• 用户采纳率（业主继续追问的比例）

小结

AI Agent 在光伏电站运营的落地路径是「替人做重复工作，把决策留给人」。

本文 5 个用例（日报 / 告警归因 / 巡检调度 / 业主问答 / 设备健康度）均已在生产环境运行。建议团队按"高频低价值"优先级排序，从日报、告警归因这两个 ROI 较高的开始。

参考

• IEC 61724 光伏系统性能监测
• LangChain / LlamaIndex tool calling 文档
• 各家时序数据库性能对比报告