import os import gradio as gr import pandas as pd import numpy as np import re import json import plotly.graph_objects as go from huggingface_hub import InferenceClient # Load or download spaCy English model dynamically import spacy try: nlp = spacy.load("en_core_web_sm") except OSError: import spacy.cli spacy.cli.download("en_core_web_sm") nlp = spacy.load("en_core_web_sm") def load_data(file_obj): """Safely loads CSV, Excel, or TXT file into a Pandas DataFrame.""" if file_obj is None: return None, gr.update(choices=[], visible=False), "Please upload a file." file_path = file_obj.name ext = os.path.splitext(file_path)[1].lower() try: if ext == '.csv': df = pd.read_csv(file_path) elif ext in ['.xls', '.xlsx']: df = pd.read_excel(file_path) elif ext == '.txt': with open(file_path, 'r', encoding='utf-8') as f: content = f.read() df = pd.DataFrame({'text': [content]}) else: return None, gr.update(choices=[], visible=False), "Unsupported file format. Please upload .csv, .xlsx, or .txt." string_cols = [col for col in df.columns if df[col].dtype == 'object' or df[col].astype(str).str.len().mean() > 5] if not string_cols: string_cols = list(df.columns) return df, gr.update(choices=string_cols, value=string_cols[0], visible=True), f"Successfully loaded dataset with {len(df)} rows." except Exception as e: return None, gr.update(choices=[], visible=False), f"Error loading file: {str(e)}" def run_local_kg(text, min_edge_weight=1, max_nodes=25): """Local SpaCy-based co-occurrence extractor that builds a Concept Knowledge Graph.""" doc = nlp(text) # Extract entities and key noun chunks as concept nodes concepts = [] for ent in doc.ents: if ent.label_ in ["PERSON", "ORG", "GPE", "NORP", "FAC", "PRODUCT", "EVENT", "WORK_OF_ART"]: concepts.append(ent.text.strip()) for chunk in doc.noun_chunks: # Filter out pronouns and very short chunks chunk_text = chunk.text.strip().lower() if len(chunk_text.split()) <= 3 and chunk.root.pos_ != "PRON" and len(chunk_text) > 3: concepts.append(chunk.text.strip()) # Standardize concept names (capitalize first letters) concepts = [c.title() for c in concepts if len(c) > 2] # We find which concepts co-occur within the same sentence sentences = list(doc.sents) edges = {} for sent in sentences: sent_text = sent.text.title() # Find which unique concepts appear in this sentence present_concepts = list(set([c for c in concepts if c in sent_text])) # Build pairwise links for i in range(len(present_concepts)): for j in range(i+1, len(present_concepts)): c1, c2 = present_concepts[i], present_concepts[j] if c1 == c2: continue pair = tuple(sorted([c1, c2])) edges[pair] = edges.get(pair, 0) + 1 # Filter edges by minimum weight filtered_edges = {k: v for k, v in edges.items() if v >= min_edge_weight} if not filtered_edges: return pd.DataFrame(), pd.DataFrame(), None # Get top nodes based on degree node_degrees = {} for (source, target), weight in filtered_edges.items(): node_degrees[source] = node_degrees.get(source, 0) + weight node_degrees[target] = node_degrees.get(target, 0) + weight top_nodes = sorted(node_degrees.items(), key=lambda x: x[1], reverse=True)[:max_nodes] top_nodes_list = [n[0] for n in top_nodes] # Keep only edges containing top nodes final_edges = [] for (source, target), weight in filtered_edges.items(): if source in top_nodes_list and target in top_nodes_list: final_edges.append({ "Source": source, "Target": target, "Relationship": "Co-occurrence", "Weight": weight }) df_edges = pd.DataFrame(final_edges) df_nodes = pd.DataFrame([{"Node": n, "Importance (Degree)": d} for n, d in top_nodes]) # Build Plotly Network Layout (Circular Layout) fig = go.Figure() # 1. Position nodes in a circle node_positions = {} n_nodes = len(top_nodes_list) for idx, node in enumerate(top_nodes_list): angle = 2 * np.pi * idx / n_nodes x = np.cos(angle) y = np.sin(angle) node_positions[node] = (x, y) # 2. Draw edge lines edge_x = [] edge_y = [] for edge in final_edges: x0, y0 = node_positions[edge["Source"]] x1, y1 = node_positions[edge["Target"]] edge_x.extend([x0, x1, None]) edge_y.extend([y0, y1, None]) fig.add_trace(go.Scatter( x=edge_x, y=edge_y, line=dict(width=1.5, color='#334155'), hoverinfo='none', mode='lines' )) # 3. Draw nodes markers node_x = [] node_y = [] node_text = [] node_sizes = [] for node, degree in top_nodes: x, y = node_positions[node] node_x.append(x) node_y.append(y) node_text.append(f"{node} (Degree: {degree})") # scale marker size node_sizes.append(15 + degree * 3) fig.add_trace(go.Scatter( x=node_x, y=node_y, mode='markers+text', hoverinfo='text', text=top_nodes_list, textposition="top center", textfont=dict(color='#f3f4f6', size=10), hovertext=node_text, marker=dict( showscale=True, colorscale='Viridis', color=node_sizes, size=node_sizes, colorbar=dict( thickness=15, title='Concept Connectivity', xanchor='left', titleside='right' ), line_width=2 ) )) fig.update_layout( title="Interactive Concept Knowledge Graph", showlegend=False, hovermode='closest', margin=dict(b=20,l=5,r=5,t=40), xaxis=dict(showgrid=False, zeroline=False, showticklabels=False), yaxis=dict(showgrid=False, zeroline=False, showticklabels=False), template="plotly_dark", height=500 ) return df_nodes, df_edges, fig def run_neural_kg(text, hf_token, model_name, max_nodes=20): """Uses advanced generative instruction model to extract rich semantic relation triples.""" if not hf_token: raise ValueError("Hugging Face API Access Token is required for Transformers mode.") client = InferenceClient(token=hf_token) prompt = f"""[INST] Extract main concept entities and their relationships from this text. Return a clean, valid JSON list of objects with the keys "Source", "Relationship", and "Target" (limit to top {max_nodes} relationships). Do not output extra text, markdown indicators, or commentary. Text to parse: "{text}" [/INST]""" try: response = client.text_generation(prompt, model=model_name, max_new_tokens=500, temperature=0.2) # Parse JSON json_clean = re.sub(r'```json\s*|\s*```', '', response).strip() data = json.loads(json_clean) df_edges = pd.DataFrame(data) # Standardize columns df_edges.columns = ["Source", "Relationship", "Target"] df_edges["Weight"] = 1 # constant weight # Extract unique nodes nodes = list(set(df_edges["Source"].tolist() + df_edges["Target"].tolist())) df_nodes = pd.DataFrame([{"Node": n, "Importance (Degree)": 1} for n in nodes]) # Circular layout Plotly graph fig = go.Figure() node_positions = {} n_nodes = len(nodes) for idx, node in enumerate(nodes): angle = 2 * np.pi * idx / n_nodes x = np.cos(angle) y = np.sin(angle) node_positions[node] = (x, y) edge_x = [] edge_y = [] for idx, row in df_edges.iterrows(): x0, y0 = node_positions[row["Source"]] x1, y1 = node_positions[row["Target"]] edge_x.extend([x0, x1, None]) edge_y.extend([y0, y1, None]) fig.add_trace(go.Scatter( x=edge_x, y=edge_y, line=dict(width=1.5, color='#475569'), hoverinfo='none', mode='lines' )) node_x = [] node_y = [] for node in nodes: x, y = node_positions[node] node_x.append(x) node_y.append(y) fig.add_trace(go.Scatter( x=node_x, y=node_y, mode='markers+text', hoverinfo='text', text=nodes, textposition="top center", textfont=dict(color='#f3f4f6', size=10), hovertext=nodes, marker=dict( color='#818cf8', size=20, line_width=2 ) )) fig.update_layout( title="Interactive Concept Knowledge Graph", showlegend=False, hovermode='closest', margin=dict(b=20,l=5,r=5,t=40), xaxis=dict(showgrid=False, zeroline=False, showticklabels=False), yaxis=dict(showgrid=False, zeroline=False, showticklabels=False), template="plotly_dark", height=500 ) return df_nodes, df_edges, fig except Exception as e: raise RuntimeError(f"Hugging Face API or parsing error: {str(e)}") def analyze_kg(text_input, file_obj, text_col, method, hf_token, hf_model, min_weight, max_nodes): docs = [] if file_obj is not None: df, _, _ = load_data(file_obj) if df is not None and text_col in df.columns: docs = df[text_col].astype(str).fillna("").tolist() elif text_input and text_input.strip(): docs = [text_input] if not docs: return None, None, None, None, "Please enter text or upload a valid dataset first." try: if method == "Local Noun-Chunk Parser (CPU & Fast)": df_nodes, df_edges, fig = run_local_kg(docs[0], min_weight, max_nodes) else: df_nodes, df_edges, fig = run_neural_kg(docs[0], hf_token, hf_model, max_nodes) if df_nodes.empty: return None, None, None, None, "No semantic concepts were successfully extracted. Try entering longer text or lowering the 'Min Co-occurrence' filter." # Save edge CSV csv_edges = "extracted_concept_edges.csv" df_edges.to_csv(csv_edges, index=False) status_md = f"Successfully generated Concept Knowledge Graph with **{len(df_nodes)}** nodes and **{len(df_edges)}** relationships." return df_nodes, df_edges, fig, csv_edges, status_md except Exception as e: return None, None, None, None, f"Execution failed: {str(e)}" custom_css = """ body { background-color: #0b0f19; color: #f3f4f6; } .gradio-container { font-family: 'Inter', sans-serif !important; } h1, h2 { color: #6366f1 !important; } """ with gr.Blocks(theme=gr.themes.Default(primary_hue="indigo", secondary_hue="slate"), css=custom_css) as demo: df_state = gr.State() gr.HTML("""

