概要

　Anthropic 社の生成 AI サービスである Claude から Projects という機能が新しく公開されました。今回は、この Projects に特定のライブラリのリファレンスをコンテキストとして入れ、そのリファレンスをもとにこちらの質問に回答できるのかを試してみます。

モチベーション

　多くの IT エンジニアがコーディング作業で ChatGPT や Claude , Gemini などの生成 AI を活用しているかと思いますが、新しくリリースされたライブラリやリポジトリを利用する場合、生成 AI に質問をしても適切に回答してくれないことが多いです。また、活発に開発が行われているライブラリの場合、生成 AI に質問をしても回答内容が最新のものではないということも多々起こります。このような場合に、新機能である Projects にライブラリのリファレンスの情報を与え、質問するようにすることで適切にコーディング作業をサポートしてくれるようになるのでは？と考え、これを試してみることにしました。

Projects について

　Projectsは、Claude.aiのPro契約者($20/月)およびTeam契約者向けの新機能で、200K のコンテキストファイル（約500ページ分）を、作成したプロジェクトにアップロードし、そのコンテキストに対して指示・質問できます。これまので Claude のサービス上においても上限 200K のコンテキストをコピペして指示・質問することはできましたが、チャットの都度コンテキストをコピペする必要がありました。 Projects では個別に作成したプロジェクトにコンテキストファイルを一度アップロードすれば以降は質問・指示するだけですみます。

作業の流れ

指定したURLとそのURLがリンクしている同一ドメインのURL内のコンテンツを再帰的にクローリングし、ひとつのファイルにまとめる。
1 で作成したファイルを　Claude の Projects にアップロードする
Claude に質問をする

今回は Ibis という Python のデータ処理ライブラリのリファレンスで実験してみます。参考ですが Ibis は pandas や Polars , DuckDB , PySpark などのあらゆるデータ集計ライブラリを同一の Ibis API で操作することができる統合データ処理ライブラリです。新しいデータ処理ライブラリが登場するたびに発生する学習コストを低減できる等、メリットの多いライブラリです。もし興味があれば以下の記事もぜひご覧下さい。

クローリングコードについて

　指定したURLとそのURLがリンクしている同一ドメインのURL内のコンテンツを再帰的にクローリングし、ひとつのファイルにまとめるコードを作成しました。以下の Google Colab のコードリンクを貼っておくのでこちらからすぐに実行することができます。

ライブラリのインストール

!pip install -qq requests beautifulsoup4 trafilatura

コード

# @markdown 処理内容：指定したURLとそのURLとリンクする同一ドメインのURLを再帰的にクローリングする
import requests
from bs4 import BeautifulSoup
import trafilatura
import time
from urllib.parse import urljoin, urlparse, urldefrag
import csv

# 初期URLリスト
start_urls = "https://ibis-project.org/reference/" # @param {type:"string"}
start_urls = start_urls.split(',')
# @markdown - クローリング対象URL（複数の場合は , で区切る）
max_crawl_count = 100  # @param {type:"integer"}
# @markdown - 最大クローリングページ数
ng_words = "" # @param {type:"string"}
if ng_words == "":
  ng_words = []
else:
  ng_words = ng_words.split(',')
# @markdown - クローリング除外ワード（複数の場合は , で区切る、除外ワードを含むURLはクローリングしない）
ok_words = "" # @param {type:"string"}
if ok_words == "":
  ok_words = []
else:
  ok_words = ok_words.split(',')
# @markdown - クローリング指定ワード（複数の場合は , で区切る、指定ワードを含むURLをクローリングする）
specific_path_only = True  # @param {type:"boolean"}
# @markdown - Trueで指定したURLの階層以降をクローリング対象とする
# @markdown   - 指定した URL が https://xxx/yyy で True にした場合、
# @markdown    - https://xxx/yyy や https://xxx/yyy/aaa はクローリングする
# @markdown    - https://xxx や https://xxx/zzz はクローリングしない

visited_urls = set()
content_list = []

