Lang Chain

上記でもつかった架空の楽器についてかかれたテキストに対して、ベクトル検索、キーワード検索をします。
検索結果をLLMの質問に付加して、これをLM Studio（GPT-OSS-20B）のプロンプトに入力しました。
何をやっているのかわかりやすくするため、手動で作業しました。
下記コードは、検索結果をリランキングして表示するものです。
ベクトル検索とキーワード検索のクエリを同じにすることは、それぞれの性質を考えた場合、適当でないと考えたため、別のものとしました。
このような検索の仕方がいいのかどうか、このあたりは試行錯誤中です。

import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
from sentence_transformers import SentenceTransformer
import re
from typing import List, Dict, Tuple
import json

class HybridRAGSystem:
    def __init__(self, embedding_model_name='all-MiniLM-L6-v2'):
        """
        ハイブリッド検索RAGシステム
        キーワード検索（TF-IDF）とベクトル検索を組み合わせ
        """
        self.embedding_model = SentenceTransformer(embedding_model_name)
        # 日本語対応のTF-IDFベクトライザー
        self.tfidf_vectorizer = TfidfVectorizer(
            stop_words=None,  # 日本語なので英語ストップワードを無効化
            ngram_range=(2, 4),  # 2-4文字のn-gramで日本語の語彙をカバー
            max_features=10000,
            analyzer='char',  # 文字ベースの解析で日本語に対応
            token_pattern=None,  # analyzerがcharの場合は不要
            min_df=1  # 最小文書頻度を1に設定（稀な語も含める）
        )
        
        # データストレージ
        self.documents = []
        self.document_embeddings = None
        self.tfidf_matrix = None
        self.is_indexed = False
    
    def add_documents(self, texts: List[str]):
        """
        文書をシステムに追加
        """
        self.documents.extend(texts)
        print(f"追加された文書数: {len(texts)}")
        print(f"総文書数: {len(self.documents)}")
    
    def build_index(self):
        """
        インデックスを構築（ベクトル埋め込み + TF-IDFマトリックス）
        """
        if not self.documents:
            raise ValueError("文書が追加されていません")
        
        print("インデックスを構築中...")
        
        # ベクトル埋め込みの生成
        print("ベクトル埋め込みを生成中...")
        self.document_embeddings = self.embedding_model.encode(
            self.documents, 
            show_progress_bar=True,
            convert_to_numpy=True
        )
        
        # TF-IDFマトリックスの構築
        print("TF-IDFマトリックスを構築中...")
        self.tfidf_matrix = self.tfidf_vectorizer.fit_transform(self.documents)
        
        self.is_indexed = True
        print("インデックス構築完了!")
    
    def keyword_search(self, query: str, top_k: int = 10, return_all_scores: bool = False) -> List[Tuple[int, float]]:
        """
        キーワード検索（TF-IDF）
        """
        if not self.is_indexed:
            raise ValueError("インデックスが構築されていません")
        
        # コサイン類似度の計算
        query_vector = self.tfidf_vectorizer.transform([query])
        similarities = cosine_similarity(query_vector, self.tfidf_matrix)[0]
        
        if return_all_scores:
            # 全文書のスコアを返す
            return [(idx, similarities[idx]) for idx in range(len(similarities))]
        
        # スコアでソートして上位K件を返す
        top_indices = np.argsort(similarities)[::-1][:top_k]
        results = [(idx, similarities[idx]) for idx in top_indices if similarities[idx] > 0]
        
        return results
    
    def vector_search(self, query: str, top_k: int = 10) -> List[Tuple[int, float]]:
        """
        ベクトル検索（意味的類似度）
        """
        if not self.is_indexed:
            raise ValueError("インデックスが構築されていません")
        
        # クエリを埋め込みベクトルに変換
        query_embedding = self.embedding_model.encode([query])
        
        # コサイン類似度の計算
        similarities = cosine_similarity(
            query_embedding, 
            self.document_embeddings
        )[0]
        
        # スコアでソートして上位K件を返す
        top_indices = np.argsort(similarities)[::-1][:top_k]
        results = [(idx, similarities[idx]) for idx in top_indices]
        
        return results
    
    def hybrid_search(
        self, 
        query: str, 
        top_k: int = 10, 
        keyword_weight: float = 0.3, 
        vector_weight: float = 0.7,
        rerank: bool = True,
        context_lines: int = 2,
        keyword_query: str = None
    ) -> List[Dict]:
        """
        ハイブリッド検索（近隣行を含むコンテキスト付き）
        
        Args:
            query: ベクトル検索用のクエリ
            top_k: 返す結果数
            keyword_weight: キーワード検索の重み
            vector_weight: ベクトル検索の重み
            rerank: リランキングを行うか
            context_lines: 前後何行を含めるか
            keyword_query: キーワード検索専用クエリ（None の場合は query を使用）
        """
        if not self.is_indexed:
            raise ValueError("インデックスが構築されていません")
        
        # 各検索手法で結果を取得（全スコアを取得）
        # キーワード検索用クエリの決定

        actual_keyword_query = keyword_query if keyword_query is not None else query
        
        print(f"Debug: ベクトル検索クエリ: '{query}'")
        print(f"Debug: キーワード検索クエリ: '{actual_keyword_query}'")
        
        keyword_all_scores = self.keyword_search(actual_keyword_query, top_k, return_all_scores=True)
        vector_results = self.vector_search(query, top_k * 2)
        
        # キーワードスコア辞書を作成
        keyword_scores_dict = {idx: score for idx, score in keyword_all_scores}
        
        print(f"\nDebug: キーワード検索結果数: {len([s for s in keyword_all_scores if s[1] > 0])}")
        print(f"Debug: ベクトル検索結果数: {len(vector_results)}")
        if vector_results:
            print(f"Debug: ベクトル検索上位3件: {vector_results[:3]}")
        
        # スコアを統合
        combined_scores = {}
        
        # ベクトル検索の結果を追加
        for doc_idx, score in vector_results:
            combined_scores[doc_idx] = combined_scores.get(doc_idx, 0) + vector_weight * score
            # キーワードスコアも追加
            keyword_score = keyword_scores_dict.get(doc_idx, 0.0)
            combined_scores[doc_idx] += keyword_weight * keyword_score
        
        # スコアでソートして上位K件を選択
        sorted_results = sorted(
            combined_scores.items(), 
            key=lambda x: x[1], 
            reverse=True
        )[:top_k]
        
        # 結果を整形（コンテキスト行を含む）
        results = []
        vector_scores_dict = {idx: score for idx, score in vector_results}
        
        for doc_idx, combined_score in sorted_results:
            # 個別スコアも取得
            keyword_score = keyword_scores_dict.get(doc_idx, 0.0)
            vector_score = vector_scores_dict.get(doc_idx, 0.0)
            
            
            # コンテキスト行を取得
            context_content = self._get_context_lines(doc_idx, context_lines)
            
            result = {
                'document_id': doc_idx,
                'content': self.documents[doc_idx],
                'context_content': context_content,
                'combined_score': combined_score,
                'keyword_score': keyword_score,
                'vector_score': vector_score
            }
            results.append(result)
        
        # リランキング（オプション）
        if rerank:
            results = self.rerank_results(query, results)
        
        return results
    
    def _get_context_lines(self, doc_idx: int, context_lines: int) -> str:
        """
        指定された文書の前後の行を含むコンテキストを取得
        """
        start_idx = max(0, doc_idx - context_lines)
        end_idx = min(len(self.documents), doc_idx + context_lines + 1)
        
        context_parts = []
        for i in range(start_idx, end_idx):
            if i == doc_idx:
                # メイン行は強調
                context_parts.append(f">>> {self.documents[i]} <<<")
            else:
                context_parts.append(self.documents[i])
        
        return "\n".join(context_parts)
    
    def rerank_results(self, query: str, results: List[Dict]) -> List[Dict]:
        """
        検索結果のリランキング
        より詳細な類似度計算で順位を調整
        """
        if len(results) <= 1:
            return results
        
        # クエリとの詳細な類似度を再計算
        contents = [r['content'] for r in results]
        query_embedding = self.embedding_model.encode([query])
        content_embeddings = self.embedding_model.encode(contents)
        
        detailed_similarities = cosine_similarity(
            query_embedding, 
            content_embeddings
        )[0]
        
        # 新しいスコアで並び替え
        for i, result in enumerate(results):
            result['rerank_score'] = detailed_similarities[i]
            # 統合スコアを更新（重み付き平均）
            result['final_score'] = (
                0.7 * result['combined_score'] + 
                0.3 * detailed_similarities[i]
            )
        
        # 最終スコアでソート
        results.sort(key=lambda x: x['final_score'], reverse=True)
        
        return results
    
    def search_with_context(self, query: str, top_k: int = 5, keyword_query: str = None) -> str:
        """
        検索結果を使ってコンテキストを生成
        LLMに渡すためのプロンプト形式で返す
        """
        results = self.hybrid_search(query, top_k, keyword_query=keyword_query)
        
        if not results:
            return "関連する文書が見つかりませんでした。"
        
        context_parts = []
        for i, result in enumerate(results, 1):
            context_parts.append(f"[文書 {i}]")
            context_parts.append(f"スコア: {result['combined_score']:.4f}")
            context_parts.append(f"内容: {result['context_content']}")
            context_parts.append("")
        
        context = "\n".join(context_parts)
        
        prompt = f"""以下の検索結果を参考にして、質問に答えてください。

質問: {query}

検索結果:
{context}

回答:"""
        
        return prompt


