deploy multi-topic parallel langgraph pipeline

app.py

@@ -0,0 +1,163 @@
+import os
+import json
+import re
+import gradio as gr
+from typing import List, TypedDict, Annotated
+from pydantic import BaseModel, Field
+from langgraph.graph import StateGraph, START, END
+from langchain_core.prompts import ChatPromptTemplate
+from langchain_openai import ChatOpenAI
+
+# ==========================================================
+# 1. LOAD DATA & BUILD LOOKUP INDICES
+# ==========================================================
+# Hugging Face will look for this file in the root directory
+with open("gcmd_hierarchy.json", "r") as f:
+    gcmd_data = json.load(f)
+
+def build_gcmd_indices(gcmd_json):
+    topic_list = []
+    sub_tree_indices = {}
+    topics = gcmd_json.get("children", [])
+    
+    for topic_node in topics:
+        topic_name = topic_node.get("name", "").upper()
+        topic_list.append(topic_name)
+        collected_paths = []
+        
+        def recurse_sub_tree(node, current_path=""):
+            node_name = node.get("name", "")
+            node_path = f"{current_path} > {node_name}" if current_path else node_name
+            if "Variable" in node.get("level", "") or not node.get("children"):
+                definition = node.get("definition", "")
+                if definition:
+                    collected_paths.append(f"{node_path} (Definition: {definition})")
+                else:
+                    collected_paths.append(node_path)
+            for child in node.get("children", []):
+                recurse_sub_tree(child, node_path)
+
+        for term_node in topic_node.get("children", []):
+            recurse_sub_tree(term_node, current_path="")
+            
+        sub_tree_indices[topic_name] = "\n".join([f"- {path}" for path in collected_paths])
+    return topic_list, sub_tree_indices
+
+VALID_TOPICS, SUB_TREE_LOOKUP = build_gcmd_indices(gcmd_data)
+
+# ==========================================================
+# 2. LANGGRAPH MULTI-TOPIC DEFINITION
+# ==========================================================
+def merge_lists(left: list, right: list) -> list:
+    return list(set((left or []) + (right or [])))
+
+class MultiTopicState(TypedDict):
+    title: str
+    abstract: str
+    chosen_topics: List[str]
+    predicted_keywords: Annotated[List[str], merge_lists]
+    invalid_keywords: Annotated[List[str], merge_lists]
+
+class TopicsChoice(BaseModel):
+    topics: List[str] = Field(description="List of matching topic areas.")
+
+def route_multi_topic(state: MultiTopicState):
+    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) # gpt-4o-mini keeps costs low for public demos
+    structured_llm = llm.with_structured_output(TopicsChoice)
+    prompt = ChatPromptTemplate.from_messages([
+        ("system", f"You are an expert science cataloger. Identify ALL relevant major topic areas. Choose ONLY from: {', '.join(VALID_TOPICS)}"),
+        ("user", "Title: {title}\nAbstract: {abstract}\n\nSelect all relevant Topics.")
+    ])
+    result = structured_llm.invoke(prompt.format(title=state["title"], abstract=state["abstract"]))
+    valid_selected = [t.upper().strip() for t in result.topics if t.upper().strip() in VALID_TOPICS]
+    if not valid_selected:
+        valid_selected = ["FALLBACK"]
+    return {"chosen_topics": valid_selected}
+
+def classify_individual_topic(topic_name: str):
+    def node_runner(state: MultiTopicState):
+        llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
+        target_sub_tree = SUB_TREE_LOOKUP.get(topic_name, "")
+        prompt = ChatPromptTemplate.from_messages([
+            ("system", f"You are a specialist in {topic_name} data mapping. Extract exact keyword pathways present in the list."),
+            ("user", "Title: {title}\nAbstract: {abstract}\n\nValid Paths:\n{sub_tree}\n\nReturn exact matching entries as a comma-separated list.")
+        ])
+        response = llm.invoke(prompt.format(title=state["title"], abstract=state["abstract"], sub_tree=target_sub_tree))
+        raw_keywords = [k.strip() for k in response.content.split(",") if k.strip()]
+        
+        valid_set = set()
+        for line in target_sub_tree.split("\n"):
+            if line.strip():
+                path = line.replace("- ", "").strip()
+                path = re.sub(r"\s*\(Definition:.*?\)$", "", path).strip()
+                valid_set.add(path)
+                
+        valid_kws = [kw for kw in raw_keywords if kw in valid_set]
+        invalid_kws = [kw for kw in raw_keywords if kw not in valid_set]
+        return {"predicted_keywords": valid_kws, "invalid_keywords": invalid_kws}
+    return node_runner
+
+def parallel_router(state: MultiTopicState) -> List[str]:
+    return [f"classify_{topic.lower()}" for topic in state["chosen_topics"]]
+
+# Compile the Workflow
+workflow = StateGraph(MultiTopicState)
+workflow.add_node("top_router", route_multi_topic)
+for topic in VALID_TOPICS:
+    workflow.add_node(f"classify_{topic.lower()}", classify_individual_topic(topic))
+workflow.add_node("classify_fallback", lambda state: {"predicted_keywords": []})
+
+workflow.add_edge(START, "top_router")
+workflow.add_conditional_edges(
+    "top_router",
+    parallel_router,
+    {f"classify_{t.lower()}": f"classify_{t.lower()}" for t in VALID_TOPICS} | {"classify_fallback": "classify_fallback"}
+)
+for topic in VALID_TOPICS:
+    workflow.add_edge(f"classify_{topic.lower()}", END)
+workflow.add_edge("classify_fallback", END)
+
+app = workflow.compile()
+
+# ==========================================================
+# 3. GRADIO USER INTERFACE FOR DEMONSTRATION
+# ==========================================================
+def run_agent_classifier(title, abstract):
+    if not title or not abstract:
+        return "Please fill out both Title and Abstract fields.", "N/A"
+        
+    inputs = {"title": title, "abstract": abstract}
+    output = app.invoke(inputs)
+    
+    topics_str = ", ".join(output.get("chosen_topics", []))
+    keywords_list = output.get("predicted_keywords", [])
+    
+    if not keywords_list:
+        keywords_str = "No explicit keywords mapped."
+    else:
+        keywords_str = "\n".join([f"• {kw}" for kw in keywords_list])
+        
+    return topics_str, keywords_str
+
+# Build out layout blocks
+demo = gr.Interface(
+    fn=run_agent_classifier,
+    inputs=[
+        gr.Textbox(label="Journal Article Title", placeholder="Enter article title here...", lines=1),
+        gr.Textbox(label="Abstract / Body Text", placeholder="Paste abstract description here...", lines=5)
+    ],
+    outputs=[
+        gr.Textbox(label="Routed Multi-Topic Domains"),
+        gr.Textbox(label="Verified GCMD Keywords Extracted")
+    ],
+    title="GCMD Science Keyword Classifier Agent",
+    description="Proof of Concept using LangGraph and LangChain. Routes articles concurrently across science domains and runs isolated self-validation routines.",
+    examples=[
+        ["El Niño Southern Oscillation Shifting Evaporation Tracks", "Rising sea surface temperatures across equatorial waters directly trigger increased low-level cloud formation and accelerated surface winds."]
+    ]
+)
+
+# Launch configuration compatible with Hugging Face Space runtime rules
+if __name__ == "__main__":
+    demo.launch()
+

requirements.txt

@@ -0,0 +1,6 @@
+langchain-core
+langchain-openai
+langgraph
+pydantic
+gradio
+