StructBERT Web系统API文档自动生成：Swagger集成与测试用例编写

StructBERT Web系统API文档自动生成：Swagger集成与测试用例编写

1. 项目背景与需求

当你成功部署了StructBERT中文语义智能匹配系统，看着它稳定运行，提供着精准的语义相似度计算和特征提取服务时，一个新的问题可能就摆在了面前：如何让其他开发同事、外部合作伙伴或者未来的自己，能够快速、准确地使用你提供的API接口？

手动编写API文档不仅耗时耗力，还容易出错。文档更新不及时，接口改了但文档没改，这种“文档滞后”的问题在项目协作中屡见不鲜。更重要的是，缺乏规范的文档，接口的测试和验证也变得困难重重。

本文将带你解决这个问题。我们将为你的StructBERT Web系统集成Swagger（OpenAPI）文档，实现API接口的自动生成与可视化，并编写配套的测试用例，打造一个既专业又易用的API服务。整个过程不需要你成为Swagger专家，跟着步骤走，一两个小时就能搞定。

2. 环境准备与Swagger集成

2.1 理解Swagger能为我们做什么

在开始动手之前，我们先简单了解一下Swagger（现在更常被称为OpenAPI）到底是什么，以及它能给我们带来什么好处。

Swagger是一套用于描述、生成、调用和可视化RESTful API的规范与工具集。把它集成到你的Flask项目中，主要能实现三个功能：

对于我们的StructBERT系统来说，这意味着：

• 前端开发同事可以直接在浏览器里查看所有接口的详细说明，包括URL、方法、参数、返回值等。
• 测试同事可以直接在Swagger UI上测试接口，无需安装额外的测试工具。
• 你自己在几个月后回顾项目时，也能快速回忆起每个接口的用途和用法。

2.2 安装必要的Python包

首先，我们需要在现有的torch26虚拟环境中安装几个必要的Python包。打开终端，激活你的虚拟环境，然后执行以下命令：

# 激活你的torch26虚拟环境（根据你的环境名称调整）
conda activate torch26

# 安装Flask-Swagger-UI和Flask-RESTx
# Flask-Swagger-UI用于提供Swagger UI界面
# Flask-RESTx是Flask-RESTPlus的继承者，功能更强大，我们主要用它来自动生成API文档
pip install flask-swagger-ui flask-restx

安装完成后，你可以通过以下命令验证是否安装成功：

pip list | grep -E "flask-swagger-ui|flask-restx"

应该能看到类似这样的输出：

flask-restx 1.3.0
flask-swagger-ui 4.11.1

2.3 修改Flask应用代码集成Swagger

现在，我们来修改StructBERT系统的Flask应用代码。我们假设你的主应用文件是app.py（如果不是，请根据实际情况调整文件名）。

重要提示：在修改代码前，建议先备份你的原始文件。

打开app.py文件，在文件顶部添加必要的导入：

# 在原有导入的基础上，添加以下导入
from flask_restx import Api, Resource, fields, reqparse
from flask_swagger_ui import get_swaggerui_blueprint

接下来，找到Flask应用初始化的地方（通常是app = Flask(__name__)这一行），在其后添加Swagger和RESTx的配置：

# 原有的Flask应用初始化
app = Flask(__name__)
app.config['JSON_AS_ASCII'] = False # 确保中文正常显示

# ========== 新增：Swagger UI配置 ==========
SWAGGER_URL = '/api/docs' # Swagger UI的访问路径
API_URL = '/api/swagger.json' # OpenAPI规范文件的路径

# 创建Swagger UI蓝图
swaggerui_blueprint = get_swaggerui_blueprint(
SWAGGER_URL,
API_URL,
config={
'app_name': "StructBERT中文语义智能匹配系统API文档"
}
)

# 注册Swagger UI蓝图到应用
app.register_blueprint(swaggerui_blueprint, url_prefix=SWAGGER_URL)