# 使用例とデモ
def demo():
    # 外部ファイルから文書データを読み込み
    with open('lumi2.txt', 'r', encoding='utf-8') as file:
        sample_documents = [line.strip() for line in file if line.strip()]
    
    # システムの初期化
    rag_system = HybridRAGSystem()
    
    # 文書を追加
    rag_system.add_documents(sample_documents)
    
    # インデックスを構築
    rag_system.build_index()
    
    # 検索デモ
    #query = "LUMINARの演奏者について教えて"
    query = "コンサートで演奏される曲を教えて"
    print(f"\n検索クエリ: '{query}'")
    print("=" * 50)
    
    # ハイブリッド検索の実行（キーワード検索とベクトル検索で別のクエリを使用）
    #keyword_query = "演奏者"  # キーワード検索用の短いクエリ
    keyword_query = "楽曲"  # キーワード検索用の短いクエリ
    results = rag_system.hybrid_search(
        query=query, 
        top_k=20,
        keyword_query=keyword_query
    )
    
    for i, result in enumerate(results, 1):
        print(f"\n結果 {i}:")
        print(f"文書ID: {result['document_id']}")
        print(f"統合スコア: {result['combined_score']:.4f}")
        print(f"キーワードスコア: {result['keyword_score']:.4f}")
        print(f"ベクトルスコア: {result['vector_score']:.4f}")
        print(f"コンテキスト内容:")
        print(result['context_content'])
    
    # コンテキスト生成のデモ
    print("\n" + "=" * 50)
    print("LLM用プロンプト:")
    print("=" * 50)
    context_prompt = rag_system.search_with_context(query, top_k=10, keyword_query=keyword_query)
    print(context_prompt)

