下記ブログに継続します。

(Under Constraction)

サーバ起動
llama.cpp/build/bin/llama-server -m "/mnt/c/Users//.lmstudio/models/mmnga/ELYZA-japanese-Llama-2-7b-fast-instruct-gguf/ELYZA-japanese-Llama-2-7b-fast-instruct-q4_K_S.gguf" --host 0.0.0.0 --port 8080 --ctx-size 4096 --n-predict -1

from openai import OpenAI
import numpy as np
import math
import sys

client = OpenAI(
    base_url="http://localhost:8080/v1",
    api_key="not-needed"  # llama.cppではAPIキーは不要
)

#question = "富士山の高さは？"
question = "プリウスの重さは？"

# 1️⃣ 回答と log probability 情報を取得
try:
    resp = client.chat.completions.create(
        model="local-model",  # llama.cppでは任意のモデル名でOK
        messages=[{"role": "user", "content": question}],
        logprobs=True,
        top_logprobs=5,   # トップ5候補を取得してエントロピーを計算
        temperature=0,
        max_tokens=200,   # 最大200トークンに制限（無限ループ防止）
        timeout=30,       # 30秒でタイムアウト
    )
except Exception as e:
    print("=" * 60)
    print("⚠️ エラーが発生しました")
    print(f"エラー種類: {type(e).__name__}")
    print(f"エラー内容: {str(e)}")
    if "timeout" in str(e).lower():
        print("\n タイムアウトしました（30秒以内に応答が完了しませんでした）")
    print("=" * 60)
    sys.exit(1)

# デバッグ: レスポンスの内容を確認
print("=" * 60)
print("レスポンス情報:")
print(f"回答: {resp.choices[0].message.content}")

# 終了理由をチェック
finish_reason = resp.choices[0].finish_reason
print(f"\n終了理由: {finish_reason}")

if finish_reason == "length":
    print("⚠️ 最大トークン数（200トークン）に達しました。回答が途中で切れている可能性があります。")
elif finish_reason == "stop":
    print("✅ 正常に完了しました。")
elif finish_reason == "content_filter":
    print("⚠️ コンテンツフィルターにより停止されました。")
else:
    print(f"ℹ️ その他の理由で終了: {finish_reason}")

print(f"\nlogprobs: {resp.choices[0].logprobs}")
print("=" * 60)

# logprobs が None の場合の処理
if resp.choices[0].logprobs is None:
    print("\n⚠️ LMStudio は logprobs をサポートしていないようです。")
    print("回答のみ表示します。\n")
    print(f"質問: {question}")
    print(f"回答: {resp.choices[0].message.content}")
else:
    # 各トークンの情報を取得
    tokens = resp.choices[0].logprobs.content

    log_probs = []
    entropies = []

    for token_info in tokens:
        # トークンの log prob
        lp = token_info.logprob
        log_probs.append(lp)

        # トップkの確率を正規化してエントロピーを計算
        probs = [math.exp(t.logprob) for t in token_info.top_logprobs]
        probs = np.array(probs) / np.sum(probs)
        H = -np.sum(probs * np.log(probs + 1e-10))  # エントロピー
        entropies.append(H)

    # 2️⃣ スコア計算
    avg_logprob = np.mean(log_probs)              # 平均対数確率
    avg_entropy = np.mean(entropies)              # 平均エントロピー

    # 3️⃣ 正規化（0〜1スケールへ）
    # logprob → 高いほど良い → sigmoidで圧縮
    conf_from_logprob = 1 / (1 + math.exp(-avg_logprob * 2))

    # entropy → 低いほど良い → 反転して正規化
    conf_from_entropy = math.exp(-avg_entropy)

    # 4️⃣ 統合スコア
    final_confidence = 0.7 * conf_from_logprob + 0.3 * conf_from_entropy

    print(f"質問: {question}")
    print(f"平均log probability: {avg_logprob:.3f}")
    print(f"平均entropy: {avg_entropy:.3f}")
    print(f"確信度(logprob由来): {conf_from_logprob:.3f}")
    print(f"確信度(entropy由来): {conf_from_entropy:.3f}")
    print(f"最終確からしさスコア: {final_confidence:.3f}")
    print(f"回答: {resp.choices[0].message.content}")

