竞争情报智能体

本页说明竞争情报智能体如何用 Web Agent 的 WebSearch 周期性收集竞品发布、定价、文档更新、招聘信号和新闻，在应用侧校验来源与置信后写入结构化情报库，并组装成可回溯的情报简报。读完本页，你能理解公开信息收集为什么只需要静默运行时凭证、长任务如何用 run.id 断点重连，以及历史判断为什么进带版本的情报库而不是 Memory。

适用场景

产品、市场或战略团队需要持续跟踪竞品发布、定价变化、文档更新和新闻动态。这些信号散落在官网、changelog、招聘页和媒体报道中，人工汇总不仅慢，还容易把二手转述当成一手事实；没有统一来源标注的简报，事后无法核对结论从哪来。

典型触发时机：

每周或每双周的情报简报例行产出，需要覆盖固定竞品清单。
竞品发布会或大版本更新后，需要快速评估对自身路线图的影响。
竞品招聘页出现新方向的岗位信号，需要纳入战略判断。

工程挑战

二手转述与一手事实：同一条"竞品发布了 X"，官网 changelog 与媒体转述的可信度完全不同。来源不分层的简报会把猜测当结论传播，事后无法纠错。
增量判定：团队真正要看的是"这期新增了什么"。没有带版本的历史条目库，每期简报都会重复报送已知事实，或漏掉悄悄发生的变化。
长任务的中断恢复：一期覆盖十几家竞品的收集可能跑几十分钟，进程重启、部署或超时都会打断它。不能断点重连的收集任务，只能整期重跑。

模块组合

模块	角色	说明
GenAuth	核心	运行时以静默委托签发短时效凭证（所有产品调用必需）；本场景默认无需交互式确认，接入登录态或写动作时才升级 interactive，见下方"何时需要交互式确认"。
Web Agent	核心	WebSearch 跨来源收集公开信号；`run()` 返回可重连句柄，长任务保存 `run.id` 后可随时 `attach()` 续接。
GUMem	不使用	情报条目、来源和置信是团队共享的业务数据，进结构化情报库；历史判断同样在情报库里带版本管理，被证伪时标失效而不是删除。Memory 不参与本场景。

竞争情报智能体：场景架构

何时需要交互式确认

所有产品调用都需要 GenAuth 委托令牌；公开只读场景用静默委托即可。本场景只读取公开可见页面，不需要用户在 Console 交互确认：

公开读取：用 delegateToken 静默签发运行时凭证即可（products: ['webSearch']）。凭证短时效、可随时撤销，每期收集按计划重新签发；每次签发与检索都进入审计链。
登录态或写动作：本场景明确不登录竞品产品、不注册试用账号、不绕过访问控制。如果确实需要登录态下的竞品研究，那是另一个场景，应走 mode: 'interactive' 交互式委托——见认证竞品话术智能体。

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

工作流程

竞争情报智能体：工作流程

团队定义竞品清单、关注主题和简报周期，维护在带版本的采集手册（playbook）里。
应用静默签发运行时凭证，并从情报库读取当前版本的采集手册和已报送事实。
应用启动 WebSearch 收集本期信号，保存 run.id 供中断后重连。
检查点：进程重启或 wait() 超时后，用 qoni.webSearch.attach(run.id, { token }) 续接同一任务，不整期重跑。
Web Agent 跨来源检索发布、定价、文档、招聘和新闻信号，逐条保留来源 URL。
应用侧解析结果并校验：无来源条目丢弃；corroboration < 2 的重要结论标记 single_source: true，进入待核实清单而不是简报正文；来源矛盾的条目并列保留，不擅自合并。
检查点：只有一个来源支撑的关键结论应进入待核实清单，等待人工核实后才能升级为简报事实；来源互相矛盾时保留矛盾记录，不擅自取舍。
校验后的条目连同采集手册版本写入结构化情报库；与历史条目比对，突出真正的增量。
应用组装情报简报（事实、推断、建议分层），附来源、置信标注和 audit id 交付分析负责人。
检查点：简报中每条结论都应能回溯到情报库中的具体条目与来源；无来源支撑的结论不应进入交付物。

示例代码

