引言:算力的尽头是电力
"人工智能的尽头是算力,算力的尽头是电力。"这句曾在业界广为流传的判断,在2026年终于获得了政策层面的正式确认。
2026年政府工作报告首次明确提出实施"算电协同"新基建工程,“十五五"规划纲要亦将推动绿色电力与算力协同布局列为核心任务。2026年4月8日,国家发展改革委、国家能源局、工业和信息化部、国家数据局四部门联合发布《关于促进人工智能与能源双向赋能的行动方案》,标志着算电协同正式从行业探索升级为国家战略统筹。
这一战略转向背后有着深刻的数据支撑:2023年至2025年,我国算力需求负荷年均增长率高达28%;2025年全国算力中心用电量达1960亿千瓦时,同比增长18.1%;预计2030年数据中心用电量占比将从当前的约3%攀升至5%以上。与此同时,全国风电、光伏总装机容量已分别达到640GW和1200GW,但发电量占比仅为22%,弃风弃光问题亟需稳定负荷消纳。
算电协同,正是破解这道双向难题的密钥。
一、算电协同的三层技术架构
算电协同并非简单的"给数据中心拉电线”,而是一个涵盖能源层、传输层、负载层的系统性工程。理解这一架构,是把握算电协同技术本质的关键。
1.1 能源层:绿电直供的多元模式
能源层解决的是"电从何处来"的问题。当前绿电供给已形成三种主流模式:
模式一:物理直供(专线供电)
增量负荷新建专属110kV线路,实现新能源电场到数据中心的点对点直送。这种模式最大程度减少了输电损耗和中间环节,是绿电利用率最高的形式。
宁夏中卫的标杆项目即为典型案例:大唐中卫云基地数据中心绿电供应项目,总装机200万千瓦(50万千瓦光伏+150万千瓦风电),通过专属线路直供数据中心,年均可提供22.9亿千瓦时稳定绿电。该项目电价约0.36元/千瓦时,较东部工业电价低0.25-0.4元,降幅达30%-50%。
模式二:双边交易(虚拟直供)
存量负荷依托电力市场,通过绿电双边交易实现虚拟直供。企业与新能源项目签订长期购电协议(PPA),绿证与电力分离输送,保障新能源项目获得稳定收益预期。
模式三:聚合供应
通过"光伏直连+风电交易+电网备容"的组合模式,中卫数据中心集群绿电占比稳定超过80%。这种模式兼具灵活性与稳定性,是当前推广最广的方案。

绿电供给成本优化模型(示例代码)

import pandas as pd
from scipy.optimize import linprog
import numpy as np

def optimize_green_power_supply(
    load_profile: np.ndarray,  # 小时级负荷曲线 (24小时)
    solar_capacity: float,     # 光伏装机容量 (MW)
    wind_capacity: float,      # 风电装机容量 (MW)
    storage_capacity: float,   # 储能容量 (MWh)
    grid_price: float,         # 电网电价 (元/kWh)
    green_power_price: float   # 绿电协议价 (元/kWh)
) -> dict:
    """
    算电协同场景下的绿电供给优化模型
    目标:最小化购电总成本
    约束:负荷平衡、储能SOC、新能源出力上限
    """
    hours = len(load_profile)
    
    # 生成新能源出力曲线(示例用正弦曲线模拟)
    solar_gen = solar_capacity * np.sin(np.linspace(0, np.pi, hours))
    wind_gen = wind_capacity * np.abs(np.sin(np.linspace(0, 2*np.pi, hours)))
    
    # 决策变量定义 [grid_buy, storage_charge, storage_discharge] × 24小时
    num_vars = hours * 3
    
    # 目标函数系数(最小化购电成本)
    c = np.zeros(num_vars)
    c[:hours] = grid_price  # 电网购电成本
    c[hours:2*hours] = 0    # 充电成本(假设为0)
    c[2*hours:] = -green_power_price  # 放电收益
    
    # 等式约束:负荷平衡
    A_eq = np.zeros((hours, num_vars))
    for t in range(hours):
        A_eq[t, t] = 1              # 电网购电
        A_eq[t, 2*hours + t] = -1   # 储能放电
    b_eq = load_profile - solar_gen - wind_gen
    
    # 不等式约束:储能SOC管理
    A_ub = np.zeros((2*hours, num_vars))
    b_ub = np.zeros(2*hours)
    for t in range(hours):
        # 充电约束
        A_ub[t, hours + t] = 1
        A_ub[t, 2*hours + t] = 1
        b_ub[t] = storage_capacity
        
        # 放电约束
        A_ub[hours + t, 2*hours + t] = -1
        b_ub[hours + t] = 0
    
    # 变量边界约束
    bounds = [(0, None) for _ in range(num_vars)]  # 所有变量非负
    
    # 求解线性规划
    res = linprog(c, A_ub=A_ub, b_ub=b_ub, A_eq=A_eq, b_eq=b_eq, bounds=bounds)
    
    return {
        'total_cost': res.fun,
        'grid_purchase': res.x[:hours],
        'storage_charge': res.x[hours:2*hours],
        'storage_discharge': res.x[2*hours:],
        'solar_generation': solar_gen,
        'wind_generation': wind_gen
    }
 
