优化目录结构

This commit is contained in:
2026-07-03 20:49:36 +08:00
parent 04fefcf488
commit 8073dd63a7
10 changed files with 220 additions and 238 deletions
+1 -68
View File
@@ -1,12 +1,8 @@
import ast
import asyncio
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime
import json
import math
import operator
from typing import Any
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
from pydantic import BaseModel, ConfigDict, Field, ValidationError
@@ -18,12 +14,6 @@ class CurrentTimeArguments(BaseModel):
timezone: str = Field(default="UTC", description="IANA timezone, e.g. Asia/Shanghai")
class CalculatorArguments(BaseModel):
model_config = ConfigDict(extra="forbid")
expression: str = Field(description="Arithmetic expression to evaluate")
@dataclass(frozen=True, slots=True)
class ToolSpec:
name: str
@@ -59,13 +49,7 @@ class ToolRegistry:
description="Get the current date and time in an IANA timezone.",
arguments_model=CurrentTimeArguments,
handler=_get_current_time,
),
ToolSpec(
name="calculate",
description="Safely evaluate a basic arithmetic expression.",
arguments_model=CalculatorArguments,
handler=_calculate,
),
)
]
self._specs = {spec.name: spec for spec in specs}
@@ -118,54 +102,3 @@ def _get_current_time(arguments: BaseModel) -> dict[str, object]:
raise ValueError("unknown timezone") from exc
now = datetime.now(timezone)
return {"timezone": arguments.timezone, "datetime": now.isoformat()}
_BINARY_OPERATORS: dict[type[ast.operator], Callable[[Any, Any], Any]] = {
ast.Add: operator.add,
ast.Sub: operator.sub,
ast.Mult: operator.mul,
ast.Div: operator.truediv,
ast.FloorDiv: operator.floordiv,
ast.Mod: operator.mod,
ast.Pow: operator.pow,
}
_UNARY_OPERATORS: dict[type[ast.unaryop], Callable[[Any], Any]] = {
ast.UAdd: operator.pos,
ast.USub: operator.neg,
}
_MAX_EXPRESSION_LENGTH = 200
_MAX_ABSOLUTE_RESULT = 1e100
_MAX_EXPONENT = 100
def _calculate(arguments: BaseModel) -> dict[str, object]:
assert isinstance(arguments, CalculatorArguments)
expression = arguments.expression.strip()
if not expression or len(expression) > _MAX_EXPRESSION_LENGTH:
raise ValueError("expression is empty or too long")
tree = ast.parse(expression, mode="eval")
result = _evaluate_node(tree.body)
if isinstance(result, float) and not math.isfinite(result):
raise ValueError("result is not finite")
if abs(result) > _MAX_ABSOLUTE_RESULT:
raise ValueError("result is too large")
return {"expression": expression, "result": result}
def _evaluate_node(node: ast.AST) -> int | float:
if isinstance(node, ast.Constant):
if isinstance(node.value, bool) or not isinstance(node.value, (int, float)):
raise ValueError("only numeric constants are allowed")
return node.value
if isinstance(node, ast.UnaryOp) and type(node.op) in _UNARY_OPERATORS:
return _UNARY_OPERATORS[type(node.op)](_evaluate_node(node.operand))
if isinstance(node, ast.BinOp) and type(node.op) in _BINARY_OPERATORS:
left = _evaluate_node(node.left)
right = _evaluate_node(node.right)
if isinstance(node.op, ast.Pow) and abs(right) > _MAX_EXPONENT:
raise ValueError("exponent is too large")
result = _BINARY_OPERATORS[type(node.op)](left, right)
if abs(result) > _MAX_ABSOLUTE_RESULT:
raise ValueError("intermediate result is too large")
return result
raise ValueError("unsupported expression")