这是一个基于Strands Agent框架的AI考试生成器，可以根据用户指定的参数生成各种类型的考试题目。 技术博客地址：https://aws.amazon.com/cn/blogs/china/evaluation-of-question-generation-and-agent-performance-based-on-strands-agent-framework/
- 支持多种题型：单选题、多选题、填空题
- 支持复合题型：可以同时生成多种类型的题目，系统会自动分配题目数量
- 支持不同难度级别：简单、中等、困难
- 支持参考资料处理：可以提供URL或文本作为参考资料
- 支持交互式HTML渲染：生成的考试可以在浏览器中交互式查看
- 支持中英文双语界面
- 支持并行生成题目：多个题目同时生成，提高效率
- 支持题目缓存：避免重复生成相似的题目
- 支持格式自动修复：自动修复格式问题，确保生成的考试内容格式正确
系统由以下几个主要组件组成：
- 后端服务：基于Strands Agent框架，提供考试生成API
- Agent组件：协调整个考试生成流程
- 工具组件：提供各种功能，如题目生成、参考资料处理、格式验证等
- 缓存组件：缓存生成的题目，避免重复生成
- 前端界面：基于React的用户界面，用于输入参数和显示结果
- 渲染服务：基于Flask的服务，将Markdown格式的考试内容转换为交互式HTML
- Python 3.8+
- Node.js 16+
- npm 8+
- AWS账户（用于访问Bedrock服务）
在使用系统之前，需要设置AWS凭证以访问Bedrock服务
# 安装所有依赖（自动检测并使用Python 3.10）
./setup.sh方法1：一键启动整个系统
./start.sh这将启动后端服务、Flask渲染服务和前端开发服务器。
方法2：选择性启动组件
# 只启动后端和Flask服务
./start.sh backend flask
# 只启动前端服务
./start.sh frontend
# 启动所有组件（与不带参数相同）
./start.sh all# 停止所有服务
./stop.sh
# 选择性停止组件
./stop.sh backend # 只停止后端服务
./stop.sh flask # 只停止Flask渲染服务
./stop.sh frontend # 只停止前端服务- 启动系统后，在浏览器中访问前端界面（通常是 http://*:5173）
- 填写考试参数：
- 年级
- 科目
- 题型（可以选择多种题型）
- 题目数量
- 难度级别
- 主题
- 参考资料（可选）
- 教师备注（可选）
- 点击"生成考试"按钮
- 等待考试生成完成
- 点击"查看考试"按钮，在新窗口中查看生成的考试
如果您想生成包含多种题型的考试，可以在"题型"选项中选择多种题型，系统会自动分配题目数量。例如：
- 如果您选择了"单选题"和"多选题"，并设置题目数量为5，系统会生成3道单选题和2道多选题
- 如果您选择了"单选题"、"多选题"和"填空题"，并设置题目数量为10，系统会生成4道单选题、3道多选题和3道填空题
您可以提供URL或文本作为参考资料，系统会自动处理参考资料并生成相关题目。例如：
- 如果您提供了一个网页URL，系统会自动获取网页内容并生成相关题目
- 如果您提供了文本，系统会直接使用该文本生成相关题目
测试文件组织如下：
tests/
├── __init__.py # 将tests目录标记为Python包
├── test_exam_tools.py # 题目生成工具的单元测试
└── test_workflow.py # 完整工作流程的集成测试
test_exam_tools.py: 测试题目生成工具的功能，包括单选题、多选题和填空题的生成，以及缓存机制和API调用test_workflow.py: 测试完整的工作流程，从用户请求到生成考试内容，再到渲染HTML
可以使用提供的测试脚本测试API：
./test_api.sh系统提供了单元测试和集成测试，可以使用以下命令运行测试：
# 运行所有测试
python run_tests.py
# 运行特定测试
python -m unittest tests.test_exam_tools
python -m unittest tests.test_workflow测试覆盖以下关键功能：
- 题目生成功能：测试各种题型的生成是否正确
- 缓存机制：测试缓存的读写和过期功能
- API调用：测试与Claude API的交互，包括重试逻辑
- 格式处理：测试格式验证和自动修复功能
- 工作流程：测试完整的考试生成流程
.
├── exam_generator/ # 后端代码
│ ├── __init__.py
│ ├── agent.py # Agent组件
│ ├── config.py # 配置组件
│ ├── server.py # 服务器组件
│ ├── tools/ # 工具组件
│ │ ├── __init__.py
│ │ ├── content_tools.py # 内容处理工具
│ │ ├── exam_tools.py # 题目生成工具
│ │ ├── reference_tools.py # 参考资料处理工具
│ │ └── render_tools.py # 渲染工具
│ └── utils/ # 工具函数
│ ├── __init__.py
│ ├── error_utils.py # 错误处理工具
│ ├── logging_utils.py # 日志工具
│ └── task_manager.py # 任务管理工具
├── flask-service/ # Flask渲染服务
├── frontend/ # 前端代码
├── tests/ # 测试代码
│ ├── __init__.py
│ ├── test_exam_tools.py # 题目生成工具测试
│ └── test_workflow.py # 工作流程测试
├── cache/ # 缓存目录
├── run_tests.py # 测试运行脚本
├── start.sh # 统一启动脚本（可选择组件）
├── stop.sh # 统一停止脚本（可选择组件）
└── test_api.sh # API测试脚本
┌─────────────┐ HTTP请求 ┌─────────────┐ 调用 ┌─────────────┐
│ │ ───────────────> │ │ ───────────> │ │
│ 前端界面 │ │ 后端服务 │ │ Agent框架 │
│ (React) │ <─────────────── │ (Flask) │ <─────────── │ (Strands) │
└─────────────┘ HTTP响应 └─────────────┘ 返回 └──────┬──────┘
│
│ 调用
▼
┌─────────────┐ HTTP请求 ┌─────────────┐ ┌─────────────┐
│ │ <─────────────── │ │ <─────────── │ │
│ 渲染服务 │ │ 工具函数 │ │ Claude API │
│ (Flask) │ ───────────────> │ (Python) │ ───────────> │ (AWS) │
└─────────────┘ HTTP响应 └─────────────┘ └─────────────┘
│ │
│ │
▼ ▼
┌─────────────┐ ┌─────────────┐
│ │ │ │
│ HTML页面 │ │ 缓存系统 │
│ (Browser) │ │ (本地文件) │
└─────────────┘ └─────────────┘
┌─────────────┐
│ 用户请求 │
└──────┬──────┘
│
▼
┌─────────────┐ 调用 ┌─────────────┐
│ │ ───────────> │ │
│ Server.py │ │ Agent.py │
│ │ <─────────── │ │
└─────────────┘ 返回 └──────┬──────┘
│
│ 调用工具
▼
┌───────────────────────────────────────────────┐
│ │
│ Tools │
│ │
├─────────────┬─────────────┬─────────────┐ │
│ content_ │ exam_ │ reference_ │ │
│ tools.py │ tools.py │ tools.py │ │
├─────────────┼─────────────┼─────────────┤ │
│ - extract_ │ - generate_ │ - fetch_url_│ │
│ metadata │ question │ content │ │
│ - validate_ │ - cache │ - process_ │ │
│ format │ mechanism │ reference │ │
│ - plan_exam_│ - parallel │ │ │
│ content │ processing│ │ │
└─────────────┴─────────────┴─────────────┘ │
│ │
└───────────────────────────────────────────────┘
│
│ 调用
▼
┌─────────────┐ │ ┌─────────────┐
│ │ │ │ │
│ Claude API │ <──┴──> │ 缓存系统 │
│ │ │ │
└─────────────┘ └─────────────┘
-
后端：
- Strands Agent框架：提供Agent能力，协调工具调用
- Flask：提供HTTP服务和渲染服务
- Python：主要编程语言
- AWS Bedrock (Claude模型)：提供大语言模型能力
- 并行处理：使用ThreadPoolExecutor实现并行生成题目
- 缓存机制：使用本地文件系统实现题目缓存
-
前端：
- React：前端框架
- TypeScript：类型安全的JavaScript
- Material-UI：UI组件库
- Vite：构建工具
系统实现了一个完整的任务评估系统，用于跟踪和评估Agent工具调用的执行情况。这个系统基于Strands Agent的回调机制，能够提供详细的工具调用统计和性能指标。
Strands Agent框架提供了强大的回调机制，允许我们监听和处理Agent的各种事件，特别是工具调用事件。我们利用这一机制实现了以下功能：
- 工具调用跟踪：记录每个工具调用的开始、完成和失败事件
- 执行时间统计：计算每个工具调用的执行时间
- 状态管理：维护工具调用的状态（运行中、完成、失败）
- 自动完成逻辑：自动处理未明确标记为完成的工具调用
回调处理函数的核心逻辑如下：
def callback_handler(**kwargs):
if "current_tool_use" in kwargs:
tool_use = kwargs["current_tool_use"]
tool_id = tool_use.get("toolUseId")
tool_name = tool_use.get("name")
# 记录工具调用开始
tool_call_id = task_manager.record_tool_call(
workflow_id=workflow_id,
step_id=step_id,
tool_name=tool_name,
input_data=tool_use.get("input")
)
# 如果有输出，则视为工具调用完成
if "output" in tool_use and tool_use["output"] is not None:
task_manager.complete_tool_call(
workflow_id=workflow_id,
step_id=step_id,
tool_call_id=tool_call_id,
output_data=tool_use.get("output")
)TaskManager是任务评估系统的核心组件，负责管理工作流、步骤和工具调用的生命周期。它的工作流程如下：
┌─────────────────┐
│ 开始工作流 │
│ start_workflow()│
└────────┬────────┘
│
▼
┌─────────────────┐
│ 添加步骤 │
│ add_step() │
└────────┬────────┘
│
▼
┌─────────────────┐
│ 开始步骤 │
│ start_step() │
└────────┬────────┘
│
▼
┌─────────────────┐
│ 记录工具调用 │
│record_tool_call()│
└────────┬────────┘
│
▼
┌─────────────────┐
│ 完成/失败工具调用│
│complete_tool_call│
│fail_tool_call() │
└────────┬────────┘
│
▼
┌─────────────────┐
│ 完成步骤 │
│ complete_step() │
└────────┬────────┘
│
▼
┌─────────────────┐
│ 完成工作流 │
│complete_workflow│
└────────┬────────┘
│
▼
┌─────────────────┐
│ 生成评估报告 │
│generate_report() │
└─────────────────┘
TaskManager与主流程的集成如下图所示：
┌─────────────┐ HTTP请求 ┌─────────────┐
│ │ ───────────────> │ │
│ 前端界面 │ │ server.py │
│ (React) │ <─────────────── │ │
└─────────────┘ HTTP响应 └──────┬──────┘
│
│ 创建工作流
▼
┌─────────────────────────────────────────────────────┐
│ │
│ TaskManager │
│ │
├─────────────┐ ┌─────────────┐ ┌───────────┐ │
│ 工作流管理 │ │ 步骤管理 │ │ 工具调用 │ │
│ │ │ │ │ 管理 │ │
│- 创建工作流 │ │- 添加步骤 │ │- 记录调用 │ │
│- 完成工作流 │ │- 开始步骤 │ │- 完成调用 │ │
│- 失败工作流 │ │- 完成步骤 │ │- 失败调用 │ │
│- 生成报告 │ │- 失败步骤 │ │ │ │
└─────────────┘ └─────────────┘ └───────────┘ │
│ │
└───────────────────┬─────────────────────────────────┘
│
│ 创建回调
▼
┌─────────────┐ 调用 ┌─────────────┐
│ │ ───────────> │ │
│ agent.py │ │ Strands Agent│
│ │ <─────────── │ │
└──────┬──────┘ 返回 └──────┬──────┘
│ │
│ 调用工具 │ 触发回调
▼ ▼
┌─────────────┐ ┌─────────────┐
│ │ │ │
│ 工具函数 │ │ 回调处理函数 │
│ (Tools) │ │(callback_ │
│ │ │ handler) │
└─────────────┘ └─────────────┘
系统生成的评估报告包含以下内容：
-
工作流统计：
- 工作流ID和名称
- 工作流状态（完成、失败）
- 总执行时间
-
工具调用统计：
- 总工具调用次数
- 成功的工具调用次数
- 失败的工具调用次数
- 工具调用成功率
-
工具分布统计：
- 各工具的调用次数
- 各工具的成功/失败次数
- 各工具的平均执行时间
-
步骤统计：
- 总步骤数
- 完成的步骤数
- 失败的步骤数
- 步骤完成率
-
性能指标：
- 平均工具执行时间
评估报告可以通过API获取：GET /evaluation/report?workflow_id=xxx
生成的数据，会使用QuickSight进行可视化展示：
系统实现了多层次的错误处理和重试机制：
- 工具层面：
call_claude函数实现了指数退避重试策略，特别针对API限流错误 - Agent层面：
generate_exam函数实现了整体重试逻辑，确保即使在Strands框架层面遇到API限流也能恢复 - 错误分类：系统区分不同类型的错误，对API限流错误采用特殊处理
缓存系统具有以下特点：
- 基于文件：使用本地文件系统存储缓存，每个题目对应一个缓存文件
- MD5键值：使用题目主题、难度、类型和参考资料的MD5哈希作为缓存键
- TTL机制：缓存有30天的有效期，过期后自动失效
- 自动创建：系统自动创建缓存目录，无需手动设置
并行处理具有以下特点：
- 线程池：使用ThreadPoolExecutor实现并行处理
- 限制并发：限制最大并发数为3，避免API限流
- 错误处理：即使部分题目生成失败，也不会影响其他题目的生成
- 备用内容：对于生成失败的题目，提供备用内容确保整体流程不中断