在前四节中，你学习了工具的定义与注册、Schema设计、安全沙箱和错误处理。这些知识是分开讲解的，但在实际开发中，你需要将它们整合起来应用。
本节将构建一个完整的实战案例：为Agent编写5个实用工具。这些工具的设计覆盖了不同的分类、权限级别、错误处理策略和安全考量。通过这个案例，你将看到第4章的所有原则如何在一个具体的工具集中落地。
假设你在为一个智能办公助手Agent构建工具集。这个Agent帮助用户处理日常办公任务：查询日程、发送消息、管理任务、搜索文档、生成报告。
这5个工具分别是：
| 工具名 | 分类 | 权限 | 重试策略 |
|---|---|---|---|
query_calendar | READ | AUTO | 可重试（指数退避） |
send_message | COMMUNICATE | USER_CONFIRM | 不重试（非幂等） |
create_task | WRITE | AUTO | 可重试（幂等设计） |
search_documents | READ | AUTO | 可重试（固定间隔） |
execute_calculation | EXECUTE | AUTO | 不重试（确定性操作） |
这5个工具覆盖了不同的副作用级别和安全需求。让我们逐一实现。
设计考量：
READ类工具，无副作用，可以自动执行
需要处理日期格式的复杂性——LLM可能填"明天"、"下周三"、"2026-09-10"等各种格式
网络请求可能超时，需要指数退避重试
from datetime import datetime, timedelta
import re
from typing import List, Optional
def query_calendar(start_date: str, end_date: str,
calendar_id: str = "primary") -> str:
"""
查询指定时间范围内的日历事件。
参数start_date和end_date支持以下格式：
- YYYY-MM-DD（如"2026-09-01"）
- 相对日期（如"today"、"tomorrow"、"next week"）
"""
# 解析日期
start = _parse_flexible_date(start_date)
end = _parse_flexible_date(end_date)
if start is None or end is None:
return f"日期格式无法解析：start={start_date}, end={end_date}。请使用YYYY-MM-DD格式。"
if start > end:
return f"开始日期（{start_date}）晚于结束日期（{end_date}）。请调整日期范围。"
# 模拟日历数据
events = [
{"date": "2026-09-03", "time": "10:00", "title": "产品评审会", "attendees": 6},
{"date": "2026-09-05", "time": "14:30", "title": "一对一沟通", "attendees": 2},
{"date": "2026-09-08", "time": "09:00", "title": "周会", "attendees": 12},
]
# 过滤时间范围内的事件
filtered = []
for event in events:
event_date = datetime.strptime(event["date"], "%Y-%m-%d")
if start <= event_date <= end:
filtered.append(event)
if not filtered:
return f"{start_date} 到 {end_date} 期间没有日程安排。"
result_lines = [f"{start_date} 到 {end_date} 期间的日程："]
for event in filtered:
result_lines.append(
f"- {event['date']} {event['time']} | {event['title']}（{event['attendees']}人）"
)
return "\n".join(result_lines)
def _parse_flexible_date(date_str: str) -> Optional[datetime]:
"""解析灵活的日期格式"""
today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
# YYYY-MM-DD格式
if re.match(r'^\d{4}-\d{2}-\d{2}$', date_str):
try:
return datetime.strptime(date_str, "%Y-%m-%d")
except ValueError:
return None
# 相对日期
relative_dates = {
"today": today,
"tomorrow": today + timedelta(days=1),
"yesterday": today - timedelta(days=1),
"next week": today + timedelta(days=7),
"last week": today - timedelta(days=7),
"next month": today + timedelta(days=30),
}
if date_str.lower() in relative_dates:
return relative_dates[date_str.lower()]
return None
# 注册工具
@registry.tool(
description=(
"查询指定时间范围内的日历事件。"
"当用户询问'我下周有什么安排'、'明天有没有会议'、'查看日历'等问题时使用此工具。"
"参数start_date和end_date支持YYYY-MM-DD格式或相对日期（today/tomorrow/next week等）。"
"如果用户只提供了一个日期，将start_date和end_date设为同一天。"
),
category=ToolCategory.READ,
permission=ToolPermission.AUTO,
timeout_seconds=10,
max_retries=3,
tags=["calendar", "schedule", "query"],
examples=[
"query_calendar(start_date='2026-09-01', end_date='2026-09-07')",
"query_calendar(start_date='tomorrow', end_date='tomorrow')",
],
)
def query_calendar_tool(start_date: str, end_date: str, calendar_id: str = "primary") -> str:
"""查询日历"""
return query_calendar(start_date, end_date, calendar_id)
值得注意的设计：
第一：灵活日期解析。 _parse_flexible_date支持多种日期格式，并处理了解析失败的情况。这很重要——LLM可能填"明天"而不是"2026-09-06"，工具应该能处理这种灵活性。
第二：日期范围的合理性检查。 工具验证了start <= end。如果LLM填反了日期（比如start="2026-09-07"而end="2026-09-01"），工具返回清晰的错误信息，LLM可以据此修正。
第三：示例中包含具体参数。 examples字段展示了工具的典型用法，这会在描述中帮助LLM理解如何正确调用。
设计考量：
COMMUNICATE类工具，有外部副作用，需要用户确认
非幂等——发送两次消息，接收者收到两条。因此不自动重试
需要验证接收者的有效性
def send_message(to_user: str, content: str,
channel: str = "internal") -> str:
"""
向指定用户发送消息。
参数to_user是接收者的用户名或邮箱。
参数channel是发送渠道：internal=站内消息，email=邮件。
"""
# 验证接收者
valid_users = {"zhangsan", "lisi", "wangwu", "zhaoliu"}
# 提取用户名（可能输入的是邮箱）
username = to_user.lower()
if "@" in username:
username = username.split("@")[0]
if username not in valid_users:
return f"未找到用户 '{to_user}'。可用的用户：{', '.join(sorted(valid_users))}"
# 验证消息内容
if len(content) > 2000:
return f"消息内容过长（{len(content)}字符），最大允许2000字符。请精简消息。"
if not content.strip():
return "消息内容不能为空。"
# 模拟发送
# 实际场景中这里会调用消息服务API
send_result = {
"message_id": f"MSG-{datetime.now().strftime('%Y%m%d%H%M%S')}",
"to": username,
"channel": channel,
"status": "sent",
"timestamp": datetime.now().isoformat(),
}
return (
f"消息已发送。"
f"接收者：{username}，渠道：{channel}，"
f"消息ID：{send_result['message_id']}"
)
@registry.tool(
description=(
"向指定用户发送消息。"
"当用户要求'通知某人'、'发消息给...'、'告诉...'时使用此工具。"
"参数to_user是接收者的用户名或邮箱。"
"参数content是消息内容，使用纯文本格式。"
"参数channel是发送渠道：internal=站内消息（默认），email=邮件。"
"注意：此工具会向真实用户发送消息，执行前需要用户确认。"
"消息发送后不可撤回。"
),
category=ToolCategory.COMMUNICATE,
permission=ToolPermission.USER_CONFIRM,
timeout_seconds=15,
max_retries=0, # 非幂等，不重试
tags=["message", "communication", "notify"],
metadata={"idempotent": False},
examples=[
"send_message(to_user='zhangsan', content='请查看最新的项目报告')",
],
)
def send_message_tool(to_user: str, content: str, channel: str = "internal") -> str:
"""发送消息"""
return send_message(to_user, content, channel)
值得注意的设计：
第一：验证接收者。 工具在"发送"之前先验证接收者是否存在。这防止了LLM填一个不存在的用户，导致消息丢失。注意工具返回的错误信息（"未找到用户...可用的用户：..."）足够LLM理解并做出调整。
第二：非幂等的明确标记。 max_retries=0和metadata={"idempotent": False}双重标记了此工具的非幂等特性。即使发生网络超时，也不会自动重试——宁可发送失败，也不能重复发送。
第三：USER_CONFIRM权限。 这个工具的调用会触发用户确认流程。Agent提议发送消息，但最终由用户批准。
设计考量：
WRITE类工具，有副作用但可接受自动执行
设计为幂等——相同的task_id不会创建重复任务
需要验证优先级和截止日期的合理性
def create_task(title: str, due_date: str,
priority: str = "medium",
assignee: str = "me",
task_id: Optional[str] = None) -> str:
"""
创建一个新任务。
参数task_id是可选的幂等键。如果提供了task_id且该任务已存在，不会创建重复任务。
"""
# 验证标题
if len(title) < 3:
return f"任务标题过短（{len(title)}字符）。请提供至少3字符的描述性标题。"
if len(title) > 200:
return f"任务标题过长（{len(title)}字符）。请精简到200字符以内。"
# 验证优先级
valid_priorities = {"low", "medium", "high", "urgent"}
if priority not in valid_priorities:
return f"无效的优先级 '{priority}'。可选值：{', '.join(sorted(valid_priorities))}"
# 验证截止日期
parsed_date = _parse_flexible_date(due_date)
if parsed_date is None:
return f"截止日期无法解析：{due_date}。请使用YYYY-MM-DD格式。"
today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
if parsed_date < today:
return f"截止日期（{due_date}）已过期。请设置未来的日期。"
# 幂等检查
if task_id:
existing = _find_task_by_idempotency_key(task_id)
if existing:
return f"任务已存在（任务ID：{existing['id']}），不重复创建。"
# 模拟创建任务
generated_id = task_id or f"TASK-{datetime.now().strftime('%Y%m%d%H%M%S')}"
task = {
"id": generated_id,
"title": title,
"due_date": due_date,
"priority": priority,
"assignee": assignee,
"status": "pending",
"created_at": datetime.now().isoformat(),
}
# 存储任务（实际场景中是数据库写入）
_store_task(task)
return (
f"任务已创建。"
f"任务ID：{task['id']}，"
f"标题：{title}，"
f"截止日期：{due_date}，"
f"优先级：{priority}"
)
# 模拟存储
_task_storage = {}
def _store_task(task: dict):
_task_storage[task["id"]] = task
def _find_task_by_idempotency_key(task_id: str) -> Optional[dict]:
return _task_storage.get(task_id)
@registry.tool(
description=(
"创建一个新任务。"
"当用户要求'创建任务'、'添加待办'、'记一下...'时使用此工具。"
"参数title是任务标题，至少3字符。"
"参数due_date是截止日期，支持YYYY-MM-DD格式或相对日期。"
"参数priority是优先级：low/medium/high/urgent，默认为medium。"
"参数task_id是可选的幂等键，用于防止重复创建。如果不确定，可以不填。"
),
category=ToolCategory.WRITE,
permission=ToolPermission.AUTO,
timeout_seconds=10,
max_retries=2, # 幂等操作，可以重试
tags=["task", "todo", "write"],
metadata={"idempotent": True},
examples=[
"create_task(title='准备季度报告', due_date='2026-09-15', priority='high')",
],
)
def create_task_tool(title: str, due_date: str,
priority: str = "medium",
assignee: str = "me",
task_id: Optional[str] = None) -> str:
"""创建任务"""
return create_task(title, due_date, priority, assignee, task_id)
值得注意的设计：
第一：幂等设计。 通过可选的task_id参数实现幂等性。如果LLM或Harness因为网络原因重试了工具调用，同一个task_id不会创建重复任务。这是WRITE类工具允许重试的前提条件。
第二：多维度验证。 工具验证了标题长度、优先级枚举值、截止日期的合理性和未来性。这些验证确保LLM填的参数是业务上合理的，而不仅仅是格式上正确的。
第三：允许重试但有限度。 max_retries=2——因为幂等设计保证了重试安全，但也不应该无限重试。
设计考量：
READ类工具，无副作用
搜索结果可能很大，需要在工具内部做摘要和截断
需要处理"无结果"的情况
def search_documents(query: str, max_results: int = 5,
doc_type: str = "all") -> str:
"""
搜索文档库。
参数doc_type可选值：all=所有类型，report=报告，meeting=会议记录，spec=技术文档
"""
# 模拟文档库
documents = [
{"id": "DOC-001", "title": "2026年Q2产品路线图", "type": "report",
"content": "Q2聚焦AI功能开发...", "relevance": 0.95},
{"id": "DOC-002", "title": "7月产品评审会纪要", "type": "meeting",
"content": "讨论了用户反馈和优先级...", "relevance": 0.87},
{"id": "DOC-003", "title": "搜索架构设计文档", "type": "spec",
"content": "描述了搜索系统的技术架构...", "relevance": 0.82},
{"id": "DOC-004", "title": "2026年H1销售报告", "type": "report",
"content": "上半年销售数据分析...", "relevance": 0.78},
{"id": "DOC-005", "title": "8月战略讨论会纪要", "type": "meeting",
"content": "确定了Q3的战略重点...", "relevance": 0.75},
]
# 验证搜索词
if len(query) < 2:
return f"搜索词过短（'{query}'）。请提供至少2字符的搜索词。"
# 过滤类型
if doc_type != "all":
documents = [d for d in documents if d["type"] == doc_type]
# 模拟相关性搜索
results = [d for d in documents if query.lower() in d["title"].lower()
or query.lower() in d["content"].lower()]
# 按相关性排序
results.sort(key=lambda x: x["relevance"], reverse=True)
# 限制结果数量
results = results[:max_results]
if not results:
return f"未找到与 '{query}' 相关的文档。建议尝试更宽泛的搜索词，或检查搜索词的拼写。"
# 构建结果摘要
lines = [f"搜索 '{query}' 找到 {len(results)} 个相关文档："]
for doc in results:
# 内容摘要（前50字符）
content_summary = doc["content"][:50] + "..." if len(doc["content"]) > 50 else doc["content"]
lines.append(
f"- [{doc['id']}] {doc['title']}（{doc['type']}，相关度：{doc['relevance']:.2f}）\n"
f" 摘要：{content_summary}"
)
return "\n".join(lines)
@registry.tool(
description=(
"搜索文档库中的文档。"
"当用户询问'有没有关于...的文档'、'查找...资料'、'搜索文档'时使用此工具。"
"参数query是搜索关键词，至少2字符。"
"参数doc_type是文档类型过滤：all=所有，report=报告，meeting=会议记录，spec=技术文档。"
"参数max_results是返回的最大结果数，默认为5。"
"返回文档的标题、类型和相关度摘要。"
),
category=ToolCategory.READ,
permission=ToolPermission.AUTO,
timeout_seconds=10,
max_retries=3,
tags=["search", "document", "query"],
examples=[
"search_documents(query='产品路线图', doc_type='report')",
],
)
def search_documents_tool(query: str, max_results: int = 5, doc_type: str = "all") -> str:
"""搜索文档"""
return search_documents(query, max_results, doc_type)
值得注意的设计：
第一：工具内部做摘要。 搜索工具返回的是摘要（标题+内容前50字符+相关度），而不是完整的文档内容。这样设计的原因：LLM拿到摘要后，如果认为某个文档相关，可以再请求完整内容（通过其他工具）。搜索工具返回摘要，文档读取工具返回全文——责任分离。
第二："无结果"也是有用的信息。 工具返回"未找到相关文档"并给出建议（尝试更宽泛的搜索词），这比返回空字符串更有用。LLM看到"未找到"后可以调整搜索策略。
设计考量：
EXECUTE类工具，执行代码。安全风险最高的工具
需要使用安全沙箱隔离执行
确定性操作，不需要重试
import ast
import math
def execute_calculation(expression: str) -> str:
"""
安全地计算数学表达式。
只支持基本的数学运算：加减乘除、括号、指数、常用数学函数。
不支持变量赋值、循环、函数定义等。
"""
# 安全验证：只允许安全的AST节点
allowed_nodes = (
ast.Expression,
ast.BinOp, # 二元运算：+, -, *, /
ast.UnaryOp, # 一元运算：-x, +x
ast.Constant, # 常量
ast.Call, # 函数调用
ast.Name, # 变量名（用于math函数）
ast.Load,
ast.Add, ast.Sub, ast.Mult, ast.Div, ast.Pow, ast.Mod,
ast.USub, ast.UAdd,
)
allowed_functions = {
"abs": abs, "round": round, "min": min, "max": max,
"sqrt": math.sqrt, "sin": math.sin, "cos": math.cos,
"tan": math.tan, "log": math.log, "log10": math.log10,
"exp": math.exp, "floor": math.floor, "ceil": math.ceil,
}
try:
# 解析表达式
tree = ast.parse(expression, mode='eval')
# 验证AST节点
for node in ast.walk(tree):
if not isinstance(node, allowed_nodes):
return f"表达式包含不允许的操作：{type(node).__name__}。只支持基本的数学运算。"
if isinstance(node, ast.Call):
if isinstance(node.func, ast.Name):
if node.func.id not in allowed_functions:
return f"不允许的函数：{node.func.id}。允许的函数：{', '.join(allowed_functions.keys())}"
else:
return "不支持复杂的函数调用。"
if isinstance(node, ast.Name):
if node.id not in allowed_functions:
return f"不允许的变量：{node.id}。只支持数学函数。"
# 在受限环境中执行
safe_globals = {"__builtins__": {}}
safe_globals.update(allowed_functions)
result = eval(compile(tree, '<expression>', 'eval'), safe_globals, {})
# 格式化结果
if isinstance(result, float):
return f"{result:.10g}"
return str(result)
except SyntaxError as e:
return f"表达式语法错误：{e}"
except ZeroDivisionError:
return "除零错误。"
except Exception as e:
return f"计算出错：{type(e).__name__}: {e}"
@registry.tool(
description=(
"安全地计算数学表达式。"
"当用户需要进行数学计算时使用此工具。"
"参数expression是数学表达式，如'(3+5)*2'、'sqrt(16)'、'2**10'。"
"支持基本运算（+,-,*,/,**,%）、括号和常用数学函数（sqrt, sin, cos, abs, round等）。"
"不支持变量赋值、循环、文件操作。"
),
category=ToolCategory.EXECUTE,
permission=ToolPermission.AUTO,
timeout_seconds=5, # 计算应该很快
max_retries=0, # 确定性操作，重试无意义
tags=["calculation", "math", "execute"],
metadata={"idempotent": True}, # 确定性操作，天然幂等
examples=[
"execute_calculation(expression='(3+5)*2')",
"execute_calculation(expression='sqrt(144)')",
],
)
def execute_calculation_tool(expression: str) -> str:
"""执行安全计算"""
return execute_calculation(expression)
值得注意的设计：
第一：基于AST的安全验证。 这是最安全的方式——不依赖字符串黑名单，而是解析代码结构，只允许白名单中的节点类型。任何不在白名单中的操作（如导入模块、变量赋值）都会被拒绝。
第二：受限的执行环境。 eval的globals只包含允许的函数，__builtins__被设为空字典。即使AST验证有漏洞，执行环境本身也是一道防线。
第三：确定性操作不需要重试。 计算结果不会因为重试而改变。如果执行失败（如表达式语法错误），重试也不会修正问题。
将5个工具注册到同一个ToolRegistry中：
# examples/office_agent_tools.py
from harness.tools import ToolRegistry, ToolCategory, ToolPermission
def create_office_tool_registry() -> ToolRegistry:
"""创建办公助手的工具注册表"""
registry = ToolRegistry(
default_timeout=10,
default_max_memory=256,
default_max_output=5000,
)
# 注册5个工具
# （使用上面定义的装饰器函数）
registry.register(ToolDefinition(
name="query_calendar",
description="查询指定时间范围内的日历事件。当用户询问日程安排时使用。",
func=query_calendar,
parameters={
"type": "object",
"properties": {
"start_date": {
"type": "string",
"description": "开始日期。YYYY-MM-DD格式或相对日期（today/tomorrow/next week）。"
},
"end_date": {
"type": "string",
"description": "结束日期。YYYY-MM-DD格式或相对日期。"
},
"calendar_id": {
"type": "string",
"description": "日历ID，默认为'primary'。通常不需要修改。"
}
},
"required": ["start_date", "end_date"]
},
category=ToolCategory.READ,
permission=ToolPermission.AUTO,
timeout_seconds=10,
max_retries=3,
tags=["calendar", "schedule"],
))
# ... 注册其他4个工具 ...
return registry
测试这5个工具
创建一个测试Agent来验证工具集：
# examples/test_office_agent.py
def test_office_agent():
"""测试办公助手Agent"""
registry = create_office_tool_registry()
config = HarnessConfig(
llm=LLMConfig(
api_key=os.getenv("OPENAI_API_KEY", ""),
model="gpt-4o-mini",
temperature=0.0,
),
runtime=RuntimeConfig(
max_iterations=15,
timeout_seconds=120,
retry_attempts=2,
)
)
memory = MemorySystem(
system_prompt=(
"你是一个智能办公助手。你可以帮助用户：\n"
"1. 查询日历和日程安排\n"
"2. 向团队成员发送消息\n"
"3. 创建和管理任务\n"
"4. 搜索文档\n"
"5. 执行数学计算\n\n"
"在使用工具前，确保你理解了用户的需求。"
"如果信息不足，先询问用户。"
)
)
tracer = Tracer()
runtime = AgentRuntime(
config=config,
tool_registry=registry,
memory=memory,
tracer=tracer,
)
# 测试场景
test_inputs = [
"帮我查一下下周的日程安排",
"计算 (15 + 7) * 3 的结果",
"搜索包含'产品'的文档",
"创建一个任务：周五前完成项目报告，优先级高",
]
for user_input in test_inputs:
print(f"\n{'='*50}")
print(f"用户：{user_input}")
result = runtime.run(user_input)
print(f"状态：{result.status}")
print(f"回答：{result.content}")
print(f"迭代：{result.iterations}，耗时：{result.duration_ms}ms")
运行上面的测试，观察Agent如何处理需要多个工具协作的请求。例如：
用户："查一下我下周三的日程，如果有空的话帮我创建一个准备演示文稿的任务。"
Agent的行为可能是：
1. 调用query_calendar(start_date='next week', end_date='next week')
→ 返回：下周三有"产品评审会"（10:00-11:30）
2. 判断：下周三上午有会，下午空闲
3. 调用create_task(
title='准备演示文稿',
due_date='2026-09-09', # 下周三
priority='high'
)
→ 返回：任务已创建
4. 总结回答："下周三上午10点有产品评审会，下午有空。"
"我已帮你创建了一个'准备演示文稿'的高优先级任务，截止日期为下周三。"
这个例子展示了工具之间的协作：一个工具的输出（日历查询）影响了另一个工具的调用决策（是否创建任务、截止日期怎么定）。这正是Agent相对于传统程序的核心优势——动态决策、灵活组合。
回顾这5个工具，可以提炼出几个可复用的设计模式：
模式一：验证优先。 每个工具在执行核心逻辑之前，都先验证输入的合理性。验证失败时返回清晰的错误信息，让LLM能够理解并修正。
模式二：无结果也是结果。 查询类工具在"无结果"时返回有意义的信息（"未找到..."+"建议..."），而非空字符串。这让LLM能够基于"无结果"这一事实做出合理的后续决策。
模式三：副作用明确。 有副作用的工具（send_message）明确标记了权限级别和非幂等特性。无副作用的工具（query_calendar、search_documents）可以安全地自动执行和重试。
模式四：安全执行。 execute_calculation展示了如何安全地执行不可信代码——AST白名单验证+受限执行环境双层防御。
模式五：内部摘要。 可能返回大量数据的工具（search_documents）在工具内部做摘要，而非返回原始数据。这节省了上下文窗口，也让LLM更容易处理。
本节构建了5个实用工具，覆盖了工具设计的核心模式：
| 工具 | 关键设计模式 |
|---|---|
| query_calendar | 灵活输入解析、范围验证 |
| send_message | 用户确认、非幂等保护 |
| create_task | 幂等设计、多维验证 |
| search_documents | 内部摘要、无结果处理 |
| execute_calculation | AST安全验证、受限执行 |
这些模式的核心思想：工具不仅要"能用"，还要"安全地用"和"被LLM正确地用"。 好的工具设计让LLM更容易做正确的事，更难做错误的事。
第4章到此结束。在下一章中，我们将讨论记忆与会话管理——如何让Agent记住之前的对话、跨会话保持信息、并在有限的上下文窗口中高效管理记忆。
评论专区
评论加载中...登录后即可发表评论