# 简化模型参数
solar_profile = np.array([0.0, 0.0, 0.0, 0.0, 0.0, 0.1, 0.3, 0.6, 0.8, 0.9, 
                           1.0, 0.9, 0.8, 0.6, 0.4, 0.2, 0.1, 0.0, 0.0, 0.0,
                           0.0, 0.0, 0.0, 0.0])  # 光伏出力曲线(归一化)
wind_profile = np.array([0.6, 0.7, 0.8, 0.7, 0.6, 0.5, 0.4, 0.5, 0.6, 0.7,
                          0.7, 0.6, 0.5, 0.6, 0.7, 0.8, 0.9, 0.9, 0.8, 0.7,
                          0.7, 0.6, 0.6, 0.7])  # 风电出力曲线(归一化)

# 优化变量:各时段购电量
# 目标函数系数(成本)
c = np.array([green_power_price if i % 2 == 0 else grid_price 
              for i in range(hours)])

# 约束条件矩阵
A_ub = None
b_ub = None

# 等式约束:负荷平衡
# 绿电 + 储能放电 + 电网购电 = 负荷
A_eq = np.eye(hours)
b_eq = load_profile

# 变量边界:购电量 >= 0
bounds = [(0, None) for _ in range(hours)]

result = linprog(c, A_ub=A_ub, b_ub=b_ub, A_eq=A_eq, b_eq=b_eq, 
                 bounds=bounds, method='highs')

return {
    'optimal_purchase': result.x,
    'total_cost': result.fun,
    'green_power_usage': 1 - np.sum(result.x) / np.sum(load_profile)
}

1.2 传输层:算电协同的智能调度
传输层解决的是"电如何高效流动"的问题。传统电网是单向的"电跟算走",而在算电协同模式下,算力负荷成为可调度的柔性资源,实现"算随电动"的双向互动。
核心机制一:算力负荷预测与调度
算力任务具有天然的可中断、可迁移、可调度特征。通过机器学习算法预测算力负载与电力价格的时序关系,可在电价低谷时段调度更多计算任务,电价高峰时段降低负载参与需求响应。

算力-电力协同调度优化框架

from typing import Tuple, List, Dict
import numpy as np
from dataclasses import dataclass
from heapq import heappush, heappop

@dataclass
class ComputeTask:
    """算力任务描述"""
    task_id: str
    required_flops: float       # 所需算力 (PFLOPS)
    deadline: int                # 截止时间 (小时)
    priority: int                # 优先级 (1-10)
    flexible: bool               # 是否可延迟

@dataclass
class PowerSlot:
    """电力时隙"""
    hour: int
    grid_price: float           # 电网价格 (元/kWh)
    green_power_available: float  # 可用绿电 (MWh)
    carbon_intensity: float      # 碳排放强度 (gCO2/kWh)

