using Microsoft.AI.MachineLearning;
 
// ONNX モデルをNPUで実行
async Task RunOnNPU()
{
    // モデルの読み込み
    var model = await LearningModel.LoadFromFilePath("model.onnx");
 
    // NPU デバイスを指定
    var device = new LearningModelDevice(
        LearningModelDeviceKind.DirectXHighPerformance // NPU優先
    );
    var session = new LearningModelSession(model, device);
 
    // 入力データの準備
    var binding = new LearningModelBinding(session);
    var inputTensor = TensorFloat.CreateFromArray(
        new long[] { 1, 3, 224, 224 },
        imageData
    );
    binding.Bind("input", inputTensor);
 
    // 推論実行
    var result = await session.EvaluateAsync(binding, "inference");
    var output = result.Outputs["output"] as TensorFloat;
 
    Console.WriteLine($"推論完了: {output.GetAsVectorView()[0]}");
}

# Ollama のインストールと実行
 
# モデルのダウンロードと実行
ollama pull llama3.1:8b        # Llama 3.1 8B (4.7GB)
ollama pull gemma2:9b          # Gemma 2 9B (5.4GB)
ollama pull phi3:mini           # Phi-3 Mini 3.8B (2.3GB)
 
# 対話開始
ollama run llama3.1:8b
 
# API経由で利用（REST API）
curl http://localhost:11434/api/generate -d '{
  "model": "llama3.1:8b",
  "prompt": "Pythonでクイックソートを実装してください",
  "stream": false
}'

# Python からOllamaを利用
import ollama
 
# チャット
response = ollama.chat(model='llama3.1:8b', messages=[
    {'role': 'system', 'content': '日本語で簡潔に回答してください。'},
    {'role': 'user', 'content': 'AI PCのNPUとは何ですか？'}
])
print(response['message']['content'])
 
# ストリーミング
for chunk in ollama.chat(
    model='llama3.1:8b',
    messages=[{'role': 'user', 'content': 'Hello'}],
    stream=True
):
    print(chunk['message']['content'], end='', flush=True)

# llama.cpp のビルド（Vulkan GPU対応）
git clone https://github.com/ggerganov/llama.cpp
cd llama.cpp
cmake -B build -DGGML_VULKAN=ON  # GPUバックエンド
cmake --build build --config Release
 
# 量子化モデルの実行（Q4_K_M: 品質とサイズのバランス）
./build/bin/llama-cli \
  -m models/llama-3.1-8b-q4_k_m.gguf \
  -p "AI PCの利点を3つ挙げてください：" \
  -n 256 \
  --n-gpu-layers 33 \    # GPU層数
  --threads 8 \           # CPUスレッド数
  --temp 0.7

# llama-cpp-python で利用
from llama_cpp import Llama
 
llm = Llama(
    model_path="models/llama-3.1-8b-q4_k_m.gguf",
    n_ctx=4096,        # コンテキスト長
    n_gpu_layers=33,   # GPU層（-1で全層GPU）
    n_threads=8,       # CPUスレッド数
    verbose=False
)
 
output = llm(
    "AI PCのNPUについて説明してください。",
    max_tokens=256,
    temperature=0.7,
    stop=["。\n\n"]
)
print(output['choices'][0]['text'])
# トークン/秒を確認（NPU/GPU活用時は20-40 tok/s）

import onnxruntime as ort
import numpy as np
 
# 利用可能なプロバイダを確認
print("利用可能:", ort.get_available_providers())
# → ['DmlExecutionProvider', 'QNNExecutionProvider', 'CPUExecutionProvider']
 
# NPU (QNN) > GPU (DML) > CPU の優先順位で実行
providers = ['QNNExecutionProvider', 'DmlExecutionProvider', 'CPUExecutionProvider']
 
session = ort.InferenceSession(
    "model_int8.onnx",
    providers=providers,
    provider_options=[
        {'backend_path': 'QnnHtp.dll'},  # NPU (Qualcomm HTP)
        {},                               # GPU (DirectML)
        {}                                # CPU
    ]
)
 