if __name__ == "__main__":
    demo()

RAGの仕組みは、LLMの回答の精度に大きく関わるので、とても興味があります。
まだまだ第一歩でしょうか。

Neo4Jは下記の環境
https://decode.red/net/archives/2198

インストール
pip install sentence-transformers
pip install matplotlib
pip install seaborn
pip install neo4j
pip install japanize-matplotlib
pip install faiss-gpu-cu12

from neo4j import GraphDatabase
from sentence_transformers import SentenceTransformer
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

from sklearn.cluster import KMeans
from sklearn.decomposition import PCA
import matplotlib.pyplot as plt
import japanize_matplotlib

import seaborn as sns

import faiss

documents = [
    "機械学習はAIの基本技術です",
    "機械学習はデータから自動的にパターンを学習する技術です",
    "深層学習はニューラルネットワークを多層化した手法です",
    "自然言語処理は人間の言語を計算機で処理する技術です",
    "今日は天気が良く、散歩日和です",
    "明日は雨が降る予報が出ています",
    "桜の花が美しく咲いています",
    "データサイエンスは統計学と計算機科学の融合分野です",
    "ビッグデータの解析には高性能な計算機が必要です"
]

model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')
# 文埋め込み生成
embeddings = model.encode(documents)

class DocumentGraphBuilder:
    def __init__(self, uri, user, password):
        self.driver = GraphDatabase.driver(uri, auth=(user, password))
    
    def close(self):
        self.driver.close()
    
    def create_document_nodes(self, documents):
        """文書ノードを作成"""
        with self.driver.session() as session:
            for i, (doc, embedding) in enumerate(zip(documents, embeddings)):
                session.run(
                    "CREATE (d:Document {id: $id, text: $text, embedding: $embedding})",
                    id=i, text=doc, embedding=embedding.tolist()
                )
    
    def create_similarity_edges(self, threshold=0.7):
        """類似性に基づいてエッジを作成"""
        with self.driver.session() as session:
            # 全文書ペアの類似性を計算してエッジ作成
            result = session.run("MATCH (d1:Document), (d2:Document) WHERE id(d1) < id(d2) RETURN d1, d2")
            
            for record in result:
                d1_embedding = np.array(record["d1"]["embedding"])
                d2_embedding = np.array(record["d2"]["embedding"])
                
                similarity = cosine_similarity([d1_embedding], [d2_embedding])[0][0]
                
                if similarity > threshold:
                    session.run(
                        """
                        MATCH (d1:Document {id: $id1}), (d2:Document {id: $id2})
                        CREATE (d1)-[:SIMILAR {score: $score}]->(d2)
                        """,
                        id1=record["d1"]["id"], 
                        id2=record["d2"]["id"], 
                        score=similarity
                    )

