File size: 4,666 Bytes
dde517d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 | """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})
|