06 · 实践：知识库研究助手

本课交付：支持第二个主题的策略与测试。先运行本页基线，再完成 L6：替换策略，保留边界。最终作品按 结课项目验收。

本页保留最小同步基线，以下“未实现”均指本页示例。新增的异步执行与模型适配见 08，真实文档检索见 09；两个版本的状态与测试口径不同，请分别记录。

一、场景：这一步要解决什么问题

前五章已经定义了一个研究助手的职责：查找授权笔记、读取证据、决定是否继续，并在结束前检查结果。现在把这些设计放进一个能运行、能观察、能故意弄错的最小程序。

先明确范围：本章交付的是离线执行器，不是上线的智能问答服务。它的脚本决策器按固定步骤运行，目的是让控制流程可复现；接入真实模型是后续独立步骤。

二、目标与方案设计

本章目标：把接口、状态、权限和验证组合为可检查的教学系统。新增示例使用脚本化决策器，不调用模型；先观察执行控制，再明确接入模型时需要替换什么。资料核对：2026-09-06。

目标与完成条件

让助手回答“延时任务有哪些方案，什么情况下选哪种”，只读取授权笔记，给出条件化比较、来源及证据不足之处。此章是本专题原创教学设计；前半部分提供控制循环伪代码，后半部分提供可运行的离线示例，没有部署在线 Agent，也不需要填写密钥阅读。

完成条件：回答与问题相关；每条主要判断有可追溯证据；无法证实的部分明确标注；任务状态是 completed，而不是仅仅生成了一段文字。

最小组件

组件	职责
请求入口	获取用户目标与可信身份
控制循环	调用模型、选择分支、限制预算
工具层	搜索、读取、校验权限与参数
状态存储	保存证据、工具结果、运行状态
验收器	检查输出结构与证据覆盖

第一版只提供 search_notes 和 read_note 两个只读工具。不加入多 Agent、长期记忆和自动发布，先用最少组件测出哪里确实需要改进。

状态记录的关键字段

task_id / user_id / goal / status
model_version / prompt_version
tool_calls：call_id、工具、参数、结果、状态
evidence：文档 ID、版本、片段、来源
open_questions / final_answer / stop_reason

状态存储要能区分“未开始、进行中、已成功、明确失败、结果未知”。这个区别对未来增加写工具尤其重要。

分三次实现

1. 固定流程基线：搜索一次、读取、生成；先跑正常与空结果用例。
2. 增加决策循环：允许在证据不足时补充搜索，对比基线的收益和成本。
3. 加入可靠性检查：覆盖权限、限流、循环和冲突证据，验证停止状态。

每一步使用 评估章节 的同一组用例。最后再判断是否需要多 Agent，而不是先增加角色数量。

后端开发者的对应关系

Agent 编排可以理解为带模型决策的应用服务；工具是受控业务接口；状态记录类似任务表；检查点类似可恢复执行进度。模型的不确定性不会消除接口契约、身份隔离、幂等与可观测性的需要。

三、实现拆解

可运行示例：先把执行闭环跑通

提供两个可下载文件，放在同一目录：

• research-agent.mjs：执行器、两条固定笔记和脚本决策器。
• research-agent.test.mjs：运行时边界测试。

使用支持内置测试运行器与 structuredClone 的 Node.js 运行。在本项目根目录执行：

node docs/public/examples/research-agent.mjs
node --test docs/public/examples/research-agent.test.mjs

如果下载到其他目录，直接运行 node research-agent.mjs 和 node --test research-agent.test.mjs。

脚本不联网、不读写业务文件、不要求密钥。它使用内置教学数据，不会自动搜索本站文章。默认结果包含 completed、两条工具轨迹和一条已读取文档的引用。

阅读示例代码

这里展示与下载文件相同的源码，避免教程和示例分别维护：

// Offline teaching harness: a scripted decision adapter, NOT an LLM agent.
import { pathToFileURL } from 'node:url';

export const NOTES = [
  { id: 'delay', owner: 'alice', revision: '1', title: '延时任务',
    text: '选择方案前，需要明确允许延迟、故障恢复和重复执行的业务影响。' },
  { id: 'private', owner: 'bob', revision: '1', title: '内部计划',
    text: 'Only Bob may read this fixture.' }
];

// Small, explicit contract checks for these two teaching tools, not a general Schema engine.
function hasExactKeys(value, keys) {
  return value !== null && typeof value === 'object' && !Array.isArray(value)
    && Object.keys(value).length === keys.length
    && keys.every(key => Object.hasOwn(value, key));
}

