SiameseUniNLU实战手册：基于app.py源码定制返回字段与响应格式开发指引-育师

SiameseUniNLU实战手册：基于app.py源码定制返回字段与响应格式开发指引

1. 为什么需要定制SiameseUniNLU的返回格式？

当你第一次使用SiameseUniNLU模型时，可能会发现它的API返回格式比较固定。比如，你调用命名实体识别，它返回的是标准的结构化数据。但实际项目中，我们往往需要更多信息：比如每个实体的置信度、在原文中的位置、或者按照业务需求重新组织返回格式。

举个例子，如果你在做智能客服系统，不仅需要识别出用户提到的“产品名称”，还需要知道用户对这个产品的“情感倾向”，并且要把这两个信息打包成一个完整的“用户意图”返回给下游系统。这时候，默认的返回格式就不够用了。

这就是为什么我们需要深入app.py源码，学习如何定制返回字段和响应格式。通过本文的实战指引，你将掌握：

理解SiameseUniNLU的核心处理流程
找到修改返回数据的关键代码位置
学会添加自定义字段（如置信度、位置信息）
掌握重新组织响应结构的技巧
了解不同任务类型的定制差异

无论你是要将模型集成到现有系统，还是需要满足特定的数据格式要求，这篇指南都能给你清晰的路径。

2. 快速理解SiameseUniNLU的工作机制

在动手修改之前，我们先花几分钟搞清楚这个模型是怎么工作的。这样你修改代码时，就知道每一处改动的影响是什么。

2.1 核心思路：提示（Prompt）+文本（Text）

SiameseUniNLU的聪明之处在于它用一个统一的框架处理多种任务。它的核心思路很简单：

你告诉它要做什么：通过Schema（模式）来提示。比如{"人物":null,"地理位置":null}就是在说：“请从文本里找出人物和地理位置”。
你给它要看的内容：就是输入的文本。
它用指针网络找答案：模型内部使用指针网络（Pointer Network）在文本中“划出”相关的片段（Span）。比如“谷爱凌在北京冬奥会获得金牌”，指针网络会识别出“谷爱凌”对应“人物”，“北京冬奥会”可能关联到“事件”或“地理位置”（取决于你的Schema）。

这种“提示+文本”的方式，让同一个模型能处理命名实体识别、关系抽取、情感分类等十多种任务，而不需要为每个任务单独训练一个模型。

2.2 app.py源码的骨架解析

我们来看看/root/nlp_structbert_siamese-uninlu_chinese-base/app.py这个文件的主要结构。理解这个结构，你就能快速定位到需要修改的地方。

# 以下是app.py的核心结构示意（非完整代码） from flask import Flask, request, jsonify import torch from transformers import AutoTokenizer, AutoModel # ... 其他导入 app = Flask(__name__) # 1. 模型加载部分 print("正在加载模型...") model_path = "/root/ai-models/iic/nlp_structbert_siamese-uninlu_chinese-base" tokenizer = AutoTokenizer.from_pretrained(model_path) model = AutoModel.from_pretrained(model_path) print("模型加载完成！") # 2. 核心预测函数 def predict(text, schema): """ 这里是实际调用模型进行预测的核心逻辑 输入：文本和schema 输出：预测结果 """ # 将文本和schema编码为模型可理解的格式 inputs = tokenizer(text, return_tensors="pt") # 模型推理 with torch.no_grad(): outputs = model(**inputs) # 后处理：根据schema类型解析输出 result = post_process(outputs, schema) return result # 3. 后处理函数（关键！） def post_process(model_outputs, schema): """ 将模型的原始输出转换为用户友好的格式 这是我们要重点修改的部分 """ # 默认实现：提取实体、关系等 # ... 具体解析逻辑 return formatted_result # 4. Flask API路由 @app.route('/api/predict', methods=['POST']) def api_predict(): data = request.get_json() text = data.get('text', '') schema = data.get('schema', '{}') # 调用预测函数 result = predict(text, schema) # 返回JSON响应（这是另一个可以修改格式的地方） return jsonify({ "code": 200, "msg": "success", "data": result }) # 5. 启动Web界面（Gradio） # ... Gradio界面定义代码 if __name__ == '__main__': app.run(host='0.0.0.0', port=7860)

