STORM実践ガイド - スタンフォード大学発LLM知識キュレーションシステム

# config.py
import os
from dotenv import load_dotenv

load_dotenv()

class StormConfig:
    # LLM設定
    LLM_PROVIDER = "openai"  # "openai", "anthropic", "local"
    LLM_MODEL = "gpt-4"
    LLM_TEMPERATURE = 0.7
    
    # 検索エンジン設定
    SEARCH_ENGINE = "google"  # "google", "bing", "duckduckgo"
    SEARCH_API_KEY = os.getenv("SEARCH_API_KEY")
    MAX_SEARCH_RESULTS = 10
    
    # STORM設定
    MAX_PERSPECTIVES = 5  # 生成するペルソナ数
    MAX_QUESTIONS_PER_PERSPECTIVE = 3
    MAX_OUTLINE_DEPTH = 3
    MIN_SECTION_LENGTH = 200
    MAX_SECTION_LENGTH = 1000
    
    # 出力設定
    OUTPUT_FORMAT = "markdown"  # "markdown", "html", "json"
    INCLUDE_CITATIONS = True
    INCLUDE_IMAGES = True
    
    # API Keys
    OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
    ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY")

# api_setup.py
from knowledge_storm import Storm
from knowledge_storm.llm import OpenAIModel, AnthropicModel
from knowledge_storm.search import GoogleSearch, BingSearch

def setup_storm_api(config: StormConfig):
    """STORM APIの初期設定"""
    
    # LLMモデルの設定
    if config.LLM_PROVIDER == "openai":
        llm = OpenAIModel(
            api_key=config.OPENAI_API_KEY,
            model=config.LLM_MODEL,
            temperature=config.LLM_TEMPERATURE
        )
    elif config.LLM_PROVIDER == "anthropic":
        llm = AnthropicModel(
            api_key=config.ANTHROPIC_API_KEY,
            model="claude-3-opus",
            temperature=config.LLM_TEMPERATURE
        )
    
    # 検索エンジンの設定
    if config.SEARCH_ENGINE == "google":
        search = GoogleSearch(
            api_key=config.SEARCH_API_KEY,
            max_results=config.MAX_SEARCH_RESULTS
        )
    elif config.SEARCH_ENGINE == "bing":
        search = BingSearch(
            api_key=config.SEARCH_API_KEY,
            max_results=config.MAX_SEARCH_RESULTS
        )
    
    # STORMインスタンスの作成
    storm = Storm(
        llm=llm,
        search_engine=search,
        config=config
    )
    
    return storm

# test_storm.py
import asyncio
from knowledge_storm import Storm

async def test_storm_generation():
    """STORMの基本動作テスト"""
    
    # 設定読み込み
    config = StormConfig()
    storm = setup_storm_api(config)
    
    # テストトピック
    topic = "量子コンピューティングの現状と未来"
    
    print(f"Generating article for: {topic}")
    print("-" * 50)
    
    try:
        # 記事生成
        article = await storm.generate_article(
            topic=topic,
            language="ja",
            target_length=2000
        )
        
        # 結果表示
        print(f"Title: {article.title}")
        print(f"Sections: {len(article.sections)}")
        print(f"Total words: {article.word_count}")
        print(f"Citations: {len(article.citations)}")
        
        # 最初のセクションを表示
        if article.sections:
            first_section = article.sections[0]
            print(f"\nFirst section: {first_section.title}")
            print(first_section.content[:200] + "...")
        
        return article
        
    except Exception as e:
        print(f"Error: {e}")
        return None

# テスト実行
if __name__ == "__main__":
    asyncio.run(test_storm_generation())

