smqjh-agent/smqjh-admin-agent/src/main/assistant/assistantOrchestrator.ts

import type {
  AssistantResultTable,
  AssistantRunRequest,
  AssistantRunResponse,
  AssistantToolCallRecord,
  PriceComparisonRow
} from "../../shared/types";
import type { AgentRunner } from "../agent/runner";
import { listTasks } from "../agent/tasks";
import type {
  AssistantToolDescriptor,
  AssistantToolObservation,
  AssistantToolPlan
} from "../deepseek/deepSeekClient";
import type { LogStore } from "../agent/logStore";
import type { McpClient, McpToolDescriptor } from "../mcp/mcpClient";

export class AssistantOrchestrator {
  constructor(
    private readonly deps: {
      mcpClient: McpClient;
      runner: AgentRunner;
      logStore: LogStore;
    }
  ) {}

  async run(request: AssistantRunRequest): Promise<AssistantRunResponse> {
    const tools = await this.loadToolDescriptors();
    const byName = new Map(tools.map((tool) => [tool.name, tool]));
    const observations: AssistantToolObservation[] = [];
    const seenCalls = new Set<string>();
    const maxSteps = 6;

    for (let step = 0; step < maxSteps; step += 1) {
      const plan = await this.planViaMcp(request, tools, observations);

      if (plan.mode !== "tool" || plan.toolCalls.length === 0) {
        break;
      }

      const executableCalls = plan.toolCalls.filter((call) => {
        const signature = toolCallSignature(call.name, call.arguments);
        if (seenCalls.has(signature)) {
          this.deps.logStore.add("warn", `智能体跳过重复工具调用：${call.name}`);
          return false;
        }
        seenCalls.add(signature);
        return true;
      });

      if (executableCalls.length === 0) {
        break;
      }

      this.deps.logStore.add("info", "Agent loop 执行工具步骤", {
        step: step + 1,
        tools: executableCalls.map((call) => call.name).join(", ")
      });

      for (const call of executableCalls) {
        const resolvedName = resolveToolName(call.name, byName);
        const descriptor = resolvedName ? byName.get(resolvedName) : undefined;
        if (!descriptor) {
          observations.push({
            name: call.name,
            source: "local",
            arguments: call.arguments,
            ok: false,
            error: "DeepSeek 选择了未注册工具"
          });
          continue;
        }

        observations.push(await this.executeTool(descriptor, call.arguments));
      }
    }

    if (observations.length === 0) {
      const result = await this.chatViaMcp(request);
      return {
        content: result.content,
        model: result.model,
        usedMcp: true,
        toolCalls: []
      };
    }

    const toolCalls: AssistantToolCallRecord[] = observations.map((item) => ({
      name: item.name,
      source: item.source,
      arguments: item.arguments,
      ok: item.ok,
      result: item.result,
      error: item.error
    }));
    const tables = extractTables(toolCalls);
    const comparisonRows = extractComparisonRows(toolCalls);

    let content: string;
    let model = "tool-only";
    try {
      const summary = await this.summarizeViaMcp(request, observations);
      content = summary.content;
      model = summary.model;
    } catch (error) {
      const message = error instanceof Error ? error.message : String(error);
      this.deps.logStore.add("warn", `MCP DeepSeek 工具结果总结失败：${message}`);
      content = fallbackToolSummary(toolCalls, tables, comparisonRows);
    }

    return {
      content,
      model,
      usedMcp: toolCalls.some((item) => item.source === "mcp"),
      toolCalls,
      tables: tables.length ? tables : undefined,
      comparisonRows: comparisonRows.length ? comparisonRows : undefined
    };
  }

  private async loadToolDescriptors(): Promise<AssistantToolDescriptor[]> {
    const tools: AssistantToolDescriptor[] = [];
    try {
      const mcpTools = await this.deps.mcpClient.listTools();
      tools.push(...mcpTools.filter((tool) => !tool.name.startsWith("smqjh.ai.")).map(toMcpDescriptor));
    } catch (error) {
      const message = error instanceof Error ? error.message : String(error);
      this.deps.logStore.add("warn", `MCP 工具列表读取失败：${message}`);
    }

    const localTasks = listTasks();
    const localNames = tools.length > 0 ? new Set(["product.price.compare"]) : new Set(localTasks.map((task) => task.id));
    for (const task of localTasks) {
      if (!localNames.has(task.id)) {
        continue;
      }
      tools.push({
        name: task.id,
        source: "local",
        title: task.title,
        description: task.description,
        inputSchema: {
          type: "object",
          properties: Object.fromEntries(
            task.params.map((param) => [
              param.key,
              {
                type: "string",
                description: [param.label, param.placeholder].filter(Boolean).join("；")
              }
            ])
          ),
          required: task.params.filter((param) => param.required).map((param) => param.key),
          additionalProperties: false
        }
      });
    }

    return tools;
  }

