const { conversation } = require('@assistant/conversation');
const functions = require('firebase-functions');
 
const app = conversation();
 
// メインインテントの処理
app.handle('greeting', (conv) => {
  conv.add('こんにちは！AIアシスタントガイドへようこそ。');
  conv.add('何についてお手伝いしましょうか？');
});
 
// 天気照会のインテント処理
app.handle('weather_query', async (conv) => {
  const location = conv.intent.params.location?.resolved;
  const date = conv.intent.params.date?.resolved;
 
  // 外部API呼び出し
  const weather = await fetchWeather(location, date);
 
  conv.add(`${location}の${date}の天気は${weather.condition}、`
    + `気温は${weather.temp}℃の予報です。`);
 
  if (weather.rain_probability > 50) {
    conv.add('傘をお持ちになることをおすすめします。');
  }
});
 
exports.ActionsOnGoogleFulfillment = functions.https.onRequest(app);

from ask_sdk_core.skill_builder import SkillBuilder
from ask_sdk_core.dispatch_components import AbstractRequestHandler
from ask_sdk_core.utils import is_intent_name
 
sb = SkillBuilder()
 
class RecipeIntentHandler(AbstractRequestHandler):
    """料理レシピを提案するスキル"""
 
    def can_handle(self, handler_input):
        return is_intent_name("RecipeIntent")(handler_input)
 
    def handle(self, handler_input):
        slots = handler_input.request_envelope.request.intent.slots
        ingredient = slots["ingredient"].value
 
        # LLMでレシピ生成（Bedrock経由）
        import boto3
        bedrock = boto3.client('bedrock-runtime')
        response = bedrock.invoke_model(
            modelId='anthropic.claude-3-sonnet',
            body=json.dumps({
                "prompt": f"{ingredient}を使った簡単なレシピを1つ提案してください。",
                "max_tokens": 200
            })
        )
        recipe = json.loads(response['body'].read())['completion']
 
        speech = f"{ingredient}を使ったレシピをご紹介します。{recipe}"
        return handler_input.response_builder.speak(speech).response
 
sb.add_request_handler(RecipeIntentHandler())
lambda_handler = sb.lambda_handler()

import Intents
import IntentsUI
 
// カスタムIntentの定義（Xcode Intent Definition File）
class OrderCoffeeIntentHandler: NSObject, OrderCoffeeIntentHandling {
 
    func handle(intent: OrderCoffeeIntent,
                completion: @escaping (OrderCoffeeIntentResponse) -> Void) {
 
        let coffeeType = intent.coffeeType ?? "ラテ"
        let size = intent.size ?? .medium
 
        // 注文処理
        CoffeeAPI.placeOrder(type: coffeeType, size: size) { result in
            switch result {
            case .success(let order):
                let response = OrderCoffeeIntentResponse(code: .success,
                    userActivity: nil)
                response.orderNumber = order.id
                response.estimatedTime = "\(order.waitMinutes)分"
                completion(response)
 
            case .failure:
                completion(OrderCoffeeIntentResponse(code: .failure,
                    userActivity: nil))
            }
        }
    }
 
    // Siriへの提案
    func resolveCoffeeType(for intent: OrderCoffeeIntent,
        with completion: @escaping (INStringResolutionResult) -> Void) {
        if let type = intent.coffeeType {
            completion(.success(with: type))
        } else {
            completion(.needsValue())
        }
    }
}

import asyncio
import websockets
import json
import base64
import pyaudio
 
