| """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 |
|
|
| |
| 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" |
|
|
|
|
| |
| _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: |
| 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 |
| |
| 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() |
|
|