# 推論
input_data = np.random.randn(1, 3, 224, 224).astype(np.float32)
result = session.run(None, {"input": input_data})
 
# 実行プロバイダを確認
active = session.get_providers()
print(f"使用中: {active[0]}")  # → 'QNNExecutionProvider'（NPU）

# Windows AI Foundation Model API（Windows 11 24H2以降）
import windowsai
 
async def use_phi_silica():
    """Copilot+ PC 内蔵の Phi Silica モデルを使用"""
 
    # オンデバイスLLM（Phi Silica）にアクセス
    model = await windowsai.LanguageModel.create_async()
 
    # テキスト生成
    result = await model.generate_async(
        prompt="この会議メモを要約してください：...",
        max_tokens=200
    )
    print(result.text)
 
    # ストリーミング生成
    async for token in model.generate_stream_async(
        prompt="メールの返信案を書いてください",
        max_tokens=300
    ):
        print(token, end="", flush=True)
 
# ※ Copilot+ PC (NPU 40+ TOPS) でのみ実行可能

import openvino as ov
import numpy as np
 
def run_on_intel_npu(model_path: str, input_data: np.ndarray):
    """Intel Core Ultra の NPU (AI Boost) で推論"""
 
    core = ov.Core()
 
    # 利用可能なデバイスを確認
    devices = core.available_devices
    print(f"利用可能デバイス: {devices}")
    # → ['CPU', 'GPU', 'NPU']
 
    # モデルの読み込みとコンパイル
    model = core.read_model(model_path)
 
    # NPU向けに最適化してコンパイル
    compiled_model = core.compile_model(
        model,
        device_name="NPU",
        config={
            "NPU_COMPILER_TYPE": "DRIVER",
            "PERFORMANCE_HINT": "LATENCY",
        }
    )
 
    # 推論リクエスト
    infer_request = compiled_model.create_infer_request()
    infer_request.set_input_tensor(ov.Tensor(input_data))
 
    # 推論実行と計測
    import time
    start = time.perf_counter()
    infer_request.infer()
    elapsed = (time.perf_counter() - start) * 1000
 
    output = infer_request.get_output_tensor().data
    print(f"推論時間: {elapsed:.2f}ms (NPU)")
 
    return output
 
# INT8量子化モデルで推論
input_image = np.random.randn(1, 3, 224, 224).astype(np.float32)
result = run_on_intel_npu("mobilenet_v3_int8.xml", input_image)

# AMD Ryzen AI Software を使ったNPU推論
import vitis_ai_runtime as vai
 
def run_on_ryzen_ai(model_path: str, input_data):
    """AMD Ryzen AI (XDNA 2 NPU) で推論"""
 
    # Vitis AI ランタイムの初期化
    runner = vai.Runner(model_path)
 
    # 入力テンソルの準備
    input_tensors = runner.get_input_tensors()
    output_tensors = runner.get_output_tensors()
 
    # NPUで推論実行
    job_id = runner.execute_async(input_data, output_tensors)
    runner.wait(job_id)
 
    return output_tensors
 
# ONNX → Vitis AI 変換フロー
# 1. ONNX モデルを準備
# 2. Vitis AI Quantizer でINT8量子化
# 3. Vitis AI Compiler でXDNA向けにコンパイル
# 4. ランタイムで実行
 
# 変換コマンド例:
# vai_q_onnx quantize --model model.onnx \
#     --output_model model_int8.onnx \
#     --calibration_data_reader calibration_data

import time
import subprocess
import json
from dataclasses import dataclass
 
@dataclass
class BenchmarkResult:
    model_name: str
    quantization: str
    tokens_per_second: float
    time_to_first_token: float
    memory_usage_gb: float
    backend: str
 
