hyreal-gen-pic/server/agent/loop.js

'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 };