async def realtime_voice_assistant():
    """OpenAI Realtime API でリアルタイム音声アシスタント"""
 
    url = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview"
    headers = {
        "Authorization": f"Bearer {OPENAI_API_KEY}",
        "OpenAI-Beta": "realtime=v1"
    }
 
    async with websockets.connect(url, extra_headers=headers) as ws:
        # セッション設定
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "modalities": ["text", "audio"],
                "instructions": "あなたは親切な日本語アシスタントです。",
                "voice": "alloy",
                "input_audio_format": "pcm16",
                "output_audio_format": "pcm16",
                "turn_detection": {
                    "type": "server_vad",  # サーバー側VAD
                    "threshold": 0.5
                }
            }
        }))
 
        # マイク入力を送信
        audio = pyaudio.PyAudio()
        stream = audio.open(format=pyaudio.paInt16,
                          channels=1, rate=24000,
                          input=True, frames_per_buffer=1024)
 
        async def send_audio():
            while True:
                data = stream.read(1024, exception_on_overflow=False)
                encoded = base64.b64encode(data).decode()
                await ws.send(json.dumps({
                    "type": "input_audio_buffer.append",
                    "audio": encoded
                }))
                await asyncio.sleep(0.04)
 
        async def receive_response():
            while True:
                msg = json.loads(await ws.recv())
                if msg["type"] == "response.audio.delta":
                    audio_data = base64.b64decode(msg["delta"])
                    # スピーカーに出力
                    play_audio(audio_data)
 
        await asyncio.gather(send_audio(), receive_response())
 
asyncio.run(realtime_voice_assistant())

import whisper
import ollama
import pyttsx3
import sounddevice as sd
import numpy as np
 
class LocalVoiceAssistant:
    """完全ローカルで動作する音声アシスタント"""
 
    def __init__(self):
        # Whisper（音声認識）
        self.asr_model = whisper.load_model("base")
        # TTS エンジン
        self.tts = pyttsx3.init()
        self.tts.setProperty('rate', 180)
 
    def listen(self, duration=5, sample_rate=16000):
        """マイクから音声を録音"""
        print("聞いています...")
        audio = sd.rec(int(duration * sample_rate),
                      samplerate=sample_rate, channels=1,
                      dtype='float32')
        sd.wait()
        return audio.flatten()
 
    def transcribe(self, audio):
        """音声をテキストに変換（Whisper）"""
        result = self.asr_model.transcribe(
            audio, language="ja", fp16=False
        )
        return result["text"]
 
    def think(self, user_text, context=None):
        """ローカルLLMで応答生成（Ollama）"""
        messages = [
            {"role": "system",
             "content": "簡潔に日本語で回答してください。"},
        ]
        if context:
            messages.append({"role": "assistant", "content": context})
        messages.append({"role": "user", "content": user_text})
 
        response = ollama.chat(model="gemma2:9b", messages=messages)
        return response['message']['content']
 
    def speak(self, text):
        """テキストを音声で読み上げ"""
        print(f"アシスタント: {text}")
        self.tts.say(text)
        self.tts.runAndWait()
 
    def run(self):
        """メインループ"""
        print("ローカル音声アシスタント起動（Ctrl+Cで終了）")
        context = None
        while True:
            audio = self.listen()
            text = self.transcribe(audio)
            print(f"ユーザー: {text}")
 
            if "終了" in text or "さようなら" in text:
                self.speak("さようなら。")
                break
 
            response = self.think(text, context)
            context = response
            self.speak(response)
 
assistant = LocalVoiceAssistant()
assistant.run()

import AppIntents
 
/// App Intents フレームワークによるSiri統合（iOS 16+）
/// 従来の SiriKit Intent Definition よりも簡潔で型安全
struct SearchRecipeIntent: AppIntent {
    static var title: LocalizedStringResource = "レシピ検索"
    static var description = IntentDescription("食材からレシピを検索します")
 
    // Siriに自然言語で問いかけるとこのパラメータが自動抽出される
    @Parameter(title: "食材")
    var ingredient: String
 
    @Parameter(title: "調理時間（分）", default: 30)
    var maxCookingTime: Int
 
    // Siri Shortcuts アプリにも自動表示される
    static var parameterSummary: some ParameterSummary {
        Summary("「\(\.$ingredient)」を使った\(\.$maxCookingTime)分以内のレシピ")
    }
 
    func perform() async throws -> some IntentResult & ProvidesDialog & ShowsSnippetView {
        // レシピ検索ロジック
        let recipes = try await RecipeService.search(
            ingredient: ingredient,
            maxTime: maxCookingTime
        )
 
        guard let topRecipe = recipes.first else {
            return .result(
                dialog: "\(ingredient)を使ったレシピが見つかりませんでした。"
            )
        }
 
        // Siri応答 + リッチUIスニペット
        return .result(
            dialog: "\(topRecipe.name)はいかがですか？調理時間は\(topRecipe.cookingTime)分です。",
            view: RecipeSnippetView(recipe: topRecipe)
        )
    }
}
 
