Agents SDKを使った判例検索 (2)

投稿者:

OpenAI Agents SDK と Python の理解を深めるため、Chatgpt が生成した database_query.py の import と DB接続部分を解析する。

1. import

(1) 総論

まず database_query.py の冒頭、import文が何を意味しているのかを整理する。ここは処理ではなく、設計思想と安全装置が詰まった重要な部分である。

# database_query.py
from __future__ import annotations

import sqlite3
from pathlib import Path
from typing import List, Optional, Tuple

from pydantic import BaseModel, Field
from agents.agent import Agent
from agents.run import Runner
from agents.tool import function_tool

(2) 各論

以下、コードの働きを１行ずつ解析する。

from __future__ import annotations

型注釈（List[str] など）を、実行時に評価せず「文字列扱い」に寄せる設定。循環参照や前方参照が楽になり、型ヒントが書きやすくなる

import sqlite3

SQLite データベース（cases.db）を Python から操作するための標準ライブラリをインポート。LLMの推測を排し、DBにある事実だけで答えさせる（幻覚を回避する）ための土台となる。DB接続、SQL実行、結果取得に使う。

from pathlib import Path

ファイルパス操作用の Path を使えるようにする。"cases.db" の存在確認や絶対パス表示などが書きやすくなる。

from typing import List, Optional, Tuple

型ヒントで使う List（リスト）、 Optional（Noneの可能性）、Tuple（固定長の組）を使えるようにする。

from pydantic import BaseModel, Field

ツール入力を型・制約付きで定義する。

  • BaseModel：入力データの型・バリデーション（検証）をする「型つき入れ物」
  • Field：説明文、デフォルト、範囲制約（ge, le）などのメタ情報を付ける
from agents.agent import Agent

OpenAI Agents SDK の Agent クラス。エージェント本体（名前・指示・使えるツール）を定義するために使う。

from agents.run import Runner

エージェントを実行するための Runner。このコードでは Runner.run_sync(...) で同期実行する。

from agents.tool import function_tool

Python関数を「エージェントが呼び出せるツール」に変換するためのデコレータ。@function_tool を付けた関数がツール化され、LLMが呼べる選択肢になる。

2. データベース接続

(1) 総論

判例データベース（cases.db）が確実に存在することを確認したうえで、安全にSQLiteへ接続し、検索結果を列名で扱える接続オブジェクトを返すための共通処理。

# =====================
# DB 接続
# =====================

DB_PATH = Path("cases.db")


def _connect() -> sqlite3.Connection:
    if not DB_PATH.exists():
        raise FileNotFoundError(f"DB not found: {DB_PATH.resolve()}")
    conn = sqlite3.connect(str(DB_PATH))
    conn.row_factory = sqlite3.Row
    return conn

(2) 各論

以下、コードの働きを１行ずつ解析する。

DB_PATH = Path("cases.db")

DBファイルの場所を表す Path オブジェクトを作る。作業ディレクトリ直下の cases.db を指す。

def _connect() -> sqlite3.Connection:

DBに接続して sqlite3.Connection（接続オブジェクト）を返す関数を定義する。先頭の _ は「このファイル内部で使う想定（内部関数）」という慣習的な印。

if not DB_PATH.exists():

cases.db が実際に存在するか確認する。存在しないなら次の行でエラーにする。

raise FileNotFoundError(f"DB not found: {DB_PATH.resolve()}")

DBが存在しない場合に、間違って処理を続ける事故を防ぐため、即座に処理を中断し、原因が一目で分かるエラー（FileNotFoundError）を出す。raise は「ここで処理を強制終了せよ」という命令。DB_PATH.resolve()により相対パス→絶対パスに変換されたDBの場所がエラーと共に表示される。

conn = sqlite3.connect(str(DB_PATH))

SQLiteに接続する（この行で cases.db を開く）。sqlite3.connect は文字列パスを要求するので Path を str(...) に変換して渡す。

conn.row_factory = sqlite3.Row

SQLの検索結果をタプル（番号）ではなく辞書っぽく（列名で）扱えるようにする設定。これにより row["doc_id"] のように列名でデータベースにアクセスできる。DBの行を人間とLLMが読みやすい形に変換するためのスイッチ。

return conn

作成した接続オブジェクトを呼び出し元へ返す。以降の with _connect() as conn: で使われる想定。