LangGraph 完全学習ガイド

from typing import TypedDict

# 📋 回覧ファイルの項目を定義
class SimpleState(TypedDict):
    user_input: str   # ユーザーからの質問
    response: str     # AIの回答

# 👤 社員（Node）が回覧ファイルに記入
def answer_node(state: SimpleState) -> dict:
    question = state["user_input"]  # 読む
    answer = f"{question}への回答です"
    return {"response": answer}       # 書く（responseを更新）
    # ↑ user_inputは書かない → 自動で引き継がれる

class ChatState(TypedDict):
    user_input: str
    history: list      # 会話履歴
    response: str

def chat_node(state: ChatState) -> dict:
    new_message = {"role": "user", "content": state["user_input"]}
    # ❌ これだと古い履歴が消える！
    return {
        "history": [new_message],  # 新しいメッセージだけになる
        "response": "回答..."
    }

# 実行するとこうなる...
# 1回目: history = [{"role": "user", "content": "こんにちは"}] ← OK
# 2回目: history = [{"role": "user", "content": "元気？"}]   ← 1回目が消えた！
# 3回目: history = [{"role": "user", "content": "天気は？"}] ← 2回目も消えた！

# 本当は欲しい結果：
# history = [1回目, 2回目, 3回目] ← 全部残したい！

from typing import TypedDict, Annotated
from operator import add  # リストを結合する関数

class ChatState(TypedDict):
    user_input: str
    # ↓ ここがポイント！「list型で、addで処理してね」
    history: Annotated[list, add]
    response: str

def chat_node(state: ChatState) -> dict:
    new_message = {"role": "user", "content": state["user_input"]}
    # ✅ これで既存履歴 + 新メッセージになる！
    return {
        "history": [new_message],  # 追加したいものだけ返せばOK
        "response": "回答..."
    }

# 実行結果：
# 1回目: history = [1回目のメッセージ]
# 2回目: history = [1回目, 2回目]  ← 追加された！
# 3回目: history = [1回目, 2回目, 3回目] ← 全部残ってる！

from typing import TypedDict, Annotated, Literal
from operator import add

# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# パターン①：シンプルなQ&A（履歴なし）
# → 1回の質問に1回答えるだけのシンプルなもの
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
class SimpleQAState(TypedDict):
    question: str      # ユーザーの質問
    answer: str        # AIの回答

# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# パターン②：チャットボット（履歴あり）
# → 会話の流れを覚えておきたい場合
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
class ChatbotState(TypedDict):
    user_input: str                         # 今回の入力
    messages: Annotated[list, add]          # 会話履歴（追加される）
    response: str                            # 今回の回答

# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# パターン③：分類・振り分け
# → 質問の種類によって処理を変えたい場合
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
class RouterState(TypedDict):
    user_input: str
    # Literalで選択肢を限定（タイポ防止）
    category: Literal["質問", "雑談", "クレーム"]
    response: str

# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# パターン④：RAG（検索拡張生成）
# → ドキュメント検索してから回答する場合
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
class RAGState(TypedDict):
    query: str             # ユーザーの質問
    retrieved_docs: list   # 検索で見つけた文書
    context: str           # 文書を整形したもの
    answer: str            # 最終回答
    sources: list          # 参照元（出典）

# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# パターン⑤：エージェント（ツール使用）
# → 必要に応じてツールを呼び出す場合
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
class AgentState(TypedDict):
    user_input: str
    messages: Annotated[list, add]          # 会話履歴
    tool_calls: list                         # 呼び出すツール
    tool_results: Annotated[list, add]      # ツールの結果（追加）
    final_response: str                     # 最終回答
    iteration_count: int                    # ループ回数
    max_iterations: int                     # 最大ループ回数

# 自分で「どう処理するか」を関数で定義
def keep_recent_15(existing: list, new: list) -> list:
    """既存 + 新規を結合し、直近15件だけ残す"""
    combined = (existing or []) + (new or [])
    return combined[-15:]  # 後ろから15件

# 使い方：addの代わりに自分の関数を指定
class ChatState(TypedDict):
    user_input: str
    messages: Annotated[list, keep_recent_15]  # ←ここ
    response: str

# 動作：
# [1,2,3...14,15] + [16] → [2,3,4...15,16]（1が消える）
# 常に15件以内をキープ！

from pydantic import BaseModel, Field
from typing import List

# 「こういう形式で返してね」という設計図
class MeetingTopics(BaseModel):
    """会議の議題抽出結果"""

    # topics: 議題のリスト（文字列の配列）
    topics: List[str] = Field(
        description="抽出された議題のリスト"  # LLMへのヒント
    )

    # main_topic: 最重要議題（文字列1つ）
    main_topic: str = Field(
        description="最も重要な議題"
    )

from langchain_google_genai import ChatGoogleGenerativeAI

# 1. LLMを用意
llm = ChatGoogleGenerativeAI(model="gemini-1.5-flash")

# 2. 「この形式で返してね」とLLMに伝える
structured_llm = llm.with_structured_output(MeetingTopics)
#                  ↑ さっき定義した形式

# 3. 普通に呼び出す
result = structured_llm.invoke("今日の会議では予算と人事について話し合いました。予算が最重要です。")

# 4. 結果は指定した形式で返ってくる！
print(result.topics)      # ['予算', '人事']  ← リスト形式
print(result.main_topic)  # '予算'           ← 文字列

# ドット記法でアクセスできる！
# result["topics"] ではなく result.topics
# ↑ これがPydanticの便利なところ

# Stateの定義
class MeetingState(TypedDict):
    meeting_text: str   # 入力：会議テキスト
    topics: list        # 出力：議題リスト
    main_topic: str     # 出力：最重要議題

