Cohere Embed 4で作る！パワポ資料に強いRAGシステム

$ pip install cohere google-genai python-dotenv tqdm faiss-cpu pymupdf

code/
├── sample.py      
├── .env           
├── slides/        
│   └── company-profile.pdf
└── images/        

.env

COHERE_API_KEY="Cohere Platformで発行したキー">
GEMINI_API_KEY="Google AI Studioで発行したキー">

sample.py

from pathlib import Path
import fitz

def convert_pdf_to_images(pdf_path, output_dir, zoom=2.0):
    """
    指定されたPDFファイルをスライド単位で画像（PNG）に変換する関数です。

    Parameters:
    ----------
    pdf_path : str or Path
        入力PDFファイルのパス。PowerPointからエクスポートしたPDFを想定しています。

    output_dir : str or Path
        変換後のPNG画像を保存するディレクトリ。存在しない場合は自動で作成されます。
    
    zoom : float (default=2.0)
        出力画像の拡大率。2.0 はおおよそ 200dpi 相当です。高精度な画像ベクトル化を行いたい場合に調整可能です。

    Returns:
    -------
    image_paths : list of str
        変換された各スライド画像（PNG）のファイルパス一覧。
        ファイル名は `_.png` の形式になります。
    """
    output_dir = Path(output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)

    pdf_name = Path(pdf_path).stem  

    doc = fitz.open(pdf_path)
    image_paths = []

    for i, page in enumerate(doc):
        mat = fitz.Matrix(zoom, zoom)
        pix = page.get_pixmap(matrix=mat)
        image_path = output_dir / f"{pdf_name}_{i+1}.png"
        pix.save(str(image_path))
        image_paths.append(str(image_path))

    doc.close()
    return image_paths

sample.py

import cohere
import base64
from dotenv import load_dotenv

load_dotenv()

co = cohere.ClientV2()



def get_image_embedding(image_path):
    """
    指定した画像ファイル（PNG）を Cohere Embed 4 を使ってベクトル化する関数です。
    画像は base64 形式でエンコードされた上で、Cohere API に送信され、画像内容に基づいた埋め込みベクトル（数値ベクトル）が返されます。

    Parameters:
    ----------
    image_path : str or Path
        ベクトル化したい画像ファイルのパス。
        今回はスライドをPNGに変換した画像が対象です。

    Returns:
    -------
    response : cohere.EmbedByTypeResponseEmbeddings
        Cohere API から返されるレスポンスオブジェクト。
        ベクトルは `response.embeddings.float_[0]` から取得できます。
    """
    with open(image_path, "rb") as f:
        image_data = f.read()
    stringified_buffer = base64.b64encode(image_data).decode("utf-8")
    content_type = "image/png"
    image_base64 = f"data:{content_type};base64,{stringified_buffer}"

    response = co.embed(
        model="embed-v4.0",
        input_type="image",
        embedding_types=["float"],
        images=[image_base64],
    )
    return response

sample.py

import faiss
import numpy as np
from pathlib import Path
from tqdm import tqdm

load_dotenv()

co = cohere.ClientV2()

def build_faiss_index(image_paths):
    """
    画像ファイルから埋め込みベクトルを生成し、それらを使って FAISS のインデックスを構築する関数です。

    Cohere Embed 4 でベクトル化されたスライド画像をローカル環境で検索可能にするために、
    FAISS（Facebook AI Similarity Search）を使ってインメモリのベクトル検索インデックスを生成します。

    Parameters:
    ----------
    image_paths : list of str or Path
        ベクトル化したい画像ファイル（今回はPNG）のパスのリスト。

    Returns:
    -------
    index : faiss.IndexFlatL2
        L2距離（ユークリッド距離）を用いたFAISSのベクトル検索インデックス。
    
    image_paths : list of str
        元の画像パスリスト（検索結果とインデックス番号を対応させるために返します）。
    """
    vectors = []
    for path in tqdm(image_paths, desc="埋め込み生成中"):
        res = get_image_embedding(path)
        embedding = res.embeddings.float_[0]
        vectors.append(np.array(embedding, dtype="float32"))

    index = faiss.IndexFlatL2(len(vectors[0]))
    index.add(np.stack(vectors))
    return index, image_paths

def search_faiss_with_text(index, image_paths, text_query, k=3):
    """
    テキストクエリに基づいて、FAISSインデックス内の画像との類似検索を行う関数です。

    Cohere Embed 4 を使用して、ユーザーのテキストクエリを埋め込みベクトルに変換し、
    すでに登録されたスライド画像のベクトルと比較して、意味的に近い画像を上位k件まで返します。

    Parameters:
    ----------
    index : faiss.Index
        事前に構築されたFAISSインデックス（画像のベクトルが登録済み）。

    image_paths : list of str
        FAISSインデックスに対応する画像ファイルのパスリスト。検索結果とファイルの対応づけに使います。

    text_query : str
        類似画像を探したい自然言語クエリ。

    k : int (default=3)
        検索して取得する類似画像の件数。上位k件をスコア順に返します。

    Returns:
    -------
    List[str]
        類似度の高い画像ファイルのパスリスト（最大k件）。

    """
    response = co.embed(
        model="embed-v4.0",
        input_type="text",
        embedding_types=["float"],
        texts=[text_query],
    )
    query_vec = np.array(response.embeddings.float_[0], dtype="float32").reshape(1, -1)
    D, I = index.search(query_vec, k)

    return [image_paths[i] for i in I[0]]

