客服知识智能体

本页说明客服知识智能体如何在客户级委托下查询公开文档和授权可见的工单系统，结合该客户的历史上下文生成带来源的答案草稿。读完本页，你能理解这个场景为什么三个模块都是核心、标准答案为什么以知识库为准而不是记忆，以及客户数据的裁剪边界。

适用场景

客服团队需要 Agent 回答产品问题、排查常见故障，或根据客户历史记录给出更贴合的处理建议。答案素材分散在产品文档、状态页、社区帖和工单系统中，人工逐个翻找拖慢响应；直接把客服账号交给脚本，则工单系统里客户数据的访问范围完全失控。Agent 的产出是给支持团队复核的答案草稿，不直接回复客户。

典型触发时机：

• 高频问题涌入（例如版本发布后），需要快速产出带来源的标准答案草稿。
• 疑难工单需要汇总历史相似工单、文档和社区讨论后再答复。
• 产品公告或已知问题更新后，需要核对现有答案是否已经过期。

工程挑战

• 答案素材新旧混杂：产品文档、状态页、社区帖各自的更新节奏不同，一条公告发布，散落各处的旧答案就过期了。草稿不带来源链接和时间，客服无法判断每条结论还可不可信。
• 两类知识容易混淆：标准答案和产品知识是团队维护、按版本更新的知识库（canonical）内容；该客户的环境事实和历史处理记录才是客户上下文。把知识库内容抄进客户记忆，产品一更新就变成一批过期私货，且无法集中修正。
• 客户数据的裁剪边界：客服账号能看到的工单远多于本次问题需要的。传给网页任务的上下文必须裁剪到本次问题所需，无关客户隐私一旦进入任务就无法收回。

模块组合

模块	角色	说明
GenAuth	核心	交互式委托限定在当前客户上下文；撤销与审计链覆盖全部访问尝试。
Web Agent	核心	查询公开文档、状态页和社区帖（WebSearch 定位来源），并在授权范围内检索相似工单。
GUMem	核心	只承载该客户的历史问题、处理记录要点和已确认环境事实；标准答案与产品知识属于知识库，不进 Memory。

权限与委托边界

Agent 本身不持有任何固有权限。每次任务的实际权限是三个集合的交集：客服人员真实权限 ∩ 本次显式委托范围 ∩ 企业批准边界。落到这个场景：

• 委托范围只覆盖"读取当前客户相关的工单、历史记录和公开知识来源"，不包含回复客户、关闭工单或修改客户数据。
• 委托凭证短时效，单次答疑任务建议分钟级有效期，过期后需重新委托。
• 客服人员或管理员可随时撤销授权；撤销后新的工单读取请求立即失败。
• 越权尝试（例如读取与本次问题无关的其他客户工单）会被拒绝并留痕——审计链覆盖全部尝试，不只是成功行为。

注意：本样例申请 doAnything 产品权限和 GUMem 读写 scope。工单系统的域名清单和客户范围等业务边界需要由应用和下游服务配置、校验；产品 scope 和 prompt 中的规则不提供这些细粒度限制。本样例未展示该配置。授权语义见 Delegate Token 与缩权。

工作流程

1. 客服人员打开客户会话并触发答疑任务。

2. GenAuth 通过交互式委托让客服人员确认授权，签发限定在该客户上下文的委托凭证。

3. GUMem 召回该客户的历史问题、处理记录要点和已确认环境事实——不召回知识库内容。

4. Web Agent 查询最新产品文档、状态页和社区讨论，并在授权范围内检索相似工单。

   检查点：传给 Web Agent 的任务上下文必须经过裁剪，只含本次问题所需信息，不携带无关客户隐私。

5. Agent 汇总各来源信息，生成答案草稿，每条关键结论附来源链接和时间。

6. 客服人员复核草稿，修改后回复客户；Agent 不直接触达客户。

   检查点：草稿中每条结论都应能回溯到具体来源；引用的文档如与产品公告冲突，标注待核实而不是擅自取舍。

7. 复核通过的标准答案更新到知识库（由团队维护）；该客户的环境事实与处理要点写回 GUMem。

示例代码

本页使用已发布到 npm 的 @qoniai/qoni 0.9.0；下载包附带同版本 SDK，由 npm ci 安装。默认回答一张虚构工单：如何在桌面开始使用 Firefox；来源是 Mozilla 官方入门文档，偏好是简短回复。公开网页由真实 SDK 读取，业务输入在 scenarios.ts 中标记为 public-demo。