从上面可以看出，我们要定制返回格式，主要关注两个地方：

post_process函数：这里决定了从模型原始输出到结构化结果的转换逻辑。
API返回部分：jsonify调用处，这里决定了最终的HTTP响应格式。

3. 实战开始：修改返回字段与格式

现在我们来动手修改。我会以“命名实体识别”任务为例，展示三种常见的定制需求。你可以根据自己的需要选择修改。

3.1 案例一：为实体添加置信度分数

默认情况下，模型只返回识别出的实体，但没有告诉咱们这个识别有多大的把握。在很多业务场景里，置信度很重要——比如低置信度的结果可能需要人工复核。

修改步骤：

找到后处理函数：在app.py中搜索post_process或类似名称的函数。
理解原始输出：在函数内部，模型输出通常包含每个位置的预测分数。我们需要提取这些分数。
修改代码：

# 修改后的post_process函数片段（以命名实体识别为例） def post_process_ner(model_outputs, schema, text): """ 处理命名实体识别任务，添加置信度 """ # 假设model_outputs包含预测分数（logits） # logits形状通常是 [batch_size, seq_length, num_labels] logits = model_outputs.logits predictions = torch.argmax(logits, dim=-1) # 获取预测标签 probabilities = torch.softmax(logits, dim=-1) # 转换为概率 entities = [] current_entity = None # 遍历每个位置 for i in range(len(predictions[0])): label_id = predictions[0][i].item() label_name = id_to_label(label_id) # 需要实现这个映射函数 # 获取当前位置的置信度（取最大概率值） confidence = torch.max(probabilities[0][i]).item() if label_name.startswith("B-"): # 实体开始 if current_entity: entities.append(current_entity) current_entity = { "text": text[i], # 需要从token映射回原始文本 "type": label_name[2:], # 去掉"B-"前缀 "start": i, "confidence": confidence } elif label_name.startswith("I-") and current_entity: # 实体继续，更新文本和置信度（取平均或最大） current_entity["text"] += text[i] current_entity["confidence"] = max(current_entity["confidence"], confidence) elif label_name == "O" and current_entity: # 实体结束 current_entity["end"] = i - 1 entities.append(current_entity) current_entity = None # 处理最后一个实体 if current_entity: current_entity["end"] = len(predictions[0]) - 1 entities.append(current_entity) return entities # 在API返回处，使用新的处理函数 @app.route('/api/predict', methods=['POST']) def api_predict(): data = request.get_json() text = data.get('text', '') schema = data.get('schema', '{}') # 解析schema判断任务类型 schema_dict = json.loads(schema) if "人物" in schema_dict or "地理位置" in schema_dict: # 简单判断是否为NER # 调用带置信度的处理函数 result = post_process_ner(model_outputs, schema_dict, text) else: result = predict(text, schema) # 其他任务使用默认处理 return jsonify({ "code": 200, "msg": "success", "data": { "entities": result, "text": text, "schema": schema_dict, "timestamp": datetime.now().isoformat() # 添加时间戳 } })

修改后的效果：调用API后，返回的数据会包含每个实体的置信度分数（0到1之间），以及额外的元数据如时间戳。

3.2 案例二：重新组织响应结构

有时候，下游系统需要特定格式的数据。比如前端希望实体按照类型分组，或者需要扁平化的结构。

需求示例：默认返回格式：

[ {"text": "谷爱凌", "type": "人物", "start": 0, "end": 2}, {"text": "北京冬奥会", "type": "事件", "start": 3, "end": 5} ]

希望改为按类型分组：

{ "人物": ["谷爱凌"], "事件": ["北京冬奥会"], "统计": { "总实体数": 2, "人物数量": 1, "事件数量": 1 } }

修改方法：

