个人模块 README — B2

模块 B2：Skill 工具函数模块 | 负责人：黄明涛 20235916

---

1. 模块概述

1.1 模块名称

B2 — Skill 工具函数模块 (Skill Functions)

1.2 模块说明

B2 是 Agent 系统的工具执行层，负责实现 LLM 可调用的全部具体工具函数。本模块提供 8 个 Skill，覆盖计算、文件读写、搜索、表格分析、格式转换、安全代码执行和组合管道，是 Agent 从"思考"到"行动"的桥梁。

B2 的核心职责：
1. 实现 8 个独立 Skill 函数，每个都有明确的输入/输出/异常处理
2. 提供统一的 run_skill() 调度入口，支持重试和守卫限制
3. 通过 B3 被 LLM 调用，每个 Skill 对应 tools.yaml 中的一个工具定义
4. 支持复合 Skill 管道（Pipeline），实现多步骤串联执行
5. 错误分类系统（15 种错误码 + 3 级分类），精确区分可恢复/输入/致命错误

1.3 完成情况概览

类型	完成情况
基础要求	✅ 全部完成：5 个基础 Skill（calculator/file_reader/local_file_search/table_analyzer/format_converter），每个有正常+异常两种输入样例，JSON 序列化输出，日志记录
进阶要求	✅ 5 项全部完成：增强搜索(TF-IDF+正则)、沙箱代码执行(safe_executor)、复合Skill管道(Pipeline)、错误分类与重试、安全守卫(Guard)
可独立运行的演示	✅ 每个 Skill 独立 CLI 入口 + Web Demo（端口 8081）
与团队系统集成情况	✅ 通过 B3 的 execute_tool_calls() 间接调用，B2 的 run_skill() 是 B3 执行工具的唯一入口

---

2. 环境、模型与数据依赖

2.1 运行环境

项目	要求
Python 版本	3.10
必要依赖	PyYAML；进阶功能需要 jieba（中文分词）、rank-bm25（BM25 检索）、numpy、sentence-transformers（向量检索）
是否需要模型	基础功能不需要；进阶 safe_executor 的 AST 沙箱也不需要模型
是否需要 GPU	不需要
是否需要外部数据集	不需要

2.2 模型依赖

B2 本身不加载模型。所有 Skill 都是纯 Python 函数。

2.3 数据集或样例数据依赖

数据或文件	来源	项目内相对路径	用途
tool_input_*.json	项目自带	agent/data/tool_inputs/	每个 Skill 的正常+异常输入样例
docs/agent_intro.txt	项目自带	agent/data/docs/	file_reader 读取目标
tables/results.csv	项目自带	agent/data/tables/	table_analyzer 分析目标
pipeline_config.json	项目自带	agent/data/	管道配置样例

2.4 安装步骤

conda create -n b2_env python=3.10 -y
conda activate b2_env
cd agent
pip install -r requirements.txt
# 进阶功能额外依赖：
pip install jieba rank-bm25 sentence-transformers

---

3. 文件结构与接口边界

3.1 文件结构

agent/
├── code/
│   ├── b2_run_skill.py          # 核心：Skill 统一调度入口（184行）
│   ├── b2_error_codes.py        # 进阶：错误分类系统（15种错误码+3级分类，149行）
│   ├── b2_skill_guard.py        # 进阶：安全守卫（风险等级+超时+输入输出限制，187行）
│   └── b2_pipeline.py           # 进阶：复合Skill管道（176行）
├── skills/
│   ├── __init__.py              # 公共：resolve_data_path() 路径解析
│   ├── calculator.py            # Skill: 安全数学表达式计算
│   ├── file_reader.py           # Skill: 读取 txt/md 文件
│   ├── local_file_search.py     # Skill: TF-IDF 关键词搜索+正则+文件名筛选
│   ├── table_analyzer.py        # Skill: CSV/TSV 表格分析（行列数+列名+统计）
│   ├── format_converter.py      # Skill: 文本转 Markdown/JSON 并生成文件
│   ├── safe_executor.py         # 进阶: AST 沙箱安全代码执行
│   ├── read_and_convert.py      # 进阶: 组合Skill（读文件+转换格式）
│   └── search_and_analyze.py    # 进阶: 组合Skill（搜索+表格分析）
├── demo/
│   ├── b2_demo_server.py        # Web Demo 服务器（端口 8081）
│   └── templates/b2_index.html  # Web Demo 前端
├── data/tool_inputs/            # 每个 Skill 的输入样例
└── B2_VERIFICATION.md           # 模块验证说明书

