Create app.py

app.py

@@ -0,0 +1,247 @@
+# (paste into app.py)
+import os
+import json
+from functools import lru_cache
+from typing import List, Tuple, Optional, Any, Dict
+
+import gradio as gr
+import numpy as np
+import pandas as pd
+
+from rdflib import Graph, URIRef, Literal
+from rdflib.namespace import RDFS, RDF, SKOS, DCTERMS
+
+ONTOLOGY_PATH = os.getenv("ONTOLOGY_PATH", "narratives.ttl")
+DEFAULT_SEARCH_METHOD = os.getenv("SEARCH_METHOD", "keyword")
+DEFAULT_STYLE = os.getenv("PROMPT_STYLE", "balanced")
+TOP_K_CONCEPTS = int(os.getenv("TOP_K_CONCEPTS", "8"))
+EXPANSION_DEPTH = int(os.getenv("EXPANSION_DEPTH", "1"))
+INCLUDE_RELATIONS = os.getenv("INCLUDE_RELATIONS", "true").lower() == "true"
+
+COHERE_EMBED_MODEL = os.getenv("COHERE_EMBED_MODEL", "embed-english-v3.0")
+COHERE_CHAT_MODEL = os.getenv("COHERE_CHAT_MODEL", "command-r")
+
+class OntologyEntry:
+    def __init__(self, uri: str, labels: List[str], alt_labels: List[str], description: str, types: List[str]):
+        self.uri = uri; self.labels = labels; self.alt_labels = alt_labels
+        self.description = description; self.types = types
+
+def _lit2text(lit: Any) -> Optional[str]:
+    if isinstance(lit, Literal): return str(lit)
+    if isinstance(lit, str): return lit
+    return None
+
+@lru_cache(maxsize=1)
+def _load_graph(path: str) -> Graph:
+    g = Graph()
+    if not os.path.exists(path):
+        raise FileNotFoundError(f"Ontology file not found at: {path}")
+    g.parse(path, format="turtle")
+    return g
+
+@lru_cache(maxsize=1)
+def _index_entries(path: str) -> List[OntologyEntry]:
+    g = _load_graph(path)
+    entries: Dict[str, OntologyEntry] = {}
+    for s in set(g.subjects()):
+        uri = str(s)
+        labels, alt_labels, desc = set(), set(), []
+        for _,_,o in g.triples((s, RDFS.label, None)):
+            t=_lit2text(o);  labels.add(t) if t else None
+        for _,_,o in g.triples((s, SKOS.prefLabel, None)):
+            t=_lit2text(o);  labels.add(t) if t else None
+        for _,_,o in g.triples((s, SKOS.altLabel, None)):
+            t=_lit2text(o);  alt_labels.add(t) if t else None
+        for p in [RDFS.comment, SKOS.definition, DCTERMS.description]:
+            for _,_,o in g.triples((s, p, None)):
+                t=_lit2text(o);  desc.append(t) if t else None
+        types = [str(o) for _,_,o in g.triples((s, RDF.type, None)) if isinstance(o, URIRef)]
+        if labels or alt_labels or desc:
+            entries[uri] = OntologyEntry(uri, sorted(labels), sorted(alt_labels), " ".join(desc), types)
+    return list(entries.values())
+
+def _neighbors(g: Graph, node: URIRef) -> List[URIRef]:
+    neigh = set()
+    for p in [SKOS.broader, SKOS.narrower, SKOS.related, RDFS.seeAlso]:
+        for _,_,o in g.triples((node, p, None)):
+            if isinstance(o, URIRef): neigh.add(o)
+        for s,_,_ in g.triples((None, p, node)):
+            if isinstance(s, URIRef): neigh.add(s)
+    return list(neigh)
+
+def expand_concepts(path: str, seeds: List[OntologyEntry], depth: int = 1) -> List[OntologyEntry]:
+    if depth <= 0 or not seeds: return seeds
+    g = _load_graph(path)
+    idx = {e.uri: e for e in _index_entries(path)}
+    frontier = [URIRef(s.uri) for s in seeds]
+    visited=set(frontier); collected=set([s.uri for s in seeds])
+    for _ in range(depth):
+        nxt=[]
+        for n in frontier:
+            for nb in _neighbors(g, n):
+                if nb not in visited:
+                    visited.add(nb); nxt.append(nb); collected.add(str(nb))
+        frontier=nxt
+    return [idx[u] for u in collected if u in idx]
+
+def _normalise_text(s: str) -> str:
+    return " ".join(s.lower().strip().split())
+
+def keyword_scores(prompt: str, entries: List[OntologyEntry]) -> List[Tuple[OntologyEntry, float]]:
+    p = _normalise_text(prompt); scored=[]
+    for e in entries:
+        best=0.0
+        for t in e.labels + e.alt_labels + ([e.description] if e.description else []):
+            if not t: continue
+            tnorm=_normalise_text(t)
+            overlap=sum(1 for tok in p.split() if tok in tnorm.split())
+            score=100.0*overlap/max(1,len(p.split()))
+            best=max(best,score)
+        if best>0: scored.append((e,best))
+    scored.sort(key=lambda x:x[1], reverse=True); return scored
+
+def _get_cohere_client():
+    api_key=os.getenv("COHERE_API_KEY")
+    if not api_key: raise RuntimeError("COHERE_API_KEY not set")
+    import cohere; return cohere.Client(api_key)
+
+def _normalize_cohere_chat_model(name: Optional[str]) -> str:
+    if not name: return "command-r"
+    n=name.strip().lower()
+    if n in {"r","commandr","command_r"}: return "command-r"
+    if n in {"r+","r-plus","commandr+","commandr-plus","command_r_plus"}: return "command-r-plus"
+    return name
+
+def embed_texts(texts: List[str], model: str) -> np.ndarray:
+    client=_get_cohere_client()
+    res=client.embed(texts=texts, model=model)
+    vecs=getattr(res,"embeddings",None) or (res.get("embeddings") if isinstance(res,dict) else None)
+    if vecs is None: raise RuntimeError("Unexpected response from Cohere embed()")
+    return np.array(vecs,dtype=float)
+
+def cosine_sim(a: np.ndarray, b: np.ndarray) -> np.ndarray:
+    a=a.astype(float); b=b.astype(float)
+    a/= (np.linalg.norm(a,axis=1,keepdims=True)+1e-12)
+    b/= (np.linalg.norm(b,axis=1,keepdims=True)+1e-12)
+    return a @ b.T
+
+def embedding_scores(prompt: str, entries: List[OntologyEntry], embed_model: str) -> List[Tuple[OntologyEntry, float]]:
+    labels=[]; idx=[]
+    for i,e in enumerate(entries):
+        text=" | ".join(e.labels+e.alt_labels)
+        if e.description: text += " | "+e.description[:300]
+        labels.append(text if text else e.uri); idx.append(i)
+    if not labels: return []
+    vp=embed_texts([prompt], embed_model)
+    vl=embed_texts(labels, embed_model)
+    sims=cosine_sim(vp, vl)[0]
+    scored=[(entries[i],float(s)) for i,s in zip(idx,sims)]
+    scored.sort(key=lambda x:x[1], reverse=True); return scored
+
+def make_enhanced_prompt(original_prompt: str, matches: List[Tuple[OntologyEntry, float]], style: str = "balanced") -> str:
+    if style=="minimal":
+        lines=[original_prompt.strip()]
+        if matches:
+            lines.append("\\nConsider these related concepts:")
+            for e,_ in matches[:8]:
+                label=e.labels[0] if e.labels else e.uri.rsplit("/",1)[-1]
+                lines.append(f"- {label}")
+        return "\\n".join(lines)
+    lines=[
+        "You are to answer the user succinctly and accurately.",
+        "First, consider these ontology cues to interpret the request more broadly; then answer plainly.\\n",
+        "User request:",
+        f"\"\"\"{original_prompt.strip()}\"\"\"\\n"
+    ]
+    if matches:
+        lines.append("Ontology cues possibly relevant:")
+        for e,_ in matches[:10]:
+            label=e.labels[0] if e.labels else e.uri.rsplit("/",1)[-1]
+            note=e.description[:140].strip().replace("\\n"," ") if e.description else ""
+            if note and not note.endswith("."): note+="."
+            lines.append(f"- {label} — {note}")
+    else:
+        lines.append("No strong ontology matches were found; proceed with general best practices.")
+    lines += [
+        "\\nWhen responding, please:",
+        "- Make assumptions explicit; surface trade-offs and impacts if relevant.",
+        "- Use precise terms; avoid vague growth/technological-fix framings unless justified.",
+        "- If uncertain, state limits and what evidence would resolve them.",
+        "\\nNow provide your answer:"
+    ]
+    return "\\n".join(lines)
+
+def _find_matches(user_prompt: str, method: str, top_k: int, expansion_depth: int) -> List[Tuple[OntologyEntry, float]]:
+    entries=_index_entries(ONTOLOGY_PATH)
+    if method=="embedding":
+        try: base=embedding_scores(user_prompt, entries, COHERE_EMBED_MODEL)[:top_k]
+        except Exception: base=keyword_scores(user_prompt, entries)[:top_k]
+    else:
+        base=keyword_scores(user_prompt, entries)[:top_k]
+    if expansion_depth>0 and base:
+        expanded=expand_concepts(ONTOLOGY_PATH, [b[0] for b in base], depth=expansion_depth)
+        rescored=keyword_scores(user_prompt, expanded)
+        by_uri={}
+        for e,sc in base+rescored:
+            if (e.uri not in by_uri) or (sc>by_uri[e.uri][1]): by_uri[e.uri]=(e,sc)
+        return sorted(by_uri.values(), key=lambda x:x[1], reverse=True)[:top_k]
+    return base
+
+def _llm_chat(prompt: str, model: Optional[str] = None, temperature: float = 0.2) -> str:
+    mdl=_normalize_cohere_chat_model(model or COHERE_CHAT_MODEL)
+    try:
+        client=_get_cohere_client()
+    except Exception as e:
+        return f"[LLM disabled: {e}]"
+    try:
+        res=client.chat(model=mdl, message=prompt, temperature=temperature)
+        return getattr(res, "text", str(res))
+    except Exception as e:
+        if "not found" in str(e).lower() and mdl!="command-r":
+            try:
+                res=client.chat(model="command-r", message=prompt, temperature=temperature)
+                return getattr(res, "text", str(res))
+            except Exception as e2:
+                return f"[LLM error after fallback: {e2}]"
+        return f"[LLM error: {e}]"
+
+def enhance_prompt_tool(user_prompt: str,
+                        search_method: str = DEFAULT_SEARCH_METHOD,
+                        top_k: int = TOP_K_CONCEPTS,
+                        expansion_depth: int = EXPANSION_DEPTH,
+                        style: str = DEFAULT_STYLE,
+                        call_llm: bool = False,
+                        temperature: float = 0.2,
+                        chat_model: Optional[str] = None):
+    matches=_find_matches(user_prompt, method=search_method, top_k=top_k, expansion_depth=expansion_depth)
+    enhanced=make_enhanced_prompt(user_prompt, matches, style=style)
+    out={
+        "original_prompt": user_prompt,
+        "enhanced_prompt": enhanced,
+        "matches":[{"uri":e.uri,"label":(e.labels[0] if e.labels else e.uri.rsplit("/",1)[-1]),"score":score}
+                   for e,score in matches]
+    }
+    if call_llm:
+        out["original_reply"]=_llm_chat(user_prompt, model=chat_model, temperature=temperature)
+        out["enhanced_reply"]=_llm_chat(enhanced, model=chat_model, temperature=temperature)
+    return out
+
+with gr.Blocks(title="Ontology Prompt Enhancer (MCP)") as demo:
+    gr.Markdown("# Ontology Prompt Enhancer (MCP)")
+    with gr.Row():
+        p = gr.Textbox(label="Your prompt")
+    with gr.Row():
+        m = gr.Radio(choices=["keyword","embedding"], value=DEFAULT_SEARCH_METHOD, label="Search method")
+        st = gr.Radio(choices=["minimal","balanced","verbose"], value=DEFAULT_STYLE, label="Prompt style")
+    with gr.Row():
+        k = gr.Slider(1, 20, value=TOP_K_CONCEPTS, step=1, label="Top-K concepts")
+        d = gr.Slider(0, 3, value=EXPANSION_DEPTH, step=1, label="Expansion depth")
+    with gr.Row():
+        call = gr.Checkbox(False, label="Also call LLM (Cohere)")
+        temp = gr.Slider(0.0, 1.0, value=0.2, step=0.05, label="Temperature")
+        model = gr.Textbox(value=COHERE_CHAT_MODEL, label="Cohere chat model")
+    out = gr.JSON(label="Result")
+    gr.Button("Enhance").click(fn=enhance_prompt_tool, inputs=[p,m,k,d,st,call,temp,model], outputs=out)
+
+if __name__ == "__main__":
+    demo.launch(server_name="0.0.0.0", server_port=int(os.getenv("PORT", "7860")), mcp_server=True)