・・・
終了理由: length
⚠️ 最大トークン数（200トークン）に達しました。
・・・
質問: プリウスの重さは？
平均log probability: -0.137
平均entropy: 0.201
確信度(logprob由来): 0.432
確信度(entropy由来): 0.818
最終確からしさスコア: 0.548
回答: プリウスの車両重量は、グレードによって異なります。
・・・

結果は、なかなか判定が難しいということでした。もう少しいろいろな実験をしたいと思います。ただ不確かな答えの場合、出力トークンが多くなるという特性はあるようです。いろんな条件の元、総合的に判定することはできる可能性はあるように感じました。

参考）
https://qiita.com/kanata564/items/e8aaf6d4aeb99842dc62

学習データ(new_knowledge.json)

学習

# Unsloth + Hugging Face + QLoRA で知識注入
import torch
from unsloth import FastLanguageModel
from datasets import load_dataset
from trl import SFTTrainer, SFTConfig

# ① ベースモデルをロード
model, tokenizer = FastLanguageModel.from_pretrained(
    "meta-llama/Meta-Llama-3-8B-Instruct",
    load_in_4bit=True,  # QLoRA用：4bit量子化
    max_seq_length=2048,
)

# ② データセットをロード
dataset = load_dataset("json", data_files="new_knowledge.json", split="train")

# データセットをフォーマット（テキストフィールドを追加）
def format_dataset(example):
    # Llama 3のチャット形式を使用
    text = f"<|begin_of_text|><|start_header_id|>user<|end_header_id|>\n\n{example['instruction']}<|eot_id|><|start_header_id|>assistant<|end_header_id|>\n\n{example['output']}<|eot_id|>"
    return {"text": text}

dataset = dataset.map(format_dataset, remove_columns=dataset.column_names)

# ③ LoRA アダプタを作成
model = FastLanguageModel.get_peft_model(
    model,
    r=16,                        # rank を増やして学習能力向上
    target_modules=["q_proj", "k_proj", "v_proj", "o_proj",
                    "gate_proj", "up_proj", "down_proj"],  # 全ての線形層を対象
    lora_alpha=16,
    lora_dropout=0,
    bias="none",
    use_gradient_checkpointing="unsloth",
)

# ④ ファインチューニング
trainer = SFTTrainer(
    model=model,
    tokenizer=tokenizer,
    train_dataset=dataset,
    dataset_text_field="text",
    max_seq_length=2048,
    args=SFTConfig(
        per_device_train_batch_size=1,  # バッチサイズを小さく
        gradient_accumulation_steps=4,
        warmup_steps=10,
        num_train_epochs=20,  # 小さいデータセットなのでエポックを増やす
        max_steps=60,  # 最大ステップ数を制限
        learning_rate=5e-4,  # 学習率を上げて強く学習
        fp16=not torch.cuda.is_bf16_supported(),
        bf16=torch.cuda.is_bf16_supported(),
        logging_steps=1,
        optim="adamw_8bit",
        weight_decay=0.001,  # weight decayを下げる
        lr_scheduler_type="linear",
        seed=3407,
        output_dir="outputs",
        save_strategy="steps",
        save_steps=20,
        save_total_limit=3,
        load_best_model_at_end=False,
    ),
)

trainer.train()

# ⑤ 保存
model.save_pretrained("llama3-x1000-lora")
tokenizer.save_pretrained("llama3-x1000-lora")

推論

# Unsloth LoRA チャットボット推論プログラム
import torch
from unsloth import FastLanguageModel

# ① モデルとLoRAアダプタをロード
model, tokenizer = FastLanguageModel.from_pretrained(
    model_name="llama3-x1000-lora",  # 学習済みLoRAアダプタのパス
    max_seq_length=2048,
    load_in_4bit=True,
)

# 推論モードに設定
FastLanguageModel.for_inference(model)