def benchmark_ollama_model(model_name: str, prompt: str, num_runs: int = 5):
    """Ollama モデルの性能ベンチマーク"""
    results = []
 
    for i in range(num_runs):
        start = time.perf_counter()
 
        # Ollama API を呼び出し
        response = subprocess.run(
            ["curl", "-s", "http://localhost:11434/api/generate",
             "-d", json.dumps({
                 "model": model_name,
                 "prompt": prompt,
                 "stream": False,
                 "options": {"num_predict": 128}
             })],
            capture_output=True, text=True
        )
 
        elapsed = time.perf_counter() - start
        data = json.loads(response.stdout)
 
        tokens = data.get("eval_count", 0)
        eval_duration = data.get("eval_duration", 1) / 1e9  # ns → s
        prompt_eval_duration = data.get("prompt_eval_duration", 0) / 1e9
 
        results.append({
            "tokens": tokens,
            "tokens_per_sec": tokens / eval_duration if eval_duration > 0 else 0,
            "ttft": prompt_eval_duration,
            "total_time": elapsed,
        })
 
    # 平均値を計算
    avg_tps = sum(r["tokens_per_sec"] for r in results) / len(results)
    avg_ttft = sum(r["ttft"] for r in results) / len(results)
 
    print(f"=== {model_name} ベンチマーク結果 ===")
    print(f"  平均生成速度: {avg_tps:.1f} tokens/sec")
    print(f"  平均TTFT: {avg_ttft*1000:.0f}ms")
    print(f"  実行回数: {num_runs}")
 
    return BenchmarkResult(
        model_name=model_name,
        quantization="Q4_K_M",
        tokens_per_second=avg_tps,
        time_to_first_token=avg_ttft,
        memory_usage_gb=0,
        backend="ollama"
    )
 
# 複数モデルのベンチマーク
models = ["phi3:mini", "llama3.1:8b", "gemma2:9b", "mistral:7b"]
prompt = "Explain the concept of NPU in AI PCs in 3 sentences."
 
for model in models:
    benchmark_ollama_model(model, prompt)

import ollama
import chromadb
from sentence_transformers import SentenceTransformer
 
class LocalRAGSystem:
    """
    AI PC上で完全ローカルに動作するRAGシステム
    プライバシー保護: データが端末外に出ない
    """
 
    def __init__(self):
        # 埋め込みモデル（ローカル実行）
        self.embedder = SentenceTransformer('all-MiniLM-L6-v2')
        # ベクトルDB（ローカル永続化）
        self.chroma = chromadb.PersistentClient(path="./local_vectordb")
        self.collection = self.chroma.get_or_create_collection("documents")
 
    def add_document(self, text: str, metadata: dict = None):
        """ドキュメントをベクトルDBに追加"""
        embedding = self.embedder.encode(text).tolist()
        doc_id = f"doc_{self.collection.count()}"
        self.collection.add(
            documents=[text],
            embeddings=[embedding],
            ids=[doc_id],
            metadatas=[metadata or {}]
        )
        print(f"ドキュメント追加: {doc_id}")
 
    def query(self, question: str, n_results: int = 3) -> str:
        """質問に対してRAG応答を生成"""
        # 類似ドキュメントを検索
        query_embedding = self.embedder.encode(question).tolist()
        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=n_results
        )
 
        context = "\n\n".join(results['documents'][0])
 
        # ローカルLLMで応答生成
        response = ollama.chat(
            model='llama3.1:8b',
            messages=[
                {'role': 'system',
                 'content': f'以下のコンテキストに基づいて回答してください。\n\nコンテキスト:\n{context}'},
                {'role': 'user', 'content': question}
            ]
        )
        return response['message']['content']
 
# 使用例
rag = LocalRAGSystem()
rag.add_document("AI PCにはNPUが搭載され、40 TOPS以上の性能を持つ...")
rag.add_document("Copilot+ PCはMicrosoftが定義したAI PC規格...")
answer = rag.query("Copilot+ PCの要件は？")
print(answer)

from llama_cpp import Llama
from flask import Flask, request, jsonify
 
app = Flask(__name__)
 
# コード補完専用モデルをロード
llm = Llama(
    model_path="models/codellama-7b-instruct-q4_k_m.gguf",
    n_ctx=4096,
    n_gpu_layers=-1,  # 全層GPU
    n_threads=8,
)
 
