"""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"(?