export function run(decide, { user = 'alice', notes = NOTES, maxSteps = 8 } = {}) {
  if (!Number.isInteger(maxSteps) || maxSteps < 1) throw new Error('Invalid step limit');
  const state = { status: 'running', evidence: [], trace: [], feedback: [] };
  const finish = (status, extra = {}) => ({ ...state, status, ...extra });
  const seenIds = new Set();

  for (let step = 0; step < maxSteps; step++) {
    // The adapter receives a copy; it cannot mutate trusted runtime state.
    let decision;
    try { decision = decide(structuredClone(state)); }
    catch { return finish('failed', { reason: 'adapter_error' }); }

    if (!decision || typeof decision !== 'object') {
      state.feedback.push('invalid_decision');
      continue;
    }
    if (decision.kind === 'ask_user' && typeof decision.question === 'string'
        && decision.question.trim()) {
      return finish('needs_input', { question: decision.question });
    }
    if (decision.kind === 'final') {
      const refs = decision.citations;
      const valid = typeof decision.answer === 'string' && decision.answer.trim()
        && Array.isArray(refs) && refs.length > 0
        && refs.every(ref => ref && typeof ref.id === 'string' && typeof ref.revision === 'string'
          && state.evidence.some(e => e.id === ref.id && e.revision === ref.revision));
      if (valid) return finish('completed', { answer: decision.answer, citations: refs });
      state.feedback.push('answer_requires_read_evidence');
      continue;
    }
    if (decision.kind === 'insufficient') {
      return finish('insufficient_evidence');
    }
    if (decision.kind !== 'tool' || typeof decision.callId !== 'string' || !decision.callId) {
      state.feedback.push('invalid_tool_request');
      continue;
    }
    if (seenIds.has(decision.callId)) {
      state.feedback.push('duplicate_call_id');
      continue;
    }
    seenIds.add(decision.callId);
    const { name, args, callId } = decision;
    if (!['search_notes', 'read_note'].includes(name)) {
      return finish('denied', { reason: 'tool_not_allowed' });
    }
    let result;
    if (name === 'search_notes') {
      if (!hasExactKeys(args, ['query', 'limit'])
          || typeof args.query !== 'string' || !args.query.trim()
          || !Number.isInteger(args.limit) || args.limit < 1 || args.limit > 5) {
        state.feedback.push('invalid_search_arguments');
        continue;
      }
      result = notes.filter(n => n.owner === user && n.title.includes(args.query))
        .slice(0, args.limit).map(({ id, revision, title }) => ({ id, revision, title }));
    } else {
      if (!hasExactKeys(args, ['id', 'revision'])
          || typeof args.id !== 'string' || !args.id.trim()
          || typeof args.revision !== 'string' || !args.revision.trim()) {
        state.feedback.push('invalid_read_arguments');
        continue;
      }
      // No disclosure of whether another user's document exists.
      const note = notes.find(n => n.id === args.id && n.owner === user);
      if (!note) return finish('denied', { reason: 'not_found_or_not_allowed' });
      if (note.revision !== args.revision) {
        state.feedback.push('stale_revision');
        continue;
      }
      const { owner, ...publicFields } = note;
      result = publicFields;
      if (!state.evidence.some(e => e.id === note.id && e.revision === note.revision)) {
        state.evidence.push(result);
      }
    }
    state.trace.push({ callId, name, result });
  }
  return finish('exhausted', { reason: 'step_limit' });
}

export function scriptedDecider(state) {
  if (state.trace.length === 0) return {
    kind: 'tool', callId: 'search-1', name: 'search_notes', args: { query: '延时任务', limit: 3 }
  };
  if (state.trace.length === 1) {
    const first = state.trace[0].result[0];
    return first ? { kind: 'tool', callId: 'read-1', name: 'read_note',
      args: { id: first.id, revision: first.revision } } : { kind: 'insufficient' };
  }
  const evidence = state.evidence[0];
  return { kind: 'final', answer: evidence.text,
    citations: [{ id: evidence.id, revision: evidence.revision }] };
}

if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) {
  console.log(JSON.stringify(run(scriptedDecider), null, 2));
}

重点观察三个设计：模型适配器只能看到状态副本；资源身份由运行时传入；最终引用必须指向已实际读取的文档版本。搜索命中并不会自动进入证据集合，需要完成读取动作。

脚本里的 scriptedDecider 只按阶段返回固定动作。它用于验证程序的控制边界，不具备理解问题、规划或判断证据的模型能力，也不能用这些测试成绩宣称 Agent 的任务成功率。

控制循环伪代码

以下函数名都是抽象接口。真实实现需要接入模型、参数校验器、存储和工具执行器，并保留模型原始工具请求与对应调用 ID。

