工具库设计与集成

一个强大的 Agent 需要一个丰富且设计良好的工具库。本章讲解如何设计和集成实用工具。

通用工具库架构

安全的代码执行器

代码执行是 Agent 最强大也最危险的工具，必须用沙箱保护。

"""
安全代码执行器 - 使用受限环境
"""
import io
import sys
from contextlib import redirect_stdout, redirect_stderr
from typing import Any
class SafeCodeExecutor:
"""安全的 Python 代码执行器"""
# 允许使用的模块
ALLOWED_MODULES = {
"math", "statistics", "json", "re",
"datetime", "collections", "itertools",
"functools", "operator", "string",
}
# 禁止的关键词
BLOCKED_KEYWORDS = {
"import os", "import sys", "import subprocess",
"exec(", "eval(", "__import__",
"open(", "file(", "input(",
"breakpoint", "compile",
}
def __init__(self, timeout: int = 10):
self.timeout = timeout
def execute(self, code: str) -> dict:
"""
安全执行 Python 代码
Returns:
{"success": bool, "output": str, "error": str}
"""
# 安全检查
for keyword in self.BLOCKED_KEYWORDS:
if keyword in code:
return {
"success": False,
"output": "",
"error": f"安全限制: 不允许使用 '{keyword}'",
}
# 准备受限环境
safe_globals = {"__builtins__": {}}
# 注入允许的模块
for module_name in self.ALLOWED_MODULES:
try:
safe_globals[module_name] = __import__(module_name)
except ImportError:
pass
# 注入安全的内置函数
safe_builtins = {
"print": print, "len": len, "range": range,
"int": int, "float": float, "str": str,
"list": list, "dict": dict, "set": set,
"tuple": tuple, "bool": bool, "type": type,
"enumerate": enumerate, "zip": zip,
"map": map, "filter": filter, "sorted": sorted,
"min": min, "max": max, "sum": sum, "abs": abs,
"round": round, "isinstance": isinstance,
}
safe_globals["__builtins__"] = safe_builtins
# 捕获输出
stdout = io.StringIO()
stderr = io.StringIO()
try:
with redirect_stdout(stdout), redirect_stderr(stderr):
exec(code, safe_globals)
return {
"success": True,
"output": stdout.getvalue(),
"error": "",
}
except Exception as e:
return {
"success": False,
"output": stdout.getvalue(),
"error": f"{type(e).__name__}: {e}",
}
# 使用
executor = SafeCodeExecutor()
# 安全执行
result = executor.execute("""
data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
avg = sum(data) / len(data)
print(f"平均值: {avg}")
print(f"最大值: {max(data)}")
print(f"标准差: {statistics.stdev(data):.2f}")
""")
print(result["output"])
# 平均值: 5.5
# 最大值: 10
# 标准差: 3.03

Web 搜索工具

"""
Web 搜索工具集成
"""
import json
from dataclasses import dataclass
@dataclass
class SearchResult:
title: str
url: str
snippet: str
class WebSearchTool:
"""Web 搜索工具"""
def __init__(self, provider: str = "tavily"):
self.provider = provider
def search(self, query: str, max_results: int = 5) -> list[SearchResult]:
"""执行搜索"""
if self.provider == "tavily":
return self._search_tavily(query, max_results)
else:
return self._search_fallback(query, max_results)
def _search_tavily(self, query: str, max_results: int) -> list[SearchResult]:
"""使用 Tavily API 搜索"""
from tavily import TavilyClient
client = TavilyClient()
response = client.search(query, max_results=max_results)
results = []
for item in response.get("results", []):
results.append(SearchResult(
title=item.get("title", ""),
url=item.get("url", ""),
snippet=item.get("content", "")[:300],
))
return results
def _search_fallback(self, query: str, max_results: int) -> list[SearchResult]:
"""备用搜索方案"""
return [SearchResult(
title=f"关于 {query} 的结果",
url="https://example.com",
snippet=f"搜索 '{query}' 的相关信息...",
)]
def format_results(self, results: list[SearchResult]) -> str:
"""格式化搜索结果为文本"""
lines = []
for i, r in enumerate(results, 1):
lines.append(f"[{i}] {r.title}")
lines.append(f"    URL: {r.url}")
lines.append(f"    {r.snippet}\n")
return "\n".join(lines)