sample.py

import os
from dotenv import load_dotenv
from pathlib import Path
from google import genai
from google.genai import types



def generate_answer_with_gemini(image_paths, text_prompt):
    """
    複数の画像とテキストプロンプトを使って、Google Gemini API によるマルチモーダル応答を生成する関数です。

    指定された画像群（例：検索結果のスライド画像）と自然言語プロンプトを組み合わせて、
    Gemini モデルに送信し、画像に基づいた自然言語の回答を取得します。

    Parameters:
    ----------
    image_paths : list of str or Path
        入力画像ファイルのパスリスト。FAISSで検索されたスライド画像のパス群。

    text_prompt : str
        Geminiモデルに渡す自然言語のプロンプト（質問）。

    Returns:
    -------
    response_text : str
        Gemini APIから返された応答。API呼び出しに失敗した場合はエラーメッセージを返します。

    Notes:
    -----
    - 画像はすべてバイナリとして読み込み、`Part.from_bytes()` でGemini形式のマルチモーダル入力に変換されます。
    - `model="gemini-2.5-pro-exp-03-25"` を使用しています（複数画像対応）。
    - `system_instruction` によって、「スライド画像に基づく回答」のような文脈を指示しています。
    - 環境変数 `GEMINI_API_KEY` が `.env` に設定されている必要があります。
    - API呼び出し時にエラーが発生した場合は例外をキャッチし、ユーザー向けのエラーメッセージを返します。
    """
    load_dotenv()
    api_key = os.getenv("GEMINI_API_KEY")
    if not api_key:
        raise ValueError("GEMINI_API_KEY is not set in the environment.")

    client = genai.Client(api_key=api_key)
    contents = [text_prompt]

    for path in image_paths:
        mime_type = "image/png" if path.endswith(".png") else "image/jpeg"
        with open(path, "rb") as f:
            image_bytes = f.read()
        part = types.Part.from_bytes(data=image_bytes, mime_type=mime_type)
        contents.append(part)

    try:
        response = client.models.generate_content(
            model="gemini-2.5-pro-exp-03-25",
            contents=contents,
            config=types.GenerateContentConfig(
                system_instruction="あなたはスライド画像に関する質問に答えるAIアシスタントです。スライド画像の内容を理解し、質問に答えてください。回答にはスライド画像の内容のみを使用してください。",
                ),
            )
        return response.text
    except Exception as e:
        return f"[Gemini APIエラー] {str(e)}"

sample.py

if __name__ == "__main__":
    slides_dir = Path("./slides")
    images_dir = Path("./images")
    all_image_paths = []

    for pdf_file in slides_dir.glob("*.pdf"):
        image_paths = convert_pdf_to_images(pdf_file, images_dir)
        all_image_paths.extend(image_paths)

    print(f"{len(all_image_paths)} 枚のスライド画像を生成しました。")
    print("画像を埋め込んでFAISSインデックスを構築中...")
    index, img_list = build_faiss_index(all_image_paths)
    print("FAISSインデックスの構築が完了しました。")
    
    while True:
        text_query = input("テキストクエリを入力してください（終了するには 'quit' と入力）: ")
        if text_query.strip().lower() == "quit":
            print("終了します。")
            break

        results = search_faiss_with_text(index, img_list, text_query)
        print("テキストクエリによる類似画像:")
        for r in results:
            print(r)

        print("\nGeminiによる回答:")
        answer = generate_answer_with_gemini(results, text_query)
        print(answer)

入力した質問

NTTデータの強みはなんですか

検索結果

テキストクエリによる類似画像:
images/company-profile_6.png
images/company-profile_7.png
images/company-profile_11.png

Geminiによる回答

NTTデータの強みは、スライドによると以下の点が挙げられます。

1.  **Foresight 起点の目利き力:** 社会や業界、顧客の変化、そしてあるべき姿に基づき、戦略を考え、最適な技術を提案する力。
2.  **We Create つくる力:** 幅広い業界のお客様とのビジネスを通じた豊富な業務ノウハウと高度な開発力。
3.  **システム管理/運用力:** 社会基盤となる重要なITシステムの安定的な稼働を実現してきたシステム管理・運用力。
4.  **We Connect つなぐ力:** データセンター、ネットワーク、マネージドサービスを主としたEdge to Cloudのサービス・オペレーション力。
5.  **人材・組織力:** 19万人超のあらゆる分野に精通するプロフェッショナル集団と、組織のフル活用によりプロジェクトを完遂する力。

また、競争優位性・差別化要因としては以下の点も挙げられています。

*   **多様性を持つビジネスポートフォリオ:** 幅広い国・地域/業種で事業を可能とする。
*   **世界最高水準のエンジニアリング力:** 新技術活用を含む。
*   **マネジメント力:** 大規模なプロジェクトをまとめる力。
*   **グローバル有数のインフラサービスプレイヤー:** データセンターやネットワーク等。
*   **顧客との長期にわたる揺るぎない関係性 (Long-Term Relationships)**

さらに、市場でのポジションや実績も強みとして示されています。

*   **グローバルブランド力:** IT Services brands 第6位。
*   **日本市場におけるリーダーシップ:** 日本市場ITサービス市場ランキング 第1位。
*   **世界トップレベルのデジタル技術:** 生成AIをはじめとする各種リーダー評価を多数獲得。
*   **データセンター事業者シェア:** 世界第3位。
*   **ITサービス提供の歴史:** 57年。