def run_agent(goal, trusted_user):
    state = new_state(goal, trusted_user)
    deadline = monotonic() + 60  # 教学预算

    for step in range(8):
        remaining = deadline - monotonic()
        if remaining <= 0:
            return stop(state, "budget_exhausted")

        try:
            decision = model_decide(
                context=select_context(state),
                tools=READ_ONLY_TOOLS,
                timeout=remaining,
            )
            validate_decision(decision)

            if decision.kind == "ask_user":
                return pause(state, decision.question)

            if decision.kind == "final":
                verdict = check_answer(decision.answer, state.evidence)
                if verdict.accepted:
                    return complete(state, decision.answer)
                record_feedback(state, verdict)
                continue

            # 此教学版本只接受每轮一个工具请求
            validate_tool_call(decision.call, READ_ONLY_TOOLS)
            authorize(trusted_user, decision.call)
            remaining = deadline - monotonic()
            if remaining <= 0:
                return stop(state, "budget_exhausted")
            result = execute_tool(
                decision.call,
                identity=trusted_user,
                timeout=min(5, remaining),
            )
            record_call_and_result(state, decision.call, result)
            checkpoint(state)

        except PermissionDenied:
            return stop(state, "permission_denied")
        except InvalidDecision as error:
            record_feedback(state, safe_error(error))
        except TransientError as error:
            # 继续循环也消耗轮数；正式实现应增加独立重试计数和退避
            record_feedback(state, safe_error(error))

    return stop(state, "step_limit")

check_answer 的结构检查可自动完成，引用是否真正支持结论仍需要语义评估。这段代码没有实现流式输出、多工具并发、完整异常分类或生产级恢复，不能直接复制当作完整服务。

源码应该从哪里开始读

先读 scriptedDecider 的三条分支，再读 run，不用从第一行一直读到最后一行。这样可以先看到调用方期望什么，再检查执行器如何限制它。

代码位置	关注的问题	运行中应观察到什么
state.trace.length === 0	第一条动作如何产生	请求搜索，而不是直接生成结论
name === 'search_notes'	搜索是否带权限过滤	只返回当前用户可读的候选
read_note 对应分支	什么时候成为证据	所有者和版本检查通过后加入 evidence
decision.kind === 'final'	如何验收引用	引用必须匹配已读取 ID 与版本
循环结束处	非法请求是否会无限运行	返回 exhausted，保留反馈

默认执行结果的关键字段如下，省略了正文和其他字段：

{
  "status": "completed",
  "trace": [
    { "callId": "search-1", "name": "search_notes" },
    { "callId": "read-1", "name": "read_note" }
  ],
  "citations": [{ "id": "delay", "revision": "1" }]
}

这里实际上有三次决策、两次工具执行；最终回答不算工具调用。completed 在当前实现中只表示通过已实现的结构验收，不代表已经完成真实研究任务。默认回答只是教学笔记正文，还没有三种方案的完整比较。

按现象排查，而不是反复重跑

现象	先检查	正确理解
找不到文件	是否位于项目根目录，下载文件是否同目录	这是启动路径问题，不是 Agent 决策问题
返回 insufficient_evidence	搜索结果是否为空	没有资料可以是正常结束
返回 denied	工具是否在白名单，资源是否属于用户	不应通过换用户参数绕过拒绝
返回 exhausted	feedback 中的错误与调用 ID	修正请求策略，不能只增加轮数
有引用但结论错误	原文是否支持具体判断	当前脚本不具备语义验收能力

四、验证与练习

测试覆盖与一个刻意暴露的局限

测试检查正常搜索读取、空结果、私有文档隔离、禁止工具、伪造引用、旧版本、无效参数、重复调用 ID、状态篡改、暂停与轮数上限。

2026-09-06 本地执行上述测试命令：17 项通过、0 项失败。新增用例覆盖额外字段、非对象参数、缺失字段和空白文档标识；这是离线执行器测试结果，不是模型评测分数。

最后一个测试特别重要：把最终答案换成与原文无关的句子，只要引用存在，结构验收仍可能通过。这个测试记录的是已知能力边界，而不是证明语义验收正确。接入真实模型后，需要使用 评估章 的语义评分与人工标注来补齐。

同理，本示例的白名单阻止任意工具执行，但没有真实模型，因此没有验证模型抵抗提示注入的能力。测试结果只适用于这里的确定性执行器。

扩展任务

先把一个无关问题交给脚本决策器，观察它仍按固定步骤搜索“延时任务”。这能直接看到它与真实 Agent 的区别。然后用测试注入不同决策器，练习参数错误后的修正、明确资料不足的结束方式，以及用户补充信息后的重新运行。

完成确定性控制测试后，再引入真实模型评估。不要同时加入多 Agent、长期记忆、写工具和新检索器，否则一次失败很难判断来自哪项变化。