def reorganize_entities_by_type(entities): """ 将实体列表按类型重新组织 """ grouped = {} stats = { "total": len(entities), "by_type": {} } for entity in entities: entity_type = entity["type"] entity_text = entity["text"] # 按类型分组 if entity_type not in grouped: grouped[entity_type] = [] grouped[entity_type].append(entity_text) # 统计 if entity_type not in stats["by_type"]: stats["by_type"][entity_type] = 0 stats["by_type"][entity_type] += 1 return { "grouped_entities": grouped, "statistics": stats } # 在API返回处调用 @app.route('/api/predict', methods=['POST']) def api_predict(): # ... 获取数据和预测 # 如果是NER任务且需要分组格式 format_type = data.get('format', 'default') # 通过参数控制格式 if format_type == 'grouped': reorganized = reorganize_entities_by_type(result) final_result = reorganized else: final_result = result return jsonify({ "code": 200, "msg": "success", "data": final_result, "format": format_type # 返回使用的格式类型 })

使用方式：现在你可以通过API参数控制返回格式：

import requests data = { "text": "谷爱凌在北京冬奥会获得金牌", "schema": '{"人物":null,"事件":null}', "format": "grouped" # 指定分组格式 } response = requests.post("http://localhost:7860/api/predict", json=data)

3.3 案例三：为不同任务定制不同字段

SiameseUniNLU支持多种任务，不同任务可能需要不同的附加信息。比如：

关系抽取：需要关系类型和置信度
情感分类：需要正面/负面分数和阈值判断
阅读理解：需要答案片段和上下文支持度

实现思路：

def custom_post_process(model_outputs, schema, text, task_type): """ 根据任务类型定制后处理 """ if task_type == "ner": # 命名实体识别：添加位置和置信度 result = process_ner_with_confidence(model_outputs, text) # 添加额外字段 for entity in result: entity["source"] = "SiameseUniNLU" entity["version"] = "1.0" return result elif task_type == "relation": # 关系抽取：添加关系强度和证据 result = process_relation_extraction(model_outputs, text) for relation in result: relation["strength"] = calculate_relation_strength(relation) relation["evidence"] = extract_evidence(text, relation) return result elif task_type == "sentiment": # 情感分类：添加分数和标签 scores = process_sentiment(model_outputs) result = { "scores": scores, "label": "正向" if scores["positive"] > 0.5 else "负向", "confidence": abs(scores["positive"] - 0.5) * 2 # 置信度计算 } return result else: # 默认处理 return default_post_process(model_outputs, schema) def detect_task_type(schema_dict): """ 根据schema自动检测任务类型 """ schema_str = json.dumps(schema_dict) if '"人物"' in schema_str and '"比赛项目"' in schema_str: return "relation" elif '"情感分类"' in schema_str: return "sentiment" elif '"人物"' in schema_str or '"地理位置"' in schema_str: return "ner" else: return "unknown" # 在预测函数中使用 def predict(text, schema): schema_dict = json.loads(schema) if isinstance(schema, str) else schema task_type = detect_task_type(schema_dict) # 模型推理... model_outputs = model(**inputs) # 使用定制后处理 result = custom_post_process(model_outputs, schema_dict, text, task_type) return result

4. 完整示例：修改后的app.py关键部分

为了让修改更清晰，这里提供一个修改后的app.py关键部分示例。请注意，这只是一个示例，你需要根据实际模型输出结构进行调整。

import json from datetime import datetime from flask import Flask, request, jsonify import torch from transformers import AutoTokenizer, AutoModelForTokenClassification # 根据实际模型调整 app = Flask(__name__) # 加载模型 model_path = "/root/ai-models/iic/nlp_structbert_siamese-uninlu_chinese-base" tokenizer = AutoTokenizer.from_pretrained(model_path) model = AutoModelForTokenClassification.from_pretrained(model_path) model.eval() def predict_with_custom_format(text, schema, format_type="default"): """ 支持自定义格式的预测函数 """ # 1. 准备输入 inputs = tokenizer(text, return_tensors="pt", truncation=True, max_length=512) # 2. 模型推理 with torch.no_grad(): outputs = model(**inputs) # 3. 解析schema确定任务 try: schema_dict = json.loads(schema) if isinstance(schema, str) else schema except: schema_dict = {} # 4. 根据任务和格式要求处理后处理 task_type = detect_task_type(schema_dict) if task_type == "ner": # 基础实体识别 entities = extract_entities(outputs, text) # 根据格式要求进一步处理 if format_type == "with_confidence": entities = add_confidence_scores(entities, outputs) elif format_type == "grouped": entities = group_entities_by_type(entities) elif format_type == "detailed": entities = add_detailed_info(entities, text) result = entities elif task_type == "relation": # 关系抽取 relations = extract_relations(outputs, text, schema_dict) if format_type == "detailed": relations = add_relation_evidence(relations, text) result = relations else: # 其他任务使用默认处理 result = default_extraction(outputs, schema_dict) return { "task": task_type, "result": result, "original_text": text, "schema": schema_dict, "processed_at": datetime.now().isoformat() } @app.route('/api/v2/predict', methods=['POST']) def api_predict_v2(): """ 增强版API，支持格式定制 """ try: data = request.get_json() if not data: return jsonify({ "code": 400, "msg": "请求体不能为空", "data": None }) text = data.get('text', '') schema = data.get('schema', '{}') format_type = data.get('format', 'default') # 新增：格式参数 include_metadata = data.get('include_metadata', True) # 新增：是否包含元数据 if not text: return jsonify({ "code": 400, "msg": "文本内容不能为空", "data": None }) # 调用预测函数 prediction_result = predict_with_custom_format(text, schema, format_type) # 构建响应 response_data = { "prediction": prediction_result["result"] } if include_metadata: response_data["metadata"] = { "task_type": prediction_result["task"], "text_length": len(text), "schema_used": prediction_result["schema"], "processing_time": prediction_result["processed_at"], "model_version": "nlp_structbert_siamese-uninlu_chinese-base", "format": format_type } return jsonify({ "code": 200, "msg": "success", "data": response_data }) except Exception as e: return jsonify({ "code": 500, "msg": f"处理失败: {str(e)}", "data": None }) # 辅助函数定义 def extract_entities(model_outputs, text): """提取实体基础函数""" # 这里需要根据实际模型输出结构实现 # 示例伪代码 entities = [] # ... 实体提取逻辑 return entities def add_confidence_scores(entities, model_outputs): """为实体添加置信度""" for entity in entities: # 计算置信度的逻辑 entity["confidence"] = 0.95 # 示例值 return entities def detect_task_type(schema_dict): """检测任务类型""" # 简单检测逻辑 keys = list(schema_dict.keys()) if schema_dict else [] if not keys: return "unknown" # 根据常见schema模式判断 if "情感分类" in keys: return "sentiment" elif any(key in ["人物", "组织机构", "地理位置"] for key in keys): return "ner" elif any("{" in str(value) for value in schema_dict.values()): return "relation" else: return "classification" if __name__ == '__main__': app.run(host='0.0.0.0', port=7860, debug=False)

5. 测试你的修改

修改代码后，一定要充分测试。这里提供几个测试方案：

5.1 基础功能测试

import requests import json # 测试1：基础NER功能 print("测试1：基础命名实体识别") data = { "text": "马云是阿里巴巴集团的创始人，公司位于杭州。", "schema": '{"人物":null,"组织机构":null,"地理位置":null}' } response = requests.post("http://localhost:7860/api/v2/predict", json=data) print("基础响应:", json.dumps(response.json(), indent=2, ensure_ascii=False)) # 测试2：带置信度的格式 print("\n测试2：带置信度的实体识别") data = { "text": "马云是阿里巴巴集团的创始人，公司位于杭州。", "schema": '{"人物":null,"组织机构":null,"地理位置":null}', "format": "with_confidence" } response = requests.post("http://localhost:7860/api/v2/predict", json=data) print("带置信度响应:", json.dumps(response.json(), indent=2, ensure_ascii=False)) # 测试3：分组格式 print("\n测试3：分组格式实体识别") data = { "text": "马云是阿里巴巴集团的创始人，公司位于杭州。", "schema": '{"人物":null,"组织机构":null,"地理位置":null}', "format": "grouped" } response = requests.post("http://localhost:7860/api/v2/predict", json=data) print("分组格式响应:", json.dumps(response.json(), indent=2, ensure_ascii=False)) # 测试4：不包含元数据 print("\n测试4：简洁响应（无元数据）") data = { "text": "这部电影真的很精彩，演员演技出色。", "schema": '{"情感分类":null}', "include_metadata": False } response = requests.post("http://localhost:7860/api/v2/predict", json=data) print("简洁响应:", json.dumps(response.json(), indent=2, ensure_ascii=False))

5.2 错误处理测试

# 测试错误情况 test_cases = [ {"name": "空文本", "data": {"text": "", "schema": '{"人物":null}'}}, {"name": "无效JSON", "data": {"text": "测试文本", "schema": '{人物:null}'}}, # 无效JSON {"name": "超大文本", "data": {"text": "测试" * 1000, "schema": '{"人物":null}'}}, ] for test in test_cases: print(f"\n测试: {test['name']}") try: response = requests.post("http://localhost:7860/api/v2/predict", json=test['data'], timeout=5) print(f"状态码: {response.status_code}") print(f"响应: {response.text[:200]}...") # 只打印前200字符 except Exception as e: print(f"请求异常: {e}")

6. 部署与维护建议

6.1 部署修改后的服务

# 1. 备份原始文件 cp /root/nlp_structbert_siamese-uninlu_chinese-base/app.py /root/nlp_structbert_siamese-uninlu_chinese-base/app.py.backup # 2. 将修改后的代码保存为app_custom.py # 3. 测试新服务 cd /root/nlp_structbert_siamese-uninlu_chinese-base python3 app_custom.py # 4. 如果测试正常，替换原文件（或直接使用新文件） # 5. 后台运行 nohup python3 app_custom.py > server_custom.log 2>&1 & # 6. 查看日志 tail -f server_custom.log

6.2 性能考虑

当你添加了额外的处理逻辑后，可能会影响性能。以下是一些优化建议：

缓存处理结果：对于相同的文本和schema组合，可以缓存处理结果。
批量处理：如果可能，修改代码支持批量文本处理。
异步处理：对于耗时的后处理，可以考虑使用异步任务。
配置开关：通过配置参数控制哪些增强功能开启。

# 示例：添加配置开关 ENABLED_FEATURES = { "confidence_scores": True, "grouping": False, # 默认关闭分组功能 "metadata": True, "caching": True } # 在代码中根据配置决定是否执行某些操作 if ENABLED_FEATURES["confidence_scores"]: result = add_confidence_scores(result)

6.3 版本管理建议

由于你修改了原始代码，建议做好版本管理：

创建分支：在代码管理系统中创建自定义版本分支。
记录修改：在文件头部添加修改记录注释。
配置分离：将可配置的部分提取到单独的文件中。
兼容性保持：保留原始API接口，新增增强接口。

# 在文件头部添加修改记录 """ SiameseUniNLU自定义版本 基于原始app.py修改 修改日期: 2024-01-15 修改内容: 1. 添加实体置信度计算 2. 支持多种返回格式 3. 增强API错误处理 4. 添加元数据信息 原始版本: nlp_structbert_siamese-uninlu_chinese-base 修改者: [你的名字] """ # 保持原始API兼容 @app.route('/api/predict', methods=['POST']) def api_predict_original(): """保持与原始版本兼容的API""" # 调用原始处理逻辑 return jsonify(original_predict_logic()) @app.route('/api/v2/predict', methods=['POST']) def api_predict_enhanced(): """增强版API""" # 调用自定义处理逻辑 return jsonify(enhanced_predict_logic())