タグ: Python

  • 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を活用した、外部データベースと連携するチャットボットの作り方を紹介します。

  • 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サイトの商品検索、カスタマーサポート、社内ツールの自動化など、応用範囲は広大です。

  • 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として優秀で、小〜中規模のドキュメントであれば十分な性能を発揮します。

  • 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のバックエンド、マイクロサービス、プロトタイプ開発など幅広い場面で活躍します。

IP: 取得中...
216.73.216.31216.73.216.31