""" LegacyScribe — Family Memory Agent Gradio Space entry point """ import gradio as gr import html as _html import json import os import re import time from pathlib import Path from gradio_client import Client # # ── Model loading ───────────────────────────────────────────────────────────── # MODEL_REPO = os.environ.get("MODEL_REPO", "build-small-hackathon/legacyscribe-Qwen3.5-9B.Q4_K_M") # MODEL_FILE = os.environ.get("MODEL_FILE", "Qwen3.5-9B.Q4_K_M.gguf") # N_CTX = 4096 # N_GPU_LAYERS = int(os.environ.get("N_GPU_LAYERS", "0")) BACKEND_URL = os.environ.get("BACKEND_URL", "https://swohamkayastha--legacystribe-backend-legacyscribeserver-predict.modal.run/") # print(f"Loading model from {MODEL_REPO}/{MODEL_FILE}...") # model_path = hf_hub_download(repo_id=MODEL_REPO, filename=MODEL_FILE) # llm = Llama(model_path=model_path, n_ctx=N_CTX, n_gpu_layers=N_GPU_LAYERS, verbose=False) # print("Model ready.") _client = None def get_client(): global _client if _client is None: _client = Client(BACKEND_URL) return _client # ── System prompts ──────────────────────────────────────────────────────────── SYSTEM_PROMPTS = { "questioner": ( "You are a gentle memory guide helping an elderly person tell their life story. " "Ask exactly one warm, open follow-up question. Never ask more than one question. " "Be patient, kind, and culturally sensitive to Nepali and South Asian contexts." ), "extractor": ( "You are an extractor agent. Given a memory fragment, extract structured information " "as JSON with keys relevant to the content (who, when, where, what, emotion). " "Output only valid JSON, nothing else." ), "arcdetector": ( "You are an arc detector agent. Given a memory fragment, identify the narrative stage. " "Output one word only: setup, tension, turn, or meaning." ), "publisher": ( "You are a publisher agent. Given memory notes, synthesize them into a single warm, " "narrative paragraph suitable for a family memory book. Write in first person. " "Use natural, unhurried language. Output only the paragraph, nothing else." ), } def _parse_think(raw: str): """Split raw LLM output into (thinking, output). Handles two formats the backend may emit: 1. \n\noutput (standard CoT tag) 2. Thinking Process:\n1. …\n\noutput (tag-free fallback) """ raw = raw.strip() if "" in raw: parts = raw.split("", 1) return parts[0].replace("", "").strip(), parts[1].strip() if re.match(r"^Thinking Process:", raw): # Split at first blank line NOT followed by a numbered list item parts = re.split(r"\n\n(?!\d+[\.\)])", raw, maxsplit=1) if len(parts) > 1: return parts[0].strip(), parts[1].strip() return "", raw import requests def call_agent(agent: str, user_text: str, max_tokens: int = -1, temp: float = -1.0): """Returns (output, thinking) — thinking is '' when no block present.""" try: print(f"Calling backend: agent={agent} url={BACKEND_URL}") # 1. Package the variables into a clean JSON dictionary matching the backend payload = { "agent": agent, "user_text": user_text, "max_tokens": int(max_tokens), "temp": float(temp) } # 2. Fire a real HTTP POST request with a generous timeout for cold-starts response = requests.post(BACKEND_URL, json=payload, timeout=300) # 3. Parse the JSON response back out response_data = response.json() if "error" in response_data: print(f"Backend returned an internal error [{agent}]: {response_data['error']}") return "", "" result = response_data.get("response", "") # 4. Pass the string text right back into your existing parser thinking, output = _parse_think(result or "") print(f"Backend output [{agent}]: {output[:80]}") return output, thinking except Exception as e: print(f"Backend error [{agent}]: {type(e).__name__}: {e}") return "", "" # ── Session state ───────────────────────────────────────────────────────────── def init_state(): return {"turns": [], "all_notes": [], "chapters": [],"chapter_counter":1, "notes_consumed_idx":0, "chapter_num": 1, "current_chapter_idx": -1, } MIN_NOTES_PER_CHAPTER = 3 def process_turn(user_input: str, state: dict): if not user_input.strip(): return state, "", "", "", [] t0 = time.time() think_steps = [] # --- arc detection --- arc_out, arc_think = call_agent("arcdetector", user_input, max_tokens=-1, temp=0.1) arc = arc_out.lower().strip() if arc not in ("setup", "tension", "turn", "meaning"): arc = "setup" think_steps.append({"t": "arc", "arc": arc, "label": ARC_LABELS.get(arc, arc), "thinking": arc_think}) # --- extraction --- try: raw, ext_think = call_agent("extractor", user_input, max_tokens=-1, temp=0.1) notes = json.loads(raw) note_lines = [f"{k}: {v}" for k, v in notes.items() if v] except Exception: ext_think = "" note_lines = [user_input[:120]] think_steps.append({"t": "extract", "items": note_lines, "thinking": ext_think}) # add new notes to the global pool state["all_notes"].extend(note_lines) # --- generate new chapter if enough fresh notes have arrived --- new_note_count = len(state["all_notes"]) - state["notes_consumed_idx"] if new_note_count >= MIN_NOTES_PER_CHAPTER: # use the last 6 notes for continuity (but we will only generate one new chapter) notes_text = "\n".join( f"{i+1}. {n}" for i, n in enumerate(state["all_notes"][-6:]) ) chapter_text, pub_think = call_agent("publisher", f"Notes:\n{notes_text}", max_tokens=-1, temp=0.4) if chapter_text and chapter_text.strip(): state["chapters"].append(chapter_text) # advance the counter: number of chapters we now have state["chapter_counter"] = len(state["chapters"]) + 1 state["notes_consumed_idx"] = len(state["all_notes"]) if len(state["chapters"]) == 1: state["current_chapter_idx"] = 0 think_steps.append({ "t": "chapter", "num": len(state["chapters"]), "thinking": pub_think, }) # --- generate follow-up question (uses last 3 turns) --- history_ctx = "\n".join(f"Memory: {t['user']}" for t in state["turns"][-3:]) question_ctx = f"{history_ctx}\nMemory: {user_input}" if history_ctx else f"Memory: {user_input}" question, q_think = call_agent("questioner", question_ctx, max_tokens=-1, temp=0.7) think_steps.append({"t": "question", "text": question, "thinking": q_think}) elapsed = round(time.time() - t0, 1) state["turns"].append({ "user": user_input, "question": question, "arc": arc, "notes": note_lines, "think_secs": elapsed, "think_steps": think_steps, }) return state, question, arc, state["chapters"][-1] if state["chapters"] else "", note_lines #---- def build_page_from_chapter(chapters, idx): """Return HTML for the page showing chapter at index idx (no inline scripts).""" if not chapters or idx < 0 or idx >= len(chapters): # Empty / waiting state return '''
— waiting for your story —
Chapter One
Your memories are gathering.
Share a few more and I will begin to write.
''' chapter_text = chapters[idx] chapter_num = idx + 1 words = chapter_text.split() lines, current = [], [] chars = 0 for w in words: if chars + len(w) + 1 > 52: lines.append(" ".join(current)) current, chars = [w], len(w) else: current.append(w) chars += len(w) + 1 if current: lines.append(" ".join(current)) text_html = "".join(f'
{line}
' for line in lines) # No scripts inside! Only the book HTML. return f'''
Chapter {chapter_num}
{text_html}
''' # ── HTML builders ───────────────────────────────────────────────────────────── ARC_ICONS = {"setup": "◎", "tension": "◈", "turn": "◉", "meaning": "✦"} ARC_LABELS = {"setup": "Setting the scene", "tension": "Something changed", "turn": "A new direction", "meaning": "What it meant"} def build_chat_html(turns): if not turns: return '''
📖
Tell me a memory — anything at all.
A meal, a person, a festival, a place.
I am here to listen.
''' html = '
' for t in turns: arc = t.get("arc", "setup") icon = ARC_ICONS.get(arc, "◎") label = ARC_LABELS.get(arc, arc) secs = t.get("think_secs", "") steps = t.get("think_steps", []) def _think_block(thinking, output_html): out = "" if thinking: escaped = _html.escape(thinking).replace("\n", "
") out += f'
{escaped}
' out += f'
{output_html}
' return out steps_html = "" for s in steps: if s["t"] == "arc": inner = _think_block( s.get("thinking", ""), f'output {_html.escape(s["arc"])}  ·  {_html.escape(s["label"])}', ) steps_html += f'
arc detector
{inner}
' elif s["t"] == "extract": pairs = "".join( f'
' f'{_html.escape(p.split(": ",1)[0])}' f'{_html.escape(p.split(": ",1)[1] if ": " in p else p)}' f'
' for p in s["items"] ) inner = _think_block(s.get("thinking", ""), f'output{pairs}') steps_html += f'
memory extractor
{inner}
' elif s["t"] == "question": inner = _think_block( s.get("thinking", ""), f'output{_html.escape(s["text"])}', ) steps_html += f'
follow-up composer
{inner}
' elif s["t"] == "chapter": inner = _think_block( s.get("thinking", ""), f'output chapter {s["num"]} drafted to notebook', ) steps_html += f'
publisher
{inner}
' thought_html = f'''
Thought for {secs}s ›
{steps_html}
''' if secs else "" html += f'''
{t["user"]}
{icon}{label}
{thought_html}
{t["question"]}
''' html += '
' html += '' return html def build_notes_html(notes): if not notes: return '' html = '
Memory fragments extracted
' for n in notes[-5:]: parts = n.split(": ", 1) if len(parts) == 2: html += f'
{parts[0]}{parts[1]}
' else: html += f'
{n}
' html += '
' return html LINE_H = 28 # px per ruled line def build_page_html(chapters, chapter_counter): if not chapters: return '''
— waiting for your story —
Chapter One
Your memories are gathering.
Share a few more and I will begin to write.
''' # take the most recent chapter chapter_text = chapters[-1] chapter_num = len(chapters) # chapter number = position in list # same line‑breaking logic as before words = chapter_text.split() lines, current = [], [] chars = 0 for w in words: if chars + len(w) + 1 > 52: lines.append(" ".join(current)) current, chars = [w], len(w) else: current.append(w) chars += len(w) + 1 if current: lines.append(" ".join(current)) text_html = "".join(f'
{line}
' for line in lines) return f'''
Chapter {chapter_num}
{text_html}
''' # ── CSS / HEAD ───────────────────────────────────────────────────────────────── HEAD = """ { /* Pure animation — no interaction. Just remove overlay from DOM after CSS is done. */ /* Total: 3.5s delay + 0.9s fade = 4.4s. Add small buffer. */ setTimeout(function() { var el = document.getElementById('ls-intro'); if (el) el.style.display = 'none'; }, 4600); } """ # ── Gradio app ───────────────────────────────────────────────────────────────── with gr.Blocks() as demo: gr.HTML(f"") state = gr.State(init_state()) # ── Book intro overlay — pure CSS animation, no interaction ───────────────── gr.HTML("""
LegacyScribe
Every family has a story worth keeping
✦ · ✦
LegacyScribe
A Family Memory Journal
· 2025 ·
Family Memory Agent
""") # ── Header ── gr.HTML("""
LegacyScribe.
Every family has a story worth keeping
Off the Grid Qwen3.5 · 9B LoRA Fine-tuned Nepali · English 5-Agent Pipeline Build Small 2025
""") # ── Persistent swipe & flip script (runs once, watches DOM changes) ── gr.HTML(''' ''') def prev_chapter(state): idx = state.get("current_chapter_idx", -1) if idx > 0: state["current_chapter_idx"] = idx - 1 page_html = build_page_from_chapter(state["chapters"], state.get("current_chapter_idx", -1)) indicator_html = update_indicator(state) return state, page_html, indicator_html def next_chapter(state): idx = state.get("current_chapter_idx", -1) chapters = state.get("chapters", []) if idx < len(chapters) - 1: state["current_chapter_idx"] = idx + 1 page_html = build_page_from_chapter(state["chapters"], state.get("current_chapter_idx", -1)) indicator_html = update_indicator(state) return state, page_html, indicator_html def update_indicator(state): chapters = state.get("chapters", []) idx = state.get("current_chapter_idx", -1) if chapters and 0 <= idx < len(chapters): return f'
Chapter {idx+1} of {len(chapters)}
' return '
No chapters yet
' # ── Main two-column layout ── with gr.Row(elem_classes=["ls-main"]): # Left — conversation with gr.Column(elem_classes=["ls-conversation"]): gr.HTML('') chat_display = gr.HTML(build_chat_html([])) notes_display = gr.HTML("") with gr.Column(elem_classes=["ls-input-wrap"]): user_input = gr.Textbox( placeholder="Tell me about a person, a festival, a meal, a place — anything you remember…", lines=3, show_label=False, container=False, ) with gr.Row(elem_classes=["ls-btn-row"]): submit_btn = gr.Button("Share this memory →", elem_classes=["ls-btn", "ls-btn-primary"]) clear_btn = gr.Button("Start over", elem_classes=["ls-btn", "ls-btn-secondary"]) thinking = gr.HTML( '' ) # Right — memory book with gr.Column(elem_classes=["ls-book"]): gr.HTML('') page_display = gr.HTML(build_page_from_chapter([], -1)) # Navigation row with gr.Row(): prev_btn = gr.Button("◀ Previous", elem_classes=["ls-nav-btn"], size="sm") chapter_label = gr.HTML('
Chapter —
') next_btn = gr.Button("Next ▶", elem_classes=["ls-nav-btn"], size="sm") # Download section (two-step but reliable) with gr.Row(): download_btn = gr.Button("↓ Download memory book", elem_classes=["ls-dl-btn"]) download_file = gr.File(label="Your memory book", visible=False, interactive=False) # Functions for download and indicator def export_memory_fn(state): import tempfile from pathlib import Path chapters = state.get("chapters", []) all_notes = state.get("all_notes", []) content = "# My Memory Book\n\n" if chapters: for idx, ch in enumerate(chapters, start=1): content += f"## Chapter {idx}\n\n{ch}\n\n---\n\n" else: content += "## (No chapters yet)\n\nKeep sharing memories — I will write your story.\n\n---\n\n" if all_notes: content += "### Memory fragments\n\n" + "\n".join(f"- {n}" for n in all_notes) else: content += "### Memory fragments\n\n(Start a conversation to see fragments appear.)" path = Path(tempfile.gettempdir()) / "legacyscribe_memory.txt" path.write_text(content, encoding="utf-8") return str(path) def on_download(state): path = export_memory_fn(state) return gr.update(visible=True, value=path) download_btn.click(fn=on_download, inputs=[state], outputs=[download_file]) # ── Footer ── gr.HTML(""" """) prev_btn.click(fn=prev_chapter, inputs=[state], outputs=[state, page_display, chapter_label]) next_btn.click(fn=next_chapter, inputs=[state], outputs=[state, page_display, chapter_label]) # ── Handlers ── def on_submit(user_text, state): if not user_text.strip(): return (state, build_chat_html(state["turns"]), "", build_page_from_chapter(state["chapters"], state.get("current_chapter_idx", -1)), update_indicator(state), "") new_state, question, arc, chapter, notes = process_turn(user_text, state) return (new_state, build_chat_html(new_state["turns"]), build_notes_html(notes), build_page_from_chapter(new_state["chapters"], new_state.get("current_chapter_idx", -1)), update_indicator(new_state), "") def on_clear(): s = init_state() return (s, build_chat_html([]), "", build_page_from_chapter([], -1), update_indicator(s), "") submit_btn.click( fn=on_submit, inputs=[user_input, state], outputs=[state, chat_display, notes_display, page_display, chapter_label, user_input], ) user_input.submit( fn=on_submit, inputs=[user_input, state], outputs=[state, chat_display, notes_display, page_display, chapter_label, user_input], ) clear_btn.click( fn=on_clear, inputs=[], outputs=[state, chat_display, notes_display, page_display, chapter_label, user_input], ) print("=== Launching Gradio demo ===") demo.launch(ssr_mode=False)