本场景先为独立演示用户写入并召回已确认偏好，再把召回结果加入网页任务。DoAnything 直接打开所给页面，按场景任务读取并形成输出。接入自己的数据时，用 --input 指定 JSON 文件；需要用户同意时用 --interactive 发起委托，站点登录与交互处理见 Qoni SDK。

下载完整可运行样例，或在文档仓库中执行：

cd examples/qoni
npm ci
npm run case -- support-knowledge-agent
# 使用自己的输入
npm run case -- support-knowledge-agent --input /path/to/input.json

运行前设置服务端 QONI_ACCESS_KEY、QONI_SECRET_KEY。QONI_USER_ID 可指定业务已识别的 GenAuth 用户；未指定时，本地演示从绑定用户池取用户。涉及演示 Memory 写入时，程序创建独立用户以避免修改业务用户的偏好。

本场景的真实执行入口：

import { cliOptions } from '../runtime.js'
import { inputFile, runScenario } from '../run-case.js'

// 召回已确认偏好，再从官方文档生成客服答案草稿。
const report = await runScenario('support-knowledge-agent', cliOptions(), inputFile())
// report.items：结论、来源引用和采集时间，交客服人员复核。
// 校验字段：conclusion, sourceRef, capturedAt.
console.log(JSON.stringify(report, null, 2))

入口按场景 ID 读取下面这段定义，代码直接引用实际执行的 scenarios.ts，注释按页面语言显示：任务描述、输出字段、来源字段、Web Search 查询（如有）、是否使用 Memory，以及演示输入。执行链会在任务描述后追加输入数据、搜索来源、召回的 Memory 和通用安全约束，拼成最终的任务描述，完整拼装逻辑见下方执行链。

// 召回已确认偏好，再从官方文档生成客服答案草稿。
browser('support-knowledge-agent',
  'Answer the synthetic support question with sourced public documentation. Return [{conclusion,sourceRef,capturedAt}]. Identify uncertainty rather than inventing instructions. Draft only: do not reply to a customer, close a ticket or change any account.',
  ['conclusion','sourceRef','capturedAt'], 'sourceRef', sample([support], { ticketId:'demo-ticket-1',question:'How can I start using Firefox on a desktop?',confirmedPreferences:['Keep replies concise.'] }), true),

这里的 runScenario()、browser()、research() 和 sample() 是样例包的应用函数。实际 SDK 调用都在下面的执行链中：委托与权限查询 → 所需的 Memory/搜索 → 网页任务或监控 → 校验与保存。fields、sourceField 规定本场景如何校验业务输出；它们由应用使用。辅助函数的完整实现在样例包的 runtime.ts 中。

查看实际 SDK 执行链

// 场景构造器：browser() 只调用 DoAnything；research() 先 Web Search 再 DoAnything；sample() 标记 public-demo 演示输入。
const firefox = 'https://www.mozilla.org/en-US/firefox/new/'
const manifesto = 'https://www.mozilla.org/en-US/about/manifesto/'
const privacy = 'https://www.mozilla.org/en-US/privacy/firefox/'
const support = 'https://support.mozilla.org/en-US/kb/get-started-firefox-overview-main-features'
// 演示文案规则会加入 prompt；它们不替代服务端权限或业务规则校验。
const policy = {
  version: 'demo-2026-09',
  approvedClaims: ['Describe only features supported by the cited page.'],
  forbiddenClaims: ['guaranteed security', '100% private', 'unverified pricing or performance'],
  voice: 'concise and warm',
}
const sample = (pages: string[], business: JsonObject = {}): JsonObject => ({
  dataset: 'public-demo', pages, policy, business,
  notice: 'Business records are synthetic demonstration inputs. Referenced websites and SDK execution are real.',
})
// fields 是必填输出字段；sourceField 指定需要校验 URL 格式的字段，memory 控制是否调用 GUMem。
const browser = (id: string, task: string, fields: string[], sourceField: string | undefined, input: JsonObject, memory = false): Scenario => ({
  id, products: ['doAnything'], task, fields, sourceField, input, memory,
})
const research = (id: string, task: string, fields: string[], sourceField: string, queries: string[], input: JsonObject, memory = false): Scenario => ({
  id, products: ['webSearch', 'doAnything'], task, fields, sourceField, queries, input, memory,
})