class ComputePowerScheduler:
    """
    算电协同调度器
    核心功能:
    - 基于电力价格和碳排放的加权成本模型
    - 支持任务优先级调度
    - 支持可延迟任务的弹性调度
    """
    
    def __init__(self, power_slots: List[PowerSlot], 
                 carbon_weight: float = 0.3,
                 price_weight: float = 0.7):
        self.power_slots = sorted(power_slots, key=lambda x: x.hour)
        self.carbon_weight = carbon_weight
        self.price_weight = price_weight
        self.schedule = {}
        
    def _calculate_slot_cost(self, slot: PowerSlot) -> float:
        """计算时隙综合成本"""
        normalized_price = slot.grid_price / max(s.grid_price for s in self.power_slots)
        normalized_carbon = slot.carbon_intensity / max(s.carbon_intensity for s in self.power_slots)
        return (self.price_weight * normalized_price + 
                self.carbon_weight * normalized_carbon)
    
    def _get_optimal_slots(self, duration: int) -> List[PowerSlot]:
        """获取最优时间窗口"""
        slot_costs = [(i, self._calculate_slot_cost(slot)) 
                     for i, slot in enumerate(self.power_slots)]
        
        # 滑动窗口寻找最低成本区间
        min_cost = float('inf')
        best_window = []
        
        for i in range(len(slot_costs) - duration + 1):
            window_cost = sum(cost for _, cost in slot_costs[i:i+duration])
            if window_cost < min_cost:
                min_cost = window_cost
                best_window = self.power_slots[i:i+duration]
        
        return best_window
    
    def schedule_task(self, task: ComputeTask) -> Dict[int, float]:
        """
        调度单个任务
        返回: {时隙: 分配算力}
        """
        required_hours = np.ceil(task.required_flops / 1.0)  # 假设1 PFLOPS/h算力
        
        if task.flexible:
            optimal_slots = self._get_optimal_slots(int(required_hours))
        else:
            # 非弹性任务必须在deadline前完成
            available_slots = [s for s in self.power_slots if s.hour <= task.deadline]
            optimal_slots = sorted(
                available_slots[:int(required_hours)],
                key=self._calculate_slot_cost
            )
        
        allocation = {slot.hour: 1.0 for slot in optimal_slots}  # 均匀分配
        self.schedule[task.task_id] = allocation
        return allocation
    
    def batch_schedule(self, tasks: List[ComputeTask]) -> Dict[str, Dict[int, float]]:
        """批量调度任务(按优先级排序)"""
        sorted_tasks = sorted(tasks, 
                            key=lambda t: (-t.priority, t.deadline))
        
        for task in sorted_tasks:
            self.schedule_task(task)
            
        return self.schedule
 