graph_builder = DocumentGraphBuilder("bolt://192.168.0.180:7687", "neo4j", "password")
graph_builder.create_document_nodes(documents)
graph_builder.create_similarity_edges(threshold=0.8)
graph_builder.close()


print(f"埋め込みの形状: {embeddings.shape}")

# クラスタリング
kmeans = KMeans(n_clusters=3, random_state=42)
clusters = kmeans.fit_predict(embeddings)
    
# 次元削減と可視化
pca = PCA(n_components=2)
embeddings_2d = pca.fit_transform(embeddings)
    
#plt.figure(figsize=(12, 8))
plt.figure(figsize=(10, 6))
scatter = plt.scatter(embeddings_2d[:, 0], embeddings_2d[:, 1], 
                         c=clusters, cmap='viridis')
plt.colorbar(scatter)
    
# 各点にテキストラベルを追加
for i, doc in enumerate(documents):
    plt.annotate(f"{i}: {doc[:20]}...", 
                (embeddings_2d[i, 0], embeddings_2d[i, 1]),
                xytext=(5, 5), textcoords='offset points', fontsize=8)
    
plt.title('文書クラスタリング結果')
plt.xlabel('PCA Component 1')
plt.ylabel('PCA Component 2')
plt.show()


# 類似性行列の計算
similarity_matrix = cosine_similarity(embeddings)
    
# 結果の可視化
plt.figure(figsize=(10, 6))
sns.heatmap(similarity_matrix, 
            xticklabels=range(len(documents)),
            yticklabels=range(len(documents)),
            annot=True, cmap='coolwarm')
plt.title('文章間の類似性行列')
plt.show()



class FastSimilaritySearch:
    def __init__(self):
        self.index = None
    
    def build_index(self, documents):
        """文書インデックスを構築"""
        # FAISS インデックス構築
        dimension = embeddings.shape[1]
        self.index = faiss.IndexFlatIP(dimension)  # 内積ベース
        
        # 正規化（コサイン類似度のため）
        faiss.normalize_L2(embeddings)
        self.index.add(embeddings.astype('float32'))
        
        print(f"インデックスに {len(documents)} 文書を追加しました")
    
    def search(self, query, k=5):
        """クエリに対する類似文書検索"""
        if self.index is None:
            raise ValueError("まずbuild_index()を実行してください")
        
        # クエリの埋め込み生成
        query_embedding = model.encode([query])
        faiss.normalize_L2(query_embedding)
        
        # 検索実行
        scores, indices = self.index.search(query_embedding.astype('float32'), k)
        
        # 結果整理
        results = []
        for score, idx in zip(scores[0], indices[0]):
            if idx != -1:  # 有効なインデックス
                results.append({
                    'document': documents[idx],
                    'score': score,
                    'index': idx
                })
        
        return results

search_engine = FastSimilaritySearch()
search_engine.build_index(documents)
query = "AIと機械学習について"
results = search_engine.search(query, k=3)
    
print(f"クエリ: {query}")
print("検索結果:")
for i, result in enumerate(results):
    print(f"{i+1}. スコア: {result['score']:.3f}")
    print(f"   文書: {result['document']}")
    print()

上書きされるので、実行のたびに削除

MATCH (n) DETACH DELETE n

このような学ぶ道具としてのAIの活用は、本当に便利に感じます。
これだったらどんな難解なコーディングでもいけそうな気がしてきます。

https://decode.red/net/archives/2208

参考）https://note.com/genaird/n/n89954c59c7e6

モデルは LMStudioのGemmaを使用します。

インストール

