跳转至

Tool 系统

Tool 系统为 Agent 提供与外部世界交互的能力,包括搜索、抓取、Artifact 操作等。

模块结构

src/tools/
├── base.py              # 工具基类和权限定义
├── xml_parser.py        # XML 工具调用解析
├── xml_formatter.py     # XML 格式化(工具说明 + 结果序列化)
└── implementations/     # 具体工具实现
    ├── call_subagent.py
    ├── web_search.py
    ├── web_fetch.py
    └── artifact_ops.py

权限系统 (base.py)

ToolPermission

两级权限模型:

class ToolPermission(Enum):
    AUTO = "auto"        # 自动执行,无需用户确认
    CONFIRM = "confirm"  # 执行前需用户确认(通过 interrupt 暂停)

权限处理流程(在 engine.py_execute_tools 中):

flowchart TD
    A[获取工具权限] --> B{权限级别}
    B -->|AUTO| C[直接执行]
    B -->|CONFIRM| D[发送 PERMISSION_REQUEST]
    D --> E[interrupt 暂停]
    E --> F{用户响应}
    F -->|approved| C
    F -->|denied| G[返回拒绝结果]
    C --> H[发送 TOOL_COMPLETE]

ToolParameter

工具参数定义:

@dataclass
class ToolParameter:
    name: str                    # 参数名
    type: str                    # 类型:string, integer, boolean, array, object
    description: str             # 参数描述
    required: bool = True        # 是否必需
    default: Any = None          # 默认值

ToolResult

工具执行结果:

@dataclass
class ToolResult:
    success: bool                              # 执行是否成功
    data: Any = None                           # 结果数据
    error: Optional[str] = None                # 错误信息
    metadata: Dict[str, Any] = field(default_factory=dict)  # 元数据

    def to_dict(self) -> Dict[str, Any]:
        """转换为字典格式"""

BaseTool 基类

定义工具

class BaseTool(ABC):
    def __init__(
        self,
        name: str,
        description: str,
        permission: ToolPermission = ToolPermission.AUTO,
        **kwargs
    ):
        self.name = name
        self.description = description
        self.permission = permission
        self.config = kwargs

    @abstractmethod
    def get_parameters(self) -> List[ToolParameter]:
        """返回参数定义列表"""
        pass

    @abstractmethod
    async def execute(self, **params) -> ToolResult:
        """执行工具"""
        pass

    def validate_params(self, params: Dict[str, Any]) -> Optional[str]:
        """
        验证参数(可选覆盖)

        Returns:
            None 表示验证通过,字符串表示错误信息
        """
        # 默认实现:检查必需参数和未知参数
        ...

    async def __call__(self, **params) -> ToolResult:
        """使工具可调用(内部调用 validate_params 和 execute)"""

示例:WebSearchTool

class WebSearchTool(BaseTool):
    def __init__(self):
        super().__init__(
            name="web_search",
            description="搜索互联网获取信息",
            permission=ToolPermission.AUTO
        )

    def get_parameters(self) -> List[ToolParameter]:
        return [
            ToolParameter(
                name="query",
                type="string",
                description="搜索查询词",
                required=True
            ),
            ToolParameter(
                name="max_results",
                type="integer",
                description="最大结果数量",
                required=False,
                default=10
            )
        ]

    async def execute(self, query: str, max_results: int = 10, **kwargs) -> ToolResult:
        try:
            results = await self._do_search(query, max_results)
            return ToolResult(success=True, data=results)
        except Exception as e:
            return ToolResult(success=False, error=str(e))

工具管理

工具以 Dict[str, BaseTool] 管理,和 agents 一样是启动时加载的全局单例。

初始化流程

全局工具在 dependencies.py_load_tools() 中加载,请求级工具(artifact 操作)在 _create_controller() 中合并:

# dependencies.py — 启动时加载一次
_tools = {t.name: t for t in [CallSubagentTool(...), WebSearchTool(), WebFetchTool()]}

# chat.py — 每次请求合并
artifact_tools = create_artifact_tools(artifact_manager)
all_tools = {**get_tools(), **{t.name: t for t in artifact_tools}}
# 传入 ExecutionController → engine → context_manager

Engine 通过 tools.get(name) 查找工具,agent 工具白名单由 Agent MD 的 tools frontmatter 控制。

XML 解析器 (xml_parser.py)

工具调用格式

Agent 使用 XML 格式发起工具调用,所有参数值使用 CDATA 包裹:

<tool_call>
  <name>web_search</name>
  <params>
    <query><![CDATA[Python async best practices]]></query>
    <max_results><![CDATA[10]]></max_results>
  </params>
</tool_call>

为什么使用 CDATA?

  • 避免特殊字符(<, >, &)导致解析失败
  • 支持多行内容(如代码片段)
  • 保持内容原样,无需转义

XMLToolCallParser

@dataclass
class ToolCall:
    """工具调用数据结构"""
    name: str
    params: Dict[str, Any]
    raw_text: str = ""

class XMLToolCallParser:
    @staticmethod
    def parse_tool_calls(text: str) -> List[ToolCall]:
        """
        解析 XML 工具调用

        Returns:
            ToolCall 对象列表
        """
        # 使用 xml.etree.ElementTree 解析
        # 自动处理 CDATA
        # 自动类型转换(bool, int, float, string)
        # 支持 fallback 正则解析(处理 LLM 格式不严格的情况)

# 便捷函数
def parse_tool_calls(text: str) -> List[ToolCall]:
    return XMLToolCallParser.parse_tool_calls(text)

类型转换规则

原始值 转换结果
"true" / "false" bool
"123" int
"3.14" float
其他 str

XML 格式化器 (xml_formatter.py)

xml_parser.py 互为 formatter / parser 对。提供两个模块级函数:

def generate_tool_instruction(tools: List[BaseTool]) -> str:
    """
    生成完整的工具使用说明(注入 system prompt)

    包含:
    - 可用工具列表
    - 每个工具的参数说明
    - XML 调用格式示例(使用 CDATA)
    """

def format_result(name: str, result: Dict[str, Any]) -> str:
    """格式化工具执行结果为 XML(注入 context 消息)"""

生成示例(实际输出格式):

### web_search
Description: Search the web for information using Bocha AI search engine
Parameters:
  - query: string (required) - Search query using keywords.
  - freshness: string (optional) - Time range filter: 'noLimit', 'oneDay', ...
    Default: noLimit
  - count: integer (optional) - Number of results to return (1-50)
    Default: 10
Example:
<tool_call>
  <name>web_search</name>
  <params>
    <query><![CDATA[your_query_here]]></query>
    <freshness><![CDATA[noLimit]]></freshness>
    <count><![CDATA[10]]></count>
  </params>
</tool_call>

内置工具

工具名 说明 权限
call_subagent CallSubagentTool 路由到 SubAgent(验证参数后设置路由) AUTO
web_search WebSearchTool 互联网搜索(博查 AI) AUTO
web_fetch WebFetchTool 网页/PDF 内容抓取(Jina Reader API,BS4/DocConverter+pymupdf 降级) CONFIRM
create_artifact CreateArtifactTool 创建新 Artifact AUTO
read_artifact ReadArtifactTool 读取内容(支持历史版本) AUTO
update_artifact UpdateArtifactTool 增量更新(支持模糊匹配) AUTO
rewrite_artifact RewriteArtifactTool 完全重写内容 AUTO

注意call_subagent 是特殊的路由工具。Agent 在 base.py 中检测到此工具时,会先调用 execute() 验证参数(如 agent_name 是否有效),验证通过后再设置路由信息,由 Graph 路由到目标 SubAgent。

添加新工具

参见 Extension Guide