def save_content_to_file(data, filename):
    with open(filename, 'w', encoding='utf-8') as f:
        for content in data:
            f.write("------------------------------------------------------\n")
            f.write(f"<url>{content['url']}</url>\n\n")
            f.write("<content>\n")
            f.write(content['text'])
            f.write("\n</content>\n")
            f.write("\n")

def save_urls_to_csv(urls, filename):
    with open(filename, 'w', newline='', encoding='utf-8') as f:
        writer = csv.writer(f)
        writer.writerow(['URL'])
        for url in urls:
            writer.writerow([url])

def is_url_allowed(url, ng_words):
    """NGワードがURLに含まれていないか確認する関数"""
    return not any(ng_word in url for ng_word in ng_words)

def is_url_included(url, ok_words):
    """OKワードがURLに含まれていないか確認する関数"""

    if ok_words == []:
      return True
    else:
      return any(ok_word in url for ok_word in ok_words)

def is_within_specific_path(url, start_url, specific_path_only):
    """指定された階層以降のURLか確認する関数"""
    if not specific_path_only:
        return True
    start_path = urlparse(start_url).path
    return urlparse(url).path.startswith(start_path)

def crawl(url, max_crawl_count, start_url):
    if url not in visited_urls and len(visited_urls) < max_crawl_count:
        visited_urls.add(url)
        print(f"Crawling {len(visited_urls)}: {url}")
        try:
            response = requests.get(url)
            response.raise_for_status()

            soup = BeautifulSoup(response.content, 'html.parser')
            title_tag = soup.title.string if soup.title else 'No Title'
            raw_content = trafilatura.extract(response.text, output_format='markdown')
            if raw_content:
                content_list.append({
                    'url': url,
                    'title': title_tag,
                    'text': raw_content
                })

            # このページ内のリンクを取得
            links = [a['href'] for a in soup.find_all('a', href=True)]

            # さらにリンク先をクロール
            for link in links:
                if len(visited_urls) >= max_crawl_count:
                    break
                absolute_url = urljoin(url, link)
                absolute_url, _ = urldefrag(absolute_url)  # フラグメントを除去
                parsed_url = urlparse(absolute_url)

                if (parsed_url.hostname == urlparse(start_url).hostname
                        and absolute_url not in visited_urls
                        and is_url_allowed(absolute_url, ng_words)
                        and is_url_included(absolute_url, ok_words)
                        and is_within_specific_path(absolute_url, start_url, specific_path_only)):
                    # クロール先の負荷調整のため時間を調整する場合
                    time.sleep(0.5)
                    crawl(absolute_url, max_crawl_count, start_url)

        except requests.RequestException as e:
            print(f"Error crawling {url}: {e}")

for start_url in start_urls:
    crawl(start_url, max_crawl_count, start_url)

save_content_to_file(content_list, 'crawled_content.txt')
save_urls_to_csv(visited_urls, 'crawled_urls.csv')

# 総単語数を計算
total_word_count = sum(len(content['text'].split()) for content in content_list)
print(f"Crawled {len(content_list)} pages.")
print(f"Total word count: {total_word_count}")

コードの各種パラメータ

start_urls
クローリング対象URLを入力する（複数の場合は , で区切る）
max_crawl_count
最大クローリングページ数を指定する（最初は 10 とかで動作確認すると良いです）
ng_words
クローリング除外ワードを指定する（複数の場合は , で区切る、除外ワードを含むURLはクローリングしない）
ok_words
クローリング指定ワードを指定する（複数の場合は , で区切る、指定ワードを含むURLをクローリングする）
specific_path_only
指定したURLのパス階層以降をクローリング対象とする
- 指定した URL が https://xxx/yyy で True にした場合、
  - https://xxx/yyy や https://xxx/yyy/aaa はクローリングする
  - https://xxx や https://xxx/zzz はクローリングしない

クローリングの実行

　以下のURLを指定してコードを実行しました。
本当はトップドメインの https://ibis-project.org/ を指定したかったのですがこれだとコンテキスト上限の 200K をオーバーしてしまうため、reference 以下の URL のみをクローリングするようにしました。

https://ibis-project.org/reference/

クローリング結果（ print 出力結果）