# ========== 新增：Flask-RESTx API初始化 ==========
# 创建API实例
api = Api(
app,
version='1.0',
title='StructBERT中文语义智能匹配系统API',
description='基于StructBERT孪生网络模型的中文语义相似度计算与特征提取服务',
doc='/api/docs' # 注意：这里我们使用自定义的Swagger UI，所以不启用RESTx自带的文档
)

# 定义命名空间（用于组织接口）
ns_similarity = api.namespace('similarity', description='语义相似度计算相关接口')
ns_feature = api.namespace('feature', description='特征提取相关接口')

2.4 重构API接口并添加Swagger注解

现在我们需要重构现有的API接口，使用Flask-RESTx的Resource类和装饰器来添加Swagger注解。

首先，我们定义请求和响应的数据模型。在API初始化代码后面添加：

# ========== 新增：定义数据模型 ==========
# 相似度计算请求模型
similarity_request_model = api.model('SimilarityRequest', {
'text1': fields.String(required=True, description='第一段文本', example='今天天气真好'),
'text2': fields.String(required=True, description='第二段文本', example='阳光明媚的早晨'),
'threshold_high': fields.Float(required=False, description='高相似度阈值，默认0.7', example=0.7),
'threshold_low': fields.Float(required=False, description='低相似度阈值，默认0.3', example=0.3)
})

# 相似度计算响应模型
similarity_response_model = api.model('SimilarityResponse', {
'text1': fields.String(description='第一段文本'),
'text2': fields.String(description='第二段文本'),
'similarity_score': fields.Float(description='语义相似度得分（0-1）'),
'similarity_level': fields.String(description='相似度等级：high/medium/low'),
'inference_time': fields.Float(description='推理耗时（秒）'),
'status': fields.String(description='请求状态：success/error'),
'message': fields.String(description='状态信息')
})

# 特征提取请求模型
feature_request_model = api.model('FeatureRequest', {
'text': fields.String(required=True, description='需要提取特征的文本', example='这是一个测试文本'),
'batch_texts': fields.List(fields.String, required=False, description='批量文本列表（用于批量提取）', example=['文本1', '文本2', '文本3'])
})

# 特征提取响应模型
feature_response_model = api.model('FeatureResponse', {
'text': fields.String(description='输入文本'),
'feature_vector': fields.List(fields.Float, description='768维特征向量'),
'feature_preview': fields.List(fields.Float, description='前20维特征预览'),
'inference_time': fields.Float(description='推理耗时（秒）'),
'status': fields.String(description='请求状态：success/error'),
'message': fields.String(description='状态信息')
})

# 批量特征提取响应模型
batch_feature_response_model = api.model('BatchFeatureResponse', {
'results': fields.List(fields.Nested(feature_response_model), description='批量处理结果列表'),
'total_time': fields.Float(description='总处理耗时（秒）'),
'total_texts': fields.Integer(description='处理文本总数'),
'status': fields.String(description='请求状态：success/error'),
'message': fields.String(description='状态信息')
})

接下来，我们需要重构原有的路由处理函数。找到你处理API请求的函数（可能是calculate_similarity()、extract_feature()等），将它们重构为Resource类。

在app.py中添加以下代码（请根据你的实际函数名和逻辑进行调整）：

# ========== 新增：语义相似度计算接口 ==========
@ns_similarity.route('/calculate')
class SimilarityCalculate(Resource):
@ns_similarity.expect(similarity_request_model)
@ns_similarity.marshal_with(similarity_response_model)
def post(self):
"""
计算两段中文文本的语义相似度

使用StructBERT孪生网络模型计算文本相似度，返回0-1之间的相似度得分和等级判定。
支持自定义相似度阈值。
"""
try:
# 获取请求数据
data = api.payload

text1 = data.get('text1', '').strip()
text2 = data.get('text2', '').strip()
threshold_high = data.get('threshold_high', 0.7)
threshold_low = data.get('threshold_low', 0.3)

