Initial commit

Last3Mon/64/workflow-lab/src/miniflow/tools.py

@@ -0,0 +1,62 @@
+"""Tools a workflow can call. Each tool reads local sample data, so no network is needed."""
+
+import json
+from collections.abc import Callable
+from pathlib import Path
+from typing import Any
+
+DATA_DIR = Path(__file__).resolve().parents[2] / "data"
+
+
+class ToolError(RuntimeError):
+    pass
+
+
+def _load(name: str) -> Any:
+    return json.loads((DATA_DIR / name).read_text(encoding="utf-8"))
+
+
+def get_weather(city: str) -> dict[str, Any]:
+    """Return today's sample weather for a city."""
+    table = _load("weather.json")
+    if city not in table:
+        raise ToolError(f"Unknown city '{city}'. Available: {', '.join(table)}")
+    return {"city": city, **table[city]}
+
+
+def search_faq(query: str, top_k: int = 2) -> list[dict[str, Any]]:
+    """Return FAQ entries ranked by keyword overlap with the query."""
+    words = {word for word in query.lower().split() if len(word) > 1}
+    scored = []
+    for item in _load("faq.json"):
+        keywords = {keyword.lower() for keyword in item["keywords"]}
+        score = sum(1 for word in words if any(keyword in word or word in keyword for keyword in keywords))
+        if score:
+            scored.append({"score": score, **item})
+    scored.sort(key=lambda item: item["score"], reverse=True)
+    return scored[: int(top_k)]
+
+
+def count_chars(text: str) -> dict[str, int]:
+    """Count characters and words of a text."""
+    return {"chars": len(text), "words": len(text.split())}
+
+
+TOOLS: dict[str, Callable[..., Any]] = {
+    "get_weather": get_weather,
+    "search_faq": search_faq,
+    "count_chars": count_chars,
+}
+
+
+def call_tool(name: str, args: dict[str, Any]) -> Any:
+    if name not in TOOLS:
+        raise ToolError(f"Tool '{name}' is not registered. Available: {', '.join(TOOLS)}")
+    try:
+        return TOOLS[name](**args)
+    except TypeError as error:
+        raise ToolError(f"Wrong arguments for tool '{name}': {error}") from error
+
+
+def describe_tools() -> list[tuple[str, str]]:
+    return [(name, (func.__doc__ or "").strip()) for name, func in TOOLS.items()]