五、从教学示例走向真实应用

如何接入真实模型

保留 run 的职责边界，把决策器改造成模型适配器，并将同步循环改成可等待调用的异步控制器。适配器把提供方响应规范化为 tool、final、ask_user、insufficient 等内部事件。不要让提供方的 SDK 类型渗透到每个业务工具中。

接入时逐项处理：

1. 工具 Schema 与提供方实际支持的字段一致。
2. 原始模型工具请求与工具结果通过调用 ID 成对保存。
3. 多工具请求有明确的顺序、并发和失败策略。
4. 模型与工具调用均使用剩余任务时限，接入取消信号。
5. 区分提供方错误、网络错误、业务错误与验收失败。
6. 验收成功后再将结果交给用户或下一条工作流。

示例中的循环只有轮数限制，没有网络超时、Token 计费和持久化恢复。前文伪代码说明了总时限思路；生产实现还应检查模型返回后的剩余预算，且设置超时不等于远端操作必然取消。

从内存状态走向持久化

在进程内保存 state 只能支持本次运行。需要跨进程恢复时，必须保存任务标识、已执行动作、证据版本、剩余工作和运行配置。LangGraph 使用检查点等机制持久化图执行状态；应按所用版本核对恢复行为。Persistence 文档。

本案例可以用以下事件模型讨论恢复：

TaskCreated → ToolRequested → ToolSucceeded → EvidenceAdded
                          ↘ ToolFailed
                          ↘ ToolOutcomeUnknown
AnswerProposed → AnswerRejected / TaskCompleted

如果进程在工具成功后、状态保存前退出，恢复时可能重放调用。只读操作通常影响成本与时效；写操作则可能重复产生结果。外部幂等键与业务对账需要单独设计，不能仅依赖任务存储。AWS 幂等 API。

给后端项目划分模块

可以把模型适配、知识库访问和状态存储放在基础设施层，把“允许怎样的研究动作”和验收条件放在应用或领域逻辑中。控制器只负责接收请求、身份和任务状态查询。这里是本专题的模块划分建议，并非要求采用某个框架。

api/             提交任务、查询进度、取消
application/     研究循环、预算、验收协调
model-adapter/   模型请求与内部事件的转换
tools/           搜索和读取的授权、校验、执行
storage/         文档、任务、证据和轨迹
evaluation/      固定用例、评分规则、回归报告

六、接入验收与实验记录

接入前的验收清单：避免一次改太多

将真实模型接入视为独立里程碑，不需要把本站改成在线服务才能理解这套设计：

层次	最小产物	如何验证	本站状态
工具执行边界	搜索、读取、字段与所有者校验	确定性单元测试	已提供离线示例
模型适配	响应转内部事件、调用 ID 关联	桩响应契约测试 + 真实小样本	设计，未接入
真正检索	文章解析、片段版本与索引	标注查询集 Recall@k	设计，未实现
答案支持检查	结论到证据的映射	人工标注与评分器校准	未实现语义验收
跨进程恢复	持久状态与重复操作处理	在动作前后注入退出故障	设计，未实现

模型适配器测试至少覆盖：空响应、非法 JSON、未知工具、同一轮多个调用、结果 ID 错配、服务拒绝、流式参数尚未完整就断线。流式输出的半段参数不能提前执行；收到完整结构并完成校验后才能分派工具。接入哪个提供方，就按其当时文档实现适配，不把本章内部事件当成通用 API 字段。

把实验结果写回博客

每次实验记录：问题、假设、固定条件、唯一改动、实际命令、结果、失败样本、不能证明的事情。比如“拒绝额外字段的测试通过”只证明对应校验分支，不证明系统具备真实身份认证。

当前示例继续保留“引用存在但语义无关也可能 completed”的反例。这个边界不应该靠给状态换名字掩盖；后续接入真正验收器时，再增加能拒绝该答案的测试并更新完成语义。

想把这些工程约束扩展到写操作与用户委托，可继续读安全与身份边界，以及SSO 与 Agent Identity 项目设计。后者是模拟项目，不是本站已上线能力。

七、本章小结

到这里，你能运行一个搜索、读取、引用检查的闭环，也能解释它为什么停下来。保留前文列出的能力边界，再逐项接入模型、持久化和真实检索。优先验证每次新增能力的收益，不要把教学执行器包装成生产级 Agent。

八、参考资料

• LangGraph Persistence：持久化与任务恢复的框架实例。
• LangGraph Interrupts：暂停恢复与副作用的处理边界。
• Making retries safe with idempotent APIs：重复执行与业务一致性。

返回 Agent 专题目录。本文场景与推演为原创教学设计，参考写法及资料边界见专题目录。