# 議題抽出Node
def extract_topics_node(state: MeetingState) -> dict:
    """会議テキストから議題を抽出"""

    # 構造化出力を設定したLLM
    structured_llm = llm.with_structured_output(MeetingTopics)

    # LLMを呼び出し
    result = structured_llm.invoke(state["meeting_text"])

    # Stateを更新（resultから必要な値を取り出す）
    return {
        "topics": result.topics,       # リストがそのまま入る
        "main_topic": result.main_topic # 文字列がそのまま入る
    }

meeting_summarizer/
│
│  ── 実践セクションで作成済み ──
├── state.py             # State定義（そのまま使う）
├── nodes.py             # Node関数（そのまま使う）
├── routing.py           # ルーティング（そのまま使う）
│
│  ── このセクションで追加 ──
├── config.py            # 🆕 設定・環境変数・認証
├── main.py              # 🔧 API用に改造
├── requirements.txt     # 🔧 ライブラリ追加
│
│  ── 認証方式に応じて追加 ──
└── auth.py              # Firebase認証を使う場合のみ

# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# Cloud Run Functions に必要なライブラリ
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
functions-framework==3.*          # Cloud Run Functions用
langgraph>=0.2.0                  # 実践で使用
langchain-google-vertexai>=1.0.0  # Gemini API
google-cloud-aiplatform>=1.38.0   # Vertex AI

# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# セクション5: APIキー認証（お客さん向け）
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# 有効なAPIキーのリスト（カンマ区切り）
# 例: "sk-customer1-abc123,sk-customer2-def456"
# ★★★ 本番では Secret Manager の使用を推奨 ★★★
VALID_API_KEYS = os.getenv("VALID_API_KEYS", "").split(",")
VALID_API_KEYS = [k.strip() for k in VALID_API_KEYS if k.strip()]

def is_valid_api_key(api_key: str) -> bool:
    """APIキーが有効かチェック"""
    # APIキー認証が無効（キーが設定されていない）場合はスキップ
    if not VALID_API_KEYS:
        return True
    return api_key in VALID_API_KEYS

def get_customer_from_key(api_key: str) -> str:
    """APIキーからお客さん名を取得（ログ用）"""
    # 例: "sk-customer1-abc123" → "customer1"
    parts = api_key.split("-")
    return parts[1] if len(parts) >= 2 else "unknown"

from config import is_valid_api_key, get_customer_from_key

# ━━━ 認証チェック（APIキー方式）━━━
def check_api_key_auth(request) -> tuple[bool, str]:
    """APIキー認証をチェック"""
    # ヘッダーからAPIキーを取得
    api_key = request.headers.get("X-API-Key", "")

    if not is_valid_api_key(api_key):
        return False, ""

    customer = get_customer_from_key(api_key)
    return True, customer

# main関数内で使用
@functions_framework.http
def main(request):
    # ... CORS処理 ...

    # ━━━ 2. 認証チェック（APIキー）━━━
    is_valid, customer = check_api_key_auth(request)
    if not is_valid:
        return (jsonify({"error": "Invalid API Key"}), 401, headers)

import requests

# 御社から発行されたAPIキー
API_KEY = "sk-customer1-abc123"
ENDPOINT = "https://your-function-url.run.app/api/chat"

response = requests.post(
    ENDPOINT,
    headers={
        "X-API-Key": API_KEY,
        "Content-Type": "application/json"
    },
    json={"message": "こんにちは"}
)
print(response.json())

firebase-admin>=6.0.0

# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# auth.py - Firebase認証ヘルパー
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# ★★★ このファイルでJWTの署名検証を行う ★★★
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

import firebase_admin
from firebase_admin import auth, credentials
from config import is_allowed_user

# Firebase Admin SDK 初期化（1回だけ）
# Cloud Run では Application Default Credentials が自動で使われる
if not firebase_admin._apps:
    firebase_admin.initialize_app()

def verify_firebase_token(request) -> tuple[bool, str, str]:
    """
    FirebaseのJWTを検証し、ユーザー情報を取得

    Returns:
        (成功したか, メールアドレス, エラーメッセージ)
    """
    auth_header = request.headers.get("Authorization", "")

    if not auth_header.startswith("Bearer "):
        return False, "", "Authorizationヘッダーがありません"

    token = auth_header.split(" ")[1]

    try:
        # ★★★ ここで署名検証が行われる ★★★
        # 偽のJWTは auth.verify_id_token で弾かれる
        decoded_token = auth.verify_id_token(token)

        email = decoded_token.get("email", "")

        # ドメイン/メール制限チェック
        if not is_allowed_user(email):
            return False, email, "アクセス権がありません"

        return True, email, ""

    except auth.InvalidIdTokenError:
        return False, "", "無効なトークンです"
    except auth.ExpiredIdTokenError:
        return False, "", "トークンの有効期限が切れています"

from auth import verify_firebase_token

@functions_framework.http
def main(request):
    headers = get_cors_headers()

    if request.method == "OPTIONS":
        return ("", 204, headers)

    # ━━━ 2. Firebase認証チェック ━━━
    is_valid, email, error_msg = verify_firebase_token(request)
    if not is_valid:
        return (jsonify({"error": error_msg}), 401, headers)

    # 以降、emailを使ってログなどに活用可能

// Firebase Auth でログイン済みの場合
import { getAuth } from "firebase/auth";

async function callChatAPI(message) {
    const auth = getAuth();
    const user = auth.currentUser;

    // ★★★ Firebase から署名付きJWTを取得 ★★★
    const token = await user.getIdToken();

    const response = await fetch("https://your-function.run.app/api/chat", {
        method: "POST",
        headers: {
            "Authorization": `Bearer ${token}`,  // JWTを付ける
            "Content-Type": "application/json"
        },
        body: JSON.stringify({ message })
    });
}