ブログ

  • PythonのFastAPIで爆速REST API開発:非同期処理とバリデーション

    FastAPIはPython製のWebフレームワークで、型ヒントを活用した自動バリデーション、自動ドキュメント生成、非同期処理対応が特徴です。Flask比で200%以上のパフォーマンス向上が見込めます。

    FastAPIの特徴

    • Python型ヒントによる自動バリデーション
    • OpenAPI(Swagger)ドキュメント自動生成
    • async/await対応の非同期処理
    • Pydanticモデルによるデータシリアライゼーション
    • StarletteベースのASGIフレームワーク

    インストールと最初のAPI

    pip install "fastapi[standard]"
    # main.py
    from fastapi import FastAPI, HTTPException
    from pydantic import BaseModel, Field
    from typing import Optional
    import uvicorn
    
    app = FastAPI(title="商品管理API", version="1.0.0")
    
    class Product(BaseModel):
        name: str = Field(..., min_length=1, max_length=100)
        price: int = Field(..., gt=0, description="価格(円)")
        description: Optional[str] = None
        in_stock: bool = True
    
    # インメモリDB
    products: dict[int, Product] = {}
    next_id = 1
    
    @app.post("/products", status_code=201)
    async def create_product(product: Product):
        global next_id
        product_id = next_id
        products[product_id] = product
        next_id += 1
        return {"id": product_id, **product.model_dump()}
    
    @app.get("/products/{product_id}")
    async def get_product(product_id: int):
        if product_id not in products:
            raise HTTPException(status_code=404, detail="商品が見つかりません")
        return {"id": product_id, **products[product_id].model_dump()}
    
    @app.get("/products")
    async def list_products(
        min_price: Optional[int] = None,
        max_price: Optional[int] = None,
        in_stock: Optional[bool] = None,
    ):
        result = []
        for pid, p in products.items():
            if min_price and p.price  max_price:
                continue
            if in_stock is not None and p.in_stock != in_stock:
                continue
            result.append({"id": pid, **p.model_dump()})
        return result
    
    if __name__ == "__main__":
        uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

    自動ドキュメント

    FastAPIは起動するだけで以下のドキュメントが自動生成されます。

    • Swagger UI: http://localhost:8000/docs – インタラクティブなAPI操作画面
    • ReDoc: http://localhost:8000/redoc – 読みやすいドキュメント形式

    非同期処理

    FastAPIはasync/awaitをネイティブサポートしています。DB操作やHTTPリクエストなどのI/O処理を非同期で実行できます。

    import httpx
    
    @app.get("/external-data")
    async def get_external_data():
        async with httpx.AsyncClient() as client:
            response = await client.get("https://api.example.com/data")
            return response.json()

    まとめ

    FastAPIは、型安全性・パフォーマンス・開発体験の全てにおいて優れたフレームワークです。AI APIのバックエンド、マイクロサービス、プロトタイプ開発など幅広い場面で活躍します。

  • Fly.ioでNode.jsアプリを無料デプロイ:Heroku代替の本命

    Herokuの無料プランが廃止されて以降、個人開発者のデプロイ先として注目を集めているFly.io。東京リージョンがあり、日本からのレイテンシも低い。Node.jsアプリのデプロイ手順を実際のプロジェクトを例に解説します。

    Fly.ioの特徴

    • 東京リージョン(nrt)あり
    • Dockerベースでデプロイ
    • 無料枠: 3つのshared-cpu-1x VM、160GB転送量/月
    • PostgreSQL、Redis、S3互換ストレージも提供
    • 自動HTTPS対応

    セットアップ

    # Fly CLIインストール
    curl -L https://fly.io/install.sh | sh
    
    # ログイン
    fly auth login
    
    # プロジェクト作成
    cd my-node-app
    fly launch

    Dockerfile の準備

    FROM node:20-alpine
    WORKDIR /app
    COPY package*.json ./
    RUN npm ci --omit=dev
    COPY src/ ./src/
    EXPOSE 3000
    CMD ["node", "src/index.js"]

    fly.tomlの設定

    app = "my-awesome-app"
    primary_region = "nrt"  # 東京リージョン
    
    [build]
    
    [http_service]
      internal_port = 3000
      force_https = true
      auto_stop_machines = "stop"
      auto_start_machines = true
      min_machines_running = 0
    
    [env]
      NODE_ENV = "production"

    シークレット管理

    # 環境変数(シークレット)の設定
    fly secrets set DATABASE_URL="postgresql://..."
    fly secrets set API_KEY="sk-..."
    
    # 設定済みシークレット一覧
    fly secrets list

    デプロイ

    # デプロイ実行
    fly deploy
    
    # ログ確認
    fly logs
    
    # ステータス確認
    fly status
    
    # ブラウザで開く
    fly open

    カスタムドメイン設定

    # カスタムドメイン追加
    fly certs create example.com
    
    # DNS設定を確認
    fly certs show example.com

    まとめ

    Fly.ioは「flyctl deploy」一発でDockerアプリをデプロイできる手軽さと、東京リージョンの低レイテンシが魅力です。個人プロジェクトからスタートアップまで、幅広い用途で活用できます。

  • RAG(検索拡張生成)をゼロから実装する:LangChainとFAISSで社内ドキュメント検索

    RAG(Retrieval-Augmented Generation)は、LLMに外部知識を与えて回答精度を向上させる手法です。社内ドキュメントやFAQをベクトルデータベースに格納し、質問に関連する情報を検索してからLLMに回答させます。

    RAGの仕組み

    RAGは大きく2つのフェーズで動作します。

    1. インデックス作成: ドキュメントをチャンクに分割→ベクトル化→ベクトルDBに格納
    2. 検索+生成: ユーザーの質問をベクトル化→類似チャンクを検索→検索結果+質問をLLMに渡して回答生成

    環境準備

    pip install langchain langchain-openai faiss-cpu tiktoken

    ドキュメントの読み込みとチャンク分割

    from langchain.text_splitter import RecursiveCharacterTextSplitter
    from langchain_community.document_loaders import TextLoader
    
    # ドキュメント読み込み
    loader = TextLoader("company_faq.txt", encoding="utf-8")
    documents = loader.load()
    
    # チャンク分割(500文字ごと、100文字オーバーラップ)
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,
        chunk_overlap=100,
        separators=["nn", "n", "。", "、", " "]
    )
    chunks = splitter.split_documents(documents)
    print(f"チャンク数: {len(chunks)}")

    ベクトルDBの構築

    from langchain_openai import OpenAIEmbeddings
    from langchain_community.vectorstores import FAISS
    
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    
    # FAISSインデックスを作成
    vectorstore = FAISS.from_documents(chunks, embeddings)
    
    # ローカルに保存(永続化)
    vectorstore.save_local("faiss_index")
    
    # 読み込み
    vectorstore = FAISS.load_local("faiss_index", embeddings, allow_dangerous_deserialization=True)

    RAGチェーンの構築

    from langchain_openai import ChatOpenAI
    from langchain.chains import RetrievalQA
    
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
    
    qa_chain = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="stuff",
        retriever=retriever,
        return_source_documents=True,
    )
    
    result = qa_chain.invoke({"query": "有給休暇の申請方法を教えてください"})
    print(result["result"])
    print("---参照ドキュメント---")
    for doc in result["source_documents"]:
        print(f"  - {doc.page_content[:100]}...")

    精度向上のコツ

    • チャンクサイズ: 小さすぎると文脈が失われ、大きすぎるとノイズが増える。300-500文字が目安
    • オーバーラップ: チャンク境界での情報欠落を防ぐ。チャンクサイズの20%程度
    • 検索件数(k): 多すぎるとコンテキストウィンドウを圧迫。3-5件が推奨
    • リランキング: 検索結果をLLMで再評価してから使うと精度が上がる

    まとめ

    RAGを使えば、LLMが学習していない最新情報や社内固有の知識に基づいた回答が可能になります。FAISSは無料で使えるベクトルDBとして優秀で、小〜中規模のドキュメントであれば十分な性能を発揮します。

  • Next.js 14 App RouterでSSR・SSG・ISRを使い分ける実践ガイド

    Next.js 14のApp Routerでは、レンダリング方式の選択が柔軟になりました。SSR(サーバーサイドレンダリング)、SSG(静的サイト生成)、ISR(増分静的再生成)をページ単位・コンポーネント単位で使い分ける方法を解説します。

    App Routerの基本

    Next.js 13以降のApp Routerでは、デフォルトですべてのコンポーネントがServer Componentになります。これにより、クライアントに送信されるJavaScriptの量が大幅に削減されます。

    // app/page.tsx - デフォルトでServer Component
    export default async function HomePage() {
      // サーバー側で実行される
      const data = await fetch("https://api.example.com/posts");
      const posts = await data.json();
    
      return (
        <main>
          <h1>最新記事</h1>
          {posts.map(post => (
            <article key={post.id}>
              <h2>{post.title}</h2>
              <p>{post.excerpt}</p>
            </article>
          ))}
        </main>
      );
    }

    SSG(静的生成)

    ビルド時にHTMLを生成する方式です。ブログ記事やドキュメントなど、更新頻度が低いコンテンツに最適です。

    // app/blog/[slug]/page.tsx
    export async function generateStaticParams() {
      const posts = await fetch("https://api.example.com/posts").then(r => r.json());
      return posts.map(post => ({ slug: post.slug }));
    }
    
    export default async function BlogPost({ params }) {
      const post = await fetch(
        `https://api.example.com/posts/${params.slug}`,
        { cache: "force-cache" } // SSG: ビルド時にキャッシュ
      ).then(r => r.json());
    
      return <article><h1>{post.title}</h1><div>{post.content}</div></article>;
    }

    SSR(サーバーサイドレンダリング)

    リクエストごとにサーバーでHTMLを生成します。ユーザーごとに異なるコンテンツを表示する場合に使います。

    // app/dashboard/page.tsx
    export default async function Dashboard() {
      const data = await fetch("https://api.example.com/user/dashboard", {
        cache: "no-store", // SSR: キャッシュしない
        headers: { Authorization: `Bearer ${getToken()}` }
      }).then(r => r.json());
    
      return <div>ようこそ、{data.user.name}さん</div>;
    }

    ISR(増分静的再生成)

    SSGとSSRの良いとこ取り。静的に生成されたページを一定時間後にバックグラウンドで再生成します。

    // app/products/page.tsx
    export default async function Products() {
      const products = await fetch("https://api.example.com/products", {
        next: { revalidate: 3600 } // ISR: 1時間ごとに再生成
      }).then(r => r.json());
    
      return (
        <div>
          {products.map(p => <div key={p.id}>{p.name} - ¥{p.price}</div>)}
        </div>
      );
    }

    使い分けの判断基準

    どのレンダリング方式を使うかは、コンテンツの特性で決めます。更新頻度が低い(ブログ・ドキュメント)→ SSG、リアルタイム性が必要(ダッシュボード)→ SSR、適度に更新される(商品一覧・ニュース)→ ISR、が基本的な選び方です。

    まとめ

    App Routerでは、fetchのオプションを変えるだけでレンダリング方式を切り替えられます。ページの特性に合わせて最適な方式を選ぶことで、パフォーマンスとユーザー体験を両立できます。

  • Docker ComposeでNode.js + PostgreSQL環境を5分で構築する

    開発環境の構築は地味に時間がかかる作業です。Docker Composeを使えば、Node.jsアプリケーションとPostgreSQLデータベースを含む開発環境を、たった1コマンドで立ち上げることができます。

    前提条件

    • Docker Desktop がインストール済み
    • docker compose コマンドが使える状態

    プロジェクト構成

    my-app/
    ├── docker-compose.yml
    ├── Dockerfile
    ├── package.json
    ├── src/
    │   └── index.js
    └── .env

    docker-compose.yml

    version: "3.8"
    services:
      app:
        build: .
        ports:
          - "3000:3000"
        environment:
          - DATABASE_URL=postgresql://myuser:mypass@db:5432/mydb
          - NODE_ENV=development
        volumes:
          - ./src:/app/src
        depends_on:
          db:
            condition: service_healthy
    
      db:
        image: postgres:16-alpine
        environment:
          POSTGRES_USER: myuser
          POSTGRES_PASSWORD: mypass
          POSTGRES_DB: mydb
        ports:
          - "5432:5432"
        volumes:
          - pgdata:/var/lib/postgresql/data
        healthcheck:
          test: ["CMD-SHELL", "pg_isready -U myuser -d mydb"]
          interval: 5s
          timeout: 5s
          retries: 5
    
    volumes:
      pgdata:

    Dockerfile

    FROM node:20-alpine
    WORKDIR /app
    COPY package*.json ./
    RUN npm ci
    COPY src/ ./src/
    EXPOSE 3000
    CMD ["node", "src/index.js"]

    Node.jsアプリケーション

    // src/index.js
    const { Pool } = require("pg");
    const http = require("http");
    
    const pool = new Pool({
      connectionString: process.env.DATABASE_URL,
    });
    
    const server = http.createServer(async (req, res) => {
      try {
        const result = await pool.query("SELECT NOW() as current_time");
        res.writeHead(200, { "Content-Type": "application/json" });
        res.end(JSON.stringify({
          message: "Hello from Docker!",
          db_time: result.rows[0].current_time,
        }));
      } catch (err) {
        res.writeHead(500);
        res.end(JSON.stringify({ error: err.message }));
      }
    });
    
    server.listen(3000, () => {
      console.log("Server running on port 3000");
    });

    起動と確認

    # ビルド&起動
    docker compose up -d
    
    # ログ確認
    docker compose logs -f app
    
    # 動作確認
    curl http://localhost:3000

    開発時のTips

    • volumesでsrcディレクトリをマウントしておけば、ファイル変更が即座に反映されます(nodemonと組み合わせると便利)
    • healthcheckを設定しておくと、DBが起動する前にアプリが接続しようとするエラーを防げます
    • PostgreSQLのデータはnamed volumeに保存されるので、コンテナを再作成してもデータは消えません

    まとめ

    Docker Composeを使えば、チームメンバー全員が同じ環境で開発できます。新メンバーのオンボーディングも「git cloneしてdocker compose up」だけで完了。環境差異によるトラブルから解放されます。

  • OpenAI Function Callingの実装ガイド:AIに外部ツールを使わせる

    OpenAIのFunction Calling(関数呼び出し)機能を使うと、AIが外部APIやデータベースと連携して、リアルタイムの情報を取得できるようになります。天気予報の取得、商品検索、データベースクエリなど、実践的なユースケースを解説します。

    Function Callingとは

    Function Callingは、GPT-4やGPT-4oに「使える関数」を定義しておくと、ユーザーの質問に応じて適切な関数を呼び出してくれる機能です。AIが直接関数を実行するのではなく、「この関数をこの引数で呼んでください」という指示を返します。

    基本的な実装

    商品検索を例に実装してみましょう。

    from openai import OpenAI
    import json
    
    client = OpenAI()
    
    # 利用可能な関数を定義
    tools = [
        {
            "type": "function",
            "function": {
                "name": "search_products",
                "description": "商品データベースからキーワードで商品を検索する",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "keyword": {
                            "type": "string",
                            "description": "検索キーワード(例: 赤いTシャツ)"
                        },
                        "max_price": {
                            "type": "integer",
                            "description": "最大価格(円)"
                        }
                    },
                    "required": ["keyword"]
                }
            }
        }
    ]
    
    # 実際の検索関数
    def search_products(keyword, max_price=None):
        # 本番ではDBクエリを実行
        products = [
            {"name": "赤いTシャツ", "price": 2980},
            {"name": "青いTシャツ", "price": 3480},
        ]
        if max_price:
            products = [p for p in products if p["price"] <= max_price]
        return [p for p in products if keyword in p["name"]]

    AIとの対話フロー

    Function Callingの対話は3ステップで進みます。

    # Step 1: ユーザーの質問をAIに送信
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "3000円以下の赤いTシャツはありますか?"}],
        tools=tools,
    )
    
    # Step 2: AIが関数呼び出しを要求
    tool_call = response.choices[0].message.tool_calls[0]
    args = json.loads(tool_call.function.arguments)
    # args = {"keyword": "赤いTシャツ", "max_price": 3000}
    
    # Step 3: 関数を実行して結果をAIに返す
    result = search_products(**args)
    final_response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "user", "content": "3000円以下の赤いTシャツはありますか?"},
            response.choices[0].message,
            {"role": "tool", "tool_call_id": tool_call.id, "content": json.dumps(result, ensure_ascii=False)}
        ],
        tools=tools,
    )
    print(final_response.choices[0].message.content)

    複数関数の定義

    実際のアプリケーションでは複数の関数を定義し、AIに状況に応じて使い分けてもらいます。商品検索に加えて、注文状況の確認や在庫確認なども追加できます。AIは質問の内容から最適な関数を自動で選択します。

    注意点とベストプラクティス

    • 関数のdescriptionは詳しく書く。AIはこれを見て関数を選択します
    • パラメータのdescriptionも具体例を含めると精度が上がります
    • AIが不要な関数呼び出しをしないよう、tool_choice パラメータで制御可能
    • 関数の実行結果は構造化されたJSONで返すと、AIの応答品質が向上します

    まとめ

    Function Callingを使うことで、AIは単なるテキスト生成を超えて、実際のデータやサービスと連携する強力なアシスタントになります。ECサイトの商品検索、カスタマーサポート、社内ツールの自動化など、応用範囲は広大です。

  • Claude APIとPythonで作るAIチャットボット入門

    Anthropic社のClaude APIを使って、Pythonで動作するAIチャットボットを作成する方法を解説します。2024年以降、ClaudeはGPT-4oと並ぶ高性能LLMとして注目を集めています。

    Claude APIとは

    Claude APIは、Anthropic社が提供する大規模言語モデル(LLM)のAPIです。ChatGPTのOpenAI APIと同様に、HTTPリクエストでAIとの対話が可能です。特にClaude 3.5 Sonnetは、コーディング支援や長文処理において高い性能を発揮します。

    環境構築

    まずAnthropicの公式サイトでAPIキーを取得し、Pythonの環境を準備します。

    pip install anthropic
    export ANTHROPIC_API_KEY="your-api-key-here"

    基本的なチャットボットの実装

    以下がClaude APIを使った最もシンプルなチャットボットの実装です。

    import anthropic
    
    client = anthropic.Anthropic()
    
    def chat(user_message: str) -> str:
        message = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            messages=[
                {"role": "user", "content": user_message}
            ]
        )
        return message.content[0].text
    
    # 対話ループ
    while True:
        user_input = input("You: ")
        if user_input.lower() in ["quit", "exit"]:
            break
        response = chat(user_input)
        print(f"Claude: {response}")

    会話履歴を保持する

    実用的なチャットボットでは、会話の文脈を保持する必要があります。messagesリストに過去のやり取りを蓄積することで実現できます。

    class ChatBot:
        def __init__(self, system_prompt="あなたは親切なアシスタントです。"):
            self.client = anthropic.Anthropic()
            self.system = system_prompt
            self.messages = []
    
        def send(self, user_message: str) -> str:
            self.messages.append({"role": "user", "content": user_message})
            response = self.client.messages.create(
                model="claude-sonnet-4-20250514",
                max_tokens=2048,
                system=self.system,
                messages=self.messages
            )
            assistant_msg = response.content[0].text
            self.messages.append({"role": "assistant", "content": assistant_msg})
            return assistant_msg
    
    bot = ChatBot("あなたはPythonプログラミングの専門家です。")
    print(bot.send("リスト内包表記について教えてください"))
    print(bot.send("具体例をもう少し見せてください"))

    エラーハンドリング

    本番環境ではレート制限やネットワークエラーへの対策が必要です。anthropicライブラリは自動リトライ機能を備えていますが、明示的なエラーハンドリングも重要です。

    import anthropic
    from anthropic import RateLimitError, APIConnectionError
    
    try:
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            messages=[{"role": "user", "content": "Hello"}]
        )
    except RateLimitError:
        print("レート制限に達しました。しばらく待ってから再試行してください。")
    except APIConnectionError:
        print("API接続エラー。ネットワーク状態を確認してください。")

    まとめ

    Claude APIは直感的なインターフェースで、少ないコード量でAIチャットボットを構築できます。次回はFunction Callingを活用した、外部データベースと連携するチャットボットの作り方を紹介します。

IP: 取得中...
216.73.216.41216.73.216.41