3.2 接口边界

类型	来源 / 去向	数据格式	说明
输入	B3 → run_skill()	Python dict（skill_name + input_data）	工具名和参数
输入	CLI → b2_run_skill.py	JSON 文件路径 + --skill 参数	独立运行
输出	run_skill() → B3	SkillResult dict（status/output/error/latency_ms）	统一 JSON 格式
输出	CLI → 磁盘	{skill_name}_result.json + skill_run_log.jsonl	结果文件+日志

---

4. 基础要求实现与演示

4.1 基础功能说明

实现了 5 个基础 Skill，每个都有明确的输入校验、异常处理和结构化输出：

Skill	功能	输入	输出
calculator	安全计算数学表达式	expression（字符串）	result（数值）
file_reader	读取 txt/md 文件	path + max_chars	content + num_chars + truncated
local_file_search	关键词搜索本地文件	query + root_dir + top_k	results[]（路径/分数/片段）
table_analyzer	分析 CSV/TSV 表格	path + describe	num_rows + columns + preview + describe
format_converter	文本格式转换	text + target_format	formatted_text + generated_file_path

4.2 基础功能实现路径

文件 / 函数	作用
b2_run_skill.py:run_skill()	统一调度：动态导入 → 参数注入 → 执行 → 错误处理 → SkillResult
b2_run_skill.py:SKILL_MODULES	8 个 Skill 的注册表（名称→模块路径映射）
skills/__init__.py:resolve_data_path()	路径解析：将相对路径拼到 data_root 下，安全检查
skills/calculator.py:calculator()	安全计算：白名单运算符+禁止危险函数
skills/file_reader.py:file_reader()	文件读取：类型校验+截断+编码处理
skills/local_file_search.py:local_file_search()	文件搜索：TF-IDF 评分+高亮片段+多格式支持
skills/table_analyzer.py:table_analyzer()	表格分析：CSV/TSV 解析+数值统计
skills/format_converter.py:format_converter()	格式转换：key:value→JSON / 逐行→Markdown，含文件生成

4.3 基础功能演示命令

cd agent/code

# calculator
python b2_run_skill.py --skill calculator \
  --input ../data/tool_inputs/tool_input_calculator.json \
  --outdir ../outputs/B2_skills

# file_reader
python b2_run_skill.py --skill file_reader \
  --input ../data/tool_inputs/tool_input_file_reader.json \
  --outdir ../outputs/B2_skills

# local_file_search
python b2_run_skill.py --skill local_file_search \
  --input ../data/tool_inputs/tool_input_file_search.json \
  --outdir ../outputs/B2_skills

# table_analyzer
python b2_run_skill.py --skill table_analyzer \
  --input ../data/tool_inputs/tool_input_table_analyzer.json \
  --outdir ../outputs/B2_skills

# format_converter
python b2_run_skill.py --skill format_converter \
  --input ../data/tool_inputs/tool_input_format_converter.json \
  --outdir ../outputs/B2_skills

观察要点：

• 每个命令输出 {skill_name}_result.json，status=success
• skill_run_log.jsonl 追加写入运行记录
• 异常输入样例同样被正常捕获，status=error，CLI 返回 0

4.4 基础功能输出格式

输出文件	格式	说明
{skill_name}_result.json	SkillResult JSON	skill_name, status, input, output, error, latency_ms
skill_run_log.jsonl	JSONL	时间戳、Skill名、状态、路径、耗时、错误码

---

5. 进阶要求实现与演示

5.1 选择的进阶要求