# 参数校验
if not text1 or not text2:
return {
'text1': text1,
'text2': text2,
'similarity_score': 0.0,
'similarity_level': 'error',
'inference_time': 0.0,
'status': 'error',
'message': '文本内容不能为空'
}, 400

# 记录开始时间
start_time = time.time()

# 调用你的相似度计算函数
# 注意：这里需要替换成你实际的相似度计算函数
similarity_score = calculate_similarity_score(text1, text2)

# 根据阈值判定相似度等级
if similarity_score >= threshold_high:
similarity_level = 'high'
elif similarity_score >= threshold_low:
similarity_level = 'medium'
else:
similarity_level = 'low'

# 计算推理时间
inference_time = time.time() – start_time

return {
'text1': text1,
'text2': text2,
'similarity_score': round(similarity_score, 4),
'similarity_level': similarity_level,
'inference_time': round(inference_time, 4),
'status': 'success',
'message': '相似度计算成功'
}

except Exception as e:
return {
'text1': data.get('text1', ''),
'text2': data.get('text2', ''),
'similarity_score': 0.0,
'similarity_level': 'error',
'inference_time': 0.0,
'status': 'error',
'message': f'相似度计算失败：{str(e)}'
}, 500

# ========== 新增：单文本特征提取接口 ==========
@ns_feature.route('/extract')
class FeatureExtract(Resource):
@ns_feature.expect(feature_request_model)
@ns_feature.marshal_with(feature_response_model)
def post(self):
"""
提取单文本的768维语义特征向量

使用StructBERT模型提取文本的语义特征，返回768维向量和前20维预览。
"""
try:
data = api.payload
text = data.get('text', '').strip()

if not text:
return {
'text': '',
'feature_vector': [],
'feature_preview': [],
'inference_time': 0.0,
'status': 'error',
'message': '文本内容不能为空'
}, 400

start_time = time.time()

# 调用你的特征提取函数
# 注意：这里需要替换成你实际的特征提取函数
feature_vector = extract_text_feature(text)

# 获取前20维作为预览
feature_preview = feature_vector[:20] if len(feature_vector) >= 20 else feature_vector

inference_time = time.time() – start_time

return {
'text': text,
'feature_vector': [round(float(x), 6) for x in feature_vector],
'feature_preview': [round(float(x), 6) for x in feature_preview],
'inference_time': round(inference_time, 4),
'status': 'success',
'message': '特征提取成功'
}

except Exception as e:
return {
'text': data.get('text', ''),
'feature_vector': [],
'feature_preview': [],
'inference_time': 0.0,
'status': 'error',
'message': f'特征提取失败：{str(e)}'
}, 500

# ========== 新增：批量特征提取接口 ==========
@ns_feature.route('/batch-extract')
class BatchFeatureExtract(Resource):
@ns_feature.expect(feature_request_model)
@ns_feature.marshal_with(batch_feature_response_model)
def post(self):
"""
批量提取多文本的768维语义特征向量

一次性处理多个文本，返回每个文本的特征向量和预览。
建议单次批量不超过100条文本。
"""
try:
data = api.payload
batch_texts = data.get('batch_texts', [])

if not batch_texts or not isinstance(batch_texts, list):
return {
'results': [],
'total_time': 0.0,
'total_texts': 0,
'status': 'error',
'message': '批量文本列表不能为空且必须是列表格式'
}, 400

# 过滤空文本
valid_texts = [text.strip() for text in batch_texts if text and str(text).strip()]

if not valid_texts:
return {
'results': [],
'total_time': 0.0,
'total_texts': 0,
'status': 'error',
'message': '没有有效的文本内容'
}, 400

total_start_time = time.time()
results = []

# 批量处理文本
for text in valid_texts:
try:
text_start_time = time.time()

# 调用你的特征提取函数
feature_vector = extract_text_feature(text)
feature_preview = feature_vector[:20] if len(feature_vector) >= 20 else feature_vector
inference_time = time.time() – text_start_time

