Phase 2 Build — Multi-Source Retrieval Oracle

{
  "id": 1, "title": "Ashfall Reign",

  // → SQL: things you would filter or sort on
  "rating": 4.6, "chapters": 812, "cadence_days": 2, "genre": "cultivation",

  // → FAISS: prose nobody would ever write a WHERE clause against
  "reviews": "Slow start, but the arc after ch.300 is peak. No filler.",

  // → Neo4j: edges to other entities, not attributes of this one
  "author": "Lin Wu", "translator": "Skybound TL", "shares_universe_with": [3]
}

# generate_data.py
import sqlite3, json
import numpy as np, faiss
from sentence_transformers import SentenceTransformer
from neo4j import GraphDatabase

MODEL = SentenceTransformer("all-MiniLM-L6-v2")
NOVELS = [{"id": 1, "title": "Ashfall Reign", "rating": 4.6, "chapters": 812,
           "cadence_days": 2, "genre": "cultivation",
           "reviews": "Slow start but the arc after ch.300 is peak.",
           "author": "Lin Wu", "shares_universe_with": [3]},
          # ... 24 more ...
]

def build_sql():
    con = sqlite3.connect("data/novels.db")
    con.execute("""CREATE TABLE IF NOT EXISTS novels(
        id INTEGER PRIMARY KEY, title TEXT, rating REAL,
        chapters INTEGER, cadence_days INTEGER, genre TEXT)""")
    for n in NOVELS:
        # INSERT OR REPLACE keeps re-runs idempotent
        con.execute("INSERT OR REPLACE INTO novels VALUES (?,?,?,?,?,?)",
                    (n["id"], n["title"], n["rating"],
                     n["chapters"], n["cadence_days"], n["genre"]))
    con.commit(); con.close()
    print(f"[SQL]   inserted {len(NOVELS)} rows")

def build_faiss():
    texts = [n["reviews"] for n in NOVELS]
    vecs = MODEL.encode(texts, normalize_embeddings=True)
    # normalized vectors + inner product == cosine similarity
    index = faiss.IndexFlatIP(vecs.shape[1])
    index.add(np.array(vecs, dtype="float32"))
    faiss.write_index(index, "data/reviews.faiss")
    # FAISS stores vectors only — keep your own id/text sidecar
    json.dump([{"id": n["id"], "title": n["title"], "text": n["reviews"]}
               for n in NOVELS], open("data/reviews_meta.json", "w"))
    print(f"[FAISS] embedded {len(texts)} review chunks")

def build_graph():
    driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "password"))
    with driver.session() as s:
        s.run("MATCH (n) DETACH DELETE n")          # fresh graph every run
        for n in NOVELS:
            # MERGE, not CREATE — no duplicate authors
            s.run("MERGE (nv:Novel {id:$id, title:$title}) "
                  "MERGE (a:Author {name:$author}) "
                  "MERGE (a)-[:WROTE]->(nv)",
                  id=n["id"], title=n["title"], author=n["author"])
            for other in n["shares_universe_with"]:
                s.run("MATCH (a:Novel {id:$a}),(b:Novel {id:$b}) "
                      "MERGE (a)-[:SHARES_UNIVERSE]->(b)", a=n["id"], b=other)
    driver.close(); print("[NEO4J] wrote nodes + edges")

if __name__ == "__main__":
    build_sql(); build_faiss(); build_graph()
    print("All three stores populated.")

# engines.py
import sqlite3, json
import numpy as np, faiss
from sentence_transformers import SentenceTransformer
from neo4j import GraphDatabase

MODEL = SentenceTransformer("all-MiniLM-L6-v2")

def query_sql(title: str) -> str:
    try:
        con = sqlite3.connect("data/novels.db")
        row = con.execute("SELECT rating,chapters,cadence_days,genre "
                          "FROM novels WHERE title=?", (title,)).fetchone()
        con.close()
        if not row:
            return "[SQL] no structured record found"
        r, ch, cad, g = row
        return f"[SQL] rating={r}/5, chapters={ch}, cadence={cad}d, genre={g}"
    except Exception as e:
        return f"[SQL ERROR] {e}"          # tagged, never raised

def query_faiss(title: str, k: int = 3) -> str:
    try:
        index = faiss.read_index("data/reviews.faiss")
        meta = json.load(open("data/reviews_meta.json"))

        # keyword precision: exact titles a vector search can miss
        exact = [m for m in meta if m["title"].lower() == title.lower()]

        # dense recall: semantically close reviews sharing no words
        qv = MODEL.encode([title], normalize_embeddings=True)
        _, I = index.search(np.array(qv, dtype="float32"), k)
        semantic = [meta[i] for i in I[0]]

        hits = (exact + semantic)[:k]        # hybrid fusion, exact first
        return "[FAISS] " + " | ".join(h["text"] for h in hits)
    except Exception as e:
        return f"[FAISS ERROR] {e}"

def query_graph(title: str) -> str:
    try:
        driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "password"))
        with driver.session() as s:
            # OPTIONAL MATCH so a novel with no siblings still returns
            rec = s.run("MATCH (a:Author)-[:WROTE]->(n:Novel {title:$t}) "
                        "OPTIONAL MATCH (n)-[:SHARES_UNIVERSE]->(o:Novel) "
                        "RETURN a.name AS author, collect(o.title) AS universe",
                        t=title).single()
        driver.close()
        if not rec:
            return "[GRAPH] no graph record found"
        return f"[GRAPH] author={rec['author']}, shared_universe={rec['universe']}"
    except Exception as e:
        return f"[GRAPH ERROR] {e}"       # e.g. Neo4j container down