进阶要求	是否完成	对应文件	简要说明
进阶1：增强搜索	✅	skills/local_file_search.py	TF-IDF 关键词搜索 + 正则匹配 + 文件名筛选 + 分数阈值
进阶2：沙箱代码执行	✅	skills/safe_executor.py	AST 白名单校验 + 禁止 import/eval/open + 超时控制
进阶3：复合Skill管道	✅	code/b2_pipeline.py	多步骤顺序执行，$step_id.output.field 引用前步结果
进阶4：错误分类与重试	✅	code/b2_error_codes.py	15 种错误码 + 3 级分类（可恢复/输入/致命）+ 指数退避重试
进阶5：安全守卫	✅	code/b2_skill_guard.py	4 级风险等级 + 输入输出大小限制 + 超时控制

5.2 进阶1：增强搜索（TF-IDF + 正则 + 文件名筛选）

local_file_search.py 实现了完整的 TF-IDF 搜索引擎：

• 关键词模式：中文分词 + TF-IDF 评分 + 排序
• 正则模式：pattern 参数覆盖 query，按匹配次数评分
• 文件名筛选：filename_pattern 参数过滤文件名
• 分数阈值：min_score 过滤低相关性结果
• 高亮片段：**匹配词** 标记，snippet 返回上下文

5.3 进阶2：沙箱代码执行（safe_executor）

safe_executor.py 通过 AST 白名单实现安全 Python 执行：

• 白名单节点：允许基本运算、列表推导、函数调用等
• 黑名单：禁止 import/eval/exec/open/breakpoint 等
• 禁止 dunder 属性访问、try/except、类定义、with 语句
• 超时控制：线程执行 + 超时中断（1-30秒可配置）
• 输出截断：max_output_chars 限制 stdout+stderr

5.4 进阶3：复合Skill管道（Pipeline）

b2_pipeline.py 支持多步骤顺序执行：

{
  "pipeline_id": "read_and_convert_demo",
  "steps": [
    {"step_id": "read", "skill": "file_reader", "args": {"path": "docs/agent_intro.txt", "max_chars": 2000}},
    {"step_id": "convert", "skill": "format_converter", "args": {"text": "$read.output.content", "target_format": "markdown"}}
  ]
}

$read.output.content 语法引用前步结果，管道在任一步失败时自动中断。

5.5 进阶4：错误分类与重试

b2_error_codes.py 定义了 15 种错误码：

错误码	含义	分类
E001-E006	输入参数/文件/目录/格式错误	input
E007	操作超时	recoverable
E009	守卫限制触发	recoverable
E010	沙箱违规	input
E011	运行时内部错误	fatal
E099	未分类错误	unknown

retry_with_backoff() 只对 recoverable 类错误做指数退避重试，input/fatal 类直接抛出。

5.6 进阶5：安全守卫（Guard）

b2_skill_guard.py 按风险等级限制执行：

Skill	风险等级	超时	输出上限	文件上限
calculator	low	10s	50KB	-
file_reader	medium	30s	100KB	10MB
format_converter	high	60s	500KB	50MB
safe_executor	critical	5s	10KB	禁止

---

6. 与团队系统的集成说明

B2 通过 run_skill() 函数被 B3 调用。B3 的 execute_tool_calls() 在内部调用 B2：

B3 → execute_tool_calls() → b2_run_skill_fn() → run_skill()

B2 的 SKILL_MODULES 注册表被 B3 引用（b3_tool_layer.py 第 16 行），B3 据此判断工具是否走 B2 的 guarded 执行路径。

run_skill() 函数签名：

def run_skill(
    skill_name: str,       # 工具名（如 "calculator"）
    input_data: dict,      # 参数字典（如 {"expression": "23*17+9"}）
    data_root: str | None, # 数据根目录（B3 从 tools.yaml 注入）
    output_dir: str | None,# 输出目录（B3 注入）
    retry: int = 0,        # 重试次数
    retry_delay: float = 1.0,
    guarded: bool = False, # 是否启用守卫
) -> dict:                 # SkillResult

---

7. 已知问题与后续改进

问题	当前原因	后续改进
file_reader 只支持 txt/md	设计如此，避免尝试读取二进制文件	可扩展支持更多文本格式（json/yaml/log）
Pipeline 不支持并行步骤	当前为顺序执行	可添加 DAG 依赖解析，支持并行
safe_executor 不支持 try/except	AST 白名单限制，安全考虑	可评估开放 try/except 的风险
format_converter 普通文本转 JSON	兜底逻辑按行拆分	可增加 LLM 辅助的结构化提取