/// Shortcuts アプリで表示されるショートカットプロバイダ
struct RecipeShortcuts: AppShortcutsProvider {
    static var appShortcuts: [AppShortcut] {
        AppShortcut(
            intent: SearchRecipeIntent(),
            phrases: [
                "「\(.applicationName)」で\(\.$ingredient)のレシピを探して",
                "「\(.applicationName)」で簡単な料理を提案して",
            ],
            shortTitle: "レシピ検索",
            systemImageName: "fork.knife"
        )
    }
}

import openai
import json
import requests
 
class FunctionCallingAssistant:
    """
    Function Calling を使った高度なAIアシスタント
    LLMが適切なAPIを自動選択して実行する
 
    なぜ Function Calling か:
    - 従来のインテント分類では対応できない複雑なクエリに対応
    - LLMが文脈に応じて適切な関数を自動選択
    - 複数の関数を順次呼び出すチェーン実行が可能
    """
 
    def __init__(self, api_key):
        self.client = openai.OpenAI(api_key=api_key)
        self.tools = [
            {
                "type": "function",
                "function": {
                    "name": "get_weather",
                    "description": "指定した場所の天気予報を取得する",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "location": {"type": "string", "description": "都市名"},
                            "date": {"type": "string", "description": "日付（YYYY-MM-DD）"}
                        },
                        "required": ["location"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "search_restaurant",
                    "description": "レストランを検索する",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "location": {"type": "string"},
                            "cuisine": {"type": "string", "description": "料理のジャンル"},
                            "budget": {"type": "integer", "description": "予算（円）"}
                        },
                        "required": ["location"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "set_reminder",
                    "description": "リマインダーを設定する",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "title": {"type": "string"},
                            "datetime": {"type": "string"},
                            "priority": {"type": "string", "enum": ["low", "medium", "high"]}
                        },
                        "required": ["title", "datetime"]
                    }
                }
            }
        ]
 
    def chat(self, user_message, conversation_history=None):
        """ユーザーメッセージを処理し、必要に応じてAPIを呼び出す"""
        if conversation_history is None:
            conversation_history = []
 
        messages = [
            {"role": "system", "content": "あなたは親切な日本語アシスタントです。"
             "ユーザーの要望に応じて適切なツールを使用してください。"},
            *conversation_history,
            {"role": "user", "content": user_message}
        ]
 
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=self.tools,
            tool_choice="auto"
        )
 
        message = response.choices[0].message
 
        # Function Call が要求された場合
        if message.tool_calls:
            results = []
            for tool_call in message.tool_calls:
                func_name = tool_call.function.name
                args = json.loads(tool_call.function.arguments)
 
                # 関数を実行
                result = self._execute_function(func_name, args)
                results.append({
                    "tool_call_id": tool_call.id,
                    "role": "tool",
                    "content": json.dumps(result, ensure_ascii=False)
                })
 
            # 関数の結果をLLMに渡して最終応答を生成
            messages.append(message)
            messages.extend(results)
 
            final_response = self.client.chat.completions.create(
                model="gpt-4o",
                messages=messages
            )
            return final_response.choices[0].message.content
 
        return message.content
 
    def _execute_function(self, name, args):
        """関数を実行して結果を返す"""
        if name == "get_weather":
            return {"condition": "晴れ", "temp": 22, "rain_prob": 10}
        elif name == "search_restaurant":
            return {"name": "鮨かねさか", "rating": 4.8, "price": "¥15,000"}
        elif name == "set_reminder":
            return {"status": "success", "id": "rem_123"}
        return {"error": "Unknown function"}
 
# 使用例
assistant = FunctionCallingAssistant(api_key="sk-...")
# 複合クエリ: 天気確認 + レストラン検索を自動で実行
response = assistant.chat(
    "明日の東京の天気を調べて、天気が良ければ表参道でランチのお店を探して"
)
print(response)

"""
Home Assistant Wyoming Protocol を使ったローカル音声パイプライン
 
Wyoming Protocol は音声処理コンポーネント間の通信プロトコル:
  マイク → Wake Word (openWakeWord) → ASR (Whisper) →
  Intent → TTS (Piper) → スピーカー
 
全てローカルで動作し、クラウド不要
"""
import asyncio
import json
from wyoming.server import AsyncServer
from wyoming.asr import Transcribe, Transcript
from wyoming.wake import Detection
import whisper
import numpy as np
 