results.append({
'text': text,
'feature_vector': [round(float(x), 6) for x in feature_vector],
'feature_preview': [round(float(x), 6) for x in feature_preview],
'inference_time': round(inference_time, 4),
'status': 'success',
'message': '特征提取成功'
})

except Exception as e:
results.append({
'text': text,
'feature_vector': [],
'feature_preview': [],
'inference_time': 0.0,
'status': 'error',
'message': f'特征提取失败：{str(e)}'
})

total_time = time.time() – total_start_time

return {
'results': results,
'total_time': round(total_time, 4),
'total_texts': len(results),
'status': 'success',
'message': f'批量处理完成，成功{len([r for r in results if r["status"] == "success"])}条，失败{len([r for r in results if r["status"] == "error"])}条'
}

except Exception as e:
return {
'results': [],
'total_time': 0.0,
'total_texts': 0,
'status': 'error',
'message': f'批量处理失败：{str(e)}'
}, 500

重要说明：上面的代码中，calculate_similarity_score()和extract_text_feature()是你原有的相似度计算和特征提取函数。你需要确保这些函数已经正确导入，或者将函数实现直接放在app.py中。

2.5 添加OpenAPI规范文件路由

为了让Swagger UI能够获取到API的描述信息，我们需要添加一个生成OpenAPI规范文件的路由。在app.py的最后（在if __name__ == '__main__':之前）添加：

# ========== 新增：生成OpenAPI规范文件 ==========
@app.route('/api/swagger.json')
def swagger_json():
"""
生成OpenAPI规范文件

Swagger UI通过这个端点获取API的完整描述信息。
"""
return jsonify(api.__schema__)

2.6 测试Swagger集成

完成以上修改后，保存app.py文件，然后重启你的Flask应用：

# 如果你之前是直接运行python app.py
python app.py

# 或者如果你使用gunicorn等WSGI服务器
gunicorn -w 4 -b 0.0.0.0:6007 app:app

应用启动后，打开浏览器，访问以下地址：

• Web界面：http://你的服务器IP:6007
• Swagger API文档：http://你的服务器IP:6007/api/docs

你应该能看到一个美观的Swagger UI界面，左侧是API接口列表，右侧是接口的详细信息和测试界面。

3. 编写API测试用例

有了Swagger文档，我们还需要编写测试用例来确保API的稳定性和正确性。这里我们使用Python的unittest框架和requests库来编写测试。

3.1 创建测试目录和文件

在你的项目根目录下创建测试目录和文件：

mkdir tests
touch tests/__init__.py
touch tests/test_api.py

3.2 编写基础测试类

打开tests/test_api.py，编写基础测试类：

import unittest
import json
import time
import requests

class TestStructBERTAPI(unittest.TestCase):
"""StructBERT API测试类"""

# API基础URL – 根据你的实际部署地址修改
BASE_URL = "http://localhost:6007"

# 测试用的文本数据
TEST_TEXT1 = "今天天气真好，阳光明媚"
TEST_TEXT2 = "天气晴朗，万里无云"
TEST_TEXT3 = "我喜欢吃苹果和香蕉"
TEST_TEXT4 = "水果中我最喜欢苹果"

def setUp(self):
"""测试前的准备工作"""
self.session = requests.Session()
# 设置请求超时时间
self.timeout = 30

def tearDown(self):
"""测试后的清理工作"""
self.session.close()

def test_01_api_health(self):
"""测试API服务是否正常"""
print("\\n=== 测试API服务健康状态 ===")

try:
response = self.session.get(f"{self.BASE_URL}/", timeout=self.timeout)
self.assertEqual(response.status_code, 200)
print("✓ API服务健康检查通过")
except Exception as e:
self.fail(f"API服务不可用: {str(e)}")

def test_02_swagger_ui(self):
"""测试Swagger UI是否可访问"""
print("\\n=== 测试Swagger UI ===")

