グリーンソフトウェアエンジニアリング: 2026年のAIワークロードにおけるエネルギー効率の最適化

pip install codecarbon

from codecarbon import EmissionsTracker
import time

tracker = EmissionsTracker(
    project_name="my-llm-inference",
    output_dir="./carbon-logs",
    log_level="warning",
)

tracker.start()

# Your AI workload here
response = llm_client.chat.completions.create(
    model="mistral-7b-instruct",
    messages=[{"role": "user", "content": "Explain async Python in 3 sentences."}],
)

emissions = tracker.stop()
print(f"Carbon emitted: {emissions * 1000:.4f} gCO2eq")
# Output: Carbon emitted: 0.0023 gCO2eq

# Install and connect to AWS
npm install -g @cloud-carbon-footprint/cli
ccf --startDate 2026-08-01 --endDate 2026-08-31 --groupBy service

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig

quantization_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_compute_dtype=torch.bfloat16,
    bnb_4bit_use_double_quant=True,    # nested quantization for extra compression
    bnb_4bit_quant_type="nf4",         # NormalFloat4 — better accuracy than int4
)

model = AutoModelForCausalLM.from_pretrained(
    "mistralai/Mistral-7B-Instruct-v0.2",
    quantization_config=quantization_config,
    device_map="auto",
)
tokenizer = AutoTokenizer.from_pretrained("mistralai/Mistral-7B-Instruct-v0.2")

# Verify memory savings
print(f"Model memory footprint: {model.get_memory_footprint() / 1e9:.2f} GB")
# Output: Model memory footprint: 3.83 GB  (vs 14 GB at FP16)

from evaluate import load
import json

# Load your test set
with open("test_prompts.json") as f:
    test_cases = json.load(f)

# Evaluate both models
for model_name, model in [("fp16", fp16_model), ("int4", int4_model)]:
    correct = 0
    for case in test_cases:
        output = generate(model, case["prompt"])
        if case["expected"] in output:
            correct += 1
    print(f"{model_name}: {correct}/{len(test_cases)} correct ({correct/len(test_cases)*100:.1f}%)")

# Typical output:
# fp16: 47/50 correct (94.0%)
# int4: 46/50 correct (92.0%)

import asyncio
from dataclasses import dataclass, field
from typing import Any
import time

@dataclass
class BatchProcessor:
    model: Any
    max_batch_size: int = 32
    max_wait_ms: float = 50.0   # max 50ms queue wait — acceptable latency increase
    _queue: asyncio.Queue = field(default_factory=asyncio.Queue)

    async def infer(self, prompt: str) -> str:
        """Add to batch queue and wait for result."""
        future: asyncio.Future = asyncio.get_event_loop().create_future()
        await self._queue.put((prompt, future))
        return await future

    async def _batch_loop(self):
        """Drain queue in batches for energy-efficient inference."""
        while True:
            batch = []
            deadline = time.monotonic() + self.max_wait_ms / 1000

            # Collect items until batch full or deadline reached
            while len(batch) < self.max_batch_size:
                timeout = deadline - time.monotonic()
                if timeout <= 0:
                    break
                try:
                    item = await asyncio.wait_for(self._queue.get(), timeout=timeout)
                    batch.append(item)
                except asyncio.TimeoutError:
                    break

            if not batch:
                await asyncio.sleep(0.001)
                continue

            # Process batch in one GPU pass
            prompts = [item[0] for item in batch]
            futures = [item[1] for item in batch]

            outputs = self.model.generate_batch(prompts)   # single GPU call
            for future, output in zip(futures, outputs):
                future.set_result(output)

import boto3

# Use AWS Spot in Oregon (low carbon) for batch embedding jobs
ec2 = boto3.client("ec2", region_name="us-west-2")

spot_response = ec2.request_spot_instances(
    InstanceCount=1,
    LaunchSpecification={
        "ImageId": "ami-0abcdef1234567890",
        "InstanceType": "g4dn.xlarge",   # GPU instance
        "KeyName": "my-key",
    },
    SpotPrice="0.50",   # max price/hr
)

import httpx

async def get_grid_intensity(zone: str = "US-CAL-CISO") -> float:
    """Returns gCO2eq/kWh for the given grid zone."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"https://api.electricitymap.org/v3/carbon-intensity/latest?zone={zone}",
            headers={"auth-token": "YOUR_TOKEN"},
        )
        return resp.json()["carbonIntensity"]

async def should_run_batch_job(threshold_gco2: float = 200.0) -> bool:
    """Only run non-urgent batch jobs when the grid is clean."""
    intensity = await get_grid_intensity()
    print(f"Current grid: {intensity:.0f} gCO2eq/kWh (threshold: {threshold_gco2})")
    return intensity < threshold_gco2

# In your batch job scheduler:
if await should_run_batch_job():
    await run_embedding_backfill()
else:
    print("Grid too carbon-intensive, deferring to next window")
    await schedule_retry_in(hours=2)

# .github/workflows/carbon-check.yml
name: Carbon Budget Check

on: [pull_request]

jobs:
  carbon:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Run inference benchmark with carbon tracking
        run: |
          pip install codecarbon
          python scripts/benchmark_inference.py --track-carbon
        
      - name: Check carbon budget
        run: |
          python scripts/check_carbon_budget.py \
            --budget-gco2-per-1k-tokens 0.5 \
            --results carbon-logs/emissions.csv

# scripts/check_carbon_budget.py
import pandas as pd
import sys
import argparse

parser = argparse.ArgumentParser()
parser.add_argument("--budget-gco2-per-1k-tokens", type=float, required=True)
parser.add_argument("--results", required=True)
args = parser.parse_args()

df = pd.read_csv(args.results)
actual = df["emissions_kg"].iloc[-1] * 1000 * 1000  # kg → g, then per 1k tokens

print(f"Carbon per 1k tokens: {actual:.4f} gCO2eq (budget: {args.budget_gco2_per_1k_tokens})")

if actual > args.budget_gco2_per_1k_tokens:
    print(f"❌ Carbon budget exceeded by {actual - args.budget_gco2_per_1k_tokens:.4f} gCO2eq")
    sys.exit(1)
else:
    print("✅ Within carbon budget")

from anthropic import Anthropic

client = Anthropic()

def classify_query_complexity(query: str) -> str:
    """Route to the smallest model that can handle this query."""
    word_count = len(query.split())
    has_code = "```" in query or "def " in query or "import " in query
    
    if word_count < 20 and not has_code:
        return "claude-haiku-3-5"      # ~10x cheaper + greener than Sonnet
    elif word_count < 100:
        return "claude-sonnet-4-5"
    else:
        return "claude-opus-4-5"       # only for genuinely complex requests

def smart_completion(prompt: str) -> str:
    model = classify_query_complexity(prompt)
    response = client.messages.create(
        model=model,
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}],
    )
    print(f"Routed to {model}")
    return response.content[0].text