4.训练数据API改造方案.md 38 KB
unified_api.py 训练数据管理API改造方案（详细版）
1. 需求概述
本方案基于对 unified_api.py 文件的深入分析，提出以下两个API改造与新增需求：
- 需求一：新增
/api/v0/training_data/update，支持先删除再插入的训练数据更新操作 - 需求二：新增
/api/v0/training_data/upload，支持上传多种格式文件并自动解析入库 - 需求三：新增
/api/v0/training_data/combine，合并相同的 langchain_pg_embedding.document 记录
2. 当前实现分析
2.1 现有API结构
unified_api.py 目前实现了4个训练数据相关的API端点：
GET /api/v0/training_data/stats(lines 1205-1234) - 基础统计信息POST /api/v0/training_data/query(lines 1236-1294) - 分页查询POST /api/v0/training_data/create(lines 1296-1404) - 批量创建（最多50条）POST /api/v0/training_data/delete(lines 1406-1471) - 批量删除（最多50条）
2.2 数据库架构
表结构关系：
-- langchain_pg_collection (集合表)
CREATE TABLE langchain_pg_collection (
uuid uuid PRIMARY KEY,
name varchar NOT NULL UNIQUE,
cmetadata json
);
-- langchain_pg_embedding (训练数据表)
CREATE TABLE langchain_pg_embedding (
id varchar PRIMARY KEY,
collection_id uuid REFERENCES langchain_pg_collection(uuid) ON DELETE CASCADE,
embedding vector,
document varchar,
cmetadata jsonb
);
集合名称映射：
sql集合 → SQL查询和问答对ddl集合 → 数据库表结构定义documentation集合 → 表和字段文档error_sql集合 → 错误SQL示例
3. 详细改造方案
3.1 /api/v0/training_data/update (新增的API)
功能描述：
- 支持单条更新训练数据
- 采用先删除后插入的策略确保数据一致性
- 更新方式是先删除再插入，删除是以 langchain_pg_embedding.id字段为依据的
- langchain_pg_embedding.id字段是varchar类型，它是uuid+类型的格式，比如："
50bb6b17-d5be-48ab-8125-58ab8bb076-sql" - 返回新创建记录的ID，便于前端更新界面
WEB UI使用流程：
- 用户选中一行记录，点击修改按钮
- 前端显示编辑表单，用户修改内容
- 提交时，根据这一行的id，删除这条记录
- 然后调用create参数，生成一条新的记录
- 返回的JSON中包括这条记录的新的id
API规格：
请求参数：
{
"id": "6982ba18-8d0a-4cce-9d3b-922ac6ee10ac-sql",
"training_data_type": "sql",
"question": "如何查询所有用户？",
"sql": "SELECT * FROM users;"
}
参数说明：
id: 要删除的原始记录ID（必需）training_data_type: 训练数据类型，支持 "sql", "ddl", "documentation", "error_sql"- 其他字段根据类型不同而不同：
sql类型：question（可选）,sql（必需）error_sql类型：question（必需）,sql（必需）documentation类型：content（必需）ddl类型：ddl（必需）
实现代码框架：
@app.route('/api/v0/training_data/update', methods=['POST'])
def training_data_update():
"""更新训练数据API - 支持单条更新，采用先删除后插入策略"""
try:
req = request.get_json(force=True)
# 1. 参数验证
original_id = req.get('id')
if not original_id:
return jsonify(bad_request_response(
response_text="缺少必需参数：id",
missing_params=["id"]
)), 400
training_type = req.get('training_data_type')
if not training_type:
return jsonify(bad_request_response(
response_text="缺少必需参数：training_data_type",
missing_params=["training_data_type"]
)), 400
# 2. 先删除原始记录
try:
success = vn.remove_training_data(original_id)
if not success:
return jsonify(bad_request_response(
response_text=f"原始记录 {original_id} 不存在或删除失败"
)), 400
except Exception as e:
return jsonify(internal_error_response(
response_text=f"删除原始记录失败: {str(e)}"
)), 500
# 3. 根据类型验证和准备新数据
try:
if training_type == 'sql':
sql = req.get('sql')
if not sql:
return jsonify(bad_request_response(
response_text="SQL字段是必需的",
missing_params=["sql"]
)), 400
# SQL语法检查
is_valid, error_msg = validate_sql_syntax(sql)
if not is_valid:
return jsonify(bad_request_response(
response_text=f"SQL语法错误: {error_msg}"
)), 400
question = req.get('question')
if question:
training_id = vn.train(question=question, sql=sql)
else:
training_id = vn.train(sql=sql)
elif training_type == 'error_sql':
question = req.get('question')
sql = req.get('sql')
if not question or not sql:
return jsonify(bad_request_response(
response_text="question和sql字段都是必需的",
missing_params=["question", "sql"]
)), 400
training_id = vn.train_error_sql(question=question, sql=sql)
elif training_type == 'documentation':
content = req.get('content')
if not content:
return jsonify(bad_request_response(
response_text="content字段是必需的",
missing_params=["content"]
)), 400
training_id = vn.train(documentation=content)
elif training_type == 'ddl':
ddl = req.get('ddl')
if not ddl:
return jsonify(bad_request_response(
response_text="ddl字段是必需的",
missing_params=["ddl"]
)), 400
training_id = vn.train(ddl=ddl)
else:
return jsonify(bad_request_response(
response_text=f"不支持的训练数据类型: {training_type}"
)), 400
except Exception as e:
return jsonify(internal_error_response(
response_text=f"创建新训练数据失败: {str(e)}"
)), 500
# 4. 获取更新后的总记录数
current_total = get_total_training_count()
return jsonify(success_response(
response_text="训练数据更新成功",
data={
"original_id": original_id,
"new_training_id": training_id,
"type": training_type,
"current_total_count": current_total
}
))
except Exception as e:
logger.error(f"training_data_update执行失败: {str(e)}")
return jsonify(internal_error_response(
response_text="更新训练数据失败，请稍后重试"
)), 500
返回格式：
成功响应：
{
"code": 200,
"success": true,
"message": "操作成功",
"data": {
"response": "训练数据更新成功",
"original_id": "6982ba18-8d0a-4cce-9d3b-922ac6ee10ac-sql",
"new_training_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890-sql",
"type": "sql",
"current_total_count": 1250
}
}
错误响应示例：
{
"code": 400,
"success": false,
"message": "请求参数错误",
"data": {
"response": "原始记录 6982ba18-8d0a-4cce-9d3b-922ac6ee10ac-sql 不存在或删除失败",
"error_type": "INVALID_PARAMS",
"timestamp": "2025-01-15T10:30:00.000Z",
"can_retry": false
}
}
错误响应示例：
{
"code": 400,
"success": false,
"message": "请求参数错误",
"data": {
"response": "缺少必需参数：data",
"error_type": "INVALID_PARAMS",
"timestamp": "2025-01-15T10:30:00.000Z",
"can_retry": false
}
}
安全性和性能考虑：
事务安全：
- 删除和插入操作在同一个函数中执行
- 如果插入失败，原始记录已被删除，需要回滚或重试机制
- 建议在应用层实现重试逻辑
数据一致性：
- 先删除后插入确保ID的唯一性
- 新记录会获得新的UUID，避免ID冲突
- 保持训练数据类型的正确性
错误处理：
- 详细的参数验证
- 原始记录存在性检查
- SQL语法验证（针对sql类型）
- 完整的异常处理和错误信息
单条处理：
- 只支持单条更新，简化逻辑
- 减少并发冲突的可能性
- 更精确的错误定位和处理
性能优化：
- 复用现有的训练函数
- 避免重复的数据库连接
- 合理的错误处理和日志记录
使用建议：
前端集成：
// 更新训练数据的示例 const updateTrainingData = async (originalId, newData) => { const response = await fetch('/api/v0/training_data/update', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ id: originalId, training_data_type: 'sql', question: newData.question, sql: newData.sql }) }); const result = await response.json(); if (result.success) { // 更新成功，使用新的training_id更新界面 const newId = result.data.new_training_id; updateUIWithNewId(newId); } };- 错误处理：
- 检查返回的
success字段 - 处理各种错误情况（参数错误、记录不存在、SQL语法错误等）
显示详细的错误信息给用户
最佳实践：
更新前验证原始记录是否存在
更新后验证新记录是否创建成功
使用返回的新ID更新前端界面
3.2
/api/v0/training_data/upload新增（重点设计）功能描述：
- 支持上传多种格式的训练数据文件（.ddl/.md/.json/.sql）
- 直接内存处理，无需临时文件存储
- 文件大小限制500KB（约包含100-500条记录）
- 自动识别文件类型并解析内容
- 复用现有trainer模块的成熟功能
- 提供详细的处理结果反馈
支持的文件类型：
基于现有
data_pipeline/trainer模块的成熟功能，API 支持以下文件类型的自动识别和处理：文件类型 扩展名 文件命名规则 复用函数 解析方式 ddl .ddl任意名称 train_ddl_file()→read_file_by_delimiter()建表语句，以“；”为分隔符 documentation .md任意名称 train_documentation_file()→read_markdown_file_by_sections()Markdown以"##"为单位分割，普通文档以"---"分隔 question_sql_pairs .json必须包含 _pair或_pairstrain_json_pairs_file()→train_json_question_sql_pairs()问答对训练（JSON格式） formatted_pairs .sql必须包含 _pair或_pairstrain_formatted_pairs_file()→train_formatted_question_sql_pairs()问答对训练（文本格式），每对Question/SQL之间使用空行分隔 sql_examples .sql任意名称（不包含 _pair）train_sql_file()→read_file_by_delimiter()SQL示例，以"；"为分隔符 1. DDL文件 (*.ddl)：
建表语句，解析时以“；”为分隔符。
-- 建表语句 CREATE TABLE users ( id SERIAL PRIMARY KEY, name VARCHAR(100) NOT NULL, email VARCHAR(255) UNIQUE ); CREATE TABLE orders ( id SERIAL PRIMARY KEY, user_id INTEGER REFERENCES users(id), amount DECIMAL(10,2) );
2. 文档文件 (*.md)：
以"##"为单位进行分割，每个"##"是一个表名：
## users表
用户基本信息表，存储系统用户数据。
字段说明
- id: 用户唯一标识
- name: 用户姓名
- email: 电子邮箱
## orders表
订单信息表，记录用户购买记录。
3. JSON问答对 (*_pair.json)：
"question"/"sql" 大小写不敏感。
[
{
"question": "如何查询活跃用户数量？",
"sql": "SELECT COUNT(*) FROM users WHERE active = true;"
},
{
"question": "如何统计每个分类的产品数量？",
"sql": "SELECT category, COUNT(*) FROM products GROUP BY category;"
}
]
4. 格式化问答对 (*_pair.sql)：
注意，Question和SQL，大小写不敏感，但是两个问答对之间要以空行作为分隔。
Question: 如何查询活跃用户数量？
SQL:SELECT COUNT(*) FROM users WHERE active = true;
Question: 如何统计每个分类的产品数量？
SQL:SELECT category, COUNT(*) FROM products GROUP BY category;
API规格：
请求参数：
file: 上传的文件（multipart/form-data）- 文件大小限制：最大500KB
- 文件编码：UTF-8
- 自动识别文件类型：基于文件扩展名和命名规则
实现代码框架：
from werkzeug.utils import secure_filename
from data_pipeline.trainer.run_training import (
read_file_by_delimiter,
read_markdown_file_by_sections,
train_json_question_sql_pairs,
train_formatted_question_sql_pairs
)
from data_pipeline.trainer.vanna_trainer import (
train_ddl, train_documentation, train_question_sql_pair,
train_sql_example, flush_training
)
from common.result import success_response, bad_request_response, internal_error_response
import json
import tempfile
import os
import logging
logger = logging.getLogger(__name__)
@app.route('/api/v0/training_data/upload', methods=['POST'])
def upload_training_data():
try:
# 1. 参数验证
if 'file' not in request.files:
return jsonify(bad_request_response("未提供文件"))
file = request.files['file']
if file.filename == '':
return jsonify(bad_request_response("未选择文件"))
# 2. 文件大小验证 (500KB)
file.seek(0, 2)
file_size = file.tell()
file.seek(0)
if file_size > 500 * 1024: # 500KB
return jsonify(bad_request_response("文件大小不能超过500KB"))
# 3. 创建临时文件（复用现有函数需要文件路径）
filename = secure_filename(file.filename)
temp_file_path = None
try:
with tempfile.NamedTemporaryFile(mode='w+b', delete=False, suffix=os.path.splitext(filename)[1]) as tmp_file:
file.save(tmp_file.name)
temp_file_path = tmp_file.name
# 4. 根据文件类型调用现有的训练函数
filename_lower = filename.lower()
success_count = 0
if filename_lower.endswith('.ddl'):
success_count = train_ddl_file(temp_file_path)
elif filename_lower.endswith(('.md', '.markdown')):
success_count = train_documentation_file(temp_file_path)
elif filename_lower.endswith('_pair.json') or filename_lower.endswith('_pairs.json'):
success_count = train_json_pairs_file(temp_file_path)
elif filename_lower.endswith('_pair.sql') or filename_lower.endswith('_pairs.sql'):
success_count = train_formatted_pairs_file(temp_file_path)
elif filename_lower.endswith('.sql'):
success_count = train_sql_file(temp_file_path)
else:
return jsonify(bad_request_response("不支持的文件类型"))
# 5. 刷新批处理
flush_training()
return jsonify(success_response(
response_text=f"文件上传并训练成功：处理了 {success_count} 条记录",
data={
"filename": filename,
"file_size": file_size,
"records_processed": success_count,
"status": "completed"
}
))
except Exception as e:
logger.error(f"训练失败: {str(e)}")
return jsonify(internal_error_response(f"训练失败: {str(e)}"))
finally:
# 清理临时文件
if temp_file_path and os.path.exists(temp_file_path):
try:
os.unlink(temp_file_path)
except Exception as e:
logger.warning(f"清理临时文件失败: {str(e)}")
except Exception as e:
logger.error(f"文件上传失败: {str(e)}")
return jsonify(internal_error_response(f"文件上传失败: {str(e)}"))
# 复用现有训练函数的包装器
def train_ddl_file(file_path: str) -> int:
"""复用 run_training.py 中的 DDL 训练逻辑"""
success_count = 0
try:
ddl_statements = read_file_by_delimiter(file_path, ";")
for ddl in ddl_statements:
try:
train_ddl(ddl)
success_count += 1
except Exception as e:
logger.error(f"训练DDL失败: {ddl[:50]}..., 错误: {str(e)}")
except Exception as e:
logger.error(f"读取DDL文件失败: {str(e)}")
return success_count
def train_documentation_file(file_path: str) -> int:
"""复用 run_training.py 中的文档训练逻辑"""
success_count = 0
try:
# 检查是否为Markdown文件
if file_path.lower().endswith(('.md', '.markdown')):
sections = read_markdown_file_by_sections(file_path)
for section in sections:
try:
train_documentation(section)
success_count += 1
except Exception as e:
logger.error(f"训练文档失败: {section[:50]}..., 错误: {str(e)}")
else:
# 非Markdown文件使用传统的---分隔
doc_blocks = read_file_by_delimiter(file_path, "---")
for doc in doc_blocks:
try:
train_documentation(doc)
success_count += 1
except Exception as e:
logger.error(f"训练文档失败: {doc[:50]}..., 错误: {str(e)}")
except Exception as e:
logger.error(f"读取文档文件失败: {str(e)}")
return success_count
def train_json_pairs_file(file_path: str) -> int:
"""复用 run_training.py 中的 JSON 问答对训练逻辑"""
try:
# 直接调用现有函数
train_json_question_sql_pairs(file_path)
# 计算成功数量
with open(file_path, 'r', encoding='utf-8') as f:
pairs = json.load(f)
return len(pairs) if isinstance(pairs, list) else 1
except Exception as e:
logger.error(f"训练JSON问答对失败: {str(e)}")
return 0
def train_formatted_pairs_file(file_path: str) -> int:
"""复用 run_training.py 中的格式化问答对训练逻辑"""
try:
# 直接调用现有函数
train_formatted_question_sql_pairs(file_path)
# 计算成功数量
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
pairs = content.split('\n\n')
return len([p for p in pairs if p.strip()])
except Exception as e:
logger.error(f"训练格式化问答对失败: {str(e)}")
return 0
def train_sql_file(file_path: str) -> int:
"""复用 run_training.py 中的 SQL 训练逻辑"""
success_count = 0
try:
sql_statements = read_file_by_delimiter(file_path, ";")
for sql in sql_statements:
try:
train_sql_example(sql)
success_count += 1
except Exception as e:
logger.error(f"训练SQL失败: {sql[:50]}..., 错误: {str(e)}")
except Exception as e:
logger.error(f"读取SQL文件失败: {str(e)}")
return success_count
返回格式：
成功响应：
{
"code": 200,
"success": true,
"message": "操作成功",
"data": {
"response": "文件上传并训练成功：处理了 45 条记录",
"filename": "training_data.sql",
"file_size": 12345,
"records_processed": 45,
"status": "completed"
}
}
错误响应：
{
"code": 400,
"success": false,
"message": "请求参数错误",
"data": {
"response": "文件大小不能超过500KB",
"error_type": "INVALID_PARAMS",
"timestamp": "2025-01-15T10:30:00.000Z",
"can_retry": false
}
}
安全性和性能考虑：
文件处理优化：
- 使用临时文件处理，确保内存安全
- 文件大小限制500KB，避免内存压力
- 自动清理临时文件，防止磁盘空间泄露
- 完整的错误处理和日志记录
自动类型识别：
- 基于文件扩展名和命名规则
- 智能解析不同格式内容
- 大小写不敏感的格式处理
复用成熟功能：
- 直接使用
data_pipeline.trainer.run_training中的成熟函数 - 复用
read_file_by_delimiter和read_markdown_file_by_sections解析逻辑 - 调用
train_json_question_sql_pairs和train_formatted_question_sql_pairs处理问答对 - 继承批处理优化机制，保持训练逻辑的一致性
- 避免重复造轮子，提高代码维护性
- 直接使用
错误处理：
- 详细的错误日志记录
- 逐条处理失败时的容错机制
- 统计处理成功和失败的记录数
- 临时文件清理保证不留垃圾文件
性能优化：
- 复用现有的高效解析函数
- 批处理机制提高训练效率
- 自动刷新确保数据及时写入
- 避免重复开发，减少维护成本
使用建议：
文件准备：
- 确保文件使用UTF-8编码
- 控制文件大小在500KB以内
- 按照规定的格式和命名规则准备文件
文件类型选择：
- DDL文件：用于训练表结构信息，复用
train_ddl_statements逻辑 - Markdown文档：用于训练表和字段说明，复用
train_documentation_blocks逻辑 - JSON问答对：用于训练结构化问答数据，复用
train_json_question_sql_pairs逻辑 - 格式化问答对：用于训练文本格式问答数据，复用
train_formatted_question_sql_pairs逻辑 - SQL文件：用于训练SQL示例，复用
train_sql_examples逻辑
- DDL文件：用于训练表结构信息，复用
最佳实践：
- 建议分批上传，每次处理一种类型的文件
- 上传前检查文件格式的正确性
- 关注返回结果中的处理统计信息
- 复用现有训练逻辑，确保数据处理的一致性
3.3 /api/v0/training_data/combine (新增的API)
功能描述：
- 合并 langchain_pg_embedding 表中相同 document 字段的重复记录
- 支持按集合名称过滤合并范围
- 提供预览模式（dry_run）和实际执行模式
- 支持多种保留策略（first/last/by_metadata_time）
技术分析：
训练数据写入机制中，不同类型的数据在 document 字段中存储不同格式的内容：
- SQL类型：JSON格式
{"question": "...", "sql": "..."} - DDL类型：直接存储DDL语句原文
- Documentation类型：直接存储文档内容原文
- Error_SQL类型：JSON格式错误SQL示例
合并策略采用 document 字段精确匹配，因为：
- embedding 字段是向量，比较成本高且可能有微小差异
- document 字段存储确定性文本内容，便于精确比较
- 相同的 document 内容必然产生相同的训练效果
API规格：
请求参数：
{
"collection_names": ["sql", "ddl", "documentation", "error_sql"],
"dry_run": true,
"keep_strategy": "first"
}
参数说明：
collection_names: 要处理的集合名称数组，支持 ["sql", "ddl", "documentation", "error_sql"]dry_run: 是否为预览模式，默认 true（安全）keep_strategy: 保留策略"first": 保留第一条记录（按ID排序，推荐默认）"last": 保留最后一条记录（按ID排序）"by_metadata_time": 按 cmetadata.createdat 时间排序保留最新记录
实现代码框架：
@app.route('/api/v0/training_data/combine', methods=['POST'])
def combine_training_data():
try:
# 1. 参数验证
data = request.get_json()
if not data:
return jsonify(bad_request_response("请求体不能为空"))
collection_names = data.get('collection_names', [])
if not collection_names or not isinstance(collection_names, list):
return jsonify(bad_request_response("collection_names 参数必须是非空数组"))
# 验证集合名称
valid_collections = ['sql', 'ddl', 'documentation', 'error_sql']
invalid_collections = [name for name in collection_names if name not in valid_collections]
if invalid_collections:
return jsonify(bad_request_response(f"不支持的集合名称: {invalid_collections}"))
dry_run = data.get('dry_run', True)
keep_strategy = data.get('keep_strategy', 'first')
if keep_strategy not in ['first', 'last', 'by_metadata_time']:
return jsonify(bad_request_response("keep_strategy 必须是 'first', 'last' 或 'by_metadata_time'"))
# 2. 获取数据库连接
conn = get_db_connection()
cursor = conn.cursor()
# 3. 查找重复记录
duplicate_groups = []
total_before = 0
total_duplicates = 0
collections_stats = {}
for collection_name in collection_names:
# 获取集合ID
cursor.execute(
"SELECT uuid FROM langchain_pg_collection WHERE name = %s",
(collection_name,)
)
collection_result = cursor.fetchone()
if not collection_result:
continue
collection_id = collection_result[0]
# 统计该集合的记录数
cursor.execute(
"SELECT COUNT(*) FROM langchain_pg_embedding WHERE collection_id = %s",
(collection_id,)
)
collection_before = cursor.fetchone()[0]
total_before += collection_before
# 查找重复记录
cursor.execute("""
SELECT document, COUNT(*) as duplicate_count,
array_agg(id ORDER BY %s) as record_ids
FROM langchain_pg_embedding
WHERE collection_id = %s
GROUP BY document
HAVING COUNT(*) > 1
""" % (
"id" if keep_strategy in ['first', 'last'] else "COALESCE((cmetadata->>'createdat')::timestamp, '1970-01-01'::timestamp) DESC, id",
"%s"
), (collection_id,))
collection_duplicates = 0
for row in cursor.fetchall():
document_content, duplicate_count, record_ids = row
collection_duplicates += duplicate_count - 1 # 减去要保留的一条
# 根据保留策略选择要保留的记录
if keep_strategy == 'first':
keep_id = record_ids[0]
remove_ids = record_ids[1:]
elif keep_strategy == 'last':
keep_id = record_ids[-1]
remove_ids = record_ids[:-1]
else: # by_metadata_time
keep_id = record_ids[0] # 已经按时间排序
remove_ids = record_ids[1:]
duplicate_groups.append({
"collection_name": collection_name,
"document_content": document_content[:100] + "..." if len(document_content) > 100 else document_content,
"duplicate_count": duplicate_count,
"kept_record_id" if not dry_run else "records_to_keep": keep_id,
"removed_record_ids" if not dry_run else "records_to_remove": remove_ids
})
total_duplicates += collection_duplicates
collections_stats[collection_name] = {
"before": collection_before,
"after": collection_before - collection_duplicates,
"duplicates_removed" if not dry_run else "duplicates_to_remove": collection_duplicates
}
# 4. 执行合并操作（如果不是dry_run）
if not dry_run:
try:
conn.autocommit = False # 开始事务
for group in duplicate_groups:
remove_ids = group["removed_record_ids"]
if remove_ids:
cursor.execute(
"DELETE FROM langchain_pg_embedding WHERE id = ANY(%s)",
(remove_ids,)
)
conn.commit()
except Exception as e:
conn.rollback()
return jsonify(internal_error_response(f"合并操作失败: {str(e)}"))
finally:
conn.autocommit = True
# 5. 构建响应
total_after = total_before - total_duplicates
summary = {
"total_records_before": total_before,
"total_records_after": total_after,
"duplicates_removed" if not dry_run else "duplicates_to_remove": total_duplicates,
"collections_stats": collections_stats
}
if dry_run:
response_text = f"发现 {total_duplicates} 条重复记录，预计删除后将从 {total_before} 条减少到 {total_after} 条记录"
data_key = "duplicate_groups"
else:
response_text = f"成功合并重复记录，删除了 {total_duplicates} 条重复记录，从 {total_before} 条减少到 {total_after} 条记录"
data_key = "merged_groups"
return jsonify(success_response(
response_text=response_text,
data={
"dry_run": dry_run,
"collections_processed": collection_names,
"summary": summary,
data_key: duplicate_groups
}
))
except Exception as e:
return jsonify(internal_error_response(f"合并操作失败: {str(e)}"))
finally:
if 'cursor' in locals():
cursor.close()
if 'conn' in locals():
conn.close()
返回格式（dry_run: true）：
{
"code": 200,
"success": true,
"message": "操作成功",
"data": {
"response": "发现 50 条重复记录，预计删除后将从 1000 条减少到 950 条记录",
"dry_run": true,
"collections_processed": ["sql", "ddl"],
"summary": {
"total_records_before": 1000,
"total_records_after": 950,
"duplicates_to_remove": 50,
"collections_stats": {
"sql": {
"before": 600,
"after": 580,
"duplicates_to_remove": 20
},
"ddl": {
"before": 400,
"after": 370,
"duplicates_to_remove": 30
}
}
},
"duplicate_groups": [
{
"collection_name": "sql",
"document_content": "SELECT * FROM users WHERE id = ?",
"duplicate_count": 3,
"records_to_keep": "uuid1-sql",
"records_to_remove": ["uuid2-sql", "uuid3-sql"]
}
]
}
}
返回格式（dry_run: false）：
{
"code": 200,
"success": true,
"message": "操作成功",
"data": {
"response": "成功合并重复记录，删除了 50 条重复记录，从 1000 条减少到 950 条记录",
"dry_run": false,
"collections_processed": ["sql", "ddl"],
"summary": {
"total_records_before": 1000,
"total_records_after": 950,
"duplicates_removed": 50,
"collections_stats": {
"sql": {
"before": 600,
"after": 580,
"duplicates_removed": 20
},
"ddl": {
"before": 400,
"after": 370,
"duplicates_removed": 30
}
}
},
"merged_groups": [
{
"collection_name": "sql",
"document_content": "SELECT * FROM users WHERE id = ?",
"duplicate_count": 3,
"kept_record_id": "uuid1-sql",
"removed_record_ids": ["uuid2-sql", "uuid3-sql"]
}
]
}
}
错误响应示例：
{
"code": 400,
"success": false,
"message": "请求参数错误",
"data": {
"response": "collection_names 参数必须是非空数组",
"error_type": "INVALID_PARAMS",
"timestamp": "2025-01-15T10:30:00.000Z",
"can_retry": false
}
}
安全性和性能考虑：
事务安全：
- 整个合并过程在数据库事务中执行
- 发生错误时自动回滚
- 避免数据不一致状态
预览模式：
- 默认启用 dry_run 模式
- 用户可以先查看合并计划再决定是否执行
- 提供详细的影响分析
保留策略：
- 支持多种保留策略满足不同需求
- first 策略最简单可靠，推荐作为默认
- by_metadata_time 策略使用真实创建时间
性能优化：
- 使用数组聚合函数减少数据库查询次数
- 批量删除操作提高效率
- 分集合处理避免长时间锁表
错误处理：
- 详细的参数验证
- 完整的异常处理和错误信息
- 数据库连接的正确关闭
使用建议：
推荐流程： ```bash
1. 先预览合并计划
POST /api/v0/training_data/combine { "collection_names": ["sql", "ddl"], "dry_run": true, "keep_strategy": "first" }
# 2. 确认无误后执行实际合并 POST /api/v0/training_data/combine {
"collection_names": ["sql", "ddl"],
"dry_run": false,
"keep_strategy": "first"
}
2. **保留策略选择**：
- 一般情况：使用 `"first"` 策略
- 需要保留最新数据：使用 `"by_metadata_time"` 策略
- 测试环境：可以使用 `"last"` 策略
3. **批量处理**：
- 大量数据时建议分批处理不同集合
- 可以先处理较小的集合验证效果
- 重要数据建议先备份
---
## 4. 补充建议
### 4.1 错误处理增强
建议在所有API中添加更详细的错误处理：
```python
# 通用错误处理装饰器
def handle_api_errors(f):
@wraps(f)
def decorated_function(*args, **kwargs):
try:
return f(*args, **kwargs)
except ValueError as e:
return jsonify(bad_request_response(str(e)))
except Exception as e:
logger.error(f"API错误: {str(e)}", exc_info=True)
return jsonify(internal_error_response(f"系统错误: {str(e)}"))
return decorated_function
4.2 批量操作优化
对于大量数据的处理，建议引入任务队列：
# 异步任务支持
@app.route('/api/v0/training_data/upload_async', methods=['POST'])
def upload_training_data_async():
# 创建后台任务
task_id = str(uuid.uuid4())
# 将任务放入队列
# 返回任务ID供客户端查询进度
return jsonify(success_response(
response_text="文件上传任务已创建",
data={"task_id": task_id}
))
@app.route('/api/v0/training_data/task_status/<task_id>', methods=['GET'])
def get_task_status(task_id):
# 查询任务状态
pass
4.3 数据验证增强
建议添加更严格的数据验证：
# SQL语法验证
def validate_sql(sql):
try:
import sqlparse
parsed = sqlparse.parse(sql)
if not parsed:
raise ValueError("SQL语法错误")
return True
except Exception as e:
raise ValueError(f"SQL验证失败: {str(e)}")
# 数据类型验证
def validate_training_data(data_type, content):
if data_type == 'sql':
return validate_sql(content)
elif data_type == 'ddl':
return validate_ddl(content)
# 其他验证逻辑
5. 实施建议
5.1 开发顺序
- 第一阶段：实现增强的stats API（相对简单）
- 第二阶段：实现update API（中等复杂度）
- 第三阶段：实现upload API（最复杂，需要充分测试）
5.2 测试策略
- 单元测试：每个解析函数和训练函数
- 集成测试：完整的API调用流程
- 性能测试：大文件上传和处理
- 错误测试：各种异常情况的处理
5.3 部署注意事项
- 文件上传限制：配置Nginx/Apache的文件大小限制
- 临时目录：确保有足够的磁盘空间
- 数据库连接：优化数据库连接池设置
- 日志监控：添加详细的操作日志
6. 结论
本改造方案通过充分复用现有代码，提供了更加完整和实用的解决方案：
- update API：简化为单条更新，采用先删除后插入策略，返回新记录ID
- upload API：复用
data_pipeline/trainer模块的成熟函数，支持多种文件格式的自动解析和导入 - combine API：支持合并重复训练数据，提供预览和执行模式
主要优势：
避免重复造轮子：
- 直接复用
run_training.py中的成熟解析函数 - 保持与现有训练逻辑的一致性
- 减少代码维护成本
- 直接复用
提高代码质量：
- 复用经过验证的文件处理逻辑
- 继承现有的错误处理机制
- 保持日志格式的统一性
性能优化：
- 利用现有的批处理机制
- 复用高效的文件解析算法
- 减少开发和测试时间
特别是upload API的设计，通过复用现有的trainer模块，不仅避免了重复开发，还确保了与现有系统的完美兼容性，可以满足生产环境的需求。