app.py

"""topoloop Finch demo: the 0.8B agent answering a travel question with live tool calls over the topoloop API.

Runs the exact episode loop from geogym (core.harness.run_question) with the model served in-process.
"""
import asyncio
import json
import os
import queue
import re
import threading
import time
import types

# The flags Finch was trained and evaluated with. Must be set before the harness is imported.
for k, v in {"BENCH_CONTRACT": "v2", "BENCH_TWO_TOOL": "1", "BENCH_BLOG_TOOLS": "1", "BENCH_SEARCH_BUDGET": "3",
             "BENCH_WEB_BUDGET": "2", "BENCH_FETCH_BUDGET": "1", "BENCH_LOCALITY": "1", "BENCH_CHUNK_CHARS": "2600",
             "WPLACES_SEARCH_RADIUS_KM": "300", "BENCH_NO_JUDGE": "1", "BENCH_NO_TEMPLATE_KWARGS": "1"}.items():
    os.environ.setdefault(k, v)
if os.environ.get("TOPOLOOP_API_KEY"):
    base = os.environ.get("TOPOLOOP_API_BASE", "https://api.topoloop.com").rstrip("/")
    os.environ.setdefault("WPLACES_API_ENDPOINT", base)
    os.environ.setdefault("WPLACES_API_KEY", os.environ["TOPOLOOP_API_KEY"])
    os.environ.setdefault("WSEARCH_EP", base)

import gradio as gr
import httpx
import spaces
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer

from geogym_blogq.lab.core import harness as H

MODEL_ID = os.environ.get("FINCH_MODEL", "topoloop/finch-0.1-0.8B")
tok = AutoTokenizer.from_pretrained(MODEL_ID)
model = AutoModelForCausalLM.from_pretrained(MODEL_ID, dtype=torch.bfloat16)
model.eval()
try:
    model.to("cuda")
except Exception:
    pass

EXAMPLES = [
    "Where can I get soft serve ice cream to eat by the Cannon River in Northfield?",
    "Is there a quiet, air-conditioned cafe in Tulum pueblo that also has a selection of books?",
    "Where in Tsim Sha Tsui can I watch the sunset over the Hong Kong Island skyline?",
    "I'm looking for a nature reserve in Bermuda where I can go snorkeling, where is this?",
    "A restaurant in Bratislava's old town with traditional Slovak folk music in a cellar near Michael's Gate?",
]


def _render(messages, tools):
    msgs = []
    for m in messages:
        m = dict(m)
        if m.get("tool_calls"):
            calls = []
            for tc in m["tool_calls"]:
                fn = dict(tc.get("function") or {})
                if isinstance(fn.get("arguments"), str):
                    try:
                        fn["arguments"] = json.loads(fn["arguments"])
                    except Exception:
                        pass
                calls.append({"id": tc.get("id"), "type": "function", "function": fn})
            m["tool_calls"] = calls
        msgs.append(m)
    return tok.apply_chat_template(msgs, tools=tools, add_generation_prompt=True, tokenize=False, enable_thinking=False)


@spaces.GPU(duration=90)
def _generate(prompt, max_new_tokens, temperature):
    ids = tok(prompt, return_tensors="pt").to(model.device)
    with torch.no_grad():
        out = model.generate(**ids, max_new_tokens=max_new_tokens, do_sample=temperature > 0,
                             temperature=temperature if temperature > 0 else None, pad_token_id=tok.eos_token_id)
    new = out[0][ids["input_ids"].shape[1]:]
    return tok.decode(new, skip_special_tokens=True), ids["input_ids"].shape[1], len(new)


class _Completions:
    """The one method of the OpenAI client the harness uses, backed by the in-process model."""

    async def create(self, model, messages, tools=None, temperature=0.0, max_tokens=1500, extra_body=None):
        prompt = _render(messages, tools)
        loop = asyncio.get_event_loop()
        text, n_in, n_out = await loop.run_in_executor(None, _generate, prompt, max_tokens, temperature)
        msg = types.SimpleNamespace(content=text, tool_calls=None, role="assistant")
        return types.SimpleNamespace(choices=[types.SimpleNamespace(message=msg, finish_reason="stop", index=0)],
                                     usage=types.SimpleNamespace(prompt_tokens=n_in, completion_tokens=n_out))


class LocalLLM:
    def __init__(self):
        self.chat = types.SimpleNamespace(completions=_Completions())
        self.base_url = "local"


# Stream tool activity to the UI by wrapping the harness's tool functions.
_events = threading.local()


def _emit(kind, text):
    q = getattr(_events, "q", None)
    if q is not None:
        q.put((kind, text))


def _wrap_async(name, label):
    orig = getattr(H, name)

    async def w(http, *a, **kw):
        _emit("call", label(*a, **kw))
        r = await orig(http, *a, **kw)
        _emit("result", _summ(name, r))
        return r
    setattr(H, name, w)


def _summ(name, r):
    if name == "wp_search":
        names = [c.get("name") for c in r if isinstance(c, dict) and c.get("name")]
        return f"{len(names)} places: " + ", ".join(names[:6])
    if name == "wp_nearest":
        return f"{len(r)} nearby places"
    if name == "wp_details":
        return f"details for {', '.join(list(r)[:5])}"
    if name == "ws_search":
        heads = re.findall(r"^\[w\d+\] (.+)$", r or "", re.M)
        return f"{len(heads)} pages: " + "; ".join(h[:60] for h in heads[:4])
    return str(r)[:120]