import { QoniScopes, type JsonObject, type RunResult } from '@qoniai/qoni'
import { readFileSync } from 'node:fs'
import { getScenario, type Scenario } from './scenarios.js'
import { appendTrace, checkInputCoverage, cleanupDemoUser, createContext, delegate, handleInteraction, inputEntryCount, isolateDemoUser, object, readWithRetry, renderScreenshot,
  save, saveArtifacts, searchHits, settled, settleRun, validateItems, withCleanup, type Context, type Options } from './runtime.js'

export async function runScenario(id: string, options: Options = {}, input?: JsonObject) {
  // 按场景 ID 读取真实任务定义；--input 只替换业务输入。
  const scenario = getScenario(id)
  const data = input ?? scenario.input
  if (scenario.memory && data.dataset !== 'public-demo' && options.mode !== 'interactive' && !options.userId && !process.env.QONI_USER_ID) {
    throw new Error('Business Memory writes require the current QONI_USER_ID; do not select an arbitrary bound user')
  }
  const context = await createContext(id, { ...options,
    skipUserResolution: scenario.memory && data.dataset === 'public-demo' })
  return withCleanup(context, async register => {
    register('isolated demonstration user', () => cleanupDemoUser(context))
    // 演示偏好写入隔离用户；业务 Memory 必须属于明确识别的当前用户。
    if (scenario.memory && data.dataset === 'public-demo') await isolateDemoUser(context)
    return await executeScenario(context, scenario, data)
  })
}