  private async planViaMcp(
    request: AssistantRunRequest,
    tools: AssistantToolDescriptor[],
    observations: AssistantToolObservation[]
  ): Promise<AssistantToolPlan> {
    const result = await this.deps.mcpClient.callTool("smqjh.ai.tool.plan", {
      request,
      tools,
      observations
    });
    const record = asRecord(result.structuredContent);
    const rawCalls = Array.isArray(record.toolCalls) ? record.toolCalls : [];
    const toolCalls = rawCalls
      .map((item): AssistantToolPlan["toolCalls"][number] | undefined => {
        const call = asRecord(item);
        const name = typeof call.name === "string" ? call.name.trim() : "";
        if (!name) {
          return undefined;
        }
        return {
          name,
          arguments: repairToolArguments(name, asRecord(call.arguments), request.message)
        };
      })
      .filter((item): item is AssistantToolPlan["toolCalls"][number] => Boolean(item))
      .slice(0, 3);

    const mode = record.mode === "tool" && toolCalls.length > 0 ? "tool" : "chat";
    this.deps.logStore.add("info", "MCP DeepSeek 工具规划完成", {
      mode,
      toolCount: toolCalls.length,
      reason: typeof record.reason === "string" ? record.reason : ""
    });
    return {
      mode,
      toolCalls: mode === "tool" ? toolCalls : [],
      replyWhenNoTool: typeof record.replyWhenNoTool === "string" ? record.replyWhenNoTool : undefined,
      reason: typeof record.reason === "string" ? record.reason : undefined
    };
  }

  private async summarizeViaMcp(request: AssistantRunRequest, observations: AssistantToolObservation[]): Promise<{ content: string; model: string }> {
    const result = await this.deps.mcpClient.callTool("smqjh.ai.tool.summarize", {
      request,
      observations
    });
    const record = asRecord(result.structuredContent);
    return {
      content: typeof record.content === "string" ? record.content : "工具已执行，但 MCP DeepSeek 没有生成有效总结。",
      model: typeof record.model === "string" ? record.model : "mcp-deepseek"
    };
  }

  private async chatViaMcp(request: AssistantRunRequest): Promise<{ content: string; model: string }> {
    const result = await this.deps.mcpClient.callTool("smqjh.ai.chat", {
      request
    });
    const record = asRecord(result.structuredContent);
    return {
      content: typeof record.content === "string" ? record.content : "MCP DeepSeek 没有返回有效内容。",
      model: typeof record.model === "string" ? record.model : "mcp-deepseek"
    };
  }

  private async executeTool(descriptor: AssistantToolDescriptor, args: Record<string, unknown>): Promise<AssistantToolObservation> {
    try {
      if (descriptor.source === "mcp") {
        const result = await this.deps.mcpClient.callTool(descriptor.name, args);
        this.deps.logStore.add("info", `MCP 工具执行完成：${descriptor.name}`);
        return {
          name: descriptor.name,
          source: "mcp",
          arguments: args,
          ok: true,
          result: result.structuredContent ?? result
        };
      }

      const result = await this.deps.runner.run({
        taskId: descriptor.name,
        params: toStringParams(args)
      });
      return {
        name: descriptor.name,
        source: "local",
        arguments: args,
        ok: result.success,
        result,
        error: result.success ? undefined : result.message
      };
    } catch (error) {
      const message = error instanceof Error ? error.message : String(error);
      this.deps.logStore.add("error", `工具执行失败：${descriptor.name}：${message}`);
      return {
        name: descriptor.name,
        source: descriptor.source,
        arguments: args,
        ok: false,
        error: message
      };
    }
  }
}

function toMcpDescriptor(tool: McpToolDescriptor): AssistantToolDescriptor {
  return {
    name: tool.name,
    source: "mcp",
    title: tool.title,
    description: tool.description,
    inputSchema: tool.inputSchema
  };
}

function asRecord(value: unknown): Record<string, unknown> {
  return value && typeof value === "object" && !Array.isArray(value) ? (value as Record<string, unknown>) : {};
}

function resolveToolName(name: string, byName: Map<string, AssistantToolDescriptor>): string | undefined {
  const candidates = [
    name,
    name.startsWith("smqjh.") ? name.slice("smqjh.".length) : `smqjh.${name}`,
    name === "smqjh.product.price.compare" ? "product.price.compare" : "",
    name === "product.lookup.summary" ? "smqjh.product.lookup.summary" : "",
    name === "order.count.query" ? "smqjh.order.count.query" : "",
    name === "database.readonly.query" ? "smqjh.database.readonly.query" : "",
    name === "cloud.health" ? "smqjh.cloud.health" : ""
  ].filter(Boolean);
  return candidates.find((candidate) => byName.has(candidate));
}