# retrieve.py
import time
from concurrent.futures import ThreadPoolExecutor
from engines import query_sql, query_faiss, query_graph

def parallel_retrieve(title: str) -> dict:
    with ThreadPoolExecutor(max_workers=3) as ex:
        futures = {"sql":   ex.submit(query_sql,   title),
                   "faiss": ex.submit(query_faiss, title),
                   "graph": ex.submit(query_graph, title)}
        # each engine already returns a tagged string, so .result()
        # never raises and one dead store cannot take down the request
        return {k: f.result() for k, f in futures.items()}

if __name__ == "__main__":
    t = time.time()
    print(parallel_retrieve("Ashfall Reign"))
    print(f"latency {time.time()-t:.2f}s (slowest engine, not the sum)")

# synth.py
import dspy

teacher = dspy.LM("openai/gpt-4o",      temperature=0.7)   # generates demos
student = dspy.LM("openai/gpt-4o-mini", temperature=0.0)   # serves traffic
dspy.configure(lm=student)

class ReadVerdict(dspy.Signature):
    """Decide BINGE or DROP for a web novel, reasoning across
    structured stats, reader reviews, and graph context."""
    sql_context:   str = dspy.InputField(desc="ratings, chapters, cadence")
    faiss_context: str = dspy.InputField(desc="reader review snippets")
    graph_context: str = dspy.InputField(desc="author / shared universe")

    verdict:   str = dspy.OutputField(desc="exactly BINGE or DROP")
    score:     int = dspy.OutputField(desc="confidence 0-100")
    rationale: str = dspy.OutputField(desc="2-3 sentence justification")

class Oracle(dspy.Module):
    def __init__(self):
        super().__init__()
        # ChainOfThought makes the model reason before committing
        self.synth = dspy.ChainOfThought(ReadVerdict)

    def forward(self, ctx: dict):
        return self.synth(sql_context=ctx["sql"],
                          faiss_context=ctx["faiss"],
                          graph_context=ctx["graph"])

# judge.py
BANNED = {"maybe", "unsure", "cannot decide"}   # no fence-sitting

def judge(example, pred, trace=None) -> bool:
    v = (pred.verdict or "").strip().upper()
    if v not in {"BINGE", "DROP"}:            # required enum
        return False
    if not 0 <= int(pred.score) <= 100:         # valid range
        return False
    if any(b in pred.rationale.lower() for b in BANNED):
        return False
    return len(pred.rationale.split()) >= 8       # required substance

# compile.py
import dspy
from dspy.teleprompt import BootstrapFewShot
from retrieve import parallel_retrieve
from synth import Oracle, teacher
from judge import judge

MANIFEST = ["Ashfall Reign", "Voidheart Saga", "Nine Sigil Gate"]

# real retrieval, so demos are grounded in real context shapes
trainset = []
for title in MANIFEST:
    ctx = parallel_retrieve(title)
    trainset.append(dspy.Example(ctx=ctx).with_inputs("ctx"))

optimizer = BootstrapFewShot(metric=judge,
                             teacher_settings=dict(lm=teacher),
                             max_bootstrapped_demos=4)
compiled = optimizer.compile(Oracle(), trainset=trainset)
compiled.save("data/oracle_compiled.json")     # ← the evidence
print("Saved compiled state to data/oracle_compiled.json")

# compare.py
from synth import Oracle
from retrieve import parallel_retrieve

ctx = parallel_retrieve("Emberclad Throne")   # held out on purpose

baseline = Oracle()                              # uncompiled
tuned    = Oracle(); tuned.load("data/oracle_compiled.json")

b, t = baseline(ctx), tuned(ctx)
print("BEFORE:", b.verdict, "|", b.rationale)
print("AFTER :", t.verdict, "|", t.rationale)

# app.py — run with: streamlit run app.py
import streamlit as st
from retrieve import parallel_retrieve
from synth import Oracle

oracle = Oracle(); oracle.load("data/oracle_compiled.json")

st.title("Web Novel Oracle — BINGE or DROP")
title = st.text_input("Novel title", "Ashfall Reign")

if st.button("Run"):
    ctx = parallel_retrieve(title)

    # evidence first, in the same three columns every time
    c1, c2, c3 = st.columns(3)
    c1.subheader("SQL");   c1.code(ctx["sql"])
    c2.subheader("FAISS"); c2.code(ctx["faiss"])
    c3.subheader("Graph"); c3.code(ctx["graph"])

    # then the verdict the model drew from it
    pred = oracle(ctx)
    st.header(f"Verdict: {pred.verdict}  ({pred.score}/100)")
    st.write(pred.rationale)

# Neo4j via Docker (README §2)
docker run -d --name neo4j \
  -p 7474:7474 -p 7687:7687 \
  -e NEO4J_AUTH=neo4j/password \
  neo4j:5

# then, from the repo root
pip install -r requirements.txt
python generate_data.py         # populate all three stores
streamlit run app.py            # launch the transparency UI