executing-plans
executing-plans
Use when executing an implementation plan in the current session as the implementer yourself — your human partner chose inline execution, or no subagent tool is available
ds-topic-modeling"Implements topic modeling using Latent Dirichlet Allocation (LDA) Non-negative Matrix Factorization (NMF), and other topic extraction methods"
Independent GitHub snapshot · Automated classification · Original documentation governs
name: ds-topic-modeling compatibility: opencode completeness: 95 content-types:
Comprehensive guide to topic modeling in machine learning and data science workflows.
Topic Modeling is a critical component of the machine learning workflow. This skill covers:
import pandas as pd
import numpy as np
from sklearn.feature_extraction.text import CountVectorizer
from sklearn.decomposition import LatentDirichletAllocation
from typing import List, Dict, Any
def basic_topic_modeling(texts: List[str], n_topics: int = 5) -> Dict[str, Any]:
"""
Perform basic topic modeling using Latent Dirichlet Allocation (LDA).
Args:
texts: List of raw text documents
n_topics: Number of topics to extract
Returns:
Dictionary containing fitted model, vectorizer, and topic-word distributions
"""
if not texts:
raise ValueError("Input texts list cannot be empty")
# Vectorize text documents
vectorizer = CountVectorizer(max_df=0.95, min_df=2, stop_words='english')
doc_term_matrix = vectorizer.fit_transform(texts)
# Initialize and fit LDA model
lda_model = LatentDirichletAllocation(
n_components=n_topics
max_iter=10
learning_method='online'
random_state=42
)
lda_model.fit(doc_term_matrix)
# Extract top words per topic
feature_names = vectorizer.get_feature_names_out()
topics = {}
for idx, topic in enumerate(lda_model.components_):
top_words_idx = topic.argsort()[:-10:-1]
topics[f'topic_{idx}'] = [feature_names[i] for i in top_words_idx]
return {
'model': lda_model
'vectorizer': vectorizer
'topics': topics
'doc_term_matrix': doc_term_matrix
}
import logging
import pandas as pd
import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.decomposition import NMF
from sklearn.pipeline import Pipeline
from typing import Any, Dict, List, Optional
import warnings
warnings.filterwarnings('ignore')
logger = logging.getLogger(__name__)
class TopicModeling:
"""Production-grade implementation of Topic Modeling using NMF"""
def __init__(self, n_topics: int = 5, max_features: int = 1000, random_state: int = 42):
self.n_topics = n_topics
self.max_features = max_features
self.random_state = random_state
self.pipeline: Optional[Pipeline] = None
self.feature_names: List[str] = []
def _validate_input(self, data: pd.DataFrame, text_column: str) -> None:
if text_column not in data.columns:
raise ValueError(f"Column '{text_column}' not found in DataFrame")
if data[text_column].isnull().any():
logger.warning("Dropping rows with missing text data")
data = data.dropna(subset=[text_column])
if len(data) < self.n_topics:
raise ValueError("Dataset size must be greater than n_topics")
def _build_pipeline(self) -> Pipeline:
vectorizer = TfidfVectorizer(
max_features=self.max_features
stop_words='english'
ngram_range=(1, 2)
)
nmf_model = NMF(
n_components=self.n_topics
init='nndsvd'
random_state=self.random_state
max_iter=200
)
return Pipeline([('tfidf', vectorizer), ('nmf', nmf_model)])
def execute(self, data: pd.DataFrame, text_column: str = 'text') -> Dict[str, Any]:
"""Execute Topic Modeling on data"""
self._validate_input(data, text_column)
self.pipeline = self._build_pipeline()
tfidf_matrix = self.pipeline.named_steps['tfidf'].fit_transform(data[text_column])
self.pipeline.fit(tfidf_matrix)
feature_names = self.pipeline.named_steps['tfidf'].get_feature_names_out()
topic_words = {}
for i, topic in enumerate(self.pipeline.named_steps['nmf'].components_):
top_indices = topic.argsort()[:-10:-1]
topic_words[f'topic_{i}'] = [feature_names[idx] for idx in top_indices]
reconstructed = self.pipeline.named_steps['nmf'].transform(tfidf_matrix) @ \
self.pipeline.named_steps['nmf'].components_
inertia = float(np.linalg.norm(tfidf_matrix.toarray() - reconstructed))
return {
'topics': topic_words
'inertia': inertia
'pipeline': self.pipeline
'document_topics': self.pipeline.named_steps['nmf'].transform(tfidf_matrix)
}
| Pitfall | Problem | Solution | |
Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content.