LoginSignup
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@cherrygnb12120404(ソラマメ 王子12世)

「Python × Wikipedia × ローカルLLM（Gemma）で要約付き相関図を描く」

Posted at

※ご注意：Gemmaによる要約処理は現在開発中です。ネットワーク図生成までは動作確認済み。

Wikipediaの単語から関連語を自動取得して、AIで要約し、ネットワーク図にしてみた

🧠 この記事でやること

Wikipediaの任意の単語を指定すると、

  1. そのページに出てくる リンク付きの単語を抽出
  2. 各リンク先のページ本文を ローカルLLM（例：Gemma）で要約
  3. 関連単語同士を ネットワーク図として可視化

実装が完了しているもの (チェックが入っているものが実装完了したものです)

  • そのページに出てくる リンク付きの単語を抽出
  • 各リンク先のページ本文を ローカルLLM（例：Gemma）で要約
  • 関連単語同士を ネットワーク図として可視化

🔧 使用技術

  • httpx（非同期Webリクエスト）
  • asyncio（非同期制御）
  • BeautifulSoup（HTMLパース）
  • networkx + matplotlib（ネットワーク図描画）
  • Ollama + gemma:12b（ローカルLLMによる要約生成）

🚀 実装編はこちら ↓

全体のプログラム像が下記です。
それぞれ、個別に何をしているのか説明します。

import httpx
from bs4 import BeautifulSoup
import asyncio
from urllib.parse import urljoin
import json
from collections import Counter
import argparse
from urllib.parse import urljoin, quote
import networkx as nx
from networkx.drawing.nx_agraph import graphviz_layout
import matplotlib.pyplot as plt
from matplotlib import font_manager as fm
import matplotlib
import numpy as np


from logging import getLogger, Formatter, StreamHandler, FileHandler, DEBUG, INFO