# ② チャット関数
def chat(instruction):
    # Llama 3のチャット形式でプロンプトをフォーマット
    prompt = f"<|begin_of_text|><|start_header_id|>user<|end_header_id|>\n\n{instruction}<|eot_id|><|start_header_id|>assistant<|end_header_id|>\n\n"

    # トークナイズ
    inputs = tokenizer([prompt], return_tensors="pt").to("cuda")
    input_length = inputs.input_ids.shape[1]

    # 生成
    outputs = model.generate(
        **inputs,
        max_new_tokens=256,
        temperature=0.3,  # 温度を下げて正確性向上
        top_p=0.9,
        do_sample=True,
        repetition_penalty=1.15,
        use_cache=True,
        eos_token_id=tokenizer.eos_token_id,
    )

    # 入力部分を除いて新しく生成された部分のみデコード
    generated_ids = outputs[0][input_length:]
    response = tokenizer.decode(generated_ids, skip_special_tokens=True)

    return response.strip()

# ③ 対話ループ
def main():
    print("=" * 50)
    print("Unsloth LoRA チャットボット")
    print("=" * 50)
    print("終了するには 'exit' または 'quit' を入力してください")
    print()

    while True:
        # ユーザー入力
        user_input = input("あなた: ").strip()

        # 終了チェック
        if user_input.lower() in ['exit', 'quit', '終了']:
            print("チャットボットを終了します。")
            break

        if not user_input:
            continue

        # 推論実行
        print("AI: ", end="", flush=True)
        response = chat(user_input)
        print(response)
        print()

if __name__ == "__main__":
    main()

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

参考）
https://learn.microsoft.com/ja-jp/dotnet/standard/native-interop/pinvoke

コンソールアプリケーションのプロジェクト作成

dotnet new console -n interop

上記サイトのサンプルから、アンマネージライブラリを使ってプロセスIDの取得とディレクトリ情報の取得をします。

using System;
using System.Runtime.InteropServices;

namespace PInvokeSamples
{
    public static partial class Program
    {
        // Define a delegate that has the same signature as the native function.
        private delegate int DirClbk(string fName, ref Stat stat, int typeFlag);

        // Import the libSystem shared library and define the method
        // corresponding to the native function.
        [LibraryImport("libSystem.dylib")]
        private static partial int getpid();

        // Import the libc and define the method to represent the native function.
        [LibraryImport("libSystem.dylib", StringMarshalling = StringMarshalling.Utf16)]
        private static partial int ftw(string dirpath, DirClbk cl, int descriptors);



        // Implement the above DirClbk delegate;
        // this one just prints out the filename that is passed to it.
        private static int DisplayEntry(string fName, ref Stat stat, int typeFlag)
        {
            Console.WriteLine("{0} / {1}", fName, stat.UserID);
            return 0;
        }

        public static void Main(string[] args)
        {
            // Invoke the function and get the process ID.
            int pid = getpid();
            Console.WriteLine(pid);

            // Call the native function.
            // Note the second parameter which represents the delegate (callback).
            ftw(".", DisplayEntry, 10);
        }
    }

    // The native callback takes a pointer to a struct. This type
    // represents that struct in managed code.
    [StructLayout(LayoutKind.Sequential)]
    public struct Stat
    {
        public uint DeviceID;
        public uint InodeNumber;
        public uint Mode;
        public uint HardLinks;
        public uint UserID;
        public uint GroupID;
        public uint SpecialDeviceID;
        public ulong Size;
        public ulong BlockSize;
        public uint Blocks;
        public long TimeLastAccess;
        public long TimeLastModification;
        public long TimeLastStatusChange;
    }
}

実行

dotnet run

以下はディレクトリ情報取得の動作確認用のCプログラムです。

#define _XOPEN_SOURCE 500  // ftwのために必要
#include <ftw.h>
#include <stdio.h>
#include <stdlib.h>

// コールバック関数: ファイルごとに呼ばれる
int list(const char *fpath, const struct stat *sb, int typeflag) {
    printf("%s %u %lld\n", fpath, sb->st_uid, sb->st_size);
    return 0; // 0を返すと続行、非0を返すと中断
}