try:
response = self.session.get(f"{self.BASE_URL}/api/docs", timeout=self.timeout)
self.assertEqual(response.status_code, 200)
print("✓ Swagger UI可正常访问")
except Exception as e:
self.fail(f"Swagger UI不可访问: {str(e)}")

def test_03_openapi_spec(self):
"""测试OpenAPI规范文件是否可访问"""
print("\\n=== 测试OpenAPI规范文件 ===")

try:
response = self.session.get(f"{self.BASE_URL}/api/swagger.json", timeout=self.timeout)
self.assertEqual(response.status_code, 200)

# 验证返回的是有效的JSON
spec = response.json()
self.assertIn('openapi', spec)
self.assertIn('info', spec)
self.assertIn('paths', spec)
print("✓ OpenAPI规范文件有效")
except Exception as e:
self.fail(f"OpenAPI规范文件无效: {str(e)}")

3.3 编写语义相似度计算测试

继续在test_api.py中添加相似度计算测试：

def test_04_similarity_calculate_basic(self):
"""测试基础语义相似度计算"""
print("\\n=== 测试基础语义相似度计算 ===")

url = f"{self.BASE_URL}/api/similarity/calculate"
headers = {'Content-Type': 'application/json'}

# 测试用例1：相似文本
print("测试用例1：相似文本（天气相关）")
data = {
"text1": self.TEST_TEXT1,
"text2": self.TEST_TEXT2
}

try:
start_time = time.time()
response = self.session.post(url, json=data, headers=headers, timeout=self.timeout)
elapsed_time = time.time() – start_time

self.assertEqual(response.status_code, 200)
result = response.json()

# 验证返回字段
self.assertIn('similarity_score', result)
self.assertIn('similarity_level', result)
self.assertIn('inference_time', result)
self.assertIn('status', result)
self.assertEqual(result['status'], 'success')

# 验证相似度得分在合理范围内
self.assertGreaterEqual(result['similarity_score'], 0)
self.assertLessEqual(result['similarity_score'], 1)

# 验证推理时间合理
self.assertGreater(result['inference_time'], 0)
self.assertLess(result['inference_time'], 5) # 假设单次推理不超过5秒

print(f" 相似度得分: {result['similarity_score']}")
print(f" 相似度等级: {result['similarity_level']}")
print(f" 推理时间: {result['inference_time']}秒")
print(f" 请求总耗时: {elapsed_time:.2f}秒")

except Exception as e:
self.fail(f"相似文本测试失败: {str(e)}")

# 测试用例2：不相似文本
print("\\n测试用例2：不相似文本（天气 vs 水果）")
data = {
"text1": self.TEST_TEXT1,
"text2": self.TEST_TEXT3
}

try:
response = self.session.post(url, json=data, headers=headers, timeout=self.timeout)
self.assertEqual(response.status_code, 200)
result = response.json()

# 不相似文本的得分应该较低
self.assertLess(result['similarity_score'], 0.5)
print(f" 不相似文本得分: {result['similarity_score']} (预期<0.5)")

except Exception as e:
self.fail(f"不相似文本测试失败: {str(e)}")

# 测试用例3：空文本
print("\\n测试用例3：空文本处理")
data = {
"text1": "",
"text2": "测试文本"
}

try:
response = self.session.post(url, json=data, headers=headers, timeout=self.timeout)
# 空文本应该返回400错误
self.assertEqual(response.status_code, 400)
result = response.json()
self.assertEqual(result['status'], 'error')
print(" 空文本正确处理，返回错误状态")

except Exception as e:
self.fail(f"空文本测试失败: {str(e)}")

def test_05_similarity_calculate_with_threshold(self):
"""测试带阈值的语义相似度计算"""
print("\\n=== 测试带阈值的语义相似度计算 ===")

url = f"{self.BASE_URL}/api/similarity/calculate"
headers = {'Content-Type': 'application/json'}

