/** * Agent Chat Route — 通用 Agent 对话 SSE 端点 * * 存储架构:conversations/agents/{agentId}/{conversationId}/ * ├── meta.json ← 对话元数据 * ├── {messageId}/messages.log ← 每轮对话日志 * └── history.json ← 有序的消息列表(供前端展示 + 上下文恢复) * * 多轮对话: * - ClaudeCliProtocol 持久进程 → 天然多轮 * - 服务器重启后自动恢复:从 history.json 加载历史,注入 system prompt + 历史摘要 * * 使用 @mas/acp-core 的 ClaudeCliProtocol。 */ const express = require('express'); const path = require('path'); const fs = require('fs'); const { randomUUID } = require('crypto'); const { ClaudeCliProtocol } = require('@mas/acp-core'); const { Parser } = require('../src/core/parser'); const AgentManager = require('../src/core/agent-manager'); const systemConfig = require('../config'); const router = express.Router(); const DATA_DIR = path.resolve(__dirname, '../../data/agents'); // CONV_ROOT is now dynamically evaluated per-agent based on DATA_DIR. function loadSystemConfig() { const configDir = path.resolve(__dirname, '../config'); const backends = JSON.parse(fs.readFileSync(path.join(configDir, 'backends.json'), 'utf-8')); const { loadConfigFromSkills } = require('../src/core/skill-config-loader'); const skillsDir = systemConfig.skillsCodeDir; return loadConfigFromSkills(skillsDir, backends); } const fullSystemConfig = loadSystemConfig(); // 初始化 AgentManager 用于 Sub-Agent 调用 const agentManager = new AgentManager(fullSystemConfig, path.resolve(__dirname, '../../')); // ═══════════════════════════════════════ // Bootstrap 工具 // ═══════════════════════════════════════ function loadFile(agentDir, filename) { try { return fs.readFileSync(path.join(agentDir, filename), 'utf-8'); } catch { return ''; } } function buildSystemPrompt(agentDir) { const files = ['agent.md', 'soul.md', 'identity.md', 'user.md', 'tools.md']; const parts = []; const agentListText = Object.entries(fullSystemConfig.agents || {}) .filter(([name]) => name !== 'master' && name !== 'innovator') .map(([name, cfg]) => { const typeLabel = cfg.type === 'orchestrator' ? '子编排' : cfg.type; return `- ${name} (${typeLabel}): ${cfg.description || ''}`; }) .join('\n'); for (const f of files) { let c = loadFile(agentDir, f); if (c.trim()) { // 替换代理占位符模板 if (c.includes('{{AVAILABLE_AGENTS}}')) { c = c.replace('{{AVAILABLE_AGENTS}}', agentListText); } parts.push(c.trim()); } } const memory = loadFile(agentDir, 'memory/memory.md'); if (memory.trim()) parts.push('---\n\n# 我的记忆\n\n' + memory.trim()); let mainSkill = loadFile(agentDir, '.claude/skills/main_skill/SKILL.md'); if (mainSkill.trim()) { if (mainSkill.includes('{{AVAILABLE_AGENTS}}')) { mainSkill = mainSkill.replace('{{AVAILABLE_AGENTS}}', agentListText); } parts.push('---\n\n# 我的核心主技能 (执行准则)\n\n' + mainSkill.trim()); } const skills = loadFile(agentDir, '.claude/skills/skills.md'); if (skills.trim()) parts.push('---\n\n' + skills.trim()); return parts.join('\n\n---\n\n'); } function loadAgentConfig(agentDir) { try { return JSON.parse(fs.readFileSync(path.join(agentDir, 'config.json'), 'utf-8')); } catch { return null; } } function loadIdentity(agentDir) { const content = loadFile(agentDir, 'identity.md'); const r = { name: 'Agent', emoji: '🤖', greeting: '' }; const nm = content.match(/名称:\s*(.+)/); if (nm) r.name = nm[1].trim(); const em = content.match(/Emoji:\s*(.+)/); if (em) r.emoji = em[1].trim(); const gm = content.match(/## 问候语\n"?([^"]*)"?/s); if (gm) r.greeting = gm[1].trim().replace(/^"|"$/g, ''); return r; } function setupSSE(res) { res.writeHead(200, { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', 'Connection': 'keep-alive', 'X-Accel-Buffering': 'no', }); return (event, data) => res.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`); } // ═══════════════════════════════════════ // 对话存储 — conversationId/messageId 两层 // ═══════════════════════════════════════ function getConvDir(agentId, convId) { const dir = path.join(DATA_DIR, agentId, 'conversations', convId); if (!fs.existsSync(dir)) fs.mkdirSync(dir, { recursive: true }); return dir; } /** 读写 meta.json */ function loadMeta(agentId, convId) { try { return JSON.parse(fs.readFileSync(path.join(DATA_DIR, agentId, 'conversations', convId, 'meta.json'), 'utf-8')); } catch { return null; } } function saveMeta(agentId, convId, meta) { const dir = getConvDir(agentId, convId); meta.updatedAt = new Date().toISOString(); fs.writeFileSync(path.join(dir, 'meta.json'), JSON.stringify(meta, null, 2)); } /** 读写 history.json — 有序消息列表 */ function loadHistory(agentId, convId) { try { return JSON.parse(fs.readFileSync(path.join(DATA_DIR, agentId, 'conversations', convId, 'history.json'), 'utf-8')); } catch { return []; } } function appendHistory(agentId, convId, msg) { const history = loadHistory(agentId, convId); history.push(msg); const dir = getConvDir(agentId, convId); fs.writeFileSync(path.join(dir, 'history.json'), JSON.stringify(history, null, 2)); } /** 写日志(流式或单次) */ function logMessageChunk(agentId, convId, content) { const dir = getConvDir(agentId, convId); fs.appendFileSync(path.join(dir, 'messages.log'), content, 'utf-8'); } /** 传统单次写日志 (为了兼容) */ function logMessage(agentId, convId, role, content) { const ts = new Date().toISOString(); const tag = role === 'user' ? 'USER' : 'AGENT'; const line = `[${ts}] [${tag}]\n${content}\n\n`; logMessageChunk(agentId, convId, line); } function listConversations(agentId) { const agentConvDir = path.join(DATA_DIR, agentId, 'conversations'); if (!fs.existsSync(agentConvDir)) return []; try { return fs.readdirSync(agentConvDir, { withFileTypes: true }) .filter(d => d.isDirectory()) .map(d => { const m = loadMeta(agentId, d.name); if (m) m.id = m.id || m.conversationId || d.name; return m; }) .filter(Boolean) .sort((a, b) => new Date(b.updatedAt).getTime() - new Date(a.updatedAt).getTime()); } catch { return []; } } // ═══════════════════════════════════════ // ACP Session 池 — 支持多轮 + 上下文恢复 // ═══════════════════════════════════════ // key: `${agentId}:${conversationId}` → { protocol, messageCount } const sessions = new Map(); async function getOrCreateProtocol(agentId, convId) { const key = `${agentId}:${convId}`; let entry = sessions.get(key); if (entry && entry.protocol.isConnected) { return entry; } // 关闭旧的 if (entry) { try { entry.protocol.disconnect(); } catch {} } const agentDir = path.join(DATA_DIR, agentId); const identity = loadIdentity(agentDir); const protocol = new ClaudeCliProtocol({ name: identity.name }); await protocol.connect(null, agentDir); await protocol.initialize(); await protocol.newSession({ cwd: agentDir }); const newEntry = { protocol, messageCount: 0, agentId, convId }; sessions.set(key, newEntry); console.log(`[AgentChat] 🧬 ${identity.name} ACP 会话已创建 (conv: ${convId})`); return newEntry; } /** * 构建 prompt — 根据是否首次消息 & 是否有历史上下文 * * 场景 A: 全新对话的第一条消息 → system prompt + 用户消息 * 场景 B: 同一进程中的后续消息 → 直接发消息(ACP 自带上下文) * 场景 C: 服务重启后恢复对话 → system prompt + 历史摘要 + 用户消息 */ function buildPrompt(agentDir, agentId, convId, userMessage, isNewSession) { if (!isNewSession) { // 场景 B: ACP 进程还在,多轮上下文自动保持 return userMessage; } const systemPrompt = buildSystemPrompt(agentDir); const history = loadHistory(agentId, convId); if (history.length === 0) { // 场景 A: 全新对话 return `${systemPrompt}\n\n---\n\n## 用户消息\n\n${userMessage}`; } // 场景 C: 恢复上下文 — 注入历史摘要 const recentHistory = history.slice(-20); // 最近 20 条 const historyText = recentHistory.map(m => { const role = m.role === 'user' ? 'Human' : 'Assistant'; const content = m.content.length > 500 ? m.content.slice(0, 500) + '...(截断)' : m.content; return `${role}: ${content}`; }).join('\n\n'); return `${systemPrompt}\n\n---\n\n## 之前的对话历史(供你参考以保持上下文连贯)\n\n${historyText}\n\n---\n\n## 用户新消息\n\n${userMessage}`; } // ═══════════════════════════════════════ // 路由 // ═══════════════════════════════════════ router.get('/list', (req, res) => { try { if (!fs.existsSync(DATA_DIR)) return res.json([]); const entries = fs.readdirSync(DATA_DIR, { withFileTypes: true }); const agents = entries.filter(e => e.isDirectory()).map(e => { const agentDir = path.join(DATA_DIR, e.name); const config = loadAgentConfig(agentDir); if (!config) return null; const identity = loadIdentity(agentDir); return { id: config.id || e.name, name: identity.name, emoji: identity.emoji, type: config.type, status: config.status, greeting: identity.greeting }; }).filter(Boolean); res.json(agents); } catch (err) { res.status(500).json({ error: err.message }); } }); router.get('/:agentId/info', (req, res) => { const agentDir = path.join(DATA_DIR, req.params.agentId); if (!fs.existsSync(agentDir)) return res.status(404).json({ error: 'Agent not found' }); const config = loadAgentConfig(agentDir); const identity = loadIdentity(agentDir); res.json({ id: config?.id || req.params.agentId, ...identity, type: config?.type, status: config?.status, bootstrapFiles: ['agent.md','soul.md','identity.md','user.md','tools.md','heartbeat.md'].filter(f => loadFile(agentDir, f).trim().length > 0), }); }); /** 列出对话 */ router.get('/:agentId/conversations', (req, res) => { res.json(listConversations(req.params.agentId)); }); /** 获取对话详情 (meta + history) */ router.get('/:agentId/conversations/:convId', (req, res) => { const { agentId, convId } = req.params; const meta = loadMeta(agentId, convId); if (!meta) return res.status(404).json({ error: 'Not found' }); const messages = loadHistory(agentId, convId); res.json({ ...meta, messages }); }); /** 删除对话 */ router.delete('/:agentId/conversations/:convId', (req, res) => { const { agentId, convId } = req.params; try { const dir = path.join(DATA_DIR, agentId, 'conversations', convId); if (fs.existsSync(dir)) fs.rmSync(dir, { recursive: true, force: true }); const key = `${agentId}:${convId}`; const entry = sessions.get(key); if (entry && entry.protocol) entry.protocol.disconnect(); sessions.delete(key); res.json({ success: true }); } catch (err) { res.status(500).json({ error: err.message }); } }); /** 停止当前正在生成的任务 */ router.post('/:agentId/chat/:convId/stop', (req, res) => { const { agentId, convId } = req.params; const key = `${agentId}:${convId}`; const entry = sessions.get(key); if (entry && entry.protocol) { entry.protocol.disconnect(); // 关闭底层 claude cli 进程 sessions.delete(key); // 下次从冷启动重新加载 return res.json({ success: true, message: 'Agent 已中断' }); } res.json({ success: false, message: '未找到运行中的 Agent 进程' }); }); /** 读取 bootstrap file */ router.get('/:agentId/bootstrap/:filename', (req, res) => { const allowed = ['agent.md','soul.md','identity.md','user.md','tools.md','heartbeat.md']; if (!allowed.includes(req.params.filename)) return res.status(400).json({ error: 'Invalid' }); res.json({ filename: req.params.filename, content: loadFile(path.join(DATA_DIR, req.params.agentId), req.params.filename) }); }); // ═══════════════════════════════════════ // POST /api/agent/:agentId/chat — SSE 流式对话 // ═══════════════════════════════════════ router.post('/:agentId/chat', async (req, res) => { const { message, conversationId: reqConvId } = req.body; if (!message) return res.status(400).json({ error: 'message is required' }); const agentId = req.params.agentId; const agentDir = path.join(DATA_DIR, agentId); if (!fs.existsSync(agentDir)) return res.status(404).json({ error: 'Agent not found' }); const convId = reqConvId || `conv-${Date.now().toString(36)}`; const msgId = `msg-${Date.now()}-${randomUUID().substring(0, 8)}`; const sendSSE = setupSSE(res); const identity = loadIdentity(agentDir); // 确保 meta.json 存在 let meta = loadMeta(agentId, convId); if (!meta) { meta = { conversationId: convId, agentId, title: message.slice(0, 50), createdAt: new Date().toISOString(), updatedAt: new Date().toISOString(), status: 'running', stats: { messageCount: 0, toolCallCount: 0 }, }; } if (!meta.stats) meta.stats = { messageCount: 0, toolCallCount: 0 }; meta.stats.messageCount = (meta.stats.messageCount || 0) + 1; meta.status = 'running'; saveMeta(agentId, convId, meta); // 写用户消息到两层存储 logMessage(agentId, convId, 'user', message); appendHistory(agentId, convId, { role: 'user', content: message, messageId: msgId, timestamp: new Date().toISOString() }); sendSSE('session', { conversationId: convId, agentId, messageId: msgId }); sendSSE('thinking', { message: `${identity.emoji} ${identity.name} 思考中...` }); let fullOutput = ''; // 记录 Agent 开始回复的标识头 logMessageChunk(agentId, convId, `[${new Date().toISOString()}] [AGENT]\n`); try { // 获取或创建 ACP 会话(per conversation) const entry = await getOrCreateProtocol(agentId, convId); const isNewSession = entry.messageCount === 0; // 构建 prompt(自动处理首次/后续/恢复三种场景) let prompt = buildPrompt(agentDir, agentId, convId, message, isNewSession); entry.messageCount++; let loopCount = 0; while (loopCount < 10) { loopCount++; fullOutput = ''; // Reset fullOutput for this turn // 监听流式事件 const onUpdate = (payload) => { const update = payload.update; if (update?.sessionUpdate === 'agent_message_chunk' && update.content?.text) { fullOutput += update.content.text; logMessageChunk(agentId, convId, update.content.text); sendSSE('agent_output', { agent: agentId, message: update.content.text }); } else if (update?.sessionUpdate === 'agent_thought_chunk' && update.content?.text) { logMessageChunk(agentId, convId, update.content.text); sendSSE('thinking', { message: update.content.text }); } }; const onAcpEvent = (event) => { if (event.method === 'claude/tool_use_start') { const toolMsg = `\n🔧 [TOOL_CALL] ${event.params.name} (ID: ${event.params.toolUseId || 'unknown'})\n`; logMessageChunk(agentId, convId, toolMsg); sendSSE('tool_call', { tool: event.params.name, message: `🔧 调用工具: ${event.params.name}` }); meta.stats.toolCallCount = (meta.stats.toolCallCount || 0) + 1; } else if (event.method === 'claude/tool_use_success') { logMessageChunk(agentId, convId, `✅ [TOOL_SUCCESS] ${event.params.name}\n`); } else if (event.method === 'claude/tool_use_failure') { logMessageChunk(agentId, convId, `❌ [TOOL_ERROR] ${event.params.error}\n`); } else if (event.method === 'session/result') { if (!fullOutput && event.params?.result) { fullOutput = event.params.result; sendSSE('agent_output', { agent: agentId, message: event.params.result }); } } }; entry.protocol.on('session_update', onUpdate); entry.protocol.on('acp_event', onAcpEvent); // 发送 prompt 并等待完成 await entry.protocol.sendPrompt(prompt); // 清理监听 entry.protocol.off('session_update', onUpdate); entry.protocol.off('acp_event', onAcpEvent); // 持久化当前轮次的回复 if (fullOutput) { logMessageChunk(agentId, convId, '\n\n'); appendHistory(agentId, convId, { role: 'assistant', content: fullOutput, messageId: `msg-${Date.now()}-${loopCount}`, timestamp: new Date().toISOString() }); } // 解析是否带有 [CALL: sub_agent] 标记 const calls = Parser.parseCalls(fullOutput, { allowParallel: true, agents: fullSystemConfig.agents }); if (calls.length === 0) { break; // 没有后续调用,当前回合结束 } // 存在 Sub-Agent 嵌套调用 logMessageChunk(agentId, convId, `\n\n[System] 执行子 Agent 编排调度(并行)...\n`); sendSSE('thinking', { message: `🔄 分析到系统编排指令,正在并行启动 ${calls.length} 个子代理...` }); const resultsText = []; // 并行执行所有的 calls await Promise.all(calls.map(async (call) => { const callId = `sub-${randomUUID().substring(0, 6)}`; sendSSE('sub_agent', { event: 'start', callId, agent: call.agent, message: `🚀 开始执行: ${call.agent}...` }); logMessageChunk(agentId, convId, `\n--- 调度 [${callId}]: ${call.agent} ---\n任务: ${call.task}\n\n`); try { // 调用 Sub Agent (l1/l2/l3 或其他) const res = await agentManager.callAgent(call.agent, call.task, { metadata: { parentConvId: convId, layer: 'sub' }, onData: (chunk) => { sendSSE('sub_agent', { event: 'chunk', callId, agent: call.agent, chunk }); } }); resultsText.push(`[Sub Agent ${call.agent} (${callId}) 返回结果]\n${res.output}\n[/Sub Agent ${call.agent} (${callId}) 返回结果]`); logMessageChunk(agentId, convId, `✅ 子 Agent ${call.agent} [${callId}] 执行成功。\n`); sendSSE('sub_agent', { event: 'done', callId, agent: call.agent, status: 'success', message: `✅ [${call.agent}] 执行完毕.` }); } catch (subErr) { resultsText.push(`[Sub Agent ${call.agent} (${callId}) 执行失败]\n${subErr.message}\n[/Sub Agent ${call.agent} (${callId}) 执行失败]`); logMessageChunk(agentId, convId, `❌ 子 Agent ${call.agent} [${callId}] 失败: ${subErr.message}\n`); sendSSE('sub_agent', { event: 'done', callId, agent: call.agent, status: 'error', message: `❌ [${call.agent}] 抛出异常: ${subErr.message}` }); } })); // 构造下一次反思与后续动作的 Prompt prompt = ` ## Sub Agent 执行结果反馈 ${resultsText.join('\n\n')} --- 请对本次执行结果进行 Reflection 反思,并在确定所有步骤完成时,记得在对话结尾加上 [RESULT: DONE]。 `; // 更新消息计数和记录 User 视角的 system 反馈消息 entry.messageCount++; logMessageChunk(agentId, convId, `\n[${new Date().toISOString()}] [SYSTEM FEEDBACK]\n${prompt}\n\n`); appendHistory(agentId, convId, { role: 'user', content: prompt, messageId: `sys-${Date.now()}`, timestamp: new Date().toISOString() }); // 循环继续,下一次将会把这个反馈 prompt 发给 Innovator logMessageChunk(agentId, convId, `[${new Date().toISOString()}] [AGENT]\n`); } meta.status = 'completed'; saveMeta(agentId, convId, meta); sendSSE('result', { conversationId: convId, agentId, messageId: msgId, output: fullOutput }); } catch (err) { console.error('[AgentChat] Error:', err.message); logMessageChunk(agentId, convId, `\n\n[ERROR]\n${err.message}\n\n`); meta.status = 'failed'; meta.error = err.message; saveMeta(agentId, convId, meta); // 清理断开的 session const key = `${agentId}:${convId}`; sessions.delete(key); sendSSE('error', { message: err.message }); } finally { res.end(); } }); module.exports = router;