Concept Knowledge Graph Builder

Map out networks of people, locations, events, and abstract ideas in computational humanities. Automatically extract concept connections and interact with them in a live network graph.

""") with gr.Row(): with gr.Column(scale=1): gr.Markdown("### 1. Upload Source Text") with gr.Tabs(): with gr.TabItem("Paste Raw Text"): text_input = gr.Textbox( label="Source Text", placeholder="Paste your text draft or chapter here to build a knowledge network...", lines=12 ) with gr.TabItem("Upload Dataset File"): file_input = gr.File(label="Upload (.csv, .xlsx, .txt)", file_types=[".csv", ".xlsx", ".txt"]) text_column_selector = gr.Dropdown( label="Target Text Column", choices=[], visible=False, interactive=True ) status_text = gr.Markdown("No file uploaded yet.") gr.Markdown("### 2. Configure Extraction") method_selector = gr.Radio( choices=["Local Noun-Chunk Parser (CPU & Fast)", "Transformers (AI Mode)"], value="Local Noun-Chunk Parser (CPU & Fast)", label="Extraction Parser" ) with gr.Group() as token_group: hf_token_input = gr.Textbox( label="Hugging Face API Token", placeholder="hf_...", type="password", visible=False, info="Required to extract deep semantic relation triples. Get one free at huggingface.co." ) hf_model_input = gr.Dropdown( choices=[ "Qwen/Qwen2.5-7B-Instruct", "meta-llama/Llama-3-8b-instruct" ], value="Qwen/Qwen2.5-7B-Instruct", label="Transformer Model (HF API)", visible=False ) with gr.Row(): min_weight = gr.Slider(minimum=1, maximum=10, value=1, step=1, label="Min Co-occurrence Weight") max_nodes = gr.Slider(minimum=5, maximum=40, value=20, step=1, label="Max Displayed Nodes") run_btn = gr.Button("Build Knowledge Graph", variant="primary") with gr.Column(scale=2): gr.Markdown("### 3. Concept Knowledge Graph Visualization") status_markdown = gr.Markdown("Enter text and click 'Build Knowledge Graph' to run.") with gr.Tabs(): with gr.TabItem("Interactive Graph"): chart_output = gr.Plot(label="Knowledge Graph Network") with gr.TabItem("Nodes Table (Concepts)"): nodes_table = gr.Dataframe( headers=["Node", "Importance (Degree)"], datatype=["str", "number"], interactive=False ) with gr.TabItem("Edges Table (Relationships)"): edges_table = gr.Dataframe( headers=["Source", "Target", "Relationship", "Weight"], datatype=["str", "str", "str", "number"], interactive=False ) gr.Markdown("### 4. Export") download_edges = gr.File(label="Download Concept Edges Table (CSV)") # Show/hide token field depending on model def toggle_method_fields(method): if method == "Transformers (AI Mode)": return gr.update(visible=True), gr.update(visible=True) else: return gr.update(visible=False), gr.update(visible=False) method_selector.change( fn=toggle_method_fields, inputs=method_selector, outputs=[hf_token_input, hf_model_input] ) file_input.change( fn=load_data, inputs=file_input, outputs=[df_state, text_column_selector, status_text] ) run_btn.click( fn=analyze_kg, inputs=[text_input, file_input, text_column_selector, method_selector, hf_token_input, hf_model_input, min_weight, max_nodes], outputs=[nodes_table, edges_table, chart_output, download_edges, status_markdown] ) if __name__ == "__main__": demo.launch()