这篇能做出什么
做完这套东西，你的推理服务会多出三个能力：
1. 拿到边际碳强度：从 Electricity Maps、WattTime 这类数据源按区域拉取"边际排放因子"（MOER，marginal operating emissions rate），统一折算成 gCO2eq/kWh，带缓存和过期兜底。
2. 按碳强度调度：一个请求进来，网关在满足延迟 SLO 的候选区域里挑一个碳强度相对较低的区域去跑，同时避免在两个区域之间反复横跳。
3. 输出单请求碳账本：每个请求返回一个 request_id，凭它可以查到这一笔推理消耗了多少 Wh 电、按当时的碳强度折算出多少克碳排放，以及这个数字是拿哪些原始参数、按什么公式算出来的——别人可以拿同样的参数复算一遍，结果对得上。
为什么用边际碳强度而不是平均碳强度？举个具体例子：
- 区域 A：平均 300 g/kWh（煤电为主），但此刻风电富余，边际机组是燃气，边际 120 g/kWh。
- 区域 B：平均 200 g/kWh（水电占比大），但水库已经跑满，你多要一度电就得多开一台煤机，边际 650 g/kWh。
如果你按平均值选，会把请求送到 B，实际反而多排了碳。路由改变的是"新增的那一度电"，所以边际信号才和你的决策对得上。这就是整套系统从数据层就要强调"is_marginal"的原因。
下面全部代码都可以离线跑通：默认用一张手写的静态碳强度表做数据源，等拿到 API token 再切到真实数据源。
前置条件清单
- Python 3 及
fastapi、uvicorn、httpx、pydantic（版本以官方文档当前版本为准）。 - 两个以上可承载推理的区域，每个区域记录三样东西：P95 延迟、单卡运行功率（W）、PUE。没有真集群就先用假数据，把链路跑通。
- 一份碳强度数据源的凭据：Electricity Maps 或 WattTime 的 token，放进环境变量
CARBON_API_TOKEN，不要写进代码。接口地址、参数名、响应字段、计费方式都以官方文档当前版本为准。 - 能读到功率的地方：裸金属可以采样
nvidia-smi，Kubernetes 上可以用 DCGM exporter；读不到就先用厂商规格里的功率上限做保守估计，并在账本里标清楚这是估算值。 - 一个 SQLite 文件足够存账本；量大再换 Postgres。
建议的目录：
```text
carbon_router/
├── intensity.py # 碳强度接入层
├── routing.py # 路由策略
├── ledger.py # 碳账本模型 + 存储
└── app.py # FastAPI 入口
```
分步骤
第 1 步：把碳强度接入层做成可替换的适配器
不同数据源的响应结构差别很大：有的直接给 gCO2eq/kWh，有的给 lbs CO2/MWh。所以适配器不要写死字段，用可配置的 JSON 路径 + 单位来解耦。
```python
intensity.py
from __future__ import annotations
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any
import httpx
LB_TO_GRAM = 453.59237
def to_g_per_kwh(value: float, unit: str) -> float:
"""把数据源返回的碳强度统一折算成 gCO2eq/kWh。"""
u = unit.strip().lower().replace(" ", "")
if u in ("g/kwh", "gco2/kwh", "gco2eq/kwh", "kg/mwh"):
return float(value)
if u in ("lb/mwh", "lbs/mwh", "lbco2/mwh"):
return float(value) * LB_TO_GRAM / 1000.0
raise ValueError(f"未知碳强度单位: {unit}")
def dig(data: Any, path: str) -> Any:
"""按 'a.b.0.c' 这样的路径取值，避免把某家 API 的响应结构写死。"""
cur = data
for part in path.split("."):
cur = cur[int(part)] if isinstance(cur, list) else cur[part]
return cur
@dataclass(frozen=True)
class CarbonPoint:
region: str
g_per_kwh: float
ts: datetime
source: str
is_marginal: bool
class CarbonSource:
name = "base"
def __init__(self, mapping: dict[str, str]):
self.mapping = mapping # 业务区域名 -> 数据源里的区域 / BA 标识
async def fetch(self, region: str) -> CarbonPoint:
raise NotImplementedError
class HttpCarbonSource(CarbonSource):
"""通用 HTTP 适配器：URL、参数、鉴权头、JSON 路径、单位全部走配置。"""
def __init__(self, name, url, mapping, json_path, unit,
params=None, headers=None, timeout=5.0):
super().__init__(mapping)
self.name = name
self.url = url
self.json_path = json_path
self.unit = unit
self.params = params or {}
self.headers = headers or {}
self.timeout = timeout
async def fetch(self, region: str) -> CarbonPoint:
params = dict(self.params)
params["region"] = self.mapping[region] # 参数名按数据源文档调整
async with httpx.AsyncClient(timeout=self.timeout) as client:
resp = await client.get(self.url, params=params, headers=self.headers)
resp.raise_for_status()
payload = resp.json()
raw = dig(payload, self.json_path)
return CarbonPoint(
region=region,
g_per_kwh=to_g_per_kwh(raw, self.unit),
ts=datetime.now(timezone.utc),
source=self.name,
is_marginal=True,
)
class StaticSource(CarbonSource):
"""离线开发用：一张手写表，单位 gCO2eq/kWh。"""
name = "static"
def __init__(self, table: dict[str, float]):
super().__init__({r: r for r in table})
self.table = table
self.cursor = 0
async def fetch(self, region: str) -> CarbonPoint:
self.cursor += 1
jitter = ((hash((region, self.cursor)) % 21) - 10) / 100.0
return CarbonPoint(
region=region,
g_per_kwh=self.table[region] * (1 + jitter),
ts=datetime.now(timezone.utc),
source=self.name,
is_marginal=True,
)
```
注意两点：is_marginal 这个字段必须显式记录。数据源里混着平均值和边际值，账本里写不清来源，后面所有折算都不可信。
第 2 步：给接入层加缓存和过期兜底
碳强度不需要每个请求都拉一次，几百个请求共用一个 5 分钟窗口的值完全够用。但数据源会抖、会限流、会超时，所以要允许"用旧值"。
```python
intensity.py（续）
class IntensityCache:
def __init__(self, source: CarbonSource, ttl: int = 300, max_stale: int = 3600):
self.source = source
self.ttl = ttl
self.max_stale = max_stale
self._points: dict[str, tuple[float, CarbonPoint]] = {}
async def get(self, region: str) -> CarbonPoint:
now = time.monotonic()
hit = self._points.get(region)
if hit and now - hit[0] < self.ttl:
return hit[1]
try:
point = await self.source.fetch(region)
self._points[region] = (now, point)
return point
except Exception as exc:
if hit and now - hit[0] < self.max_stale:
return hit[1] # 数据源挂了，用最近一次有效值，账本里会带上旧时间戳
raise RuntimeError(f"{region} 碳强度不可用: {exc}") from exc
```
关键设计：兜底返回旧值时不要改时间戳。账本记录的是 ts，用户一看就知道这条数据是哪一刻的，能不能接受由他判断。
第 3 步：路由决策——先过滤，再打分
不要在"碳最低"和"延迟最低"之间做二选一。正确顺序是：先用延迟 SLO 过滤掉不能用的区域，再在剩下的区域里按碳强度排序。
```python
routing.py
from dataclasses import dataclass
from intensity import CarbonPoint, IntensityCache
@dataclass
class RegionProfile:
name: str
p95_latency_ms: float
gpu_power_w: float
pue: float
@dataclass
class Decision:
region: str
reason: str
point: CarbonPoint
candidates: dict[str, float]
async def choose_region(
cache: IntensityCache,
profiles: dict[str, RegionProfile],
slo_ms: float,
current: str | None = None,
hysteresis: float = 0.08,
) -> Decision:
1) 延迟过滤
feasible = [r for r, p in profiles.items() if p.p95_latency_ms <= slo_ms]
if not feasible:
feasible = list(profiles) # 全都不达标时退化为全部候选，并在 reason 里说明
2) 拉碳强度
points: dict[str, CarbonPoint] = {}
for r in feasible:
try:
points[r] = await cache.get(r)
except RuntimeError:
continue
if not points:
raise RuntimeError("没有可用区域的碳强度数据")
3) 排序
ranked = sorted(points, key=lambda r: points[r].g_per_kwh)
best = ranked[0]
4) 迟滞：当前区域如果和最好区域差距在阈值内，就不切
if current in points:
gap = (points[current].g_per_kwh - points[best].g_per_kwh) / max(points[best].g_per_kwh, 1e-6)
if gap <= hysteresis:
return Decision(current, "keep-current(within-hysteresis)",
points[current], {r: points[r].g_per_kwh for r in ranked})
reason = "min-carbon" if len(feasible) == len(profiles) else "min-carbon-within-slo"
return Decision(best, reason, points[best],
{r: points[r].g_per_kwh for r in ranked})
```
hysteresis=0.08 的意思是：新区域比当前区域只低不到 8% 就不值得切。没有这一步，碳强度曲线一抖动，请求就会在两区域之间来回弹，冷启动和缓存未命中带来的额外开销可能比省下的碳还多。
第 4 步：定义单请求碳账本的字段
账本要"可核对"，秘诀只有一条：把公式里的每一个输入都存下来，把假设显式写出来。
```python
ledger.py
import json
import sqlite3
from dataclasses import asdict, dataclass
METHOD = "gpu-power-x-duration/pue/marginal/v1"
@dataclass
class LedgerEntry:
request_id: str
region: str
started_at: str # ISO8601 UTC
duration_s: float # 该请求占用加速卡的墙钟时间
gpu_power_w: float # 运行功率假设值
pue: float # 区域 PUE 假设值
share: float # 批处理分摊比例，1.0 表示独占
energy_wh: float
intensity_g_per_kwh: float
intensity_source: str
intensity_ts: str
is_marginal: bool
emission_g: float
method: str = METHOD
def build_entry(request_id, region, started_at, duration_s, profile,
point, share=1.0) -> LedgerEntry:
energy_wh = profile.gpu_power_w * duration_s / 3600.0 * profile.pue * share
emission_g = energy_wh / 1000.0 * point.g_per_kwh
return LedgerEntry(
request_id=request_id,
region=region,
started_at=started_at,
duration_s=round(duration_s, 6),
gpu_power_w=profile.gpu_power_w,
pue=profile.pue,
share=round(share, 6),
energy_wh=round(energy_wh, 6),
intensity_g_per_kwh=round(point.g_per_kwh, 3),
intensity_source=point.source,
intensity_ts=point.ts.isoformat(),
is_marginal=point.is_marginal,
emission_g=round(emission_g, 6),
)
def recompute(entry: dict) -> float:
"""任何人拿账本里的字段都能复算一遍，用来做核对。"""
energy_wh = (entry["gpu_power_w"] * entry["duration_s"]
/ 3600.0 * entry["pue"] * entry["share"])
return energy_wh / 1000.0 * entry["intensity_g_per_kwh"]
```
公式本身很朴素：功率 × 时间 × PUE × 分摊比例 = 电量，电量 × 边际碳强度 = 排放。它的价值不在于精确，而在于口径固定、逐项可查。功率取的是规格上限还是实测采样，要在 gpu_power_w 的取值来源上写清楚；这决定了账本是"保守上界"还是"实测估算"，两种都能用，但不能混着用还不标注。
第 5 步：把账本落库并做核对接口
```python
ledger.py（续）
SCHEMA = """
CREATE TABLE IF NOT EXISTS ledger (
request_id TEXT PRIMARY KEY,
region TEXT NOT NULL,
started_at TEXT NOT NULL,
duration_s REAL NOT NULL,
gpu_power_w REAL NOT NULL,
pue REAL NOT NULL,
share REAL NOT NULL,
energy_wh REAL NOT NULL,
intensity_g_per_kwh REAL NOT NULL,
intensity_source TEXT NOT NULL,
intensity_ts TEXT NOT NULL,
is_marginal INTEGER NOT NULL,
emission_g REAL NOT NULL,
method TEXT NOT NULL,
payload TEXT NOT NULL
);
"""
def save(conn: sqlite3.Connection, entry: LedgerEntry) -> None:
row = asdict(entry)
conn.execute(
f"INSERT INTO ledger ({','.join(row)}) "
f"VALUES ({','.join('?' * len(row))})",
list(row.values()),
)
conn.commit()
def load(conn: sqlite3.Connection, request_id: str) -> dict | None:
cur = conn.execute("SELECT * FROM ledger WHERE request_id = ?", (request_id,))
cols = [c[0] for c in cur.description]
row = cur.fetchone()
return dict(zip(cols, row)) if row else None
def summary(conn: sqlite3.Connection) -> list[dict]:
cur = conn.execute(
"SELECT region, COUNT(*) AS n, SUM(emission_g) AS g, "
"SUM(energy_wh) AS wh FROM ledger GROUP BY region ORDER BY g DESC"
)
cols = [c[0] for c in cur.description]
return [dict(zip(cols, r)) for r in cur.fetchall()]
```
第 6 步：组装成 HTTP 服务
```python
app.py
import asyncio
import sqlite3
import time
import uuid
from datetime import datetime, timezone
from fastapi import FastAPI, HTTPException, Request
from intensity import HttpCarbonSource, IntensityCache, StaticSource
from ledger import SCHEMA, build_entry, load, recompute, save, summary
from routing import RegionProfile, choose_region
PROFILES = {
"cn-north-1": RegionProfile("cn-north-1", p95_latency_ms=180, gpu_power_w=400, pue=1.35),
"cn-east-1": RegionProfile("cn-east-1", p95_latency_ms=90, gpu_power_w=400, pue=1.25),
"cn-south-1": RegionProfile("cn-south-1", p95_latency_ms=140, gpu_power_w=400, pue=1.30),
}
有 token 用真实数据源，没有就用静态表跑通链路
source = StaticSource({"cn-north-1": 620, "cn-east-1": 480, "cn-south-1": 210})
cache = IntensityCache(source, ttl=300, max_stale=3600)
conn = sqlite3.connect("ledger.db", check_same_thread=False)
conn.execute(SCHEMA)
conn.commit()
app = FastAPI()
@app.post("/v1/chat")
async def chat(request: Request):
body = await request.json()
slo_ms = float(body.get("slo_ms", 200))
tokens = int(body.get("max_tokens", 256))
decision = await choose_region(
cache, PROFILES, slo_ms=slo_ms,
current=request.headers.get("x-current-region"),
)
started_at = datetime.now(timezone.utc).isoformat()
t0 = time.perf_counter()
await asyncio.sleep(0.25) # 占位：替换为真实推理调用
duration_s = time.perf_counter() - t0
profile = PROFILES[decision.region]
request_id = f"req_{uuid.uuid4().hex[:16]}"
entry = build_entry(request_id, decision.region, started_at,
duration_s, profile, decision.point)
save(conn, entry)
return {
"request_id": request_id,
"region": decision.region,
"reason": decision.reason,
"carbon": {
"emission_g": entry.emission_g,
"energy_wh": entry.energy_wh,
"intensity_g_per_kwh": entry.intensity_g_per_kwh,
"intensity_source": entry.intensity_source,
"intensity_ts": entry.intensity_ts,
"is_marginal": entry.is_marginal,
"method": entry.method,
},
}
@app.get("/ledger/{request_id}")
async def get_ledger(request_id: str):
row = load(conn, request_id)
if not row:
raise HTTPException(status_code=404, detail="not found")
row["recomputed_emission_g"] = round(recompute(row), 6)
row["consistent"] = abs(row["recomputed_emission_g"] - row["emission_g"]) < 1e-6
return row
@app.get("/ledger")
async def get_summary():
return summary(conn)
```
第 7 步：跑一遍，验证账本能对上
```bash
bash
uvicorn app:app --reload --port 8000
```
另开一个终端：
```bash
bash
curl -s -X POST http://localhost:8000/v1/chat \
-H 'content-type: application/json' \
-d '{"slo_ms": 200, "max_tokens": 256}'
curl -s http://localhost:8000/ledger/req_xxxxxxxxxxxxxxxx
```
第二条命令返回里会有一个 recomputed_emission_g 和 consistent: true。这一步就是"可核对"的落地方式：账本不是一段自说自话的日志，而是一组能被第三方代入公式复算的参数。
再把 slo_ms 从 200 调到 100，观察路由结果变化——延迟不达标的区域会被过滤掉，reason 字段会从 min-carbon 变成 min-carbon-within-slo，碳排放数字也随之上升。这个对比就是碳感知路由的全部价值所在：用可度量的方式展示"延迟换碳"的取舍。
切到真实数据源时，只改一行：
```python
python
source = HttpCarbonSource(
name="electricity-maps",
url="<按官方文档填写>",
mapping={"cn-north-1": "<区域标识>", "cn-east-1": "<区域标识>", "cn-south-1": "<区域标识>"},
json_path="<响应中碳强度字段的路径>",
unit="g/kwh",
headers={"auth-token": os.environ["CARBON_API_TOKEN"]},
)
```
具体 URL、区域标识、字段路径和鉴权头名称，都以数据源官方文档当前版本为准。
常见坑与排错
单位换算错一千倍。 最常见的就是把 lbs CO2/MWh 当成 g/kWh 直接用了。1 磅约等于 453.59 克，1000 lbs/MWh ≈ 453.59 g/kWh。看到账本里单请求排放是几百克而不是零点几克，先查单位。
平均值和边际值混用。 数据源可能同时提供历史平均、当前边际、未来边际预测三类数据。混着用会导致同一份报表里两个口径的数字相加。解决办法是强制在 CarbonPoint 里带 is_marginal，并在汇总时按它分组。
时区和时间戳。 全部用 UTC 存储、接口返回 ISO8601 带时区。碳强度数据的窗口通常是 5 分钟或 1 小时粒度，如果你的账本时间是本地时间且没有偏移量，跨时区核对时永远差几个小时。
路由抖动。 没有迟滞参数时，两个区域碳强度交叉一次就切一次，冷启动、连接池、KV 缓存全部失效。用 hysteresis 加一个百分比阈值，或者限定"每小时最多切换 N 次"。
数据源限流。 按区域拉取，别按请求拉取。缓存 TTL 设到 5 分钟以上，多区域并发拉取时用信号量限并发。被限流后要有降级路径，而不是整个网关 500。
账本里只有结论没有参数。 只存了 emission_g 而没有 gpu_power_w、duration_s、pue、share、method，这份账本三个月后没人能解释。方法版本号 method 一定要有，公式一改就要升版本，否则历史数据不可比。
功率假设前后不一致。 上午用规格上限 400W、下午改成实测 210W，同一份月度报表里的数字就不可加。要么全用保守上界并在报表里注明是上界，要么全用实测采样，别混。
批处理没做分摊。 一个 batch 里 8 个请求，如果每个都按整段 GPU 时间记账，总量会翻 8 倍。用 token 数或序列长度做 share，并保证同一批次内 share 之和等于 1。
下一步建议
- 接入预测值做前瞻调度：边际碳强度数据源通常提供未来若干小时的预测。对延迟不敏感的离线推理、批量重排、向量重建这类任务，可以排队等到预测低谷再跑，收益比实时路由大得多。
- 把账本接到真实计量上：用 DCGM 或
nvidia-smi按秒采样功率，替换掉常数假设，账本精度会提升一个量级，同时保留"假设模式"作为对照口径。 - 区分可延迟与不可延迟流量：给请求打上
deadline或flexible标签，硬实时走低延迟优先，柔性流量走碳优先，两条策略共用一套账本。 - 做月度对账：按区域、按模型、按租户聚合
emission_g，和云厂商提供的碳报告做量级比对。对不上时差异通常是 PUE 口径和功率假设，这个排查过程本身就能帮你把口径定清楚。 - 考虑碳强度之外的成本：区域之间的电价、出口流量费、模型副本部署成本都会影响最终决策。把成本作为第二目标写进打分函数，先过滤、再加权，结构不用改。