class AsyncWikiWebCrawler:
    def __init__(self):
        self.base_url = 'https://ja.wikipedia.org/wiki/'
        self.client = None
        self.seen = set()
        self.nx = nx.Graph()
        self.ollama_api_url = "http://localhost:11434/api/generate"
        self.edges = []
        self.summaries = {}

    async def query_ollama(self, prompt : str):
        prompt = f'''
                    あなたは優秀なテキストの内容を要約してくれる要約者です。下記記事の内容を必ず"50文字以内"で要約しなさい
                    {prompt}
                    '''
        response = await self.client.post(
            self.ollama_api_url,
            json = {
                'model': 'gemma:12b',
                'prompt': prompt,
                'stream': False
            }
        )
        if response.status_code == 200:
            data = response.json()
            summary = data.get("response", "").strip()
            print(f'Response : {summary}')
            return summary
        else:
            print("Error:", response.status_code, response.text)
            return None

    async def get_wiki_connection_words(self, word, parent_word=None):
        counter = Counter()
        links = []
        deduplicated = []
        page_seen = set()
        
        encoded_word = quote(word).replace('%20', '_')
        print(f'{self.base_url}{encoded_word}')
        
        response = await self.client.get(f'{self.base_url}{encoded_word}')
        html = response.text

        if word not in self.summaries:
            summary = await self.query_ollama(html[:1000])
            self.summaries[word] = summary
        else:
            summary = self.summaries[word]

        soup = BeautifulSoup(html, 'html.parser')
        content = soup.find('div', id='mw-content-text')

        # td trを含めると、他の画面上に表示されていないものまで取得されてしまう問題がある。
        for parent in content.find_all(['dd', 'p', 'td', 'tr']):
            for a in parent.find_all('a', href=True, title=True):
                full_url = urljoin(self.base_url, a['href'])
                links.append({'title': a['title'], 'url': full_url})
 
        for link in links:
            link_str = json.dumps(link, sort_keys=True)
            if link_str not in page_seen:
                page_seen.add(link_str)
                deduplicated.append(link)

        for data in deduplicated:
            link_str = json.dumps(data, sort_keys=True)
            counter[link_str] +=1

        for link_str, count in counter.items():
            if count > 1:
                print(f'重複データ : {json.loads(link_str)} -> {count}回')

        all_unique = True
        for count in counter.values():
            if count > 1:
                all_unique = False
                break

        if all_unique:
            print('最終的なデータに重複はありません')
            print(deduplicated)
        else:
            print('最終的なデータに重複があります')
            print(deduplicated)

        for link in deduplicated:
            if parent_word:
                title = link['title']
                summary = self.summaries.get(title)
                label = f"{title}\n{summary}" if summary else title
                self.edges.append((parent_word, label))

        return deduplicated
    
    def draw_graph(self):

        jpn_fonts=list(np.sort([ttf for ttf in fm.findSystemFonts() if 'ipaexg' in ttf or 'msgothic' in ttf or 'japan' in ttf or 'ipafont' in ttf]))
        jpn_font=jpn_fonts[0]
        prop = fm.FontProperties(fname=jpn_font)
        print(jpn_font)

        print("相関図を描画中...")
        self.nx.add_edges_from(self.edges)

        plt.figure(figsize=(24, 18))
        degrees = dict(self.nx.degree())
        node_sizes = [degrees[n] * 100 for n in self.nx.nodes()]
        pos = nx.kamada_kawai_layout(self.nx)
        nx.draw_networkx_nodes(self.nx, pos, node_size=node_sizes)
        nx.draw_networkx_edges(self.nx, pos)
        for node, (x, y) in pos.items():
            plt.text(x, y, node, fontsize=9, fontproperties=prop, ha='center', va='center')
        plt.rcParams['font.family'] = prop.get_name()
        plt.title("Wikipedia 関連単語ネットワーク", fontproperties=prop)
        plt.axis('off')
        # plt.show()
        plt.savefig("graph.png")

        return
    

    async def __aenter__(self):
        self.client = httpx.AsyncClient()
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.client.aclose()
        return
    
    async def main(self, word, frequency):
        to_crawl = [word]
        crawled = set()
        count = 0
        
        while count < frequency and to_crawl:
            current_batch = []
            while to_crawl:
                current_word = to_crawl.pop(0)
                if current_word in crawled:
                    continue
                crawled.add(current_word)
                current_batch.append(current_word)
            if not current_batch:
                break

            print(f'現在クロール中: {current_batch}')

            tasks = []
            for w in current_batch:
                task = self.get_wiki_connection_words(w, parent_word=w)
                tasks.append(task)
            
            results = await asyncio.gather(*tasks, return_exceptions=True)

            for links in results:
                if isinstance(links, Exception):
                    print(f'エラー発生 : {links}')
                    continue

                for link in links:
                    title = link['title']

                    if ':' in title:
                        print(f'スキップ : {title} (特殊ページ)')
                        continue

                    if title not in crawled:
                        to_crawl.append(title)
            count += 1
            with open("summaries.json", "w", encoding="utf-8") as f:
                json.dump(self.summaries, f, ensure_ascii=False, indent=2)
            print("要約結果を書き出しました → summaries.json")

        self.draw_graph()

        return
    


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="非同期Wikipedia単語クローラー")
    parser.add_argument('--word', type=str, default='プログラミング', help='検索するWikipediaの単語（デフォルト: プログラミング）')
    parser.add_argument('--frequency', type=int, default=1, help='再帰処理を実施する回数')

    args = parser.parse_args()

    async def run():
        async with AsyncWikiWebCrawler() as crawler:
            await crawler.main(args.word, args.frequency)

    asyncio.run(run())

1️⃣ Wikipediaから関連語を抜き出す処理

