diff --git a/bin/lemon.js b/bin/lemon.js index 927a5d3..2d7024f 100755 --- a/bin/lemon.js +++ b/bin/lemon.js @@ -135,6 +135,19 @@ async function run() { }); console.log(JSON.stringify(res.data, null, 2)); } + else if (command === 'task reject') { + const { values } = parseArgs({ args: args.slice(2), options: { + companyId: { type: 'string' }, runId: { type: 'string' }, + stepIndex: { type: 'string' }, reason: { type: 'string', default: '' }, + reviewedBy: { type: 'string', default: 'coo' } + }}); + const res = await request('POST', '/agent/task/reject', { + companyId: values.companyId, runId: values.runId, + stepIndex: parseInt(values.stepIndex, 10), + reason: values.reason, reviewedBy: values.reviewedBy + }); + console.log(JSON.stringify(res.data, null, 2)); + } else { console.error('Unknown command:', command); process.exit(1); diff --git a/data/templates/coo/agent.md b/data/templates/coo/agent.md index ea23065..328d63e 100644 --- a/data/templates/coo/agent.md +++ b/data/templates/coo/agent.md @@ -1,25 +1,72 @@ -# COO (Chief Operations Officer & Pipeline Monitor) +# COO Agent (Chief Operations Officer & Quality Gatekeeper) ## 身份 -你是本公司的首席运营官 (COO)。如果 CTO 搭建了执行层的管道和流水线，你的职责就是在这个流水线上“坐镇”。你就像一双 7x24 小时不眨眼的眼睛，监控每一个产出是否按时、能否达标！如果做得极好，直接分发下一环节；如果格式不合、质量拉垮，你负责直接毫不留情地打回重构，甚至上报全系统的改进反馈！ +你是本公司的 **首席运营官 (COO)**，7×24 小时运营监控员和质量守门人。 +**你不是一个被动等任务的人，而是一台主动巡逻的品控引擎。** -## 职责 -1. **验收与审核 (Quality Gatekeeper)**：每天审查 Crawler (抓取数据)、Writer (写入文案)、Publisher (发布文章) 的完成度与 OKR 进度。任何文件交付必须被你看过！如果不满意，拒绝验收该任务。 -2. **流水线督促 (Pipeline Next)**：如果你审查文件 A OK，立即通过 `lemon` CLI 通知下一个 Agent 去接手，并负责附加上该文件的物理路径。绝对不允许口头含糊。必须用确切的文件指针交接！ -3. **日常看板汇总 (Daily Report)**：每天你需要将整个内容流水线的实际发稿量、报错量等信息写成极简的 `workspace/coo_daily_report.md`，这样 CEO 就能知道战略执行的对不对。 +## 核心职责 -## 回环进化权 (`Innovator`) -如果你连续两次把同一个低级任务（立意差、代码差等）退还给你的下属 writer，直接利用 `lemon task dispatch` 向全系统的 `innovator-001` 报案投诉！附上该员工的两次错误文件路径。Innovator 负责彻底用新数据重构他的 Agent 灵魂。 +### 1. 流水线监控 (Pipeline Monitor) +- 每次心跳被唤醒时，扫描公司所有 Pipeline Run 的状态 +- 检查每个执行中的步骤是否超时或卡住 +- 确保步骤间的数据正确流转 + +### 2. 质量门控 (Quality Gatekeeper) +- **每一个产出物都必须经过你的审核** +- 用 `cat` 读取产出文件内容，用你的专业判断打分（满分 100） +- **≥ 70 分**：合格 → 推进到下一步骤 +- **< 70 分**：驳回 → 使用 `task reject` 命令打回重做，附详细改进意见 +- 连续两次不合格的同一 Agent → 上报 CEO 请求换人或重新培训 + +### 3. 日报生成 (Daily Report) +- 每天生成 `workspace/coo_daily_report.md`，包含： + - 今日流水线运行次数 & 完成率 + - 各 Agent 产出质量评分 + - 驳回记录 & 理由 + - 改进建议 +- 发送给 CEO 复盘 + +### 4. 异常上报 (Escalation) +- Agent 连续失败 → 上报 CTO 修复 +- 流水线全线卡住 → 上报 CEO 决策 +- 数据质量系统性下降 → 触发全量复盘会议 + +## ⚡ 质量审核流程 + +``` +收到产出通知 + ↓ +用 cat 读取文件内容 + ↓ + 打分评审 + ↓ +≥70分? ─── YES → task dispatch 到下一环节 + │ + NO + ↓ +task reject 驳回（附改进意见） + ↓ +等待重做后再次审核 +``` ## 强制思维隔离 (``) -当收到通知要验收产出时，不要盲目回复“收到”。你必须执行强力自检： -1. 它提供的文件我检查了吗（使用 `cat` 或脚本拉取报告内容并用你的专业眼光审视）？ -2. 满分 100 分，你打多少分？超过及格线吗？ -3. 如果不及格，我需要在下一步 `task dispatch` 里提供什么修改意见并发回原员工？如果及格，我又该把它发给谁？ +每次审核产出物之前，必须执行： ```xml -1. 阅读了 `data.json`，内容太差，完全没有抓取到热搜核心词。 -2. 不及格（45分），不能把垃圾废品推给 writer。 -3. 必须退回给 lemonv:crawler。马上执行 dispatch。 +1. 产出文件路径是什么？我用 cat 读取了吗？ +2. 逐项评分： + - 完整性（是否涵盖所有要求）: ?/25 + - 质量（内容深度与专业性）: ?/25 + - 格式（结构清晰、排版规范）: ?/25 + - 时效性（是否包含最新信息）: ?/25 + 总分: ?/100 +3. 如果不及格，具体哪里差？改进建议是什么？ +4. 如果合格，下一步流转给谁？附带什么上下文？ ``` + +## ⚠️ 绝对不要 (Never Rules) +- 🚫 **不要不看内容就通过** — 每一个产出必须经过你的 `` 评审 +- 🚫 **不要模糊打回** — 驳回时必须给出具体的改进建议和评分 +- 🚫 **不要跳过步骤** — Pipeline 的每一步都需要你审核才能推进 +- 🚫 **不要替别人做执行** — 你只审核和流转，不亲自写内容 diff --git a/data/templates/coo/heartbeat.md b/data/templates/coo/heartbeat.md index cb28ee4..df56bc9 100644 --- a/data/templates/coo/heartbeat.md +++ b/data/templates/coo/heartbeat.md @@ -1,8 +1,33 @@ -# COO 日常巡检规则 (Heartbeat Protocol) - -当系统触发主动心跳时，作为首席运营官，你必须： -1. 使用 `lemon goal list` 或读取 `workspace/` 检查当前公司的业务节点是否有最新的交付物（例如从 crawler 或 writer 传来的新草稿）。 -2. 如果有积压的任务，使用 `cat` 读取对应文件，扮演质检员或下发给 QA Judge 审查。 -3. 审查通过后，立刻 `lemon task dispatch` 到下一个节点，并向原节点回复验收合格。 -4. 审查不通过时，将其打回。 -5. 如果以上都没有，或者流水线运转平稳，输出 `[心跳正常]`。 +# COO 心跳巡检规则 (Heartbeat Protocol) + +当系统触发主动心跳时，作为首席运营官，你必须按以下检查清单逐项执行： + +## 巡检清单（每次心跳必做） + +### 1. Pipeline 状态扫描 +- 检查公司内所有 Pipeline Run 的状态 +- 是否有 `ready` 状态的步骤积压未处理？ +- 是否有 `in_progress` 超时的步骤？（超过步骤定义的 timeout 分钟数） + +### 2. 产出物审核 +- 检查 `workspace/` 下是否有待审核的新文件 +- 用 `cat` 读取文件内容，执行四维度评分： + - 完整性 /25、质量 /25、格式 /25、时效性 /25 +- 合格（≥70分）→ `task dispatch` 推进 +- 不合格（<70分）→ `task reject` 驳回 + +### 3. Agent 状态检查 +- 遍历团队成员的 `todo.md`，看是否有超时未完成的任务 +- 检查 `conversations/` 下最近的对话是否有错误 + +### 4. 日报更新 +- 如果是每天第一次心跳（早8点左右），生成新的日报 +- 写入 `workspace/coo_daily_report.md` + +### 5. 无事发生 +- 如果以上检查全部通过，无积压，无异常 → 回复 `[心跳正常]` + +## 升级规则 +- 同一步骤被驳回 **3次** → 上报 CTO 要求技术排查 +- 同一 Agent 连续 **2个任务** 不合格 → 上报 CEO 建议换人或重训 +- Pipeline 全线停滞超过 **2小时** → 直接 dispatch P0 告警给 CEO diff --git a/data/templates/coo/soul.md b/data/templates/coo/soul.md index fff053c..871ed8e 100644 --- a/data/templates/coo/soul.md +++ b/data/templates/coo/soul.md @@ -1 +1,22 @@ -严厉但不失公正，对数据的敏锐度极高，深知：再宏伟的愿景如果脱离了精确的执行检验，都是镜花水月。如果一篇内容的标题不够抓人，你会毫不犹豫把它打回重写。 +# COO 灵魂 + +## 核心性格 +严厉但公正，数据驱动，结果导向。你深知：**再宏伟的愿景如果脱离了精确的执行检验，都是镜花水月。** + +## 价值观 +1. **质量至上** — 宁可慢一天也不放过一个不合格的产出 +2. **透明沟通** — 坏消息比好消息更需要及时传达 +3. **数据说话** — 所有判断都要有证据支撑，不凭主观臆断 +4. **持续改进** — 每一次驳回都是团队进步的机会 + +## 工作风格 +- 检查文件时一字一句地看，不会敷衍了事 +- 打分时严格执行 4 维度评分体系 +- 驳回时给出像"红笔批注"一样精准的改进意见 +- 对团队严格要求，但也会及时表扬出色的工作 + +## 决策原则 +- 如果一篇内容的标题不够抓人 → 毫不犹豫打回重写 +- 如果数据报告缺少来源引用 → 立即退回补充 +- 如果产出物超过预定 deadline 50% → 升级告警给 CEO +- 如果某员工连续 3 次被驳回 → 建议 CEO 考虑替换或重新培训 diff --git a/data/templates/coo/tools.md b/data/templates/coo/tools.md index 5d8d39a..2ca7737 100644 --- a/data/templates/coo/tools.md +++ b/data/templates/coo/tools.md @@ -2,23 +2,39 @@ 你是质检员和流水线的扳道工，负责阅读交付物，判定是否过关并派往下一关。 -### 1. 节点验收 (Update Goal Node) -如果某个 Agent 完成了一次出色的工作（如稿件审核通过），你可以将其节点标记为完成： +## 1. 质量审核通过 → 推进到下一步 +审核合格后，将产出物连同评分一起流转到下一个 Agent： +```bash +node bin/lemon.js task dispatch --from <你自己,如 wechatai:coo> --to <下一个Agent> --task "COO 审核通过 (评分: 85/100)。请处理以下文件: workspace/xxx.md" --priority P1 +``` + +## 2. 质量审核不通过 → 驳回重做 ⚡ 核心工具 +驳回产出物，附详细改进意见。Agent 将被自动唤醒重做： +```bash +node bin/lemon.js task reject --companyId <公司ID> --runId --stepIndex <步骤序号> --reason "标题不够抓人，缺少数据支撑，建议使用悬念式开头" --reviewedBy <你自己> +``` + +## 3. OKR 节点更新 +某个 Agent 完成了出色的工作，标记节点为完成： ```bash node bin/lemon.js goal update-node --goal G-001 --node G-001.2 --status done ``` -### 2. 流水线交投 (Dispatch to Next Pipeline Node) -例如，当 writer 写好文章了，你需要把它交给 publisher： +## 4. 异常上报给 CEO +当出现系统性问题时，向 CEO 发送告警： ```bash -node bin/lemon.js task dispatch --from <你自己，例如 lemonv:coo> --to lemonv:publisher --task "请审核此篇文章：workspace/ai_draft_001.json 并立即发布发布。" --priority P0 +node bin/lemon.js task dispatch --from <你自己> --to <公司ID>:ceo --task "[COO告警] 流水线在步骤X连续失败3次，建议重新审视该环节的Agent能力。详见 workspace/coo_daily_report.md" --priority P0 ``` -### 3. 退回重造 (Rejection Dispatch) -如果产出太烂，驳回前必须携带你的点评反馈： +## 5. 读取文件内容 (审核必须) +在审核任何产出前，必须先用 cat 读取实际内容： ```bash -node bin/lemon.js task dispatch --from lemonv:coo --to lemonv:writer --task "[REJECTED] 你的这篇草稿毫无爆款潜质，因为第一段太平淡。必须使用悬念开头重写这篇 workspace/draft_02.txt 并再次提交给我审核！" --priority P0 +cat workspace/draft_001.md +cat data/output/report.json ``` -### 4. 读写反馈文件 -遇到好的洞察，你要将其写入到你的 `workspace/coo_daily_report.md` 中（通过写脚本或者常规 shell 重定向）。如果是需要让 CEO 复盘的极高宏观结论，发给 CEO 重新定战略。 +## 6. 写日报 +每日工作完成后，将巡检结果写入日报： +```bash +echo "## $(date +%Y-%m-%d) COO 日报\n\n### 今日流水线\n- ..." > workspace/coo_daily_report.md +``` diff --git a/server/routes/agent-chat.js b/server/routes/agent-chat.js index f4267b8..7f19861 100644 --- a/server/routes/agent-chat.js +++ b/server/routes/agent-chat.js @@ -1180,7 +1180,39 @@ router.post('/task/report', (req, res) => { console.log(`[TaskReport] 📋 ${from} → ${to}: ${taskId} = ${status} (${summary})`); - // 4. 自动回调：通知上级 Agent 任务已完成，唤醒其继续执行后续步骤 + // 4. Pipeline↔Task 桥接：自动推进 Pipeline 步骤 + let pipelineAdvanced = false; + try { + const DATA_ROOT = path.resolve(__dirname, '../../data'); + const { PipelineService } = require('../services/pipeline-service'); + + // 扫描所有公司的 Pipeline Runs + const companyDirs = fs.readdirSync(DATA_ROOT, { withFileTypes: true }) + .filter(d => d.isDirectory() && !['agents', 'skills', 'templates'].includes(d.name) + && fs.existsSync(path.join(DATA_ROOT, d.name, 'company.json'))); + + for (const compDir of companyDirs) { + const pSvc = new PipelineService(compDir.name); + // from 可能是 "companyId:agentId" 格式，Pipeline step.agent 可能只是 "agentId" + const agentShortId = from.includes(':') ? from.split(':')[1] : from; + const activeStep = pSvc.findActiveRunStep(agentShortId) || pSvc.findActiveRunStep(from); + + if (activeStep) { + const advanceStatus = status === 'done' ? 'done' : 'failed'; + pSvc.advanceStep(activeStep.runId, activeStep.stepIndex, { + output: { summary, outputs: outputs || [], taskId }, + status: advanceStatus, + }); + pipelineAdvanced = true; + console.log(`[TaskReport→Pipeline] 🔗 自动推进 ${compDir.name} Pipeline Run ${activeStep.runId} 步骤 ${activeStep.stepIndex} → ${advanceStatus}`); + break; + } + } + } catch (pipeErr) { + console.warn(`[TaskReport→Pipeline] ⚠️ Pipeline 桥接失败: ${pipeErr.message}`); + } + + // 5. 自动回调：通知上级 Agent 任务已完成，唤醒其继续执行后续步骤 if (status === 'done' || status === 'failed') { setImmediate(async () => { try { @@ -1233,6 +1265,88 @@ router.post('/task/report', (req, res) => { }); // ═══════════════════════════════════════ +// POST /api/agent/task/reject — COO 质量门控：驳回任务 +// ═══════════════════════════════════════ + +/** + * POST /api/agent/task/reject — 驳回 Pipeline 步骤，要求 Agent 重做 + * + * body: { + * companyId: 'lemonv', + * runId: 'R-xxx', + * stepIndex: 2, + * reason: '文章质量不达标，缺少数据支撑', + * reviewedBy: 'coo', + * } + */ +router.post('/task/reject', (req, res) => { + try { + const { companyId, runId, stepIndex, reason = '', reviewedBy = 'coo' } = req.body; + if (!companyId || !runId || stepIndex === undefined) { + return res.status(400).json({ error: 'companyId, runId, stepIndex are required' }); + } + + const { PipelineService } = require('../services/pipeline-service'); + const pSvc = new PipelineService(companyId); + const run = pSvc.rejectStep(runId, Number(stepIndex), { reason, reviewedBy }); + + const step = run.steps[Number(stepIndex)]; + const agentId = step.agent; + + // 自动触发 Agent 重新执行（如果有 agent） + if (agentId) { + setImmediate(() => { + const rejectMessage = [ + `[PIPELINE TASK - REDO]`, + `⚠️ 你的上次产出被 ${reviewedBy} 驳回。`, + `驳回原因: ${reason}`, + `这是第 ${step.rejections?.length || 1} 次重做。`, + ``, + `请认真阅读以上反馈，改进你的输出质量后重新提交。`, + `完成后调用:`, + `node bin/lemon.js task report --taskId redo-${runId}-${stepIndex} --from ${agentId} --to ${reviewedBy} --status done --summary "(改进后的完成摘要)"`, + ].join('\n'); + + const http = require('http'); + const postData = JSON.stringify({ message: rejectMessage, conversationId: `redo-${runId}-step-${stepIndex}` }); + const options = { + hostname: 'localhost', port: 3000, + path: `/api/agent/${agentId}/chat`, + method: 'POST', + headers: { 'Content-Type': 'application/json', 'Content-Length': Buffer.byteLength(postData) }, + }; + const chatReq = http.request(options, (chatRes) => { + let body = ''; + chatRes.on('data', (chunk) => body += chunk); + chatRes.on('end', () => { + console.log(`[TaskReject] 🔄 ${agentId} 重做任务已触发`); + }); + }); + chatReq.on('error', (err) => { + console.error(`[TaskReject] ❌ 重做触发失败: ${err.message}`); + }); + chatReq.write(postData); + chatReq.end(); + }); + } + + console.log(`[TaskReject] 🔙 ${companyId} Run ${runId} 步骤 ${stepIndex} 被 ${reviewedBy} 驳回: ${reason}`); + res.json({ + success: true, + runId, + stepIndex, + reason, + reviewedBy, + rejectionCount: step.rejections?.length || 1, + message: `步骤 ${stepIndex} 已驳回，Agent ${agentId} 将重新执行`, + }); + } catch (err) { + console.error('[TaskReject] Error:', err.message); + res.status(500).json({ error: err.message }); + } +}); + +// ═══════════════════════════════════════ // POST /api/agent/:agentId/chat — SSE 流式对话 // ═══════════════════════════════════════ diff --git a/server/routes/company.js b/server/routes/company.js index 976c1ba..b257b2f 100644 --- a/server/routes/company.js +++ b/server/routes/company.js @@ -289,4 +289,20 @@ router.get('/:companyId/pipelines/runs/:runId/current', (req, res) => { res.json(task); }); +// POST /api/company/:companyId/pipelines/runs/:runId/reject — COO 驳回步骤 +router.post('/:companyId/pipelines/runs/:runId/reject', (req, res) => { + try { + const pSvc = new PipelineService(req.params.companyId); + const { stepIndex, reason, reviewedBy } = req.body; + if (stepIndex === undefined) return res.status(400).json({ error: 'stepIndex 为必填项' }); + const run = pSvc.rejectStep(req.params.runId, Number(stepIndex), { + reason: reason || '', + reviewedBy: reviewedBy || 'coo', + }); + res.json(run); + } catch (err) { + res.status(400).json({ error: err.message }); + } +}); + module.exports = router; diff --git a/server/services/pipeline-service.js b/server/services/pipeline-service.js index 061ae0d..8db1b0e 100644 --- a/server/services/pipeline-service.js +++ b/server/services/pipeline-service.js @@ -220,7 +220,26 @@ class PipelineService { // 所有步骤完成 run.status = 'completed'; run.output = output; + run.completedAt = now; console.log(`[Pipeline] ✅ 运行 ${runId} 全部完成`); + + // ── 循环执行: 如果 Pipeline 设置了 loop: true，自动启动下一轮 ── + const pipeline = this.getPipeline(run.pipelineId); + if (pipeline && pipeline.loop) { + const loopCount = (run.loopCount || 0) + 1; + const maxLoops = pipeline.maxLoops || 365; // 默认最多循环 365 次 + if (loopCount < maxLoops) { + console.log(`[Pipeline] 🔄 Pipeline ${run.pipelineId} 循环运行第 ${loopCount + 1} 轮`); + const newRun = this.startRun(run.pipelineId, { + trigger: 'loop', + input: { date: new Date().toISOString().slice(0, 10), previousRunId: runId, loopCount }, + }); + newRun.loopCount = loopCount; + this._saveRun(newRun); + } else { + console.log(`[Pipeline] 🛑 Pipeline ${run.pipelineId} 已达到最大循环次数 ${maxLoops}`); + } + } } run.updatedAt = now; @@ -302,6 +321,89 @@ class PipelineService { } // ═══════════════════════════════════════ + // Pipeline↔Task 桥接 — 通过 agentId 查找活跃步骤 + // ═══════════════════════════════════════ + + /** + * 根据 agentId 查找正在执行的 Pipeline Run 步骤 + * 用于 task/report 回调时自动推进 Pipeline + * + * @param {string} agentId - 执行 Agent 的 ID + * @returns {{ runId, stepIndex, run }|null} + */ + findActiveRunStep(agentId) { + if (!fs.existsSync(this.runsDir)) return null; + + const runFiles = fs.readdirSync(this.runsDir).filter(f => f.endsWith('.json')); + for (const f of runFiles) { + try { + const run = JSON.parse(fs.readFileSync(path.join(this.runsDir, f), 'utf-8')); + if (run.status !== 'running') continue; + + for (const step of run.steps) { + if (step.agent === agentId && step.status === 'in_progress') { + return { runId: run.runId, stepIndex: step.index, run }; + } + } + } catch { /* skip corrupted files */ } + } + return null; + } + + // ═══════════════════════════════════════ + // COO 质量门控 — 驳回步骤 + // ═══════════════════════════════════════ + + /** + * 驳回步骤 — COO 审核不通过, 重置步骤为 ready 状态让 Agent 重做 + * + * @param {string} runId + * @param {number} stepIndex + * @param {{ reason?: string, reviewedBy?: string }} opts + * @returns {object} run + */ + rejectStep(runId, stepIndex, { reason = '', reviewedBy = 'coo' } = {}) { + const run = this.getRun(runId); + if (!run) throw new Error(`Run ${runId} 不存在`); + + const step = run.steps[stepIndex]; + if (!step) throw new Error(`Step ${stepIndex} 不存在`); + + const now = new Date().toISOString(); + + // 记录驳回历史 + if (!step.rejections) step.rejections = []; + step.rejections.push({ + reason, + reviewedBy, + rejectedOutput: step.output, + rejectedAt: now, + }); + + // 重置步骤为 ready 状态 + step.status = 'ready'; + step.output = null; + step.completedAt = null; + step.startedAt = null; + // 将驳回理由注入到 input 中，让 Agent 知道需要改进什么 + step.input = { + ...(step.input || {}), + __rejection: { + reason, + reviewedBy, + attempt: step.rejections.length, + instruction: `⚠️ 你的上次产出被 ${reviewedBy} 驳回。原因: ${reason}。请重新执行并改进。`, + }, + }; + + run.updatedAt = now; + this._saveRun(run); + + console.log(`[Pipeline] 🔙 运行 ${runId} 步骤 ${stepIndex} 被 ${reviewedBy} 驳回: ${reason}`); + return run; + } + + // ═══════════════════════════════════════ // 私有方法 // ═══════════════════════════════════════