本页使用已发布到 npm 的 @qoniai/qoni 0.9.0；下载包附带同版本 SDK，由 npm ci 安装。默认研究 Firefox 官方产品与隐私资料；演示竞争对手是 Mozilla Firefox。公开网页由真实 SDK 读取，业务输入在 scenarios.ts 中标记为 public-demo。

本演示不读取或写入 GUMem；任务输入来自页面清单和明确提供的业务数据。Web Search 先取得 results[] 来源，再由 DoAnything 阅读和分析。接入自己的数据时，用 --input 指定 JSON 文件；需要用户同意时用 --interactive 发起委托，站点登录与交互处理见 Qoni SDK。

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

bash

cd examples/qoni
npm ci
npm run case -- competitive-intelligence-agent
# 使用自己的输入
npm run case -- competitive-intelligence-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('competitive-intelligence-agent', cliOptions(), inputFile())
// report.items：竞品、主张、来源、采集时间、置信度及支撑来源数。
// 校验字段：competitor, claim, sourceUrl, capturedAt, confidence, corroboration.
console.log(JSON.stringify(report, null, 2))

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

// 搜索并交叉检查竞品主张，保留不确定性。
research('competitive-intelligence-agent',
  'Use the supplied search results and official pages to produce sourced product intelligence. Return [{competitor,claim,sourceUrl,capturedAt,confidence,corroboration}]. Set corroboration to the number of actual cited independent sources, and retain uncertain claims as low confidence. Do not invent publication dates.',
  ['competitor','claim','sourceUrl','capturedAt','confidence','corroboration'], 'sourceUrl', ['Mozilla Firefox browser privacy official'], sample([firefox,privacy], { competitors:['Mozilla Firefox'], playbookVersion:'demo-v1' })),

这里的 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/competitive-intelligence-agent/report.json。report.items 包含竞品、主张、来源、采集时间、置信度及支撑来源数，字段为 competitor, claim, sourceUrl, capturedAt, confidence, corroboration；audit 保存委托 ID、审计 ID 和实际授权范围，http.json 记录脱敏请求状态。应用解析 DoAnything 输出，并检查必填字段和来源 URL 格式。内容判断仍需业务人员结合原始来源复核。

数据与记忆边界

这个场景涉及四类数据，没有一类属于 GUMem：

版本化规则：竞品清单、关注主题、置信分层标准——放在采集手册（playbook）里按版本管理，每期简报引用手册版本号。
业务状态：情报条目、来源指针、置信标注和历史判断——进结构化情报库；判断被证伪时旧条目标失效并指向新条目，演变可追溯。
审计记录：凭证签发与每次检索构成的行为链——由 GenAuth 维护。
用户 Memory：本场景不使用 GUMem。情报是团队共享的业务数据而不是某个用户的偏好；写进 Memory 会失去版本对账和团队共享能力。

失败处理

情况	推荐处理
收集任务中断（进程重启、部署、超时）	用保存的 `run.id` 调用 `attach()` 重连续接，不整期重跑。
来源页面无法访问或已下线	保留失效记录并降低相关条目置信度，不引用缓存内容冒充现行事实。
多来源对同一事实说法矛盾	并列写入各来源与差异，标注为待人工判断，不合并成单一结论。
重要结论只有单一来源支撑（`corroboration < 2`）	标记 `single_source: true` 进入待核实清单，人工核实后再升级为简报事实。
输出条目缺少来源	应用侧校验直接丢弃该条目，并在简报中标注丢弃数量。

生产注意点

市场判断应区分事实、推断和建议，避免把推断写成确定结论；分层发生在应用组装简报时，依据是情报库里逐条的来源与置信。收集只覆盖公开可见页面，不注册账号、不绕过访问控制，对目标站点的抓取频率应设置上限。定价页等固定目标如需长期盯守，参见供应商监控智能体的 Track 形态。

下一步

阅读快速开始跑通 Agent 身份与委托的最短路径。
阅读授权与浏览器沙盒了解受控会话的安全边界。
继续查看认证竞品话术智能体了解登录态下的相邻研究场景。

竞争情报智能体 ​

适用场景 ​

工程挑战 ​

模块组合 ​

何时需要交互式确认 ​

工作流程 ​

示例代码 ​

数据与记忆边界 ​

失败处理 ​

生产注意点 ​

下一步 ​