Crawling 1: https://ibis-project.org/reference/
Crawling 2: https://ibis-project.org/reference/expression-tables
Crawling 3: https://ibis-project.org/reference/selectors
Crawling 4: https://ibis-project.org/reference/expression-generic
Crawling 5: https://ibis-project.org/reference/expression-numeric
Crawling 6: https://ibis-project.org/reference/expression-strings
Crawling 7: https://ibis-project.org/reference/expression-temporal
Crawling 8: https://ibis-project.org/reference/expression-collections
Crawling 9: https://ibis-project.org/reference/expression-geospatial
Crawling 10: https://ibis-project.org/reference/datatypes
Crawling 11: https://ibis-project.org/reference/schemas
Crawling 12: https://ibis-project.org/reference/scalar-udfs
Crawling 13: https://ibis-project.org/reference/connection
Crawling 14: https://ibis-project.org/reference/interactive
Crawling 15: https://ibis-project.org/reference/options
Crawling 16: https://ibis-project.org/reference/repr
Crawling 17: https://ibis-project.org/reference/sql
Crawling 18: https://ibis-project.org/reference/operations
Crawling 19: https://ibis-project.org/reference/backendtest
Crawling 20: https://ibis-project.org/reference/servicebackendtest
Crawled 20 pages.
Total word count: 56629

Google Colab 上で以下の画像のように「crawled_content.txt」というファイルが出力されるのでこれを projects にてアップロードし、質問・指示をしてみます。

質疑応答の結果

Ibis に関して以下の4つの質問をしてみました。

Prompt1:

Ibisがサポートしているバックエンドエンジンの種類をすべて教えて下さい。

Response:

Ibisは多様なバックエンドエンジンをサポートしています。以下に、Ibisがサポートしている主なバックエンドエンジンの種類を列挙します：

DuckDB - デフォルトのバックエンド、高速なin-memoryデータベース
SQLite - 軽量なファイルベースのデータベース
PostgreSQL - 強力なオープンソースリレーショナルデータベース
MySQL - 広く使用されているオープンソースリレーショナルデータベース
ClickHouse - 列指向の分析データベース
BigQuery - Googleのクラウドデータウェアハウス
Impala - Hadoopベースの高性能分析データベース
Snowflake - クラウドデータプラットフォーム
Apache Spark - 大規模データ処理エンジン
PySpark - SparkのPythonインターフェース
Datafusion - Rustで書かれた高性能データ処理フレームワーク
Polars - 高性能データ操作ライブラリ
Pandas - Pythonのデータ分析ライブラリ
Dask - 大規模データセット用の柔軟な並列計算ライブラリ
Trino (旧Presto) - 分散SQLクエリエンジン
Oracle - 商用リレーショナルデータベース管理システム
Microsoft SQL Server - Microsoftの商用リレーショナルデータベース管理システム
Exasol - 高性能なin-memoryデータベース
Risingwave - ストリーミングデータベース
Flink - 分散ストリーム処理フレームワーク
Hive - Hadoopベースのデータウェアハウス

これらのバックエンドは、それぞれ異なる特性と強みを持っています。プロジェクトの要件、データの規模、処理の種類、既存のインフラストラクチャなどに応じて適切なバックエンドを選択することが重要です。また、Ibisの開発は活発に行われているため、新しいバックエンドが追加されたり、既存のバックエンドの機能が拡張されたりする可能性があります。

Prompt2:

Ibisでバックエンドの処理エンジンにPolarsを指定する方法を教えて下さい。

Response:

Ibisでバックエンドの処理エンジンにPolarsを指定するには、いくつかのステップがあります。以下に手順を示します：

1.必要なライブラリをインストールする：

まず、ibisとpolarsをインストールする必要があります。以下のコマンドを使用してインストールできます：

pip install ibis-framework[polars]

2.Polarsバックエンドを使用してIbisに接続する：

import ibis

# Polarsバックエンドを使用して接続
con = ibis.polars.connect()

3.データを読み込む：

Polarsバックエンドを使用してデータを読み込むには、以下のようにします：

# CSVファイルを読み込む例
table = con.read_csv("path/to/your/file.csv")