@app.route('/v1/completions', methods=['POST'])
def code_completion():
    """VS Code / JetBrains 互換のコード補完API"""
    data = request.json
    prompt = data.get('prompt', '')
    max_tokens = data.get('max_tokens', 128)
 
    output = llm(
        prompt,
        max_tokens=max_tokens,
        temperature=0.2,  # コード補完は低温推奨
        stop=["\n\n", "```", "def ", "class "],
    )
 
    return jsonify({
        "choices": [{
            "text": output['choices'][0]['text'],
            "finish_reason": output['choices'][0]['finish_reason']
        }],
        "usage": output['usage']
    })
 
# VS Code の Continue 拡張機能と連携可能
# config.json: {"model": "codellama", "apiBase": "http://localhost:5000"}
if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

import onnxruntime as ort
import cv2
import numpy as np
import time
 
class NPUVideoProcessor:
    """NPUでリアルタイム映像処理（背景ぼかし、物体検出）"""
 
    def __init__(self):
        # 背景セグメンテーションモデル（NPU実行）
        self.seg_session = ort.InferenceSession(
            "selfie_segmentation_int8.onnx",
            providers=['QNNExecutionProvider', 'DmlExecutionProvider', 'CPUExecutionProvider']
        )
 
        # 物体検出モデル（NPU実行）
        self.det_session = ort.InferenceSession(
            "yolov8n_int8.onnx",
            providers=['QNNExecutionProvider', 'DmlExecutionProvider', 'CPUExecutionProvider']
        )
 
        print(f"セグメンテーション: {self.seg_session.get_providers()[0]}")
        print(f"物体検出: {self.det_session.get_providers()[0]}")
 
    def process_frame(self, frame):
        """1フレームを処理（背景ぼかし + 物体検出）"""
        # 背景セグメンテーション
        input_seg = cv2.resize(frame, (256, 256))
        input_seg = input_seg.astype(np.float32) / 255.0
        input_seg = np.transpose(input_seg, (2, 0, 1))[np.newaxis]
 
        mask = self.seg_session.run(None, {"input": input_seg})[0]
        mask = cv2.resize(mask[0, 0], (frame.shape[1], frame.shape[0]))
 
        # 背景ぼかし適用
        blurred = cv2.GaussianBlur(frame, (21, 21), 0)
        mask_3d = np.stack([mask] * 3, axis=-1)
        result = (frame * mask_3d + blurred * (1 - mask_3d)).astype(np.uint8)
 
        return result
 
    def run_camera(self):
        """カメラ入力でリアルタイム処理"""
        cap = cv2.VideoCapture(0)
        fps_counter = []
 
        while cap.isOpened():
            ret, frame = cap.read()
            if not ret:
                break
 
            start = time.perf_counter()
            result = self.process_frame(frame)
            elapsed = time.perf_counter() - start
 
            fps_counter.append(1.0 / elapsed)
            if len(fps_counter) > 30:
                fps_counter.pop(0)
            avg_fps = sum(fps_counter) / len(fps_counter)
 
            cv2.putText(result, f"FPS: {avg_fps:.1f} (NPU)",
                       (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)
            cv2.imshow("AI PC NPU Demo", result)
 
            if cv2.waitKey(1) & 0xFF == ord('q'):
                break
 
        cap.release()
 
# NPU実行時: 30+ FPS (背景ぼかし + 物体検出)
# CPU実行時: 5-10 FPS

# 1. 基本環境
winget install Python.Python.3.12
winget install Git.Git
winget install Ollama.Ollama
 
# 2. CUDA (NVIDIA GPU の場合)
# https://developer.nvidia.com/cuda-downloads からインストール
 
# 3. Python AI パッケージ
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu121
pip install onnxruntime-directml  # DirectML (GPU + NPU)
pip install transformers accelerate
pip install llama-cpp-python  # ローカルLLM
 
# 4. Ollama モデルダウンロード
ollama pull llama3.1:8b
ollama pull codellama:7b-instruct
ollama pull nomic-embed-text  # 埋め込みモデル
 
# 5. 動作確認
python -c "import torch; print(f'CUDA: {torch.cuda.is_available()}')"
python -c "import onnxruntime as ort; print(ort.get_available_providers())"
curl http://localhost:11434/api/tags  # Ollamaモデル一覧

# 演習1: 基本実装のテンプレート
class Exercise1:
    """基本的な実装パターンの演習"""
 
    def __init__(self):
        self.data = []
 
    def validate_input(self, value):
        """入力値の検証"""
        if value is None:
            raise ValueError("入力値がNoneです")
        return True
 
    def process(self, value):
        """データ処理のメインロジック"""
        self.validate_input(value)
        self.data.append(value)
        return self.data
 
    def get_results(self):
        """処理結果の取得"""
        return {
            'count': len(self.data),
            'data': self.data
        }
 
# テスト
def test_exercise1():
    ex = Exercise1()
    assert ex.process(1) == [1]
    assert ex.process(2) == [1, 2]
    assert ex.get_results()['count'] == 2
 
    try:
        ex.process(None)
        assert False, "例外が発生するべき"
    except ValueError:
        pass
 
    print("全テスト合格!")
 
test_exercise1()

# 演習2: 応用パターン
from typing import List, Dict, Optional
from datetime import datetime
 
class AdvancedExercise:
    """応用パターンの演習"""
 
    def __init__(self, max_size: int = 100):
        self._items: List[Dict] = []
        self._max_size = max_size
        self._created_at = datetime.now()
 
    def add(self, key: str, value: any) -> bool:
        """アイテムの追加（サイズ制限付き）"""
        if len(self._items) >= self._max_size:
            return False
        self._items.append({
            'key': key,
            'value': value,
            'timestamp': datetime.now().isoformat()
        })
        return True
 
    def find(self, key: str) -> Optional[Dict]:
        """キーによる検索"""
        for item in reversed(self._items):
            if item['key'] == key:
                return item
        return None
 
    def remove(self, key: str) -> bool:
        """キーによる削除"""
        for i, item in enumerate(self._items):
            if item['key'] == key:
                self._items.pop(i)
                return True
        return False
 
    def stats(self) -> Dict:
        """統計情報"""
        return {
            'total_items': len(self._items),
            'max_size': self._max_size,
            'usage_percent': len(self._items) / self._max_size * 100,
            'uptime': str(datetime.now() - self._created_at)
        }
 
# テスト
def test_advanced():
    ex = AdvancedExercise(max_size=3)
    assert ex.add("a", 1) == True
    assert ex.add("b", 2) == True
    assert ex.add("c", 3) == True
    assert ex.add("d", 4) == False  # サイズ制限
    assert ex.find("b")['value'] == 2
    assert ex.remove("b") == True
    assert ex.find("b") is None
    stats = ex.stats()
    assert stats['total_items'] == 2
    print("応用テスト全合格!")
 
test_advanced()

# 演習3: パフォーマンス最適化
import time
from functools import lru_cache
 
# 最適化前（O(n^2)）
def slow_search(data: list, target: int) -> int:
    """非効率な検索"""
    for i in range(len(data)):
        for j in range(i + 1, len(data)):
            if data[i] + data[j] == target:
                return (i, j)
    return (-1, -1)
 
# 最適化後（O(n)）
def fast_search(data: list, target: int) -> tuple:
    """ハッシュマップを使った効率的な検索"""
    seen = {}
    for i, num in enumerate(data):
        complement = target - num
        if complement in seen:
            return (seen[complement], i)
        seen[num] = i
    return (-1, -1)
 
# ベンチマーク
def benchmark():
    import random
    data = list(range(5000))
    random.shuffle(data)
    target = data[100] + data[4000]
 
    start = time.time()
    result1 = slow_search(data, target)
    slow_time = time.time() - start
 
    start = time.time()
    result2 = fast_search(data, target)
    fast_time = time.time() - start
 
    print(f"非効率版: {slow_time:.4f}秒")
    print(f"効率版:   {fast_time:.6f}秒")
    print(f"高速化率: {slow_time/fast_time:.0f}倍")
 
benchmark()

# 設計判断の記録テンプレート
class ArchitectureDecisionRecord:
    """ADR (Architecture Decision Record) の作成"""
 
    def __init__(self, title: str):
        self.title = title
        self.context = ""
        self.decision = ""
        self.consequences = []
        self.alternatives = []
 
    def set_context(self, context: str):
        """背景と課題の記述"""
        self.context = context
        return self
 
    def set_decision(self, decision: str):
        """決定内容の記述"""
        self.decision = decision
        return self
 
    def add_consequence(self, consequence: str, positive: bool = True):
        """結果の追加"""
        self.consequences.append({
            'description': consequence,
            'type': 'positive' if positive else 'negative'
        })
        return self
 
    def add_alternative(self, name: str, reason_rejected: str):
        """却下した代替案の追加"""
        self.alternatives.append({
            'name': name,
            'reason_rejected': reason_rejected
        })
        return self
 
    def to_markdown(self) -> str:
        """Markdown形式で出力"""
        md = f"# ADR: {self.title}\n\n"
        md += f"## 背景\n{self.context}\n\n"
        md += f"## 決定\n{self.decision}\n\n"
        md += "## 結果\n"
        for c in self.consequences:
            icon = "✅" if c['type'] == 'positive' else "⚠️"
            md += f"- {icon} {c['description']}\n"
        md += "\n## 却下した代替案\n"
        for a in self.alternatives:
            md += f"- **{a['name']}**: {a['reason_rejected']}\n"
        return md

# チーム間のAPI契約定義
from dataclasses import dataclass
from typing import List, Optional
from enum import Enum
 
class Priority(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"
 
@dataclass
class APIContract:
    """チーム間のAPI契約"""
    endpoint: str
    method: str
    owner_team: str
    consumers: List[str]
    sla_ms: int  # レスポンスタイムSLA
    priority: Priority
 
    def validate_sla(self, actual_ms: int) -> bool:
        """SLA準拠の確認"""
        return actual_ms <= self.sla_ms
 
    def to_openapi(self) -> dict:
        """OpenAPI形式で出力"""
        return {
            'path': self.endpoint,
            'method': self.method,
            'x-owner': self.owner_team,
            'x-consumers': self.consumers,
            'x-sla-ms': self.sla_ms
        }
 
# 使用例
contracts = [
    APIContract(
        endpoint="/api/v1/users",
        method="GET",
        owner_team="user-team",
        consumers=["order-team", "notification-team"],
        sla_ms=200,
        priority=Priority.HIGH
    ),
    APIContract(
        endpoint="/api/v1/orders",
        method="POST",
        owner_team="order-team",
        consumers=["payment-team", "inventory-team"],
        sla_ms=500,
        priority=Priority.CRITICAL
    )
]

# 複数GPUが存在する場合のデバイス列挙と明示的選択
import onnxruntime as ort
 
# 利用可能な全ExecutionProviderを確認
providers = ort.get_available_providers()
print(f"利用可能: {providers}")
# 例: ['TensorrtExecutionProvider', 'CUDAExecutionProvider',
#       'DmlExecutionProvider', 'QNNExecutionProvider', 'CPUExecutionProvider']
 
# dGPU (CUDA) を優先し、なければ iGPU (DirectML)、最後にNPU
preferred_providers = [
    ('CUDAExecutionProvider', {'device_id': 0}),
    ('DmlExecutionProvider', {'device_id': 0}),  # iGPU
    ('QNNExecutionProvider', {}),                  # NPU
    ('CPUExecutionProvider', {}),
]
 
session = ort.InferenceSession("model.onnx", providers=preferred_providers)
active = session.get_providers()
print(f"実際に使用中: {active[0]}")

# Windows で電源プランをプログラムから確認・変更
powercfg /getactivescheme
# AI処理実行前に高性能モードへ切り替え
powercfg /setactive 8c5e7fda-e8bf-4a96-9a85-a6e23a8c635c