function toStringParams(args: Record<string, unknown>): Record<string, string> {
  return Object.fromEntries(
    Object.entries(args).map(([key, value]) => [key, value === undefined || value === null ? "" : String(value)])
  );
}

function toolCallSignature(name: string, args: Record<string, unknown>): string {
  return `${name}:${stableStringify(args)}`;
}

function stableStringify(value: unknown): string {
  if (!value || typeof value !== "object") {
    return JSON.stringify(value);
  }
  if (Array.isArray(value)) {
    return `[${value.map(stableStringify).join(",")}]`;
  }
  const record = value as Record<string, unknown>;
  return `{${Object.keys(record)
    .sort()
    .map((key) => `${JSON.stringify(key)}:${stableStringify(record[key])}`)
    .join(",")}}`;
}

function repairToolArguments(name: string, args: Record<string, unknown>, message: string): Record<string, unknown> {
  if (!isProductTool(name)) {
    return args;
  }
  const current = typeof args.productKeyword === "string" ? args.productKeyword.trim() : "";
  const extracted = extractProductKeywordFromMessage(message);
  if (!current || (extracted && !productKeywordMatchesMessage(current, message))) {
    return {
      ...args,
      productKeyword: extracted || current
    };
  }
  return args;
}

function isProductTool(name: string): boolean {
  return name === "smqjh.product.lookup.summary" || name === "product.price.compare" || name === "smqjh.product.price.compare";
}

function productKeywordMatchesMessage(keyword: string, message: string): boolean {
  const normalizedKeyword = normalizeProductText(keyword).replace(/\d+(?:\.\d+)?(?:ml|毫升|l|升|g|克|kg|千克)?/g, "");
  const normalizedMessage = normalizeProductText(message);
  const meaningful = normalizedKeyword.replace(/[^\u4e00-\u9fa5a-z]/g, "");
  if (meaningful.length < 2) {
    return false;
  }
  if (normalizedMessage.includes(meaningful)) {
    return true;
  }
  const chars = Array.from(new Set(meaningful.split("")));
  const overlap = chars.filter((char) => normalizedMessage.includes(char)).length;
  return overlap >= Math.min(3, chars.length);
}