この部分では、指定した単語のWikipediaページから、リンクされている単語（＝aタグ）を抽出します。

    async def get_wiki_connection_words(self, word, parent_word=None):
        counter = Counter()
        links = []
        deduplicated = []
        page_seen = set()
        
        encoded_word = quote(word).replace('%20', '_')
        print(f'{self.base_url}{encoded_word}')
        
        response = await self.client.get(f'{self.base_url}{encoded_word}')
        html = response.text

        if word not in self.summaries:
            summary = await self.query_ollama(html[:1000])
            self.summaries[word] = summary
        else:
            summary = self.summaries[word]

        soup = BeautifulSoup(html, 'html.parser')
        content = soup.find('div', id='mw-content-text')

        # td trを含めると、他の画面上に表示されていないものまで取得されてしまう問題がある。
        for parent in content.find_all(['dd', 'p', 'td', 'tr']):
            for a in parent.find_all('a', href=True, title=True):
                full_url = urljoin(self.base_url, a['href'])
                links.append({'title': a['title'], 'url': full_url})
 
        for link in links:
            link_str = json.dumps(link, sort_keys=True)
            if link_str not in page_seen:
                page_seen.add(link_str)
                deduplicated.append(link)

        for data in deduplicated:
            link_str = json.dumps(data, sort_keys=True)
            counter[link_str] +=1

        for link_str, count in counter.items():
            if count > 1:
                print(f'重複データ : {json.loads(link_str)} -> {count}回')

        all_unique = True
        for count in counter.values():
            if count > 1:
                all_unique = False
                break

        if all_unique:
            print('最終的なデータに重複はありません')
            print(deduplicated)
        else:
            print('最終的なデータに重複があります')
            print(deduplicated)

        for link in deduplicated:
            if parent_word:
                title = link['title']
                summary = self.summaries.get(title)
                label = f"{title}\n{summary}" if summary else title
                self.edges.append((parent_word, label))

        return deduplicated

工夫した点

  • Wikipediaのページを正しくリクエストするために、スペース文字をアンダーバーに置き換える必要があります。
encoded_word = quote(word).replace('%20', '_')
  • Wikipediaのリンクの中には、重複したurlが複数存在しているパターンがあった為、重複を排除する必要があります。
        for link in links:
            link_str = json.dumps(link, sort_keys=True)
            if link_str not in page_seen:
                page_seen.add(link_str)
                deduplicated.append(link)

        for data in deduplicated:
            link_str = json.dumps(data, sort_keys=True)
            counter[link_str] +=1

        for link_str, count in counter.items():
            if count > 1:
                print(f'重複データ : {json.loads(link_str)} -> {count}回')

2️⃣wikipediaの中身をgemma3に要約させる処理

    async def query_ollama(self, prompt : str):
        prompt = f'''
                    あなたは優秀なテキストの内容を要約してくれる要約者です。下記記事の内容を必ず"50文字以内"で要約しなさい
                    {prompt}
                    '''
        response = await self.client.post(
            self.ollama_api_url,
            json = {
                'model': 'gemma:12b',
                'prompt': prompt,
                'stream': False
            }
        )
        if response.status_code == 200:
            data = response.json()
            summary = data.get("response", "").strip()
            print(f'Response : {summary}')
            return summary
        else:
            print("Error:", response.status_code, response.text)
            return None

工夫した点

  • どんな形で要約されたかをテキストで確認をしたかった為、jsonでdumpする処理を追加しました。
        if response.status_code == 200:
            data = response.json()
            summary = data.get("response", "").strip()
            print(f'Response : {summary}')
            return summary

3️⃣全体処理 (非同期処理でWikipediaページを並列取得)

工夫した点

  • Wikipediaの複数ページを順番に取得していたら、処理が終わらないので、
    Pythonの asyncio + httpx.AsyncClient を使って非同期化しています。
async with httpx.AsyncClient() as client:
    response = await client.get(url)
  • 非同期対応の httpx.AsyncClient() を aenter / aexit で管理することで、安全なセッション制御を行っています。
    async def __aenter__(self):
        self.client = httpx.AsyncClient()
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.client.aclose()
        return

4️⃣グラフ描画処理（networkx + matplotlib）

工夫した点

  • ノード間の距離や配置は kamada_kawai_layout を使って、自然な見た目になるように調整しています。

  • 日本語が文字化けしないように、日本語フォントの自動探索と設定を行っています

       jpn_fonts=list(np.sort([ttf for ttf in fm.findSystemFonts() if 'ipaexg' in ttf or 'msgothic' in ttf or 'japan' in ttf or 'ipafont' in ttf]))
        jpn_font=jpn_fonts[0]
        prop = fm.FontProperties(fname=jpn_font)

問題点

  • gemma3による要約が、それぞれのリンクに対して実行されない
  • 取得したデータ量が多すぎて、完成した画像がよくわからない原子みたいな画像になる
  • LLMの要約がまだ不安定

graph.png

1
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?