# または、既存のPolars DataFrameを使用する場合
import polars as pl
df = pl.read_csv("path/to/your/file.csv")
table = con.from_dataframe(df)

4.クエリを実行する：

# クエリを定義
result = table.filter(table.column > 0).groupby('category').agg(total=table.value.sum())

# 実行
df_result = result.to_pandas()

5.デフォルトバックエンドとして設定する：

セッション中にPolarsをデフォルトのバックエンドとして使用したい場合は、以下のように設定できます：

ibis.options.default_backend = "polars"

注意点：

Polarsバックエンドは比較的新しく、すべての機能がサポートされているわけではありません。
Polarsは高速な処理を提供しますが、大規模なデータセットの場合はメモリ使用量に注意が必要です。
Polarsバックエンドを使用する際は、Polarsの特性（例：NULLの扱い方）を理解しておくことが重要です。
一部のIbis操作は、PolarsのDataFrameに直接変換されて実行される可能性があります。

Polarsバックエンドを使用することで、Ibisの表現力豊かなAPIとPolarsの高速な処理能力を組み合わせることができます。ただし、プロジェクトの要件や特性に応じて、適切なバックエンドを選択することが重要です。

Prompt3:

Ibisで、csvを読み込む方法を教えて下さい。

Response:

IbisでCSVファイルを読み込むには、ibis.read_csv()関数を使用します。基本的な使い方は以下の通りです:

import ibis

# CSVファイルを読み込む
table = ibis.read_csv("path/to/your/file.csv")

具体的な例を挙げると:

1.基本的な読み込み:

import ibis

# CSVファイルを読み込む
table = ibis.read_csv("/tmp/data.csv")

# 最初の数行を表示
print(table)

2.オプションを指定して読み込む:

table = ibis.read_csv(
    "/tmp/data.csv",
    header=True,  # CSVにヘッダーがある場合
    delimiter=",",  # 区切り文字を指定
    parse_dates=["date_column"]  # 日付列を指定
)

3.複数のCSVファイルを読み込む:

table = ibis.read_csv(["/tmp/data1.csv", "/tmp/data2.csv"])

注意点:

ibis.read_csv()は遅延評価されます。実際のデータの読み込みは、計算が必要になるまで行われません。
デフォルトでは、Ibisは内部でDuckDBを使用してCSVを読み込みます。他のバックエンドを使用する場合は、それに応じた方法で読み込む必要があります。
大規模なCSVファイルを扱う場合、Ibisは効率的に処理できますが、システムのリソースに注意が必要です。

これらの方法を使って、IbisでCSVファイルを簡単に読み込むことができます。

Prompt4:

Ibisで2つのテーブルを特定の列をキーにマージする方法を教えて下さい。

Response:

Ibisで2つのテーブルをマージ（結合）するには、主にjoinメソッドを使用します。以下にいくつかの例を示します：

1.内部結合（Inner Join）:

import ibis

# 2つのテーブルを定義
table1 = ibis.table([('id', 'int64'), ('value1', 'string')], name='table1')
table2 = ibis.table([('id', 'int64'), ('value2', 'string')], name='table2')

# 'id'列をキーにして内部結合
merged = table1.join(table2, table1.id == table2.id)

2.左外部結合（Left Outer Join）:

merged = table1.join(table2, table1.id == table2.id, how='left')

3.右外部結合（Right Outer Join）:

merged = table1.join(table2, table1.id == table2.id, how='right')

4.完全外部結合（Full Outer Join）:

merged = table1.join(table2, table1.id == table2.id, how='outer')

5.複数の列をキーにして結合:

merged = table1.join(table2, [table1.id == table2.id, table1.date == table2.date])

6.結合後に特定の列を選択:

merged = table1.join(table2, table1.id == table2.id)
result = merged.select([table1.id, table1.value1, table2.value2])

7.列名の衝突を避ける:

merged = table1.join(
    table2,
    table1.id == table2.id,
    lname='t1',  # table1の列に't1_'というプレフィックスを付ける
    rname='t2'   # table2の列に't2_'というプレフィックスを付ける
)

注意点：

