'use strict'; /** * 智能问数智能体 · 主循环(harness 核心) * ---------------------------------------------------------------------------- * 每回合:LLM → parseAgentAction → 护栏(求交/钳制/预算/循环侦测)→ 工具执行 * → 观察回注 → …,直到模型给出 final 或触发预算强制收敛。 * 消息形状保持严格 user/assistant 交替;工具观察以 [工具结果] 文本回注(digest * 已在工具内裁剪)。取消在回合间隙检查(进行中的 HTTP 调用不可中断,属已知边界)。 */ const llm = require('./llm.js'); const guard = require('./guard.js'); const prompt = require('./prompt.js'); const registry = require('./tools/registry.js'); const skills = require('./skills.js'); const metrics = require('./metrics.js'); const provenance = require('./provenance.js'); const trace = require('./trace.js'); const nlQuery = require('../../nl-query.js'); const z2 = (n) => String(n).padStart(2, '0'); function cancelledError() { const e = new Error('已取消'); e.code = 'AGENT_CANCELLED'; return e; } /** 「应用到报表」载荷:取最后一次成功查询的钳制后参数,前端据此落报表筛选 */ function appliedFrom(tool, args) { if (!args) return null; if (tool === 'query_report') { return { mode: args.mode, from: args.from, to: args.to, shopIds: args.shopIds || [], metrics: args.metrics }; } if (tool === 'compare_periods') { return { mode: 'shop', from: args.from, to: args.to, shopIds: args.shops || [], metrics: args.metrics }; } return null; } function fallbackAnswer(rec, reason) { const lastOk = [...rec.steps].reverse().find((s) => s.type === 'tool' && s.ok); if (lastOk && lastOk.summary) { return `(${reason},未生成完整结论)最后一步取数结果:\n\n${String(lastOk.summary).slice(0, 800)}`; } return `(${reason})本次未能完成分析,请重试或换一个更具体的问题。`; } /** * 运行一次智能体回合循环。 * @param {object} o { rec, svc, cfg: {chatApiBase,chatApiKey,chatModel}, mock: {next(rec)}|null } * @returns {object} { answer, chart, appliedFrom, thought } */ async function runAgent(o) { const { rec, svc, cfg, mock } = o; const meta = svc.store.getEntity(rec.entity) || {}; const shops = meta.shops || []; const dateRange = nlQuery.lakeDateRange(rec.entity, svc); const ctx = { svc, entityKey: rec.entity, meta, shops, dateRange, cloud: o.cloud || null }; const today = (() => { const d = new Date(); return `${d.getFullYear()}${z2(d.getMonth() + 1)}${z2(d.getDate())}`; })(); const messages = [{ role: 'system', content: prompt.buildSystemPrompt({ svc, entityName: meta.name || rec.entity, shops, dateRange, today, toolsText: registry.manifestText(), skillsText: skills.manifestText(), metricsText: metrics.manifestText(), pointedSkill: rec.skill ? skills.get(rec.skill) : null, fewShotsText: o.fewShotsText || '', }), }]; for (const h of rec.history || []) messages.push(h); messages.push({ role: 'user', content: rec.question }); const startedAt = Date.now(); let repairs = 0; let repeatWarned = false; let lastApplied = null; let lastReportUrl = null; const detector = new guard.LoopDetector(); const callModel = async (msgs) => { rec.usage.llmCalls += 1; trace.pushEvent(rec, 'llm.call', { turns: msgs.length, mock: !!mock }); const t0 = Date.now(); if (mock) { return { reply: mock.next(rec), usage: null, ms: Date.now() - t0 }; } // 模型路由:深度分析优先用专用模型(chatModelAgent),缺省回落主对话模型 const out = await llm.chatCompletionDetailed({ apiBase: cfg.chatApiBase, apiKey: cfg.chatApiKey, model: cfg.chatModelAgent || cfg.chatModel, messages: msgs, timeoutMs: guard.BUDGET.llmTimeoutMs, temperature: 0.2, jsonMode: !!cfg.chatJsonMode, }); // 治理:run 级 token 用量累计(成本展示) if (out.usage) { rec.usage.tokensIn = (rec.usage.tokensIn || 0) + (out.usage.promptTokens || 0); rec.usage.tokensOut = (rec.usage.tokensOut || 0) + (out.usage.completionTokens || 0); } return { reply: out.content, usage: out.usage, ms: Date.now() - t0 }; }; const finishWith = (parsed) => { // 数字溯源 post-check(零 token 规则):回答中的数值必须能在工具结果里找到出处 const prov = provenance.checkProvenance(parsed.final.answer, rec.steps); rec.provenance = { total: prov.total, untracedCount: prov.untraced.length, untraced: prov.untraced, ratio: prov.ratio }; return { answer: parsed.final.answer, chart: guard.clampChart(parsed.final.chart, svc), appliedFrom: lastApplied, reportUrl: lastReportUrl, thought: parsed.thought || '', }; }; /** 工具执行(主循环与复用打法共用):留痕/verdict 归档/应用与报告链接捕获 */ const executeTool = async (toolName, tool, args) => { rec.usage.toolCalls += 1; const step = { type: 'tool', tool: toolName, args, ok: false, ms: 0 }; rec.steps.push(step); trace.pushEvent(rec, 'tool.start', { tool: toolName }); const t0 = Date.now(); try { const out = await tool.handler(ctx, args); step.ok = true; step.summary = String(out.summary || '').slice(0, 4000); step.data = out.data || null; if (out.data && out.data.totalRows != null) step.rows = out.data.totalRows; step.bytes = JSON.stringify(out.data || null).length; // 校验类工具的 verdict 归档(quality/identity/cross-layer/freshness/coverage/cloud) if (out.data && Array.isArray(out.data.verdicts)) { for (const v of out.data.verdicts) { rec.verdicts.push({ id: `v${rec.verdicts.length + 1}`, tool: toolName, ...v }); } } const applied = appliedFrom(toolName, args); if (applied) lastApplied = applied; if (toolName === 'build_report' && out.data && out.data.url) lastReportUrl = out.data.url; } catch (e) { step.error = String(e.message || e).slice(0, 500); } finally { step.ms = Date.now() - t0; trace.pushEvent(rec, 'tool.done', { tool: toolName, ok: step.ok, ms: step.ms, error: step.error }); trace.persistRun(rec); // 逐步落盘:每个工具(查询/提取/计算)完成即对轮询可见 } return step; }; const forceFinal = async (reason) => { messages.push({ role: 'user', content: `${reason}:请立即基于以上已有信息输出 final(JSON),不要再调用工具。` }); try { const turn = await callModel(messages); const parsed = llm.parseAgentAction(turn.reply); if (parsed.type === 'final') return finishWith(parsed); } catch { /* 强制收敛失败走兜底话术 */ } return { answer: fallbackAnswer(rec, reason), chart: null, appliedFrom: lastApplied, reportUrl: lastReportUrl, thought: '' }; }; // 打法复用:命中已验证查询序列 → 直接重跑(缓存的是打法,数据按当前湖重算)→ 单次收敛 if (o.plan && Array.isArray(o.plan.steps) && o.plan.steps.length) { rec.replayFrom = o.plan.question; rec.steps.push({ type: 'thought', text: `命中已验证打法「${o.plan.question}」,直接复用其查询序列(数据按当前湖重算)`, ms: 0 }); trace.persistRun(rec); for (const ps of o.plan.steps) { if (rec.cancelRequested) throw cancelledError(); const tool = registry.get(ps.tool); if (!tool) continue; let args; try { args = tool.sanitize(ps.args || {}, ctx); } catch { continue; } // 口径变更等导致的参数失配:跳过该步 const step = await executeTool(ps.tool, tool, args); messages.push({ role: 'user', content: `[工具结果] ${ps.tool}\n${step.summary || step.error || ''}` }); } messages.push({ role: 'user', content: '以上为按既有打法重跑的最新数据结果,请立即基于其输出 final(JSON),不要再调用工具。' }); try { const turn = await callModel(messages); const parsed = llm.parseAgentAction(turn.reply); if (parsed.type === 'final') return finishWith(parsed); } catch { /* 收敛失败走兜底 */ } return { answer: fallbackAnswer(rec, '复用打法收敛失败'), chart: null, appliedFrom: lastApplied, reportUrl: lastReportUrl, thought: '' }; } while (true) { if (rec.cancelRequested) throw cancelledError(); // 预算(轮次/工具次数/墙钟)触发强制收敛:给模型最后一次输出 final 的机会 if (rec.usage.llmCalls >= guard.BUDGET.maxTurns || rec.usage.toolCalls >= guard.BUDGET.maxToolCalls || Date.now() - startedAt >= guard.BUDGET.runTimeoutMs) { return forceFinal('已达预算上限'); } let reply; let turnMeta = null; try { turnMeta = await callModel(messages); reply = turnMeta.reply; } catch (e) { throw new Error('对话模型调用失败:' + (e.message || e)); } let parsed; try { parsed = llm.parseAgentAction(reply); repairs = 0; } catch (e) { if (repairs >= guard.BUDGET.maxRepairs) throw new Error('模型连续输出无法解析:' + (e.message || e)); repairs += 1; trace.pushEvent(rec, 'notice', { text: '解析失败,回注修复' }); messages.push({ role: 'assistant', content: String(reply).slice(0, 2000) }); messages.push({ role: 'user', content: `你的输出无法解析(${e.message})。请重新输出:只输出一个 JSON 对象,action 或 final 二选一。` }); continue; } rec.steps.push({ type: 'thought', turn: rec.usage.llmCalls, text: parsed.thought, ms: turnMeta.ms, tokensIn: turnMeta.usage ? turnMeta.usage.promptTokens : null, tokensOut: turnMeta.usage ? turnMeta.usage.completionTokens : null, }); trace.persistRun(rec); // 逐步落盘:前端轮询实时看到思考/查询/计算推进 messages.push({ role: 'assistant', content: String(reply).slice(0, 4000) }); if (parsed.type === 'final') { return finishWith(parsed); } // 工具名与清单求交(防幻觉工具名) const tool = registry.get(parsed.tool); if (!tool) { if (repairs >= guard.BUDGET.maxRepairs) throw new Error(`模型连续引用不存在的工具(${parsed.tool})`); repairs += 1; trace.pushEvent(rec, 'notice', { text: `未知工具 ${parsed.tool},回注清单` }); messages.push({ role: 'user', content: `工具「${parsed.tool}」不存在。可用工具:\n${registry.manifestText()}\n请重新输出动作 JSON。`, }); continue; } // 参数钳制(sanitize 抛错 = 参数非法,回注让模型修正) let args; try { args = tool.sanitize(parsed.args, ctx); } catch (e) { if (repairs >= guard.BUDGET.maxRepairs) throw new Error(`工具参数连续非法(${parsed.tool}):${e.message || e}`); repairs += 1; trace.pushEvent(rec, 'notice', { text: `${parsed.tool} 参数非法,回注修复` }); messages.push({ role: 'user', content: `工具「${parsed.tool}」参数非法:${e.message || e}。请修正后重新输出动作 JSON。` }); continue; } // 循环侦测:完全相同的调用先提醒,再犯强制收敛 const verdict = detector.record(parsed.tool, args); if (verdict === 'loop') { if (repeatWarned) { trace.pushEvent(rec, 'notice', { text: '重复调用循环,强制收敛' }); return { answer: fallbackAnswer(rec, '检测到重复调用'), chart: null, appliedFrom: lastApplied, reportUrl: lastReportUrl, thought: '' }; } repeatWarned = true; messages.push({ role: 'user', content: '提醒:你已连续多次发出完全相同的调用。若信息已足够请立即输出 final;确需重试请更换参数。' }); continue; } const step = await executeTool(parsed.tool, tool, args); messages.push(step.ok ? { role: 'user', content: `[工具结果] ${parsed.tool}\n${step.summary}` } : { role: 'user', content: `[工具失败] ${parsed.tool}:${step.error}。请修正参数重试或换路径。` }); if (step.ok && verdict === 'repeat') { messages.push({ role: 'user', content: '提醒:这与上一次调用完全相同。若信息已足够请输出 final;确需重试请更换参数。' }); } } } module.exports = { runAgent };