"""

def __init__(self, tasks: List[ComputeTask], power_slots: List[PowerSlot]):
    self.tasks = tasks
    self.power_slots = power_slots
    self.schedule = {}
    
def optimize_schedule(self) -> dict:
    """生成最优调度方案"""
    # 1. 按优先级和截止时间排序任务
    sorted_tasks = sorted(
        self.tasks, 
        key=lambda t: (-t.priority, t.deadline)
    )
    
    # 2. 对每个时隙计算调度价值
    slot_values = []
    for slot in self.power_slots:
        # 综合价值 = 低电价 + 高绿电 + 低碳排
        green_score = slot.green_power_available / max(
            [s.green_power_available for s in self.power_slots]
        )
        price_score = 1 - slot.grid_price / max(
            [s.grid_price for s in self.power_slots]
        )
        carbon_score = 1 - slot.carbon_intensity / max(
            [s.carbon_intensity for s in self.power_slots]
        )
        
        slot_value = 0.4 * green_score + 0.4 * price_score + 0.2 * carbon_score
        slot_values.append(slot_value)
    
    # 3. 贪心调度:将任务分配给最高价值时隙
    self.schedule = {}
    remaining_capacity = {slot.hour: slot.green_power_available 
                           for slot in self.power_slots}
    
    for task in sorted_tasks:
        best_hours = np.argsort(slot_values)[::-1][:5]  # Top 5 时隙
        
        for hour in best_hours:
            slot = self.power_slots[hour]
            estimated_energy = task.required_flops * 1e6 / 1e12  # 简化估算
            
            if remaining_capacity.get(hour, 0) >= estimated_energy:
                if hour not in self.schedule:
                    self.schedule[hour] = []
                self.schedule[hour].append(task)
                remaining_capacity[hour] -= estimated_energy
                break
    
    return self.schedule

def calculate_metrics(self) -> dict:
    """计算调度指标"""
    total_cost = 0
    total_carbon = 0
    green_power_usage = 0
    total_energy = 0
    
    for hour, tasks in self.schedule.items():
        slot = self.power_slots[hour]
        for task in tasks:
            energy = task.required_flops * 1e6 / 1e12
            total_energy += energy
            total_cost += energy * 1000 * slot.grid_price
            total_carbon += energy * 1000 * slot.carbon_intensity
            if slot.green_power_available > 0:
                green_power_usage += min(energy, slot.green_power_available)
    
    return {
        'total_cost': total_cost,
        'total_carbon_kg': total_carbon / 1000,
        'green_power_ratio': green_power_usage / total_energy if total_energy > 0 else 0,
        'task_completion_rate': len(sum(self.schedule.values(), [])) / len(self.tasks)
    }

核心机制二:虚拟电厂聚合调度
数据中心作为稳定的7×24小时运行负载,是电力系统的优质基荷资源。通过虚拟电厂技术聚合后,可参与电网调峰调频,获取辅助服务收益。
上海、杭州等地已有智算中心通过数据异地迁移参与虚拟电厂调峰响应。在贵州贵安新区的南方能源大数据中心,系统可实时监测不同区域数据中心的交易电价,自动将算力任务调度至电价更低地区的服务器。
1.3 负载层:智算中心的能效跃迁
负载层解决的是"如何高效用好每一度电"的问题。智算中心正经历从"粗放耗电"到"精细用能"的关键转型。
PUE管控的量化目标
根据政策要求,2026年起国家枢纽节点新建数据中心须满足:
.西部枢纽:PUE ≤ 1.20
.东部枢纽:PUE ≤ 1.25
.绿电使用比例:≥ 80%
不达标项目将面临不予审批或关停整改的硬约束。
二、PUE优化:从风冷到液冷的技术跃迁
数据中心能源利用效率(PUE)的持续优化,是算电协同的技术核心。当前PUE优化主要聚焦三大技术路径。
2.1 液冷技术:突破散热瓶颈
2026年3月,国内高密算力场景液冷渗透率正式突破78%,AI训练服务器液冷渗透率达74%。液冷从"可选项"演变为高密智算的"必选项",背后是三重因素的共振:
驱动因素一:芯片功耗的指数级攀升

芯片代际与功耗适配性对比

芯片代际 典型功耗 风冷适配性
NVIDIA A100 400-700W 勉强支撑
NVIDIA H100 700W 接近极限
NVIDIA B200 1200W 严重不足
Rubin R300(预计) 4000W 完全失效

单机柜功率密度从传统数据中心的5-10kW飙升至AI集群的120-200kW,传统风冷30kW/柜的散热上限已被彻底突破。
驱动因素二:政策刚性约束
"东数西算"2026年工作要点明确,新建超大型数据中心PUE不得超过1.2,且70%须采用液冷技术。北京、上海等城市已禁止新建风冷数据中心。
驱动因素三:成本拐点到来
液冷与风冷的成本差距已大幅缩小。2026年,冷板式液冷改造成本约为风冷的1.2-1.5倍,单位千瓦冷却能力改造成本约380-520元。以10年运营周期测算,高密场景中液冷全生命周期成本已优于风冷。

液冷技术路线对比

技术路线 市场份额 单柜功率密度 PUE范围 适用场景
冷板式液冷 ~80% 40-100kW 1.10-1.15 主流高密场景、存量改造
单相浸没式 ~15% 100-150kW 1.05-1.10 超高密场景
两相浸没式 ~5% 200-900kW 1.03-1.08 极致高密、Megawatt级

数据中心液冷系统配置示例

class LiquidCoolingSystem:
   def __init__(self, config):
       self.name = config["datacenter"]["name"]
       self.location = config["datacenter"]["location"]
       
       # IT负载配置
       self.total_power = config["datacenter"]["it_load"]["total_power"]
       self.rack_count = config["datacenter"]["it_load"]["rack_count"]
       self.avg_rack_density = config["datacenter"]["it_load"]["avg_rack_density"]
       
       # 冷却系统配置
       cooling_config = config["datacenter"]["cooling_system"]
       self.cooling_type = cooling_config["type"]
       
       # CDU配置
       self.cdu_count = cooling_config["cdu_spec"]["count"]
       self.cdu_capacity = cooling_config["cdu_spec"]["capacity"]
       self.redundancy = cooling_config["cdu_spec"]["redundancy"]
       self.flow_rate = cooling_config["cdu_spec"]["flow_rate"]
       
       # 冷却介质参数
       self.coolant_type = cooling_config["cooling_medium"]["primary"]
       self.supply_temp = cooling_config["cooling_medium"]["supply_temp"]
       self.return_temp = cooling_config["cooling_medium"]["return_temp"]
       
       # 热排放系统
       self.heat_rejection_type = cooling_config["heat_rejection"]["type"]
       self.waterside_economizer = cooling_config["heat_rejection"]["waterside economizer"]
       self.ambient_temp_threshold = cooling_config["heat_rejection"]["ambient_temp_threshold"]
       
       # 监控参数
       self.monitoring_metrics = cooling_config["monitoring"]["metrics"]
       self.alert_thresholds = cooling_config["monitoring"]["alert_thresholds"]

   def calculate_pue(self, actual_power_usage):
       """
       计算实际PUE值
       """
       return actual_power_usage / self.total_power

   def check_cooling_status(self, current_temp, current_flow):
       """
       检查冷却系统状态
       """
       alerts = []
       if current_flow < self.alert_thresholds["flow_rate_low"]:
           alerts.append("流量过低警报")
       if current_temp > self.alert_thresholds["temp_high"]:
           alerts.append("温度过高警报")
       return alerts if alerts else "系统运行正常"

# 配置数据
config = {
   "datacenter": {
       "name": "某智算中心液冷方案",
       "location": "西部枢纽节点",
       "it_load": {
           "total_power": 10000,
           "rack_count": 100,
           "avg_rack_density": 100
       },
       "cooling_system": {
           "type": "冷板式液冷 + 间接蒸发冷却",
           "cdu_spec": {
               "count": 4,
               "capacity": 2500,
               "redundancy": "2N",
               "flow_rate": 180
           },
           "cooling_medium": {
               "primary": "乙二醇溶液 (35%)",
               "supply_temp": 22,
               "return_temp": 32
           },
           "heat_rejection": {
               "type": "自然冷却 + 冷水机组",
               "waterside economizer": True,
               "ambient_temp_threshold": 15
           },
           "monitoring": {
               "metrics": [
                   "flow_rate",
                   "pressure",
                   "temperature",
                   "conductivity",
                   "leak_detection"
               ],
               "alert_thresholds": {
                   "flow_rate_low": 150,
                   "temp_high": 35
               }
           }
       },
       "expected_performance": {
           "pue_target": 1.12,
           "green_power_ratio": 0.85
       }
   }
}

# 实例化冷却系统
cooling_system = LiquidCoolingSystem(config)

# 示例使用
print(f"数据中心名称: {cooling_system.name}")
print(f"冷却系统类型: {cooling_system.cooling_type}")
print(f"当前系统状态: {cooling_system.check_cooling_status(30, 160)}")

2.2 HVDC供电:提升配电效率
高压直流(HVDC)供电方案省去了传统UPS的逆变-整流双重转换环节,系统效率提升3%-5%。国产HVDC方案已成熟,输出电压等级从240V演进至400V+,适配更大功率密度需求。
2.3 AI驱动的能效优化
华为等厂商已部署AI驱动的iCooling算法,通过深度学习预测负载分布、动态调整制冷参数,实测可降低PUE 0.05-0.1。以10MW数据中心测算,年节省电费可达数百万元。
三、数据中心的新角色:虚拟电厂与能源节点
算电协同正在重塑数据中心的角色定位——从被动"用电大户"转变为主动"调节资源"。
3.1 虚拟电厂的商业逻辑
2026年发布的《电力辅助服务市场基本规则》首次明确将算力负荷纳入可调度资源范畴。数据中心可在电网负荷高峰时段通过降低自身负载参与辅助服务交易,获得相应经济补偿。
一个10MW规模的算力集群,通过参与电网需求响应,年额外收益可超200万元。这一收益来源于两方面:
削峰收益:高峰期主动降载获得的补贴
调频收益:快速响应电网频率波动获得的服务费
3.2 源网荷储算一体化
内蒙古和林格尔数据中心集群已建成智能调控平台,实现风、光、储、算实时联动:
发电高峰时,引导储能充电,最大化消纳绿电
发电低谷时,储能快速放电,同时联动电网协同托底
这种"源网荷储算一体化"模式,使数据中心从单一的能源消费者,转变为电力系统的智能调节节点。
四、技术挑战与演进路径
算电协同虽然前景广阔,但在规模化落地过程中仍面临若干技术挑战:
挑战一:绿电波动性与算力需求刚性的矛盾
风光发电具有间歇性和波动性,而智算中心需要7×24小时稳定供电。尽管储能和多能互补可部分缓解这一问题,但完全消除时间错配仍需技术突破。
挑战二:跨区域调度的通信延迟
算力任务跨区域迁移需要毫秒级响应,但当前东西部之间存在8-30ms的传输时延。如何在保证服务质量的前提下实现跨区域算电协同调度,是工程实践的难点。
挑战三:能碳协同核算
绿电使用比例的精确计量、碳排放的可溯源核算,仍缺乏统一标准。2026年发布的《行动方案》明确将推进绿电消费占比统计及碳排放核算工作,但落地细则仍在完善中。
五、未来展望:源网荷储一体化
展望2030年,算电协同将向"源网荷储"四维一体的终极形态演进:
能源侧:核能、氢能等更多清洁能源将纳入算力供给。2026年《行动方案》已明确鼓励探索核电、氢能直连算力设施。
电网侧:特高压和智能电网支撑跨区域绿电调度,实现"电从远方来"与"电从身边来"的双通道供给。
负荷侧:**数据中心全面拥抱液冷,PUE降至1.1以下;算力负荷深度参与电网调度,成为新型电力系统的核心调节资源。
储能侧:储能从"可选配置"升级为"标准配置",专为智算中心配套的MW级储能系统将迎来爆发式增长。
算力侧:**绿电溯源、碳核算等技术成熟,算力真正成为可计量、可定价、可交易的绿色商品。
算电协同,正在重新定义算力与能源的关系——从"消耗"到"协同",从"成本"到"资产"。这不仅是中国AI产业的系统性护城河,更是全球数字经济时代基础设施竞争的核心战场。

参考来源
国家发展改革委等四部门《关于促进人工智能与能源双向赋能的行动方案》(2026年4月)
《2026年政府工作报告》
中国信息通信研究院《全球科学计算行业发展研究报告》
国家数据局日均Token调用量统计数据
国际能源署(IEA)全球数据中心电力消耗预测报告
各省市算电协同试点项目公开资料

Logo

有“AI”的1024 = 2048,欢迎大家加入2048 AI社区

更多推荐