class LocalASRServer:
    """Whisper ベースのローカルASRサーバー"""
 
    def __init__(self, model_name="base"):
        self.model = whisper.load_model(model_name)
        print(f"Whisper {model_name} モデルロード完了")
 
    async def handle_client(self, reader, writer):
        """Wyoming プロトコルでASRリクエストを処理"""
        # 音声データを受信
        audio_data = await self._receive_audio(reader)
 
        # Whisper で認識
        audio_np = np.frombuffer(audio_data, dtype=np.int16).astype(np.float32) / 32768.0
        result = self.model.transcribe(audio_np, language="ja", fp16=False)
 
        transcript = result["text"].strip()
        print(f"認識結果: {transcript}")
 
        # Wyoming プロトコルで結果を返す
        response = Transcript(text=transcript)
        await self._send_response(writer, response)
 
    async def _receive_audio(self, reader):
        """音声データの受信"""
        chunks = []
        while True:
            data = await reader.read(4096)
            if not data:
                break
            chunks.append(data)
        return b"".join(chunks)
 
    async def _send_response(self, writer, response):
        """応答の送信"""
        writer.write(json.dumps({"text": response.text}).encode())
        await writer.drain()
        writer.close()
 
async def main():
    server = LocalASRServer(model_name="small")
    srv = await asyncio.start_server(
        server.handle_client, "0.0.0.0", 10300  # Wyoming ASR ポート
    )
    print("Wyoming ASR サーバー起動: port 10300")
    async with srv:
        await srv.serve_forever()
 
# 起動: python wyoming_asr.py
# Home Assistant で Wyoming 統合を追加して接続

class MultimodalAssistant:
    """
    テキスト + 音声 + 画像を統合するマルチモーダルアシスタント
    例: 「この写真の料理のカロリーを教えて」と音声で質問 + 写真撮影
    """
    def __init__(self):
        self.asr = whisper.load_model("base")
        self.vlm_client = openai.OpenAI()  # GPT-4V
 
    async def process_multimodal_query(self, audio_data, image_data=None):
        """音声 + 画像のマルチモーダルクエリを処理"""
 
        # 1. 音声認識
        text = self.asr.transcribe(audio_data, language="ja")["text"]
        print(f"認識テキスト: {text}")
 
        # 2. 画像がある場合はマルチモーダルLLMで処理
        if image_data:
            import base64
            image_b64 = base64.b64encode(image_data).decode()
 
            response = self.vlm_client.chat.completions.create(
                model="gpt-4o",
                messages=[
                    {
                        "role": "user",
                        "content": [
                            {"type": "text", "text": text},
                            {
                                "type": "image_url",
                                "image_url": {
                                    "url": f"data:image/jpeg;base64,{image_b64}"
                                }
                            }
                        ]
                    }
                ]
            )
            return response.choices[0].message.content
 
        # 3. テキストのみの場合
        response = self.vlm_client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": text}]
        )
        return response.choices[0].message.content

