1. 项目概述与工程化改造路径
Edict系统经过前期的部署启动和使用实战阶段后，正式进入二次开发与扩展阶段。这个阶段的核心目标是将系统从"可用"推进到"可迭代"的平台级别，建立完整的工程化改造路径。不同于简单的功能堆砌，真正的工程化改造需要关注以下几个关键维度：
- 代码结构清晰度：建立明确的代码地图和职责边界
- 扩展性设计：前后端分离的契约式开发模式
- 稳定性保障：状态机设计、异常处理和回滚机制
- 交付标准化：构建、测试、发布的完整流水线
- 运维可视化：日志、监控、配置的集中管理
当前系统已经形成两条明确的开发主线：
- 根目录看板主线：位于项目根目录下，包含dashboard、scripts等目录，特点是改动直观、验证成本低，适合快速迭代面板扩展和轻量级功能增强。
- edict/全栈主线：位于edict目录下，包含完整的后端架构和前端实现，适合长期架构演进和服务层抽象。
2. 代码结构与职责边界
2.1 核心目录结构解析
进入二次开发阶段，首先需要建立完整的代码地图。以下是关键目录的职责划分：
agents/ # Agent人格定义、职责边界、输入输出规范 scripts/ # 数据同步脚本、定时任务、运维脚本 dashboard/ # 看板服务、轻量级接口聚合 edict/backend/ # FastAPI核心后端、服务层抽象 edict/frontend/ # 前端页面与组件 data/ # 看板依赖的JSON数据源 tests/ # 单元测试与端到端测试2.2 修改落点判断逻辑
当需要进行功能修改或扩展时，可按以下决策树确定修改位置：
- 改页面展示：优先查看edict/frontend
- 改接口返回：
- 简单接口：dashboard/server.py
- 核心接口：edict/backend
- 改任务状态：
- 检查dashboard/server.py
- 同步修改store.ts
- 更新后端状态模型
- 改角色行为：
- 修改agents/*/SOUL.md
- 更新权限矩阵
- 改数据同步：
- 调整scripts/下的同步脚本
- 更新run_loop.py中的任务列表
3. 后端扩展实践
3.1 契约优先的开发模式
后端扩展必须遵循"契约优先"原则，在编码前明确三类核心契约：
- 状态契约：
class TaskState(str, Enum): TAIZI = "Taizi" # 待办状态 ZHONGSHU = "Zhongshu" # 审议中 MENXIA = "Menxia" # 复核中 DOING = "Doing" # 执行中 QA = "QA" # 质量检查 DONE = "Done" # 已完成- 数据契约：
class TaskBase(BaseModel): id: str = Field(..., min_length=3) title: str = Field(..., max_length=100) state: TaskState assignee: Optional[str] created_at: datetime = Field(default_factory=datetime.now)- 接口契约：
@app.get("/tasks/{task_id}", response_model=TaskOut) async def get_task(task_id: str = Path(..., min_length=3)): """获取任务详情""" task = await TaskService.get(task_id) if not task: raise HTTPException(404, "Task not found") return task3.2 状态机设计与实现
状态流转是后端核心逻辑，需要严格实现：
STATE_TRANSITIONS = { "Taizi": ["Zhongshu"], "Zhongshu": ["Menxia", "Taizi"], "Menxia": ["Assigned", "Zhongshu"], "Assigned": ["Doing"], "Doing": ["QA"], "QA": ["Review", "Doing"], "Review": ["Done", "Doing"] } def validate_transition(current: str, next_state: str) -> bool: """验证状态流转是否合法""" allowed = STATE_TRANSITIONS.get(current, []) return next_state in allowed注意事项：状态变更必须保证原子性，通常需要结合数据库事务实现。对于关键业务状态，建议记录完整的状态变更历史。
4. 前端扩展策略
4.1 组件化开发模式
前端扩展采用标准的组件化开发流程：
- 数据准备层：在scripts/下添加数据生成脚本
- 接口层：在后端添加API路由
- 前端API层：封装接口调用方法
- 组件层：实现展示组件
- 状态管理：集成到全局store
- 页面集成：注册到导航路由
4.2 质量面板实现示例
以新增质量雷达面板为例，完整实现链路如下：
- 数据脚本(scripts/sync_quality_metrics.py):
# 生成模拟质量数据 metrics = { "completion_rate": random.randint(85, 98), "rejection_rate": random.randint(2, 15), "avg_duration": random.randint(30, 120) } Path("data/quality.json").write_text(json.dumps(metrics))- 后端接口(dashboard/server.py):
@app.get("/api/quality") def get_quality(): file = Path("data/quality.json") if not file.exists(): return {"error": "Data not ready"} return JSONResponse(json.loads(file.read_text()))- 前端组件(edict/frontend/src/components/QualityRadar.tsx):
const QualityRadar = () => { const [metrics, setMetrics] = useState<QualityMetrics>(); useEffect(() => { fetchQualityMetrics().then(setMetrics); }, []); return ( <div className="quality-panel"> <RadarChart data={transformForChart(metrics)} /> <MetricTable data={metrics} /> </div> ); };5. Agent扩展体系
5.1 Agent定义规范
每个Agent必须包含完整的SOUL.md定义：
# 审计司 (Audit Office) ## 职责 - 结果合规性检查 - 风险识别与评估 - 质量审计报告生成 ## 输入规范 - 任务执行结果JSON - 相关附件清单 - 执行上下文信息 ## 输出规范 - 审计报告MD格式 - 问题清单 - 整改建议 ## 权限范围 - 可读取所有任务数据 - 可生成审计标记 - 不可修改原始结果5.2 权限矩阵配置
在config/agents.py中维护权限矩阵：
AGENT_PERM_MATRIX = { "audit_office": { "read": ["*"], "write": ["audit_mark"], "deny": ["result_edit"] } }6. 调度与Worker扩展
6.1 异常处理策略
实现三级故障处理机制：
- 重试策略：
def retry_policy(failure_count: int) -> bool: return failure_count < 3 # 最多重试3次- 升级策略：
def escalate_policy(wait_seconds: int) -> bool: return wait_seconds > 300 # 5分钟未处理则升级- 回滚策略：
def rollback(task: Task) -> bool: return ( task.state == "QA" and task.duration > timedelta(hours=2) ) # QA阶段超过2小时则回滚6.2 调度器实现
基于Redis Streams的分布式调度器：
class TaskScheduler: def __init__(self, redis: Redis): self.redis = redis self.stream = "edict:tasks" async def schedule(self, task: Task): await self.redis.xadd(self.stream, task.dict()) async def consume(self): while True: messages = await self.redis.xreadgroup( "edict_workers", "worker1", {self.stream: ">"}, count=1 ) if messages: await process_message(messages[0])7. PyQt启动器实现
7.1 架构设计
启动器采用经典MVC模式：
edict_launcher.py # 入口文件 edict_pyqt_ui.py # 主界面(View) service_controller.py # 服务控制(Controller) config_manager.py # 配置管理(Model)7.2 核心功能实现
- ��务控制：
class ServiceController: def start(self): if self._check_port(): raise RuntimeError("Port already in use") self._process = subprocess.Popen( ["python", "dashboard/server.py"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT ) self._start_output_thread() def _start_output_thread(self): def read_output(): while self._process.poll() is None: line = self._process.stdout.readline() self.sig_output.emit(line.decode()) Thread(target=read_output, daemon=True).start()- 日志展示：
class LogViewer(QTextEdit): def __init__(self): super().__init__() self.setReadOnly(True) self.setFont(QFont("Consolas", 10)) def append_log(self, text: str): self.moveCursor(QTextCursor.End) self.insertPlainText(text) self.ensureCursorVisible()8. 数据兼容与迁移
8.1 兼容层设计
对于历史数据，实现智能兼容层：
def normalize_task(raw: dict) -> Task: """将不同版本的任务数据标准化""" return Task( id=raw.get("id") or raw["task_id"], title=raw["title"], state=normalize_state(raw["state"]), created_at=parse_date(raw) ) def parse_date(raw: dict) -> datetime: """兼容多种日期格式""" if isinstance(raw["created_at"], datetime): return raw["created_at"] return datetime.strptime(raw["created_at"], "%Y-%m-%dT%H:%M:%S")8.2 迁移脚本示例
数据库变更迁移脚本：
def migrate_qa_state(): """将历史任务中的Review状态拆分为QA和Review""" with Session() as session: tasks = session.query(Task).filter_by(state="Review").all() for task in tasks: if needs_qa(task): # 根据业务规则判断 task.state = "QA" session.commit()9. 测试与发布流程
9.1 测试金字塔实现
- 单元测试：
def test_state_transition(): assert validate_transition("Doing", "QA") assert not validate_transition("Done", "Doing")- 集成测试：
def test_task_flow(): task = create_task() assert task.state == "Taizi" move_to_state(task, "Zhongshu") assert task.state == "Zhongshu"- E2E测试：
def test_create_task_ui(): browser = Chrome() browser.visit("/") browser.click("New Task") browser.fill("title", "Test Task") browser.click("Submit") assert browser.has_text("Test Task")9.2 发布检查清单
- [ ] 所有测试通过
- [ ] 数据库迁移脚本验证
- [ ] 版本号更新
- [ ] 文档更新
- [ ] 回滚方案准备
- [ ] 监控指标配置
10. 运维与安全实践
10.1 配置管理
采用分层配置策略：
# config.py class Config: def __init__(self): self._config = { **self._load_defaults(), **self._load_env(), **self._load_secrets() } def _load_secrets(self): return dotenv_values(".env.secret")10.2 安全实践
密钥管理：
- 禁止硬编码密钥
- 使用环境变量或密钥管理服务
- 桌面启动器中的密钥输入做内存级处理
日志规范：
- 敏感信息脱敏
- 统一日志格式
- 按服务拆分日志文件
监控指标：
- 服务存活状态
- 任务积压数量
- 异常发生频率
- 资源使用情况
11. 工程化改造路线图
建议按照以下阶段逐步推进：
基础设施阶段(1-2周)
- 统一配置管理
- 建立CI流水线
- 实现基础监控
核心能力阶段(2-3周)
- 完善状态机实现
- 构建调度系统
- 实现基础测试套件
扩展能力阶段(3-4周)
- 新增业务面板
- 扩展Agent类型
- 增强桌面启动器
治理能力阶段(持续迭代)
- 完善文档体系
- 建立发布规范
- 优化运维流程
在实际操作中，我发现最有效的改造方式是"小步快跑"——每个迭代周期控制在1-2周，每次聚焦一个明确的改进目标。例如先集中精力完善状态机和工作流，再转向UI增强和工具链优化。这种渐进式改造既保证了系统持续可用，又能稳步提升工程化水平。
