Local LLMをPython開発環境に活用 備忘録

概要

開発作業フォルダ内のDataBaseからLLMに作ってもらったSQLでDataFrameを返却させる。複雑なSQLが苦手なオイラ用のバッチスクリプト。サクサク変更して関数化したいのでMCPサーバなどは立てず簡単に済ませる。SELECT文のみ受け付けるガードレール付き。
インターネット環境や最近重宝しているantigravityが使えないときなどに、主にJupyter環境から呼び出して使用する。

• 実施期間: 2026年2月
• 環境：Ubuntu22.04LTS
• LLM host: Ollama
• LLM: qwen2.5-coderなど、coding instraction済みのモデル
• package: pip install ollamaだけ

1. LLM_db.py

LLM: Ollamaの"qwen2.5-coder:7b"
db: SQLite3で作成
このdbスキーマ情報（table名 or table名+フィールド名）はget_db_schema()で取得してsystem promptに入れてLLMへ渡すが、dbが大きかったりOllamaをCPUで動かすならtable名だけを渡すようにする。
今回使用したdbにはyfinanceで取得している株価情報が入っている。

import os
import ollama
import sqlite3
import re
import pandas as pd

# --- Configuration ---
# 適宜変更
MODEL = "qwen2.5-coder:7b
DB_PATH = "stock_data.db"

def get_db_schema(db_path, with_details=False):
    """
    DBからテーブル情報を取得してテキスト化する。
    with_details=True の場合、各テーブルのCREATE文(フィールド情報付き)も含める。
    """

    # ファイルの存在チェック
    if not os.path.exists(db_path):
        print(f"Error: DB file not found: {os.path.abspath(db_path)}")
        return None 

    conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
    cursor = conn.cursor()
    
    if with_details:
        # テーブル名とCREATE文（スキーマ詳細）を取得
        query = "SELECT name, sql FROM sqlite_schema WHERE type='table' AND name NOT LIKE 'sqlite_%';"
    else:
        # テーブル名のみを取得
        query = "SELECT name FROM sqlite_schema WHERE type='table' AND name NOT LIKE 'sqlite_%';"
        
    cursor.execute(query)
    tables = cursor.fetchall()
    conn.close()
    
    schema_text = "Available Tables:\n"
    for row in tables:
        if with_details:
            table_name, create_sql = row
            # schema_text += f"- Table: {table_name}\n  Schema: {create_sql}\n\n"
            # CREATE文をそのまま渡すのが、LLMにとって最も情報量が多い
            schema_text += f"{create_sql}\n\n"
        else:
            table_name = row[0]
            schema_text += f"- {table_name}\n"
    
    return schema_text


def get_sql_from_llm(user_query, schema_info):
    """
    取得したスキーマ情報をプロンプトに動的に埋め込む
    Rulesは用途に応じて変更する
    """
    system_prompt = f"""
You are an expert SQL engineer. Generate a SQLite SELECT statement.

[Database Schema]
{schema_info}

[Rules]
1. Respond ONLY with the raw SQL string. No markdown, no explanation.
2. Only "SELECT" statements are allowed.
3. If the user asks for 'latest' or 'recent', use: ORDER BY date DESC.
4. The field names 'symbol' in the 'info_tbl' and 'ticker' in the 'hist_tbl' both refer to the ticker code field.
"""
    
    response = ollama.chat(model=MODEL, messages=[
        {'role': 'system', 'content': system_prompt},
        {'role': 'user', 'content': user_query}
    ])
    
    # sql = response['message']['content'].strip()
    sql = response['message']['content']
    match = re.search(r"(SELECT.*?;)", sql, re.DOTALL | re.IGNORECASE)
    if match:
        sql = match.group(1)
    return sql

def execute_readonly_query(sql):
    """
    SQLを実行し、Pandas DataFrameとして返す
    """
    # 基本的なセーフティチェック
    forbidden_keywords = ["INSERT", "UPDATE", "DELETE", "DROP", "ALTER", "TRUNCATE"]
    if not sql.upper().startswith("SELECT"):
        return None, "Error: Only SELECT statements are allowed."
    
    for word in forbidden_keywords:
        if word in sql.upper():
            return None, f"Error: Forbidden keyword '{word}' detected."

    try:
        conn = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True) # 読み取り専用モードで接続
        df = pd.read_sql_query(sql, conn)
        conn.close()
        return df, None
    except Exception as e:
        return None, str(e)


def get_df_by_prompt(user_query, db_path):
    """
    プロンプトからSQLを生成し、DBからDataFrameを取得する
    main()を置き換える関数で、jupyter notebookで利用する
    """
    schema_info = get_db_schema(db_path) # 1. 起動時にDBから構造を読み取る
    if schema_info is None:
        return None

    generated_sql = get_sql_from_llm(user_query, schema_info) # 2. LLMに最新の構造を伝える
    print(f"Generated SQL:\n{generated_sql}")
    df, error = execute_readonly_query(generated_sql)
    if error:
        print(f"Execution Error: {error}")
        return None # エラーの場合はNoneを返す
    elif df.empty:
        print("No data found.")
        return None # データがない場合はNoneを返す

    return df # データがある場合はDataFrameを返す


def main_loop():
    print(f"--- Local DB Query Assistant (Model: {MODEL}) ---")
    print(f"Target DB: {DB_PATH}\n")

    while True:
        try:
            # --- 実行フロー ---
            schema_info = get_db_schema(DB_PATH) # 1. 起動時にDBから構造を読み取る

            user_input = input("プロンプトを入力してください (exitで終了): ")
            if user_input.lower() in ['exit', 'quit', 'q']:
                break

            print(f"Generating SQL...", end="\r")
            generated_sql = get_sql_from_llm(user_input, schema_info) # 2. LLMに最新の構造を伝える
            print(f"Generated SQL:\n{generated_sql}")
            
            print(f"Generated SQL: {generated_sql}")
            print("-" * 30)

            df, error = execute_readonly_query(generated_sql)
            
            if error:
                print(f"Execution Error: {error}")
            elif df is not None:
                if df.empty:
                    print("No data found.")
                else:
                    print(df)
            
            print("-" * 30 + "\n")

        except KeyboardInterrupt:
            print("\nExiting...")
            break

if __name__ == "__main__":
    main_loop()

2. get_df_by_prompt()をJupyterから呼ぶ

初回の関数呼び出し時はLLMのロードに時間がかかるが、それ以降はGPU環境で1sec未満、CPU環境で5sec程度でdataframeが返却される。
SQLを考えたりDBをconnectしたりする部分が省略できるので十分使えている。

import pandas as pd 
from llm_db import get_df_by_prompt

DB_PATH = "stock_data.db"

2.1 Close値の平均値をdfで返却させる

df = get_df_by_prompt(
    "hist_tblのDateが2026年2月1日以降のTickerが'7741.T'のCloseの平均値を表示してください。", 
    DB_PATH)
display(df)

2.2 指定期間のClose値と会社名をdfで返却させる

df = get_df_by_prompt(
    "hist_tblのDateが2026年2月1日以降のTickerが'7741.T'のDateとCloseを取得し、'info_tbl'の'shortName'とつなげて表示してください。", 
    DB_PATH)
display(df)

以上