生产环境问题捕获和同步方案
📊 方案概述
从生产环境自动收集真实问题，生成报告和任务，同步到开发环境进行修复。
🏗️ 系统架构
┌─────────────────────────────────────────────────────────┐
│ 生产环境 (mbe.hi-maker.com) │
├─────────────────────────────────────────────────────────┤
│ 1. 真实用户交互 │
│ - 对话数据 │
│ - 响应时间 │
│ - 错误日志 │
│ - 专家切换 │
│ │
│ 2. 数据收集器 │
│ ✅ TITANS 记忆系统（已有） │
│ ✅ 性能监控（已有） │
│ ✅ 错误追踪（已有） │
│ 🆕 生产问题收集器（新增） │
└─────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ 问题分析和报告生成 │
├─────────────────────────────────────────────────────────┤
│ 3. 自动分析引擎 │
│ - 定时任务（每小时/每天） │
│ - 分析最近N小时的生产数据 │
│ - 生成测试报告 │
│ - 提取修改任务 │
└─────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ 任务审批和执行 │
├─────────────────────────────────────────────────────────┤
│ 4. 人工审批 │
│ - Web UI 查看任务 │
│ - 批准/拒绝任务 │
│ │
│ 5. 开发环境测试 │
│ - 同步任务到开发版 │
│ - 在开发环境复现和修复 │
│ - 测试修复方案 │
│ │
│ 6. 代码修改 │
│ - OpenRouter AI 生成修改 │
│ - 代码审查 │
│ - 提交和部署 │
└─────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ 部署到生产 │
└─────────────────────────────────────────────────────────┘
🔧 实现细节
1. 生产环境数据收集
已有的数据源
A. TITANS 记忆系统
- ✅ 记录所有对话
- ✅ 用户ID、设备ID
- ✅ 时间戳
- ✅ 专家选择
B. 性能监控
- ✅ 响应时间
- ✅ 知识库查询时间
- ✅ LLM 调用时间
C. 错误日志
- ✅ 异常记录
- ✅ 错误堆栈
- ✅ 失败请求
需要增强的部分
D. 生产问题标记 🆕
# 自动标记问题请求
class ProductionIssueCollector:
def collect_issues(self, timeframe_hours=24):
"""收集生产环境问题"""
issues = []
# 1. 慢响应
slow_requests = self.find_slow_requests(timeframe_hours, threshold=5.0)
# 2. 错误请求
error_requests = self.find_error_requests(timeframe_hours)
# 3. 专家切换
expert_switches = self.find_expert_switches(timeframe_hours)
# 4. 用户重试
user_retries = self.find_user_retries(timeframe_hours)
return issues
2. 定时自动分析
方案A: Cron 定时任务（推荐）
# docker-compose.prod.yml
services:
mbe-scheduler:
image: mises-behavior-engine-mbe-api
command: python scripts/production_monitor.py
environment:
- MONITOR_INTERVAL=3600 # 每小时
- MONITOR_ENVIRONMENT=production
depends_on:
- mbe-api
- postgres
方案B: Celery 定时任务
# src/tasks/production_monitor.py
from celery import Celery
from celery.schedules import crontab
app = Celery('mbe-tasks')
@app.task
@app.on_after_finalize.connect
def setup_production_monitor(sender, **kwargs):
# 每小时运行
sender.add_periodic_task(
crontab(minute=0), # 每小时的第0分钟
monitor_production.s(),
)
@app.task
def monitor_production():
"""监控生产环境"""
from src.feedback.auto_test_report import AutoTestReportGenerator
from src.feedback.task_manager import get_task_manager
# 1. 生成报告
generator = AutoTestReportGenerator()
report = await generator.generate_report(
start_time=datetime.now() - timedelta(hours=1),
end_time=datetime.now(),
environment="production"
)
# 2. 提取任务
manager = get_task_manager()
tasks = manager.create_tasks_from_report(report.to_dict())
# 3. 发送通知（可选）
if tasks:
send_notification(f"发现 {len(tasks)} 个生产环境问题")
3. 环境标识
区分生产和开发环境的数据:
# 在报告中添加环境标签
@dataclass
class AutoTestReport:
environment: str = "production" # or "development", "staging"
# 在任务中记录来源环境
@dataclass
class ModificationTask:
source_environment: str = "production"
affected_users: int = 0 # 生产环境受影响用户数
severity_score: float = 0.0 # 基于真实影响计算的严重性
4. 生产到开发的同步
方式1: 数据库同步
# 同步生产问题到开发数据库
class ProductionDevSync:
def sync_tasks(self):
"""同步生产任务到开发环境"""
# 从生产数据库读取
prod_tasks = self.get_production_tasks()
# 写入开发数据库
for task in prod_tasks:
dev_task = task.copy()
dev_task.environment = "development"
dev_task.is_from_production = True
self.save_to_dev(dev_task)
方式2: API 同步
# 生产环境 API
@router.post("/tasks/export")
async def export_production_tasks():
"""导出生产任务"""
tasks = get_task_manager().get_pending_tasks()
return {
"environment": "production",
"tasks": [t.to_dict() for t in tasks]
}
# 开发环境 API
@router.post("/tasks/import")
async def import_production_tasks(data: Dict):
"""导入生产任务"""
manager = get_task_manager()
for task_data in data["tasks"]:
task_data["source_environment"] = "production"
manager.import_task(task_data)
5. 开发环境复现
# 在开发环境复现生产问题
class ProductionIssueReproducer:
def reproduce_in_dev(self, production_task):
"""在开发环境复现生产问题"""
# 1. 提取生产环境的上下文
context = production_task.source_issue.get("context", {})
# 2. 在开发环境重放
test_cases = self.extract_test_cases(context)
# 3. 运行测试
results = self.run_tests(test_cases)
# 4. 验证问题复现
if self.verify_reproduction(results):
return "问题已复现"
else:
return "问题未复现，可能是环境差异"
📋 实施步骤
阶段1: 增强生产监控（1天）
添加生产问题收集器
src/feedback/production_collector.py配置环境标识
# .env ENVIRONMENT=production # or development测试数据收集
python scripts/test_production_collector.py
阶段2: 定时分析（1天）
创建定时监控脚本
scripts/production_monitor.py配置定时任务
- 选择 Cron 或 Celery
- 设置运行频率
测试自动分析
python scripts/production_monitor.py --test
阶段3: 环境同步（1天）
实现同步机制
scripts/sync_prod_to_dev.py配置同步策略
- 自动同步 vs 手动同步
- 同步频率
- 数据过滤
测试同步流程
python scripts/sync_prod_to_dev.py --dry-run
🎯 配置示例
生产环境配置
# docker-compose.prod.yml
services:
mbe-api:
environment:
# 环境标识
- ENVIRONMENT=production
# 监控配置
- ENABLE_PRODUCTION_MONITOR=true
- MONITOR_INTERVAL=3600 # 每小时
- MONITOR_THRESHOLD_SLOW_RESPONSE=5.0 # 5秒
- MONITOR_THRESHOLD_ERROR_RATE=0.05 # 5%
# 同步配置
- SYNC_TO_DEV=true
- DEV_API_URL=http://localhost:8001
开发环境配置
# docker-compose.dev.yml
services:
mbe-api-dev:
environment:
# 环境标识
- ENVIRONMENT=development
# 接收生产数据
- ACCEPT_PRODUCTION_TASKS=true
- PRODUCTION_API_URL=http://localhost:8000
📊 监控Dashboard（可选）
创建一个监控面板显示生产健康状况：
┌─────────────────────────────────────────┐
│ 生产环境健康监控 │
├─────────────────────────────────────────┤
│ 平均响应时间: 2.3s ✅ │
│ 错误率: 0.5% ✅ │
│ 专家切换率: 12% ⚠️ │
│ 待修复问题: 3 🔴 │
├─────────────────────────────────────────┤
│ 最近24小时趋势 │
│ [响应时间图表] │
│ [错误率图表] │
└─────────────────────────────────────────┘
✅ 使用流程
日常运行
自动监控
- 生产环境持续运行
- 每小时自动分析数据
- 发现问题自动生成报告和任务
人工审查
- 登录 http://localhost:8000/admin/tasks
- 查看生产环境问题
- 批准需要修复的任务
开发环境修复
- 任务自动同步到开发环境
- 在开发环境复现和修复
- 测试修复方案
部署到生产
- 修复验证通过后
- 部署到生产环境
- 持续监控效果
🔒 安全考虑
数据脱敏
# 同步时移除敏感信息 def anonymize_production_data(task): task.source_issue["user_id"] = hash_user_id(task.source_issue["user_id"]) task.source_issue["user_query"] = anonymize_text(task.source_issue["user_query"]) return task访问控制
- 开发环境不能直接访问生产数据库
- 通过 API 同步
- 需要认证和授权
📝 总结
可以捕获生产版问题吗？
✅ 可以！ 通过：
- TITANS 记忆系统（已有）
- 性能监控（已有）
- 生产问题收集器（新增）
- 定时自动分析（新增）
可以统一到开发版吗？
✅ 可以！ 通过：
- 数据库同步
- API 同步
- 任务导入导出
- 环境标识和隔离
🚀 下一步
我可以立即为您实现以下功能：
- ✅ 生产问题收集器
- ✅ 定时监控脚本
- ✅ 环境同步机制
- ✅ 开发环境复现工具
请告诉我是否开始实现？ 🎯