# 测试自定义阈值
data = {
"text1": self.TEST_TEXT4,
"text2": "苹果是一种常见的水果",
"threshold_high": 0.8, # 提高高相似度阈值
"threshold_low": 0.4 # 提高低相似度阈值
}

try:
response = self.session.post(url, json=data, headers=headers, timeout=self.timeout)
self.assertEqual(response.status_code, 200)
result = response.json()

print(f" 文本1: {data['text1']}")
print(f" 文本2: {data['text2']}")
print(f" 相似度得分: {result['similarity_score']}")
print(f" 相似度等级: {result['similarity_level']}")
print(f" 使用阈值: 高={data['threshold_high']}, 低={data['threshold_low']}")

# 验证阈值生效
score = result['similarity_score']
if score >= 0.8:
self.assertEqual(result['similarity_level'], 'high')
elif score >= 0.4:
self.assertEqual(result['similarity_level'], 'medium')
else:
self.assertEqual(result['similarity_level'], 'low')

print(" 阈值设置生效")

except Exception as e:
self.fail(f"带阈值测试失败: {str(e)}")

3.4 编写特征提取测试

继续添加特征提取测试：

def test_06_feature_extract_single(self):
"""测试单文本特征提取"""
print("\\n=== 测试单文本特征提取 ===")

url = f"{self.BASE_URL}/api/feature/extract"
headers = {'Content-Type': 'application/json'}

# 测试正常文本
data = {
"text": "深度学习是人工智能的一个重要分支"
}

try:
start_time = time.time()
response = self.session.post(url, json=data, headers=headers, timeout=self.timeout)
elapsed_time = time.time() – start_time

self.assertEqual(response.status_code, 200)
result = response.json()

# 验证返回字段
self.assertIn('feature_vector', result)
self.assertIn('feature_preview', result)
self.assertIn('inference_time', result)
self.assertIn('status', result)
self.assertEqual(result['status'], 'success')

# 验证特征向量维度
self.assertEqual(len(result['feature_vector']), 768)
self.assertEqual(len(result['feature_preview']), 20)

# 验证预览是前20维
self.assertEqual(result['feature_preview'], result['feature_vector'][:20])

print(f" 输入文本: {data['text']}")
print(f" 特征向量维度: {len(result['feature_vector'])}")
print(f" 特征预览维度: {len(result['feature_preview'])}")
print(f" 推理时间: {result['inference_time']}秒")
print(f" 请求总耗时: {elapsed_time:.2f}秒")

# 验证特征值在合理范围内
for value in result['feature_preview']:
self.assertIsInstance(value, float)
# BERT特征通常在一定范围内
self.assertGreater(value, -10)
self.assertLess(value, 10)

print(" 特征值范围验证通过")

except Exception as e:
self.fail(f"单文本特征提取测试失败: {str(e)}")

# 测试空文本
print("\\n测试空文本处理")
data = {"text": ""}

try:
response = self.session.post(url, json=data, headers=headers, timeout=self.timeout)
self.assertEqual(response.status_code, 400)
result = response.json()
self.assertEqual(result['status'], 'error')
print(" 空文本正确处理，返回错误状态")

except Exception as e:
self.fail(f"空文本测试失败: {str(e)}")

def test_07_feature_extract_batch(self):
"""测试批量特征提取"""
print("\\n=== 测试批量特征提取 ===")

url = f"{self.BASE_URL}/api/feature/batch-extract"
headers = {'Content-Type': 'application/json'}

# 准备测试数据
batch_texts = [
"今天天气真好，适合出去散步",
"人工智能正在改变世界",
"机器学习是人工智能的核心技术",
"自然语言处理让计算机理解人类语言",
"计算机视觉让机器看懂世界"
]

data = {
"batch_texts": batch_texts
}

try:
start_time = time.time()
response = self.session.post(url, json=data, headers=headers, timeout=self.timeout)
elapsed_time = time.time() – start_time

