Python ローカルLLM

OllamaはWebのAPIを内蔵しており、localhost:11434 にpostでアクセスすることで問い合わせが可能。

import requests
import json

question = "人工知能とは何ですか？"

response = requests.post("http://localhost:11434/api/generate", 
                            json={"model": "llama3.2", "prompt": question})

for r in response.iter_lines() :
    res = json.loads(r)
    print(res['response'], end='');

Ollamaのライブラリ利用

OllamaのPythonライブラリを利用するとより簡単にOllamaをPythonで利用できる。

インストール

pip install ollama

利用例

import ollama

# 使用例
query = "人工知能とは何ですか？"

response = ollama.generate(model='llama3.2', prompt=query)
print( response['response'] )

過去の会話の反映

LLMとのやりとりでは原則としてその一度きりの内容で判断した会話になる。これを過去の会話を反映させるには過去のやりとりも毎回LLMに送信する必要がある。

このときに利用するのがmessagesである。これは過去の会話を含めたリストであり、roleがuserのときにはユーザの入力、assistantのときにはLLMの解答を表す。

messages=[
    {"role": "user", "content": "今日は金曜です。明日は?"},
    {"role": "assistant", "content": "土曜です"},
    {"role": "user", "content": "では明後日は?"},
]

messagesを利用して問い合わせを行うには、ollama.chat()を使用する。このとき、引数streamをTrueにすることで、解答を順次得られるので、それを少しずつ画面に出すようにする。

# LLMに問い合わせ
stream = ollama.chat(
    model='llama3.2',
    messages=messages,
    stream=True
)

# 内容の表示
for chunk in stream:
    content = chunk['message']['content']
    print(content, end='', flush=True)

チャット

ユーザが入力した内容を元にチャットを行うコードを作成する。会話内容をmessagesに保存していき、過去の会話内容を元に現在の会話を行えるようにする。

import ollama

messages = []

while True:
    user_input = input("\nあなた: ")
    
    if user_input == 'bye':
        print("チャットを終了します。")
        break

    messages.append({'role': 'user', 'content': user_input})

    stream = ollama.chat(
        model='llama3.2',
        messages=messages,
        stream=True
    )

    print("\nAI: ", end='', flush=True)
    
    ai_response = "" # 最終的な解答文字列を保存

    for chunk in stream:
        content = chunk['message']['content']
        print(content, end='', flush=True)
        ai_response += content

    messages.append({'role': 'assistant', 'content': ai_response})

Streamlitを利用し、WebでOllamaを利用する。

import streamlit as st
import ollama

st.title("Ollama Chat サンプル")

# セッション状態の初期化
if "messages" not in st.session_state:
    st.session_state.messages = []

# チャット履歴の表示
for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

# ユーザー入力
if prompt := st.chat_input("メッセージを入力してください"):
    # ユーザーメッセージの追加
    st.session_state.messages.append({"role": "user", "content": prompt})
    with st.chat_message("user"):
        st.markdown(prompt)

    # Ollama API呼び出しの準備
    with st.chat_message("assistant"):
        message_placeholder = st.empty()
        full_response = ""
        
        # Ollamaクライアントを使用してストリーミング応答を取得
        stream = ollama.chat(
            model='llama3.2',
            messages=st.session_state.messages,
            stream=True
        )

        for chunk in stream:
            content = chunk['message']['content']
            message_placeholder.markdown(full_response + "┃")
            full_response += content
        
        message_placeholder.markdown(full_response)

    # アシスタントの応答をメッセージ履歴に追加
    st.session_state.messages.append({"role": "assistant", "content": full_response})

LLMが元々学習した内容には限りがあり、専門知識や組織内のローカルな内容、最新の情報は学習していない。これらの情報をプロンプトに付加することで適切な回答を得ることが出来る。

ただし、プロンプトの最大長には限界がある(Llama3で8kトークン、GPT-4で128Kトークンなど)。また、プロンプトが長くなることで応答に時間がかかるようになり、また、応答の精度も落ちる。

RAG（Retrieval Augmented Generation：検索拡張生成）とは、LLMに問い合わせを行う際に、情報源から別途検索・抽出した内容をプロンプトに付加すること。これらの情報を付加することで適切な回答を得ることが出来る。

RAGの流れ

1. 元となる情報の取得

   テキストファイルやPDFファイル、Webからのダウンロード、データベースの検索で情報元のデータを取得する。

2. 必要な情報の抽出

   一般的に、元となるデータは膨大なため、全てをプロンプト内に納めることは困難である。このため、必要な情報のみを抽出する。
   具体的には問合せ内容と似た内容の文章を抽出する。この際には文章をベクトル化し類似度の計算を行う。

3. プロンプトに付加

   
   プロンプト内で「この情報を元に回答を行うように」と指示し、抽出情報を付加する。