joinメソッドは新しいテーブル式を返します。元のテーブルは変更されません。
結合条件は通常、等価結合（==）を使用しますが、他の条件も可能です。
大規模なテーブルを結合する場合、パフォーマンスに影響を与える可能性があります。適切なインデックスの使用や、結合前のフィルタリングなどを検討してください。
結合後のテーブルでは、同じ名前の列が存在する可能性があります。selectメソッドを使用して必要な列だけを選択するか、lnameとrnameパラメータを使用して列名の衝突を避けることができます。

これらの方法を使用して、Ibisで効果的に2つのテーブルをマージできます。実際の使用例に応じて、最適な結合方法を選択してください。

参考

Projects を使わず、通常のチャット（Claude 3.5 Sonnet）で Prompt4 の質問をしてみました。

Prompt4:（通常のチャットでの質問）

Ibisで2つのテーブルを特定の列をキーにマージする方法を教えて下さい。

Response:

Ibisで2つのテーブルを特定の列をキーにマージする方法をご説明します。これはjoinメソッドを使用して行います。以下に基本的な例を示します：

Copyimport ibis

# テーブルを取得すると仮定
table1 = ...
table2 = ...

# 'key_column'をキーとしてテーブルをマージ
merged_table = table1.join(
    table2,
    table1.key_column == table2.key_column
)

# 結果を表示
print(merged_table)

この例では：

table1に対してjoinメソッドを呼び出します。
最初の引数としてtable2を渡します。
2番目の引数として、結合条件を指定します。ここでは両テーブルの'key_column'が一致する行を結合します。

joinのタイプ（内部結合、左外部結合など）を指定したい場合は、howパラメータを使用できます：

Copy# 左外部結合の例
merged_table = table1.join(
    table2,
    table1.key_column == table2.key_column,
    how='left'
)

複数の列をキーとして使用する場合は、条件を&で結合します：

Copymerged_table = table1.join(
    table2,
    (table1.key_column1 == table2.key_column1) &
    (table1.key_column2 == table2.key_column2)
)

感想と今後の期待

　Projects にコンテキストを入力し、質問することで正しく応答してくれることを確認できました。参考に Prompt4 については Projects でコンテキストを渡す場合、通常のチャットで質問する場合も試しました。通常のチャットでも正しく回答はしてくれましたが、コンテキストを渡して質問するほうが回答内容がより詳細に感じました。

今回は Ibis のリファレンスを題材に実験をしてみましたが、生成 AI が確実に学習していない内容を題材に質問をしたほうが違いがよく分かりやすそうです。

また、今後に期待することとしては、現状ではコンテキスト上限が 200K であり、リファレンスをすべてコンテキストとして渡す場合に上限オーバーしてしまうことがあるため、より長いコンテキストに対応できるようになると活用の幅がもっと広がりそうです。

（参考）その他の活用案

参考ですが、その他の活用案についての紹介です。今回はライブラリのリファレンスをクローリングしてみましたが以下のような用途でも使えそうです。

企業研究
就職志望者が、志望先の会社のテックブログをすべてクローリング、 Projects にコンテキストをアップロードし「この企業に入社するために必要なスキルセットを列挙して下さい」と質問してみるとか。逆に、企業側もこれまで掲載した自社テックブログの傾向分析をすることにも使えそうです。
リポジトリの内容理解
こちらの記事とコンセプトは同じですが、特定のリポジトリの内容をクローリングしてリポジトリの詳細について回答してもらう。

最後に

Claude から新しく公開された Projects という機能をライブラリのリファレンス理解に活用する例を紹介させていただきました。他にも良い活用方法があればまた検証してみたいと思います！

Claude の Projects にライブラリのリファレンスを丸ごと入れてみる

概要

モチベーション

Projects について

作業の流れ

クローリングコードについて

コードの各種パラメータ

クローリングの実行

質疑応答の結果

Prompt1:

Response:

Prompt2:

Response:

Prompt3:

Response:

Prompt4:

Response:

参考

Prompt4:（通常のチャットでの質問）

Response:

感想と今後の期待

（参考）その他の活用案

最後に