_wrap_async("wp_search", lambda query, lat, lng, constraints=None: f"search_places({query!r}, {lat:.3f}, {lng:.3f})")
_wrap_async("wp_nearest", lambda lat, lng, limit=10: f"nearest({lat:.3f}, {lng:.3f})")
_wrap_async("wp_details", lambda refs: f"place_details({', '.join(refs[:5])})")
_wrap_async("ws_search", lambda query, limit=5, handles=None: f"web_search({query!r})")
_orig_ws_page = H.ws_page


def _ws_page(url):
    _emit("call", f"web_fetch({url})")
    r = _orig_ws_page(url)
    _emit("result", f"{len(r)} chars")
    return r


H.ws_page = _ws_page

ARGS = types.SimpleNamespace(model="finch", temperature=0.0, max_tokens=1500, max_tools=8, judge_model="")


def _run(question, q):
    _events.q = q
    try:
        from geogym_blogq.lab.core.bank import normalize
        qrow = normalize({"id": "demo", "category": "demo", "question": question, "gold": {}, "source": {}})

        async def go():
            async with httpx.AsyncClient(timeout=httpx.Timeout(60.0, connect=10.0)) as http:
                sem = asyncio.Semaphore(1)
                return await H.run_question(LocalLLM(), None, http, qrow, ARGS, sem)
        row = asyncio.new_event_loop().run_until_complete(go())
        q.put(("done", row))
    except Exception as e:  # noqa: BLE001
        q.put(("error", f"{type(e).__name__}: {e}"))


def _final_markdown(row):
    final = row.get("final") or ""
    m = re.search(r"\{\s*\"map\".*\}\s*$", final, re.S)
    prose = final[: m.start()].rstrip() if m else final
    handles = {}
    for msg in row.get("messages") or []:
        if msg.get("role") == "tool":
            for h, u in re.findall(r"^(w\d+): (https?://\S+)$", msg.get("content") or "", re.M):
                handles[h] = u
    # the model cites as [site](wN); turn the handle into the page URL
    prose = re.sub(r"\]\((w\d+)\)", lambda mm: "](" + handles.get(mm.group(1), "#") + ")", prose)
    prose = re.sub(r"(?<!\]\()\[(w\d+)\](?!\()", lambda mm: f"[{mm.group(1)}]({handles.get(mm.group(1), '#')})", prose)
    refs = []
    if m:
        try:
            refs = list(dict.fromkeys(n.get("ref") for n in json.loads(m.group(0)).get("map", []) if n.get("ref")))
        except Exception:
            pass
    tail = ("\n\nPinned: " + ", ".join(f"[{r}](https://www.openstreetmap.org/{ {'N': 'node', 'W': 'way', 'R': 'relation'}.get(r[0], 'node') }/{r[1:]})" for r in refs) + " on OpenStreetMap.") if refs else ""
    return prose + tail


def ask(question):
    question = (question or "").strip()
    if not question:
        yield "", "Type a question."
        return
    q = queue.Queue()
    t = threading.Thread(target=_run, args=(question, q), daemon=True)
    t.start()
    log, t0 = [], time.time()
    yield "", "thinking…"
    while True:
        try:
            kind, payload = q.get(timeout=0.5)
        except queue.Empty:
            if not t.is_alive():
                break
            continue
        if kind == "call":
            log.append(f"→ {payload}")
        elif kind == "result":
            log.append(f"   {payload}")
        elif kind == "error":
            log.append(f"error: {payload}")
            yield "\n".join(log), "Something went wrong."
            return
        elif kind == "done":
            row = payload
            log.append(f"✓ answered in {time.time() - t0:.0f}s, {row.get('n_tools', 0)} tool calls")
            yield "\n".join(log), _final_markdown(row)
            return
        yield "\n".join(log), "thinking…"


with gr.Blocks(title="topoloop Finch") as demo:
    gr.Markdown(
        "# topoloop Finch\n"
        "A 0.8B open-weights model that answers travel questions over OpenStreetMap and a travel-blog corpus, "
        "with every claim cited to something it actually retrieved. This is the same model and the same tool loop as "
        "[Ask Maps Basic](https://huggingface.co/datasets/topoloop/ask-maps-basic), where it ties Gemini 3.8 Flash. "
        "Tools come from [api.topoloop.com](https://topoloop.com/docs/); on a phone it runs fully offline. "
        "[Weights](https://huggingface.co/topoloop/finch-0.1-0.8B) · [Environment](https://github.com/topoloop-ai/geogym) · "
        "[How it was trained](https://jonready.com/blog/posts/rl-on-a-0_8b-travel-agent.html)"
    )
    with gr.Row():
        inp = gr.Textbox(label="Ask about a place", placeholder=EXAMPLES[0], scale=5)
        btn = gr.Button("Ask", variant="primary", scale=1)
    with gr.Row():
        log = gr.Textbox(label="Tool calls, live", lines=10, interactive=False)
        out = gr.Markdown(label="Answer")
    gr.Examples(EXAMPLES, inputs=inp)
    btn.click(ask, inputs=inp, outputs=[log, out])
    inp.submit(ask, inputs=inp, outputs=[log, out])

demo.queue(default_concurrency_limit=4).launch()