export async function executeScenario(context: Context, scenario: Scenario, input: JsonObject) {
  const pages = input.pages
  if (!Array.isArray(pages) || !pages.length || pages.some(page => typeof page !== 'string' || !/^https:\/\//.test(page))) {
    throw new Error('Input pages must be an array of HTTPS URLs')
  }
  if (input.requiresLogin === true && context.mode !== 'interactive') {
    throw new Error('Targets that require sign-in need --interactive and user-controlled login')
  }
  const memoryScopes = scenario.memory
    ? [QoniScopes.GUMEM_MEMORY_READ, QoniScopes.GUMEM_MEMORY_WRITE, QoniScopes.GUMEM_MESSAGE_WRITE] : []
  // delegate() 是应用函数，内部调用 delegateToken / completeDelegateToken。
  const grant = await delegate(context, scenario.id, scenario.products, memoryScopes)
  // 查询实际授权范围；readWithRetry() 只对可重试的读取错误有限重试。
  const { data: tokenInfo } = await readWithRetry(context, 'delegation introspection',
    () => context.qoni.genauth.introspectDelegationToken({ token: grant.token }))
  const info = object(tokenInfo)
  if (info.active !== true) throw new Error('The delegation token is not active')
  const audit = { grantId: grant.grantId, auditId: grant.auditId, scopes: info.scope }
  let memory: unknown
  if (scenario.memory) {
    // Session 将本次对话关联到当前用户，sessionId 由应用生成。
    const sessionId = `${scenario.id}-${Date.now()}`
    await context.qoni.gumem.createSession({
      token: grant.token, userId: context.userId, sessionId, title: scenario.id,
    })
    const preferences = object(input.business ?? {}).confirmedPreferences
    if (Array.isArray(preferences) && preferences.length) {
      // 只写入业务已经确认的偏好；sync: true 请求同步处理这次写入。
      await context.qoni.gumem.addMessages({ token: grant.token, userId: context.userId, sessionId, sync: true,
        messages: [{ role: 'user', content: `Confirmed demonstration preferences: ${preferences.join('; ')}` }] })
    }
    // 召回与任务有关的偏好，作为后续 prompt 的上下文。
    memory = (await readWithRetry(context, 'GUMem recall', () => context.qoni.gumem.recall({ token: grant.token, sessionId,
      query: 'Confirmed preferences relevant to this task', details: true }))).data
    save(context, 'memory.json', { sessionId, context: memory })
    if (Array.isArray(preferences) && preferences.length && !preferences.every(value => JSON.stringify(memory).includes(String(value)))) {
      throw new Error('Recall did not include the confirmed preferences just written by this demo')
    }
  }

  if (scenario.products.includes('track')) return runMonitor(context, scenario, input, grant.token, audit)

  let hits: ReturnType<typeof searchHits> = []
  if (scenario.queries) {
    // Web Search 返回真实 results[]；搜索来源将提供给 DoAnything 阅读。
    const search = await context.qoni.webSearch.run({ token: grant.token, prompt: scenario.queries, maxResultsPerQuery: 3 })
    save(context, 'search-ref.json', { runId: search.id, audit })
    await withCleanup(context, async register => {
      register('Web Search run', () => search.cancel('Documentation demonstration cleanup'))
      const result = await settleRun(context, search)
      save(context, 'search-result.json', result)
      settled(result)
      hits = searchHits(result.output)
    })
  }

  // 逐项对应输入的场景按输入项数要求条数，其余场景最多两项。
  const requiredItems = inputEntryCount(scenario.id, input)
  // 拼装场景任务、业务输入、搜索来源和 Memory；这些都是应用约定的上下文。
  const prompt = [scenario.task, `Task inputs: ${JSON.stringify(input)}`,
    `Search sources: ${JSON.stringify(hits)}`, `Confirmed memory: ${JSON.stringify(memory ?? null)}`,
    `Actual collection time: ${new Date().toISOString()}`,
    requiredItems === undefined
      ? 'Inspect the supplied sources. Return at most two items in the requested JSON array, without prose or Markdown.'
      : `Inspect the supplied sources. Return exactly ${requiredItems} item${requiredItems === 1 ? '' : 's'} in the requested JSON array, one per input entry, without prose or Markdown.`,
    'Keep synthetic demonstration data identified as synthetic. Do not send messages, publish, pay, edit accounts or submit forms.',
    input.requiresLogin === true ? 'Request user sign-in through an interaction when required; never enter credentials yourself.' : 'Public demonstration sources only; do not sign in.',
  ].join('\n\n')
  // 使用同一委托启动 Agent；capture 接收服务端交付的截图，不保证每步都有图。
  const run = await context.qoni.doAnything.run({ token: grant.token, prompt, capture: { screenshots: true } })
  save(context, 'run-ref.json', { runId: run.id, session: run.sessionRef, audit })
  let result: RunResult
  const trace = (event: { type: string; data: unknown }) => {
    context.eventCounts[event.type] = (context.eventCounts[event.type] ?? 0) + 1
    if (['progress','message','done'].includes(event.type)) appendTrace(context, event)
    if (event.type === 'browserLiveUrlChanged') {
      const liveUrl = object(event.data).liveUrl
      if (typeof liveUrl === 'string') context.browserUrl = liveUrl
    }
  }
  return withCleanup(context, async register => {
    register('DoAnything run', () => run.cancel('Documentation demonstration cleanup'))
    if (context.delivery === 'events') {
      // --events 实时读取事件；普通交互对象先转换为 SDK 句柄再交给用户处理。
      for await (const event of run.events({ signal: AbortSignal.any([context.abort.signal, AbortSignal.timeout(context.timeoutMs)]) })) {
        trace(event)
        if (event.type === 'screenshot') renderScreenshot(context, event.image)
        if (event.type === 'interaction') await handleInteraction(context, run.interactionHandle(event.data))
      }
      result = await settleRun(context, run)
    } else {
      // 回调模式在 wait 内接收同一次任务的事件；这里的辅助函数会保存图像和询问用户。
      result = await settleRun(context, run, { onEvent: trace,
        onScreenshot: (image, index) => renderScreenshot(context, image, index),
        onInteraction: interaction => handleInteraction(context, interaction) })
    }
    save(context, 'result.json', result)
    settled(result)
    await saveArtifacts(context, result)
    if (scenario.id === 'landing-page-audit-agent' && context.screenshots === 0) throw new Error('The landing-page audit did not deliver the requested screenshot')
    // 应用校验必填字段和来源 URL 格式；事实准确性仍需业务复核。
    const items = validateItems(result.output, scenario.fields, scenario.sourceField)
    // 要求逐项对应输入的场景，再核对每个输入项都有对应输出。
    checkInputCoverage(scenario.id, items, input)
    const report = { scenario: scenario.id, dataset: input.dataset, passed: true, runId: run.id,
      status: result.status, items, audit, screenshots: context.screenshots, interactions: context.interactions,
      events: context.eventCounts, artifactIds: result.artifacts.map(artifact => artifact.id) }
    save(context, 'report.json', report)
    return report
  })
}

async function runMonitor(context: Context, scenario: Scenario, input: JsonObject, token: string, audit: JsonObject) {
  // Track 单独创建 monitor：声明目标、抽取字段并请求每小时调度。
  const monitor = await context.qoni.track.create({ token, prompt: scenario.task,
    targetUrls: input.pages, extractionSchema: { heading: 'string', source_url: 'string' },
    tickInstructions: `Open the target URLs and read the actual visible heading. Return a JSON object with heading and source_url. ${scenario.task}`,
    triggerDsl: { on: 'change' }, schedule: { kind: 'interval', intervalSeconds: 3600 } })
  save(context, 'monitor-ref.json', { id: monitor.id, audit })
  return withCleanup(context, async register => {
    register('Track monitor', () => monitor.delete())
    const definition = await monitor.get()
    save(context, 'monitor-definition.json', definition)
    if (object(definition.schedule).intervalSeconds !== 3600) throw new Error('Track did not persist the requested schedule interval')
    // 立即运行一次，再读取该 runId 的实际抽取；completed 本身不足以证明取数成功。
    const tick = await monitor.runNow()
    save(context, 'tick.json', tick)
    if (tick.state !== 'completed') throw new Error(`Track execution failed: ${tick.state} / ${tick.error ?? ''}`)
    const runId = tick.runId
    if (typeof runId !== 'string') throw new Error('Track tick did not return a runId')
    const detail = await monitor.run(runId)
    save(context, 'tick-detail.json', detail)
    if (detail.state !== 'completed' || !detail.extracted || !Object.keys(object(detail.extracted)).length) {
      throw new Error('Track did not extract page data')
    }
    const extracted = object(detail.extracted)
    if (typeof extracted.heading !== 'string' || !extracted.heading.trim() ||
      typeof extracted.source_url !== 'string' || !/^https:\/\//.test(extracted.source_url)) {
      throw new Error('Track extraction is missing a heading or source URL')
    }
    const normalizeUrl = (value: string) => { const url = new URL(value); url.hash = ''; return url.href.replace(/\/$/, '') }
    if (!(input.pages as string[]).some(url => normalizeUrl(url) === normalizeUrl(String(extracted.source_url)))) {
      throw new Error('The Track source URL is not a configured target')
    }
    // 检查暂停/恢复是否保存；withCleanup() 在退出时调用 monitor.delete()。
    await monitor.pause()
    if ((await monitor.get()).status !== 'paused') throw new Error('Track did not persist the paused state')
    await monitor.resume()
    if ((await monitor.get()).status !== 'active') throw new Error('Track did not persist the active state')
    const report = { scenario: scenario.id, dataset: input.dataset, passed: true, monitorId: monitor.id,
      runId, state: detail.state, outcome: detail.outcome, extracted: detail.extracted, audit }
    save(context, 'report.json', report)
    return report
  })
}

export function inputFile(): JsonObject | undefined {
  const index = process.argv.indexOf('--input')
  return index >= 0 ? object(JSON.parse(readFileSync(process.argv[index + 1], 'utf8'))) : undefined
}

结果写入 output/support-knowledge-agent/report.json。report.items 包含结论、来源引用和采集时间，交客服人员复核，字段为 conclusion, sourceRef, capturedAt；audit 保存委托 ID、审计 ID 和实际授权范围，http.json 记录脱敏请求状态。应用解析 DoAnything 输出，并检查必填字段和来源 URL 格式。内容判断仍需业务人员结合原始来源复核。

记忆策略

• 进 Memory：该客户的历史问题、处理记录要点和已确认环境事实（带来源与时间）——只属于这个客户的上下文。
• 不进 Memory：标准答案、产品知识和故障排查手册。它们是知识库（canonical）内容，由团队按版本维护、任务时注入；抄进 Memory 会在产品更新后变成无法集中修正的过期副本。
• 更正：客户环境变化导致旧事实失效时（例如客户升级了版本），把旧事实标记为已失效并指向新记忆，而不是物理删除，保证历史答复可追溯。

失败处理

情况	推荐处理
工单系统登录态失效	任务挂起，通知客服人员重新登录，从断点继续。
各来源答案互相矛盾	如实呈现分歧并标注置信度，交客服人员判断，不硬猜。
委托范围外的客户数据请求	直接拒绝并记录，事后可在审计链中查到未遂访问。
召回的客户事实与最新工单冲突	以本次工单确认的事实为准，把旧事实标记为已失效并写回 GUMem。

生产注意点

标准答案与产品知识以知识库为准：复核通过的答案更新到知识库供全团队复用，不写入单个客户的 Memory。客户隐私数据不要传给不需要的网页任务，Web Agent 只应接收经过裁剪的任务上下文。Agent 的产出始终是给支持团队的草稿，不应配置为直接回复客户；对外答复的最终责任在复核的客服人员。

下一步

• 阅读 授权与浏览器沙盒 了解受控会话的安全边界。
• 阅读 快速开始 跑通 Agent 身份与委托的最短路径。
• 继续查看 客户入驻智能体 了解相邻场景。