品牌一致性智能体

本页说明品牌一致性智能体如何在只读委托下读取官网、帮助中心、社媒主页、市场平台展示页或 CMS 草稿，对照版本化的术语表和宣传主张规则跨渠道检查一致性，并输出带来源的差异清单。读完本页，你能理解这个场景需要哪些模块、权限边界如何收窄，以及术语规则和巡检基线各自应该放在哪里。

适用场景

Marketing、内容和产品团队需要在发布前或周期性检查多个页面是否符合品牌规范。品牌表达散落在官网、社媒、商店页和帮助中心等多个渠道，各渠道由不同团队维护，术语漂移和过期宣传主张很难靠人工发现；渠道改版后差异还会持续累积。

典型触发时机：

• 品牌术语表或宣传主张口径更新后，需要排查全部渠道的存量表达。
• 周期性品牌巡检，需要对官网、社媒和商店页做一致性抽查。
• 某个渠道页面改版后，需要确认改动没有偏离品牌规范。

工程挑战

• 渠道分散且各自演进：官网、社媒、商店页、帮助中心由不同团队维护，术语漂移是渐进发生的，人工抽查只能看到单页快照，看不到跨渠道差异的全貌。
• 差异需要证据而不是印象："这个页面语气不对"驱动不了改写；每条差异必须落到具体页面、具体表述和触发的具体规则，否则内容 owner 无从下手。
• 一致性基准本身在变：术语表和宣传主张口径持续更新，巡检结论必须绑定规则版本、基线必须按版本归档，否则新旧两期巡检结果无法对比，也说不清一条差异是新漂移还是旧问题。

模块组合

模块	角色	说明
GenAuth	核心	CMS、渠道后台等授权页面的只读委托、撤销与审计链；周期性巡检每次触发新签发凭证。
Web Agent	核心	受控会话逐页抽取文本、CTA 和视觉上下文并保留来源；关键页面可用 Track 持续盯守变化——变化判定由确定性规则决定，不由模型措辞决定。
GUMem	不使用	术语表和宣传主张规则是版本化配置，放在你的规则库（policy store）按版本注入；巡检基线和差异结论是业务状态，归档到你的结构化存储。这个场景没有需要进入 Memory 的用户长期偏好。

权限与委托边界

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

• 委托范围只覆盖"读取指定站点、CMS collection 和渠道页面"，不包含编辑内容、发布页面或修改 listing。
• 委托凭证短时效，单次巡检任务建议分钟级有效期；周期性盯守由每次触发时新签发的凭证执行。
• 用户或管理员可随时撤销授权；撤销后新的页面读取请求立即失败。
• 越权尝试（例如读取委托清单外的 workspace）会被拒绝并留痕——审计链覆盖全部尝试，不只是成功行为。

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

工作流程

1. 用户选择要检查的站点、渠道或页面列表，以及巡检维度（术语、语气、宣传主张），并在 Qoni Console 确认本次交互式授权。

2. GenAuth 为本次任务签发最小权限的只读委托凭证。

3. 应用从规则库读取当前版本的术语表、品牌 voice 和宣传主张规则，注入任务描述。

4. Web Agent 逐页抽取各渠道页面的文本、CTA 和视觉上下文，保留来源 URL 与截图。

   检查点：渠道后台或 CMS 预览遇到登录墙、验证码或风控页时，Web Agent 应升级给人处理，而不是静默绕过。

5. Agent 对照规则比对各渠道表达，标注术语漂移、语气偏离和过期宣传主张，每条附页面来源和触发的规则编号。

6. 应用侧校验输出契约：缺少页面来源或规则编号的差异直接丢弃；本期巡检基线按规则版本归档到你的结构化存储。

7. 对需要持续盯守的关键页面，配置 Track 按确定性规则监测后续变化。

8. Agent 输出差异清单、建议改写和来源列表，附规则版本号与 audit id。

   检查点：差异清单中每条不一致项都应能回溯到具体页面来源；无来源支撑的结论不应进入交付物。

示例代码

本页使用已发布到 npm 的 @qoniai/qoni 0.9.0；下载包附带同版本 SDK，由 npm ci 安装。默认比较 Firefox 产品页与 Mozilla 宣言，并应用演示品牌规则检查证据和措辞。公开网页由真实 SDK 读取，业务输入在 scenarios.ts 中标记为 public-demo。

本演示不读取或写入 GUMem；任务输入来自页面清单和明确提供的业务数据。DoAnything 直接打开所给页面，按场景任务读取并形成输出。接入自己的数据时，用 --input 指定 JSON 文件；需要用户同意时用 --interactive 发起委托，站点登录与交互处理见 Qoni SDK。

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

cd examples/qoni
npm ci
npm run case -- brand-consistency-agent
# 使用自己的输入
npm run case -- brand-consistency-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('brand-consistency-agent', cliOptions(), inputFile())
// report.items：页面、差异或观察、原文、规则 ID 和来源。
// 校验字段：page, diff, quote, ruleId, sourceUrl.
console.log(JSON.stringify(report, null, 2))

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

// 对照同一套品牌规则检查多个公开页面。
browser('brand-consistency-agent',
  'Compare product wording on the supplied pages against the approved policy. Return [{page,diff,quote,ruleId,sourceUrl}]. For every inspected page, diff must explain the observed difference or supported alignment in non-empty text; never return null. If no inconsistency is found, describe the alignment supported by quote. Capture and attach a screenshot as evidence. Do not edit or publish anything.',
  ['page','diff','quote','ruleId','sourceUrl'], 'sourceUrl', sample([firefox,manifesto], { rules:[{id:'evidence-and-voice',text:'Claims should be supported and use clear wording.'}] })),

这里的 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/brand-consistency-agent/report.json。report.items 包含页面、差异或观察、原文、规则 ID 和来源，字段为 page, diff, quote, ruleId, sourceUrl；audit 保存委托 ID、审计 ID 和实际授权范围，http.json 记录脱敏请求状态。diff 必须是非空说明；未发现差异时，也应结合 quote 说明符合了哪条规则，不能返回 null。应用解析 DoAnything 输出，并检查必填字段和来源 URL 格式。内容判断仍需业务人员结合原始来源复核。

数据与记忆边界

这个场景涉及四类数据，都不需要进入 GUMem：

• 版本化规则：术语表、品牌 voice、宣传主张规则——放在规则库（policy store）里按版本管理，每条差异引用规则版本号。
• 业务状态：巡检基线、差异清单、页面截图——按规则版本归档到你的结构化存储，供跨期对比与追溯。
• 审计记录：grantId 与 auditId 构成的委托与行为链——由 GenAuth 维护。
• 用户 Memory（可选）：这个场景默认不召回也不写回；如果未来出现需要跨任务保留的用户长期偏好，再评估是否引入 GUMem。

失败处理

情况	推荐处理
渠道后台登录态失效	任务挂起，通知用户重新登录，从断点继续。
页面结构变化导致抽取失败	按失败处理并回放会话记录，不输出无证据的差异项。
委托范围外的 workspace 请求	直接拒绝并记录，事后可在审计链中查到未遂访问。
输出差异缺少页面来源或规则编号	应用侧校验直接丢弃该条目，并在差异清单中标注丢弃数量。

生产注意点

不要自动覆盖发布内容。批量修改前需要内容 owner 确认。Agent 只输出差异清单与建议，不直接修改任何渠道的线上内容；差异清单应带规则版本号，术语表更新后旧基线不追溯改判。对外部渠道页面的巡检频率应设置上限，避免对目标站点造成压力。

下一步

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