4. LLMへ問い合わせ

テキストファイルをRAGとして使用

テキストファイルを読み込み、そのテキストを文章単位に分割する。分割した文章をチャンクと呼ぶ。

まず、テキストファイルを読み込む関数 load_document を作成する。

def load_document(file_path):
    with open(file_path, 'r', encoding='utf-8') as file:
        return file.read()

文章を読み込みチャンクに分割する関数 load_chunk を作成する。

import spacy

nlp = spacy.load("ja_ginza")

def load_chunks(text, max_length=300, overlap=30):
    doc = nlp(text)
    chunks = []
    current_chunk = ""
    
    for sent in doc.sents:
        if len(current_chunk) + len(sent.text) <= max_length:
            current_chunk += sent.text
        else:
            chunks.append(current_chunk)
            current_chunk = sent.text[-overlap:]
    
    if current_chunk:
        chunks.append(current_chunk)
    
    return chunks

続いて、クエリーとチャンクのベクトル化を行い、そのコサイン類似度が高いチャンクを抽出する。

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

def get_similar_chunks(query, chunks, top_n=10):
    # クエリと全チャンクをベクトル化
    query_embedding = [nlp(query).vector]
    chunk_embeddings = [nlp(chunk).vector for chunk in chunks]
    
    # コサイン類似度を計算
    similarities = cosine_similarity(query_embedding, chunk_embeddings)[0]
    
    # 類似度が高いtop_n件を取得
    top_indices = np.argsort(similarities)[-top_n:][::-1]
    
    return [chunks[i] for i in top_indices]

最後にメイン処理を作成。

import ollama

if __name__ == "__main__":
    text = load_document('kousoku.txt')
    chunks = load_chunks(text)

    while(True):
        query = input('質問：')
        if query == 'q':
            break

        # 質問と似たチャンクを取得
        similar_chunks = get_similar_chunks(query, chunks)
        context = "\n\n".join(similar_chunks) # 各行を改行で挟んで結合
        
        # プロンプトの作成
        prompt = f"以下の情報を参考にして質問に答えてください：\n\n{context}\n\n質問: {query}"

        response = ollama.generate(model='llama3.2', prompt=prompt)
        print( response['response'] )

上記の例だと全チャンクをベクトル化することを問い合わせのたびに毎回行っている。これは一度だけ行えば良い処理であり、その結果を保存しておいて、問い合わせ時に読み込んで使用すれば良い。

そのために利用できるのがベクトルデータベースと呼ばれるものである。これを利用すれば文章とそのベクトルを保存し、類似するベクトルの検索も出来る。

ここではFacebookが開発したオープンソースのベクトルデータベースであるFAISS(Facebook AI Similarity Search)を利用し、ベクトルデータを保存し、問い合わせ時に活用する。

インストールはlangchainをインストールすることでFAISSが使用できる。

pip install langchain

ベクトルデータベースの保存

まずは先ほどと同様にload_documentとload_chunksの関数を定義し、以下のように呼び出す。

text = load_document('kousoku.txt')
chunks = load_chunks(text)

次にベクトル化を行うモデルを生成する。これはOllamaに備わっているOllamaEmbeddingsを利用する。

from langchain_community.embeddings import OllamaEmbeddings

embeddings = OllamaEmbeddings(model="llama3.2")

次にベクトルデータベース FAISSのオブジェクトを生成する。これはFAISS.from_texts でチャンクとベクトルかを行うモデルを指定するだけである。

faiss = FAISS.from_texts(chunks, embeddings)

最後に保存を行う。これはフォルダを指定する。これを後で読み込む。

faiss.save_local('./kousoku.faiss')

実行すると、指定したフォルダにベクトル化したデータが保存される。

ベクトルデータベースの利用

別ファイルで検索を行う。まず、ベクトルデータの読み込みを行う関数load_vectorstoreを作成する。

import ollama
from langchain_community.embeddings import OllamaEmbeddings
from langchain_community.vectorstores import FAISS

# ファイルからベクトルストアを読み込む
embeddings = OllamaEmbeddings(model='llama3.2')
vectorstore = FAISS.load_local(f'./kousoku.faiss', embeddings,allow_dangerous_deserialization=True)

これを使い、検索を行う。

while(True):
    query = input('質問：')
    if query == 'q':
        break
    
    # 質問と似たチャンクを取得
    similar_chunks = vectorstore.similarity_search(query, k=8)
    context = "\n".join([doc.page_content for doc in similar_chunks])

    # プロンプトの作成
    prompt = f"以下の情報を参考にして質問に答えてください：\n\n{context}\n\n質問: {query}"

    response = ollama.generate(model='llama3.2', prompt=prompt)
    print( response['response'] )