文件操作工具

"""
文件操作工具集
"""
from pathlib import Path
class FileTools:
"""文件操作工具集"""
def __init__(self, workspace: str = "./workspace"):
self.workspace = Path(workspace)
self.workspace.mkdir(parents=True, exist_ok=True)
def _safe_path(self, filepath: str) -> Path:
"""确保路径在工作区内"""
full_path = (self.workspace / filepath).resolve()
if not str(full_path).startswith(str(self.workspace.resolve())):
raise ValueError(f"路径越界: {filepath}")
return full_path
def read(self, filepath: str) -> str:
"""读取文件"""
path = self._safe_path(filepath)
if not path.exists():
return f"文件不存在: {filepath}"
return path.read_text(encoding="utf-8")
def write(self, filepath: str, content: str) -> str:
"""写入文件"""
path = self._safe_path(filepath)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content, encoding="utf-8")
return f"已写入: {filepath} ({len(content)} 字符)"
def list_dir(self, dirpath: str = ".") -> str:
"""列出目录内容"""
path = self._safe_path(dirpath)
if not path.is_dir():
return f"不是目录: {dirpath}"
items = []
for item in sorted(path.iterdir()):
prefix = "📁" if item.is_dir() else "📄"
size = item.stat().st_size if item.is_file() else 0
items.append(f"  {prefix} {item.name} ({size} bytes)")
return f"目录 {dirpath}:\n" + "\n".join(items)
def search(self, pattern: str, content_pattern: str = None) -> str:
"""搜索文件"""
matches = list(self.workspace.rglob(pattern))
if content_pattern and matches:
import re
filtered = []
for m in matches:
if m.is_file():
try:
text = m.read_text(encoding="utf-8")
if re.search(content_pattern, text):
filtered.append(m)
except Exception:
pass
matches = filtered
results = [str(m.relative_to(self.workspace)) for m in matches[:20]]
return f"找到 {len(matches)} 个文件:\n" + "\n".join(results)

整合到 Agent

"""
将工具集成到 Agent 中
"""
class ToolkitAgent:
"""集成了完整工具库的 Agent"""
def __init__(self):
self.client = OpenAI()
self.executor = SafeCodeExecutor()
self.search = WebSearchTool()
self.files = FileTools()
# 工具定义
self.tools = [
{
"type": "function",
"function": {
"name": "run_code",
"description": "执行 Python 代码，可用于数据处理和计算",
"parameters": {
"type": "object",
"properties": {
"code": {"type": "string", "description": "Python 代码"},
},
"required": ["code"],
},
},
},
{
"type": "function",
"function": {
"name": "web_search",
"description": "搜索互联网获取最新信息",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"},
},
"required": ["query"],
},
},
},
{
"type": "function",
"function": {
"name": "read_file",
"description": "读取工作区中的文件内容",
"parameters": {
"type": "object",
"properties": {
"filepath": {"type": "string", "description": "文件路径"},
},
"required": ["filepath"],
},
},
},
{
"type": "function",
"function": {
"name": "write_file",
"description": "向工作区写入文件",
"parameters": {
"type": "object",
"properties": {
"filepath": {"type": "string", "description": "文件路径"},
"content": {"type": "string", "description": "文件内容"},
},
"required": ["filepath", "content"],
},
},
},
]
# 工具路由
self._handlers = {
"run_code": lambda code: self.executor.execute(code),
"web_search": lambda query: self.search.format_results(
self.search.search(query)
),
"read_file": lambda filepath: self.files.read(filepath),
"write_file": lambda filepath, content: self.files.write(
filepath, content
),
}
def execute_tool(self, name: str, args: dict) -> str:
"""执行工具"""
handler = self._handlers.get(name)
if not handler:
return f"未知工具: {name}"
result = handler(**args)
if isinstance(result, dict):
return json.dumps(result, ensure_ascii=False)
return str(result)

本章小结

• 工具库设计要分层：信息获取、代码执行、文件操作、外部服务
• 代码执行器必须使用沙箱环境，限制危险操作
• 文件操作工具需要路径安全检查，防止越界访问
• Web 搜索为 Agent 提供获取实时信息的能力
• 工具路由用字典映射，保持代码简洁

下一章：学习 Agent 的 Memory 系统设计。