| from __future__ import annotations |
|
|
| import argparse |
| import importlib |
| import json |
| import tempfile |
| from pathlib import Path |
|
|
|
|
| def main() -> None: |
| parser = argparse.ArgumentParser() |
| parser.add_argument("--base-url", default="http://127.0.0.1:8007") |
| parser.add_argument("--model", default="local-model") |
| args = parser.parse_args() |
|
|
| package = "hcl" |
| models = importlib.import_module(f"{package}.models") |
| router_module = importlib.import_module(f"{package}.router") |
| task_interface_module = importlib.import_module(f"{package}.task_interface") |
|
|
| model = models.ChatCompletionsAPIModel( |
| name=args.model, |
| base_url=args.base_url, |
| api_key="EMPTY", |
| max_new_tokens=512, |
| answer_max_new_tokens=128, |
| temperature=0.0, |
| system_prompt=( |
| "You are a concise reasoning assistant. Use successful tool evidence " |
| "and return only the requested final answer." |
| ), |
| timeout_seconds=300, |
| max_retries=2, |
| cleanup_output=True, |
| enable_image_input=False, |
| chat_template_kwargs={"enable_thinking": False}, |
| json_object_for_structured_phases=False, |
| ) |
| task_config = task_interface_module.TaskInterfaceConfig(mode="rule") |
| task_interface = task_interface_module.TaskInterface(task_config) |
| example = { |
| "task_id": "tool_smoke/mercury", |
| "task_name": "tool_smoke", |
| "task_type": "math_word_problem", |
| "question": ( |
| "Mercury completes one revolution in 88 Earth days. " |
| "Using 365.25 days per Earth year, how many Earth years is that? " |
| "Return only a decimal rounded to two decimal places." |
| ), |
| "answer": "0.24", |
| } |
| chunk = task_interface.build_taskinterfacechunk(example, 0) |
|
|
| with tempfile.TemporaryDirectory(prefix="hcl_tool_smoke_") as temp_dir: |
| cache_root = Path(temp_dir) |
| router = router_module.Router( |
| model, |
| router_module.RouterConfig( |
| route_name="router_managed_arithmetic_smoke", |
| use_llm_workflow_selector=True, |
| use_memory_selector=False, |
| memory_always_on=False, |
| use_skill_selector=False, |
| use_tool_selector=True, |
| tool_auto_select=True, |
| use_chat_tool_calls=False, |
| tool_max_rounds=0, |
| workflow_selector_cache_path=str(cache_root / "workflow.jsonl"), |
| tool_selector_cache_path=str(cache_root / "tool_selection.jsonl"), |
| tool_argument_cache_path=str(cache_root / "tool_arguments.jsonl"), |
| final_generation_cache_path=str(cache_root / "final.jsonl"), |
| capability_enabled=True, |
| capability_execution_mode="router_managed", |
| semantic_search_enabled=False, |
| cross_modal_match_enabled=False, |
| ), |
| ) |
| result = router.run(chunk, task_config) |
|
|
| trace = result["trace"] |
| capability_results = trace.get("capability_results") or [] |
| if len(capability_results) != 1: |
| raise SystemExit(f"expected one capability result, got: {capability_results!r}") |
| capability = capability_results[0] |
| if capability.get("tool_name") != "arithmetic_calculator": |
| raise SystemExit(f"wrong selected capability: {capability!r}") |
| if capability.get("status") != "success": |
| raise SystemExit(f"capability did not execute successfully: {capability!r}") |
| arguments = capability.get("arguments") or {} |
| output = capability.get("output") or {} |
| numeric_result = float(output.get("numeric_result")) |
| if abs(numeric_result - (88 / 365.25)) > 1e-12: |
| raise SystemExit(f"wrong calculator result: {capability!r}") |
| if "0.240930869" not in str(trace.get("prompt", "")): |
| raise SystemExit("successful calculator result was not injected into final context") |
|
|
| print(json.dumps({ |
| "status": "passed", |
| "selected_tools": [item.get("tool_name") for item in trace.get("selected_tools", [])], |
| "capability": capability, |
| "final_answer": result.get("answer"), |
| }, ensure_ascii=False, indent=2)) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