self.assertEqual(response.status_code, 200)
result = response.json()

# 验证返回字段
self.assertIn('results', result)
self.assertIn('total_time', result)
self.assertIn('total_texts', result)
self.assertIn('status', result)
self.assertEqual(result['status'], 'success')

# 验证结果数量
self.assertEqual(len(result['results']), len(batch_texts))
self.assertEqual(result['total_texts'], len(batch_texts))

print(f" 批量文本数量: {len(batch_texts)}")
print(f" 处理结果数量: {len(result['results'])}")
print(f" 总处理时间: {result['total_time']}秒")
print(f" 请求总耗时: {elapsed_time:.2f}秒")

# 验证每个结果
success_count = 0
for i, item in enumerate(result['results']):
if item['status'] == 'success':
success_count += 1
self.assertEqual(len(item['feature_vector']), 768)
self.assertEqual(len(item['feature_preview']), 20)
self.assertEqual(item['feature_preview'], item['feature_vector'][:20])
else:
print(f" 第{i+1}条文本处理失败: {item['message']}")

print(f" 成功处理: {success_count}/{len(batch_texts)}")

# 验证批量处理的效率
# 批量处理总时间应该小于单条处理时间之和
self.assertLess(result['total_time'], 5 * 2) # 假设单条最多2秒，5条应该小于10秒

except Exception as e:
self.fail(f"批量特征提取测试失败: {str(e)}")

# 测试空批量
print("\\n测试空批量处理")
data = {"batch_texts": []}

try:
response = self.session.post(url, json=data, headers=headers, timeout=self.timeout)
self.assertEqual(response.status_code, 400)
result = response.json()
self.assertEqual(result['status'], 'error')
print(" 空批量正确处理，返回错误状态")

except Exception as e:
self.fail(f"空批量测试失败: {str(e)}")

# 测试无效格式
print("\\n测试无效格式处理")
data = {"batch_texts": "这不是一个列表"}

try:
response = self.session.post(url, json=data, headers=headers, timeout=self.timeout)
self.assertEqual(response.status_code, 400)
result = response.json()
self.assertEqual(result['status'], 'error')
print(" 无效格式正确处理，返回错误状态")

except Exception as e:
self.fail(f"无效格式测试失败: {str(e)}")

3.5 编写性能测试

添加性能测试用例：

def test_08_performance_test(self):
"""测试API性能"""
print("\\n=== 测试API性能 ===")

similarity_url = f"{self.BASE_URL}/api/similarity/calculate"
feature_url = f"{self.BASE_URL}/api/feature/extract"
headers = {'Content-Type': 'application/json'}

# 准备测试数据
test_cases = [
{"text1": "苹果手机", "text2": "iPhone"},
{"text1": "今天天气不错", "text2": "天气很好"},
{"text1": "人工智能", "text2": "AI技术"},
{"text1": "机器学习", "text2": "深度学习"},
{"text1": "自然语言处理", "text2": "NLP"}
]

# 测试相似度计算性能
print("测试相似度计算性能（5次请求）:")
similarity_times = []

for i, test_case in enumerate(test_cases, 1):
try:
start_time = time.time()
response = self.session.post(
similarity_url,
json=test_case,
headers=headers,
timeout=self.timeout
)
elapsed_time = time.time() – start_time

if response.status_code == 200:
result = response.json()
similarity_times.append(result['inference_time'])
print(f" 第{i}次: {result['inference_time']:.4f}秒 (相似度: {result['similarity_score']:.4f})")
else:
print(f" 第{i}次: 请求失败")

except Exception as e:
print(f" 第{i}次: 异常 – {str(e)}")

if similarity_times:
avg_time = sum(similarity_times) / len(similarity_times)
max_time = max(similarity_times)
min_time = min(similarity_times)
print(f" 平均推理时间: {avg_time:.4f}秒")
print(f" 最长推理时间: {max_time:.4f}秒")
print(f" 最短推理时间: {min_time:.4f}秒")