int main(int argc, char *argv[]) {
    if (argc < 2) {
        fprintf(stderr, "Usage: %s <directory_path>\n", argv[0]);
        exit(EXIT_FAILURE);
    }

    // ftw関数でディレクトリを再帰的に探索
    if (ftw(argv[1], list, 16) == -1) {
        perror("ftw");
        exit(EXIT_FAILURE);
    }

    return 0;
}

ビルド、実行

gcc -o ftw_sample ftw_sample.c
./ftw_sample .

InvokeSampleのstruct Statの定義が不正確のため、実際とはアライメントがずれてしまっていました。

前回に続き、.Net/Mac でした。

UnityはC#
https://decode.red/net/archives/844

かなり前にWindows以外でC#を試したときは、下記のようにmonoを使いました。
http://crossframe.iiv.jp/?s=mono

VS Code / macOS を使ったC#の開発は初めてでしたので、いろいろメモすることも目的です。

インストールファイル
dotnet-sdk-8.0.408-osx-arm64.pkg (205,144,863 バイト)

VS Codeや、Xcodeはすでに入っているものとします。（要マイクロソフトまたはGithubアカウント）

VS Code機能拡張

コマンドから新規プロジェクトを作成できます。（コンソールプロジェクト）

Program.cs

class Program
{
    static void Main()
    {
        // JSON → CSV
        JsonToCsvConverter.Convert("input.json", "output.csv");

        // CSV → JSON
        CsvToJsonConverter.Convert("input.csv", "output.json");
    }
}

input.json

[
  { "Name": "Alice", "Age": "30", "City": "Tokyo" },
  { "Name": "Bob", "Age": "25", "City": "Osaka" },
  { "Name": "Charlie", "Age": "35", "City": "Kyoto" }
]

input.csv

Name,Age,City
Alice,30,Tokyo
Bob,25,Osaka
Charlie,35,Kyoto

JsonToCsv.cs

using System;
using System.Collections.Generic;
using System.IO;
using System.Text;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;

class JsonToCsvConverter
{
    public static void Convert(string jsonFilePath, string csvFilePath)
    {
        var json = File.ReadAllText(jsonFilePath);
        var array = JArray.Parse(json);

        var sb = new StringBuilder();

        // ヘッダー
        var headers = ((JObject)array[0]).Properties();
        sb.AppendLine(string.Join(",", headers.Select(h => h.Name)));

        // データ
        foreach (var obj in array)
        {
            var values = ((JObject)obj).Properties().Select(p => p.Value.ToString());
            sb.AppendLine(string.Join(",", values));
        }

        File.WriteAllText(csvFilePath, sb.ToString());
    }
}

CsvToJson.cs

using System;
using System.Collections.Generic;
using System.IO;
using Newtonsoft.Json;

class CsvToJsonConverter
{
    public static void Convert(string csvFilePath, string jsonFilePath)
    {
        var lines = File.ReadAllLines(csvFilePath);
        var headers = lines[0].Split(',');

        var jsonList = new List<Dictionary<string, string>>();

        for (int i = 1; i < lines.Length; i++)
        {
            var values = lines[i].Split(',');
            var dict = new Dictionary<string, string>();

            for (int j = 0; j < headers.Length; j++)
            {
                dict[headers[j]] = values[j];
            }

            jsonList.Add(dict);
        }

        var json = JsonConvert.SerializeObject(jsonList, Formatting.Indented);
        File.WriteAllText(jsonFilePath, json);
    }
}

ファイル構成

パッケージインストールと実行

dotnet add package Newtonsoft.Json
dotnet run

ビルドはVSCodeの三角ボタンから（やり方はいろいろあり）

JSONは今後 System.Text.Jsonを使うことになりそうです。

https://learn.microsoft.com/ja-jp/dotnet/standard/serialization/system-text-json/migrate-from-newtonsoft?pivots=dotnet-9-0

次にMAUIも試してみました。（サンプルプログラムのビルドと実行）

workloadのインストール

sudo dotnet workload install maui
dotnet workload list

実行

dotnet run

参考）https://qiita.com/aqua_ix/items/ba9533d60633abe4c850