function extractProductKeywordFromMessage(message: string): string {
  let text = message
    .replace(/[，。！？?；;：:]/g, " ")
    .replace(/帮我|麻烦|查询一下|查一下|查询|查看|当前|业务系统|系统里面|系统里|后台|我方|我们的|商品库|商品表/g, " ")
    .replace(/和其它电商平台|和其他电商平台|其它电商平台|其他电商平台|淘宝|京东|拼多多|苏宁|慢慢买|价格对比|做个对比|对比/g, " ")
    .replace(/商品描述是什么|商品描述|描述是什么|描述|价格是多少|价格|定价|是多少|是什么|呢/g, " ")
    .replace(/\s+/g, " ")
    .trim();

  const specMatch = text.match(/[\u4e00-\u9fa5A-Za-z0-9（）()_\-\s/*×x.]+?(?:\d+(?:\.\d+)?\s*(?:ml|mL|ML|毫升|L|l|升|g|克|kg|千克)\s*(?:[*×x/]\s*\d+\s*(?:瓶|罐|箱|盒|袋|件|包)?)?|瓶|罐|箱|盒|袋|件|包)/);
  if (specMatch?.[0]) {
    text = specMatch[0].trim();
  }
  return text
    .replace(/^(的|了|在|里|中|一下|这个|那个)\s*/g, "")
    .replace(/\s+(的|了|在|里|中|一下)$/g, "")
    .replace(/\s+/g, " ")
    .trim();
}

function normalizeProductText(value: string): string {
  return value
    .toLowerCase()
    .replace(/×/g, "*")
    .replace(/\s+/g, "")
    .trim();
}

function extractTables(toolCalls: AssistantToolCallRecord[]): AssistantResultTable[] {
  const tables: AssistantResultTable[] = [];
  for (const call of toolCalls) {
    tables.push(...extractTablesFromValue(toolTitle(call.name), call.result));
  }
  return tables.slice(0, 4);
}

function extractTablesFromValue(title: string, value: unknown): AssistantResultTable[] {
  const tables: AssistantResultTable[] = [];
  const seen = new WeakSet<object>();

  function visit(node: unknown, currentTitle: string, depth: number): void {
    if (!node || typeof node !== "object" || depth > 4) {
      return;
    }
    if (seen.has(node)) {
      return;
    }
    seen.add(node);

    const record = node as Record<string, unknown>;
    const rows = Array.isArray(record.rows) ? record.rows : undefined;
    const columns = Array.isArray(record.columns) ? record.columns.map((item) => String(item)) : undefined;
    if (rows && rows.every(isRecord)) {
      const rowRecords = rows.map((row) => normalizeRow(row as Record<string, unknown>));
      const rowColumns = columns?.length ? columns : Array.from(new Set(rowRecords.flatMap((row) => Object.keys(row))));
      if (rowColumns.length > 0) {
        tables.push({
          title: buildTableTitle(currentTitle, record),
          columns: rowColumns,
          rows: rowRecords
        });
      }
    }

    for (const key of ["data", "structuredContent", "result", "record"]) {
      if (record[key]) {
        visit(record[key], currentTitle, depth + 1);
      }
    }
  }

  visit(value, title, 0);
  return tables;
}

function extractComparisonRows(toolCalls: AssistantToolCallRecord[]): PriceComparisonRow[] {
  const rows: PriceComparisonRow[] = [];
  for (const call of toolCalls) {
    collectComparisonRows(call.result, rows, 0);
  }
  return rows;
}

function collectComparisonRows(value: unknown, rows: PriceComparisonRow[], depth: number): void {
  if (!value || typeof value !== "object" || depth > 4) {
    return;
  }
  const record = value as Record<string, unknown>;
  if (Array.isArray(record.comparisonRows)) {
    for (const row of record.comparisonRows) {
      if (isPriceComparisonRow(row)) {
        rows.push(row);
      }
    }
  }
  for (const key of ["data", "structuredContent", "result"]) {
    if (record[key]) {
      collectComparisonRows(record[key], rows, depth + 1);
    }
  }
}

function isPriceComparisonRow(value: unknown): value is PriceComparisonRow {
  if (!isRecord(value)) {
    return false;
  }
  return ["platform", "productName", "price", "status", "source", "note"].every((key) => typeof value[key] === "string");
}

function isRecord(value: unknown): value is Record<string, unknown> {
  return Boolean(value && typeof value === "object" && !Array.isArray(value));
}

function normalizeRow(row: Record<string, unknown>): Record<string, unknown> {
  return Object.fromEntries(
    Object.entries(row).map(([key, value]) => [key, value === undefined || value === null ? "" : String(value)])
  );
}

function buildTableTitle(baseTitle: string, record: Record<string, unknown>): string {
  const rowCount = record.rowCount === undefined ? "" : `（${record.rowCount} 行）`;
  return `${baseTitle}${rowCount}`;
}

function toolTitle(name: string): string {
  const titles: Record<string, string> = {
    "smqjh.database.readonly.query": "数据库只读查询结果",
    "smqjh.database.smart.query": "智能数据库查询结果",
    "smqjh.schema.search": "业务表搜索结果",
    "smqjh.schema.getTable": "业务表说明",
    "smqjh.schema.businessRules": "业务规则",
    "smqjh.order.count.query": "订单统计结果",
    "smqjh.product.lookup.summary": "商品资料查询结果",
    "smqjh.settlement.enterprise.list": "月结企业清单",
    "smqjh.settlement.monthly.plan": "企业月结计划",
    "product.price.compare": "商品价格对比"
  };
  return titles[name] ?? name;
}

function fallbackToolSummary(toolCalls: AssistantToolCallRecord[], tables: AssistantResultTable[], comparisonRows: PriceComparisonRow[]): string {
  const okCount = toolCalls.filter((item) => item.ok).length;
  const failed = toolCalls.filter((item) => !item.ok);
  const lines = [`已执行 ${toolCalls.length} 个工具，成功 ${okCount} 个。`];
  if (tables.length || comparisonRows.length) {
    lines.push("结果已整理为下方表格。");
  }
  if (failed.length) {
    lines.push(`失败工具：${failed.map((item) => `${item.name}（${item.error || "调用失败"}）`).join("；")}`);
  }
  const sqlEvidence = findExecutedSql(toolCalls);
  if (sqlEvidence) {
    lines.push(`依据：${sqlEvidence}`);
  }
  return lines.join("\n\n");
}

function findExecutedSql(toolCalls: AssistantToolCallRecord[]): string {
  for (const call of toolCalls) {
    const sql = findKeyValue(call.result, "executedSql", 0);
    if (typeof sql === "string" && sql.trim()) {
      return sql.trim();
    }
  }
  return "";
}

function findKeyValue(value: unknown, key: string, depth: number): unknown {
  if (!value || typeof value !== "object" || depth > 4) {
    return undefined;
  }
  const record = value as Record<string, unknown>;
  if (record[key] !== undefined) {
    return record[key];
  }
  for (const child of Object.values(record)) {
    const result = findKeyValue(child, key, depth + 1);
    if (result !== undefined) {
      return result;
    }
  }
  return undefined;
}