pip install langchain-mcp-adapters
pip install langgraph
pip install langchain_openai

client.py

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
import asyncio

model = ChatOpenAI(base_url="http://192.168.0.xxx:1234/v1", api_key="not-needed",  temperature=0.7)

server_params = StdioServerParameters(
    command="python",
    args=["server.py"],
)

async def run_agent():
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = await load_mcp_tools(session)
            agent = create_react_agent(model, tools)
            #agent_response = await agent.ainvoke({"messages": "what's (3 + 5) x 12 ?"})
            agent_response = await agent.ainvoke({"messages": "11 + 22 ?"})
            return agent_response

if __name__ == "__main__":
    result = asyncio.run(run_agent())
    print(result)

server.py

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Math")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b

@mcp.tool()
def multiply(a: int, b: int) -> int:
    """Multiply two numbers"""
    return a * b

if __name__ == "__main__":
    mcp.run(transport="stdio")

参考サイトにある、かっこ付きの計算はできなかったので、簡単なものにしました。ToolExecutorなるものを使用するらしいのですが、ライブラリのバージョンが違いうまくいかず。。
もう少しLangGraphを勉強してからまた試したいと思います。

まずは簡単なチャットで、動作確認。

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(base_url="http://192.168.0.xxx:1234/v1", api_key="not-needed",  temperature=0.7)
text = "hello"
print(llm.invoke(text).content)

import requests
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import CharacterTextSplitter
from langchain.chains import RetrievalQA
from langchain.schema.embeddings import Embeddings
from langchain_openai import ChatOpenAI

# LM Studio の埋め込み API を使うクラスを定義
class LMStudioEmbeddings(Embeddings):
    def __init__(self, api_url="http://192.168.0.xxx:1234/v1/embeddings", model_name="nomic-embed-text-v1.5.Q4_K_M.gguf"):
        self.api_url = api_url
        self.model_name = model_name

    def embed_documents(self, texts):
        return [self.embed_query(text) for text in texts]

    def embed_query(self, text):
        response = requests.post(
            self.api_url,
            headers={"Content-Type": "application/json"},
            json={
                "input": text,
                "model": self.model_name
            }
        )
        return response.json()["data"][0]["embedding"]

with open('text.txt', 'r') as file:
    raw_text = file.read()

# テキスト分割
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
texts = text_splitter.split_text(raw_text)

# ベクトルストア
embeddings = LMStudioEmbeddings()
docsearch = Chroma.from_texts(
    texts, 
    embeddings,   
    metadatas=[{"source": str(i)} for i in range(len(texts))],
    persist_directory="./chroma_db"
)

# LLM 本体は LM Studio 経由の ChatOpenAI
llm = ChatOpenAI(base_url="http://192.168.0.xxx:1234/v1", api_key="not-needed", temperature=0.7)

# 質問応答チェーンの作成
qa = RetrievalQA.from_chain_type(llm=llm, chain_type="stuff", retriever=docsearch.as_retriever())

query = "ルミナールの奏法について、４００文字くらいで日本語で教えて"
#print(qa.run(query))
print(qa.invoke(query)['result'])

エンベディングモデルと、言語モデルと二つLM Studioから得ています。

入力テキストはChatGPTで生成した、架空の楽器にまつわる架空の話にしました。

LM Studioのログ画面

Chromaが保存した、SQLiteの中身

回答結果

さまざまな要素を使った実装で、LangChainの利便性が理解できた気がします。

Large Language Models : Ollama

RAG (Retrieval-Augmented Generation) とは、大規模言語モデル (LLM) の回答精度を向上させるための技術です。LLMが学習した範囲を超えた情報を活用するために、外部のデータベースや検索エンジンから情報を取得し、それをLLMに与えて回答を生成させます。これにより、LLMの知識の限界を補い、より正確で最新の情報に基づいた回答を生成できます。（by AI)

LLMのモデルはGemmaを選択

KANTAN Play core （かんぷれ）
スイッチサイエンスの新製品で、LLMにはない情報だろうということで、このページのテキストを読み込ませる前と後とで、結果を比べました。

他にも、電子楽器のマニュアルなども試しましたが、結構幅広いジャンルで学習していることに驚かされました。本当にWebアクセスしていないのだろうかと、WiFiを切った状態でテストをするほどでした。
RAGの読み込み、質問、回答まで、実行時間は1分程度でした。こんな簡単にRAGが試せるとは、進化のスピードがすごい！

環境）Mac mini M2