# 性能断言：平均推理时间应小于1秒
self.assertLess(avg_time, 1.0, "相似度计算平均时间超过1秒")

# 测试特征提取性能
print("\\n测试特征提取性能（5次请求）:")
feature_times = []
test_texts = [
"这是一个测试文本，用于特征提取性能测试",
"深度学习模型需要大量的计算资源",
"自然语言处理是人工智能的重要方向",
"计算机视觉可以识别图像中的物体",
"机器学习算法可以从数据中学习规律"
]

for i, text in enumerate(test_texts, 1):
try:
start_time = time.time()
response = self.session.post(
feature_url,
json={"text": text},
headers=headers,
timeout=self.timeout
)
elapsed_time = time.time() – start_time

if response.status_code == 200:
result = response.json()
feature_times.append(result['inference_time'])
print(f" 第{i}次: {result['inference_time']:.4f}秒")
else:
print(f" 第{i}次: 请求失败")

except Exception as e:
print(f" 第{i}次: 异常 – {str(e)}")

if feature_times:
avg_time = sum(feature_times) / len(feature_times)
max_time = max(feature_times)
min_time = min(feature_times)
print(f" 平均推理时间: {avg_time:.4f}秒")
print(f" 最长推理时间: {max_time:.4f}秒")
print(f" 最短推理时间: {min_time:.4f}秒")

# 性能断言：平均推理时间应小于1秒
self.assertLess(avg_time, 1.0, "特征提取平均时间超过1秒")

3.6 添加测试运行入口

在test_api.py文件末尾添加：

def run_tests():
"""运行所有测试"""
print("=" * 60)
print("StructBERT API 测试开始")
print("=" * 60)

# 创建测试套件
loader = unittest.TestLoader()
suite = loader.loadTestsFromTestCase(TestStructBERTAPI)

# 运行测试
runner = unittest.TextTestRunner(verbosity=2)
result = runner.run(suite)

print("=" * 60)
print("测试完成")
print(f"运行测试: {result.testsRun}")
print(f"失败测试: {len(result.failures)}")
print(f"错误测试: {len(result.errors)}")
print("=" * 60)

return result.wasSuccessful()

if __name__ == '__main__':
# 检查API服务是否可用
try:
response = requests.get("http://localhost:6007/", timeout=5)
if response.status_code == 200:
print("✓ API服务运行正常，开始测试…")
success = run_tests()
sys.exit(0 if success else 1)
else:
print("✗ API服务返回异常状态码")
sys.exit(1)
except requests.exceptions.ConnectionError:
print("✗ 无法连接到API服务，请确保服务已启动")
print(" 运行命令: python app.py 或 gunicorn -w 4 -b 0.0.0.0:6007 app:app")
sys.exit(1)
except Exception as e:
print(f"✗ 检查API服务时发生错误: {str(e)}")
sys.exit(1)

3.7 运行测试

确保你的StructBERT服务正在运行，然后在终端中执行：

cd /path/to/your/project
python -m tests.test_api

或者直接运行：

python tests/test_api.py

你应该能看到详细的测试输出，包括每个测试用例的执行结果和性能数据。

4. 总结与最佳实践

4.1 本文实现的核心价值

通过本文的实践，我们为StructBERT中文语义智能匹配系统实现了两个重要的工程化改进：

4.2 实际使用建议

在实际项目中，我建议你：

4.3 遇到的常见问题与解决

在实施过程中，你可能会遇到以下问题：

4.4 下一步优化方向

完成基础集成后，你还可以考虑以下优化：

通过本文的实践，你的StructBERT系统不仅功能强大，而且具备了专业级的API文档和测试保障，为团队协作和系统维护打下了坚实基础。现在，其他开发者可以轻松理解和使用你的API，你也可以自信地进行系统迭代和优化了。

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。