#!/usr/bin/env python3 """Moonshots Live — ask the corpus. Retrieval + page composition demo. Hybrid retrieval exactly as the spec calls for: dense vectors.npy, bge-large-en-v1.5, cosine on normalised vectors sparse bm25.pkl (BM25Okapi over the same 3,767 chunks) fused Reciprocal Rank Fusion — combines RANKINGS, because the two score scales are not comparable Then composes a page the way compose_page will: picks a treatment, attaches the span quote, the real clip, the speaker, the curated asset and any company logos. This is a demo of the retrieval + composition feel. It does NOT speak — no voice agent, no LLM in the loop. Everything shown is selected, never generated. PORT=8801 python3 ask.py """ import http.server, socketserver, json, os, pathlib, pickle, re, socket, threading import urllib.parse, urllib.request MOON = pathlib.Path(__file__).resolve().parent ROOT = MOON.parent # serve from here: ASSETS_WEB + Moonshots both visible IDX = MOON / "data/index" PORT = int(os.environ.get("PORT", "8801")) MODEL_NAME = "BAAI/bge-large-en-v1.5" # BGE retrieval asymmetry: queries get this prefix, passages do not. QPREFIX = "Represent this sentence for searching relevant passages: " print("loading index…", flush=True) import numpy as np VEC = np.load(IDX / "vectors.npy") CHUNK_IDS = json.loads((IDX / "chunk_ids.json").read_text()) BM25 = pickle.loads((IDX / "bm25.pkl").read_bytes()) # Span-level vectors, built by build_span_index.py. The chunk index answers # "which passage is about this"; this one answers "which LINE answers this", # which is what actually goes on the page. SPANVEC = None _sv = IDX / "span_vectors.npy" if _sv.exists(): SPANVEC = np.load(_sv) # The relevance re-rank scores each hit by its span, its claim and its topic # labels. All three are static per chunk and were being re-embedded on every # request — 147-178 texts per query, and three times that when the widening # pass fired. Harmless at 1.1s on a Mac; 11-13s on the deployed CPU instance, # which is long enough that the agent stops waiting for search_corpus and # calls render_page before there are any results. Built by # build_rerank_vectors.py, aligned to chunk_ids.json. CLAIMVEC = np.load(IDX / "claim_vectors.npy") if (IDX / "claim_vectors.npy").exists() else None TOPICVEC = np.load(IDX / "topic_vectors.npy") if (IDX / "topic_vectors.npy").exists() else None POS = {c: i for i, c in enumerate(CHUNK_IDS)} SPANS = {} for line in (MOON / "data/spans.jsonl").read_text().splitlines(): if line.strip(): d = json.loads(line) SPANS[d["chunk_id"]] = d ENTS = {} for line in (MOON / "data/entities.jsonl").read_text().splitlines(): if line.strip(): d = json.loads(line) ENTS[d["chunk_id"]] = d # Who each chunk TALKS ABOUT, which is not the same as who is speaking. "What # did they say about Elon Musk?" returned one voice and the weakest art match # of the QC set, because a name is not a topic: the query matched passages # about spaceflight rather than passages mentioning Elon. The extractor has # recorded this all along and retrieval never used it. PEOPLE = {} for _cid, _e in ENTS.items(): for _p in (_e.get("people") or []): _n = (_p if isinstance(_p, str) else _p.get("name", "")).strip().lower() if len(_n) >= 4: PEOPLE.setdefault(_n, []).append(_cid) # The hosts are named in almost every episode, so matching them would filter # nothing. A question about what a host thinks is served by the speaker field. for _h in ("alex", "salim", "peter", "dave", "emad", "imad", "alex wissner-gross", "peter diamandis", "dave blundin", "salim ismail", "emad mostaque"): PEOPLE.pop(_h, None) CHUNKS = {} for line in (MOON / "data/chunks.jsonl").read_text().splitlines(): if line.strip(): d = json.loads(line) CHUNKS[d["chunk_id"]] = d # Single source of truth for clip windows, written by quotecut.py. Computing # them in two places drifted once already: the server said span_in_clip while # the audio had been cut to a different window. # 16% of spans cross a speaker change and get one name and one face. Built by # build_speaker_trim.py: the displayed quote narrowed to the attributed # speaker's own words. The AUDIO is untouched — it still plays the exchange. SPK_TRIM = {} _st = IDX / "speaker_trim.json" if _st.exists(): SPK_TRIM = json.loads(_st.read_text()) # The residue the trim cannot fix: 35 spans where the labelled speaker says # nothing at all inside the window, so there is no run to narrow to. The label # is wrong, not the width — QC found one running Ramez Naam's sentence about # fusion regulation under Peter's name and portrait. Attributing a named guest's # words to the host is worse than showing two voices instead of three, so these # are barred from panel selection. They stay in retrieval for context. SPK_BADLABEL = set() _su = IDX / "speaker_unattributed.json" if _su.exists(): SPK_BADLABEL = set(json.loads(_su.read_text())) CUTW = {} _cw = IDX / "cut_windows.json" if _cw.exists(): CUTW = json.loads(_cw.read_text()) # build_span_lead.py: what each speaker said in the ~20s before their span, # as 1-3 sentence run-ups, shortest first. Used to give a quote back the # sentence that says what it is about — see restore_context(). SPAN_LEAD = {} _sl = IDX / "span_lead.json" if _sl.exists(): SPAN_LEAD = json.loads(_sl.read_text()) # Diarization is ~98.3% accurate by talk time; the remainder puts the wrong # host's face on a quote, which at Peter's own event is worse than a dull # headline. Scribe merged a stretch of Dave into Alex's bucket in one episode. # There is no signal in the data to detect this, so corrections are recorded by # hand, same pattern as the logo overrides. SPK_OVERRIDE = {} _so = MOON / "speaker_overrides.json" if _so.exists(): SPK_OVERRIDE = json.loads(_so.read_text()) VM = json.loads((ROOT / "visual_manifest.json").read_text()) ALIAS, PARENT, NOTCO, LOGOS = (VM["company_aliases"], VM["product_parents"], set(VM["not_companies"]), VM["logos"]) SPK = {"Peter Diamandis": "pd", "Alex Wissner-Gross": "awg", "Dave Blundin": "db", "Salim Ismail": "si", "Emad Mostaque": "em"} print(f" {len(CHUNK_IDS)} chunks · {len(SPANS)} spans · {len(VM['topics'])} topics", flush=True) _model = None _lock = threading.Lock() def model(): """Loaded on first query so the server comes up immediately.""" global _model with _lock: if _model is None: print("loading embedding model (first query only)…", flush=True) from sentence_transformers import SentenceTransformer _model = SentenceTransformer(MODEL_NAME) print(" model ready", flush=True) return _model def tok(s): return re.findall(r"[a-z0-9']+", s.lower()) # Words that carry no discrimination in a question. "AI" is not here but might # as well be — its IDF in this corpus is 1.02, below "the" at 1.71, because # nearly every chunk mentions it. The IDF floor below removes it and its kind. QSTOP = set("""a an the and or but if is are was were be been being do does did how what when where which who whom why will would can could should shall may might must have has had get got go going about into over under with without for from that this these those there their them they it its of on in to at as by so we you i he she me my our your us not no yes any some all more most much many other else new now then than too very just really actually like think thing things say said says tell talk talking know people way ways lot going gonna guys mates moonshots episode show podcast""".split()) IDF_FLOOR = 2.5 # "ai" 1.02 out, "change" 3.58 in, "education" 4.68 in MAX_TERMS = 4 PLAY_LEAD_MS = 6000 # default run-up actually played, inside a longer cut # Asking what a particular host said is not the same as asking about a topic # they come up in. "What has Peter said about abundance?" led with a Salim # quote — correct on the topic, wrong on the person, and the agent then # introduced it as Peter in his own words. HOST_ALIAS = {} for _full, _k in (("Peter Diamandis", "pd"), ("Alex Wissner-Gross", "awg"), ("Dave Blundin", "db"), ("Salim Ismail", "si"), ("Emad Mostaque", "em")): _first, _last = _full.split(" ", 1) for _a in (_full, _first, _last, _last.replace("-", " ")): HOST_ALIAS[" ".join(tok(_a))] = _full def host_named(q): """The one host this question is asking ABOUT, if it names exactly one.""" ql = " " + " ".join(tok(q)) + " " found = {HOST_ALIAS[a] for a in HOST_ALIAS if " " + a + " " in ql} return found.pop() if len(found) == 1 else None def people_in(q): """Chunk indices that mention a person the question names, longest name first so "Elon Musk" beats "Elon".""" ql = " " + " ".join(tok(q)) + " " best = None for name in sorted(PEOPLE, key=len, reverse=True): if " " + " ".join(tok(name)) + " " in ql: best = name break if not best: return None, None idx = [POS[c] for c in PEOPLE[best] if c in POS] return (best, idx) if len(idx) >= 3 else (None, None) def key_terms(q): """The question's discriminating words, rarest first. A quote can mention AI and still not be about what was asked. The terms that decide that are the rare ones — "education", "fusion", "longevity" — so rank the question's words by corpus IDF and keep the top few.""" idf = getattr(BM25, "idf", {}) ts = {t for t in tok(q) if t not in QSTOP and len(t) > 2} ts = [t for t in ts if idf.get(t, 8.0) >= IDF_FLOOR] return sorted(ts, key=lambda t: -idf.get(t, 8.0))[:MAX_TERMS] def mark_on_point(panel, q, terms, qv): """Flag quotes that name the question's subject without engaging the ask. "Are the frontier labs overvalued?" has two parts — the labs, and the valuation — and a quote carrying only the first ("every frontier lab is buying bio companies") scores well on everything we measure and still reads as filler. This is a WEAK, advisory signal, and it is deliberately not a gate. Four instruments were tried and all four failed somewhere: word match on the rarest term called Emad's "$75 million per state" on point (it says "worths") and Salim's "the value of these Frontier labs" off point (it says "value") — exactly inverted strike the term and re-embed mangling a short question moves EVERY hit away, so the delta is positive for all term vs the hit's topic labels Salim's labels say "frontier ai labs, ai regulation" and never mention valuation, so he scores below an unrelated topic claim-similarity gain measures length, not aboutness So the count only decides whether to widen the query and what to tell the agent, who reads the quotes and can judge this far better than a cosine. Never drop a quote on it.""" idf = getattr(BM25, "idf", {}) rarest = terms[0] if terms else None n = 0 for h in panel: hay = set(tok(" ".join([h.get("quote") or "", h.get("claim") or ""] + (h.get("topics") or [])))) hit = not rarest or any( t == rarest or (len(t) >= 5 and (t in rarest or rarest in t)) for t in hay) h["on_point"] = hit n += bool(hit) return n def restore_context(h, terms): """Give a quote back the sentence that says what it is about. QC put Alex's "We can't find one that survives a tsunami of AI plus 20 other Gutenberg moments" on the AI-education page. Not off topic — two sentences earlier he says "Democracy is breaking, education is broken, all the institutions", and the "one" is one of those. But the word the reader needs was outside the span, so on the page it read as a non-sequitur. Whether a quote reads that way depends on the question, which is why this runs here and not in the index build: fire only when the quote itself misses every discriminating term of the question AND a run-up carries one. Takes the shortest run-up that does, because the pull quote still has to be a pull quote. Mutates h in place.""" opts = SPAN_LEAD.get(h["chunk_id"]) if not terms or not opts: return if any(t in set(tok(h["quote"])) for t in terms): return # the quote already answers on its own for o in opts: # shortest run-up first if not any(t in set(tok(o["text"])) for t in terms): continue cut_start = h.get("cut_start_ms") if cut_start is None or o["start_ms"] < cut_start: # the audio for this run-up was never cut into the clip; showing # text the clip does not speak is worse than the non-sequitur h["lead_missing_ms"] = (cut_start - o["start_ms"]) if cut_start else None return h["quote"] = o["text"] + " " + h["quote"] h["lead_added"] = o["text"] at = round(max(0, o["start_ms"] - cut_start) / 1000, 2) h["span_in_clip"] = [at, h["span_in_clip"][1]] h["play_from"] = at # start where the text on screen starts return def logo_for(name): if name in NOTCO: return None n = PARENT.get(ALIAS.get(name, name), ALIAS.get(name, name)) e = LOGOS.get(n) return {"label": name, "resolved": n, "src": e["src"] if e else None, "tone": (e or {}).get("tone")} # Peter hands between segments constantly: "Alex, I'm gonna move us to a # conversation about energy and AI." Relevant on paper, useless as a quote — # it is the pivot, not the point, and the substance is in the next sentence. TRANSITION = re.compile( r"^(?:\w+,\s*)?(?:i'?m gonna |let'?s |we'?re gonna |i want to |i'?ll |" r"moving on|next (?:story|up)|coming up|turning to|let me |" r"let'?s (?:jump|move|take|go|talk))", re.I) # A span that opens by addressing a host BY NAME and is attributed to that same # host has straddled a speaker change — Peter asks, Alex answers, diarization # gave the whole thing to one of them. 20 spans do this. ADDRESSED = re.compile(r"^(Peter|Alex|Dave|Salim|Emad)\b\s*[,:]", re.I) FIRSTNAME = {"Peter": "Peter Diamandis", "Alex": "Alex Wissner-Gross", "Dave": "Dave Blundin", "Salim": "Salim Ismail", "Emad": "Emad Mostaque"} INTERROG = re.compile(r"^(what|who|how|why|is|are|do|does|did|can|could|would|should|tell|let)\b", re.I) DISFLUENT = re.compile(r"\b(uh|um|you know|i mean|sort of|kind of|like)\b", re.I) # Relevance and topic-match reward a line for SAYING "data centers", not for # saying anything about them. "We need to know what is actually going on" # scored rel 0.707 / tsim 0.92 and beat Dave's own "It's called Provocative AI. # The data center is water negative and carbon negative." Specificity is the # missing axis: concrete quotes carry numbers, names and units. VAGUE = re.compile(r"\b(we need to know|what'?s actually going on|things like that|" r"that kind of (?:thing|stuff)|and so forth|et cetera|or whatever|" r"a bunch of|a lot of stuff|it'?s interesting|i don'?t know if)\b", re.I) NUMBER = re.compile(r"\b\d") PROPER = re.compile(r"(? 'Asking…' Also drops a leading direct-address question when it is aimed at the very person the span is attributed to. "Alex, which one gets us to AGI faster? Um, well, it's a trick question…" is Peter asking and Alex answering; the displayed quote should be Alex's answer, not Peter's question under Alex's face. The audio still carries both, which is the conversation.""" out = q.strip() m = ADDRESSED.match(out) if m and speaker and FIRSTNAME.get(m.group(1).capitalize()) == speaker: nxt = re.search(r"[.?!]\s+", out) if nxt and len(out) - nxt.end() >= 40: out = out[nxt.end():].lstrip() for _ in range(6): # they stack; peel one at a time m = _LEAD.match(out) if not m: break cand = out[m.end():].lstrip() if len(cand) < 40: # never trim a quote into a fragment break out = cand # bare connectives with no comma stack too: "so but asking…" for _ in range(3): m = re.match(r"^(?:but|so|and|well|now)\s+(?=[a-z])", out, re.I) if not m or len(out) - m.end() < 40: break out = out[m.end():] if out and out[0].islower(): out = out[0].upper() + out[1:] return out or q def retrieve(q, k=16): """Dense + BM25, fused with RRF. Returns (ranked chunk records, query vec).""" qv = model().encode([QPREFIX + q], normalize_embeddings=True)[0] dense = VEC @ qv # vectors are unit-norm sparse = np.asarray(BM25.get_scores(tok(q)), dtype=np.float32) RRF_K = 60 fused = np.zeros(len(CHUNK_IDS), dtype=np.float32) # Weighted RRF. Equal arms let a chunk strong on dense AND bm25 outvote one # that only the span arm can see — but the span arm is the only one that # knows what the LINE says, and the line is what goes on the page. It had # the interpretability quote at rank 18 of 3767 while the fused result # never surfaced it at all. arms = [(dense, 0.8), (sparse, 0.8)] if SPANVEC is not None: arms.append((SPANVEC @ qv, 1.6)) # When the question names a person, rank the chunks that actually mention # them as a fourth arm — ordered among themselves by span similarity, so it # contributes a ranking rather than a flat boost and composes with the rest. who, pidx = people_in(q) if pidx: base = (SPANVEC @ qv) if SPANVEC is not None else dense parm = np.full(len(CHUNK_IDS), -np.inf, dtype=np.float32) parm[pidx] = base[pidx] arms.append((parm, 1.2)) for arm, w in arms: ranked = np.argsort(-arm)[:400] for r, i in enumerate(ranked): if not np.isfinite(arm[i]): break fused[i] += w / (RRF_K + r + 1) order = np.argsort(-fused)[:k] out = [] for i in order: cid = CHUNK_IDS[int(i)] sp, en = SPANS.get(cid, {}), ENTS.get(cid, {}) if not sp.get("quote"): continue comps = [logo_for(c["name"] if isinstance(c, dict) else c) for c in (en.get("companies") or [])] comps = [c for c in comps if c] # Play the CHUNK cut so the thought finishes; display the span as the # pull quote and light it up when the audio reaches it. The span cut is # kept for the typographic treatment, where the quote *is* the page. ck = CHUNKS.get(cid, {}) c_start = ck.get("start_ms", int(cid.rsplit("_", 1)[1])) c_end = ck.get("end_ms", c_start) q0, q1 = sp.get("quote_start_ms", c_start), sp.get("quote_end_ms", c_start) # Prefer the quote cut (lead-in + the whole span). The chunk cut is # capped at 20s while chunks run to 56s, so for 36% of spans the quote # starts after the chunk cut ends and is never audible. Fall back to # the chunk cut only while quotecut.py is still running. # Quote mp3s live on S3/CloudFront in deploy. Do not require them on disk # (the container image does not ship clips_quote/). cut_windows.json is # the source of truth that a quote cut exists. if cid in CUTW: cut_start, cut_end = CUTW[cid] clip_rel = f"Moonshots/data/clips_quote/{cid}.mp3" clip_dur = round((cut_end - cut_start) / 1000, 2) else: cut_start = c_start clip_rel = f"Moonshots/data/clips_chunkcut/{cid}.mp3" clip_dur = round(max(0, min(c_end, c_start + 20000) - c_start) / 1000, 1) speaker = SPK_OVERRIDE.get(cid) or sp.get("speaker") or "Clip / third party" out.append({ "chunk_id": cid, "quote": trim_lead(SPK_TRIM.get(cid) or sp["quote"], speaker), "quote_full": sp["quote"], "speaker": speaker, "key": SPK.get(speaker, "gu"), "episode": sp.get("episode_title", ""), "clip": clip_rel, "episode_mp3": f"Moonshots/data/raw/{sp.get('episode_id','')}.mp3", "clip_end_in_episode": round(cut_start / 1000 + clip_dur, 2), "clip_span": f"Moonshots/data/clips/{cid}.mp3", "context": ck.get("text", ""), "cut_start_ms": cut_start, "span_in_clip": [round(max(0, q0 - cut_start) / 1000, 2), round(max(0, q1 - cut_start) / 1000, 2)], # Clips are cut with up to 21s of run-up so a quote CAN be given # back its setup sentence, but most questions don't need it and # 21s of preamble before every quote would be unbearable. Start # PLAY_LEAD before the line; restore_context() moves this earlier # on the pages that need it. "play_from": round(max(0, (q0 - PLAY_LEAD_MS) - cut_start) / 1000, 2), "clip_dur": clip_dur, "dur": round((sp.get("quote_end_ms", 0) - sp.get("quote_start_ms", 0)) / 1000, 1), "safe": bool(sp.get("stage_safe")), "claim": en.get("claim", ""), "stance": en.get("stance", ""), "topics": en.get("topics") or [], "companies": comps, "score_raw": sp.get("score", 0), "scores": {"dense": round(float(dense[i]), 4), "bm25": round(float(sparse[i]), 2), "rrf": round(float(fused[i]), 5)}, }) # Retrieval scores the CHUNK, but the span we display is a couple of # sentences inside it and may be about something else entirely: for "AI's # effect on education" the chunk matched while the span was about AI # consciousness. Embed the spans themselves and rank on that. # # Sorting on quotability alone was the bug — it promoted a well-formed # irrelevant line over Salim's "we're still teaching people the way we # taught them 150 years ago". Relevance leads; quotability is the nudge # that settles which of two relevant lines reads better on a page. for h in out: h["scores"]["quote"] = round(quotability(h), 2) if out: try: # Three views of the same hit, because no one of them is reliable # alone. The raw span is short and often omits the subject entirely # ("we're still teaching people the way we taught them 150 years # ago" never says AI or education, and scored 0.36). The extracted # claim states the substance. The topic labels are the cleanest # signal of all — they are what the extractor decided the passage # was ABOUT. One batched encode keeps this at ~150ms. n = len(out) if CLAIMVEC is not None and TOPICVEC is not None and SPANVEC is not None: # One dot product per hit against vectors built offline. No # encoding here at all — the query was the only text embedded. rows = [POS[h["chunk_id"]] for h in out] sv = SPANVEC[rows] @ qv cv = CLAIMVEC[rows] @ qv tv = TOPICVEC[rows] @ qv else: texts = ([h["quote"] for h in out] + [(h.get("claim") or h["quote"]) for h in out] + [", ".join(h.get("topics") or ["none"]) for h in out]) ev = model().encode(texts, normalize_embeddings=True) @ qv sv, cv, tv = ev[:n], ev[n:2 * n], ev[2 * n:] for i, h in enumerate(out): rel = 0.30 * float(sv[i]) + 0.35 * float(cv[i]) + 0.35 * float(tv[i]) h["scores"]["span"] = round(float(sv[i]), 4) h["scores"]["topic"] = round(float(tv[i]), 4) h["scores"]["rel"] = round(rel, 4) except Exception as e: print(f" relevance embed failed: {e}", flush=True) for h in out: h["scores"]["rel"] = h["scores"]["dense"] # Relevance decides, full stop. Quotability only separates lines that are # genuinely tied: at 0.02 a fluent off-topic quote still outranked Salim's # "we're still teaching people the way we taught them 150 years ago" by # 0.015 of relevance. It is a tie-breaker, not a thumb on the scale. out.sort(key=lambda h: -(h["scores"].get("rel", 0) + 0.008 * h["scores"]["quote"])) return out, qv def _strong_topics(): """Topics the HOSTS cover substantially and that we have art for. Used when the corpus is thin on what was asked — she pivots to solid ground rather than serving tangential quotes with false confidence.""" have = {t for t, v in VM["topics"].items() if v["pool"]} count = {} for cid, en in ENTS.items(): sp = SPANS.get(cid) if not sp or sp.get("speaker") not in SPK: continue # hosts only for t in (en.get("topics") or []): tl = t.lower() for c in have: if c.lower() == tl or c.lower() in tl: count[c] = count.get(c, 0) + 1 return sorted([t for t, n in count.items() if n >= 6], key=lambda t: -count[t]) STRONG = _strong_topics() print(f" {len(STRONG)} well-covered topics", flush=True) _tvecs = None _svecs = None def topic_vectors(): """Embeddings for every curated topic that actually has art. Built once, on the same model as the index, so the query vector can be compared directly.""" global _tvecs if _tvecs is None: names = [t for t, v in VM["topics"].items() if v["pool"]] _tvecs = (names, model().encode(names, normalize_embeddings=True)) return _tvecs def strong_vectors(): global _svecs if _svecs is None: _svecs = (STRONG, model().encode(STRONG, normalize_embeddings=True)) return _svecs def art_topic(q, hits, qv): """Find a curated topic with art. Exact equality was far too strict: the extractor emits 3,289 free-form topics ('data center infrastructure', 'data center cooling', 'water consumption') while only 140 are curated ('data centers'), so good assets went unused and the page fell back to typographic. Widen it in tiers, loosest last, and report which tier hit.""" have = {t: v for t, v in VM["topics"].items() if v["pool"]} # Tiers were the wrong shape: "exact" is not automatically best. For # "environmental impacts of AI energy use" the top hit carried the tag # 'nanotechnology', which is curated and has art, so the exact tier took it # over 'ai infrastructure' further down. Gather candidates from every # route and score them on ONE scale — closeness to the question, with a # modest bonus for being tagged on a high-ranked hit. if qv is not None: try: cand = {} for rank, h in enumerate(hits): w = 1.0 / (1 + rank) for t in (h.get("topics") or []): tl = t.lower() if t in have: cand[t] = max(cand.get(t, 0), w) else: for c in have: if c.lower() in tl or tl in c.lower(): cand[c] = max(cand.get(c, 0), w * 0.8) names, tv = topic_vectors() sims = tv @ qv for i in np.argsort(-sims)[:6]: # semantic neighbours too cand.setdefault(names[int(i)], 0.0) simof = {n: float(s) for n, s in zip(names, sims)} scored = [(simof.get(t, 0.0) + 0.09 * cand[t], t) for t in cand] scored.sort(reverse=True) if scored and scored[0][0] >= 0.30: t = scored[0][1] tag = "tagged" if cand.get(t, 0) > 0 else "semantic" return t, f"{tag}/{simof.get(t,0):.2f}" except Exception as e: print(f" topic scoring failed: {e}", flush=True) # --- no query vector (shouldn't happen in the server path) --- # 1. exact — but weighted. Taking the most frequent curated tag picked # 'nanotechnology' for a question about AI energy use, because one hit # happened to carry it. Weight each candidate by the rank of the hit that # contributed it, then break ties on closeness to the QUESTION, so the # picture tracks what was asked rather than an incidental tag. counts = {} for rank, h in enumerate(hits): for t in h["topics"]: if t in have: counts[t] = counts.get(t, 0.0) + 1.0 / (1 + rank) if counts: cands = sorted(counts, key=lambda t: -counts[t])[:6] if qv is not None and len(cands) > 1: try: cv = model().encode(cands, normalize_embeddings=True) sims = cv @ qv best = max(range(len(cands)), key=lambda i: counts[cands[i]] * 0.5 + float(sims[i])) return cands[best], "exact" except Exception: pass return cands[0], "exact" # 2. containment — 'data center infrastructure' contains 'data center' chunk_topics = [t for h in hits for t in (h.get("topics") or [])] for ct in chunk_topics: cl = ct.lower() for t in have: tl = t.lower() if tl in cl or cl in tl: return t, "contains" # 3. word overlap — 'ai model development' / 'ai development' stop = {"ai", "the", "of", "and", "in", "for", "a"} best, score = None, 0.0 for ct in chunk_topics: cw = {w for w in re.findall(r"[a-z]+", ct.lower()) if w not in stop} for t in have: tw = {w for w in re.findall(r"[a-z]+", t.lower()) if w not in stop} if not cw or not tw: continue ov = len(cw & tw) / min(len(cw), len(tw)) if ov > score: best, score = t, ov if score >= 0.5: return best, f"overlap/{score:.2f}" # 4. semantic — nearest curated topic to the question itself try: names, tv = topic_vectors() sims = tv @ qv i = int(np.argmax(sims)) if float(sims[i]) >= 0.34: return names[i], f"semantic/{float(sims[i]):.2f}" except Exception as e: print(f" topic embed failed: {e}", flush=True) return None, "none" # A question's discriminating term often is not the corpus's word for it. The # asker says "overvalued"; Salim says "value", Peter says "stock valuation # drop", Dave says "undervaluing". One query cannot reach all three, so when # the first pass comes back with a single on-point voice, retry with the # question's own vocabulary swapped out and merge what comes back. # # Deliberately server-side. The agent could be asked to rephrase and search # again, but that adds a round-trip inside her speaking turn, and it was an # open re-search budget that once had her search eight times and then tell the # user there were no quotes. Here it is bounded, deterministic, and visible to # the QC run. # Widen by APPENDING the corpus's vocabulary, never by substituting into the # sentence — rewriting "Are the frontier labs overvalued?" around a synonym # produced "Are the frontier labs what these companies are valued at?", which # is worse than the original at the exact moment the original is struggling. REPHRASE = [ (r"\bovervalued\b|\bundervalued\b|\bvaluations?\b|\bworth\b|\bbubble\b", ["valuation stock price what they are valued at", "bubble, worth a fraction of what they were"]), (r"\bbottleneck\b|\bconstraints?\b", ["what is holding this back, the limit"]), (r"\brisks?\b|\bdangers?\b|\bworried\b", ["what could go wrong, the downside"]), (r"\btimelines?\b|\bhow (?:soon|close|long)\b", ["when this arrives, what year"]), (r"\bjobs?\b|\bemployment\b", ["work disappearing, who gets displaced"]), ] MAX_PASSES = 3 def answer(q, k=16): """Retrieve and compose, widening the query if the first pass is thin. "Thin" here is not low coverage — the valuation page scored 0.56 coverage and looked fine by every number we had, while carrying one real quote and two that merely said "labs".""" hits, qv = retrieve(q, k=k) page = compose(q, hits, qv) if page.get("treatment") == "thin" or page.get("on_point", 0) >= 2: return page tried = [q] for pat, subs in REPHRASE: if not re.search(pat, q, re.I): continue for sub in subs: if len(tried) >= MAX_PASSES: break alt = q + " — " + sub tried.append(alt) h2, _ = retrieve(alt, k=k) seen = {h["chunk_id"] for h in hits} hits = hits + [h for h in h2 if h["chunk_id"] not in seen] merged = compose(q, sorted(hits, key=lambda h: -h["scores"].get("rel", 0)), qv) if merged.get("on_point", 0) > page.get("on_point", 0): page = merged if page.get("on_point", 0) >= 2: break page["passes"] = tried return page def compose(q, hits, qv=None): """Choose a treatment and the assets, the way compose_page will.""" if not hits: return {"treatment": "typographic", "hits": [], "assets": [], "why": "no retrieval hits"} topic, how = art_topic(q, hits, qv) pool = VM["topics"][topic]["pool"] if topic else [] assets = [{"file": f, **{k: VM["assets"][f][k] for k in ("w", "h", "bucket", "orientation")}} for f in pool] comps, seen = [], set() for h in hits[:4]: for c in h["companies"]: if c["label"] not in seen: seen.add(c["label"]); comps.append(c) # It's a quintet, and one quote rarely settles a question. Take the best # quote from each DIFFERENT speaker, up to three — but only while they're # actually worth hearing. Padding to three with a weak third is worse than # two strong ones, so the third has to clear a floor. # Hosts only. A guest reading out a headline is a different voice but not a # third opinion, and it was filling the third slot. The floor is real now: # two strong voices beat three where the third is filler. # Gate on RELEVANCE, not quotability. Gating on quotability dropped Salim's # "we're still teaching people the way we taught them 150 years ago" (rel # 0.606, quotability 1.97) while admitting a less relevant line that merely # read more cleanly (rel 0.593, quotability 2.99). Quotability still has a # job — keeping out questions and mid-turn fragments — but it is a gate, # not a ranking. QUOT_GATE = 0.2 # The gate applies to the LEAD as well. On "AI and longevity" the most # relevant line was Peter's "Would love to see both of you guys. Uh, # anyway…" — rel 0.82, quotability 0.69. As the lead it also set the # relative floor, locking out Alex's far better quote just below it. A # throwaway line should never carry the page, however relevant it scores. # A segment pivot is never a quote, whatever it scores. The numeric gate # sits at 0.2 so that disfluent-but-substantive lines survive, which also # let "I'm gonna just put this up here right now" onto the page at 0.68. # Exclude transitions structurally instead of chasing the threshold. eligible = [h for h in hits if h["key"] != "gu" and h["chunk_id"] not in SPK_BADLABEL and h["scores"]["quote"] >= QUOT_GATE and not TRANSITION.match((h.get("quote_full") or h["quote"]).strip())] # The fallback used to ignore quotability entirely, which is how "Are we # going to mine asteroids?" put Salim's "You and I've, you've kind of # tracking that. You've invested and built companies around that." on a # page at coverage 0.7751. Nothing rescues a line with nothing in it, so # the emergency path keeps a floor; below it the page goes thin and the # agent says the mates mention it but have not dug in. EMERGENCY = -0.2 if not eligible: eligible = [h for h in hits if h["key"] != "gu" and h["chunk_id"] not in SPK_BADLABEL and h["scores"]["quote"] >= EMERGENCY] top_rel = max((h["scores"].get("rel", 0) for h in eligible), default=0) # A fixed 0.14 band is fine when the lead is strong and far too wide when it # is not. "Are the frontier labs overvalued?" led at 0.60, so the floor fell # to the 0.50 backstop and admitted anything mentioning labs — Peter on # buying bio companies, Dave on governance — either side of the one quote # that answers it. Scale the band to the lead so a weak lead cannot drag a # whole page of filler in behind it. # The ABSOLUTE floor was 0.50 and is now 0.46, swept over the forty in # attendee voice: 0.50 gave 26 clean with 1-VOICE the largest flag at 8; # 0.46 gives 29 clean with 1-VOICE down to 3; 0.43 also gives 29 but lets # filler back in (VAGUE reappears). 0.46 is the last value that buys voices # without buying padding — and it is only safe because emptiness() now # catches the vacuous lines the floor used to be standing in for. # # The multiplier stays at 0.90. It was moved to 0.88 to cure 1-VOICE and moved straight # back: on every 1-VOICE page in the forty-question run the binding # constraint is the ABSOLUTE 0.50 floor below, not this multiplier — # 0.88 x top_rel lands at 0.44-0.48 for those questions, so loosening it # changed nothing and admitted one more speaker-straddled quote. # # The lever for 1-VOICE is the 0.50 backstop. That is a separate decision # with a real cost, and it should be made on evidence, not by nudging. REL_FLOOR = max(0.46, top_rel - 0.14, top_rel * 0.90) # One quote per speaker means the FIRST one we meet for a host is the one # they get, and rel does not always rank their best answer first. "What # should I tell my kids to study?" put Peter's "WALL-E future or a Star Trek # future" (rel 0.542) above his "AI literacy, critical thinking, # adaptability" (rel 0.512) — so the real answer was discarded as a # duplicate speaker, the page was left weak, and the coverage gate then # refused the question outright on material the corpus covers well. # # So within a speaker, prefer the quote whose topic labels match the page's # subject, as long as it is still inside the band. subject = set(tok(topic or "")) - {"and", "of", "the", "in", "for"} def on_subject(h): return bool(subject) and any(subject & set(tok(t)) for t in (h.get("topics") or [])) by_key = {} for h in eligible: by_key.setdefault(h["key"], []).append(h) panel, seen, said = [], set(), set() for h in eligible: if h["key"] in seen: continue if panel and h["scores"].get("rel", 0) < REL_FLOOR: continue # try the next speaker, don't stop best = h if not on_subject(h): bar = max(REL_FLOOR, h["scores"].get("rel", 0) * 0.90) best = next((a for a in by_key[h["key"]] if a is not h and a["scores"].get("rel", 0) >= bar and on_subject(a)), h) # Episodes replay a teaser at the top, so the same sentence is indexed # twice — and the diarizer labelled the two copies as different people. # "Will we have self-driving cars everywhere?" put the identical line # under Salim AND Dave, 36 seconds and 99 minutes into the same episode. # One of those names is necessarily wrong; printing both is worse. fp = " ".join(tok(best["quote"]))[:120] if fp in said: continue seen.add(best["key"]) said.add(fp) panel.append(best) if len(panel) == 3: break # Rather than pad with a guest, fall back to the best hosts we have — but # never below the emergency floor, and an empty panel now means thin. if not panel: panel = [h for h in hits if h["key"] != "gu" and h["chunk_id"] not in SPK_BADLABEL and h["scores"]["quote"] >= EMERGENCY][:2] # If the question names one host, they lead. Their best eligible quote is # promoted to the front of the panel, or brought in if the topic ranking # left them out entirely — otherwise the page answers "what did Peter say" # in somebody else's voice. asked_host = host_named(q) if asked_host: mine = [h for h in eligible if h["speaker"] == asked_host] if mine: # their best ON-SUBJECT quote, not merely their best-ranked one: # "What has Peter said about abundance?" promoted his line about # SpaceX stock, which is his but is not the answer. lead = next((h for h in mine if on_subject(h)), mine[0]) # Promote them only if they actually said something relevant. If a # host never discussed the topic, a weak quote under their name is # worse than the best answer under someone else's — the agent can # say so, now that she is told who is really speaking. if lead["scores"].get("rel", 0) < 0.46: lead = None if lead is not None: panel = ([lead] + [h for h in panel if h["chunk_id"] != lead["chunk_id"]])[:3] # Only now, on the three lines that will actually be printed, check whether # each one carries the question on its own — and if not, hand it back the # sentence that does. kt = key_terms(q) for h in panel: restore_context(h, kt) n_on_point = mark_on_point(panel, q, kt, qv) rest = [h for h in hits if h not in panel] # --- coverage gate --------------------------------------------------- # Mean TOPIC similarity across the host hits: topic labels are the # extractor's judgement of what a passage is actually about, so they track # coverage better than rel, which cannot tell "answers this" from "adjacent # to this". # # The threshold is deliberately LOW. An earlier 0.63 was fitted to twelve # questions and broke on the thirteenth: "When will we reach AGI?" scored # 0.56 and was refused, because the hits are tagged "artificial general # intelligence" while the question says "AGI" — the same concept at half # the cosine, purely an acronym artefact. No threshold separates AGI # (0.563) from the black-box question (0.611), so don't try. Refuse only # what is unambiguously outside the corpus — sourdough 0.39, ocean cleanup # 0.45, car transmission 0.45 — and let the agent hedge on the middle via # if_thin. A false refusal on stage is far worse than an adjacent quote # carrying an honest caveat. tsims = [h["scores"].get("topic", 0) for h in panel] or [0] coverage = round(sum(tsims) / len(tsims), 4) THIN = 0.50 # A second pass-path keyed on the art topic was tried here and removed: any # question resolves to SOME tagged topic (sourdough lands on "biotech" at # 0.38), so it let every off-corpus question through. The false refusal it # was meant to fix turned out to be a selection bug — the panel was carrying # Peter's WALL-E line instead of his answer — and fixing that lifted this # question to 0.5318 on its own. Leave the gate alone. # A named person IS coverage, whatever the topic labels say. "What do they # think about Sam Altman?" was refused at 0.3656 while he is named in 82 # chunks and the people arm had already pulled them to the top — because # coverage compares the question to labels like "ai governance", which # never say his name. This cannot leak the way the tagged-topic path did: # a question about sourdough names nobody. PERSON_COVER = 12 _who, _pidx = people_in(q) person_covered = bool(_pidx) and len(_pidx) >= PERSON_COVER # Coverage as a cosine cannot separate "the corpus answers this" from "the # corpus is adjacent to this". Five different similarity measures were # tried and every one shifted the whole distribution instead of separating # the cases — content-words-only lifted "clean the ocean" (0.46 -> 0.57) # further than "is any of this dangerous" (0.50 -> 0.56). # # What actually differs is LEXICAL: the corpus says the word, or it does # not. Counting how many retrieved hits contain the question's rarest term # separates them cleanly where no cosine did: # # dangerous 7/20 code 8/20 moonshot 19/20 <- material exists # sourdough 0/20 transmission 1/20 behind 1/20 <- it does not # # This can only ever ADD a pass, never cause a refusal, so the downside is # a few adjacent pages the agent hedges on via if_thin. # Match on ANY of the question's strong terms, not just the rarest. Picking # the single rarest is a coin flip between near-equal words: "Should I learn # to code still?" put "learn" (idf 4.22) a hair above "code" (4.10), stemmed # the wrong one and scored 5/20 instead of 8/20. The IDF bar is higher here # than for retrieval so that a passable-but-common word like "still" (2.85) # cannot pad the count, and stems are at least four characters so "car" # cannot match "carbon". GROUND_MIN = 6 GROUND_IDF = 3.5 _idf = getattr(BM25, "idf", {}) stems = [t[:6] if len(t) > 7 else t[:4] for t in kt[:3] if len(t) >= 4 and _idf.get(t, 8.0) >= GROUND_IDF] grounded = 0 if stems: for h in hits[:20]: words = tok(" ".join([h.get("quote") or "", h.get("claim") or ""] + (h.get("topics") or []))) if any(w.startswith(st) for st in stems for w in words): grounded += 1 if (coverage < THIN and not person_covered and grounded < GROUND_MIN) or not panel: near = [] try: names, sv = strong_vectors() for i in np.argsort(-(sv @ qv))[:3]: near.append(names[int(i)]) except Exception: near = STRONG[:3] return {"treatment": "thin", "why": f"corpus is thin here (coverage {coverage})", "topic": topic, "topic_match": how, "coverage": coverage, "suggest_topics": near, "related_topics": [], "assets": assets, "companies": comps, "voices": 0, "hits": hits} if not assets: treat, why = "typographic", "no reviewed asset for any matched topic" elif len(comps) >= 3: treat, why = "logo_row", f"{len(comps)} companies named" elif len(panel) >= 3: treat, why = "panel", f"{len(panel)} hosts on the question" elif len(panel) == 2: treat, why = "split", "two hosts in the top hits" else: treat, why = "hero", "one host clears the bar on this question" # Nearest curated topics to the question. When she judges the hits too thin # she should offer the adjacent ground the mates DO cover, not tell the # user the corpus is empty — it never is, it's 3,767 chunks. related = [] try: if qv is not None: names, tv = topic_vectors() sims = tv @ qv for i in np.argsort(-sims)[:5]: if names[int(i)] != topic and float(sims[int(i)]) >= 0.28: related.append(names[int(i)]) except Exception: pass return {"treatment": treat, "why": why, "topic": topic, "topic_match": how, "coverage": coverage, "related_topics": related[:4], "assets": assets, "companies": comps, # How many of the panel answer what was ASKED rather than merely # naming its subject. The agent has no other way to tell one strong # quote plus two fillers from three strong ones, and without it she # writes a headline off the question that the page cannot cash. "on_point": n_on_point, "key_terms": kt, "voices": len(panel), "hits": panel + rest} # ---------------------------------------------------------------- voice # .env sits next to this file and is gitignored; load it so `export` isn't # needed (each tool-invoked shell is a fresh one). for _l in (MOON / ".env").read_text().splitlines() if (MOON / ".env").exists() else []: if "=" in _l and not _l.strip().startswith("#"): _k, _v = _l.split("=", 1) os.environ.setdefault(_k.strip(), _v.strip()) ELEVEN_KEY = os.environ.get("ELEVENLABS_API_KEY", "") VOICE = json.loads((MOON / "voice.json").read_text()) if (MOON / "voice.json").exists() else {} _auth_cache = {} def agent_needs_auth(agent_id): """A public agent connects by id over webrtc (lower latency). A private one needs a signed URL, and the SDK only accepts those over websocket. Ask once so the client doesn't have to guess.""" if agent_id in _auth_cache: return _auth_cache[agent_id] need = False if ELEVEN_KEY: try: req = urllib.request.Request( f"https://api.elevenlabs.io/v1/convai/agents/{agent_id}", headers={"xi-api-key": ELEVEN_KEY}) with urllib.request.urlopen(req, timeout=15) as r: d = json.loads(r.read()) need = bool(((d.get("platform_settings") or {}).get("auth") or {}) .get("enable_auth")) except Exception as e: print(f" agent auth lookup failed ({e}) — assuming public", flush=True) _auth_cache[agent_id] = need return need def signed_url(agent_id): """Short-lived signed URL for a private agent. Public agents don't need one.""" if not ELEVEN_KEY: return {"error": "no_key"} url = ("https://api.elevenlabs.io/v1/convai/conversation/get-signed-url" "?agent_id=" + agent_id) try: req = urllib.request.Request(url, headers={"xi-api-key": ELEVEN_KEY}) with urllib.request.urlopen(req, timeout=15) as r: return {"signedUrl": json.loads(r.read()).get("signed_url", "")} except Exception as e: return {"error": f"signed url request failed: {e}"} class Handler(http.server.SimpleHTTPRequestHandler): def __init__(self, *a, **kw): super().__init__(*a, directory=str(ROOT), **kw) def log_message(self, fmt, *args): if "/api/" in (self.path or ""): print(f" {self.path[:90]}", flush=True) def _json(self, obj, code=200): b = json.dumps(obj).encode() self.send_response(code) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(b))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(b) def send_head(self): """Range support. Without it, seeking into a 280MB episode makes the browser fetch from byte zero — so "keep listening" would stall for minutes. SimpleHTTPRequestHandler has no Range handling of its own.""" rng = self.headers.get("Range") if not rng or not rng.startswith("bytes="): return super().send_head() path = self.translate_path(self.path) if not os.path.isfile(path): return super().send_head() size = os.path.getsize(path) try: first, _, last = rng[6:].partition("-") start = int(first) if first else 0 end = int(last) if last else size - 1 except ValueError: return super().send_head() start = max(0, start) end = min(end, size - 1) if start > end: self.send_response(416) self.send_header("Content-Range", f"bytes */{size}") self.end_headers() return None f = open(path, "rb") f.seek(start) self.send_response(206) self.send_header("Content-Type", self.guess_type(path)) self.send_header("Content-Range", f"bytes {start}-{end}/{size}") self.send_header("Content-Length", str(end - start + 1)) self.send_header("Accept-Ranges", "bytes") self.end_headers() return _Ranged(f, end - start + 1) def do_GET(self): u = urllib.parse.urlparse(self.path) if u.path == "/api/ask": qs = urllib.parse.parse_qs(u.query) q = (qs.get("q") or [""])[0].strip() if not q: return self._json({"error": "empty query"}, 400) try: k = int((qs.get("k") or ["16"])[0]) return self._json({"q": q, **answer(q, k)}) except Exception as e: import traceback; traceback.print_exc() return self._json({"error": f"{type(e).__name__}: {e}"}, 500) if u.path == "/api/voice": aid = VOICE.get("agentId", "") return self._json({"agentId": aid, "hasKey": bool(ELEVEN_KEY), "needsAuth": agent_needs_auth(aid) if aid else False}) if u.path == "/api/signed-url": aid = (urllib.parse.parse_qs(u.query).get("agent") or [VOICE.get("agentId", "")])[0] if not aid: return self._json({"error": "no agent configured"}, 400) return self._json(signed_url(aid)) if u.path == "/api/health": return self._json({"ok": True, "chunks": len(CHUNK_IDS), "model_loaded": _model is not None, "agent": bool(VOICE.get("agentId")), "key": bool(ELEVEN_KEY)}) return super().do_GET() class _Ranged: """File wrapper that stops after n bytes, for copyfile().""" def __init__(self, f, n): self.f, self.left = f, n def read(self, size=-1): if self.left <= 0: return b"" if size is None or size < 0: size = self.left data = self.f.read(min(size, self.left)) self.left -= len(data) return data def close(self): self.f.close() class Server(socketserver.ThreadingMixIn, socketserver.TCPServer): allow_reuse_address = True daemon_threads = True def lan_ip(): try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(("8.8.8.8", 80)); ip = s.getsockname()[0]; s.close() return ip except Exception: return "127.0.0.1" if __name__ == "__main__": with Server(("0.0.0.0", PORT), Handler) as httpd: print(f"\n http://127.0.0.1:{PORT}/ask.html") print(f" http://{lan_ip()}:{PORT}/ask.html (phone on the same wifi)\n", flush=True) httpd.serve_forever()