# 多言語対応STORM
class MultilingualSTORM(StormKnowledgeGenerator):
    def __init__(self, target_language: str = "en", *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.target_language = target_language
        self.language_config = self._get_language_config()
    
    def _get_language_config(self) -> Dict:
        """言語別の設定を取得"""
        configs = {
            "ja": {
                "writing_style": "丁寧",
                "section_markers": ["第", "節"],
                "citation_format": "『{}』",
                "date_format": "%Y年%m月%d日"
            },
            "en": {
                "writing_style": "formal",
                "section_markers": ["Section", "Chapter"],
                "citation_format": "\"{}\"",
                "date_format": "%B %d, %Y"
            },
            "es": {
                "writing_style": "formal",
                "section_markers": ["Sección", "Capítulo"],
                "citation_format": "«{}»",
                "date_format": "%d de %B de %Y"
            }
        }
        
        return configs.get(self.target_language, configs["en"])
    
    async def generate_questions(self, topic: str, 
                               perspectives: List[Perspective]) -> List[Dict]:
        """多言語での質問生成"""
        # 基本質問を生成
        questions = await super().generate_questions(topic, perspectives)
        
        # 必要に応じて翻訳
        if self.target_language != "en":
            translated_questions = []
            for q in questions:
                translated = await self._translate_text(
                    q['question'], 
                    target_lang=self.target_language
                )
                q['question'] = translated
                translated_questions.append(q)
            
            return translated_questions
        
        return questions
    
    async def write_sections(self, outline: Dict, 
                           search_results: Dict) -> List[ArticleSection]:
        """言語別のスタイルで執筆"""
        sections = await super().write_sections(outline, search_results)
        
        # 言語別の後処理
        for section in sections:
            section.content = self._apply_language_style(section.content)
        
        return sections
    
    def _apply_language_style(self, content: str) -> str:
        """言語固有のスタイルを適用"""
        if self.target_language == "ja":
            # 日本語の敬語調整
            content = content.replace("です。", "です。")
            content = content.replace("ます。", "ます。")
        
        # 引用形式の調整
        citation_format = self.language_config['citation_format']
        # 実装省略
        
        return content

# スタイルカスタマイズ
class StyleCustomizedSTORM(StormKnowledgeGenerator):
    def __init__(self, style_config: Dict, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.style = style_config
    
    async def write_sections(self, outline: Dict, 
                           search_results: Dict) -> List[ArticleSection]:
        """カスタムスタイルでセクションを執筆"""
        sections = []
        
        for section_data in outline['sections']:
            # スタイル別のプロンプト生成
            style_prompt = self._generate_style_prompt()
            
            prompt = f"""
            {style_prompt}
            
            セクションタイトル: {section_data['title']}
            
            関連情報:
            {self._get_relevant_info(section_data['title'], search_results)}
            
            上記のスタイルガイドラインに従って執筆してください。
            """
            
            content = await self.llm.generate(prompt)
            
            sections.append(ArticleSection(
                title=section_data['title'],
                content=self._apply_style_formatting(content),
                subsections=[],
                citations=self._extract_citations(content),
                level=1
            ))
        
        return sections
    
    def _generate_style_prompt(self) -> str:
        """スタイルプロンプトを生成"""
        prompts = {
            "academic": """
            学術的なスタイルで執筆してください：
            - 客観的で中立的な表現
            - 専門用語の適切な使用と定義
            - 論理的な構成と根拠の明示
            - 受動態の使用を適切に
            """,
            "journalistic": """
            ジャーナリスティックなスタイルで執筆してください：
            - 読者を引き込む導入
            - 具体例とストーリーの活用
            - 簡潔で明快な文章
            - 5W1Hを明確に
            """,
            "technical": """
            技術文書のスタイルで執筆してください：
            - 正確で曖昧さのない表現
            - 手順の明確な説明
            - コード例やダイアグラムの参照
            - 前提条件と結果の明示
            """,
            "educational": """
            教育的なスタイルで執筆してください：
            - 段階的な説明
            - 例示と練習問題
            - 重要ポイントの強調
            - 理解度確認の質問
            """
        }
        
        return prompts.get(self.style.get('type', 'academic'), 
                          prompts['academic'])
    
    def _apply_style_formatting(self, content: str) -> str:
        """スタイル別のフォーマッティング"""
        if self.style.get('type') == 'academic':
            # 脚注形式の引用に変換
            content = self._convert_to_footnotes(content)
        elif self.style.get('type') == 'journalistic':
            # プルクォートの追加
            content = self._add_pull_quotes(content)
        elif self.style.get('type') == 'technical':
            # コードブロックのハイライト
            content = self._highlight_code_blocks(content)
        
        return content

# ソース制限とフィルタリング
class FilteredSTORM(StormKnowledgeGenerator):
    def __init__(self, source_filters: Dict, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.filters = source_filters
    
    async def collect_information(self, questions: List[Dict]) -> Dict:
        """フィルタリングされた情報収集"""
        search_results = {}
        
        for q_data in questions:
            # ソースフィルターを適用
            filtered_results = await self._filtered_search(q_data)
            
            # 信頼性スコアの計算
            scored_results = self._score_sources(filtered_results)
            
            # 閾値以上のソースのみ使用
            reliable_results = [
                r for r in scored_results 
                if r['reliability_score'] >= self.filters.get('min_reliability', 0.7)
            ]
            
            if reliable_results:
                answer = await self._generate_answer_from_sources(
                    q_data['question'], 
                    reliable_results
                )
                
                search_results[q_data['question']] = {
                    'answer': answer,
                    'sources': reliable_results[:3],
                    'perspective': q_data['perspective']
                }
        
        return search_results
    
    async def _filtered_search(self, question_data: Dict) -> List[Dict]:
        """フィルタリングされた検索を実行"""
        # 基本検索
        results = await self.search.search(question_data['question'])
        
        # URLフィルタリング
        if 'allowed_domains' in self.filters:
            results = [
                r for r in results 
                if any(domain in r['url'] for domain in self.filters['allowed_domains'])
            ]
        
        if 'blocked_domains' in self.filters:
            results = [
                r for r in results 
                if not any(domain in r['url'] for domain in self.filters['blocked_domains'])
            ]
        
        # 日付フィルタリング
        if 'max_age_days' in self.filters:
            cutoff_date = datetime.now() - timedelta(days=self.filters['max_age_days'])
            results = [
                r for r in results 
                if self._parse_date(r.get('date', '')) >= cutoff_date
            ]
        
        return results
    
    def _score_sources(self, sources: List[Dict]) -> List[Dict]:
        """ソースの信頼性スコアを計算"""
        for source in sources:
            score = 0.5  # ベーススコア
            
            # ドメイン権威性
            authority_domains = self.filters.get('authority_domains', {})
            domain = self._extract_domain(source['url'])
            if domain in authority_domains:
                score += authority_domains[domain]
            
            # HTTPS使用
            if source['url'].startswith('https://'):
                score += 0.1
            
            # 更新日の新しさ
            if 'date' in source:
                age_days = (datetime.now() - self._parse_date(source['date'])).days
                if age_days < 30:
                    score += 0.2
                elif age_days < 365:
                    score += 0.1
            
            # 著者情報の有無
            if source.get('author'):
                score += 0.1
            
            source['reliability_score'] = min(score, 1.0)
        
        return sorted(sources, key=lambda x: x['reliability_score'], reverse=True)