class ProactiveAssistant:
    """
    ユーザーが明示的に指示しなくても、コンテキストに基づいて
    先回りして情報を提供するプロアクティブアシスタント
 
    なぜプロアクティブか:
    - 従来のアシスタントは「質問→回答」の受動型
    - プロアクティブ型は状況を監視し、適切なタイミングで介入
    - Apple Intelligence の「Personal Context」がこの方向性
    """
    def __init__(self):
        self.context_store = {}
        self.rules = []
 
    def update_context(self, context_type, data):
        """コンテキスト情報を更新"""
        self.context_store[context_type] = {
            "data": data,
            "timestamp": time.time()
        }
        self._evaluate_rules()
 
    def _evaluate_rules(self):
        """ルールを評価してプロアクティブな提案を生成"""
        suggestions = []
 
        # ルール1: 出発時刻が近い場合、交通情報を提供
        if "calendar" in self.context_store and "location" in self.context_store:
            next_event = self.context_store["calendar"]["data"]
            if next_event.get("departure_in_minutes", float('inf')) < 30:
                weather = self._get_weather()
                traffic = self._get_traffic(next_event["location"])
                suggestions.append(
                    f"{next_event['title']}の{next_event['departure_in_minutes']}分前です。"
                    f"現在の交通状況: {traffic['duration']}分。"
                    f"{'傘を忘れずに。' if weather['rain_prob'] > 50 else ''}"
                )
 
        # ルール2: 異常な健康データを検出
        if "health" in self.context_store:
            hr = self.context_store["health"]["data"].get("heart_rate", 0)
            if hr > 100 and self.context_store.get("activity", {}).get("data", {}).get("type") == "resting":
                suggestions.append(
                    f"安静時心拍数が{hr}BPMと高めです。"
                    "ストレスや脱水の可能性があります。水分を取りましょう。"
                )
 
        # ルール3: 定期的なタスクのリマインド
        if "habits" in self.context_store:
            habits = self.context_store["habits"]["data"]
            for habit in habits:
                if habit["due"] and not habit["completed"]:
                    suggestions.append(f"まだ「{habit['name']}」が完了していません。")
 
        for suggestion in suggestions:
            self._notify_user(suggestion)
 
    def _notify_user(self, message):
        """ユーザーに通知"""
        print(f"[プロアクティブ提案] {message}")

class HybridRouter:
    """
    単純なコマンドはルールベースで即時処理、
    複雑な質問はLLMにルーティングするハイブリッド設計
 
    なぜハイブリッドか:
    - 「タイマー3分」にLLMは不要（遅延・コストの無駄）
    - 「明日の天気を考慮してコーデを提案」はLLMが適切
    """
    def __init__(self):
        self.simple_commands = {
            "タイマー": self._handle_timer,
            "アラーム": self._handle_alarm,
            "音量": self._handle_volume,
            "電話": self._handle_call,
        }
        self.llm = OllamaClient()
 
    def route(self, text):
        # 1. 単純コマンドのマッチング（正規表現ベース）
        for keyword, handler in self.simple_commands.items():
            if keyword in text:
                return handler(text), "rule"
 
        # 2. LLMにルーティング
        return self.llm.chat(text), "llm"
 
    def _handle_timer(self, text):
        import re
        match = re.search(r'(\d+)\s*分', text)
        if match:
            minutes = int(match.group(1))
            return f"タイマーを{minutes}分にセットしました。"
        return "何分のタイマーですか？"

class ContextualDialogManager:
    """
    対話コンテキストを維持し、自然な連続会話を実現する
 
    問題: 「東京の天気は？」→「じゃあ明日は？」
    → 文脈がないと「何のことですか？」になる
 
    解決: 直近の対話履歴を保持し、代名詞・省略を解決
    """
    def __init__(self, max_history=10):
        self.history = []
        self.entities = {}  # 抽出されたエンティティのキャッシュ
        self.max_history = max_history
 
    def process(self, user_input):
        # エンティティ追跡
        new_entities = self._extract_entities(user_input)
        self.entities.update(new_entities)
 
        # 省略された情報を補完
        enriched_input = self._resolve_references(user_input)
 
        # 対話履歴に追加
        self.history.append({"role": "user", "content": enriched_input})
 
        # LLMで応答生成（履歴を含む）
        response = self._generate_response()
 
        self.history.append({"role": "assistant", "content": response})
 
        # 履歴の上限管理
        if len(self.history) > self.max_history * 2:
            self.history = self.history[-self.max_history * 2:]
 
        return response
 
    def _extract_entities(self, text):
        """テキストからエンティティを抽出"""
        entities = {}
        # 場所の抽出
        locations = ["東京", "大阪", "名古屋", "福岡", "札幌"]
        for loc in locations:
            if loc in text:
                entities["location"] = loc
        # 日付の抽出
        if "明日" in text:
            entities["date"] = "tomorrow"
        elif "今日" in text:
            entities["date"] = "today"
        return entities
 
    def _resolve_references(self, text):
        """代名詞や省略を解決"""
        # 「じゃあ明日は？」→ 「じゃあ東京の明日の天気は？」
        if len(text) < 10 and "location" in self.entities:
            if "明日" in text or "今日" in text:
                text = f"{self.entities['location']}の{text}"
        return text

# 演習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()