[UPDATE] Restructure release: safetensors weights, cbt package, chat demo, clean model card
dde517d verified | """Offline stand-ins for the tools CBT was fine-tuned to call. Replace `execute_tool` with real backends as needed.""" | |
| import ast | |
| import json | |
| import operator | |
| import re | |
| from typing import Any, Dict | |
| _OPS = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv, | |
| ast.USub: operator.neg, ast.UAdd: operator.pos} | |
| _KNOWLEDGE: Dict[str, str] = { | |
| "tomato": "the edible berry of the plant Solanum lycopersicum, commonly prepared as a culinary vegetable.", | |
| "apple": "a round, edible fruit produced by an apple tree (Malus domestica), cultivated worldwide.", | |
| "potato": "a starchy, tuberous root crop of the nightshade family Solanaceae.", | |
| "coffee": "a brewed drink prepared from roasted coffee beans of flowering Coffea species.", | |
| "banana": "an elongated, edible fruit produced by several kinds of large herbaceous plants in the genus Musa.", | |
| "gravity": "a fundamental interaction causing mutual attraction between all things with mass or energy.", | |
| "photosynthesis": "a process used by plants to convert light energy into chemical energy stored in glucose.", | |
| "atom": "the basic unit of a chemical element, consisting of a dense nucleus surrounded by electrons.", | |
| "speed of light": "approximately 299,792 kilometers per second in a vacuum.", | |
| "mount everest": "Earth's highest mountain above sea level, located in the Himalayas.", | |
| } | |
| _UNITS = {("km", "miles"): 0.621371, ("miles", "km"): 1.60934, ("kg", "lbs"): 2.20462, ("lbs", "kg"): 0.453592} | |
| def _calc(node: ast.AST) -> Any: | |
| if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)): | |
| return node.value | |
| if isinstance(node, ast.BinOp) and type(node.op) in _OPS: | |
| return _OPS[type(node.op)](_calc(node.left), _calc(node.right)) | |
| if isinstance(node, ast.UnaryOp) and type(node.op) in _OPS: | |
| return _OPS[type(node.op)](_calc(node.operand)) | |
| raise ValueError("unsupported expression") | |
| def _search(query: str, online: bool) -> Dict[str, str]: | |
| q = query.lower().rstrip("s") | |
| for key, text in _KNOWLEDGE.items(): | |
| if key in q or q in key: | |
| return {"snippet": text} | |
| if online: | |
| try: | |
| import urllib.parse | |
| import urllib.request | |
| url = "https://en.wikipedia.org/api/rest_v1/page/summary/" + urllib.parse.quote(query.replace(" ", "_")) | |
| req = urllib.request.Request(url, headers={"User-Agent": "CBT-demo/1.0"}) | |
| with urllib.request.urlopen(req, timeout=3) as resp: | |
| extract = json.loads(resp.read().decode("utf-8")).get("extract") | |
| if extract: | |
| return {"snippet": extract[:220]} | |
| except Exception: | |
| pass | |
| return {"snippet": f"{query} is a topic in verified knowledge sources."} | |
| def execute_tool(call: str, online: bool = False) -> str: | |
| """Returns a JSON string for a call such as `get_weather(city="Tokyo")`.""" | |
| name = call.split("(", 1)[0].strip() | |
| args = dict(re.findall(r'(\w+)="([^"]*)"', call)) | |
| if name == "calculator": | |
| try: | |
| return json.dumps({"result": _calc(ast.parse(args.get("expr", ""), mode="eval").body)}) | |
| except ZeroDivisionError: | |
| return json.dumps({"error": "ZeroDivisionError: division by zero"}) | |
| except Exception as e: | |
| return json.dumps({"error": str(e)}) | |
| if name == "get_weather": | |
| city = args.get("city", "") | |
| if city.lower() in ("atlantis", "narnia"): | |
| return json.dumps({"status": "error", "code": 404, "message": f"City '{city}' not found"}) | |
| return json.dumps({"temp": 22, "condition": "Sunny", "city": city}) | |
| if name == "get_time": | |
| return json.dumps({"time": "14:30", "timezone": args.get("timezone", "UTC")}) | |
| if name == "unit_convert": | |
| m = re.search(r"value=([\d.]+)", call) | |
| value = float(m.group(1)) if m else 0.0 | |
| to = args.get("to_unit", "") | |
| return json.dumps({"value": round(value * _UNITS.get((args.get("from_unit", ""), to), 1.0), 2), "unit": to}) | |
| if name == "search": | |
| return json.dumps(_search(args.get("query", ""), online)) | |
| if name == "database_query": | |
| return json.dumps({"status": "success", "count": 3, "records": [{"id": 101, "status": "active"}]}) | |
| if name == "create_event": | |
| return json.dumps({"event_id": "evt_98214", "status": "confirmed"}) | |
| if name == "file_manager": | |
| return json.dumps({"status": "ok", "bytes_written": 64}) | |
| if name == "fetch_api": | |
| return json.dumps({"status": 200, "data": {"status": "healthy"}}) | |
| return json.dumps({"status": "executed", "call": call}) | |